Coverage for o2/actions/batching_actions/modify_size_rule_by_wt_action.py: 96%

1from typing_extensions import override 

2 

3from o2.actions.base_actions.add_size_rule_base_action import ( 

4 AddSizeRuleAction, 

5 AddSizeRuleBaseActionParamsType, 

6) 

7from o2.actions.base_actions.base_action import ( 

8 RateSelfReturnType, 

9) 

10from o2.actions.base_actions.modify_size_rule_base_action import ( 

11 ModifySizeRuleBaseAction, 

12 ModifySizeRuleBaseActionParamsType, 

13) 

14from o2.models.self_rating import RATING 

15from o2.models.solution import Solution 

16from o2.models.timetable import RULE_TYPE 

17from o2.store import Store 

18from o2.util.helper import select_variants 

19 

20 

21class ModifySizeRuleByWTActionParamsType(ModifySizeRuleBaseActionParamsType): 

22 """Parameter for ModifySizeRuleByWTAction.""" 

23 

24 pass 

25 

26 

27class ModifySizeRuleByWTAction(ModifySizeRuleBaseAction): 

28 """An Action to modify size batching rules based on waiting time. 

29 

30 1. Gets the tasks with the most waiting time 

31 2. Looks if theres a size rule for that task 

32 3. If there is, it decrements the size by 1 

33 """ 

34 

35 params: ModifySizeRuleByWTActionParamsType 

36 

37 @override 

38 @staticmethod 

39 def rate_self( 

40 store: "Store", input: "Solution" 

41 ) -> RateSelfReturnType["ModifySizeRuleByWTAction | AddSizeRuleAction"]: 

42 timetable = input.state.timetable 

43 avg_batching_waiting_time_per_task = input.evaluation.avg_batching_waiting_time_per_task 

44 sorted_tasks = sorted( 

45 avg_batching_waiting_time_per_task.items(), 

46 key=lambda x: x[1], 

47 reverse=True, 

48 ) 

49 for task_id, waiting_time in sorted_tasks: 

50 if (waiting_time) < 1: 

51 continue 

52 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

53 

54 selectors = timetable.get_firing_rule_selectors_for_task(task_id, rule_type=RULE_TYPE.SIZE) 

55 if not selectors: 

56 yield ( 

57 RATING.LOW, 

58 AddSizeRuleAction( 

59 AddSizeRuleBaseActionParamsType( 

60 task_id=task_id, 

61 size=2, 

62 duration_fn=duration_fn, 

63 ) 

64 ), 

65 ) 

66 for selector in select_variants(store, selectors): 

67 yield ( 

68 RATING.LOW, 

69 ModifySizeRuleByWTAction( 

70 ModifySizeRuleByWTActionParamsType( 

71 rule=selector, 

72 size_increment=-1, 

73 duration_fn=duration_fn, 

74 ) 

75 ), 

76 )