Coverage for moptipy / api / _mo_process_ss.py: 92%
98 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""A multi-objective process with different search and solution spaces."""
3from math import isfinite
4from typing import Callable, Final
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.logging import (
16 PREFIX_SECTION_ARCHIVE,
17 SCOPE_ENCODING,
18 SCOPE_SEARCH_SPACE,
19 SECTION_RESULT_X,
20 SUFFIX_SECTION_ARCHIVE_X,
21 SUFFIX_SECTION_ARCHIVE_Y,
22)
23from moptipy.api.mo_archive import MOArchivePruner, MORecord
24from moptipy.api.mo_problem import MOProblem
25from moptipy.api.space import Space, check_space
26from moptipy.utils.logger import KeyValueLogSection, Logger
29class _MOProcessSS(_MOProcessNoSS):
30 """A class implementing a process with search and solution space."""
32 def __init__(self,
33 solution_space: Space,
34 objective: MOProblem,
35 algorithm: Algorithm,
36 pruner: MOArchivePruner,
37 archive_max_size: int,
38 archive_prune_limit: int,
39 log_file: Path | None = None,
40 search_space: Space | None = None,
41 encoding: Encoding | None = None,
42 rand_seed: int | None = None,
43 max_fes: int | None = None,
44 max_time_millis: int | None = None,
45 goal_f: int | float | None = None) -> None:
46 """
47 Perform the internal initialization. Do not call directly.
49 :param solution_space: the solution space.
50 :param objective: the objective function
51 :param algorithm: the optimization algorithm
52 :param pruner: the archive pruner
53 :param archive_max_size: the maximum archive size after pruning
54 :param archive_prune_limit: the archive size above which pruning will
55 be performed
56 :param search_space: the search space.
57 :param encoding: the encoding
58 :param log_file: the optional log file
59 :param rand_seed: the optional random seed
60 :param max_fes: the maximum permitted function evaluations
61 :param max_time_millis: the maximum runtime in milliseconds
62 :param goal_f: the goal objective value. if it is reached, the
63 process is terminated
64 """
65 super().__init__(solution_space=solution_space,
66 objective=objective,
67 algorithm=algorithm,
68 pruner=pruner,
69 archive_max_size=archive_max_size,
70 archive_prune_limit=archive_prune_limit,
71 log_file=log_file,
72 rand_seed=rand_seed,
73 max_fes=max_fes,
74 max_time_millis=max_time_millis,
75 goal_f=goal_f)
77 #: The search space.
78 self._search_space: Final[Space] = check_space(search_space)
79 #: The encoding.
80 self._encoding: Final[Encoding] = check_encoding(encoding)
81 #: the internal encoder
82 self._g: Final[Callable] = encoding.decode
83 #: The holder for the currently de-coded solution.
84 self._current_y = solution_space.create()
85 #: The current best point in the search space.
86 self._current_best_x: Final = search_space.create()
87 # wrappers
88 self.create = search_space.create # type: ignore
89 self.copy = search_space.copy # type: ignore
90 self.to_str = search_space.to_str # type: ignore
91 self.is_equal = search_space.is_equal # type: ignore
92 self.from_str = search_space.from_str # type: ignore
93 self.n_points = search_space.n_points # type: ignore
94 self.validate = search_space.validate # type: ignore
95 self._create_y = solution_space.create # the y creator
97 def f_evaluate(self, x, fs: np.ndarray) -> float | int:
98 if self._terminated:
99 if self._knows_that_terminated:
100 raise ValueError("The process has been terminated and the "
101 "algorithm knows it.")
102 return self._current_best_f
104 current_y: Final = self._current_y
105 self._g(x, current_y)
106 result: Final[int | float] = self._f_evaluate(current_y, fs)
107 self._current_fes = current_fes = self._current_fes + 1
108 do_term: bool = current_fes >= self._end_fes
110 improved: bool = False
111 if result < self._current_best_f:
112 improved = True
113 self._current_best_f = result
114 copyto(self._current_best_fs, fs)
115 self.copy(self._current_best_x, x)
116 self._current_y = self._current_best_y
117 self._current_best_y = current_y
118 do_term = do_term or (result <= self._end_f)
120 if self.check_in(x, fs, True) or improved:
121 self._last_improvement_fe = current_fes
122 self._current_time_nanos = ctn = _TIME_IN_NS()
123 self._last_improvement_time_nanos = ctn
125 if do_term:
126 self.terminate()
128 return result
130 def get_copy_of_best_x(self, x) -> None:
131 if self._current_fes > 0:
132 return self.copy(x, self._current_best_x)
133 raise ValueError("No current best x available.")
135 def get_copy_of_best_y(self, y) -> None:
136 if self._current_fes > 0:
137 return self._copy_y(y, self._current_best_y)
138 raise ValueError("No current best y available.")
140 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
141 super().log_parameters_to(logger)
142 with logger.scope(SCOPE_SEARCH_SPACE) as sc:
143 self._search_space.log_parameters_to(sc)
144 with logger.scope(SCOPE_ENCODING) as sc:
145 self._encoding.log_parameters_to(sc)
147 def _write_result(self, logger: Logger) -> None:
148 with logger.text(SECTION_RESULT_X) as txt:
149 txt.write(self._search_space.to_str(self._current_best_x))
150 super()._write_result(logger)
152 def _validate_x(self) -> None:
153 """Validate x, if it exists."""
154 self._search_space.validate(self._current_best_x)
156 def _log_and_check_archive_entry(self, index: int, rec: MORecord,
157 logger: Logger) -> int | float:
158 """
159 Write an archive entry.
161 :param index: the index of the entry
162 :param rec: the record to verify
163 :param logger: the logger
164 :returns: the objective value
165 """
166 self.validate(rec.x)
167 self.f_validate(rec.fs)
168 tfs: Final[np.ndarray] = self._fs_temp
170 current_y: Final = self._current_y
171 self._g(rec.x, current_y)
172 self._solution_space.validate(current_y)
173 f: Final[int | float] = self._f_evaluate(current_y, tfs)
175 if not np.array_equal(tfs, rec.fs):
176 raise ValueError(
177 f"expected {rec.fs} but got {tfs} when re-evaluating {rec}")
178 if not isinstance(f, int | float):
179 raise type_error(f, "scalarized objective value", (int, float))
180 if not isfinite(f):
181 raise ValueError(f"scalarized objective value {f} is not finite")
183 with logger.text(f"{PREFIX_SECTION_ARCHIVE}{index}"
184 f"{SUFFIX_SECTION_ARCHIVE_X}") as lg:
185 lg.write(self.to_str(rec.x))
187 with logger.text(f"{PREFIX_SECTION_ARCHIVE}{index}"
188 f"{SUFFIX_SECTION_ARCHIVE_Y}") as lg:
189 lg.write(self._solution_space.to_str(current_y))
191 return f
193 def __str__(self) -> str:
194 return "MOProcessWithSearchSpace"