Coverage for moptipy / api / _process_ss.py: 93%
96 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 10:36 +0000
1"""An implementation of processes with different search and solution spaces."""
2from typing import Callable, Final, cast
4from pycommons.io.path import Path
6from moptipy.api._process_base import _TIME_IN_NS
7from moptipy.api._process_no_ss import _ProcessNoSS
8from moptipy.api.algorithm import Algorithm
9from moptipy.api.encoding import Encoding, check_encoding
10from moptipy.api.improvement_logger import ImprovementLogger
11from moptipy.api.logging import (
12 SCOPE_ENCODING,
13 SCOPE_SEARCH_SPACE,
14 SECTION_RESULT_X,
15)
16from moptipy.api.objective import Objective
17from moptipy.api.space import Space, check_space
18from moptipy.utils.logger import KeyValueLogSection, Logger
21class _ProcessSS(_ProcessNoSS):
22 """A class implementing a process with search and solution space."""
24 def __init__(self,
25 solution_space: Space,
26 objective: Objective,
27 algorithm: Algorithm,
28 log_file: Path | None = None,
29 search_space: Space | None = None,
30 encoding: Encoding | None = None,
31 rand_seed: int | None = None,
32 max_fes: int | None = None,
33 max_time_millis: int | None = None,
34 goal_f: int | float | None = None,
35 improvement_logger: ImprovementLogger | None = None) -> None:
36 """
37 Perform the internal initialization. Do not call directly.
39 :param solution_space: the solution space.
40 :param objective: the objective function
41 :param algorithm: the optimization algorithm
42 :param search_space: the search space.
43 :param encoding: the encoding
44 :param log_file: the optional log file
45 :param rand_seed: the optional random seed
46 :param max_fes: the maximum permitted function evaluations
47 :param max_time_millis: the maximum runtime in milliseconds
48 :param goal_f: the goal objective value. if it is reached, the
49 process is terminated
50 :param improvement_logger: an improvement logger, whose
51 :meth:`~ImprovementLogger.log_improvement` method will be invoked
52 whenever the process has registered an improvement
53 """
54 super().__init__(solution_space=solution_space,
55 objective=objective,
56 algorithm=algorithm,
57 log_file=log_file,
58 rand_seed=rand_seed,
59 max_fes=max_fes,
60 max_time_millis=max_time_millis,
61 goal_f=goal_f,
62 improvement_logger=improvement_logger)
64 #: The search space.
65 self._search_space: Final[Space] = check_space(search_space)
66 #: The encoding.
67 self._encoding: Final[Encoding] = check_encoding(encoding)
68 #: the internal encoder
69 self._g: Final[Callable] = encoding.decode
70 #: The holder for the currently de-coded solution.
71 self._current_y = solution_space.create()
72 #: The current best point in the search space.
73 self._current_best_x: Final = search_space.create()
74 # wrappers
75 self.create = search_space.create # type: ignore
76 self.copy = search_space.copy # type: ignore
77 self.to_str = search_space.to_str # type: ignore
78 self.is_equal = search_space.is_equal # type: ignore
79 self.from_str = search_space.from_str # type: ignore
80 self.n_points = search_space.n_points # type: ignore
81 self.validate = search_space.validate # type: ignore
83 def evaluate(self, x) -> float | int:
84 if self._terminated:
85 if self._knows_that_terminated:
86 raise ValueError("The process has been terminated and the "
87 "algorithm knows it.")
88 return self._current_best_f
90 current_y: Final = self._current_y
91 self._g(x, current_y)
92 result: Final[int | float] = self._f(current_y)
93 self._current_fes = current_fes = self._current_fes + 1
94 do_term: bool = current_fes >= self._end_fes
96 if result < self._current_best_f:
97 self._last_improvement_fe = current_fes
98 self._current_best_f = result
99 self.copy(self._current_best_x, x)
100 self._current_y = self._current_best_y
101 self._current_best_y = current_y
102 self._current_time_nanos = ctn = _TIME_IN_NS()
103 self._last_improvement_time_nanos = ctn
104 do_term = do_term or (result <= self._end_f)
105 if self._log_improvement:
106 self._log_improvement(
107 cast("Callable[[Logger], None]",
108 lambda lg, _x=x, _y=current_y, _f=result:
109 self._write_improvement(lg, _x, _y, _f)))
111 if do_term:
112 self.terminate()
114 return result
116 def register(self, x, f: int | float) -> None:
117 if self._terminated:
118 if self._knows_that_terminated:
119 raise ValueError("The process has been terminated and the "
120 "algorithm knows it.")
121 return
123 self._current_fes = current_fes = self._current_fes + 1
124 do_term: bool = current_fes >= self._end_fes
126 if f < self._current_best_f:
127 self._last_improvement_fe = current_fes
128 self._current_best_f = f
129 self.copy(self._current_best_x, x)
130 current_y: Final = self._current_y
131 self._g(x, current_y)
132 self._current_y = self._current_best_y
133 self._current_best_y = current_y
134 self._current_time_nanos = ctn = _TIME_IN_NS()
135 self._last_improvement_time_nanos = ctn
136 do_term = do_term or (f <= self._end_f)
137 if self._log_improvement:
138 self._log_improvement(
139 cast("Callable[[Logger], None]",
140 lambda lg, _x=x, _y=current_y, _f=f:
141 self._write_improvement(lg, _x, _y, _f)))
143 if do_term:
144 self.terminate()
146 def get_copy_of_best_x(self, x) -> None:
147 if self._current_fes > 0:
148 return self.copy(x, self._current_best_x)
149 raise ValueError("No current best x available.")
151 def get_copy_of_best_y(self, y) -> None:
152 if self._current_fes > 0:
153 return self._copy_y(y, self._current_best_y)
154 raise ValueError("No current best y available.")
156 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
157 super().log_parameters_to(logger)
158 with logger.scope(SCOPE_SEARCH_SPACE) as sc:
159 self._search_space.log_parameters_to(sc)
160 with logger.scope(SCOPE_ENCODING) as sc:
161 self._encoding.log_parameters_to(sc)
163 def _write_result(self, logger: Logger) -> None:
164 with logger.text(SECTION_RESULT_X) as txt:
165 txt.write(self._search_space.to_str(self._current_best_x))
166 super()._write_result(logger)
168 def _validate_x(self) -> None:
169 """Validate x, if it exists."""
170 self._search_space.validate(self._current_best_x)
172 def __str__(self) -> str:
173 return "ProcessWithSearchSpace"
175 def _write_improvement(
176 self, logger: Logger, x, y, f: int | float) -> None:
177 """
178 Write an improvement to the logger.
180 :param logger: the logger
181 :param x: the point in the search space
182 :param y: the point in the solution space
183 :param f: the objective value
184 """
185 super()._write_improvement(logger, self._search_space.to_str(x), y, f)