Coverage for moptipy / algorithms / restarts.py: 83%

1"""Perform restarts of an algorithm that terminates too early.""" 

2from typing import Callable, Final, TypeVar, cast 

3 

4from pycommons.types import type_error 

5 

6from moptipy.api.algorithm import Algorithm 

7from moptipy.api.mo_algorithm import MOAlgorithm 

8from moptipy.api.mo_process import MOProcess 

9from moptipy.api.process import Process 

10from moptipy.utils.logger import KeyValueLogSection 

11 

12 

13class __RestartAlgorithm(Algorithm): 

14 """A wrapper around an existing algorithm.""" 

15 

16 def __init__(self, algorithm: Algorithm) -> None: 

17 """ 

18 Create the algorithm wrapper. 

19 

20 :param algorithm: the algorithm to wrap 

21 """ 

22 super().__init__() 

23 #: the algorithm 

24 self._algo: Final[Algorithm] = algorithm 

25 self.initialize = algorithm.initialize # type: ignore # fast call 

26 

27 def __str__(self) -> str: 

28 """ 

29 Get the string representation of this algorithm. 

30 

31 :returns: `rs_` + sub-algorithm name 

32 """ 

33 return f"rs_{self._algo}" 

34 

35 def solve(self, process: Process) -> None: 

36 """ 

37 Solve a single-objective problem. 

38 

39 :param process: the process 

40 """ 

41 st: Final[Callable[[], bool]] = process.should_terminate 

42 rst: Final[Callable[[], None]] = self._algo.initialize 

43 sv: Final[Callable[[Process], None]] = self._algo.solve 

44 while not st(): 

45 rst() 

46 sv(process) 

47 

48 def log_parameters_to(self, logger: KeyValueLogSection) -> None: 

49 """ 

50 Store the parameters of this algorithm wrapper to the logger. 

51 

52 :param logger: the logger 

53 """ 

54 super().log_parameters_to(logger) 

55 with logger.scope("a") as scope: 

56 self._algo.log_parameters_to(scope) 

57 

58 

59class __RestartMOAlgorithm(__RestartAlgorithm, MOAlgorithm): 

60 """A wrapper around an existing multi-objective algorithm.""" 

61 

62 def solve_mo(self, process: MOProcess) -> None: 

63 """ 

64 Solve a multi-objective problem. 

65 

66 :param process: the process 

67 """ 

68 st: Final[Callable[[], bool]] = process.should_terminate 

69 rst: Final[Callable[[], None]] = self.initialize 

70 sv: Final[Callable[[MOProcess], None]] = cast( 

71 "MOAlgorithm", self._algo).solve_mo 

72 while not st(): 

73 rst() 

74 sv(process) 

75 

76 def __str__(self) -> str: 

77 """ 

78 Get the string representation of this algorithm. 

79 

80 :returns: `mors_` + sub-algorithm name 

81 """ 

82 return f"mors_{self._algo}" 

83 

84 

85#: the type variable for single- and multi-objective algorithms. 

86T = TypeVar("T", Algorithm, MOAlgorithm) 

87 

88 

89def restarts(algorithm: T) -> T: 

90 """ 

91 Perform restarts of an algorithm until the termination criterion is met. 

92 

93 :param algorithm: the algorithm 

94 """ 

95 if isinstance(algorithm, MOAlgorithm): 

96 return __RestartMOAlgorithm(algorithm) 

97 if isinstance(algorithm, Algorithm): 

98 return __RestartAlgorithm(algorithm) 

99 raise type_error(algorithm, "algorithm", (Algorithm, MOAlgorithm))