Coverage for o2/actions/batching_actions/modify_size_rule_by_utilization_action.py: 38%

1from o2.actions.base_actions.add_size_rule_base_action import ( 

2 AddSizeRuleAction, 

3 AddSizeRuleBaseActionParamsType, 

4) 

5from o2.actions.base_actions.base_action import ( 

6 RateSelfReturnType, 

7) 

8from o2.actions.base_actions.modify_size_rule_base_action import ( 

9 ModifySizeRuleBaseAction, 

10 ModifySizeRuleBaseActionParamsType, 

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 

16from o2.util.helper import select_variants 

17 

18 

19class ModifySizeRuleByUtilizationActionParamsType(ModifySizeRuleBaseActionParamsType): 

20 """Parameter for ModifySizeRuleByUtilizationAction.""" 

21 

22 pass 

23 

24 

25class ModifySizeRuleByLowUtilizationAction(ModifySizeRuleBaseAction): 

26 """An Action to modify size batching rules based on utilization. 

27 

28 1. Gets least utilized resources 

29 2. Looks at all resources with utilization < 0.5, smallest utilization first 

30 3. Gets a task (with batching enabled) that uses the resource 

31 4. Increases the size rule by 1 

32 """ 

33 

34 params: ModifySizeRuleByUtilizationActionParamsType 

35 

36 @staticmethod 

37 def rate_self( 

38 store: "Store", input: "Solution" 

39 ) -> RateSelfReturnType["ModifySizeRuleByLowUtilizationAction | AddSizeRuleAction"]: 

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

41 timetable = input.state.timetable 

42 evaluation = input.evaluation 

43 

44 resource_utilizations = evaluation.resource_utilizations 

45 resources_by_utilization = sorted(resource_utilizations.items(), key=lambda x: x[1]) 

46 

47 for resource_id, utilization in resources_by_utilization: 

48 if utilization > 0.5: 

49 continue 

50 

51 tasks = timetable.get_task_ids_assigned_to_resource(resource_id) 

52 selectors = timetable.get_firing_rule_selectors_for_tasks(tasks, rule_type=RULE_TYPE.SIZE) 

53 if not selectors: 

54 for task_id in select_variants(store, tasks): 

55 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

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 continue 

67 for selector in select_variants(store, selectors): 

68 duration_fn = store.constraints.get_duration_fn_for_task(selector.batching_rule_task_id) 

69 yield ( 

70 RATING.HIGH, 

71 ModifySizeRuleByLowUtilizationAction( 

72 ModifySizeRuleByUtilizationActionParamsType( 

73 rule=selector, 

74 size_increment=1, 

75 duration_fn=duration_fn, 

76 ) 

77 ), 

78 ) 

79 

80 

81class ModifySizeRuleByHighUtilizationAction(ModifySizeRuleBaseAction): 

82 """An Action to modify size batching rules based on utilization. 

83 

84 1. Gets most utilized resources 

85 2. Looks at all resources with utilization > 0.8, largest utilization first 

86 3. Gets a task (with batching enabled) that uses the resource 

87 4. Reduces the size rule by 1 

88 """ 

89 

90 params: ModifySizeRuleByUtilizationActionParamsType 

91 

92 @staticmethod 

93 def rate_self( 

94 store: "Store", input: "Solution" 

95 ) -> RateSelfReturnType["ModifySizeRuleByHighUtilizationAction"]: 

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

97 timetable = input.state.timetable 

98 evaluation = input.evaluation 

99 

100 resource_utilizations = evaluation.resource_utilizations 

101 resources_by_utilization = sorted(resource_utilizations.items(), key=lambda x: x[1], reverse=True) 

102 

103 for resource_id, utilization in resources_by_utilization: 

104 if utilization < 0.8: 

105 continue 

106 

107 tasks = timetable.get_task_ids_assigned_to_resource(resource_id) 

108 selectors = timetable.get_firing_rule_selectors_for_tasks(tasks, rule_type=RULE_TYPE.SIZE) 

109 for selector in select_variants(store, selectors): 

110 duration_fn = store.constraints.get_duration_fn_for_task(selector.batching_rule_task_id) 

111 yield ( 

112 RATING.HIGH, 

113 ModifySizeRuleByHighUtilizationAction( 

114 ModifySizeRuleByUtilizationActionParamsType( 

115 rule=selector, 

116 size_increment=-1, 

117 duration_fn=duration_fn, 

118 ) 

119 ), 

120 )