Coverage for o2/models/timetable/firing_rule.py: 94%

1from dataclasses import asdict, dataclass 

2from json import dumps 

3from typing import Generic, Optional, TypeGuard, TypeVar 

4 

5from dataclass_wizard import JSONWizard 

6 

7from o2.models.days import DAY 

8from o2.models.timetable.comparator import COMPARATOR 

9from o2.models.timetable.rule_type import RULE_TYPE 

10from o2.util.helper import hash_string 

11 

12V = TypeVar("V", DAY, int) 

13 

14 

15@dataclass(frozen=True, eq=True, order=True) 

16class FiringRule(JSONWizard, Generic[V]): 

17 """A rule for when to fire (activate) a batching rule.""" 

18 

19 attribute: RULE_TYPE 

20 comparison: COMPARATOR 

21 value: V 

22 

23 def __eq__(self, other: object) -> bool: 

24 """Check if two rules are equal.""" 

25 return ( 

26 isinstance(other, FiringRule) 

27 and self.attribute == other.attribute 

28 and self.comparison == other.comparison 

29 and self.value == other.value 

30 ) 

31 

32 def id(self) -> str: 

33 """Get a thread-safe id for the rule. 

34 

35 We need to use this and not hash, because hash will not give you the same result on different threads. 

36 """ 

37 return hash_string(dumps(asdict(self))) 

38 

39 @property 

40 def is_gte(self) -> bool: 

41 """Check if the rule is a greater than or equal rule.""" 

42 return self.comparison == COMPARATOR.GREATER_THEN_OR_EQUAL 

43 

44 @property 

45 def is_lt(self) -> bool: 

46 """Check if the rule is a less than rule.""" 

47 return self.comparison == COMPARATOR.LESS_THEN 

48 

49 @property 

50 def is_eq(self) -> bool: 

51 """Check if the rule is an equal rule.""" 

52 return self.comparison == COMPARATOR.EQUAL 

53 

54 @property 

55 def is_lte(self) -> bool: 

56 """Check if the rule is a less than or equal rule.""" 

57 return self.comparison == COMPARATOR.LESS_THEN_OR_EQUAL 

58 

59 @property 

60 def is_gt(self) -> bool: 

61 """Check if the rule is a greater than rule.""" 

62 return self.comparison == COMPARATOR.GREATER_THEN 

63 

64 @property 

65 def is_gt_or_gte(self) -> bool: 

66 """Check if the rule is a greater than or equal rule.""" 

67 return self.is_gt or self.is_gte 

68 

69 @property 

70 def is_lt_or_lte(self) -> bool: 

71 """Check if the rule is a less than or equal rule.""" 

72 return self.is_lt or self.is_lte 

73 

74 @staticmethod 

75 def eq(attribute: RULE_TYPE, value: V) -> "FiringRule[V]": 

76 """Create a FiringRule with an EQUAL comparison. 

77 

78 Creates a rule that checks if an attribute equals the given value. 

79 """ 

80 return FiringRule(attribute=attribute, comparison=COMPARATOR.EQUAL, value=value) 

81 

82 @staticmethod 

83 def gte(attribute: RULE_TYPE, value: V) -> "FiringRule[V]": 

84 """Create a FiringRule with a GREATER_THAN_OR_EQUAL comparison. 

85 

86 Creates a rule that checks if an attribute is greater than or equal to the given value. 

87 """ 

88 return FiringRule( 

89 attribute=attribute, 

90 comparison=COMPARATOR.GREATER_THEN_OR_EQUAL, 

91 value=value, 

92 ) 

93 

94 @staticmethod 

95 def lt(attribute: RULE_TYPE, value: V) -> "FiringRule[V]": 

96 """Create a FiringRule with a LESS_THAN comparison. 

97 

98 Creates a rule that checks if an attribute is less than the given value. 

99 """ 

100 return FiringRule(attribute=attribute, comparison=COMPARATOR.LESS_THEN, value=value) 

101 

102 @staticmethod 

103 def lte(attribute: RULE_TYPE, value: V) -> "FiringRule[V]": 

104 """Create a FiringRule with a LESS_THAN_OR_EQUAL comparison. 

105 

106 Creates a rule that checks if an attribute is less than or equal to the given value. 

107 """ 

108 return FiringRule(attribute=attribute, comparison=COMPARATOR.LESS_THEN_OR_EQUAL, value=value) 

109 

110 @staticmethod 

111 def gt(attribute: RULE_TYPE, value: V) -> "FiringRule[V]": 

112 """Create a FiringRule with a GREATER_THAN comparison. 

113 

114 Creates a rule that checks if an attribute is greater than the given value. 

115 """ 

116 return FiringRule(attribute=attribute, comparison=COMPARATOR.GREATER_THEN, value=value) 

117 

118 

119AndRules = list[FiringRule] 

120OrRules = list[AndRules] 

121 

122 

123def rule_is_large_wt(rule: Optional[FiringRule]) -> TypeGuard[FiringRule[int]]: 

124 """Check if a rule is a large waiting time rule.""" 

125 return rule is not None and rule.attribute == RULE_TYPE.LARGE_WT 

126 

127 

128def rule_is_week_day(rule: Optional[FiringRule]) -> TypeGuard[FiringRule[DAY]]: 

129 """Check if a rule is a week day rule.""" 

130 return rule is not None and rule.attribute == RULE_TYPE.WEEK_DAY 

131 

132 

133def rule_is_size(rule: Optional[FiringRule]) -> TypeGuard[FiringRule[int]]: 

134 """Check if a rule is a size rule.""" 

135 return rule is not None and rule.attribute == RULE_TYPE.SIZE 

136 

137 

138def rule_is_ready_wt(rule: Optional[FiringRule]) -> TypeGuard[FiringRule[int]]: 

139 """Check if a rule is a ready waiting time rule.""" 

140 return rule is not None and rule.attribute == RULE_TYPE.READY_WT 

141 

142 

143def rule_is_daily_hour(rule: Optional[FiringRule]) -> TypeGuard[FiringRule[int]]: 

144 """Check if a rule is a daily hour rule.""" 

145 return rule is not None and rule.attribute == RULE_TYPE.DAILY_HOUR