Coverage for moptipy / api / _mo_process_ss.py: 92%
103 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 10:36 +0000
1"""A multi-objective process with different search and solution spaces."""
3from math import isfinite
4from typing import Callable, Final, cast
6import numpy as np
7from numpy import copyto
8from pycommons.io.path import Path
9from pycommons.types import type_error
11from moptipy.api._mo_process_no_ss import _MOProcessNoSS
12from moptipy.api._process_base import _TIME_IN_NS
13from moptipy.api.algorithm import Algorithm
14from moptipy.api.encoding import Encoding, check_encoding
15from moptipy.api.improvement_logger import ImprovementLogger
16from moptipy.api.logging import (
17 PREFIX_SECTION_ARCHIVE,
18 SCOPE_ENCODING,
19 SCOPE_SEARCH_SPACE,
20 SECTION_RESULT_X,
21 SUFFIX_SECTION_ARCHIVE_X,
22 SUFFIX_SECTION_ARCHIVE_Y,
23)
24from moptipy.api.mo_archive import MOArchivePruner, MORecord
25from moptipy.api.mo_problem import MOProblem
26from moptipy.api.space import Space, check_space
27from moptipy.utils.logger import KeyValueLogSection, Logger
30class _MOProcessSS(_MOProcessNoSS):
31 """A class implementing a process with search and solution space."""
33 def __init__(self,
34 solution_space: Space,
35 objective: MOProblem,
36 algorithm: Algorithm,
37 pruner: MOArchivePruner,
38 archive_max_size: int,
39 archive_prune_limit: int,
40 log_file: Path | None = None,
41 search_space: Space | None = None,
42 encoding: Encoding | None = None,
43 rand_seed: int | None = None,
44 max_fes: int | None = None,
45 max_time_millis: int | None = None,
46 goal_f: int | float | None = None,
47 improvement_logger: ImprovementLogger | None = None) -> None:
48 """
49 Perform the internal initialization. Do not call directly.
51 :param solution_space: the solution space.
52 :param objective: the objective function
53 :param algorithm: the optimization algorithm
54 :param pruner: the archive pruner
55 :param archive_max_size: the maximum archive size after pruning
56 :param archive_prune_limit: the archive size above which pruning will
57 be performed
58 :param search_space: the search space.
59 :param encoding: the encoding
60 :param log_file: the optional log file
61 :param rand_seed: the optional random seed
62 :param max_fes: the maximum permitted function evaluations
63 :param max_time_millis: the maximum runtime in milliseconds
64 :param goal_f: the goal objective value. if it is reached, the
65 process is terminated
66 :param improvement_logger: an improvement logger, whose
67 :meth:`~ImprovementLogger.log_improvement` method will be invoked
68 whenever the process has registered an improvement
69 """
70 super().__init__(solution_space=solution_space,
71 objective=objective,
72 algorithm=algorithm,
73 pruner=pruner,
74 archive_max_size=archive_max_size,
75 archive_prune_limit=archive_prune_limit,
76 log_file=log_file,
77 rand_seed=rand_seed,
78 max_fes=max_fes,
79 max_time_millis=max_time_millis,
80 goal_f=goal_f,
81 improvement_logger=improvement_logger)
83 #: The search space.
84 self._search_space: Final[Space] = check_space(search_space)
85 #: The encoding.
86 self._encoding: Final[Encoding] = check_encoding(encoding)
87 #: the internal encoder
88 self._g: Final[Callable] = encoding.decode
89 #: The holder for the currently de-coded solution.
90 self._current_y = solution_space.create()
91 #: The current best point in the search space.
92 self._current_best_x: Final = search_space.create()
93 # wrappers
94 self.create = search_space.create # type: ignore
95 self.copy = search_space.copy # type: ignore
96 self.to_str = search_space.to_str # type: ignore
97 self.is_equal = search_space.is_equal # type: ignore
98 self.from_str = search_space.from_str # type: ignore
99 self.n_points = search_space.n_points # type: ignore
100 self.validate = search_space.validate # type: ignore
101 self._create_y = solution_space.create # the y creator
103 def f_evaluate(self, x, fs: np.ndarray) -> float | int:
104 if self._terminated:
105 if self._knows_that_terminated:
106 raise ValueError("The process has been terminated and the "
107 "algorithm knows it.")
108 return self._current_best_f
110 current_y: Final = self._current_y
111 self._g(x, current_y)
112 result: Final[int | float] = self._f_evaluate(current_y, fs)
113 self._current_fes = current_fes = self._current_fes + 1
114 do_term: bool = current_fes >= self._end_fes
116 improved: bool = False
117 if result < self._current_best_f:
118 improved = True
119 self._current_best_f = result
120 copyto(self._current_best_fs, fs)
121 self.copy(self._current_best_x, x)
122 self._current_y = self._current_best_y
123 self._current_best_y = current_y
124 do_term = do_term or (result <= self._end_f)
126 if self.check_in(x, fs, True) or improved:
127 self._last_improvement_fe = current_fes
128 self._current_time_nanos = ctn = _TIME_IN_NS()
129 self._last_improvement_time_nanos = ctn
130 if self._log_improvement:
131 self._log_improvement(
132 cast("Callable[[Logger], None]",
133 lambda lg, _x=x, _y=current_y, _f=result, _fs=fs:
134 self._write_improvement(lg, _x, _y, _f, _fs)))
136 if do_term:
137 self.terminate()
139 return result
141 def get_copy_of_best_x(self, x) -> None:
142 if self._current_fes > 0:
143 return self.copy(x, self._current_best_x)
144 raise ValueError("No current best x available.")
146 def get_copy_of_best_y(self, y) -> None:
147 if self._current_fes > 0:
148 return self._copy_y(y, self._current_best_y)
149 raise ValueError("No current best y available.")
151 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
152 super().log_parameters_to(logger)
153 with logger.scope(SCOPE_SEARCH_SPACE) as sc:
154 self._search_space.log_parameters_to(sc)
155 with logger.scope(SCOPE_ENCODING) as sc:
156 self._encoding.log_parameters_to(sc)
158 def _write_result(self, logger: Logger) -> None:
159 with logger.text(SECTION_RESULT_X) as txt:
160 txt.write(self._search_space.to_str(self._current_best_x))
161 super()._write_result(logger)
163 def _validate_x(self) -> None:
164 """Validate x, if it exists."""
165 self._search_space.validate(self._current_best_x)
167 def _log_and_check_archive_entry(self, index: int, rec: MORecord,
168 logger: Logger) -> int | float:
169 """
170 Write an archive entry.
172 :param index: the index of the entry
173 :param rec: the record to verify
174 :param logger: the logger
175 :returns: the objective value
176 """
177 self.validate(rec.x)
178 self.f_validate(rec.fs)
179 tfs: Final[np.ndarray] = self._fs_temp
181 current_y: Final = self._current_y
182 self._g(rec.x, current_y)
183 self._solution_space.validate(current_y)
184 f: Final[int | float] = self._f_evaluate(current_y, tfs)
186 if not np.array_equal(tfs, rec.fs):
187 raise ValueError(
188 f"expected {rec.fs} but got {tfs} when re-evaluating {rec}")
189 if not isinstance(f, int | float):
190 raise type_error(f, "scalarized objective value", (int, float))
191 if not isfinite(f):
192 raise ValueError(f"scalarized objective value {f} is not finite")
194 with logger.text(f"{PREFIX_SECTION_ARCHIVE}{index}"
195 f"{SUFFIX_SECTION_ARCHIVE_X}") as lg:
196 lg.write(self.to_str(rec.x))
198 with logger.text(f"{PREFIX_SECTION_ARCHIVE}{index}"
199 f"{SUFFIX_SECTION_ARCHIVE_Y}") as lg:
200 lg.write(self._solution_space.to_str(current_y))
202 return f
204 def __str__(self) -> str:
205 return "MOProcessWithSearchSpace"
207 def _write_improvement(self, logger: Logger, x, y,
208 f: int | float, fs: np.ndarray) -> None:
209 """
210 Write an improvement to the logger.
212 :param logger: the logger
213 :param x: the point in the search space
214 :param y: the point in the solution space
215 :param f: the objective value
216 :param fs: the vector with the objective values
217 """
218 super()._write_improvement(
219 logger, self._search_space.to_str(x), y, f, fs)