Coverage for o2/actions/legacy_optimos_actions/modify_calendar_by_it_action.py: 78%

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 ModifyCalendarByITActionParamsType(ModifyCalendarBaseActionParamsType): 

14 """Parameter for `ModifyCalendarByITAction`.""" 

15 

16 

17@dataclass(frozen=True) 

18class ModifyCalendarByITAction(ModifyCalendarBaseAction, str=False): 

19 """`ModifyCalendarByITAction` 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_idle_times` in that project. 

23 

24 It will first find the tasks with the most idle 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 end of the shifts, and if that is not possible, 

29 it will try to shift the shifts to start later. Finally it will try to add hours 

30 to be beginning & start of the shift. 

31 """ 

32 

33 @staticmethod 

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

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

36 tasks = input.evaluation.get_task_names_sorted_by_idle_time_desc() 

37 for task in tasks: 

38 days = input.evaluation.get_most_frequent_enablement_weekdays(task) 

39 resources = input.evaluation.get_most_frequent_resources(task) 

40 for day in days: 

41 for resource in resources: 

42 calendar = input.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 end of the shift 

59 new_period = fixed_day_period.add_hours_after(1) 

60 if new_period is None: 

61 continue 

62 

63 yield ( 

64 ModifyCalendarByITAction.get_default_rating(store), 

65 ModifyCalendarByITAction( 

66 ModifyCalendarByITActionParamsType( 

67 calendar_id=calendar.id, 

68 period_id=period_id, 

69 day=day, 

70 add_hours_after=1, 

71 ) 

72 ), 

73 ) 

74 

75 # Try to shift the shift to begin later 

76 new_period = fixed_day_period.shift_hours(1) 

77 if new_period is None: 

78 continue 

79 

80 yield ( 

81 ModifyCalendarByITAction.get_default_rating(store), 

82 ModifyCalendarByITAction( 

83 ModifyCalendarByITActionParamsType( 

84 calendar_id=calendar.id, 

85 period_id=period_id, 

86 day=day, 

87 shift_hours=1, 

88 ) 

89 ), 

90 ) 

91 

92 # Try to add hours to the start & end of the shift 

93 # TODO: Ask whether this shouldn't move up? 

94 new_period = fixed_day_period.add_hours_before(1) 

95 if new_period is None: 

96 continue 

97 new_period = new_period.add_hours_after(1) 

98 if new_period is None: 

99 continue 

100 

101 yield ( 

102 ModifyCalendarByITAction.get_default_rating(store), 

103 ModifyCalendarByITAction( 

104 ModifyCalendarByITActionParamsType( 

105 calendar_id=calendar.id, 

106 period_id=period_id, 

107 day=day, 

108 add_hours_before=1, 

109 add_hours_after=1, 

110 ) 

111 ), 

112 ) 

113 

114 return