Coverage for moptipy / api / _mo_process_ss_log.py: 91%

1"""A multi-objective process with solution space and logging.""" 

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

3 

4import numpy as np 

5from numpy import copyto 

6from pycommons.io.path import Path 

7from pycommons.types import type_error 

8 

9from moptipy.api._mo_process_ss import _MOProcessSS 

10from moptipy.api._process_base import _TIME_IN_NS, _check_log_time 

11from moptipy.api.algorithm import Algorithm 

12from moptipy.api.encoding import Encoding 

13from moptipy.api.improvement_logger import ImprovementLogger 

14from moptipy.api.mo_archive import MOArchivePruner 

15from moptipy.api.mo_problem import MOProblem 

16from moptipy.api.space import Space 

17from moptipy.utils.logger import Logger 

18 

19 

20class _MOProcessSSLog(_MOProcessSS): 

21 """A multi-objective process with solution space and logging.""" 

22 

23 def __init__(self, 

24 solution_space: Space, 

25 objective: MOProblem, 

26 algorithm: Algorithm, 

27 pruner: MOArchivePruner, 

28 archive_max_size: int, 

29 archive_prune_limit: int, 

30 log_file: Path | None = None, 

31 search_space: Space | None = None, 

32 encoding: Encoding | None = None, 

33 rand_seed: int | None = None, 

34 max_fes: int | None = None, 

35 max_time_millis: int | None = None, 

36 goal_f: int | float | None = None, 

37 log_all_fes: bool = False, 

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

39 """ 

40 Perform the internal initialization. Do not call directly. 

41 

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

43 :param objective: the objective function 

44 :param algorithm: the optimization algorithm 

45 :param pruner: the archive pruner 

46 :param archive_max_size: the maximum archive size after pruning 

47 :param archive_prune_limit: the archive size above which pruning will 

48 be performed 

49 :param log_file: the optional log file 

50 :param search_space: the search space. 

51 :param encoding: the encoding 

52 :param rand_seed: the optional random seed 

53 :param max_fes: the maximum permitted function evaluations 

54 :param max_time_millis: the maximum runtime in milliseconds 

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

56 is terminated 

57 :param log_all_fes: should we log all FEs? 

58 :param improvement_logger: an improvement logger, whose 

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

60 whenever the process has registered an improvement 

61 """ 

62 super().__init__(solution_space=solution_space, 

63 objective=objective, 

64 algorithm=algorithm, 

65 pruner=pruner, 

66 archive_max_size=archive_max_size, 

67 archive_prune_limit=archive_prune_limit, 

68 log_file=log_file, 

69 search_space=search_space, 

70 encoding=encoding, 

71 rand_seed=rand_seed, 

72 max_fes=max_fes, 

73 max_time_millis=max_time_millis, 

74 goal_f=goal_f, 

75 improvement_logger=improvement_logger) 

76 if not isinstance(log_file, str): 

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

78 if not isinstance(log_all_fes, bool): 

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

80 

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

82 self.__log_all: Final[bool] = log_all_fes 

83 #: The in-memory log 

84 self.__log: list[list[int | float | np.ndarray]] = [] 

85 #: the quick access to the log appending method 

86 self.__log_append = self.__log.append 

87 

88 def f_evaluate(self, x, fs: np.ndarray) -> float | int: 

89 if self._terminated: 

90 if self._knows_that_terminated: 

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

92 "the algorithm knows it.") 

93 return self._current_best_f 

94 

95 current_y: Final = self._current_y 

96 self._g(x, current_y) 

97 result: Final[int | float] = self._f_evaluate(current_y, fs) 

98 self._current_fes = current_fes = self._current_fes + 1 

99 do_term: bool = current_fes >= self._end_fes 

100 do_log: bool = self.__log_all 

101 ctn: int = 0 

102 

103 improved: bool = False 

104 if result < self._current_best_f: 

105 improved = True 

106 self._current_best_f = result 

107 copyto(self._current_best_fs, fs) 

108 self.copy(self._current_best_x, x) 

109 self._current_y = self._current_best_y 

110 self._current_best_y = current_y 

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

112 

113 if self.check_in(x, fs, True) or improved: 

114 self._last_improvement_fe = current_fes 

115 self._current_time_nanos = ctn = _TIME_IN_NS() 

116 self._last_improvement_time_nanos = ctn 

117 do_log = True 

118 if self._log_improvement: 

119 self._log_improvement( 

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

121 lambda lg, _x=x, _y=current_y, _f=result, _fs=fs: 

122 self._write_improvement(lg, _x, _y, _f, _fs))) 

123 

124 if do_log: 

125 if ctn <= 0: 

126 self._current_time_nanos = ctn = _TIME_IN_NS() 

127 self.__log_append([current_fes, ctn, result, fs.copy()]) 

128 

129 if do_term: 

130 self.terminate() 

131 

132 return result 

133 

134 def _check_timing(self) -> None: 

135 super()._check_timing() 

136 _check_log_time(self._start_time_nanos, self._current_time_nanos, 

137 self.__log) 

138 

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

140 self._write_mo_log(self.__log, self._start_time_nanos, 

141 self.__log_all, logger) 

142 del self.__log 

143 super()._write_log(logger) 

144 

145 def __str__(self) -> str: 

146 return "MOLoggingProcessWithSearchSpace"