Coverage for o2/actions/batching_actions/modify_size_rule_by_cost_fn_action.py: 50%

1from collections import defaultdict 

2from collections.abc import Mapping 

3 

4from o2.actions.base_actions.add_size_rule_base_action import ( 

5 AddSizeRuleAction, 

6 AddSizeRuleBaseActionParamsType, 

7) 

8from o2.actions.base_actions.base_action import ( 

9 RateSelfReturnType, 

10) 

11from o2.actions.base_actions.modify_size_rule_base_action import ( 

12 ModifySizeRuleBaseAction, 

13 ModifySizeRuleBaseActionParamsType, 

14) 

15from o2.models.rule_selector import RuleSelector 

16from o2.models.self_rating import RATING 

17from o2.models.solution import Solution 

18from o2.models.timetable import RULE_TYPE 

19from o2.store import Store 

20from o2.util.helper import select_variants 

21 

22 

23class ModifySizeRuleByCostFnActionParamsType(ModifySizeRuleBaseActionParamsType): 

24 """Parameter for ModifySizeRuleByCostFn.""" 

25 

26 pass 

27 

28 

29def rate_self_helper_by_metric_dict( 

30 store: "Store", 

31 task_id_metric_dict: Mapping[str, float], 

32 task_class: type[ModifySizeRuleBaseAction], 

33) -> RateSelfReturnType: 

34 """Rate ModifySizeRuleByCostFn actions.""" 

35 timetable = store.current_timetable 

36 constraints = store.constraints 

37 state = store.current_state 

38 

39 sorted_metric_values = sorted(set(task_id_metric_dict.values()), reverse=True) 

40 

41 for metric in sorted_metric_values: 

42 task_ids = [task_id for task_id, value in task_id_metric_dict.items() if value == metric] 

43 for task_id in select_variants(store, task_ids): 

44 firing_rule_selectors = timetable.get_firing_rule_selectors_for_task( 

45 task_id, rule_type=RULE_TYPE.SIZE 

46 ) 

47 size_constraints = constraints.get_batching_size_rule_constraints(task_id) 

48 if not size_constraints: 

49 continue 

50 size_constraint = size_constraints[0] 

51 

52 # In case we do not have firing_rule selectors, we might want to add a new rule 

53 if not firing_rule_selectors: 

54 new_cost = size_constraint.cost_fn_lambda(2) 

55 old_cost = size_constraint.cost_fn_lambda(1) 

56 if new_cost <= old_cost: 

57 duration_fn = store.constraints.get_duration_fn_for_task(task_id) 

58 yield ( 

59 RATING.LOW, 

60 AddSizeRuleAction( 

61 AddSizeRuleBaseActionParamsType(task_id=task_id, size=2, duration_fn=duration_fn) 

62 ), 

63 ) 

64 

65 for firing_rule_selector in select_variants(store, firing_rule_selectors, inner=True): 

66 firing_rule = firing_rule_selector.get_firing_rule_from_state(state) 

67 if firing_rule is None: 

68 continue 

69 size = firing_rule.value 

70 

71 new_cost = size_constraint.cost_fn_lambda(size + 1) 

72 old_cost = size_constraint.cost_fn_lambda(size) 

73 if new_cost <= old_cost: 

74 yield ( 

75 ModifySizeRuleBaseAction.get_default_rating(), 

76 task_class( 

77 ModifySizeRuleByCostFnActionParamsType( 

78 rule=firing_rule_selector, 

79 size_increment=1, 

80 duration_fn=size_constraint.duration_fn, 

81 ) 

82 ), 

83 ) 

84 

85 

86class ModifySizeRuleByCostFnRepetitiveTasksAction(ModifySizeRuleBaseAction): 

87 """An Action to modify size batching rules based on the cost fn. 

88 

89 If batch size increment reduces Costs => Increase Batch Size 

90 - For repetitive tasks (higher frequencies first) 

91 """ 

92 

93 params: ModifySizeRuleByCostFnActionParamsType 

94 

95 @staticmethod 

96 def rate_self( 

97 store: "Store", input: "Solution" 

98 ) -> RateSelfReturnType["ModifySizeRuleByCostFnRepetitiveTasksAction"]: 

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

100 evaluation = input.evaluation 

101 

102 task_frequencies = evaluation.task_execution_counts 

103 yield from rate_self_helper_by_metric_dict( 

104 store, task_frequencies, ModifySizeRuleByCostFnRepetitiveTasksAction 

105 ) 

106 

107 

108class ModifySizeRuleByCostFnHighCostsAction(ModifySizeRuleBaseAction): 

109 """An Action to modify size batching rules based on the cost fn. 

110 

111 If batch size increment reduces Costs => Increase Batch Size 

112 - For tasks with high costs (higher costs first). 

113 """ 

114 

115 params: ModifySizeRuleByCostFnActionParamsType 

116 

117 @staticmethod 

118 def rate_self( 

119 store: "Store", input: "Solution" 

120 ) -> RateSelfReturnType["ModifySizeRuleByCostFnHighCostsAction"]: 

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

122 evaluation = input.evaluation 

123 

124 task_costs = evaluation.get_total_cost_per_task() 

125 yield from rate_self_helper_by_metric_dict(store, task_costs, ModifySizeRuleByCostFnHighCostsAction) 

126 

127 

128class ModifySizeRuleByCostFnLowProcessingTimeAction(ModifySizeRuleBaseAction): 

129 """An Action to modify size batching rules based on the cost fn. 

130 

131 If batch size increment reduces processing time => Increase Batch Size 

132 - For tasks with low processing time (lower processing time first) 

133 """ 

134 

135 params: ModifySizeRuleByCostFnActionParamsType 

136 

137 @staticmethod 

138 def rate_self( 

139 store: "Store", input: "Solution" 

140 ) -> RateSelfReturnType["ModifySizeRuleByCostFnLowProcessingTimeAction"]: 

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

142 evaluation = input.evaluation 

143 

144 task_processing_times = evaluation.get_average_processing_time_per_task() 

145 yield from rate_self_helper_by_metric_dict( 

146 store, task_processing_times, ModifySizeRuleByCostFnLowProcessingTimeAction 

147 ) 

148 

149 

150class ModifyBatchSizeIfNoCostImprovementAction(ModifySizeRuleBaseAction): 

151 """An Action to modify size batching rules based on the cost fn. 

152 

153 If batch size decrement does not increase Costs (looking at the cost fn) => Decrease Batch Size 

154 """ 

155 

156 params: ModifySizeRuleByCostFnActionParamsType 

157 

158 @staticmethod 

159 def rate_self( 

160 store: "Store", input: "Solution" 

161 ) -> RateSelfReturnType["ModifyBatchSizeIfNoCostImprovementAction | AddSizeRuleAction"]: 

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

163 constraints = store.constraints 

164 timetable = input.state.timetable 

165 state = input.state 

166 

167 task_ids = timetable.get_task_ids() 

168 rule_selectors_by_cost: dict[float, list[RuleSelector]] = defaultdict(list) 

169 

170 for task_id in task_ids: 

171 size_constraints = constraints.get_batching_size_rule_constraints(task_id) 

172 if not size_constraints: 

173 continue 

174 size_constraint = size_constraints[0] 

175 

176 firing_rule_selectors = timetable.get_firing_rule_selectors_for_task( 

177 task_id, rule_type=RULE_TYPE.SIZE 

178 ) 

179 

180 # Because we are aiming to reduce batch size, we do NOT add a new rule here 

181 if not firing_rule_selectors: 

182 continue 

183 

184 for firing_rule_selector in firing_rule_selectors: 

185 firing_rule = firing_rule_selector.get_firing_rule_from_state(state) 

186 if firing_rule is None: 

187 continue 

188 size = firing_rule.value 

189 

190 new_cost = size_constraint.cost_fn_lambda(size - 1) 

191 old_cost = size_constraint.cost_fn_lambda(size) 

192 if new_cost <= old_cost: 

193 cost = old_cost - new_cost 

194 rule_selectors_by_cost[cost].append(firing_rule_selector) 

195 

196 costs_sorted = sorted(rule_selectors_by_cost.keys(), reverse=True) 

197 

198 for cost in costs_sorted: 

199 rule_selectors = rule_selectors_by_cost[cost] 

200 for rule_selector in select_variants(store, rule_selectors): 

201 duration_fn = constraints.get_duration_fn_for_task(rule_selector.batching_rule_task_id) 

202 yield ( 

203 RATING.LOW, 

204 ModifyBatchSizeIfNoCostImprovementAction( 

205 ModifySizeRuleByCostFnActionParamsType( 

206 rule=rule_selector, 

207 size_increment=-1, 

208 duration_fn=duration_fn, 

209 ) 

210 ), 

211 ) 

212 

213 

214class ModifySizeRuleByCostFnLowCycleTimeImpactAction(ModifySizeRuleBaseAction): 

215 """An Action to modify size batching rules based on the cost fn. 

216 

217 If batch size increment reduces Costs => Increase Batch Size 

218 - For Tasks which have a low impact on the cycle time (looking at the duration fn) 

219 """ 

220 

221 params: ModifySizeRuleByCostFnActionParamsType 

222 

223 @staticmethod 

224 def rate_self( 

225 store: "Store", input: "Solution" 

226 ) -> RateSelfReturnType["ModifySizeRuleByCostFnLowCycleTimeImpactAction | AddSizeRuleAction"]: 

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

228 timetable = input.state.timetable 

229 constraints = store.constraints 

230 state = input.state 

231 

232 task_ids = timetable.get_task_ids() 

233 rule_selectors_by_cycle_time_impact: dict[float, list[RuleSelector]] = defaultdict(list) 

234 

235 for task_id in task_ids: 

236 size_constraints = constraints.get_batching_size_rule_constraints(task_id) 

237 if not size_constraints: 

238 continue 

239 size_constraint = size_constraints[0] 

240 

241 firing_rule_selectors = timetable.get_firing_rule_selectors_for_task( 

242 task_id, rule_type=RULE_TYPE.SIZE 

243 ) 

244 if not firing_rule_selectors: 

245 new_cost = size_constraint.cost_fn_lambda(2) 

246 old_cost = size_constraint.cost_fn_lambda(1) 

247 

248 old_duration = size_constraint.duration_fn_lambda(1) 

249 new_duration = size_constraint.duration_fn_lambda(2) 

250 if new_cost < old_cost: 

251 cycle_time_impact = old_duration - new_duration 

252 rule_selectors_by_cycle_time_impact[cycle_time_impact].append( 

253 RuleSelector(batching_rule_task_id=task_id, firing_rule_index=None) 

254 ) 

255 

256 for firing_rule_selector in firing_rule_selectors: 

257 firing_rule = firing_rule_selector.get_firing_rule_from_state(state) 

258 if firing_rule is None: 

259 continue 

260 size = firing_rule.value 

261 

262 new_cost = size_constraint.cost_fn_lambda(size + 1) 

263 old_cost = size_constraint.cost_fn_lambda(size) 

264 

265 old_duration = size_constraint.duration_fn_lambda(size) 

266 new_duration = size_constraint.duration_fn_lambda(size + 1) 

267 if new_cost < old_cost: 

268 cycle_time_impact = old_duration - new_duration 

269 rule_selectors_by_cycle_time_impact[cycle_time_impact].append(firing_rule_selector) 

270 

271 cycle_time_impacts_sorted = sorted( 

272 rule_selectors_by_cycle_time_impact.keys(), 

273 reverse=True, 

274 ) 

275 

276 for cycle_time_impact in cycle_time_impacts_sorted: 

277 rule_selectors = rule_selectors_by_cycle_time_impact[cycle_time_impact] 

278 for rule_selector in select_variants(store, rule_selectors): 

279 duration_fn = constraints.get_duration_fn_for_task(rule_selector.batching_rule_task_id) 

280 

281 if rule_selector.firing_rule_index is not None: 

282 yield ( 

283 ModifySizeRuleBaseAction.get_default_rating(), 

284 ModifySizeRuleByCostFnLowCycleTimeImpactAction( 

285 ModifySizeRuleByCostFnActionParamsType( 

286 rule=rule_selector, 

287 size_increment=1, 

288 duration_fn=duration_fn, 

289 ) 

290 ), 

291 ) 

292 else: 

293 yield ( 

294 RATING.LOW, 

295 AddSizeRuleAction( 

296 AddSizeRuleBaseActionParamsType( 

297 task_id=rule_selector.batching_rule_task_id, 

298 size=2, 

299 duration_fn=duration_fn, 

300 ) 

301 ), 

302 ) 

303 

304 

305class ModifySizeRuleByManySimilarEnablementsAction(ModifySizeRuleBaseAction): 

306 """An Action to modify size batching rules based on the cost fn. 

307 

308 - Modify those tasks, which have many enablements at the same time 

309 """ 

310 

311 params: ModifySizeRuleByCostFnActionParamsType 

312 

313 @staticmethod 

314 def rate_self( 

315 store: "Store", input: "Solution" 

316 ) -> RateSelfReturnType["ModifySizeRuleByManySimilarEnablementsAction"]: 

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

318 evaluation = input.evaluation 

319 

320 tasks_by_number_of_duplicate_enablement_dates = ( 

321 evaluation.tasks_by_number_of_duplicate_enablement_dates 

322 ) 

323 yield from rate_self_helper_by_metric_dict( 

324 store, 

325 tasks_by_number_of_duplicate_enablement_dates, 

326 ModifySizeRuleByManySimilarEnablementsAction, 

327 )