Coverage for o2/actions/batching_actions/add_date_time_rule_by_start_action.py: 91%

1from collections import Counter 

2 

3from typing_extensions import override 

4 

5from o2.actions.base_actions.add_datetime_rule_base_action import ( 

6 AddDateTimeRuleBaseAction, 

7 AddDateTimeRuleBaseActionParamsType, 

8) 

9from o2.actions.base_actions.base_action import ( 

10 RateSelfReturnType, 

11) 

12from o2.models.self_rating import RATING 

13from o2.models.solution import Solution 

14from o2.models.timetable.time_period import TimePeriod 

15from o2.store import Store 

16from o2.util.helper import select_variants 

17 

18 

19class AddDateTimeRuleByStartActionParamsType(AddDateTimeRuleBaseActionParamsType): 

20 """Parameter for AddDateTimeRuleByStartAction.""" 

21 

22 pass 

23 

24 

25class AddDateTimeRuleByStartAction(AddDateTimeRuleBaseAction): 

26 """AddDateTimeRuleByStartAction will add new daily_hour / weekday rules based on the start time of tasks. 

27 

28 It does the following: 

29 1. Get all start times by resource (therefore we roughly know when a resource is available) 

30 2. Find the tasks with the most (batching) waiting time 

31 3. For that task get all assigned resources and their start times 

32 4. Find the most frequent intersection of those start times 

33 5. Add a new daily_hour / weekday rule for the task based on the start time 

34 """ 

35 

36 params: AddDateTimeRuleByStartActionParamsType 

37 

38 @override 

39 @staticmethod 

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

41 evaluation = store.current_evaluation 

42 timetable = store.current_timetable 

43 

44 start_times = evaluation.resource_started_weekdays 

45 

46 sorted_tasks = sorted( 

47 store.current_evaluation.total_batching_waiting_time_per_task.items(), 

48 key=lambda x: x[1], 

49 reverse=True, 

50 ) 

51 

52 for task_id, waiting_time in sorted_tasks: 

53 # If we got no waiting time, we can stop 

54 if waiting_time < 1: 

55 break 

56 

57 resources_ids = timetable.get_resources_assigned_to_task(task_id) 

58 if not resources_ids: 

59 continue 

60 

61 aggregated_start_times = { 

62 (day, hour): count 

63 for resource_id in resources_ids 

64 for day, times in start_times.get(resource_id, {}).items() 

65 for hour, count in times.items() 

66 } 

67 

68 # find most frequent day,hour combination 

69 # Lets look at the top 3 

70 most_frequent_day_hour = Counter(aggregated_start_times).most_common() 

71 if not most_frequent_day_hour: 

72 continue 

73 

74 for (day, hour), _ in select_variants(store, most_frequent_day_hour, ordered=True): 

75 yield ( 

76 RATING.HIGH, 

77 AddDateTimeRuleByStartAction( 

78 AddDateTimeRuleByStartActionParamsType( 

79 task_id=task_id, 

80 time_period=TimePeriod.from_start_end(hour, hour + 1, day), 

81 duration_fn=store.constraints.get_duration_fn_for_task(task_id), 

82 ) 

83 ), 

84 )