Coverage for o2/models/constraints/constraints_type.py: 90%
59 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-16 11:18 +0000
1from dataclasses import dataclass, field
2from typing import TYPE_CHECKING, Optional, Union
4from dataclass_wizard import JSONWizard
6from o2.models.constraints.batching_constraints import BatchingConstraints
7from o2.models.constraints.daily_hour_rule_constraints import (
8 DailyHourRuleConstraints,
9 is_daily_hour_constraint,
10)
11from o2.models.constraints.large_wt_rule_constraints import (
12 LargeWtRuleConstraints,
13 is_large_wt_constraint,
14)
15from o2.models.constraints.ready_wt_rule_constraints import (
16 ReadyWtRuleConstraints,
17 is_ready_wt_constraint,
18)
19from o2.models.constraints.size_rule_constraints import (
20 SizeRuleConstraints,
21 is_size_constraint,
22)
23from o2.models.constraints.week_day_rule_constraints import (
24 WeekDayRuleConstraints,
25 is_week_day_constraint,
26)
27from o2.models.legacy_constraints import ConstraintsResourcesItem, ResourceConstraints
28from o2.util.helper import name_is_clone_of
30if TYPE_CHECKING:
31 from o2.models.timetable import TimetableType
34@dataclass(frozen=True)
35class ConstraintsType(JSONWizard):
36 """Global Constraints Type including resource timetable and batching constraints."""
38 # TODO: Add more constraints here
39 batching_constraints: list[
40 Union[
41 SizeRuleConstraints,
42 ReadyWtRuleConstraints,
43 LargeWtRuleConstraints,
44 WeekDayRuleConstraints,
45 DailyHourRuleConstraints,
46 ]
47 ] = field(default_factory=list)
48 # Legacy Optimos constraints
49 resources: list[ConstraintsResourcesItem] = field(default_factory=list)
50 """Legacy Optimos Constraint: Resource Constraints"""
51 max_cap: int = 9999999
52 """Legacy Optimos Constraint: Max number of hours a resource can work in a week"""
53 max_shift_size: int = 24
54 """Legacy Optimos Constraint: Max number of hours in a continuous shift block"""
55 max_shift_blocks: int = 24
56 """Legacy Optimos Constraint: Max number of continuos shift block in a day"""
57 hours_in_day: int = 24
58 """Legacy Optimos Constraint: Hours/Slots per day"""
59 time_var: int = 60
60 """Legacy Optimos Constraint: Slot duration in minutes"""
62 class _(JSONWizard.Meta): # noqa: N801
63 key_transform_with_dump = "SNAKE"
64 tag_key = "rule_type"
66 def verify_legacy_constraints(self, timetable: "TimetableType") -> bool:
67 """Check if the timetable is valid against the constraints.
69 Will check resource constraints for all resources as well as the base constraints.
70 """
71 return (
72 timetable.max_total_hours_per_resource <= self.max_cap
73 and timetable.max_consecutive_hours_per_resource <= self.max_shift_size
74 and timetable.max_periods_per_day_per_resource <= self.max_shift_blocks
75 and all(
76 resource_constraints.verify_timetable(timetable) for resource_constraints in self.resources
77 )
78 )
80 def verify_batching_constraints(self, timetable: "TimetableType") -> bool:
81 """Check if the timetable is valid against the batching constraints.
83 Will check batching constraints for all firing rules.
84 """
85 return all(constraint.verify_timetable(timetable) for constraint in self.batching_constraints)
87 def get_legacy_constraints_for_resource(self, resource_id: str) -> Optional[ResourceConstraints]:
88 """Get the legacy constraints for a specific resource."""
89 return next(
90 (
91 constraint.constraints
92 for constraint in self.resources
93 if (
94 constraint.id == resource_id
95 or constraint.id == (resource_id + "timetable")
96 or name_is_clone_of(resource_id, constraint.id)
97 )
98 ),
99 None,
100 )
102 def get_batching_constraints_for_task(self, task: str) -> list[BatchingConstraints]:
103 """Get the batching constraints for a specific task."""
104 return [constraint for constraint in self.batching_constraints if task in constraint.tasks]
106 def get_batching_size_rule_constraints(self, task_id: str) -> list[SizeRuleConstraints]:
107 """Get the size rule constraints for a specific task."""
108 return [
109 constraint
110 for constraint in self.batching_constraints
111 if task_id in constraint.tasks and is_size_constraint(constraint)
112 ]
114 def get_batching_ready_wt_rule_constraints(self, task_id: str) -> list[ReadyWtRuleConstraints]:
115 """Get the ready waiting time rule constraints for a specific task."""
116 return [
117 constraint
118 for constraint in self.batching_constraints
119 if task_id in constraint.tasks and is_ready_wt_constraint(constraint)
120 ]
122 def get_batching_large_wt_rule_constraints(self, task_id: str) -> list[LargeWtRuleConstraints]:
123 """Get the large waiting time rule constraints for a specific task."""
124 return [
125 constraint
126 for constraint in self.batching_constraints
127 if task_id in constraint.tasks and is_large_wt_constraint(constraint)
128 ]
130 def get_week_day_rule_constraints(self, task_id: str) -> list[WeekDayRuleConstraints]:
131 """Get the week day rule constraints for a specific task."""
132 return [
133 constraint
134 for constraint in self.batching_constraints
135 if task_id in constraint.tasks and is_week_day_constraint(constraint)
136 ]
138 def get_daily_hour_rule_constraints(self, task_id: str) -> list[DailyHourRuleConstraints]:
139 """Get the daily hour rule constraints for a specific task."""
140 return [
141 constraint
142 for constraint in self.batching_constraints
143 if task_id in constraint.tasks and is_daily_hour_constraint(constraint)
144 ]
146 def get_fixed_cost_fn_for_task(self, task_id: str) -> str:
147 """Get the fixed cost function for a specific task."""
148 size_constraint = self.get_batching_size_rule_constraints(task_id)
149 if not size_constraint:
150 return "0"
151 first_constraint = size_constraint[0]
152 return first_constraint.cost_fn
154 def get_duration_fn_for_task(self, task_id: str) -> str:
155 """Get the duration function for a specific task."""
156 size_constraint = self.get_batching_size_rule_constraints(task_id)
157 if not size_constraint:
158 return "1"
159 first_constraint = size_constraint[0]
160 return first_constraint.duration_fn