Coverage for o2/actions/batching_actions/add_date_time_rule_by_enablement_action.py: 92%

1from typing_extensions import override 

2 

3from o2.actions.base_actions.add_datetime_rule_base_action import ( 

4 AddDateTimeRuleBaseAction, 

5 AddDateTimeRuleBaseActionParamsType, 

6) 

7from o2.actions.base_actions.base_action import ( 

8 RateSelfReturnType, 

9) 

10from o2.models.self_rating import RATING 

11from o2.models.solution import Solution 

12from o2.models.timetable.time_period import TimePeriod 

13from o2.store import Store 

14from o2.util.helper import select_variants 

15 

16 

17class AddDateTimeRuleByEnablementActionParamsType(AddDateTimeRuleBaseActionParamsType): 

18 """Parameter for AddDateTimeRuleByEnablementAction.""" 

19 

20 pass 

21 

22 

23class AddDateTimeRuleByEnablementAction(AddDateTimeRuleBaseAction): 

24 """AddDateTimeRuleByEnablementAction will add new daily_hour / weekday rules. 

25 

26 based on the enablement of the task. 

27 

28 It does the following: 

29 1. gets the tasks with the most waiting time 

30 2. finds the most frequent enablement datetime for the task 

31 3. add a new daily_hour / weekday rule for the task based on the enablement datetime 

32 to the most frequent task-datetime pair. 

33 """ 

34 

35 params: AddDateTimeRuleByEnablementActionParamsType 

36 

37 @override 

38 @staticmethod 

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

40 evaluation = store.current_evaluation 

41 task_enablement_weekdays = evaluation.task_enablement_weekdays 

42 

43 sorted_tasks = sorted( 

44 store.current_evaluation.total_batching_waiting_time_per_task.items(), 

45 key=lambda x: x[1], 

46 reverse=True, 

47 ) 

48 

49 task_enablements = { 

50 task_id: sorted( 

51 ((count, day, hour) for day, hours in dict.items() for hour, count in hours.items()), 

52 key=lambda x: x[0], 

53 reverse=True, 

54 ) 

55 for task_id, dict in task_enablement_weekdays.items() 

56 } 

57 

58 for task_id, waiting_time in sorted_tasks: 

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

60 if waiting_time < 1: 

61 break 

62 task_enablement_pairs = task_enablements[task_id] 

63 for _, day, hour in select_variants(store, task_enablement_pairs, ordered=True): 

64 yield ( 

65 RATING.HIGH, 

66 AddDateTimeRuleByEnablementAction( 

67 AddDateTimeRuleByEnablementActionParamsType( 

68 task_id=task_id, 

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

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

71 ) 

72 ), 

73 ) 

74 

75 return