Coverage for moptipy / api / _mo_process_no_ss_log.py: 92%

1"""A multi-objective process with logging.""" 

2from typing import Final 

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_no_ss import _MOProcessNoSS 

10from moptipy.api._process_base import _TIME_IN_NS, _check_log_time 

11from moptipy.api.algorithm import Algorithm 

12from moptipy.api.mo_archive import MOArchivePruner 

13from moptipy.api.mo_problem import MOProblem 

14from moptipy.api.space import Space 

15from moptipy.utils.logger import Logger 

16 

17 

18class _MOProcessNoSSLog(_MOProcessNoSS): 

19 """A multi-objective process with logging.""" 

20 

21 def __init__(self, 

22 solution_space: Space, 

23 objective: MOProblem, 

24 algorithm: Algorithm, 

25 pruner: MOArchivePruner, 

26 archive_max_size: int, 

27 archive_prune_limit: int, 

28 log_file: Path | None = None, 

29 rand_seed: int | None = None, 

30 max_fes: int | None = None, 

31 max_time_millis: int | None = None, 

32 goal_f: int | float | None = None, 

33 log_all_fes: bool = False) -> None: 

34 """ 

35 Perform the internal initialization. Do not call directly. 

36 

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

38 :param objective: the objective function 

39 :param algorithm: the optimization algorithm 

40 :param pruner: the archive pruner 

41 :param archive_max_size: the maximum archive size after pruning 

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

43 be performed 

44 :param log_file: the optional log file 

45 :param rand_seed: the optional random seed 

46 :param max_fes: the maximum permitted function evaluations 

47 :param max_time_millis: the maximum runtime in milliseconds 

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

49 is terminated 

50 :param log_all_fes: should we log all FEs? 

51 """ 

52 super().__init__(solution_space=solution_space, 

53 objective=objective, 

54 algorithm=algorithm, 

55 pruner=pruner, 

56 archive_max_size=archive_max_size, 

57 archive_prune_limit=archive_prune_limit, 

58 log_file=log_file, 

59 rand_seed=rand_seed, 

60 max_fes=max_fes, 

61 max_time_millis=max_time_millis, 

62 goal_f=goal_f) 

63 if not isinstance(log_file, str): 

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

65 if not isinstance(log_all_fes, bool): 

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

67 

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 | np.ndarray]] = [] 

72 #: the quick access to the log appending method 

73 self.__log_append = self.__log.append 

74 

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

76 if self._terminated: 

77 if self._knows_that_terminated: 

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

79 "the algorithm knows it.") 

80 return self._current_best_f 

81 

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

83 self._current_fes = current_fes = self._current_fes + 1 

84 do_term: bool = current_fes >= self._end_fes 

85 do_log: bool = self.__log_all 

86 ctn: int = 0 

87 

88 improved: bool = False 

89 if result < self._current_best_f: 

90 improved = True 

91 self._current_best_f = result 

92 copyto(self._current_best_fs, fs) 

93 self._copy_y(self._current_best_y, x) 

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

95 

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

97 self._last_improvement_fe = current_fes 

98 self._current_time_nanos = ctn = _TIME_IN_NS() 

99 self._last_improvement_time_nanos = ctn 

100 do_log = True 

101 

102 if do_log: 

103 if ctn <= 0: 

104 self._current_time_nanos = ctn = _TIME_IN_NS() 

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

106 

107 if do_term: 

108 self.terminate() 

109 

110 return result 

111 

112 def _check_timing(self) -> None: 

113 super()._check_timing() 

114 _check_log_time(self._start_time_nanos, self._current_time_nanos, 

115 self.__log) 

116 

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

118 self._write_mo_log(self.__log, self._start_time_nanos, 

119 self.__log_all, logger) 

120 del self.__log 

121 super()._write_log(logger) 

122 

123 def __str__(self) -> str: 

124 return "MOLoggingProcessWithoutSearchSpace"