Coverage for moptipy / api / subprocesses.py: 90%
341 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""
2Different ways to transform and slice processes.
4In this module, we provide some routines that can be used to slice of
5computational budgets of a given process for running algorithms.
6The following functions are included:
81. :func:`for_fes` allows for creating a sub-process that forwards all method
9 calls to the original process but will perform at most a given number of
10 objective function evaluations.
112. :func:`from_starting_point` creates a sub-process that has the current-best
12 solution pre-set to a given point in the search space and its quality. If
13 the best solution is improved upon, the provided point will be overwritten
14 in place.
153. :func:`without_should_terminate` wraps a process in such a way that the
16 termination criterion :meth:`~moptipy.api.process.Process.should_terminate`
17 does not need to be checked anymore. Instead, once the optimization must
18 stop, it will throw an internal exception and catch it again. This makes it
19 possible to pass :meth:`~moptipy.api.process.Process.evaluate` to
20 externally implemented algorithms that do not care about the `moptipy` API.
22The utility function :func:`get_remaining_fes` returns a number representing
23the remaining objective function evaluations of a given
24:class:`~moptipy.api.process.Process`. If that process does not have an
25FE-based termination criterion, it will instead return a very big number.
26"""
27import contextlib
28from typing import Any, Callable, Final, TypeVar
30import numpy as np
31from pycommons.types import type_error
33from moptipy.api.mo_process import MOProcess
34from moptipy.api.process import Process, check_max_fes
36#: the type variable for single- and multi-objective processes.
37T = TypeVar("T", Process, MOProcess)
40class __ForFEs(Process):
41 """A process searching for a fixed amount of FEs."""
43 def __init__(self, owner: Process, max_fes: int):
44 super().__init__()
45 if not isinstance(owner, Process):
46 raise type_error(owner, "owner", Process)
47 #: the owning process
48 self.__owner: Final[Process] = owner
49 self.get_random = owner.get_random # type: ignore
50 a = owner.get_consumed_time_millis # type: ignore
51 self.get_consumed_time_millis = a # type: ignore
52 a = owner.get_max_time_millis # type: ignore
53 self.get_max_time_millis = a # type: ignore
54 a = owner.get_last_improvement_time_millis # type: ignore
55 self.get_last_improvement_time_millis = a # type: ignore
56 self.has_log = owner.has_log # type: ignore
57 self.add_log_section = owner.add_log_section # type: ignore
58 self.lower_bound = owner.lower_bound # type: ignore
59 self.upper_bound = owner.upper_bound # type: ignore
60 self.create = owner.create # type: ignore
61 self.copy = owner.copy # type: ignore
62 self.to_str = owner.to_str # type: ignore
63 self.is_equal = owner.is_equal # type: ignore
64 self.from_str = owner.from_str # type: ignore
65 self.validate = owner.validate # type: ignore
66 self.n_points = owner.n_points # type: ignore
67 self.has_best = owner.has_best # type: ignore
68 self.get_copy_of_best_x = owner.get_copy_of_best_x # type: ignore
69 self.get_best_f = owner.get_best_f # type: ignore
70 self.get_log_basename = owner.get_log_basename # type: ignore
71 #: the maximum FEs
72 self.max_fes: Final[int] = check_max_fes(max_fes)
73 #: the FEs that we still have left
74 self.__fes_left: int = max_fes
75 #: did we terminate?
76 self.__terminated: bool = False
77 #: the fast call to the owner's should_terminate method
78 self.__should_terminate: Final[Callable[[], bool]] \
79 = owner.should_terminate
80 #: the fast call to the owner's evaluate method
81 self.__evaluate: Final[Callable[[Any], int | float]] \
82 = owner.evaluate
83 #: the fast call to the owner's register method
84 self.__register: Final[Callable[[Any, int | float], None]] \
85 = owner.register
86 #: the start fe
87 self.__start_fe: Final[int] = owner.get_consumed_fes()
89 def should_terminate(self) -> bool:
90 return self.__terminated or self.__should_terminate()
92 def terminate(self) -> None:
93 self.__terminated = True
95 def evaluate(self, x) -> float | int:
96 f: Final[int | float] = self.__evaluate(x)
97 fel: Final[int] = self.__fes_left - 1
98 self.__fes_left = fel
99 if fel <= 0:
100 self.__terminated = True
101 return f
103 def register(self, x, f: int | float) -> None:
104 self.__register(x, f)
105 fel: Final[int] = self.__fes_left - 1
106 self.__fes_left = fel
107 if fel <= 0:
108 self.__terminated = True
110 def get_consumed_fes(self) -> int:
111 return self.max_fes - self.__fes_left
113 def get_last_improvement_fe(self) -> int:
114 return max(1 if self.__fes_left < self.max_fes else 0,
115 self.__owner.get_last_improvement_fe() - self.__start_fe)
117 def get_max_fes(self) -> int:
118 return self.max_fes
120 def __str__(self) -> str:
121 return f"forFEs_{self.max_fes}_{self.__owner}"
124class __ForFEsMO(MOProcess):
125 """A process searching for a fixed amount of FEs."""
127 def __init__(self, owner: MOProcess, max_fes: int):
128 super().__init__()
129 if not isinstance(owner, MOProcess):
130 raise type_error(owner, "owner", MOProcess)
131 #: the owning process
132 self.__owner: Final[MOProcess] = owner
133 self.get_random = owner.get_random # type: ignore
134 a = owner.get_consumed_time_millis # type: ignore
135 self.get_consumed_time_millis = a # type: ignore
136 a = owner.get_max_time_millis # type: ignore
137 self.get_max_time_millis = a # type: ignore
138 a = owner.get_last_improvement_time_millis # type: ignore
139 self.get_last_improvement_time_millis = a # type: ignore
140 self.has_log = owner.has_log # type: ignore
141 self.add_log_section = owner.add_log_section # type: ignore
142 self.lower_bound = owner.lower_bound # type: ignore
143 self.upper_bound = owner.upper_bound # type: ignore
144 self.create = owner.create # type: ignore
145 self.copy = owner.copy # type: ignore
146 self.to_str = owner.to_str # type: ignore
147 self.is_equal = owner.is_equal # type: ignore
148 self.from_str = owner.from_str # type: ignore
149 self.validate = owner.validate # type: ignore
150 self.n_points = owner.n_points # type: ignore
151 self.has_best = owner.has_best # type: ignore
152 self.get_copy_of_best_x = owner.get_copy_of_best_x # type: ignore
153 self.get_best_f = owner.get_best_f # type: ignore
154 self.get_archive = owner.get_archive # type: ignore
155 self.check_in = owner.check_in # type: ignore
156 self.f_create = owner.f_create # type: ignore
157 self.f_validate = owner.f_validate # type: ignore
158 self.f_dtype = owner.f_dtype # type: ignore
159 self.f_dominates = owner.f_dominates # type: ignore
160 self.f_dimension = owner.f_dimension # type: ignore
161 self.get_log_basename = owner.get_log_basename # type: ignore
162 #: the maximum FEs
163 self.max_fes: Final[int] = check_max_fes(max_fes)
164 #: the FEs that we still have left
165 self.__fes_left: int = max_fes
166 #: did we terminate?
167 self.__terminated: bool = False
168 #: the fast call to the owner's should_terminate method
169 self.__should_terminate: Final[Callable[[], bool]] \
170 = owner.should_terminate
171 #: the fast call to the owner's evaluate method
172 self.__evaluate: Final[Callable[[Any], int | float]] \
173 = owner.evaluate
174 #: the fast call to the owner's register method
175 self.__register: Final[Callable[[Any, int | float], None]] \
176 = owner.register
177 #: the evaluation wrapper
178 self.__f_evaluate: Final[Callable[
179 [Any, np.ndarray], int | float]] = owner.f_evaluate
180 #: the start fe
181 self.__start_fe: Final[int] = owner.get_consumed_fes()
183 def should_terminate(self) -> bool:
184 return self.__terminated or self.__should_terminate()
186 def terminate(self) -> None:
187 self.__terminated = True
189 def evaluate(self, x) -> float | int:
190 f: Final[int | float] = self.__evaluate(x)
191 fel: Final[int] = self.__fes_left - 1
192 self.__fes_left = fel
193 if fel <= 0:
194 self.__terminated = True
195 return f
197 def register(self, x, f: int | float) -> None:
198 self.__register(x, f)
199 fel: Final[int] = self.__fes_left - 1
200 self.__fes_left = fel
201 if fel <= 0:
202 self.__terminated = True
204 def f_evaluate(self, x, fs: np.ndarray) -> float | int:
205 f: Final[int | float] = self.__f_evaluate(x, fs)
206 fel: Final[int] = self.__fes_left - 1
207 self.__fes_left = fel
208 if fel <= 0:
209 self.__terminated = True
210 return f
212 def get_consumed_fes(self) -> int:
213 return self.max_fes - self.__fes_left
215 def get_last_improvement_fe(self) -> int:
216 return max(1 if self.__fes_left < self.max_fes else 0,
217 self.__owner.get_last_improvement_fe() - self.__start_fe)
219 def get_max_fes(self) -> int:
220 return self.max_fes
222 def __str__(self) -> str:
223 return f"forFEsMO_{self.max_fes}_{self.__owner}"
226def for_fes(process: T, max_fes: int) -> T:
227 """
228 Create a sub-process that can run for the given number of FEs.
230 :param process: the original process
231 :param max_fes: the maximum number of objective function evaluations
232 :returns: the sub-process that will terminate after `max_fes` FEs and that
233 forwards all other calls the `process`.
234 """
235 max_fes = check_max_fes(max_fes, False)
236 return __ForFEsMO(process, max_fes) if isinstance(process, MOProcess) \
237 else __ForFEs(process, max_fes)
240class __FromStartingPoint(Process):
241 """A process searching from a given point."""
243 def __init__(self, owner: Process, in_and_out_x: Any,
244 f: int | float):
245 """
246 Create a sub-process searching from one starting point.
248 :param owner: the owning process
249 :param in_and_out_x: the input solution record, which will be
250 overwritten with the best encountered solution
251 :param f: the objective value corresponding to `in_and_out`
252 """
253 super().__init__()
254 if not isinstance(owner, Process):
255 raise type_error(owner, "owner", Process)
256 #: the owning process
257 self.__owner: Final[Process] = owner
258 self.get_random = owner.get_random # type: ignore
259 a = owner.get_consumed_time_millis # type: ignore
260 self.get_consumed_time_millis = a # type: ignore
261 a = owner.get_max_time_millis # type: ignore
262 self.get_max_time_millis = a # type: ignore
263 a = owner.get_last_improvement_time_millis # type: ignore
264 self.get_last_improvement_time_millis = a # type: ignore
265 self.has_log = owner.has_log # type: ignore
266 self.add_log_section = owner.add_log_section # type: ignore
267 self.lower_bound = owner.lower_bound # type: ignore
268 self.upper_bound = owner.upper_bound # type: ignore
269 self.create = owner.create # type: ignore
270 self.copy = owner.copy # type: ignore
271 self.to_str = owner.to_str # type: ignore
272 self.is_equal = owner.is_equal # type: ignore
273 self.from_str = owner.from_str # type: ignore
274 self.validate = owner.validate # type: ignore
275 self.n_points = owner.n_points # type: ignore
276 self.should_terminate = owner.should_terminate # type: ignore
277 self.terminate = owner.terminate # type: ignore
278 self.get_log_basename = owner.get_log_basename # type: ignore
279 #: the best solution
280 self.__best_x: Final[Any] = in_and_out_x
281 #: the best-so-far solution
282 self.__best_f: int | float = f
283 #: the last improvement fe
284 self.__last_improvement_fe: int = 0
285 #: the consumed FEs
286 self.__fes: int = 0
287 mfes: int | None = owner.get_max_fes()
288 if mfes is not None:
289 mfes -= owner.get_consumed_fes()
290 #: the maximum permitted FEs
291 self.__max_fes: Final[int | None] = mfes
292 #: the fast call to the owner's evaluate method
293 self.__evaluate: Final[Callable[[Any], int | float]] \
294 = owner.evaluate
295 #: the fast call to the owner's register method
296 self.__register: Final[Callable[[Any, int | float], None]] \
297 = owner.register
298 #: True as long as only the seed has been used
299 self.__only_seed_used: bool = True
301 def has_best(self) -> bool:
302 return True
304 def get_copy_of_best_x(self, x) -> None:
305 self.copy(x, self.__best_x)
307 def get_best_f(self) -> int | float:
308 return self.__best_f
310 def evaluate(self, x) -> float | int:
311 if self.__only_seed_used:
312 if self.is_equal(x, self.__best_x):
313 return self.__best_f
314 self.__only_seed_used = False
315 self.__fes = fe = self.__fes + 1
316 f: Final[int | float] = self.__evaluate(x)
317 if f <= self.__best_f:
318 self.copy(self.__best_x, x)
319 if f < self.__best_f:
320 self.__best_f = f
321 self.__last_improvement_fe = fe
322 return f
324 def register(self, x, f: int | float) -> None:
325 if self.__only_seed_used:
326 if self.is_equal(x, self.__best_x):
327 return
328 self.__only_seed_used = False
329 self.__fes = fe = self.__fes + 1
330 self.__register(x, f)
331 if f < self.__best_f:
332 self.copy(self.__best_x, x)
333 if f < self.__best_f:
334 self.__best_f = f
335 self.__last_improvement_fe = fe
337 def get_consumed_fes(self) -> int:
338 return max(1, self.__fes)
340 def get_last_improvement_fe(self) -> int:
341 return self.__last_improvement_fe
343 def get_max_fes(self) -> int | None:
344 return self.__max_fes
346 def __str__(self) -> str:
347 return f"fromStart_{self.__owner}"
350def from_starting_point(owner: Process, in_and_out_x: Any,
351 f: int | float) -> Process:
352 """
353 Create a sub-process searching from one starting point.
355 This process is especially useful in conjunction with class
356 :class:`~moptipy.operators.op0_forward.Op0Forward`. This class
357 allows forwarding the nullary search operator to the function
358 :meth:`~moptipy.api.process.Process.get_copy_of_best_x`. This way, the
359 first point that it sampled by a local search can be the point specified
360 as `in_and_out_x`, which effectively seeds the local search.
362 To dovetail with chance of seeding, no FEs are counted at the beginning of
363 the process as long as all points to be evaluated equal to the
364 `in_and_out_x`. As soon as the first point different from `in_and_out_x`
365 is evaluated, FE counting starts.
367 Equally-good solutions will also be accepted, i.e., stored into
368 `in_and_out_x`. This costs a little bit of runtime, but would normally be
369 the preferred behavior: On many problems, making neutral moves (i.e.,
370 drifting) will be beneficial over only accepting strict improvements. This
371 is why :mod:`~moptipy.algorithms.so.rls` outperforms the normal
372 :mod:`~moptipy.algorithms.so.hill_climber` on the
373 :mod:`~moptipy.examples.jssp`.
375 :param owner: the owning process
376 :param in_and_out_x: the input solution record, which will be
377 overwritten with the best encountered solution
378 :param f: the objective value corresponding to `in_and_out`
379 """
380 return __FromStartingPoint(owner, in_and_out_x, f)
383class _InternalTerminationError(Exception):
384 """A protected internal termination error."""
387class __WithoutShouldTerminate(Process):
388 """A process allowing algorithm execution ignoring `should_terminate`."""
390 def __init__(self, owner: Process):
391 super().__init__()
392 if not isinstance(owner, Process):
393 raise type_error(owner, "owner", Process)
394 #: the owning process
395 self._owner: Final[Process] = owner
396 self.get_random = owner.get_random # type: ignore
397 a = owner.get_consumed_time_millis # type: ignore
398 self.get_consumed_time_millis = a # type: ignore
399 a = owner.get_max_time_millis # type: ignore
400 self.get_max_time_millis = a # type: ignore
401 a = owner.get_last_improvement_time_millis # type: ignore
402 self.get_last_improvement_time_millis = a # type: ignore
403 a = owner.get_last_improvement_fe # type: ignore
404 self.get_last_improvement_fe = a # type: ignore
405 self.has_log = owner.has_log # type: ignore
406 self.add_log_section = owner.add_log_section # type: ignore
407 self.lower_bound = owner.lower_bound # type: ignore
408 self.upper_bound = owner.upper_bound # type: ignore
409 self.create = owner.create # type: ignore
410 self.copy = owner.copy # type: ignore
411 self.to_str = owner.to_str # type: ignore
412 self.is_equal = owner.is_equal # type: ignore
413 self.from_str = owner.from_str # type: ignore
414 self.validate = owner.validate # type: ignore
415 self.n_points = owner.n_points # type: ignore
416 self.has_best = owner.has_best # type: ignore
417 self.get_copy_of_best_x = owner.get_copy_of_best_x # type: ignore
418 self.get_best_f = owner.get_best_f # type: ignore
419 self.should_terminate = owner.should_terminate # type: ignore
420 self.terminate = owner.terminate # type: ignore
421 self.get_max_fes = owner.get_max_fes # type: ignore
422 self.get_consumed_fes = owner.get_consumed_fes # type: ignore
423 self.get_log_basename = owner.get_log_basename # type: ignore
424 #: the fast call to the owner's evaluate method
425 self.__evaluate: Final[Callable[[Any], int | float]] \
426 = owner.evaluate
427 #: the fast call to the owner's register method
428 self.__register: Final[Callable[[Any, int | float], None]] \
429 = owner.register
431 def evaluate(self, x) -> float | int:
432 if self.should_terminate():
433 raise _InternalTerminationError
434 return self.__evaluate(x)
436 def register(self, x, f: int | float) -> None:
437 if self.should_terminate():
438 raise _InternalTerminationError
439 self.__register(x, f)
441 def __str__(self) -> str:
442 return f"protect_{self._owner}"
444 def __exit__(self, exception_type, exception_value, traceback) -> bool:
445 return True
448class __WithoutShouldTerminateMO(MOProcess):
449 """A process allowing algorithm execution ignoring `should_terminate`."""
451 def __init__(self, owner: Process):
452 super().__init__()
453 if not isinstance(owner, MOProcess):
454 raise type_error(owner, "owner", MOProcess)
455 #: the owning process
456 self._owner: Final[MOProcess] = owner
457 self.get_random = owner.get_random # type: ignore
458 a = owner.get_consumed_time_millis # type: ignore
459 self.get_consumed_time_millis = a # type: ignore
460 a = owner.get_max_time_millis # type: ignore
461 self.get_max_time_millis = a # type: ignore
462 a = owner.get_last_improvement_time_millis # type: ignore
463 self.get_last_improvement_time_millis = a # type: ignore
464 a = owner.get_last_improvement_fe # type: ignore
465 self.get_last_improvement_fe = a # type: ignore
466 self.has_log = owner.has_log # type: ignore
467 self.add_log_section = owner.add_log_section # type: ignore
468 self.lower_bound = owner.lower_bound # type: ignore
469 self.upper_bound = owner.upper_bound # type: ignore
470 self.create = owner.create # type: ignore
471 self.copy = owner.copy # type: ignore
472 self.to_str = owner.to_str # type: ignore
473 self.is_equal = owner.is_equal # type: ignore
474 self.from_str = owner.from_str # type: ignore
475 self.validate = owner.validate # type: ignore
476 self.n_points = owner.n_points # type: ignore
477 self.has_best = owner.has_best # type: ignore
478 self.get_copy_of_best_x = owner.get_copy_of_best_x # type: ignore
479 self.get_best_f = owner.get_best_f # type: ignore
480 self.should_terminate = owner.should_terminate # type: ignore
481 self.terminate = owner.terminate # type: ignore
482 self.get_max_fes = owner.get_max_fes # type: ignore
483 self.get_consumed_fes = owner.get_consumed_fes # type: ignore
484 self.f_create = owner.f_create # type: ignore
485 self.f_validate = owner.f_validate # type: ignore
486 self.f_dtype = owner.f_dtype # type: ignore
487 self.f_dominates = owner.f_dominates # type: ignore
488 self.f_dimension = owner.f_dimension # type: ignore
489 self.get_archive = owner.get_archive # type: ignore
490 self.check_in = owner.check_in # type: ignore
491 self.get_log_basename = owner.get_log_basename # type: ignore
492 #: the fast call to the owner's evaluate method
493 self.__evaluate: Final[Callable[[Any], int | float]] \
494 = owner.evaluate
495 #: the fast call to the owner's register method
496 self.__register: Final[Callable[[Any, int | float], None]] \
497 = owner.register
498 #: the fast call to the owner's f_evaluate method
499 self.__f_evaluate: Final[Callable[
500 [Any, np.ndarray], int | float]] \
501 = owner.f_evaluate
503 def evaluate(self, x) -> float | int:
504 if self.should_terminate():
505 raise _InternalTerminationError
506 return self.__evaluate(x)
508 def register(self, x, f: int | float) -> None:
509 if self.should_terminate():
510 raise _InternalTerminationError
511 self.__register(x, f)
513 def f_evaluate(self, x, fs: np.ndarray) -> int | float:
514 if self.should_terminate():
515 raise _InternalTerminationError
516 return self.__f_evaluate(x, fs)
518 def __str__(self) -> str:
519 return f"protectMO_{self._owner}"
521 def __exit__(self, exception_type, exception_value, traceback) -> bool:
522 return True
525def get_remaining_fes(process: Process) -> int:
526 """
527 Get a finite number representing the remaining FEs of a process.
529 If the process has the maximum objective function evaluations (FEs) set
530 (see :meth:`~moptipy.api.process.Process.get_max_fes`), then this method
531 returns the maximum FEs minus the consumed FEs (see
532 :meth:`~moptipy.api.process.Process.get_consumed_fes`).
533 Otherwise, i.e., if :meth:`~moptipy.api.process.Process.get_max_fes`
534 returns `None`, this function returns a very large number, namely
535 `9223372036854775807`, i.e., `(2 ** 63) - 1`. This number is so high that
536 it will always be impossible to consume it in terms of FEs. But it is
537 also finite in any case. When trying to slice of budgets or computing
538 things based on the remaining budget, this makes it unnecessary for us to
539 deal with special cases.
541 :param process: the process
542 :returns: an integer representing the remaining FEs of the process. If
543 no FE limit is imposed by `process`, a very large number will be
544 returned.
546 >>> from moptipy.api.process import Process as Proc
547 >>> class X(Proc):
548 ... def get_max_fes(self):
549 ... return None
550 ... def get_consumed_fes(self):
551 ... return 123
552 >>> get_remaining_fes(X())
553 9223372036854775807
554 >>> class Y(X):
555 ... def get_max_fes(self):
556 ... return 456
557 >>> get_remaining_fes(Y())
558 333
559 """
560 mf: int | None = process.get_max_fes() # get the number of available FEs
561 if mf is None: # if a no FE limit is specified, then return a large value
562 return 9_223_372_036_854_775_807 # (2 ** 63) - 1
563 return mf - process.get_consumed_fes() # else, subtract the consumed FEs
566def without_should_terminate(algorithm: Callable[[T], Any], process: T) \
567 -> None:
568 """
569 Apply an algorithm that does not call `should_terminate` to a process.
571 If we use an algorithm from an external library, this algorithm may ignore
572 the proper usage of our API. With this method, we try to find a way to
573 make sure that these calls are consistent with the termination criterion
574 of the `moptipy` API.
576 Before calling :meth:`~moptipy.api.process.Process.evaluate`,
577 :meth:`~moptipy.api.process.Process.register`, or
578 :meth:`~moptipy.api.mo_problem.MOProblem.f_evaluate`, an optimization
579 algorithm must check if it should instead stop via
580 :meth:`~moptipy.api.process.Process.should_terminate`. If the process
581 called :meth:`~moptipy.api.process.Process.should_terminate` and was
582 told to stop but did invoke the evaluation routines anyway, an
583 exception will be thrown and the process force-terminates. If the
584 process did not call
585 :meth:`~moptipy.api.process.Process.should_terminate` but was
586 supposed to stop, the results of
587 :meth:`~moptipy.api.process.Process.evaluate` may be arbitrary (or
588 positive infinity).
590 This function here can be used to deal with processes that do not invoke
591 :meth:`~moptipy.api.process.Process.should_terminate`. It will invoke
592 this method by itself before
593 :meth:`~moptipy.api.process.Process.evaluate`,
594 :meth:`~moptipy.api.process.Process.register`, and
595 :meth:`~moptipy.api.mo_problem.MOProblem.f_evaluate` and terminate the
596 algorithm with an exception if necessary. It will then catch the
597 exception and bury it.
599 Thus, we can now use algorithms that ignore our termination criteria and
600 still force them to terminate when they should.
602 Some algorithms using this system are implemented in
603 :mod:`~moptipy.algorithms.so.vector.scipy` and
604 :mod:`~moptipy.algorithms.so.vector.pdfo`. These modules import external
605 algorithms from other libraries which, of course, know nothing about how
606 our `moptipy` works. They only accept the objective function and cannot
607 handle the beautiful
608 :meth:`~moptipy.api.process.Process.should_terminate`-based termination
609 criteria. By using :func:`without_should_terminate`, however, we can still
610 safely use them within `moptipy` compliant scenarios.
612 :param algorithm: the algorithm
613 :param process: the optimization process
614 """
615 with contextlib.suppress(_InternalTerminationError), \
616 __WithoutShouldTerminateMO(process) \
617 if isinstance(process, MOProcess) \
618 else __WithoutShouldTerminate(process) as proc:
619 algorithm(proc)