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

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

2from typing import Final 

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.objective import Objective 

11from moptipy.api.space import Space 

12from moptipy.utils.logger import Logger 

13 

14 

15class _ProcessNoSSLog(_ProcessNoSS): 

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

17 

18 def __init__(self, 

19 solution_space: Space, 

20 objective: Objective, 

21 algorithm: Algorithm, 

22 log_file: Path, 

23 rand_seed: int | None = None, 

24 max_fes: int | None = None, 

25 max_time_millis: int | None = None, 

26 goal_f: int | float | None = None, 

27 log_all_fes: bool = False) -> None: 

28 """ 

29 Perform the internal initialization. Do not call directly. 

30 

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

32 :param objective: the objective function 

33 :param algorithm: the optimization algorithm 

34 :param log_file: the optional log file 

35 :param rand_seed: the optional random seed 

36 :param max_fes: the maximum permitted function evaluations 

37 :param max_time_millis: the maximum runtime in milliseconds 

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

39 is terminated 

40 :param log_all_fes: should we log all FEs? 

41 """ 

42 super().__init__(solution_space=solution_space, 

43 objective=objective, 

44 algorithm=algorithm, 

45 log_file=log_file, 

46 rand_seed=rand_seed, 

47 max_fes=max_fes, 

48 max_time_millis=max_time_millis, 

49 goal_f=goal_f) 

50 if not isinstance(log_file, str): 

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

52 if not isinstance(log_all_fes, bool): 

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

54 

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

56 self.__log_all: Final[bool] = log_all_fes 

57 #: The in-memory log 

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

59 #: the quick access to the log appending method 

60 self.__log_append = self.__log.append 

61 

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

63 if self._terminated: 

64 if self._knows_that_terminated: 

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

66 "the algorithm knows it.") 

67 return self._current_best_f 

68 

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

70 self._current_fes = current_fes = self._current_fes + 1 

71 do_term: bool = current_fes >= self._end_fes 

72 do_log: bool = self.__log_all 

73 ctn: int = 0 

74 

75 if result < self._current_best_f: 

76 self._last_improvement_fe = current_fes 

77 self._current_best_f = result 

78 self._current_time_nanos = ctn = _TIME_IN_NS() 

79 self._last_improvement_time_nanos = ctn 

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

81 self._copy_y(self._current_best_y, x) 

82 do_log = True 

83 

84 if do_log: 

85 if ctn <= 0: 

86 self._current_time_nanos = ctn = _TIME_IN_NS() 

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

88 

89 if do_term: 

90 self.terminate() 

91 

92 return result 

93 

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

95 if self._terminated: 

96 if self._knows_that_terminated: 

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

98 "the algorithm knows it.") 

99 return 

100 

101 self._current_fes = current_fes = self._current_fes + 1 

102 do_term: bool = current_fes >= self._end_fes 

103 do_log: bool = self.__log_all 

104 ctn: int = 0 

105 

106 if f < self._current_best_f: 

107 self._last_improvement_fe = current_fes 

108 self._current_best_f = f 

109 self._current_time_nanos = ctn = _TIME_IN_NS() 

110 self._last_improvement_time_nanos = ctn 

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

112 self._copy_y(self._current_best_y, x) 

113 do_log = True 

114 

115 if do_log: 

116 if ctn <= 0: 

117 self._current_time_nanos = ctn = _TIME_IN_NS() 

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

119 

120 if do_term: 

121 self.terminate() 

122 

123 def _check_timing(self) -> None: 

124 super()._check_timing() 

125 _check_log_time(self._start_time_nanos, self._current_time_nanos, 

126 self.__log) 

127 

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

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

130 del self.__log 

131 super()._write_log(logger) 

132 

133 def __str__(self) -> str: 

134 return "LoggingProcessWithoutSearchSpace"