shepherd-agent/shepherd/scheduler.py

#!/usr/bin/env python3

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.events import *
from collections import namedtuple

from tzlocal import get_localzone

from datetime import datetime
from datetime import timezone
from datetime import timedelta

import time

import subprocess

import os
import shepherd.plugin

import io


def is_raspberry_pi(raise_on_errors=False):
    """Checks if Raspberry PI.
    :return:
    """
    try:
        with io.open('/proc/cpuinfo', 'r') as cpuinfo:
            found = False
            for line in cpuinfo:
                if line.startswith('Hardware'):
                    found = True
                    label, value = line.strip().split(':', 1)
                    value = value.strip()
                    if value not in (
                        'BCM2708',
                        'BCM2709',
                        'BCM2835',
                        'BCM2836'
                    ):
                        if raise_on_errors:
                            raise ValueError(
                                'This system does not appear to be a '
                                'Raspberry Pi.'
                            )
                        else:
                            return False
            if not found:
                if raise_on_errors:
                    raise ValueError(
                        'Unable to determine if this system is a Raspberry Pi.'
                    )
                else:
                    return False
    except IOError:
        if raise_on_errors:
            raise ValueError('Unable to open `/proc/cpuinfo`.')
        else:
            return False

    return True


_scheduler = None

joblist_path = None

JobDescription = namedtuple(
    'Job', ['jobstring', 'hour', 'minute', 'second'])


def init_scheduler(core_config):
    global _scheduler
    global joblist_path
    _scheduler = BlockingScheduler()
    joblist_path = os.path.expanduser(core_config["root_dir"])
    joblist_path = os.path.join(joblist_path, "joblist.shp")


def schedule_job(job_desc):
    if not isinstance(job_desc, JobDescription):
        raise Exception(
            "Argument to schedule_job() must be an instance of shepherd.scheduler.JobDescription")
    global _scheduler
    # resolve callable
    splitstring = job_desc.jobstring.split(':')
    if not ((len(splitstring) == 4) and (splitstring[0] == "shepherd")):
        raise Exception(
            "Could not add job - jobstring is not a valid Shepherd job")
    job_plugin, job_function, job_name = splitstring[1:]
    func = getattr(shepherd.plugin.plugin_functions[job_plugin], job_function)
    triggerstr = "!"+job_desc.hour+":"+job_desc.minute+":"+job_desc.second
    for job in _scheduler.get_jobs():
        #Jobstring must be the same, triggerstring can vary and still match
        if job.id.startswith(job_desc.jobstring):
            # Jobstring matches existing job, so update it
            print("Modifying job "+job.id)
            print("next run was "+job.next_run_time)
            job.modify(func=func, trigger=CronTrigger(
                hour=job_desc.hour, minute=job_desc.minute, second=job_desc.second))
            print("next is now "+job.next_run_time)
            break
    else:
        # No matching job found, so new job
        print("Scheduling job "+job_desc.jobstring)
        newjob=_scheduler.add_job(func, id=job_desc.jobstring+triggerstr, coalesce=True, misfire_grace_time=300, trigger=CronTrigger(
            hour=job_desc.hour, minute=job_desc.minute, second=job_desc.second))
        #print("Next scheduled for "+str(newjob.next_run_time))


# Needs to be called after plugins are initialised, so interface functions
# are available

# Remember to wipe job store and not restore on next boot if config file was changed

def save_jobs():
    joblist = _scheduler.get_jobs()
    saved_jobs = []
    next_job_time = None

    for job in joblist:
        jobstring, _, triggerstr = job.id.partition('!')
        jobstring_parts = jobstring.split(':')
        triggerstr_parts = triggerstr.split(':')
        if not ((len(jobstring_parts) == 4) and (jobstring_parts[0] == "shepherd")):
            raise Exception(
                "Could not save scheduler job "+job.id+" - ID is not a valid Shepherd job")
        if not isinstance(job.trigger, CronTrigger):
            raise Exception("Could not save scheduler job " +
                            job.id+" - Trigger is not a CronTrigger")
        saved_jobs.append(job.id)
        if next_job_time is not None:
            if job.next_run_time < next_job_time:
                next_job_time = job.next_run_time
        else:
            
            next_job_time = job.next_run_time

    with open(joblist_path+".writing", 'w+') as f:
        for saved_job in saved_jobs:
            f.write("%s\n" % saved_job)
    
    os.rename(joblist_path+".writing", joblist_path)
    return next_job_time

# Currently to wakeup functionality is based on a hard-coded dependancy on the Scout
# plugin. We'd like this to instead be independant, and provide hooks for modules to register with that provide this
# A problem with just providing a "set_alarm" hook or something is that we really need to be able to
# confirm that an alarm has been set and we _are_ going to wake up gain correctly.
# Could potentially provide a interface function that others can call to set a "Next Alarm" state variable. This can then be
# checked after the hook call to verify.


# At the moment shutdown is just triggered 1 minute after setting the alarm - mostly just to allow time for things to upload.
# Having things be properly event based instead would be better - adding random delays all around the place inevetiably leads
# "delay creep" - where all the "just in case" delays just build up and up and wind up making the whole thing take ages to do anything.
# Instead we really need some sort of "current jobs/dependancies" queue thing - containing active stuff to be dealt with that should 
# hold the system active (sidenote - this same system could easily then be used to hold things on to access them remotely)
# Would ideally have a mechanism that makes it hard to accidentally leave an item/job in the list that stops the system shutting down - 
# maybe look at a context manager used on an object in each plugin?


def _jobs_changed(event):
    next_job_time = save_jobs()
    # default to idle for 5 mins

    early_wakeup_period = timedelta(minutes=1)

    now_datetime = datetime.now(get_localzone())
    next_idle_period = timedelta(minutes=5)
    if next_job_time is not None:
        
        next_idle_period = next_job_time - now_datetime

    if next_idle_period > timedelta(hours=6):
        next_idle_period = timedelta(hours=6)

    wakeup_time = None
    if next_idle_period > timedelta(minutes=4):
        wakeup_time = now_datetime+(next_idle_period-early_wakeup_period)
    
    if wakeup_time is not None:
        alarm_str = str(int(wakeup_time.timestamp()))
        print("waking up at "+ str(wakeup_time) + " ("+alarm_str+")")
        
        retval=shepherd.plugin.plugin_functions["scout"].set_alarm(alarm_str)
        if retval is not None:
            print(retval)
            if retval == alarm_str:
                if is_raspberry_pi():
                    print("Shutting down in 1 minute")
                    time.sleep(60)

                    subprocess.run(["shutdown","now"])
            else:
                print("Alarm set response was incorrect")
        else:
            print("Did not get reply from Scout after setting alarm")

def restore_jobs():
    pass


def start():
    global _scheduler
    _scheduler.add_listener(_jobs_changed, EVENT_JOB_ADDED | EVENT_JOB_REMOVED | EVENT_JOB_MODIFIED | EVENT_JOB_EXECUTED | EVENT_JOB_MISSED | EVENT_SCHEDULER_STARTED)
    _scheduler.start()