Coverage for o2/models/solution.py: 81%

1import functools 

2import math 

3from dataclasses import dataclass, field 

4from json import dumps 

5from typing import Optional 

6 

7from o2.actions.base_actions.base_action import BaseAction 

8from o2.models.evaluation import Evaluation 

9from o2.models.settings import Settings 

10from o2.models.state import State 

11from o2.models.timetable import TimetableType 

12from o2.util.helper import hash_string, hex_id 

13from o2.util.solution_dumper import SolutionDumper 

14 

15 

16@dataclass(frozen=True) 

17class Solution: 

18 """A solution in the optimization process. 

19 

20 A solution is a pair of an evaluation, the (new) state, the old, 

21 parent state, and the actions that led to this solution. With the last 

22 action being the one that led to this solution. 

23 

24 NOTE: There are two evaluations defined in the class, with `evaluation` definition 

25 just being there to satisfy type checking. The actual evaluation is stored in 

26 `_evaluation`, which is loaded from the evaluation file when needed. 

27 """ 

28 

29 actions: list["BaseAction"] = field(default_factory=list) 

30 """Actions taken since the base state.""" 

31 

32 evaluation: Evaluation # type: ignore 

33 _evaluation: Optional[Evaluation] = field(init=False, repr=False, default=None) 

34 

35 @property 

36 def evaluation(self) -> Evaluation: 

37 """Return the evaluation of the solution.""" 

38 # In case this is a pre-archived solution (e.g. from a previous run), 

39 # we can take the evaluation from __dict__['evaluation'] 

40 if ( 

41 self._evaluation is None 

42 and "evaluation" in self.__dict__ 

43 and self.__dict__["evaluation"] is not None 

44 ): 

45 self.__dict__["_evaluation"] = self.__dict__["evaluation"] 

46 return self.__dict__["_evaluation"] 

47 # Else we just load the evaluation from the evaluation file 

48 elif self._evaluation is None and Settings.ARCHIVE_SOLUTIONS: 

49 self.__dict__["_evaluation"] = SolutionDumper.instance.load_evaluation(self) 

50 assert self._evaluation is not None 

51 return self._evaluation 

52 

53 @evaluation.setter 

54 def evaluation(self, value: Evaluation) -> None: 

55 """Set the evaluation of the solution.""" 

56 self.__dict__["_evaluation"] = value 

57 

58 state: State # type: ignore 

59 _state: Optional[State] = field(init=False, repr=False, default=None) 

60 

61 @property 

62 def timetable(self) -> TimetableType: 

63 """Return the timetable of the solution.""" 

64 return self.state.timetable 

65 

66 @property 

67 def state(self) -> State: 

68 """Return the state of the solution.""" 

69 # In case this is a pre-archived solution (e.g. from a previous run), 

70 # we can take the state from __dict__['state'] 

71 if self._state is None and "state" in self.__dict__ and self.__dict__["state"] is not None: 

72 self.__dict__["_state"] = self.__dict__["state"] 

73 return self.__dict__["_state"] 

74 # Else we just load the state from the state file 

75 elif self._state is None and Settings.ARCHIVE_SOLUTIONS: 

76 self.__dict__["_state"] = SolutionDumper.instance.load_state(self) 

77 assert self._state is not None 

78 return self._state 

79 

80 @state.setter 

81 def state(self, value: State) -> None: 

82 """Set the state of the solution.""" 

83 self.__dict__["_state"] = value 

84 

85 @property 

86 def is_base_solution(self) -> bool: 

87 """Check if this state is the base solution.""" 

88 return not self.actions 

89 

90 @property 

91 def last_action(self) -> "BaseAction": 

92 """Return the last action taken.""" 

93 return self.actions[-1] 

94 

95 @functools.cached_property 

96 def point(self) -> tuple[float, float]: 

97 """Return the evaluation as a point.""" 

98 return self.evaluation.to_tuple() 

99 

100 @functools.cached_property 

101 def pareto_x(self) -> float: 

102 """Return the pareto x of the solution.""" 

103 return self.evaluation.pareto_x 

104 

105 @functools.cached_property 

106 def pareto_y(self) -> float: 

107 """Return the pareto y of the solution.""" 

108 return self.evaluation.pareto_y 

109 

110 def distance_to(self, other: "Solution") -> float: 

111 """Calculate the euclidean distance between two evaluations.""" 

112 return math.sqrt((self.pareto_x - other.pareto_x) ** 2 + (self.pareto_y - other.pareto_y) ** 2) 

113 

114 def is_dominated_by(self, other: "Solution") -> bool: 

115 """Check if this solution is dominated by the given solution.""" 

116 if not Settings.EQUAL_DOMINATION_ALLOWED: 

117 return other.pareto_x <= self.pareto_x and other.pareto_y <= self.pareto_y 

118 return other.pareto_x < self.pareto_x and other.pareto_y < self.pareto_y 

119 

120 def has_equal_point(self, solution: "Solution") -> bool: 

121 """Check if this solution has the same point as the given solution.""" 

122 return self.point == solution.point 

123 

124 def archive(self) -> None: 

125 """Archive the solution. 

126 

127 For downwards compatibility, we need to check `__dict__['evaluation']` 

128 and `__dict__['state']` as well. 

129 """ 

130 if not Settings.ARCHIVE_SOLUTIONS: 

131 return 

132 # Solution is already archived 

133 if self._evaluation is not None or ( 

134 "evaluation" in self.__dict__ and self.__dict__["evaluation"] is not None 

135 ): 

136 # Make sure the computed fields are triggered 

137 self.pareto_x # noqa: B018 

138 self.pareto_y # noqa: B018 

139 self.point # noqa: B018 

140 SolutionDumper.instance.dump_evaluation(self) 

141 self.__dict__["_evaluation"] = None 

142 self.__dict__["evaluation"] = None 

143 if self._state is not None or ("state" in self.__dict__ and self.__dict__["state"] is not None): 

144 # Make sure the computed fields are triggered 

145 self._cache_timetable_hash() # noqa: B018 

146 SolutionDumper.instance.dump_state(self) 

147 self.__dict__["_state"] = None 

148 # Make sure that the legacy state is removed from __dict__ 

149 self.__dict__["state"] = None 

150 # If we still got the parent state (from a previous run), we need to remove it 

151 if "parent_state" in self.__dict__: 

152 self.__dict__["parent_state"] = None 

153 

154 def __eq__(self, value: object) -> bool: 

155 """Check if this solution is equal to the given object.""" 

156 if not isinstance(value, Solution): 

157 return False 

158 if Settings.CHECK_FOR_TIMETABLE_EQUALITY: 

159 return self.__hash__() == value.__hash__() 

160 return self.id == value.id 

161 

162 def _cache_timetable_hash(self) -> None: 

163 """Cache the state hash.""" 

164 self.__dict__["_timetable_hash"] = hash(self.state.timetable) 

165 

166 def __hash__(self) -> int: 

167 """Hash the solution (possibly by id or state). 

168 

169 Also we cache the hash in a __dict__ field. 

170 (manually as functools.cached_property does not work with __hash__) 

171 """ 

172 if Settings.CHECK_FOR_TIMETABLE_EQUALITY: 

173 if "_timetable_hash" not in self.__dict__: 

174 self._cache_timetable_hash() 

175 return self.__dict__["_timetable_hash"] 

176 return int(self.id, 16) 

177 

178 @functools.cached_property 

179 def is_valid(self) -> bool: 

180 """Check if the evaluation is valid.""" 

181 return ( 

182 # Ensure that there was no error running the simulation, 

183 # that results in a <= 0 value. 

184 # Or that the solution is empty 

185 self.pareto_x > 0 and self.pareto_y > 0 

186 ) 

187 

188 @functools.cached_property 

189 def id(self) -> str: 

190 """A unique identifier for the solution. 

191 

192 Generated only based on the actions. 

193 """ 

194 # When we are the base_solution, we have no actions 

195 # -> we cant hash them -> we hash the timetable 

196 if not self.actions: 

197 self._cache_timetable_hash() 

198 return hex_id(self.__dict__["_timetable_hash"]) 

199 return Solution.hash_action_list(self.actions) 

200 

201 @staticmethod 

202 def hash_action_list(actions: list["BaseAction"]) -> str: 

203 """Hash a list of actions.""" 

204 if not actions: 

205 return "" 

206 return hash_string(dumps([a.id for a in actions])) 

207 

208 @staticmethod 

209 def from_parent(parent: "Solution", action: "BaseAction") -> "Solution": 

210 """Create a new solution from a parent solution. 

211 

212 Will automatically apply the action to the parent state, 

213 and evaluate the new state. 

214 """ 

215 new_state = action.apply(parent.state, enable_prints=False) 

216 # If the action did not change the state, we mark the solution as invalid/empty 

217 # This is due to the fact that many actions will not change the state, if 

218 # the action is not valid. 

219 if new_state == parent.state: 

220 return Solution.empty_from_parent(parent, action) 

221 evaluation = new_state.evaluate() 

222 return Solution( 

223 evaluation=evaluation, 

224 state=new_state, 

225 actions=parent.actions + [action], 

226 ) 

227 

228 @staticmethod 

229 def empty(state: State, last_action: Optional["BaseAction"] = None) -> "Solution": 

230 """Create an empty solution.""" 

231 return Solution( 

232 evaluation=Evaluation.empty(), 

233 state=state, 

234 actions=[last_action] if last_action else [], 

235 ) 

236 

237 @staticmethod 

238 def empty_from_parent(parent: "Solution", last_action: Optional["BaseAction"] = None) -> "Solution": 

239 """Create an empty solution from a parent solution.""" 

240 return Solution( 

241 evaluation=Evaluation.empty(), 

242 state=parent.state, 

243 actions=(parent.actions + [last_action]) if last_action else parent.actions, 

244 )