Coverage for o2/actions/legacy_optimos_actions/modify_calendar_by_wt_action.py: 94%

1from dataclasses import dataclass 

2 

3from o2.actions.base_actions.base_action import RateSelfReturnType 

4from o2.actions.base_actions.modify_calendar_base_action import ( 

5 ModifyCalendarBaseAction, 

6 ModifyCalendarBaseActionParamsType, 

7) 

8from o2.models.solution import Solution 

9from o2.models.timetable.time_period import TimePeriod 

10from o2.store import Store 

11 

12 

13class ModifyCalendarByWTActionParamsType(ModifyCalendarBaseActionParamsType): 

14 """Parameter for `ModifyCalendarByWTAction`.""" 

15 

16 

17@dataclass(frozen=True) 

18class ModifyCalendarByWTAction(ModifyCalendarBaseAction, str=False): 

19 """`ModifyCalendarByWTAction` will modify the resource calendars based on wt. 

20 

21 This action is based on the original optimos implementation, 

22 see `solution_traces_sorting_by_waiting_times` in that project. 

23 

24 It will first find the tasks with the most waiting time, then find the days where 

25 those are executed the most, and then find the resources 

26 that does this task the most. (Although not for that day, as it was in Optimos). 

27 Then it will look at the resource calendars for those resources on those days, 

28 and first try to add hours to the start of the shifts, and if that is not possible, 

29 it will try to shift the shifts to start earlier. 

30 """ 

31 

32 @staticmethod 

33 def rate_self(store: Store, input: Solution) -> RateSelfReturnType: 

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

35 evaluation = input.evaluation 

36 tasks = evaluation.get_task_names_sorted_by_waiting_time_desc() 

37 for task in tasks: 

38 days = evaluation.get_most_frequent_enablement_weekdays(task) 

39 resources = evaluation.get_most_frequent_resources(task) 

40 for day in days: 

41 for resource in resources: 

42 calendar = store.current_timetable.get_calendar_for_resource(resource) 

43 if calendar is None: 

44 continue 

45 periods = calendar.get_periods_containing_day(day) 

46 for period in periods: 

47 period_id = period.id 

48 # We need to fix the day period to not change 

49 # change the times of other days 

50 fixed_day_period = TimePeriod( 

51 from_=day, 

52 to=day, 

53 begin_time=period.begin_time, 

54 end_time=period.end_time, 

55 probability=period.probability, 

56 ) 

57 

58 # Try to add hours to the start of the shift 

59 new_period = fixed_day_period.add_hours_before(1) 

60 if new_period is None: 

61 continue 

62 

63 yield ( 

64 ModifyCalendarByWTAction.get_default_rating(store), 

65 ModifyCalendarByWTAction( 

66 ModifyCalendarByWTActionParamsType( 

67 calendar_id=calendar.id, 

68 period_id=period_id, 

69 day=day, 

70 add_hours_before=1, 

71 ) 

72 ), 

73 ) 

74 

75 # Try to shift the shift to start earlier 

76 new_period = fixed_day_period.shift_hours(-1) 

77 if new_period is None: 

78 continue 

79 

80 yield ( 

81 ModifyCalendarByWTAction.get_default_rating(store), 

82 ModifyCalendarByWTAction( 

83 ModifyCalendarByWTActionParamsType( 

84 calendar_id=calendar.id, 

85 period_id=period_id, 

86 day=day, 

87 add_hours_before=0, 

88 shift_hours=-1, 

89 ) 

90 ), 

91 ) 

92 

93 return