Coverage for moptipy / api / _process_no_ss_log.py: 86%

1"""A process with logging, where search and solution space are the same.""" 

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 _ProcessNoSS, _write_log 

9from moptipy.api.algorithm import Algorithm 

10from moptipy.api.improvement_logger import ImprovementLogger 

11from moptipy.api.objective import Objective 

12from moptipy.api.space import Space 

13from moptipy.utils.logger import Logger 

14 

15 

16class _ProcessNoSSLog(_ProcessNoSS): 

17 """A process with logging, with the same search and solution space.""" 

18 

19 def __init__(self, 

20 solution_space: Space, 

21 objective: Objective, 

22 algorithm: Algorithm, 

23 log_file: Path, 

24 rand_seed: int | None = None, 

25 max_fes: int | None = None, 

26 max_time_millis: int | None = None, 

27 goal_f: int | float | None = None, 

28 log_all_fes: bool = False, 

29 improvement_logger: ImprovementLogger | None = None) -> None: 

30 """ 

31 Perform the internal initialization. Do not call directly. 

32 

33 :param solution_space: the search- and solution space. 

34 :param objective: the objective function 

35 :param algorithm: the optimization algorithm 

36 :param log_file: the optional log file 

37 :param rand_seed: the optional random seed 

38 :param max_fes: the maximum permitted function evaluations 

39 :param max_time_millis: the maximum runtime in milliseconds 

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

41 is terminated 

42 :param log_all_fes: should we log all FEs? 

43 :param improvement_logger: an improvement logger, whose 

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

45 whenever the process has registered an improvement 

46 """ 

47 super().__init__(solution_space=solution_space, 

48 objective=objective, 

49 algorithm=algorithm, 

50 log_file=log_file, 

51 rand_seed=rand_seed, 

52 max_fes=max_fes, 

53 max_time_millis=max_time_millis, 

54 goal_f=goal_f, 

55 improvement_logger=improvement_logger) 

56 if not isinstance(log_file, str): 

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

58 if not isinstance(log_all_fes, bool): 

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

60 

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

62 self.__log_all: Final[bool] = log_all_fes 

63 #: The in-memory log 

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

65 #: the quick access to the log appending method 

66 self.__log_append = self.__log.append 

67 

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

69 if self._terminated: 

70 if self._knows_that_terminated: 

71 raise ValueError("The process has been terminated and " 

72 "the algorithm knows it.") 

73 return self._current_best_f 

74 

75 result: Final[int | float] = self._f(x) 

76 self._current_fes = current_fes = self._current_fes + 1 

77 do_term: bool = current_fes >= self._end_fes 

78 do_log: bool = self.__log_all 

79 ctn: int = 0 

80 

81 if result < self._current_best_f: 

82 self._last_improvement_fe = current_fes 

83 self._current_best_f = result 

84 self._current_time_nanos = ctn = _TIME_IN_NS() 

85 self._last_improvement_time_nanos = ctn 

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

87 self._copy_y(self._current_best_y, x) 

88 do_log = True 

89 if self._log_improvement: 

90 self._log_improvement( 

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

92 lambda lg, _x=x, _f=result: 

93 self._write_improvement(lg, None, _x, _f))) 

94 

95 if do_log: 

96 if ctn <= 0: 

97 self._current_time_nanos = ctn = _TIME_IN_NS() 

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

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 " 

109 "the 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 do_log: bool = self.__log_all 

115 ctn: int = 0 

116 

117 if f < self._current_best_f: 

118 self._last_improvement_fe = current_fes 

119 self._current_best_f = f 

120 self._current_time_nanos = ctn = _TIME_IN_NS() 

121 self._last_improvement_time_nanos = ctn 

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

123 self._copy_y(self._current_best_y, x) 

124 do_log = True 

125 if self._log_improvement: 

126 self._log_improvement( 

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

128 lambda lg, _x=x, _f=f: 

129 self._write_improvement(lg, None, _x, _f))) 

130 

131 if do_log: 

132 if ctn <= 0: 

133 self._current_time_nanos = ctn = _TIME_IN_NS() 

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

135 

136 if do_term: 

137 self.terminate() 

138 

139 def _check_timing(self) -> None: 

140 super()._check_timing() 

141 _check_log_time(self._start_time_nanos, self._current_time_nanos, 

142 self.__log) 

143 

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

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

146 del self.__log 

147 super()._write_log(logger) 

148 

149 def __str__(self) -> str: 

150 return "LoggingProcessWithoutSearchSpace"