Coverage for o2/actions/batching_actions/modify_size_rule_by_utilization_action.py: 38%
47 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-16 11:18 +0000
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-16 11:18 +0000
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
19class ModifySizeRuleByUtilizationActionParamsType(ModifySizeRuleBaseActionParamsType):
20 """Parameter for ModifySizeRuleByUtilizationAction."""
22 pass
25class ModifySizeRuleByLowUtilizationAction(ModifySizeRuleBaseAction):
26 """An Action to modify size batching rules based on utilization.
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 """
34 params: ModifySizeRuleByUtilizationActionParamsType
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
44 resource_utilizations = evaluation.resource_utilizations
45 resources_by_utilization = sorted(resource_utilizations.items(), key=lambda x: x[1])
47 for resource_id, utilization in resources_by_utilization:
48 if utilization > 0.5:
49 continue
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 )
81class ModifySizeRuleByHighUtilizationAction(ModifySizeRuleBaseAction):
82 """An Action to modify size batching rules based on utilization.
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 """
90 params: ModifySizeRuleByUtilizationActionParamsType
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
100 resource_utilizations = evaluation.resource_utilizations
101 resources_by_utilization = sorted(resource_utilizations.items(), key=lambda x: x[1], reverse=True)
103 for resource_id, utilization in resources_by_utilization:
104 if utilization < 0.8:
105 continue
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 )