"""The algorithm execution API."""
from math import isfinite
from typing import Any, Final, TypeVar
from pycommons.io.path import Path
from pycommons.types import type_error
from moptipy.api._process_base import _ProcessBase
from moptipy.api._process_no_ss import _ProcessNoSS
from moptipy.api._process_no_ss_log import _ProcessNoSSLog
from moptipy.api._process_ss import _ProcessSS
from moptipy.api._process_ss_log import _ProcessSSLog
from moptipy.api.algorithm import Algorithm, check_algorithm
from moptipy.api.encoding import Encoding, check_encoding
from moptipy.api.objective import Objective, check_objective
from moptipy.api.process import (
Process,
check_goal_f,
check_max_fes,
check_max_time_millis,
)
from moptipy.api.space import Space, check_space
from moptipy.utils.nputils import rand_seed_check
def _check_log_file(log_file: Any, none_is_ok: bool = True) -> Path | None:
"""
Check a log file.
:param log_file: the log file
:param none_is_ok: is `None` ok for log files?
:return: the log file
>>> print(_check_log_file("/a/b.txt"))
/a/b.txt
>>> print(_check_log_file("/a/b.txt", False))
/a/b.txt
>>> print(_check_log_file("/a/b.txt", True))
/a/b.txt
>>> from pycommons.io.path import Path as Pth
>>> print(_check_log_file(Path("/a/b.txt"), False))
/a/b.txt
>>> print(_check_log_file(None))
None
>>> print(_check_log_file(None, True))
None
>>> try:
... _check_log_file(1) # noqa # type: ignore
... except TypeError as te:
... print(te)
descriptor '__len__' requires a 'str' object but received a 'int'
>>> try:
... _check_log_file(None, False)
... except TypeError as te:
... print(te)
descriptor '__len__' requires a 'str' object but received a 'NoneType'
"""
if (log_file is None) and none_is_ok:
return None
return Path(log_file)
#: The execution type variable for returning `Self`.
TExecution = TypeVar("TExecution", bound="Execution")
[docs]
class Execution:
"""
Define all the components of an experiment and then execute it.
This class follows the builder pattern. It allows us to
step-by-step store all the parameters needed to execute an
experiment. Via the method :meth:`~Execution.execute`, we can then
run the experiment and obtain the instance of
:class:`~moptipy.api.process.Process` *after* the execution of the
algorithm. From this instance, we can query the final result of the
algorithm application.
"""
def __init__(self) -> None:
"""Initialize the execution builder."""
super().__init__()
self._algorithm: Algorithm | None = None
self._solution_space: Space | None = None
self._objective: Objective | None = None
self._search_space: Space | None = None
self._encoding: Encoding | None = None
self._rand_seed: int | None = None
self._max_fes: int | None = None
self._max_time_millis: int | None = None
self._goal_f: None | int | float = None
self._log_file: Path | None = None
self._log_improvements: bool = False
self._log_all_fes: bool = False
[docs]
def set_algorithm(self: TExecution, algorithm: Algorithm) -> TExecution:
"""
Set the algorithm to be used for this experiment.
:param algorithm: the algorithm
:returns: this execution
"""
self._algorithm = check_algorithm(algorithm)
return self
[docs]
def set_solution_space(self: TExecution, solution_space: Space) \
-> TExecution:
"""
Set the solution space to be used for this experiment.
This is the space managing the data structure holding the candidate
solutions.
:param solution_space: the solution space
:returns: this execution
"""
self._solution_space = check_space(solution_space)
return self
[docs]
def set_objective(self: TExecution, objective: Objective) -> TExecution:
"""
Set the objective function to be used for this experiment.
This is the function rating the quality of candidate solutions.
:param objective: the objective function
:returns: this execution
"""
if self._objective is not None:
raise ValueError(
"Cannot add more than one objective function in single-"
f"objective optimization, attempted to add {objective} "
f"after {self._objective}.")
self._objective = check_objective(objective)
return self
[docs]
def set_search_space(self: TExecution, search_space: Space | None) \
-> TExecution:
"""
Set the search space to be used for this experiment.
This is the space from which the algorithm samples points.
:param search_space: the search space, or `None` of none shall be
used, i.e., if search and solution space are the same
:returns: this execution
"""
self._search_space = check_space(search_space, none_is_ok=True)
return self
[docs]
def set_encoding(self: TExecution, encoding: Encoding | None) \
-> TExecution:
"""
Set the encoding to be used for this experiment.
This is the function translating from the search space to the
solution space.
:param encoding: the encoding, or `None` of none shall be used
:returns: this execution
"""
self._encoding = check_encoding(encoding, none_is_ok=True)
return self
[docs]
def set_rand_seed(self: TExecution, rand_seed: int | None) -> TExecution:
"""
Set the seed to be used for initializing the random number generator.
:param rand_seed: the random seed, or `None` if a seed should
automatically be chosen when the experiment is executed
"""
self._rand_seed = None if rand_seed is None \
else rand_seed_check(rand_seed)
return self
[docs]
def set_max_fes(self: TExecution, max_fes: int, # +book
force_override: bool = False) -> TExecution:
"""
Set the maximum FEs.
This is the number of candidate solutions an optimization is allowed
to evaluate. If this method is called multiple times, then the
shortest limit is used unless `force_override` is `True`.
:param max_fes: the maximum FEs
:param force_override: the use the value given in `max_time_millis`
regardless of what was specified before
:returns: this execution
"""
max_fes = check_max_fes(max_fes)
if (self._max_fes is not None) and (max_fes >= self._max_fes) \
and (not force_override):
return self
self._max_fes = max_fes
return self
[docs]
def set_max_time_millis(self: TExecution, max_time_millis: int,
force_override: bool = False) -> TExecution:
"""
Set the maximum time in milliseconds.
This is the maximum time that the process is allowed to run. If this
method is called multiple times, the shortest time is used unless
`force_override` is `True`.
:param max_time_millis: the maximum time in milliseconds
:param force_override: the use the value given in `max_time_millis`
regardless of what was specified before
:returns: this execution
"""
max_time_millis = check_max_time_millis(max_time_millis)
if (self._max_time_millis is not None) \
and (max_time_millis >= self._max_time_millis) \
and (not force_override):
return self
self._max_time_millis = max_time_millis
return self
[docs]
def set_goal_f(self: TExecution, goal_f: int | float) -> TExecution:
"""
Set the goal objective value after which the process can stop.
If this method is called multiple times, then the largest value is
retained.
:param goal_f: the goal objective value.
:returns: this execution
"""
goal_f = check_goal_f(goal_f)
if (self._goal_f is not None) and (goal_f <= self._goal_f):
return self
self._goal_f = goal_f
return self
[docs]
def set_log_file(self: TExecution, log_file: str | None) -> TExecution:
"""
Set the log file to write to.
If a path to a log file is provided, the contents of this file will be
filled based on the structure documented at
https://thomasweise.github.io/moptipy/#log-file-sections, which
includes the algorithm parameters, the instance features, the
system settings, the final solution, the corresponding point in the
search space, etc.
This method can be called arbitrarily often.
:param log_file: the log file
"""
self._log_file = _check_log_file(log_file, True)
return self
[docs]
def set_log_improvements(self: TExecution,
log_improvements: bool = True) -> TExecution:
"""
Set whether improvements should be logged.
If improvements are logged, then the `PROGRESS` section will be added
to the log files, as documented at
https://thomasweise.github.io/moptipy/#the-section-progress.
:param log_improvements: if improvements should be logged?
:returns: this execution
"""
if not isinstance(log_improvements, bool):
raise type_error(log_improvements, "log_improvements", bool)
self._log_improvements = log_improvements
return self
[docs]
def set_log_all_fes(self: TExecution,
log_all_fes: bool = True) -> TExecution:
"""
Set whether all objective function evaluations (FEs) should be logged.
If all FEs are logged, then the `PROGRESS` section will be added to
the log files, as documented at
https://thomasweise.github.io/moptipy/#the-section-progress.
:param log_all_fes: if all FEs should be logged?
:returns: this execution
"""
if not isinstance(log_all_fes, bool):
raise type_error(log_all_fes, "log_all_fes", bool)
self._log_all_fes = log_all_fes
return self
[docs]
def execute(self) -> Process:
"""
Execute the experiment and return the process *after* the run.
The optimization process constructed with this object is executed.
This means that first, an instance of
:class:`~moptipy.api.process.Process` is constructed.
Then, the method :meth:`~moptipy.api.algorithm.Algorithm.solve` is
applied to this instance.
In other words, the optimization algorithm is executed until it
terminates.
Finally, this method returns the :class:`~moptipy.api.process.Process`
instance *after* algorithm completion.
This instance then can be queried for the final result of the run (via
:meth:`~moptipy.api.process.Process.get_copy_of_best_y`), the
objective value of this final best solution (via
:meth:`~moptipy.api.process.Process.get_best_f`), and other
information.
If a log file path was supplied to :meth:`~set_log_file`, then the
information gathered from the optimization process will be written to
the file *after* the `with` blog using the process is left. See
https://thomasweise.github.io/moptipy/#log-file-sections for a
documentation of the log file structure and sections.
:return: the process *after* the run, i.e., in the state where it can
be queried for the result
"""
algorithm: Final[Algorithm] = check_algorithm(self._algorithm)
solution_space: Final[Space] = check_space(self._solution_space)
objective: Final[Objective] = check_objective(self._objective)
search_space: Final[Space | None] = check_space(
self._search_space, self._encoding is None)
encoding: Final[Encoding | None] = check_encoding(
self._encoding, search_space is None)
rand_seed = self._rand_seed
if rand_seed is not None:
rand_seed = rand_seed_check(rand_seed)
max_time_millis = check_max_time_millis(self._max_time_millis, True)
max_fes = check_max_fes(self._max_fes, True)
goal_f = check_goal_f(self._goal_f, True)
f_lb = objective.lower_bound()
if (f_lb is not None) and isfinite(f_lb) \
and ((goal_f is None) or (f_lb > goal_f)):
goal_f = f_lb
log_all_fes = self._log_all_fes
log_improvements = self._log_improvements or self._log_all_fes
log_file = self._log_file
if log_file is None:
if log_all_fes:
raise ValueError("Log file cannot be None "
"if all FEs should be logged.")
if log_improvements:
raise ValueError("Log file cannot be None "
"if improvements should be logged.")
else:
log_file.create_file_or_truncate()
process: Final[_ProcessBase] = \
(_ProcessNoSSLog(solution_space=solution_space,
objective=objective,
algorithm=algorithm,
log_file=log_file,
rand_seed=rand_seed,
max_fes=max_fes,
max_time_millis=max_time_millis,
goal_f=goal_f,
log_all_fes=log_all_fes)
if log_improvements or log_all_fes else
_ProcessNoSS(solution_space=solution_space,
objective=objective,
algorithm=algorithm,
log_file=log_file,
rand_seed=rand_seed,
max_fes=max_fes,
max_time_millis=max_time_millis,
goal_f=goal_f)) if search_space is None else \
(_ProcessSSLog(solution_space=solution_space,
objective=objective,
algorithm=algorithm,
search_space=search_space,
encoding=encoding,
log_file=log_file,
rand_seed=rand_seed,
max_fes=max_fes,
max_time_millis=max_time_millis,
goal_f=goal_f,
log_all_fes=log_all_fes)
if log_improvements or log_all_fes else
_ProcessSS(solution_space=solution_space,
objective=objective,
algorithm=algorithm,
search_space=search_space,
encoding=encoding,
log_file=log_file,
rand_seed=rand_seed,
max_fes=max_fes,
max_time_millis=max_time_millis,
goal_f=goal_f))
try:
# noinspection PyProtectedMember
process._after_init() # finalize the created process
objective.initialize() # initialize the objective function
if encoding is not None:
encoding.initialize() # initialize the encoding
solution_space.initialize() # initialize the solution space
if search_space is not None:
search_space.initialize() # initialize the search space
algorithm.initialize() # initialize the algorithm
algorithm.solve(process) # apply the algorithm
except Exception as be:
# noinspection PyProtectedMember
if process._caught is None:
# noinspection PyProtectedMember
process._caught = be
return process