from abc import ABC, abstractmethod from dateutil import tz import pytz from apscheduler.triggers.cron import CronTrigger as APCronTrigger from configspec import * import preserve class TaskTrigger(ABC): """Abstract trigger class""" @abstractmethod def next_time(self, base_time): """ Return a time indicating the next trigger time after base_time. Return None if no more trigger events. """ @preserve.preservable(exclude_attrs=['ap_trigger']) class CronTrigger(TaskTrigger): """ Interprets Cron strings using a wrapper around the APScheduler CronTrigger (and so function is similar). Values left as default or supplied as None are set to a wildcard, unless it is a smaller unit than those supplied - where it instead gets set to it's minimum (so setting `hour` to 3 will set `minute` and `second` to 0). The trigger format will be matched against The timezone used is always the local system timezone. Details available at https://apscheduler.readthedocs.io/en/latest/modules/triggers/cron.html """ def __init__(self, month=None, day=None, day_of_week=None, hour=None, minute=None, second=None): self.month = month self.day = day self.day_of_week = day_of_week self.hour = hour self.minute = minute self.second = second self.__restore_init__() def __restore_init__(self): # Default timezone is the one from tzlocal self.ap_trigger = APCronTrigger(month=self.month, day=self.day, day_of_week=self.day_of_week, hour=self.hour, minute=self.minute, second=self.second) def next_time(self, base_time): """ Return a time indicating the next trigger time after base_time. Return None if no more trigger events. """ # Convert base_time to UTC with dateutil, then to pytz which APScheduler requires. utc_base_time = base_time.astimezone(tz.tzutc()).astimezone(pytz.utc) fire_time = self.ap_trigger.get_next_fire_time(None, utc_base_time) # Convert back to UTC, as ap_trigger returns a value with local timezone # Use DateUtil, as it doesn't add other crap to tzinfo return fire_time.astimezone(pytz.utc).astimezone(tz.tzutc()) TriggerSpec = ConfigSpecification() TriggerSpec.add_specs({ 'month': StringSpec(helptext="Month in year, 1-12"), 'day': StringSpec(helptext="Day of month, 1-32"), 'day_of_week': StringSpec(helptext="Day of week, 0-6 or mon,tue,wed,thu,fri,sat,sun"), 'hour': StringSpec(helptext="Hour in day, 0-23"), 'minute': StringSpec(helptext="Minute in hour, 0-59"), 'second': StringSpec(helptext="Second in minute, 0-59"), }, optional=True) # class IntervalTrigger(TaskTrigger): """ Triggers every x period starting from when it was first created (carries over lowpower) """ # pass # class SingleTrigger(TaskTrigger): """ Either pass a whole datetime instance, or a delta like a period that gets added to current. """ # pass class Task(): def __init__(self, trigger, interface_call): pass # InterfaceCall already handles the callables and args for us, we just need to preserve # them. Trigger is going to be multiple formats, but the most common will be Cron style. def init_tasks(core_config, applied_config, interface_functions): pass # Check if we have a session to load # Load the session - this includes resolving interfacecalls # Return a list of tasks def start_tasks(task_list): pass # Start our scheduler thread to fire off tasks, including the initial grace period checks # Due to our task constraints, our scheduler can be somewhat simplified. We know that task # triggers won't change after init, and they will be infrequent enough that we can just # make a new thread for each task. # Maybe have "cycles", where we determine what task is going to occur next, and when. # A main dispatch loop then pretty much delays until that point, or triggers the low power # stuff somehow. # Does the low power stuff occur here? Or somewhere else?