Coverage for moptipy / api / _process_ss_log.py: 89%

1"""A process with logging and different search and solution space.""" 

2from typing import Callable, Final, cast # pylint: disable=W0611 

3 

4from pycommons.io.path import Path 

5from pycommons.types import type_error 

6 

7from moptipy.api._process_base import _TIME_IN_NS, _check_log_time 

8from moptipy.api._process_no_ss import _write_log 

9from moptipy.api._process_ss import _ProcessSS 

10from moptipy.api.algorithm import Algorithm 

11from moptipy.api.encoding import Encoding 

12from moptipy.api.improvement_logger import ImprovementLogger 

13from moptipy.api.objective import Objective 

14from moptipy.api.space import Space 

15from moptipy.utils.logger import Logger 

16 

17 

18class _ProcessSSLog(_ProcessSS): 

19 """A process with logging and different search and solution space.""" 

20 

21 def __init__(self, 

22 solution_space: Space, 

23 objective: Objective, 

24 algorithm: Algorithm, 

25 log_file: Path, 

26 search_space: Space | None = None, 

27 encoding: Encoding | None = None, 

28 rand_seed: int | None = None, 

29 max_fes: int | None = None, 

30 max_time_millis: int | None = None, 

31 goal_f: int | float | None = None, 

32 log_all_fes: bool = False, 

33 improvement_logger: ImprovementLogger | 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 :param log_all_fes: should every single FE be logged? 

49 :param improvement_logger: an improvement logger, whose 

50 :meth:`~ImprovementLogger.log_improvement` method will be invoked 

51 whenever the process has registered an improvement 

52 """ 

53 super().__init__(solution_space=solution_space, 

54 objective=objective, 

55 algorithm=algorithm, 

56 log_file=log_file, 

57 search_space=search_space, 

58 encoding=encoding, 

59 rand_seed=rand_seed, 

60 max_fes=max_fes, 

61 max_time_millis=max_time_millis, 

62 goal_f=goal_f, 

63 improvement_logger=improvement_logger) 

64 if not isinstance(log_file, str): 

65 raise type_error(log_file, "log_file", str) 

66 if not isinstance(log_all_fes, bool): 

67 raise type_error(log_all_fes, "log_all_fes", bool) 

68 #: `True` if all FEs are logged, `False` to only log improvements. 

69 self.__log_all: Final[bool] = log_all_fes 

70 #: The in-memory log 

71 self.__log: list[list[int | float]] = [] 

72 #: the quick access to the log appending method 

73 self.__log_append = self.__log.append 

74 

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

76 if self._terminated: 

77 if self._knows_that_terminated: 

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

79 "algorithm knows it.") 

80 return self._current_best_f 

81 

82 current_y: Final = self._current_y 

83 self._g(x, current_y) 

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

85 self._current_fes = current_fes = self._current_fes + 1 

86 do_term: bool = current_fes >= self._end_fes 

87 do_log: bool = self.__log_all 

88 ctn: int = 0 

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 do_log = True 

100 if self._log_improvement: 

101 self._log_improvement( 

102 cast("Callable[[Logger], None]", 

103 lambda lg, _x=x, _y=current_y, _f=result: 

104 self._write_improvement(lg, _x, _y, _f))) 

105 

106 if do_log: 

107 if ctn <= 0: 

108 self._current_time_nanos = ctn = _TIME_IN_NS() 

109 self.__log_append([current_fes, ctn, result]) 

110 

111 if do_term: 

112 self.terminate() 

113 

114 return result 

115 

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 

122 

123 self._current_fes = current_fes = self._current_fes + 1 

124 do_term: bool = current_fes >= self._end_fes 

125 do_log: bool = self.__log_all 

126 ctn: int = 0 

127 

128 if f < self._current_best_f: 

129 self._last_improvement_fe = current_fes 

130 self._current_best_f = f 

131 self.copy(self._current_best_x, x) 

132 current_y: Final = self._current_y 

133 self._g(x, current_y) 

134 self._current_y = self._current_best_y 

135 self._current_best_y = current_y 

136 self._current_time_nanos = ctn = _TIME_IN_NS() 

137 self._last_improvement_time_nanos = ctn 

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

139 do_log = True 

140 if self._log_improvement: 

141 self._log_improvement( 

142 cast("Callable[[Logger], None]", 

143 lambda lg, _x=x, _y=current_y, _f=f: 

144 self._write_improvement(lg, _x, _y, _f))) 

145 

146 if do_log: 

147 if ctn <= 0: 

148 self._current_time_nanos = ctn = _TIME_IN_NS() 

149 self.__log_append([current_fes, ctn, f]) 

150 

151 if do_term: 

152 self.terminate() 

153 

154 def _check_timing(self) -> None: 

155 super()._check_timing() 

156 _check_log_time(self._start_time_nanos, self._current_time_nanos, 

157 self.__log) 

158 

159 def _write_log(self, logger: Logger) -> None: 

160 _write_log(self.__log, self._start_time_nanos, logger) 

161 del self.__log 

162 super()._write_log(logger) 

163 

164 def __str__(self) -> str: 

165 return "LoggingProcessWithSearchSpace"