Coverage for o2/actions/batching_actions/add_large_wt_rule_by_wt_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.self_rating import RATING 

13from o2.models.solution import Solution 

14from o2.models.timetable import RULE_TYPE 

15from o2.store import Store 

16 

17 

18class AddLargeWTRuleByWTActionParamsType(AddReadyLargeWTRuleBaseActionParamsType): 

19 """Parameter for AddLargeWTRuleByWTAction.""" 

20 

21 pass 

22 

23 

24class AddLargeWTRuleByWTAction(AddReadyLargeWTRuleBaseAction): 

25 """An Action to add a LargeWT rule based on the waiting time of the task. 

26 

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

28 It does the following: 

29 1. Find the task with the highest (batching) waiting time. 

30 2. Create a new LargeWT rule for 90% of the waiting time. 

31 """ 

32 

33 params: AddLargeWTRuleByWTActionParamsType 

34 

35 @override 

36 @staticmethod 

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

38 sorted_tasks = sorted( 

39 input.evaluation.total_batching_waiting_time_per_task.items(), 

40 key=lambda x: x[1], 

41 reverse=True, 

42 ) 

43 for task_id, _ in sorted_tasks: 

44 waiting_time = input.evaluation.avg_batching_waiting_time_per_task[task_id] 

45 if waiting_time == 0: 

46 continue 

47 # Cap the waiting time to 24 hours 

48 new_large_wt = min(ceil(waiting_time * 0.9), 24 * 60 * 60) 

49 yield ( 

50 RATING.VERY_LOW, 

51 AddLargeWTRuleByWTAction( 

52 AddLargeWTRuleByWTActionParamsType( 

53 task_id=task_id, 

54 waiting_time=new_large_wt, 

55 type=RULE_TYPE.LARGE_WT, 

56 ) 

57 ), 

58 )