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

1"""Providing a process without explicit logging with a single space.""" 

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

3 

4from pycommons.strings.string_conv import num_to_str 

5 

6from moptipy.api._process_base import _TIME_IN_NS, _ns_to_ms, _ProcessBase 

7from moptipy.api.logging import ( 

8 KEY_CURRENT_F, 

9 PROGRESS_CURRENT_F, 

10 PROGRESS_FES, 

11 PROGRESS_TIME_MILLIS, 

12 SECTION_PROGRESS, 

13 SECTION_RESULT_X, 

14 SECTION_RESULT_Y, 

15) 

16from moptipy.utils.logger import Logger 

17 

18 

19class _ProcessNoSS(_ProcessBase): 

20 """ 

21 An internal class process implementation. 

22 

23 This class implements a stand-alone process without explicit logging where 

24 the search and solution space are the same. 

25 """ 

26 

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

28 if self._terminated: 

29 if self._knows_that_terminated: 

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

31 "the algorithm knows it.") 

32 return self._current_best_f 

33 

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

35 self._current_fes = current_fes = self._current_fes + 1 

36 do_term: bool = current_fes >= self._end_fes 

37 

38 if result < self._current_best_f: 

39 self._last_improvement_fe = current_fes 

40 self._current_best_f = result 

41 self._current_time_nanos = ctn = _TIME_IN_NS() 

42 self._last_improvement_time_nanos = ctn 

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

44 self._copy_y(self._current_best_y, x) 

45 if self._log_improvement: 

46 self._log_improvement( 

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

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

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

50 

51 if do_term: 

52 self.terminate() 

53 

54 return result 

55 

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

57 if self._terminated: 

58 if self._knows_that_terminated: 

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

60 "the algorithm knows it.") 

61 return 

62 

63 self._current_fes = current_fes = self._current_fes + 1 

64 do_term: bool = current_fes >= self._end_fes 

65 

66 if f < self._current_best_f: 

67 self._last_improvement_fe = current_fes 

68 self._current_best_f = f 

69 self._current_time_nanos = ctn = _TIME_IN_NS() 

70 self._last_improvement_time_nanos = ctn 

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

72 self._copy_y(self._current_best_y, x) 

73 if self._log_improvement: 

74 self._log_improvement( 

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

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

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

78 

79 if do_term: 

80 self.terminate() 

81 

82 def __str__(self) -> str: 

83 return "ProcessWithoutSearchSpace" 

84 

85 def _write_improvement(self, logger: Logger, x, y, f: int | float) -> None: 

86 """ 

87 Write an improvement to the logger. 

88 

89 :param logger: the logger 

90 :param x: the point in the search space 

91 :param y: the point in the solution space 

92 :param f: the objective value 

93 """ 

94 self._write_state_and_setup( 

95 logger, ((KEY_CURRENT_F, num_to_str(f)), )) 

96 if x is not None: 

97 with logger.text(SECTION_RESULT_X) as txt: 

98 txt.write(x) 

99 with logger.text(SECTION_RESULT_Y) as txt: 

100 txt.write(self._solution_space.to_str(y)) 

101 

102 

103def _write_log(log: list[list[int | float]], 

104 start_time: int, 

105 logger: Logger) -> None: 

106 """ 

107 Write the log to a logger. 

108 

109 :param log: the log 

110 :param start_time: the start time 

111 :param logger: the logger 

112 """ 

113 if len(log) > 0: 

114 with logger.csv(SECTION_PROGRESS, 

115 [PROGRESS_FES, 

116 PROGRESS_TIME_MILLIS, 

117 PROGRESS_CURRENT_F]) as csv: 

118 for row in log: 

119 csv.row([row[0], _ns_to_ms(cast("int", row[1]) - start_time), 

120 row[2]])