Coverage for moptipy / algorithms / so / vector / surrogate / _processes.py: 88%

1"""A set of process wrappers that can be used for surrogate modeling.""" 

2 

3from math import inf, isfinite 

4from typing import Any, Callable, Final 

5 

6import numpy as np 

7 

8from moptipy.api.process import Process 

9 

10 

11class _Surrogate(Process): 

12 """A surrogate process.""" 

13 

14 def __init__(self, owner: Process, max_fes: int): 

15 super().__init__() 

16 #: the owning process 

17 self._owner: Final[Process] = owner 

18 self.get_random = owner.get_random # type: ignore 

19 a = owner.get_consumed_time_millis # type: ignore 

20 self.get_consumed_time_millis = a # type: ignore 

21 a = owner.get_max_time_millis # type: ignore 

22 self.get_max_time_millis = a # type: ignore 

23 a = owner.get_last_improvement_time_millis # type: ignore 

24 self.get_last_improvement_time_millis = a # type: ignore 

25 self.has_log = owner.has_log # type: ignore 

26 self.add_log_section = owner.add_log_section # type: ignore 

27 self.lower_bound = owner.lower_bound # type: ignore 

28 self.upper_bound = owner.upper_bound # type: ignore 

29 self.create = owner.create # type: ignore 

30 self.copy = owner.copy # type: ignore 

31 self.to_str = owner.to_str # type: ignore 

32 self.is_equal = owner.is_equal # type: ignore 

33 self.from_str = owner.from_str # type: ignore 

34 self.validate = owner.validate # type: ignore 

35 self.n_points = owner.n_points # type: ignore 

36 #: the maximum FEs 

37 self.max_fes: Final[int] = max_fes 

38 #: the FEs that we still have left 

39 self._fes_left: int = max_fes 

40 #: did we terminate? 

41 self._terminated: bool = False 

42 #: the fast call to the owner's should_terminate method 

43 self.__should_terminate: Final[Callable[[], bool]] \ 

44 = owner.should_terminate 

45 

46 def should_terminate(self) -> bool: 

47 return self._terminated or self.__should_terminate() 

48 

49 def terminate(self) -> None: 

50 self._terminated = True 

51 

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

53 raise ValueError("you should not call this function.") 

54 

55 def get_consumed_fes(self) -> int: 

56 return self.max_fes - self._fes_left 

57 

58 def get_last_improvement_fe(self) -> int: 

59 return 1 if self._fes_left < self.max_fes else 0 

60 

61 def get_max_fes(self) -> int: 

62 return self.max_fes 

63 

64 def __str__(self) -> str: 

65 return f"{self.max_fes}_{self._owner}" 

66 

67 

68class _SurrogateApply(_Surrogate): 

69 """A process running for a `n` FEs and collecting the results.""" 

70 

71 #: the internal evaluation function 

72 _evaluate: Callable[[np.ndarray], np.ndarray] 

73 

74 def __init__(self, owner: Process, max_fes: int) -> None: 

75 super().__init__(owner, max_fes) 

76 #: the best-so-far solution 

77 self._best_x: Final[np.ndarray] = owner.create() 

78 #: the best-so-far objective value 

79 self._best_f: int | float = inf 

80 

81 def has_best(self) -> bool: 

82 return isfinite(self._best_f) 

83 

84 def get_best_f(self) -> int | float: 

85 return self._best_f 

86 

87 def get_copy_of_best_x(self, x) -> None: 

88 np.copyto(x, self._best_x) 

89 

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

91 f: Final[float] = self._evaluate( 

92 x.reshape((1, x.shape[0])))[0] 

93 fel: Final[int] = self._fes_left - 1 

94 self._fes_left = fel 

95 if fel <= 0: 

96 self._terminated = True 

97 if isfinite(f): 

98 if f < self._best_f: 

99 self._best_f = f 

100 np.copyto(self._best_x, x) 

101 return f 

102 return inf 

103 

104 

105class _SurrogateWarmup(_Surrogate): 

106 """A process running for a `n` FEs and collecting the results.""" 

107 

108 def __init__(self, owner: Process, max_fes: int, 

109 point_collector: Callable[[np.ndarray], None], 

110 f_collector: Callable[[int | float], None]): 

111 super().__init__(owner, max_fes) 

112 #: the point collector 

113 self.__point_collector = point_collector 

114 #: the objective value collector 

115 self.__f_collector = f_collector 

116 self.has_best = owner.has_best # type: ignore 

117 self.get_copy_of_best_x = owner.get_copy_of_best_x # type: ignore 

118 self.get_best_f = owner.get_best_f # type: ignore 

119 #: the owner's evaluation function 

120 self.__evaluate: Final[Callable[[Any], int | float]] = owner.evaluate 

121 

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

123 f: Final[int | float] = self.__evaluate(x) 

124 self.__point_collector(x.copy()) 

125 self.__f_collector(f) 

126 fel: Final[int] = self._fes_left - 1 

127 self._fes_left = fel 

128 if fel <= 0: 

129 self._terminated = True 

130 return f 

131 

132 def lower_bound(self) -> float: 

133 return -inf 

134 

135 def upper_bound(self) -> float: 

136 return inf 

137 

138 def __str__(self) -> str: 

139 return f"surrogateWarmup_{super().__str__()}"