in source/lambda/configuration/running_period.py [0:0]
def get_desired_state(self, logger, current_dt):
"""
Test if the instance should be running at the specified dt, all conditions configured a period should be true
:param logger: logger to log the output of scheduling logic
:param current_dt: time to test
:return: desired state for the instance in the period
"""
def state_str(checked):
return "[running]" if checked else "[stopped]"
def check_running_state_str(checked_state):
return state_str(checked_state != InstanceSchedule.STATE_STOPPED)
def not_str(is_not_not):
return "" if is_not_not else "not "
# check day of month
def check_monthday(dt):
result = self.monthdays is None or dt.day in self.monthdays
if self.monthdays:
self._log_debug(DEBUG_CHECK_MONTH_DAY, state_str(result), dt.day, "" if result else "not ", set_str(self.monthdays))
return result
# check month
def check_month(dt):
result = self.months is None or dt.month in self.months
if self.months:
self._log_debug(DEBUG_CHECK_MONTH, state_str(result), configuration.MONTH_NAMES[dt.month - 1], not_str(result),
set_str(self.months, displaynames=configuration.MONTH_NAMES, offset=1))
return result
# check weekday
def check_weekday(dt):
result = self.weekdays is None or dt.weekday() in self.weekdays
if self.weekdays is not None:
self._log_debug(DEBUG_CHECK_WEEKDAYS, state_str(result), configuration.WEEKDAY_NAMES[dt.weekday()], not_str(result),
set_str(self.weekdays, displaynames=configuration.WEEKDAY_NAMES))
return result
# check time
def check_time(dt):
t = datetime.time(dt.hour, dt.minute, dt.second)
ts = time_str(t)
# no start and stop time, means running all day
if self.begintime is None and self.endtime is None:
desired_state = InstanceSchedule.STATE_RUNNING
self._log_debug(DEBUG_CHECK_DT_UNDEFINED_START_STOP, state_str(True), desired_state)
return desired_state
elif self.begintime is None:
# just the end time, stopped if later than that time
desired_state = InstanceSchedule.STATE_STOPPED if t >= self.endtime else InstanceSchedule.STATE_ANY
self._log_debug(DEBUG_CHECK_DT_STOP_TIME,
check_running_state_str(desired_state), ts,
"before" if desired_state == InstanceSchedule.STATE_ANY else "after",
time_str(self.endtime), desired_state)
return desired_state
elif self.begintime is not None and self.endtime is None:
# just the start time, running if later that that time
desired_state = InstanceSchedule.STATE_RUNNING if t >= self.begintime else InstanceSchedule.STATE_ANY
self._log_debug(DEBUG_CHECK_DT_START_TIME,
check_running_state_str(desired_state), ts,
"before" if desired_state == InstanceSchedule.STATE_ANY else "after",
time_str(self.begintime), desired_state)
return desired_state
else:
# start and stop time, test if time falls in the period defined by these times
desired_state = InstanceSchedule.STATE_RUNNING \
if self.begintime <= t < self.endtime else InstanceSchedule.STATE_STOPPED
self._log_debug(DEBUG_CHECK_DT_START_AND_STOP,
check_running_state_str(desired_state), ts,
"within" if desired_state == InstanceSchedule.STATE_RUNNING
else "outside",
time_str(self.begintime), time_str(self.endtime), desired_state)
return desired_state
self._logger = logger
state = InstanceSchedule.STATE_STOPPED
self._log_debug(DEBUG_CHECK_DT, self.name)
for check in [check_weekday, check_month, check_monthday]:
if not check(current_dt):
return state
state = check_time(current_dt)
return state