Coverage for o2/models/legacy_constraints.py: 100%

78 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-05-16 11:18 +0000

1import operator 

2from dataclasses import dataclass, replace 

3from functools import reduce 

4from typing import TYPE_CHECKING, Optional 

5 

6from dataclass_wizard import JSONWizard 

7 

8from o2.models.days import DAY 

9 

10if TYPE_CHECKING: 

11 from o2.models.timetable import ResourceCalendar, TimetableType 

12 

13 

14@dataclass(frozen=True) 

15class WorkMasks(JSONWizard): 

16 """Bitmask per day.""" 

17 

18 monday: Optional[int] = 0 

19 tuesday: Optional[int] = 0 

20 wednesday: Optional[int] = 0 

21 thursday: Optional[int] = 0 

22 friday: Optional[int] = 0 

23 saturday: Optional[int] = 0 

24 sunday: Optional[int] = 0 

25 

26 def has_hour_for_day(self, day: DAY, hour: int) -> bool: 

27 """Check if the hour is enabled for a specific day.""" 

28 return bool(self.get(day) & (1 << (23 - hour))) 

29 

30 def get(self, day: DAY) -> int: 

31 """Get the mask for a specific day.""" 

32 return getattr(self, day.name.lower()) or 0 

33 

34 def has_intersection(self, calendar: "ResourceCalendar") -> bool: 

35 """Check if the calendar has an overlap (intersection) with the work masks.""" 

36 for day, periods in calendar.split_group_by_day(): 

37 masks = [period.to_bitmask() for period in periods] 

38 day_mask = reduce(operator.or_, masks) 

39 if day_mask & self.get(day): 

40 return True 

41 return False 

42 

43 def is_super_set(self, calendar: "ResourceCalendar") -> bool: 

44 """Check if the work masks are a super set of the calendar.""" 

45 for day, periods in calendar.split_group_by_day(): 

46 masks = [period.to_bitmask() for period in periods] 

47 day_mask = reduce(operator.or_, masks) 

48 if day_mask & ~self.get(day): 

49 return False 

50 return True 

51 

52 def is_subset(self, calendar: "ResourceCalendar") -> bool: 

53 """Check if the work masks are a subset of the calendar.""" 

54 for day, periods in calendar.split_group_by_day(): 

55 masks = [period.to_bitmask() for period in periods] 

56 day_mask = reduce(operator.or_, masks) 

57 if ~day_mask & self.get(day): 

58 return False 

59 return True 

60 

61 def set_hour_for_day(self, day: DAY, hour: int) -> "WorkMasks": 

62 """Enable the hour for a specific day.""" 

63 return replace(self, **{day.name.lower(): self.get(day) | (1 << (23 - hour))}) 

64 

65 def set_hour_for_every_day(self, hour: int) -> "WorkMasks": 

66 """Enable the hour for every day.""" 

67 return replace( 

68 self, 

69 monday=(self.monday or 0) | (1 << (23 - hour)), 

70 tuesday=(self.tuesday or 0) | (1 << (23 - hour)), 

71 wednesday=(self.wednesday or 0) | (1 << (23 - hour)), 

72 thursday=(self.thursday or 0) | (1 << (23 - hour)), 

73 friday=(self.friday or 0) | (1 << (23 - hour)), 

74 saturday=(self.saturday or 0) | (1 << (23 - hour)), 

75 sunday=(self.sunday or 0) | (1 << (23 - hour)), 

76 ) 

77 

78 def set_hour_range_for_day(self, day: DAY, start: int, end: int) -> "WorkMasks": 

79 """Enable the hour range for a specific day.""" 

80 return replace( 

81 self, 

82 **{day.name.lower(): self.get(day) | ((1 << (24 - start)) - (1 << (24 - end)))}, 

83 ) 

84 

85 def set_hour_range_for_every_day(self, start: int, end: int) -> "WorkMasks": 

86 """Enable the hour range for every day.""" 

87 return replace( 

88 self, 

89 monday=(self.monday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

90 tuesday=(self.tuesday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

91 wednesday=(self.wednesday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

92 thursday=(self.thursday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

93 friday=(self.friday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

94 saturday=(self.saturday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

95 sunday=(self.sunday or 0) | ((1 << (24 - start)) - (1 << (24 - end))), 

96 ) 

97 

98 @staticmethod 

99 def all_day() -> "WorkMasks": 

100 """Enable all hours for every day.""" 

101 return WorkMasks( 

102 monday=0b111111111111111111111111, 

103 tuesday=0b111111111111111111111111, 

104 wednesday=0b111111111111111111111111, 

105 thursday=0b111111111111111111111111, 

106 friday=0b111111111111111111111111, 

107 saturday=0b111111111111111111111111, 

108 sunday=0b111111111111111111111111, 

109 ) 

110 

111 

112@dataclass(frozen=True) 

113class GlobalConstraints(JSONWizard): 

114 """'Global' constraints for a resource, independent of the day.""" 

115 

116 max_weekly_cap: float 

117 max_daily_cap: float 

118 max_consecutive_cap: float 

119 max_shifts_day: int 

120 max_shifts_week: float 

121 is_human: bool 

122 

123 def verify_timetable(self, calendar: "ResourceCalendar") -> bool: 

124 """Check if the timetable is valid against the constraints.""" 

125 return ( 

126 calendar.total_hours <= self.max_weekly_cap 

127 and calendar.max_hours_per_day <= self.max_daily_cap 

128 and calendar.max_consecutive_hours <= self.max_consecutive_cap 

129 and calendar.max_periods_per_day <= self.max_shifts_day 

130 and calendar.total_periods <= self.max_shifts_week 

131 ) 

132 

133 

134@dataclass(frozen=True) 

135class ResourceConstraints(JSONWizard): 

136 """Constraints for a resource.""" 

137 

138 global_constraints: GlobalConstraints 

139 never_work_masks: WorkMasks 

140 always_work_masks: WorkMasks 

141 

142 

143@dataclass(frozen=True) 

144class ConstraintsResourcesItem(JSONWizard): 

145 """Resource constraints for a specific resource.""" 

146 

147 id: str 

148 constraints: ResourceConstraints 

149 

150 def verify_timetable(self, timetable: "TimetableType") -> bool: 

151 """Check if the timetable is valid against the constraints. 

152 

153 NOTE: This assumes the calendar itself is valid (e.g. no overlapping periods). 

154 This should be checked before (e.g. see `calendar.is_valid()`). 

155 """ 

156 original_calendar = timetable.get_calendar_for_resource(self.id) 

157 calendars = timetable.get_calendars_for_resource_clones(self.id) 

158 if original_calendar is not None: 

159 calendars = [original_calendar, *calendars] 

160 return all(self._verify_calendar(calendar) for calendar in calendars) 

161 

162 def _verify_calendar(self, calendar: "ResourceCalendar") -> bool: 

163 return ( 

164 self.constraints.global_constraints.verify_timetable(calendar) 

165 and not self.constraints.never_work_masks.has_intersection(calendar) 

166 and self.constraints.always_work_masks.is_subset(calendar) 

167 )