Coverage for o2/actions/batching_actions/add_large_wt_rule_by_idle_action.py: 95%

1from math import ceil 

2 

3from typing_extensions import override 

4 

5from o2.actions.base_actions.add_ready_large_wt_rule_base_action import ( 

6 AddReadyLargeWTRuleBaseAction, 

7 AddReadyLargeWTRuleBaseActionParamsType, 

8) 

9from o2.actions.base_actions.base_action import ( 

10 RateSelfReturnType, 

11) 

12from o2.models.days import DAY 

13from o2.models.solution import Solution 

14from o2.models.timetable import RULE_TYPE 

15from o2.store import Store 

16from o2.util.helper import select_variants 

17 

18 

19class AddLargeWTRuleByIdleActionParamsType(AddReadyLargeWTRuleBaseActionParamsType): 

20 """Parameter for AddLargeWTRuleByIdleAction.""" 

21 

22 pass 

23 

24 

25class AddLargeWTRuleByIdleAction(AddReadyLargeWTRuleBaseAction): 

26 """An Action to add a LargeWT rule based on the idle time of the task. 

27 

28 This action will add a new LargeWT rule based on the waiting time of the task. 

29 It does the following: 

30 1. Looks at all batches with idle time, grouped by their activity 

31 2. Calculate the average ideal_proc 

32 3. For each activity, look at all resources, which could also 

33 do that batch (incl. the one which originally did it) 

34 4. For each resource, find all time intervals/periods, that are are at 

35 least avg_ideal_proc long (only work-time, skipping idle time), 

36 and start after accumulation_begin. Up to a time after the actual 

37 batch start, that is at least avg_ideal_proc long. 

38 7. Create a new LargeWT rule for every of those periods 

39 

40 TODO: Use average and do not create for every single one 

41 """ 

42 

43 params: AddLargeWTRuleByIdleActionParamsType 

44 

45 @override 

46 @staticmethod 

47 def rate_self(store: "Store", input: "Solution") -> RateSelfReturnType["AddLargeWTRuleByIdleAction"]: 

48 timetable = input.state.timetable 

49 

50 # Group all batches by activity, only include those with idle time 

51 batches_by_activity = input.evaluation.batches_by_activity_with_idle 

52 

53 # Sort the activities by the highest idle time 

54 sorted_activities = sorted( 

55 batches_by_activity.keys(), 

56 key=lambda x: sum(batch["idle_time"] for batch in batches_by_activity[x]), 

57 reverse=True, 

58 ) 

59 

60 for activity in sorted_activities: 

61 batch_group = sorted(batches_by_activity[activity], key=lambda x: x["idle_time"], reverse=True) 

62 # For each activity, look at all resources, which could also 

63 # do that batch (incl. the one which originally did it) 

64 

65 resources = timetable.get_resources_assigned_to_task(activity) 

66 for resource in select_variants(store, resources): 

67 # For each resource, find all time intervals/periods, that are are at 

68 # least avg_ideal_proc long (only work-time, skipping idle time), 

69 # and start after accumulation_begin. Up to a time after the actual 

70 # batch start, that is at least avg_ideal_proc long. 

71 calendar = timetable.get_calendar_for_resource(resource) 

72 if calendar is None: 

73 continue 

74 for batch in select_variants(store, batch_group, inner=True, ordered=True): 

75 first_enablement_weekday = DAY.from_date(batch["accumulation_begin"]) 

76 required_processing_time = ceil(batch["ideal_proc"] / 3600) 

77 

78 time_periods = calendar.get_time_periods_of_length_excl_idle( 

79 first_enablement_weekday, 

80 required_processing_time, 

81 start_time=batch["accumulation_begin"].hour, 

82 last_start_time=batch["start"].hour + required_processing_time, 

83 ) 

84 

85 proposed_waiting_times = [] 

86 

87 for period in time_periods: 

88 required_large_wt = period.begin_time_hour - batch["accumulation_begin"].hour 

89 # Idle time of 0 doesn't make sense, 

90 # as it's basically the same as no batching 

91 if required_large_wt <= 0: 

92 continue 

93 

94 waiting_time = ceil(required_large_wt) * 3600 

95 # If we already proposed this waiting time, skip 

96 if waiting_time in proposed_waiting_times: 

97 continue 

98 proposed_waiting_times.append(waiting_time) 

99 

100 for waiting_time in select_variants( 

101 store, proposed_waiting_times, inner=True, ordered=True 

102 ): 

103 yield ( 

104 AddLargeWTRuleByIdleAction.get_default_rating(), 

105 AddLargeWTRuleByIdleAction( 

106 AddLargeWTRuleByIdleActionParamsType( 

107 task_id=activity, 

108 waiting_time=waiting_time, 

109 type=RULE_TYPE.LARGE_WT, 

110 ) 

111 ), 

112 )