Coverage for moptipy / api / _process_ss.py: 93%

1"""An implementation of processes with different search and solution spaces.""" 

2from typing import Callable, Final 

3 

4from pycommons.io.path import Path 

5 

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.logging import ( 

11 SCOPE_ENCODING, 

12 SCOPE_SEARCH_SPACE, 

13 SECTION_RESULT_X, 

14) 

15from moptipy.api.objective import Objective 

16from moptipy.api.space import Space, check_space 

17from moptipy.utils.logger import KeyValueLogSection, Logger 

18 

19 

20class _ProcessSS(_ProcessNoSS): 

21 """A class implementing a process with search and solution space.""" 

22 

23 def __init__(self, 

24 solution_space: Space, 

25 objective: Objective, 

26 algorithm: Algorithm, 

27 log_file: Path | None = None, 

28 search_space: Space | None = None, 

29 encoding: Encoding | None = None, 

30 rand_seed: int | None = None, 

31 max_fes: int | None = None, 

32 max_time_millis: int | None = None, 

33 goal_f: int | float | None = None) -> None: 

34 """ 

35 Perform the internal initialization. Do not call directly. 

36 

37 :param solution_space: the solution space. 

38 :param objective: the objective function 

39 :param algorithm: the optimization algorithm 

40 :param search_space: the search space. 

41 :param encoding: the encoding 

42 :param log_file: the optional log file 

43 :param rand_seed: the optional random seed 

44 :param max_fes: the maximum permitted function evaluations 

45 :param max_time_millis: the maximum runtime in milliseconds 

46 :param goal_f: the goal objective value. if it is reached, the 

47 process is terminated 

48 """ 

49 super().__init__(solution_space=solution_space, 

50 objective=objective, 

51 algorithm=algorithm, 

52 log_file=log_file, 

53 rand_seed=rand_seed, 

54 max_fes=max_fes, 

55 max_time_millis=max_time_millis, 

56 goal_f=goal_f) 

57 

58 #: The search space. 

59 self._search_space: Final[Space] = check_space(search_space) 

60 #: The encoding. 

61 self._encoding: Final[Encoding] = check_encoding(encoding) 

62 #: the internal encoder 

63 self._g: Final[Callable] = encoding.decode 

64 #: The holder for the currently de-coded solution. 

65 self._current_y = solution_space.create() 

66 #: The current best point in the search space. 

67 self._current_best_x: Final = search_space.create() 

68 # wrappers 

69 self.create = search_space.create # type: ignore 

70 self.copy = search_space.copy # type: ignore 

71 self.to_str = search_space.to_str # type: ignore 

72 self.is_equal = search_space.is_equal # type: ignore 

73 self.from_str = search_space.from_str # type: ignore 

74 self.n_points = search_space.n_points # type: ignore 

75 self.validate = search_space.validate # type: ignore 

76 

77 def evaluate(self, x) -> float | int: 

78 if self._terminated: 

79 if self._knows_that_terminated: 

80 raise ValueError("The process has been terminated and the " 

81 "algorithm knows it.") 

82 return self._current_best_f 

83 

84 current_y: Final = self._current_y 

85 self._g(x, current_y) 

86 result: Final[int | float] = self._f(current_y) 

87 self._current_fes = current_fes = self._current_fes + 1 

88 do_term: bool = current_fes >= self._end_fes 

89 

90 if result < self._current_best_f: 

91 self._last_improvement_fe = current_fes 

92 self._current_best_f = result 

93 self.copy(self._current_best_x, x) 

94 self._current_y = self._current_best_y 

95 self._current_best_y = current_y 

96 self._current_time_nanos = ctn = _TIME_IN_NS() 

97 self._last_improvement_time_nanos = ctn 

98 do_term = do_term or (result <= self._end_f) 

99 

100 if do_term: 

101 self.terminate() 

102 

103 return result 

104 

105 def register(self, x, f: int | float) -> None: 

106 if self._terminated: 

107 if self._knows_that_terminated: 

108 raise ValueError("The process has been terminated and the " 

109 "algorithm knows it.") 

110 return 

111 

112 self._current_fes = current_fes = self._current_fes + 1 

113 do_term: bool = current_fes >= self._end_fes 

114 

115 if f < self._current_best_f: 

116 self._last_improvement_fe = current_fes 

117 self._current_best_f = f 

118 self.copy(self._current_best_x, x) 

119 current_y: Final = self._current_y 

120 self._g(x, current_y) 

121 self._current_y = self._current_best_y 

122 self._current_best_y = current_y 

123 self._current_time_nanos = ctn = _TIME_IN_NS() 

124 self._last_improvement_time_nanos = ctn 

125 do_term = do_term or (f <= self._end_f) 

126 

127 if do_term: 

128 self.terminate() 

129 

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.") 

134 

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.") 

139 

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) 

146 

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) 

151 

152 def _validate_x(self) -> None: 

153 """Validate x, if it exists.""" 

154 self._search_space.validate(self._current_best_x) 

155 

156 def __str__(self) -> str: 

157 return "ProcessWithSearchSpace"