Coverage for o2/actions/batching_actions/add_ready_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 AddReadyWTRuleByWTActionParamsType(AddReadyLargeWTRuleBaseActionParamsType): 

19 """Parameter for AddReadyWTRuleByWTAction.""" 

20 

21 pass 

22 

23 

24class AddReadyWTRuleByWTAction(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: AddReadyWTRuleByWTActionParamsType 

34 

35 @override 

36 @staticmethod 

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

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 new_large_wt = min(ceil(waiting_time * 0.9), 24 * 60 * 60) 

48 yield ( 

49 RATING.VERY_LOW, 

50 AddReadyWTRuleByWTAction( 

51 AddReadyWTRuleByWTActionParamsType( 

52 task_id=task_id, 

53 waiting_time=new_large_wt, 

54 type=RULE_TYPE.READY_WT, 

55 ) 

56 ), 

57 )