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

1"""The multi-objective algorithm execution API.""" 

2 

3from math import isfinite 

4from typing import Final, Self, cast 

5 

6from pycommons.types import check_int_range 

7 

8from moptipy.api._mo_process_no_ss import _MOProcessNoSS 

9from moptipy.api._mo_process_no_ss_log import _MOProcessNoSSLog 

10from moptipy.api._mo_process_ss import _MOProcessSS 

11from moptipy.api._mo_process_ss_log import _MOProcessSSLog 

12from moptipy.api.algorithm import Algorithm, check_algorithm 

13from moptipy.api.encoding import Encoding, check_encoding 

14from moptipy.api.execution import Execution 

15from moptipy.api.improvement_logger import ImprovementLogger 

16from moptipy.api.mo_archive import MOArchivePruner, check_mo_archive_pruner 

17from moptipy.api.mo_problem import ( 

18 MOProblem, 

19 MOSOProblemBridge, 

20 check_mo_problem, 

21) 

22from moptipy.api.mo_process import MOProcess 

23from moptipy.api.objective import Objective, check_objective 

24from moptipy.api.process import ( 

25 check_goal_f, 

26 check_max_fes, 

27 check_max_time_millis, 

28) 

29from moptipy.api.space import Space, check_space 

30from moptipy.mo.archive.keep_farthest import KeepFarthest 

31from moptipy.utils.nputils import rand_seed_check 

32 

33 

34class MOExecution(Execution): 

35 """ 

36 Define all the components of a multi-objective experiment and execute it. 

37 

38 Different from :class:`~moptipy.api.execution.Execution`, this class here 

39 allows us to construct multi-objective optimization processes, i.e., such 

40 that have more than one optimization goal. 

41 """ 

42 

43 def __init__(self) -> None: 

44 """Create the multi-objective execution.""" 

45 super().__init__() 

46 #: the maximum size of a pruned archive 

47 self._archive_max_size: int | None = None 

48 #: the archive size limit at which pruning should be performed 

49 self._archive_prune_limit: int | None = None 

50 #: the archive pruning strategy 

51 self._archive_pruner: MOArchivePruner | None = None 

52 

53 def set_archive_max_size(self, size: int) -> Self: 

54 """ 

55 Set the upper limit for the archive size (after pruning). 

56 

57 The internal archive of the multi-objective optimization process 

58 retains non-dominated solutions encountered during the search. Since 

59 there can be infinitely many such solutions, the archive could grow 

60 without bound if left untouched. 

61 Therefore, we define two size limits: the maximum archive size 

62 (defined by this method) and the pruning limit. Once the archive grows 

63 beyond the pruning limit, it is cut down to the archive size limit. 

64 

65 :param size: the maximum archive size 

66 :returns: this execution 

67 """ 

68 check_int_range(size, "maximum archive size") 

69 if (self._archive_prune_limit is not None) and \ 

70 (size > self._archive_prune_limit): 

71 raise ValueError( 

72 f"archive max size {size} must be <= than archive " 

73 f"prune limit {self._archive_prune_limit}") 

74 self._archive_max_size = size 

75 return self 

76 

77 def set_archive_pruning_limit(self, 

78 limit: int) -> Self: 

79 """ 

80 Set the size limit of the archive above which pruning is performed. 

81 

82 If the size of the archive grows above this limit, the archive will be 

83 pruned down to the archive size limit. 

84 

85 :param limit: the archive pruning limit 

86 :returns: this execution 

87 """ 

88 check_int_range(limit, "limit", 1) 

89 if (self._archive_max_size is not None) and \ 

90 (limit < self._archive_max_size): 

91 raise ValueError( 

92 f"archive pruning limit {limit} must be >= than archive " 

93 f"maximum size {self._archive_max_size}") 

94 self._archive_prune_limit = limit 

95 return self 

96 

97 def set_archive_pruner(self, 

98 pruner: MOArchivePruner) -> Self: 

99 """ 

100 Set the pruning strategy for downsizing the archive. 

101 

102 :param pruner: the archive pruner 

103 :returns: this execution 

104 """ 

105 self._archive_pruner = check_mo_archive_pruner(pruner) 

106 return self 

107 

108 def set_objective(self, objective: Objective) -> Self: 

109 """ 

110 Set the objective function in form of a multi-objective problem. 

111 

112 :param objective: the objective function 

113 :returns: this execution 

114 """ 

115 check_objective(objective) 

116 if not isinstance(objective, MOProblem): 

117 objective = MOSOProblemBridge(objective) 

118 super().set_objective(check_mo_problem(objective)) 

119 return self 

120 

121 def execute(self) -> MOProcess: 

122 """ 

123 Create a multi-objective process, apply algorithm, and return result. 

124 

125 This method is multi-objective equivalent of the 

126 :meth:`~moptipy.api.execution.Execution.execute` method. It returns a 

127 multi-objective process after applying the multi-objective algorithm. 

128 

129 :returns: the instance of :class:`~moptipy.api.mo_process.MOProcess` 

130 after applying the algorithm. 

131 """ 

132 objective: Final[MOProblem] = cast("MOProblem", self._objective) 

133 solution_space: Final[Space] = check_space(self._solution_space) 

134 search_space: Final[Space | None] = check_space( 

135 self._search_space, self._encoding is None) 

136 encoding: Final[Encoding | None] = check_encoding( 

137 self._encoding, search_space is None) 

138 rand_seed = self._rand_seed 

139 if rand_seed is not None: 

140 rand_seed = rand_seed_check(rand_seed) 

141 max_time_millis = check_max_time_millis(self._max_time_millis, True) 

142 max_fes = check_max_fes(self._max_fes, True) 

143 goal_f = check_goal_f(self._goal_f, True) 

144 f_lb = objective.lower_bound() 

145 if (f_lb is not None) and isfinite(f_lb) and \ 

146 ((goal_f is None) or (f_lb > goal_f)): 

147 goal_f = f_lb 

148 

149 log_all_fes = self._log_all_fes 

150 log_improvements = self._log_improvements or self._log_all_fes 

151 

152 log_file = self._log_file 

153 if log_file is None: 

154 if log_all_fes: 

155 raise ValueError("Log file cannot be None " 

156 "if all FEs should be logged.") 

157 if log_improvements: 

158 raise ValueError("Log file cannot be None " 

159 "if improvements should be logged.") 

160 else: 

161 log_file.create_file_or_truncate() 

162 

163 pruner: Final[MOArchivePruner] = \ 

164 self._archive_pruner if self._archive_pruner is not None \ 

165 else KeepFarthest(objective) 

166 dim: Final[int] = objective.f_dimension() 

167 size: Final[int] = self._archive_max_size if \ 

168 self._archive_max_size is not None else ( 

169 self._archive_prune_limit if 

170 self._archive_prune_limit is not None 

171 else (1 if dim == 1 else 32)) 

172 limit: Final[int] = self._archive_prune_limit if \ 

173 self._archive_prune_limit is not None \ 

174 else (1 if dim == 1 else (size * 4)) 

175 algorithm: Final[Algorithm] = check_algorithm(self._algorithm) 

176 

177 logger: Final[ImprovementLogger | None] = self._logger( 

178 rand_seed, log_file) 

179 

180 process: Final[_MOProcessNoSS] = (_MOProcessNoSSLog( 

181 solution_space=solution_space, 

182 objective=objective, 

183 algorithm=algorithm, 

184 pruner=pruner, 

185 archive_max_size=size, 

186 archive_prune_limit=limit, 

187 log_file=log_file, 

188 rand_seed=rand_seed, 

189 max_fes=max_fes, 

190 max_time_millis=max_time_millis, 

191 goal_f=goal_f, 

192 log_all_fes=log_all_fes, 

193 improvement_logger=logger) 

194 if log_improvements or log_all_fes else _MOProcessNoSS( 

195 solution_space=solution_space, 

196 objective=objective, 

197 algorithm=algorithm, 

198 pruner=pruner, 

199 archive_max_size=size, 

200 archive_prune_limit=limit, 

201 log_file=log_file, 

202 rand_seed=rand_seed, 

203 max_fes=max_fes, 

204 max_time_millis=max_time_millis, 

205 goal_f=goal_f, 

206 improvement_logger=logger)) \ 

207 if search_space is None else (_MOProcessSSLog( 

208 solution_space=solution_space, 

209 objective=objective, 

210 algorithm=algorithm, 

211 pruner=pruner, 

212 archive_max_size=size, 

213 archive_prune_limit=limit, 

214 log_file=log_file, 

215 search_space=search_space, 

216 encoding=encoding, 

217 rand_seed=rand_seed, 

218 max_fes=max_fes, 

219 max_time_millis=max_time_millis, 

220 goal_f=goal_f, 

221 log_all_fes=log_all_fes, 

222 improvement_logger=logger) 

223 if log_improvements or log_all_fes else 

224 _MOProcessSS( 

225 solution_space=solution_space, 

226 objective=objective, 

227 algorithm=algorithm, 

228 pruner=pruner, 

229 archive_max_size=size, 

230 archive_prune_limit=limit, 

231 log_file=log_file, 

232 search_space=search_space, 

233 encoding=encoding, 

234 rand_seed=rand_seed, 

235 max_fes=max_fes, 

236 max_time_millis=max_time_millis, 

237 goal_f=goal_f, 

238 improvement_logger=logger)) 

239 

240 try: 

241 # noinspection PyProtectedMember 

242 process._after_init() # finalize the created process 

243 pruner.initialize() # initialize the pruner 

244 objective.initialize() # initialize the multi-objective problem 

245 if encoding is not None: 

246 encoding.initialize() # initialize the encoding 

247 solution_space.initialize() # initialize the solution space 

248 if search_space is not None: 

249 search_space.initialize() # initialize the search space 

250 algorithm.initialize() # initialize the algorithm 

251 algorithm.solve(process) # apply the algorithm 

252 except Exception as be: # noqa 

253 # noinspection PyProtectedMember 

254 process._caught = be 

255 return process