Coverage for o2/actions/batching_actions/modify_daily_hour_rule_action.py: 84%

1from dataclasses import replace 

2 

3from typing_extensions import override 

4 

5from o2.actions.base_actions.base_action import RateSelfReturnType 

6from o2.actions.base_actions.batching_rule_base_action import ( 

7 BatchingRuleBaseAction, 

8 BatchingRuleBaseActionParamsType, 

9) 

10from o2.models.rule_selector import RuleSelector 

11from o2.models.self_rating import RATING 

12from o2.models.solution import Solution 

13from o2.models.state import State 

14from o2.models.timetable import rule_is_daily_hour 

15from o2.store import Store 

16from o2.util.helper import select_variants 

17from o2.util.logger import warn 

18 

19SIZE_OF_CHANGE = 1 

20CLOSENESS_TO_MAX_WT = 0.01 

21 

22 

23class ModifyDailyHourRuleActionParamsType(BatchingRuleBaseActionParamsType): 

24 """Parameter for ModifyDailyHourRuleAction. 

25 

26 hour_increment may also be negative to remove hours. 

27 """ 

28 

29 hour_increment: int 

30 

31 

32class ModifyDailyHourRuleAction(BatchingRuleBaseAction, str=False): 

33 """ModifyDailyHourRuleAction will add a new day to the firing rules of a BatchingRule. 

34 

35 It does this by cloning all the surrounding (AND) `FiringRule`s of 

36 the selected `FiringRule` and add one clone per `add_days` day to the BatchingRule. 

37 

38 Why are we not also removing weekdays? This would result in simply removing the 

39 firing rule, which is already implemented in RemoveRuleAction. 

40 """ 

41 

42 params: ModifyDailyHourRuleActionParamsType 

43 

44 @override 

45 def apply(self, state: State, enable_prints: bool = True) -> State: 

46 timetable = state.timetable 

47 rule_selector = self.params["rule"] 

48 hour_increment = self.params["hour_increment"] 

49 duration_fn = self.params.get("duration_fn", None) 

50 index, rule = timetable.get_batching_rule(rule_selector) 

51 if rule is None or index is None: 

52 warn(f"BatchingRule not found for {rule_selector}") 

53 return state 

54 

55 firing_rule = rule.get_firing_rule(rule_selector) 

56 if not rule_is_daily_hour(firing_rule): 

57 warn(f"Firing rule not found for {rule_selector}") 

58 return state 

59 

60 if hour_increment == 0: 

61 warn(f"No change in hours for {str(self)}") 

62 return state 

63 

64 new_hour = int(firing_rule.value) + hour_increment 

65 if (new_hour < 0) or (new_hour > 24): 

66 warn(f"Hour out of bounds ({new_hour}) for {str(self)}") 

67 return state 

68 

69 assert rule_selector.firing_rule_index is not None 

70 

71 new_batching_rule = rule.replace_firing_rule( 

72 rule_selector, 

73 replace(firing_rule, value=new_hour), 

74 duration_fn=duration_fn, 

75 ) 

76 

77 return state.replace_timetable( 

78 batch_processing=timetable.batch_processing[:index] 

79 + [new_batching_rule] 

80 + timetable.batch_processing[index + 1 :], 

81 ) 

82 

83 @override 

84 @staticmethod 

85 def rate_self(store: Store, input: Solution) -> RateSelfReturnType: 

86 selectors = [ 

87 RuleSelector( 

88 batching_rule_task_id=batching_rule.task_id, 

89 firing_rule_index=(or_rule_index, and_rule_index), 

90 ) 

91 for batching_rule in input.timetable.batch_processing 

92 for or_rule_index, or_rules in enumerate(batching_rule.firing_rules) 

93 for and_rule_index, rule in enumerate(or_rules) 

94 if rule_is_daily_hour(rule) and not rule.is_eq 

95 ] 

96 for selector in select_variants(store, selectors): 

97 for hour_increment in select_variants(store, [1, -1], inner=True): 

98 yield ( 

99 RATING.LOW, 

100 ModifyDailyHourRuleAction( 

101 ModifyDailyHourRuleActionParamsType( 

102 rule=selector, 

103 hour_increment=hour_increment, 

104 duration_fn=store.constraints.get_duration_fn_for_task( 

105 selector.batching_rule_task_id 

106 ), 

107 ), 

108 ), 

109 ) 

110 

111 return