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

1from math import ceil 

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.store import Store 

13 

14 

15class AddDateTimeRuleByAvailabilityActionParamsType(AddDateTimeRuleBaseActionParamsType): 

16 """Parameter for AddDateTimeRuleByAvailabilityAction.""" 

17 

18 pass 

19 

20 

21class AddDateTimeRuleByAvailabilityAction(AddDateTimeRuleBaseAction): 

22 """An Action to add daily_hour / weekday batching rules based on availability. 

23 

24 This action will add a new daily_hour / weekday rule based on the availability of the task. 

25 It does the following: 

26 1. Find the task with the highest processing time. 

27 2. Get the average processing time for the task. 

28 3. Find the most frequent availability time period for the task, with min. hours needed. 

29 4. Add a new daily_hour / weekday rule for the task based on the availability. 

30 """ 

31 

32 params: AddDateTimeRuleByAvailabilityActionParamsType 

33 

34 @staticmethod 

35 def rate_self( 

36 store: "Store", input: "Solution" 

37 ) -> RateSelfReturnType["AddDateTimeRuleByAvailabilityAction"]: 

38 """Generate a best set of parameters & self-evaluates this action.""" 

39 timetable = input.state.timetable 

40 

41 avg_processing_times = input.evaluation.get_average_processing_time_per_task() 

42 

43 sorted_tasks = sorted( 

44 store.current_evaluation.get_total_duration_time_per_task().items(), 

45 key=lambda x: x[1], 

46 reverse=True, 

47 ) 

48 for task_id, _ in sorted_tasks: 

49 avg_processing_time = avg_processing_times[task_id] 

50 hours_needed = ceil(avg_processing_time / 60 / 60) 

51 best_time_period = timetable.get_highest_availability_time_period(task_id, hours_needed) 

52 if best_time_period is None: 

53 continue 

54 

55 yield ( 

56 RATING.HIGH, 

57 AddDateTimeRuleByAvailabilityAction( 

58 AddDateTimeRuleByAvailabilityActionParamsType( 

59 task_id=task_id, 

60 time_period=best_time_period, 

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

62 ) 

63 ), 

64 ) 

65 return