Coverage for o2/models/json_report.py: 100%
131 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-16 11:18 +0000
1import datetime
2from dataclasses import dataclass
3from typing import Any, Optional
5from dataclass_wizard import JSONWizard
6from pydantic import BaseModel
8from o2.actions.base_actions.base_action import BaseAction
9from o2.actions.base_actions.modify_calendar_base_action import ModifyCalendarBaseAction
10from o2.actions.base_actions.modify_resource_base_action import ModifyResourceBaseAction
11from o2.models.constraints import ConstraintsType
12from o2.models.legacy_constraints import WorkMasks
13from o2.models.settings import CostType
14from o2.models.solution import Solution
15from o2.models.timetable import Resource, TimetableType
16from o2.pareto_front import ParetoFront
17from o2.store import Store
18from o2.util.helper import CLONE_REGEX
21class JSONReport(BaseModel):
22 """Class to represent report in JSON format."""
24 name: str
25 created_at: str
26 """Time when the report was created."""
28 constraints: ConstraintsType
29 bpmn_definition: str
30 task_names: dict[str, str]
31 """Mapping of task IDs to task names."""
33 base_solution: "_JSONSolution"
34 pareto_fronts: list["_JSONParetoFront"]
36 is_final: bool
38 approach: Optional[str] = None
39 cost_type: CostType
40 """The cost type used for optimization."""
42 @staticmethod
43 def from_store(store: Store, is_final: bool = False) -> "JSONReport":
44 """Create a JSONReport from a Store instance.
46 Constructs a complete report containing all solutions and pareto fronts
47 from the given store.
48 """
49 return JSONReport(
50 name=store.name,
51 constraints=store.constraints,
52 bpmn_definition=store.base_state.bpmn_definition,
53 task_names=store.base_state.get_task_names(),
54 base_solution=_JSONSolution.from_state_evaluation(store.base_solution, store),
55 pareto_fronts=[
56 _JSONParetoFront.from_pareto_front(pareto_front, store)
57 for pareto_front in store.pareto_fronts
58 ],
59 is_final=is_final,
60 approach=str(store.settings.legacy_approach),
61 cost_type=store.settings.COST_TYPE,
62 created_at=datetime.datetime.now().isoformat(),
63 )
65 class Config:
66 """pydantic config for JSONReport."""
68 frozen = True
69 ser_json_inf_nan = "strings"
70 allow_inf_nan = True
73@dataclass(frozen=True)
74class _JSONParetoFront(JSONWizard):
75 solutions: list["_JSONSolution"]
77 @staticmethod
78 def from_pareto_front(pareto_front: "ParetoFront", store: Store) -> "_JSONParetoFront":
79 return _JSONParetoFront(
80 solutions=[
81 _JSONSolution.from_state_evaluation(solution, store) for solution in pareto_front.solutions
82 ]
83 )
86@dataclass
87class _JSONResourceModifiers(JSONWizard):
88 deleted: Optional[bool]
89 added: Optional[bool]
90 shifts_modified: Optional[bool]
91 tasks_modified: Optional[bool]
94@dataclass(frozen=True)
95class _JSONResourceInfo(JSONWizard):
96 id: str
97 name: str
99 worked_time: float
100 available_time: float
101 utilization: float
102 cost_per_week: float
103 total_cost: float
104 hourly_rate: float
105 is_human: bool
106 max_weekly_capacity: float
107 max_daily_capacity: float
108 max_consecutive_capacity: float
109 # Timetable
110 timetable_bitmask: WorkMasks
111 original_timetable_bitmask: WorkMasks
112 work_hours_per_week: int
113 never_work_bitmask: WorkMasks
114 always_work_bitmask: WorkMasks
115 # Tasks
116 assigned_tasks: list[str]
117 added_tasks: list[str]
118 removed_tasks: list[str]
120 # Batching
121 total_batching_waiting_time: float
123 modifiers: "_JSONResourceModifiers"
125 @staticmethod
126 def from_resource(
127 resource: Resource,
128 solution: Solution,
129 store: Store,
130 ) -> "_JSONResourceInfo":
131 timetable = solution.state.timetable.get_calendar(
132 resource.calendar
133 ) or store.base_state.timetable.get_calendar_for_base_resource(resource.id)
134 resource_constraints = store.constraints.get_legacy_constraints_for_resource(resource.id)
135 if solution.is_base_solution:
136 original_time_table = timetable
137 else:
138 original_time_table = store.base_state.timetable.get_calendar_for_base_resource(resource.id)
140 assert timetable is not None
141 assert resource_constraints is not None
142 assert original_time_table is not None
144 relevant_actions = [
145 action
146 for action in solution.actions
147 if (isinstance(action, ModifyResourceBaseAction) and action.params["resource_id"] == resource.id)
148 or (isinstance(action, ModifyCalendarBaseAction) and action.params["calendar_id"] == timetable.id)
149 ]
151 deleted = any(
152 isinstance(action, ModifyResourceBaseAction) and action.params.get("remove_resource")
153 for action in relevant_actions
154 )
156 added = CLONE_REGEX.match(resource.id) is not None
158 shifts_modified = any(isinstance(action, ModifyCalendarBaseAction) for action in relevant_actions)
160 tasks_modified = any(
161 isinstance(action, ModifyResourceBaseAction) and action.params.get("remove_task_from_resource")
162 for action in relevant_actions
163 )
165 original_tasks = store.base_state.timetable.get_tasks(resource.id)
166 removed_tasks = [task_id for task_id in original_tasks if task_id not in resource.assigned_tasks]
167 added_tasks = [task_id for task_id in resource.assigned_tasks if task_id not in original_tasks]
169 evaluation = solution.evaluation
171 return _JSONResourceInfo(
172 id=resource.id,
173 name=resource.name,
174 worked_time=evaluation.resource_worked_times.get(resource.id, 0),
175 work_hours_per_week=timetable.total_hours,
176 available_time=evaluation.resource_available_times.get(resource.id, 0),
177 utilization=evaluation.resource_utilizations.get(resource.id, 0),
178 cost_per_week=timetable.total_hours * resource.cost_per_hour,
179 total_cost=(evaluation.resource_available_times.get(resource.id, 0) / 60 / 60)
180 * resource.cost_per_hour,
181 hourly_rate=resource.cost_per_hour,
182 is_human=resource_constraints.global_constraints.is_human,
183 max_weekly_capacity=resource_constraints.global_constraints.max_weekly_cap,
184 max_daily_capacity=resource_constraints.global_constraints.max_daily_cap,
185 max_consecutive_capacity=resource_constraints.global_constraints.max_consecutive_cap,
186 original_timetable_bitmask=original_time_table.work_masks,
187 timetable_bitmask=timetable.work_masks,
188 never_work_bitmask=resource_constraints.never_work_masks,
189 always_work_bitmask=resource_constraints.always_work_masks,
190 assigned_tasks=resource.assigned_tasks,
191 modifiers=_JSONResourceModifiers(
192 deleted=deleted,
193 added=added,
194 shifts_modified=shifts_modified,
195 tasks_modified=tasks_modified,
196 ),
197 added_tasks=added_tasks,
198 removed_tasks=removed_tasks,
199 total_batching_waiting_time=evaluation.total_batching_waiting_time_per_resource.get(
200 resource.id, 0
201 ),
202 )
205@dataclass(frozen=True)
206class _JSONGlobalInfo(JSONWizard):
207 average_cost: float
208 average_time: float
209 average_resource_utilization: float
210 total_cost: float
211 total_time: float
212 average_batching_waiting_time: float
213 average_waiting_time: float
214 total_fixed_cost: float
215 total_cost_for_available_time: float
216 total_cost_for_worked_time: float
217 total_processing_time: float
218 avg_batch_processing_time_per_task_instance: float
219 total_waiting_time: float
220 total_task_idle_time: float
221 avg_idle_wt_per_task_instance: float
224@dataclass(frozen=True)
225class _JSONAction(JSONWizard):
226 """Class to represent action in JSON format."""
228 type: str
229 params: Any
231 @staticmethod
232 def from_action(action: "BaseAction") -> "_JSONAction":
233 return _JSONAction(action.__class__.__name__, action.params)
236@dataclass(frozen=True)
237class _JSONSolution(JSONWizard):
238 is_base_solution: bool
239 solution_no: int
240 global_info: "_JSONGlobalInfo"
241 resource_info: dict[str, "_JSONResourceInfo"]
242 deleted_resources_info: dict[str, "_JSONResourceInfo"]
243 timetable: TimetableType
245 actions: list[_JSONAction]
247 @staticmethod
248 def from_state_evaluation(solution: Solution, store: Store) -> "_JSONSolution":
249 state = solution.state
250 evaluation = solution.evaluation
251 return _JSONSolution(
252 timetable=state.timetable,
253 is_base_solution=solution.is_base_solution,
254 solution_no=store.solution_tree.get_index_of_solution(solution),
255 global_info=_JSONGlobalInfo(
256 average_cost=evaluation.avg_cost_by_case,
257 average_time=evaluation.avg_cycle_time_by_case,
258 average_resource_utilization=evaluation.avg_resource_utilization_by_case,
259 total_cost=evaluation.total_cost,
260 total_time=evaluation.total_duration,
261 average_batching_waiting_time=evaluation.avg_batching_waiting_time_by_case,
262 average_waiting_time=evaluation.avg_waiting_time_by_case,
263 total_fixed_cost=evaluation.total_fixed_cost,
264 total_cost_for_available_time=evaluation.total_cost_for_available_time,
265 total_cost_for_worked_time=evaluation.total_cost_for_worked_time,
266 total_processing_time=evaluation.total_processing_time,
267 avg_batch_processing_time_per_task_instance=evaluation.avg_batch_processing_time_per_task_instance,
268 total_waiting_time=evaluation.total_waiting_time,
269 total_task_idle_time=evaluation.total_task_idle_time,
270 avg_idle_wt_per_task_instance=evaluation.avg_idle_wt_per_task_instance,
271 ),
272 resource_info={
273 resource.id: _JSONResourceInfo.from_resource(resource, solution, store)
274 for resource in state.timetable.get_all_resources()
275 },
276 deleted_resources_info={
277 resource.id: _JSONResourceInfo.from_resource(resource, solution, store)
278 for resource in state.timetable.get_deleted_resources(store.base_state)
279 },
280 actions=[_JSONAction.from_action(action) for action in solution.actions], # type: ignore
281 )