Coverage for moptipy / api / execution.py: 94%
124 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +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.objective import Objective, check_objective
16from moptipy.api.process import (
17 Process,
18 check_goal_f,
19 check_max_fes,
20 check_max_time_millis,
21)
22from moptipy.api.space import Space, check_space
23from moptipy.utils.nputils import rand_seed_check
26def _check_log_file(log_file: Any, none_is_ok: bool = True) -> Path | None:
27 """
28 Check a log file.
30 :param log_file: the log file
31 :param none_is_ok: is `None` ok for log files?
32 :return: the log file
34 >>> print(_check_log_file("/a/b.txt"))
35 /a/b.txt
36 >>> print(_check_log_file("/a/b.txt", False))
37 /a/b.txt
38 >>> print(_check_log_file("/a/b.txt", True))
39 /a/b.txt
40 >>> from pycommons.io.path import Path as Pth
41 >>> print(_check_log_file(Path("/a/b.txt"), False))
42 /a/b.txt
43 >>> print(_check_log_file(None))
44 None
45 >>> print(_check_log_file(None, True))
46 None
48 >>> try:
49 ... _check_log_file(1) # noqa # type: ignore
50 ... except TypeError as te:
51 ... print(te)
52 descriptor '__len__' requires a 'str' object but received a 'int'
54 >>> try:
55 ... _check_log_file(None, False)
56 ... except TypeError as te:
57 ... print(te)
58 descriptor '__len__' requires a 'str' object but received a 'NoneType'
59 """
60 if (log_file is None) and none_is_ok:
61 return None
62 return Path(log_file)
65class Execution:
66 """
67 Define all the components of an experiment and then execute it.
69 This class follows the builder pattern. It allows us to
70 step-by-step store all the parameters needed to execute an
71 experiment. Via the method :meth:`~Execution.execute`, we can then
72 run the experiment and obtain the instance of
73 :class:`~moptipy.api.process.Process` *after* the execution of the
74 algorithm. From this instance, we can query the final result of the
75 algorithm application.
76 """
78 def __init__(self) -> None:
79 """Initialize the execution builder."""
80 super().__init__()
81 self._algorithm: Algorithm | None = None
82 self._solution_space: Space | None = None
83 self._objective: Objective | None = None
84 self._search_space: Space | None = None
85 self._encoding: Encoding | None = None
86 self._rand_seed: int | None = None
87 self._max_fes: int | None = None
88 self._max_time_millis: int | None = None
89 self._goal_f: int | float | None = None
90 self._log_file: Path | None = None
91 self._log_improvements: bool = False
92 self._log_all_fes: bool = False
94 def set_algorithm(self, algorithm: Algorithm) -> Self:
95 """
96 Set the algorithm to be used for this experiment.
98 :param algorithm: the algorithm
99 :returns: this execution
100 """
101 self._algorithm = check_algorithm(algorithm)
102 return self
104 def set_solution_space(self, solution_space: Space) \
105 -> Self:
106 """
107 Set the solution space to be used for this experiment.
109 This is the space managing the data structure holding the candidate
110 solutions.
112 :param solution_space: the solution space
113 :returns: this execution
114 """
115 self._solution_space = check_space(solution_space)
116 return self
118 def set_objective(self, objective: Objective) -> Self:
119 """
120 Set the objective function to be used for this experiment.
122 This is the function rating the quality of candidate solutions.
124 :param objective: the objective function
125 :returns: this execution
126 """
127 if self._objective is not None:
128 raise ValueError(
129 "Cannot add more than one objective function in single-"
130 f"objective optimization, attempted to add {objective} "
131 f"after {self._objective}.")
132 self._objective = check_objective(objective)
133 return self
135 def set_search_space(self, search_space: Space | None) \
136 -> Self:
137 """
138 Set the search space to be used for this experiment.
140 This is the space from which the algorithm samples points.
142 :param search_space: the search space, or `None` of none shall be
143 used, i.e., if search and solution space are the same
144 :returns: this execution
145 """
146 self._search_space = check_space(search_space, none_is_ok=True)
147 return self
149 def set_encoding(self, encoding: Encoding | None) \
150 -> Self:
151 """
152 Set the encoding to be used for this experiment.
154 This is the function translating from the search space to the
155 solution space.
157 :param encoding: the encoding, or `None` of none shall be used
158 :returns: this execution
159 """
160 self._encoding = check_encoding(encoding, none_is_ok=True)
161 return self
163 def set_rand_seed(self, rand_seed: int | None) -> Self:
164 """
165 Set the seed to be used for initializing the random number generator.
167 :param rand_seed: the random seed, or `None` if a seed should
168 automatically be chosen when the experiment is executed
169 """
170 self._rand_seed = None if rand_seed is None \
171 else rand_seed_check(rand_seed)
172 return self
174 def set_max_fes(self, max_fes: int, # +book
175 force_override: bool = False) -> Self:
176 """
177 Set the maximum FEs.
179 This is the number of candidate solutions an optimization is allowed
180 to evaluate. If this method is called multiple times, then the
181 shortest limit is used unless `force_override` is `True`.
183 :param max_fes: the maximum FEs
184 :param force_override: the use the value given in `max_time_millis`
185 regardless of what was specified before
186 :returns: this execution
187 """
188 max_fes = check_max_fes(max_fes)
189 if (self._max_fes is not None) and (max_fes >= self._max_fes) \
190 and (not force_override):
191 return self
192 self._max_fes = max_fes
193 return self
195 def set_max_time_millis(self, max_time_millis: int,
196 force_override: bool = False) -> Self:
197 """
198 Set the maximum time in milliseconds.
200 This is the maximum time that the process is allowed to run. If this
201 method is called multiple times, the shortest time is used unless
202 `force_override` is `True`.
204 :param max_time_millis: the maximum time in milliseconds
205 :param force_override: the use the value given in `max_time_millis`
206 regardless of what was specified before
207 :returns: this execution
208 """
209 max_time_millis = check_max_time_millis(max_time_millis)
210 if (self._max_time_millis is not None) \
211 and (max_time_millis >= self._max_time_millis) \
212 and (not force_override):
213 return self
214 self._max_time_millis = max_time_millis
215 return self
217 def set_goal_f(self, goal_f: int | float) -> Self:
218 """
219 Set the goal objective value after which the process can stop.
221 If this method is called multiple times, then the largest value is
222 retained.
224 :param goal_f: the goal objective value.
225 :returns: this execution
226 """
227 goal_f = check_goal_f(goal_f)
228 if (self._goal_f is not None) and (goal_f <= self._goal_f):
229 return self
230 self._goal_f = goal_f
231 return self
233 def set_log_file(self, log_file: str | None) -> Self:
234 """
235 Set the log file to write to.
237 If a path to a log file is provided, the contents of this file will be
238 filled based on the structure documented at
239 https://thomasweise.github.io/moptipy/#log-file-sections, which
240 includes the algorithm parameters, the instance features, the
241 system settings, the final solution, the corresponding point in the
242 search space, etc.
244 This method can be called arbitrarily often.
246 :param log_file: the log file
247 """
248 self._log_file = _check_log_file(log_file, True)
249 return self
251 def set_log_improvements(self,
252 log_improvements: bool = True) -> Self:
253 """
254 Set whether improvements should be logged.
256 If improvements are logged, then the `PROGRESS` section will be added
257 to the log files, as documented at
258 https://thomasweise.github.io/moptipy/#the-section-progress.
260 :param log_improvements: if improvements should be logged?
261 :returns: this execution
262 """
263 if not isinstance(log_improvements, bool):
264 raise type_error(log_improvements, "log_improvements", bool)
265 self._log_improvements = log_improvements
266 return self
268 def set_log_all_fes(self,
269 log_all_fes: bool = True) -> Self:
270 """
271 Set whether all objective function evaluations (FEs) should be logged.
273 If all FEs are logged, then the `PROGRESS` section will be added to
274 the log files, as documented at
275 https://thomasweise.github.io/moptipy/#the-section-progress.
277 :param log_all_fes: if all FEs should be logged?
278 :returns: this execution
279 """
280 if not isinstance(log_all_fes, bool):
281 raise type_error(log_all_fes, "log_all_fes", bool)
282 self._log_all_fes = log_all_fes
283 return self
285 def execute(self) -> Process:
286 """
287 Execute the experiment and return the process *after* the run.
289 The optimization process constructed with this object is executed.
290 This means that first, an instance of
291 :class:`~moptipy.api.process.Process` is constructed.
292 Then, the method :meth:`~moptipy.api.algorithm.Algorithm.solve` is
293 applied to this instance.
294 In other words, the optimization algorithm is executed until it
295 terminates.
296 Finally, this method returns the :class:`~moptipy.api.process.Process`
297 instance *after* algorithm completion.
298 This instance then can be queried for the final result of the run (via
299 :meth:`~moptipy.api.process.Process.get_copy_of_best_y`), the
300 objective value of this final best solution (via
301 :meth:`~moptipy.api.process.Process.get_best_f`), and other
302 information.
304 If a log file path was supplied to :meth:`~set_log_file`, then the
305 information gathered from the optimization process will be written to
306 the file *after* the `with` blog using the process is left. See
307 https://thomasweise.github.io/moptipy/#log-file-sections for a
308 documentation of the log file structure and sections.
310 :return: the process *after* the run, i.e., in the state where it can
311 be queried for the result
312 """
313 algorithm: Final[Algorithm] = check_algorithm(self._algorithm)
314 solution_space: Final[Space] = check_space(self._solution_space)
315 objective: Final[Objective] = check_objective(self._objective)
316 search_space: Final[Space | None] = check_space(
317 self._search_space, self._encoding is None)
318 encoding: Final[Encoding | None] = check_encoding(
319 self._encoding, search_space is None)
320 rand_seed = self._rand_seed
321 if rand_seed is not None:
322 rand_seed = rand_seed_check(rand_seed)
323 max_time_millis = check_max_time_millis(self._max_time_millis, True)
324 max_fes = check_max_fes(self._max_fes, True)
325 goal_f = check_goal_f(self._goal_f, True)
326 f_lb = objective.lower_bound()
327 if (f_lb is not None) and isfinite(f_lb) \
328 and ((goal_f is None) or (f_lb > goal_f)):
329 goal_f = f_lb
331 log_all_fes = self._log_all_fes
332 log_improvements = self._log_improvements or self._log_all_fes
334 log_file = self._log_file
335 if log_file is None:
336 if log_all_fes:
337 raise ValueError("Log file cannot be None "
338 "if all FEs should be logged.")
339 if log_improvements:
340 raise ValueError("Log file cannot be None "
341 "if improvements should be logged.")
342 else:
343 log_file.create_file_or_truncate()
345 process: Final[_ProcessBase] = \
346 (_ProcessNoSSLog(solution_space=solution_space,
347 objective=objective,
348 algorithm=algorithm,
349 log_file=log_file,
350 rand_seed=rand_seed,
351 max_fes=max_fes,
352 max_time_millis=max_time_millis,
353 goal_f=goal_f,
354 log_all_fes=log_all_fes)
355 if log_improvements or log_all_fes else
356 _ProcessNoSS(solution_space=solution_space,
357 objective=objective,
358 algorithm=algorithm,
359 log_file=log_file,
360 rand_seed=rand_seed,
361 max_fes=max_fes,
362 max_time_millis=max_time_millis,
363 goal_f=goal_f)) if search_space is None else \
364 (_ProcessSSLog(solution_space=solution_space,
365 objective=objective,
366 algorithm=algorithm,
367 search_space=search_space,
368 encoding=encoding,
369 log_file=log_file,
370 rand_seed=rand_seed,
371 max_fes=max_fes,
372 max_time_millis=max_time_millis,
373 goal_f=goal_f,
374 log_all_fes=log_all_fes)
375 if log_improvements or log_all_fes else
376 _ProcessSS(solution_space=solution_space,
377 objective=objective,
378 algorithm=algorithm,
379 search_space=search_space,
380 encoding=encoding,
381 log_file=log_file,
382 rand_seed=rand_seed,
383 max_fes=max_fes,
384 max_time_millis=max_time_millis,
385 goal_f=goal_f))
386 try:
387 # noinspection PyProtectedMember
388 process._after_init() # finalize the created process
389 objective.initialize() # initialize the objective function
390 if encoding is not None:
391 encoding.initialize() # initialize the encoding
392 solution_space.initialize() # initialize the solution space
393 if search_space is not None:
394 search_space.initialize() # initialize the search space
395 algorithm.initialize() # initialize the algorithm
396 algorithm.solve(process) # apply the algorithm
397 except Exception as be: # noqa: BLE001
398 # noinspection PyProtectedMember
399 if process._caught is None:
400 # noinspection PyProtectedMember
401 process._caught = be
402 return process