Coverage for moptipy / api / execution.py: 94%
141 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 10:36 +0000
1"""The algorithm execution API."""
2from math import isfinite
3from typing import Any, Final, Self
5from pycommons.io.path import Path
6from pycommons.types import type_error
8from moptipy.api._process_base import _ProcessBase
9from moptipy.api._process_no_ss import _ProcessNoSS
10from moptipy.api._process_no_ss_log import _ProcessNoSSLog
11from moptipy.api._process_ss import _ProcessSS
12from moptipy.api._process_ss_log import _ProcessSSLog
13from moptipy.api.algorithm import Algorithm, check_algorithm
14from moptipy.api.encoding import Encoding, check_encoding
15from moptipy.api.improvement_logger import (
16 ImprovementLogger,
17 ImprovementLoggerFactory,
18)
19from moptipy.api.objective import Objective, check_objective
20from moptipy.api.process import (
21 Process,
22 check_goal_f,
23 check_max_fes,
24 check_max_time_millis,
25)
26from moptipy.api.space import Space, check_space
27from moptipy.utils.nputils import rand_seed_check
28from moptipy.utils.strings import sanitize_names
31def _check_log_file(log_file: Any, none_is_ok: bool = True) -> Path | None:
32 """
33 Check a log file.
35 :param log_file: the log file
36 :param none_is_ok: is `None` ok for log files?
37 :return: the log file
39 >>> print(_check_log_file("/a/b.txt"))
40 /a/b.txt
41 >>> print(_check_log_file("/a/b.txt", False))
42 /a/b.txt
43 >>> print(_check_log_file("/a/b.txt", True))
44 /a/b.txt
45 >>> from pycommons.io.path import Path as Pth
46 >>> print(_check_log_file(Path("/a/b.txt"), False))
47 /a/b.txt
48 >>> print(_check_log_file(None))
49 None
50 >>> print(_check_log_file(None, True))
51 None
53 >>> try:
54 ... _check_log_file(1) # noqa # type: ignore
55 ... except TypeError as te:
56 ... print(te)
57 descriptor '__len__' requires a 'str' object but received a 'int'
59 >>> try:
60 ... _check_log_file(None, False)
61 ... except TypeError as te:
62 ... print(te)
63 descriptor '__len__' requires a 'str' object but received a 'NoneType'
64 """
65 if (log_file is None) and none_is_ok:
66 return None
67 return Path(log_file)
70class Execution:
71 """
72 Define all the components of an experiment and then execute it.
74 This class follows the builder pattern. It allows us to
75 step-by-step store all the parameters needed to execute an
76 experiment. Via the method :meth:`~Execution.execute`, we can then
77 run the experiment and obtain the instance of
78 :class:`~moptipy.api.process.Process` *after* the execution of the
79 algorithm. From this instance, we can query the final result of the
80 algorithm application.
81 """
83 def __init__(self) -> None:
84 """Initialize the execution builder."""
85 super().__init__()
86 self._algorithm: Algorithm | None = None
87 self._solution_space: Space | None = None
88 self._objective: Objective | None = None
89 self._search_space: Space | None = None
90 self._encoding: Encoding | None = None
91 self._rand_seed: int | None = None
92 self._max_fes: int | None = None
93 self._max_time_millis: int | None = None
94 self._goal_f: int | float | None = None
95 self._log_file: Path | None = None
96 self._log_improvements: bool = False
97 self._log_all_fes: bool = False
98 self._improvement_logger: (
99 ImprovementLogger | ImprovementLoggerFactory | None) = None
101 def set_algorithm(self, algorithm: Algorithm) -> Self:
102 """
103 Set the algorithm to be used for this experiment.
105 :param algorithm: the algorithm
106 :returns: this execution
107 """
108 self._algorithm = check_algorithm(algorithm)
109 return self
111 def set_solution_space(self, solution_space: Space) \
112 -> Self:
113 """
114 Set the solution space to be used for this experiment.
116 This is the space managing the data structure holding the candidate
117 solutions.
119 :param solution_space: the solution space
120 :returns: this execution
121 """
122 self._solution_space = check_space(solution_space)
123 return self
125 def set_objective(self, objective: Objective) -> Self:
126 """
127 Set the objective function to be used for this experiment.
129 This is the function rating the quality of candidate solutions.
131 :param objective: the objective function
132 :returns: this execution
133 """
134 if self._objective is not None:
135 raise ValueError(
136 "Cannot add more than one objective function in single-"
137 f"objective optimization, attempted to add {objective} "
138 f"after {self._objective}.")
139 self._objective = check_objective(objective)
140 return self
142 def set_search_space(self, search_space: Space | None) \
143 -> Self:
144 """
145 Set the search space to be used for this experiment.
147 This is the space from which the algorithm samples points.
149 :param search_space: the search space, or `None` of none shall be
150 used, i.e., if search and solution space are the same
151 :returns: this execution
152 """
153 self._search_space = check_space(search_space, none_is_ok=True)
154 return self
156 def set_encoding(self, encoding: Encoding | None) \
157 -> Self:
158 """
159 Set the encoding to be used for this experiment.
161 This is the function translating from the search space to the
162 solution space.
164 :param encoding: the encoding, or `None` of none shall be used
165 :returns: this execution
166 """
167 self._encoding = check_encoding(encoding, none_is_ok=True)
168 return self
170 def set_rand_seed(self, rand_seed: int | None) -> Self:
171 """
172 Set the seed to be used for initializing the random number generator.
174 :param rand_seed: the random seed, or `None` if a seed should
175 automatically be chosen when the experiment is executed
176 """
177 self._rand_seed = None if rand_seed is None \
178 else rand_seed_check(rand_seed)
179 return self
181 def set_max_fes(self, max_fes: int, # +book
182 force_override: bool = False) -> Self:
183 """
184 Set the maximum FEs.
186 This is the number of candidate solutions an optimization is allowed
187 to evaluate. If this method is called multiple times, then the
188 shortest limit is used unless `force_override` is `True`.
190 :param max_fes: the maximum FEs
191 :param force_override: the use the value given in `max_time_millis`
192 regardless of what was specified before
193 :returns: this execution
194 """
195 max_fes = check_max_fes(max_fes)
196 if (self._max_fes is not None) and (max_fes >= self._max_fes) \
197 and (not force_override):
198 return self
199 self._max_fes = max_fes
200 return self
202 def set_max_time_millis(self, max_time_millis: int,
203 force_override: bool = False) -> Self:
204 """
205 Set the maximum time in milliseconds.
207 This is the maximum time that the process is allowed to run. If this
208 method is called multiple times, the shortest time is used unless
209 `force_override` is `True`.
211 :param max_time_millis: the maximum time in milliseconds
212 :param force_override: the use the value given in `max_time_millis`
213 regardless of what was specified before
214 :returns: this execution
215 """
216 max_time_millis = check_max_time_millis(max_time_millis)
217 if (self._max_time_millis is not None) \
218 and (max_time_millis >= self._max_time_millis) \
219 and (not force_override):
220 return self
221 self._max_time_millis = max_time_millis
222 return self
224 def set_goal_f(self, goal_f: int | float) -> Self:
225 """
226 Set the goal objective value after which the process can stop.
228 If this method is called multiple times, then the largest value is
229 retained.
231 :param goal_f: the goal objective value.
232 :returns: this execution
233 """
234 goal_f = check_goal_f(goal_f)
235 if (self._goal_f is not None) and (goal_f <= self._goal_f):
236 return self
237 self._goal_f = goal_f
238 return self
240 def set_log_file(self, log_file: str | None) -> Self:
241 """
242 Set the log file to write to.
244 If a path to a log file is provided, the contents of this file will be
245 filled based on the structure documented at
246 https://thomasweise.github.io/moptipy/#log-file-sections, which
247 includes the algorithm parameters, the instance features, the
248 system settings, the final solution, the corresponding point in the
249 search space, etc.
251 This method can be called arbitrarily often.
253 :param log_file: the log file
254 """
255 self._log_file = _check_log_file(log_file, True)
256 return self
258 def set_log_improvements(self,
259 log_improvements: bool = True) -> Self:
260 """
261 Set whether improvements should be logged.
263 If improvements are logged, then the `PROGRESS` section will be added
264 to the log files, as documented at
265 https://thomasweise.github.io/moptipy/#the-section-progress.
267 :param log_improvements: if improvements should be logged?
268 :returns: this execution
269 """
270 if not isinstance(log_improvements, bool):
271 raise type_error(log_improvements, "log_improvements", bool)
272 self._log_improvements = log_improvements
273 return self
275 def set_log_all_fes(self,
276 log_all_fes: bool = True) -> Self:
277 """
278 Set whether all objective function evaluations (FEs) should be logged.
280 If all FEs are logged, then the `PROGRESS` section will be added to
281 the log files, as documented at
282 https://thomasweise.github.io/moptipy/#the-section-progress.
284 :param log_all_fes: if all FEs should be logged?
285 :returns: this execution
286 """
287 if not isinstance(log_all_fes, bool):
288 raise type_error(log_all_fes, "log_all_fes", bool)
289 self._log_all_fes = log_all_fes
290 return self
292 def set_improvement_logger(
293 self, improvement_logger: ImprovementLogger
294 | ImprovementLoggerFactory | None) -> Self:
295 """
296 Set the improvement logger.
298 :param improvement_logger: the improvement logger or the
299 improvement logger factory
300 """
301 if improvement_logger is None:
302 self._improvement_logger = None
303 return None
304 if not isinstance(improvement_logger,
305 ImprovementLogger | ImprovementLoggerFactory):
306 raise type_error(improvement_logger, "improvement_logger",
307 (ImprovementLogger, ImprovementLoggerFactory))
308 self._improvement_logger = improvement_logger
309 return self
311 def _logger(self, seed: int, log_file: str | None = None) \
312 -> ImprovementLogger | None:
313 """
314 Get the improvement logger to use.
316 :param seed: the random see
317 :param log_file: the log file, if any
318 :returns: the improvement logger or `None`
319 """
320 logger: ImprovementLogger | ImprovementLoggerFactory | None = (
321 self._improvement_logger)
322 if (logger is None) or isinstance(logger, ImprovementLogger):
323 return logger
325 return logger.create(log_file, sanitize_names((
326 str(self._algorithm), f"{seed:x}")))
328 def execute(self) -> Process:
329 """
330 Execute the experiment and return the process *after* the run.
332 The optimization process constructed with this object is executed.
333 This means that first, an instance of
334 :class:`~moptipy.api.process.Process` is constructed.
335 Then, the method :meth:`~moptipy.api.algorithm.Algorithm.solve` is
336 applied to this instance.
337 In other words, the optimization algorithm is executed until it
338 terminates.
339 Finally, this method returns the :class:`~moptipy.api.process.Process`
340 instance *after* algorithm completion.
341 This instance then can be queried for the final result of the run (via
342 :meth:`~moptipy.api.process.Process.get_copy_of_best_y`), the
343 objective value of this final best solution (via
344 :meth:`~moptipy.api.process.Process.get_best_f`), and other
345 information.
347 If a log file path was supplied to :meth:`~set_log_file`, then the
348 information gathered from the optimization process will be written to
349 the file *after* the `with` blog using the process is left. See
350 https://thomasweise.github.io/moptipy/#log-file-sections for a
351 documentation of the log file structure and sections.
353 :return: the process *after* the run, i.e., in the state where it can
354 be queried for the result
355 """
356 algorithm: Final[Algorithm] = check_algorithm(self._algorithm)
357 solution_space: Final[Space] = check_space(self._solution_space)
358 objective: Final[Objective] = check_objective(self._objective)
359 search_space: Final[Space | None] = check_space(
360 self._search_space, self._encoding is None)
361 encoding: Final[Encoding | None] = check_encoding(
362 self._encoding, search_space is None)
363 rand_seed = self._rand_seed
364 if rand_seed is not None:
365 rand_seed = rand_seed_check(rand_seed)
366 max_time_millis = check_max_time_millis(self._max_time_millis, True)
367 max_fes = check_max_fes(self._max_fes, True)
368 goal_f = check_goal_f(self._goal_f, True)
369 f_lb = objective.lower_bound()
370 if (f_lb is not None) and isfinite(f_lb) \
371 and ((goal_f is None) or (f_lb > goal_f)):
372 goal_f = f_lb
374 log_all_fes = self._log_all_fes
375 log_improvements = self._log_improvements or self._log_all_fes
377 log_file = self._log_file
378 if log_file is None:
379 if log_all_fes:
380 raise ValueError("Log file cannot be None "
381 "if all FEs should be logged.")
382 if log_improvements:
383 raise ValueError("Log file cannot be None "
384 "if improvements should be logged.")
385 else:
386 log_file.create_file_or_truncate()
388 logger: Final[ImprovementLogger | None] = self._logger(
389 rand_seed, log_file)
391 process: Final[_ProcessBase] = \
392 (_ProcessNoSSLog(solution_space=solution_space,
393 objective=objective,
394 algorithm=algorithm,
395 log_file=log_file,
396 rand_seed=rand_seed,
397 max_fes=max_fes,
398 max_time_millis=max_time_millis,
399 goal_f=goal_f,
400 log_all_fes=log_all_fes,
401 improvement_logger=logger)
402 if log_improvements or log_all_fes else
403 _ProcessNoSS(solution_space=solution_space,
404 objective=objective,
405 algorithm=algorithm,
406 log_file=log_file,
407 rand_seed=rand_seed,
408 max_fes=max_fes,
409 max_time_millis=max_time_millis,
410 goal_f=goal_f,
411 improvement_logger=logger)) \
412 if search_space is None else \
413 (_ProcessSSLog(
414 solution_space=solution_space,
415 objective=objective,
416 algorithm=algorithm,
417 search_space=search_space,
418 encoding=encoding,
419 log_file=log_file,
420 rand_seed=rand_seed,
421 max_fes=max_fes,
422 max_time_millis=max_time_millis,
423 goal_f=goal_f,
424 log_all_fes=log_all_fes,
425 improvement_logger=logger)
426 if log_improvements or log_all_fes else
427 _ProcessSS(
428 solution_space=solution_space,
429 objective=objective,
430 algorithm=algorithm,
431 search_space=search_space,
432 encoding=encoding,
433 log_file=log_file,
434 rand_seed=rand_seed,
435 max_fes=max_fes,
436 max_time_millis=max_time_millis,
437 goal_f=goal_f,
438 improvement_logger=logger))
439 try:
440 # noinspection PyProtectedMember
441 process._after_init() # finalize the created process
442 objective.initialize() # initialize the objective function
443 if encoding is not None:
444 encoding.initialize() # initialize the encoding
445 solution_space.initialize() # initialize the solution space
446 if search_space is not None:
447 search_space.initialize() # initialize the search space
448 algorithm.initialize() # initialize the algorithm
449 algorithm.solve(process) # apply the algorithm
450 except Exception as be: # noqa
451 # noinspection PyProtectedMember
452 if process._caught is None:
453 # noinspection PyProtectedMember
454 process._caught = be
455 return process