Automating Hue lights at dusk/dawn, or scheduled

hue_automation.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys                         
import time
import astral
import logging
import datetime
from dateutil import parser
from threading import Thread
from beautifulhue.api import Bridge

rules = {
    'Outdoor lights on':{
        'lights':['Front Door','Back door'],
        'status':'on',
        'when':'dusk',
        'brightness':254,
        'xy':[0.4643, 0.4115]
    },
    'Lights out at sunrise':{
        'lights':['all'],
        'status':'off',
        'when':'sunrise',
        'brightness':254,
    },
    'Lights down - kids room':{
        'lights':['Kids room 1','Kids room 2'],
        'status':'on',
        'when':'18:45',
        'brightness':10,
        'xy':[0.51, 0.4148]
    }
}

class LightScheduler(Thread):
    def __init__(self):
        super(LightScheduler, self).__init__()
        self.daemon=True
        self.bridge = Bridge(device={'ip':'hue'},user={'name':'huetestuser'})

    def run(self):
        while True:
            next_time,next_rules = self.next_activation()
            logging.info("next activation : "+str(self.next_activation()))
            now = datetime.datetime.now()
            waittime = (next_time-now).total_seconds()
            logging.info('Executing rules '+str(next_rules)+' in '+str(waittime)+' seconds')
            time.sleep(waittime)
            try:
                for r in next_rules:
                    self.run_rule(r)
            except Exception as e:
                logging.exception(e)

    def when_to_absolute(self,when):
        if when in ['dusk','dawn','sunrise','sunset']:
            a = astral.Astral()
            a.solar_depression = 'civil'
            sun = a['Copenhagen'].sun(datetime.datetime.now(),local=True)
            d = sun[when].replace(tzinfo=None)
            if d < datetime.datetime.now():
                logging.info("using tomorrows date for "+when)
                sun = a['Copenhagen'].sun(datetime.date.today() + datetime.timedelta(days=1),local=True)
            return sun[when].replace(tzinfo=None)

        try:
            absolute = parser.parse(when)
            if absolute < datetime.datetime.now():
                absolute += datetime.timedelta(days=1)
            return absolute.replace(tzinfo=None)
        except Exception as e:
            logging.exception(e)

        raise 'Unable to parse activation time \"'+when+'\"'

    def next_activation(self):
        now = datetime.datetime.time(datetime.datetime.now())
        next_time = None
        next_rule = []
        for r in rules:
            rule = rules[r]
            when = self.when_to_absolute(rule['when'])
            if next_time is None or when < next_time:
                next_time = when
                next_rule = [r]
            elif when == next_time:
                next_rule.append(r)

        return next_time,next_rule

    def run_rule(self,rule_name):
        r = rules[rule_name]
        logging.info('Running rule :'+rule_name+':'+str(r))
        for l in r['lights']:
            status = r['status']
            brightness = r['brightness']
            xy = None
            if 'xy' in r:
                xy = r['xy']
            self.update_light(l,status,brightness,xy)

    def update_light(self,name='all',state=False,brightness=254, xy=None):
        resource = {'which':'all'}

        if state == 'on':
            to_state = True
        else:
            to_state = False

        lights = self.bridge.light.get({'which':'all'})
        for light in lights['resource']:
            l = self.bridge.light.get({'which':light['id']})
            l = l['resource']
            if l['name'] == name or name == 'all':
                resource = {'which':light['id'],'data':{'state':{'on':to_state, 'ct':brightness}}}
                if xy is not None:
                    resource['data']['state']['xy'] = xy
                self.bridge.light.update(resource)


def main():
    logging.basicConfig(filename='hue_automation.log',level=logging.DEBUG)
    t = LightScheduler()
    t.start()
    while True:
        time.sleep(1)


if __name__ == '__main__':
    main()