Coverage for o2/actions/batching_actions/random_action.py: 99%

1import random 

2from typing import Union 

3 

4from typing_extensions import override 

5 

6from o2.actions.base_actions.add_datetime_rule_base_action import ( 

7 AddDateTimeRuleAction, 

8 AddDateTimeRuleBaseActionParamsType, 

9) 

10from o2.actions.base_actions.add_ready_large_wt_rule_base_action import ( 

11 AddReadyLargeWTRuleAction, 

12 AddReadyLargeWTRuleBaseActionParamsType, 

13) 

14from o2.actions.base_actions.add_size_rule_base_action import ( 

15 AddSizeRuleAction, 

16 AddSizeRuleBaseActionParamsType, 

17) 

18from o2.actions.base_actions.base_action import ( 

19 BaseAction, 

20 BaseActionParamsType, 

21 RateSelfReturnType, 

22) 

23from o2.actions.base_actions.modify_size_rule_base_action import ( 

24 ModifySizeRuleAction, 

25 ModifySizeRuleBaseActionParamsType, 

26) 

27from o2.actions.batching_actions.modify_daily_hour_rule_action import ( 

28 ModifyDailyHourRuleAction, 

29 ModifyDailyHourRuleActionParamsType, 

30) 

31from o2.actions.batching_actions.remove_rule_action import ( 

32 RemoveRuleAction, 

33 RemoveRuleActionParamsType, 

34) 

35from o2.models.days import DAYS 

36from o2.models.self_rating import RATING 

37from o2.models.settings import Settings 

38from o2.models.solution import Solution 

39from o2.models.timetable import RULE_TYPE 

40from o2.models.timetable.time_period import TimePeriod 

41from o2.store import Store 

42 

43ACTIONS: list[type[BaseAction]] = [ 

44 AddDateTimeRuleAction, 

45 AddReadyLargeWTRuleAction, 

46 AddSizeRuleAction, 

47 ModifyDailyHourRuleAction, 

48 ModifySizeRuleAction, 

49 RemoveRuleAction, 

50] 

51 

52 

53class RandomActionParamsType(BaseActionParamsType): 

54 """Parameter for RandomAction.""" 

55 

56 pass 

57 

58 

59class RandomAction(BaseAction): 

60 """RandomAction will randomly select an action and apply it to the timetable. 

61 

62 This is only used by the Random Agents. 

63 """ 

64 

65 params: RandomActionParamsType 

66 

67 @override 

68 @staticmethod 

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

70 timetable = input.timetable 

71 params: Union[BaseActionParamsType, None] = None 

72 

73 # We generate infinite actions 

74 while True: 

75 # Randomly select an action 

76 action = random.choice(ACTIONS) 

77 

78 if action == AddDateTimeRuleAction: 

79 task_id = random.choice(timetable.get_task_ids()) 

80 random_day = random.choice(DAYS) 

81 random_start_time = random.randint(0, 23) 

82 end_time = random_start_time + 1 

83 time_period = TimePeriod.from_start_end(random_start_time, end_time, random_day) 

84 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

85 

86 params = AddDateTimeRuleBaseActionParamsType( 

87 task_id=task_id, 

88 time_period=time_period, 

89 duration_fn=duration_fn, 

90 ) 

91 elif action == AddReadyLargeWTRuleAction: 

92 task_id = random.choice(timetable.get_task_ids()) 

93 waiting_time = random.randint(1, 24 * 60 * 60) 

94 rule_type = random.choice([RULE_TYPE.LARGE_WT, RULE_TYPE.READY_WT]) 

95 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

96 params = AddReadyLargeWTRuleBaseActionParamsType( 

97 task_id=task_id, 

98 waiting_time=waiting_time, 

99 type=rule_type, # type: ignore 

100 duration_fn=duration_fn, 

101 ) 

102 elif action == AddSizeRuleAction: 

103 task_id = random.choice(timetable.get_task_ids()) 

104 size = 2 

105 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

106 params = AddSizeRuleBaseActionParamsType( 

107 task_id=task_id, 

108 size=size, 

109 duration_fn=duration_fn, 

110 ) 

111 elif action == ModifyDailyHourRuleAction: 

112 all_daily_hour_rule_selectors = [ 

113 rule_selector 

114 for batching_rule in timetable.batch_processing 

115 for rule_selector in batching_rule.get_firing_rule_selectors(RULE_TYPE.DAILY_HOUR) 

116 ] 

117 if not all_daily_hour_rule_selectors: 

118 continue 

119 hour_increment = random.choice([-1, 1]) 

120 random_rule_selector = random.choice(all_daily_hour_rule_selectors) 

121 params = ModifyDailyHourRuleActionParamsType( 

122 rule=random_rule_selector, 

123 hour_increment=hour_increment, 

124 duration_fn=store.constraints.get_duration_fn_for_task( 

125 random_rule_selector.batching_rule_task_id 

126 ), 

127 ) 

128 elif action == ModifySizeRuleAction: 

129 all_size_rule_selectors = [ 

130 rule_selector 

131 for batching_rule in timetable.batch_processing 

132 for rule_selector in batching_rule.get_firing_rule_selectors(RULE_TYPE.SIZE) 

133 ] 

134 if not all_size_rule_selectors: 

135 continue 

136 random_rule_selector = random.choice(all_size_rule_selectors) 

137 task_id = random.choice(timetable.get_task_ids()) 

138 size_increment = random.choice([-1, 1]) 

139 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

140 params = ModifySizeRuleBaseActionParamsType( 

141 rule=random_rule_selector, 

142 size_increment=size_increment, 

143 duration_fn=duration_fn, 

144 ) 

145 elif action == RemoveRuleAction: 

146 if Settings.DISABLE_REMOVE_ACTION_RULE: 

147 continue 

148 all_rule_selectors = [ 

149 rule_selector 

150 for batching_rule in timetable.batch_processing 

151 for rule_selector in batching_rule.get_firing_rule_selectors() 

152 ] 

153 if not all_rule_selectors: 

154 continue 

155 random_rule_selector = random.choice(all_rule_selectors) 

156 duration_fn = store.constraints.get_duration_fn_for_task( 

157 random_rule_selector.batching_rule_task_id 

158 ) 

159 params = RemoveRuleActionParamsType(rule=random_rule_selector, duration_fn=duration_fn) 

160 else: 

161 raise ValueError(f"Unknown action: {action}") 

162 

163 yield (RATING.HIGH, action(params))