Coverage for moptipy / api / mo_execution.py: 92%
88 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""The multi-objective algorithm execution API."""
3from math import isfinite
4from typing import Final, Self, cast
6from pycommons.types import check_int_range
8from moptipy.api._mo_process_no_ss import _MOProcessNoSS
9from moptipy.api._mo_process_no_ss_log import _MOProcessNoSSLog
10from moptipy.api._mo_process_ss import _MOProcessSS
11from moptipy.api._mo_process_ss_log import _MOProcessSSLog
12from moptipy.api.algorithm import Algorithm, check_algorithm
13from moptipy.api.encoding import Encoding, check_encoding
14from moptipy.api.execution import Execution
15from moptipy.api.mo_archive import MOArchivePruner, check_mo_archive_pruner
16from moptipy.api.mo_problem import (
17 MOProblem,
18 MOSOProblemBridge,
19 check_mo_problem,
20)
21from moptipy.api.mo_process import MOProcess
22from moptipy.api.objective import Objective, check_objective
23from moptipy.api.process import (
24 check_goal_f,
25 check_max_fes,
26 check_max_time_millis,
27)
28from moptipy.api.space import Space, check_space
29from moptipy.mo.archive.keep_farthest import KeepFarthest
30from moptipy.utils.nputils import rand_seed_check
33class MOExecution(Execution):
34 """
35 Define all the components of a multi-objective experiment and execute it.
37 Different from :class:`~moptipy.api.execution.Execution`, this class here
38 allows us to construct multi-objective optimization processes, i.e., such
39 that have more than one optimization goal.
40 """
42 def __init__(self) -> None:
43 """Create the multi-objective execution."""
44 super().__init__()
45 #: the maximum size of a pruned archive
46 self._archive_max_size: int | None = None
47 #: the archive size limit at which pruning should be performed
48 self._archive_prune_limit: int | None = None
49 #: the archive pruning strategy
50 self._archive_pruner: MOArchivePruner | None = None
52 def set_archive_max_size(self, size: int) -> Self:
53 """
54 Set the upper limit for the archive size (after pruning).
56 The internal archive of the multi-objective optimization process
57 retains non-dominated solutions encountered during the search. Since
58 there can be infinitely many such solutions, the archive could grow
59 without bound if left untouched.
60 Therefore, we define two size limits: the maximum archive size
61 (defined by this method) and the pruning limit. Once the archive grows
62 beyond the pruning limit, it is cut down to the archive size limit.
64 :param size: the maximum archive size
65 :returns: this execution
66 """
67 check_int_range(size, "maximum archive size")
68 if (self._archive_prune_limit is not None) and \
69 (size > self._archive_prune_limit):
70 raise ValueError(
71 f"archive max size {size} must be <= than archive "
72 f"prune limit {self._archive_prune_limit}")
73 self._archive_max_size = size
74 return self
76 def set_archive_pruning_limit(self,
77 limit: int) -> Self:
78 """
79 Set the size limit of the archive above which pruning is performed.
81 If the size of the archive grows above this limit, the archive will be
82 pruned down to the archive size limit.
84 :param limit: the archive pruning limit
85 :returns: this execution
86 """
87 check_int_range(limit, "limit", 1)
88 if (self._archive_max_size is not None) and \
89 (limit < self._archive_max_size):
90 raise ValueError(
91 f"archive pruning limit {limit} must be >= than archive "
92 f"maximum size {self._archive_max_size}")
93 self._archive_prune_limit = limit
94 return self
96 def set_archive_pruner(self,
97 pruner: MOArchivePruner) -> Self:
98 """
99 Set the pruning strategy for downsizing the archive.
101 :param pruner: the archive pruner
102 :returns: this execution
103 """
104 self._archive_pruner = check_mo_archive_pruner(pruner)
105 return self
107 def set_objective(self, objective: Objective) -> Self:
108 """
109 Set the objective function in form of a multi-objective problem.
111 :param objective: the objective function
112 :returns: this execution
113 """
114 check_objective(objective)
115 if not isinstance(objective, MOProblem):
116 objective = MOSOProblemBridge(objective)
117 super().set_objective(check_mo_problem(objective))
118 return self
120 def execute(self) -> MOProcess:
121 """
122 Create a multi-objective process, apply algorithm, and return result.
124 This method is multi-objective equivalent of the
125 :meth:`~moptipy.api.execution.Execution.execute` method. It returns a
126 multi-objective process after applying the multi-objective algorithm.
128 :returns: the instance of :class:`~moptipy.api.mo_process.MOProcess`
129 after applying the algorithm.
130 """
131 objective: Final[MOProblem] = cast("MOProblem", self._objective)
132 solution_space: Final[Space] = check_space(self._solution_space)
133 search_space: Final[Space | None] = check_space(
134 self._search_space, self._encoding is None)
135 encoding: Final[Encoding | None] = check_encoding(
136 self._encoding, search_space is None)
137 rand_seed = self._rand_seed
138 if rand_seed is not None:
139 rand_seed = rand_seed_check(rand_seed)
140 max_time_millis = check_max_time_millis(self._max_time_millis, True)
141 max_fes = check_max_fes(self._max_fes, True)
142 goal_f = check_goal_f(self._goal_f, True)
143 f_lb = objective.lower_bound()
144 if (f_lb is not None) and isfinite(f_lb) and \
145 ((goal_f is None) or (f_lb > goal_f)):
146 goal_f = f_lb
148 log_all_fes = self._log_all_fes
149 log_improvements = self._log_improvements or self._log_all_fes
151 log_file = self._log_file
152 if log_file is None:
153 if log_all_fes:
154 raise ValueError("Log file cannot be None "
155 "if all FEs should be logged.")
156 if log_improvements:
157 raise ValueError("Log file cannot be None "
158 "if improvements should be logged.")
159 else:
160 log_file.create_file_or_truncate()
162 pruner: Final[MOArchivePruner] = \
163 self._archive_pruner if self._archive_pruner is not None \
164 else KeepFarthest(objective)
165 dim: Final[int] = objective.f_dimension()
166 size: Final[int] = self._archive_max_size if \
167 self._archive_max_size is not None else (
168 self._archive_prune_limit if
169 self._archive_prune_limit is not None
170 else (1 if dim == 1 else 32))
171 limit: Final[int] = self._archive_prune_limit if \
172 self._archive_prune_limit is not None \
173 else (1 if dim == 1 else (size * 4))
174 algorithm: Final[Algorithm] = check_algorithm(self._algorithm)
176 process: Final[_MOProcessNoSS] = (_MOProcessNoSSLog(
177 solution_space, objective, algorithm, pruner, size, limit,
178 log_file, rand_seed, max_fes, max_time_millis, goal_f,
179 log_all_fes) if log_improvements or log_all_fes else
180 _MOProcessNoSS(solution_space, objective, algorithm, pruner,
181 size, limit, log_file, rand_seed, max_fes,
182 max_time_millis, goal_f)) \
183 if search_space is None else (_MOProcessSSLog(
184 solution_space, objective, algorithm, pruner, size, limit,
185 log_file, search_space, encoding, rand_seed, max_fes,
186 max_time_millis, goal_f,
187 log_all_fes) if log_improvements or log_all_fes else
188 _MOProcessSS(solution_space, objective, algorithm, pruner, size,
189 limit, log_file, search_space, encoding, rand_seed,
190 max_fes, max_time_millis, goal_f))
192 try:
193 # noinspection PyProtectedMember
194 process._after_init() # finalize the created process
195 pruner.initialize() # initialize the pruner
196 objective.initialize() # initialize the multi-objective problem
197 if encoding is not None:
198 encoding.initialize() # initialize the encoding
199 solution_space.initialize() # initialize the solution space
200 if search_space is not None:
201 search_space.initialize() # initialize the search space
202 algorithm.initialize() # initialize the algorithm
203 algorithm.solve(process) # apply the algorithm
204 except Exception as be: # noqa: BLE001
205 # noinspection PyProtectedMember
206 process._caught = be
207 return process