Coverage for o2/store.py: 91%
99 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-16 11:18 +0000
1from dataclasses import replace
2from typing import TYPE_CHECKING, Callable, Optional, TypeAlias
4from o2.models.constraints import ConstraintsType
5from o2.models.evaluation import Evaluation
6from o2.models.settings import Settings
7from o2.models.solution import Solution
8from o2.models.solution_tree import SolutionTree
9from o2.models.state import State
10from o2.pareto_front import FRONT_STATUS, ParetoFront
11from o2.util.indented_printer import print_l2
13if TYPE_CHECKING:
14 from o2.actions.base_actions.base_action import BaseAction
15 from o2.models.timetable import TimetableType
17SolutionTry: TypeAlias = tuple[FRONT_STATUS, Solution]
20class Store:
21 """The Store class is the main class that holds the state of the application.
23 It holds the current state, the constraints, the previous actions and states,
24 the Pareto Fronts, the Tabu List and the Settings.
25 """
27 def __init__(
28 self,
29 solution: Solution,
30 constraints: ConstraintsType,
31 name: str = "An Optimos Run",
32 ) -> None:
33 self.name = name
35 self.constraints = constraints
37 # This is the base Pareto Front
38 self.pareto_fronts: list[ParetoFront] = [ParetoFront()]
39 self.pareto_fronts[0].add(solution)
41 # Add another Pareto Front for the current iteration
42 # We need to do that, as the base solution might otherwise be dominated
43 # by a new solution and thereby self.pareto_fronts[0].solutions[0]
44 # will be changed.
45 self.pareto_fronts.append(ParetoFront())
46 self.pareto_fronts[-1].add(solution)
48 self.solution_tree = SolutionTree()
49 self.solution_tree.add_solution(solution, archive=False)
51 self.solution = solution
52 """The current solution of the optimization process.
54 It will be updated after any change to the pareto front or after a iteration.
55 """
57 self.settings: Settings = Settings()
59 @property
60 def current_pareto_front(self) -> ParetoFront:
61 """Returns the current Pareto Front."""
62 return self.pareto_fronts[-1]
64 @property
65 def base_solution(self) -> Solution:
66 """Returns the base solution, e.g. the fist solution with no changes."""
67 return self.pareto_fronts[0].solutions[0]
69 @property
70 def base_evaluation(self) -> Evaluation:
71 """Returns the base evaluation, e.g. the fist evaluation with no changes."""
72 return self.base_solution.evaluation
74 @property
75 def base_state(self) -> State:
76 """Returns the base state, e.g. the state before any changes."""
77 return self.base_solution.state
79 @property
80 def base_timetable(self) -> "TimetableType":
81 """Returns the base timetable, e.g. the timetable before any changes."""
82 return self.base_state.timetable
84 @property
85 def current_evaluation(self) -> Evaluation:
86 """Returns the current evaluation of the solution."""
87 return self.solution.evaluation
89 @property
90 def current_timetable(self) -> "TimetableType":
91 """Return the current timetable of the solution."""
92 return self.solution.state.timetable
94 @property
95 def current_state(self) -> State:
96 """Return the current state of the solution."""
97 return self.solution.state
99 def mark_action_as_tabu(self, action: "BaseAction") -> None:
100 """Mark an action as tabu."""
101 solution = Solution(
102 evaluation=Evaluation.empty(),
103 state=self.solution.state,
104 actions=[*self.solution.actions, action],
105 )
107 self.solution_tree.add_solution(solution)
109 def process_many_solutions(
110 self,
111 solutions: list[Solution],
112 set_new_base_evaluation_callback: Optional[Callable[[SolutionTry], None]] = None,
113 ) -> tuple[list[SolutionTry], list[SolutionTry]]:
114 """Process a list of action solutions.
116 Ignores the solutions that are dominated by the current Pareto Front.
117 Returns two lists, one with the chosen actions and one with the not chosen actions.
119 This is useful if multiple actions are evaluated at once.
121 The set_new_base_evaluation_callback is called when a new base potential base_solution
122 is found. It's assumed that the callback will update the store.solution attribute.
123 """
124 chosen_tries = []
125 not_chosen_tries = []
126 new_baseline_chosen = False
127 for solution in solutions:
128 status = self.current_pareto_front.is_in_front(solution)
130 if solution.is_valid:
131 # We directly archive the solutions that are not in the front
132 # because we most likely will not need them again.
133 should_archive = status != FRONT_STATUS.IN_FRONT and status != FRONT_STATUS.IS_DOMINATED
134 self.solution_tree.add_solution(solution, archive=should_archive)
135 else:
136 # If the solution is invalid, override the status to invalid
137 status = FRONT_STATUS.INVALID
138 self.solution_tree.add_solution_as_discarded(solution)
140 if status == FRONT_STATUS.IN_FRONT:
141 chosen_tries.append((status, solution))
142 self.current_pareto_front.add(solution)
143 elif status == FRONT_STATUS.IS_DOMINATED:
144 chosen_tries.append((status, solution))
145 self.pareto_fronts.append(ParetoFront())
146 self.current_pareto_front.add(solution)
147 # A dominating solution, will always override the current baseline
148 new_baseline_chosen = False
149 else:
150 not_chosen_tries.append((status, solution))
152 if not new_baseline_chosen and (
153 status == FRONT_STATUS.IN_FRONT or status == FRONT_STATUS.IS_DOMINATED
154 ):
155 if set_new_base_evaluation_callback is not None:
156 set_new_base_evaluation_callback((status, solution))
157 else:
158 self.solution = solution
159 new_baseline_chosen = True
161 return chosen_tries, not_chosen_tries
163 def run_action(self, action: "BaseAction") -> Optional[Solution]:
164 """Run an action and add the new solution to the store.
166 NOTE: Usually you would use the HillClimber to run actions.
167 This method should only be used in tests.
169 NOTE: This will only update the store if the action is not dominated
170 """
171 new_solution = Solution.from_parent(self.solution, action)
172 self.process_many_solutions([new_solution], None)
174 # Tries an action and returns the status of the new evaluation
175 # Does NOT modify the store
176 def try_solution(self, solution: "Solution") -> SolutionTry:
177 """Try an action and return the status of the new evaluation.
179 If the evaluation throws an exception, it returns IS_DOMINATED.
180 """
181 try:
182 if not solution.is_valid:
183 return (FRONT_STATUS.INVALID, solution)
184 if self.settings.disable_action_validity_check and (
185 not self.constraints.verify_batching_constraints(solution.state.timetable)
186 or not self.constraints.verify_legacy_constraints(solution.state.timetable)
187 ):
188 return (FRONT_STATUS.INVALID, solution)
189 status = self.current_pareto_front.is_in_front(solution)
190 return (status, solution)
191 except Exception as e:
192 print_l2(f"Error in try_action: {e}")
193 return (FRONT_STATUS.INVALID, solution)
195 def is_tabu(self, action: "BaseAction") -> bool:
196 """Check if the action is tabu."""
197 return self.solution_tree.check_if_already_done(self.solution, action)
199 @staticmethod
200 def from_state_and_constraints(
201 state: State, constraints: ConstraintsType, name: str = "An Optimos Run"
202 ) -> "Store":
203 """Create a new Store from a state and constraints."""
204 updated_state: State = replace(state, timetable=state.timetable.init_fixed_cost_fns(constraints))
205 evaluation = updated_state.evaluate()
206 solution = Solution(evaluation=evaluation, state=updated_state, actions=[])
207 return Store(solution, constraints, name)