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

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

2from typing import Final, cast 

3 

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

5from moptipy.api.logging import ( 

6 PROGRESS_CURRENT_F, 

7 PROGRESS_FES, 

8 PROGRESS_TIME_MILLIS, 

9 SECTION_PROGRESS, 

10) 

11from moptipy.utils.logger import Logger 

12 

13 

14class _ProcessNoSS(_ProcessBase): 

15 """ 

16 An internal class process implementation. 

17 

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

19 the search and solution space are the same. 

20 """ 

21 

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

23 if self._terminated: 

24 if self._knows_that_terminated: 

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

26 "the algorithm knows it.") 

27 return self._current_best_f 

28 

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

30 self._current_fes = current_fes = self._current_fes + 1 

31 do_term: bool = current_fes >= self._end_fes 

32 

33 if result < self._current_best_f: 

34 self._last_improvement_fe = current_fes 

35 self._current_best_f = result 

36 self._current_time_nanos = ctn = _TIME_IN_NS() 

37 self._last_improvement_time_nanos = ctn 

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

39 self._copy_y(self._current_best_y, x) 

40 

41 if do_term: 

42 self.terminate() 

43 

44 return result 

45 

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

47 if self._terminated: 

48 if self._knows_that_terminated: 

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

50 "the algorithm knows it.") 

51 return 

52 

53 self._current_fes = current_fes = self._current_fes + 1 

54 do_term: bool = current_fes >= self._end_fes 

55 

56 if f < self._current_best_f: 

57 self._last_improvement_fe = current_fes 

58 self._current_best_f = f 

59 self._current_time_nanos = ctn = _TIME_IN_NS() 

60 self._last_improvement_time_nanos = ctn 

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

62 self._copy_y(self._current_best_y, x) 

63 

64 if do_term: 

65 self.terminate() 

66 

67 def __str__(self) -> str: 

68 return "ProcessWithoutSearchSpace" 

69 

70 

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

72 start_time: int, 

73 logger: Logger) -> None: 

74 """ 

75 Write the log to a logger. 

76 

77 :param log: the log 

78 :param start_time: the start time 

79 :param logger: the logger 

80 """ 

81 if len(log) > 0: 

82 with logger.csv(SECTION_PROGRESS, 

83 [PROGRESS_FES, 

84 PROGRESS_TIME_MILLIS, 

85 PROGRESS_CURRENT_F]) as csv: 

86 for row in log: 

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

88 row[2]])