1
0
Fork 0

replace schedule rules with automated range calculations

master
Bob Carroll 2024-01-16 21:18:01 -06:00
parent 8dd1f1766a
commit eb2a06aa9f
4 changed files with 246 additions and 204 deletions

View File

@ -18,187 +18,258 @@
import sys import sys
import asyncio import asyncio
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from datetime import datetime, timedelta from collections import namedtuple
import logging import logging
import yaml import yaml
from dns.asyncresolver import resolve from dns.asyncresolver import resolve
from asyncio_mqtt import Client from asyncio_mqtt import Client
import umsgpack import umsgpack
from dateutil.parser import parse
Rule = namedtuple('Rule', 'bri ct')
async def apply_rule(rule, client): class SceneManager(object):
""" """
Sets the scene from a rule. Manages light state changes.
:param rule: a rule dictionary
:param client: mqtt client
"""
name = rule.get('name')
logging.debug(f'Begin applying rule: {name}')
for light, payload in rule.get('scene', {}).items():
payload = {k: v
for k, v in payload.items()
if k in ['mode', 'brightness', 'effect']}
if not len(payload):
logging.warn(f'Light {light} payload is malformed')
continue
logging.debug(f'Setting state for light {light}: {payload}')
await client.publish(f'light/{light}/state/set', umsgpack.packb(payload))
logging.debug(f'End applying rule: {name}')
async def check_elevation(trigger, state):
"""
Determines if the current sun elevation matches the trigger.
:param trigger: a trigger dictionary
:param state: shared state dictionary
:returns: True on match, False otherwise
"""
elevation = state.get('elevation')
is_rising = state.get('rising')
at = trigger.get('at', 0)
until = trigger.get('until', 0)
on_rising = at <= until
if elevation is None or is_rising is None:
return False
elif is_rising and (not on_rising or elevation < at or elevation >= until):
return False
elif not is_rising and (on_rising or elevation > at or elevation <= until):
return False
return True
async def check_time(trigger):
"""
Determines if the current time matches the trigger.
:param trigger: a trigger dictionary
:returns: True on match, False otherwise
"""
at_time = parse(trigger.get('at'))
before = at_time + timedelta(minutes=trigger.get('for', 1))
return at_time <= datetime.now() < before
async def match_trigger(trigger, state):
"""
Looks for matching triggers.
:param trigger: a trigger dictionary
:param state: shared state dictionary
:returns: True on match, False otherwise
"""
type_ = trigger.get('type')
if type_ == 'elevation':
return await check_elevation(trigger, state)
elif type_ == 'time':
return await check_time(trigger)
else:
logging.warn(f'Unknown trigger type: {type_}')
return False
async def periodic_rule_check(client, state, mutex):
"""
Periodically checks for a matching rule to apply.
:param client: mqtt client :param client: mqtt client
:param rule: new light state values
:param state: shared state dictionary :param state: shared state dictionary
:param mutex: lock for making changes to shared state
""" """
async with mutex:
rules = state['rules']
scene_name = state['scene-name']
while True: def __init__(self, client, rule, state):
logging.debug('Begin rule check') self.client = client
await mutex.acquire() self.rule = rule
active = state.get('active', -1) self.elevation = state['solar']['elevation']
self.direction = state['solar']['direction']
self.auto_on = state['options'].get('auto_on', False)
for i, r in enumerate(rules): def get_auto_on_config(self, options):
if active > -1 and i != active: """
logging.debug('Skipping, another rule is active') Gets the solar elevation and direction settings for a light.
continue
try: :param options: light configuration
logging.debug(f'Checking rule: {r.get("name")}') :returns: a tuple of elevation and direction values
matched = await match_trigger(r.get('trigger', {}), state) """
return options.get('auto_on_elevation'), options.get('auto_on_direction', 1)
if matched and active == i: def check_auto_on(self, options):
logging.debug('Rule has already been triggered') """
break Determiens whether a light should automatically turn on based on configured and current
elif matched: solar elevation and direction.
logging.info(f'Rule triggered: {r.get("name")}')
await apply_rule(r, client)
await client.publish('scene/activate', scene_name)
active = i
break
elif active == i:
logging.debug('Resetting, rule no longer matches')
active = -1
except Exception as ex:
logging.error(str(ex))
break
state['active'] = active :param options: light configuration
mutex.release() :returns: True if the light should turn on, False otherwise
logging.debug('End rule check') """
await asyncio.sleep(30) elevation, direction = self.get_auto_on_config(options)
if self.auto_on and elevation is not None and self.direction == direction:
tod = 'morning' if direction else 'evening'
logging.debug(f'Auto-on will trigger in the {tod} at {elevation} degrees')
if direction and self.elevation >= elevation:
return True
elif not direction and self.elevation <= elevation:
return True
return False
def adjust_brightness(self, state):
"""
Adjusts the rule brightness value given an offset.
:param state: light state dictionary
:returns: new brightness value
"""
offset = state['options'].get('brightness_offset', 0)
return max(min(self.rule.bri + offset, 100), 0)
def adjust_colour_temp(self, state):
"""
Adjusts the rule colour temperature value given an offset.
:param state: light state dictionary
:returns: new colour temperature value
"""
offset = state['options'].get('colour_temp_offset', 0)
return max(min(self.rule.ct + offset, 10000), 0)
async def turn_on(self, name, state):
"""
Turns on a light based on current state and configuration.
:param name: light name
:param state: light state dictionary
"""
bri = self.adjust_brightness(state)
ct = self.adjust_colour_temp(state)
payload = {'mode': 'dim',
'brightness': bri,
'effect': {'type': 'temperature',
'value': f'{ct}K'}}
if state.get('on', False) or self.check_auto_on(state['options']):
logging.debug(f'Setting state for light {name}: {payload}')
await self.client.publish(f'light/{name}/state/set', umsgpack.packb(payload))
else:
logging.debug(f'Skipping light {name} because it is off')
async def on_sun_event(state, mutex, messages): def match_step_rule(rules, current):
""" """
Event handler for sun state changes. Gets the closest rule for the given index.
:param rules: dictionary of step rules
:param current: the current step index
:returns: the matched rule
"""
return rules[min(rules.keys(), key=lambda k: abs(k - current))]
async def update_light_states(client, state):
"""
Updates the state of all lights based on the solar state.
:param client mqtt client
:param state: shared state dictionary
"""
if state['paused']:
logging.debug('Circadian automation is paused, skipping')
return
elif state['solar']['elevation'] < 0:
logging.debug('Sun is below the horizon, skipping')
return
index = (state['solar']['elevation'] / state['solar']['noon']) * 100
logging.debug(f'Current step index: {index}')
rule = match_step_rule(state['rules'], index)
logging.debug(f'Matched step rule: {rule}')
scene = SceneManager(client, rule, state)
for light_name, light_state in state['lights'].items():
await scene.turn_on(light_name, light_state)
await client.publish('scene/activate', state['scene_name'])
async def on_solar_state_change(client, state, mutex, messages):
"""
Event handler for receiving solar state changes.
:param client: mqtt client
:param state: shared state dictionary :param state: shared state dictionary
:param mutex: lock for making changes to shared state :param mutex: lock for making changes to shared state
:param messages: mqtt message generator :param messages: mqtt message generator
""" """
async for m in messages: async for m in messages:
try: try:
_, event = m.topic.split('/')
async with mutex: async with mutex:
if event == 'elevation': state['solar'] = umsgpack.unpackb(m.payload)
state['elevation'] = round(float(m.payload), 1) logging.debug(f'Received solar state: {state["solar"]}')
logging.debug(f'Received sun elevation: {state["elevation"]}') await update_light_states(client, state)
elif event == 'rising':
state['rising'] = bool(int(m.payload))
logging.debug(f'Received sun rising state: {state["rising"]}')
except Exception as ex: except Exception as ex:
logging.error(str(ex)) logging.error(str(ex))
async def on_reset(state, mutex, messages): async def on_light_event(client, state, mutex, messages):
""" """
Handler for resetting the active scene. Event handler for receiving light 'on' changes.
:param client: mqtt client
:param state: shared state dictionary :param state: shared state dictionary
:param mutex: lock for making changes to shared state :param mutex: lock for making changes to shared state
:param messages: mqtt message generator :param messages: mqtt message generator
""" """
async for m in messages: async for m in messages:
try: try:
if int(m.payload) == 1: async with mutex:
logging.debug('Resetting active scene') _, name, _ = m.topic.split('/', 3)
async with mutex: logging.debug(f'Received light {name} state: {m.payload}')
state['active'] = -1
if name in state['lights']:
current = state['lights'][name]['on']
state['lights'][name]['on'] = bool(int(m.payload))
if not current and 'solar' in state:
await update_light_states(client, state)
except Exception as ex: except Exception as ex:
logging.error(str(ex)) logging.error(str(ex))
async def on_pause(client, state, mutex, messages):
"""
Handler for pausing and unpausing the automation.
:param client: mqtt client
:param state: shared state dictionary
:param mutex: lock for making changes to shared state
:param messages: mqtt message generator
"""
async for m in messages:
try:
async with mutex:
logging.debug(f'Setting paused state: {m.payload}')
state['paused'] = bool(int(m.payload))
await client.publish('circadian/paused', m.payload, retain=True)
except Exception as ex:
logging.error(str(ex))
def compute_steps(range_min, range_max, count):
"""
Computes the intervals given a range and step.
:param range_min:
:param range_max:
:param count: the number of steps to iterate over
:returns: a list of intervals
"""
step = (range_max - range_min) / count
result = []
i = 0
while range_min + i * step <= range_max:
result.append(round(range_min + i * step))
i += 1
return result
def make_rule_map(config):
"""
Builds a mapping of step rules to light state rules.
:param config: configuration dictionary
:returns: a dictionary of rules
"""
steps = min(max(config.get('change_steps', 1), 1), 100)
elev = compute_steps(0, 100, steps)
bri = compute_steps(max(config.get('min_brightness', 0), 0),
min(config.get('max_brightness', 100), 100),
steps)
ct = compute_steps(max(config.get('min_colour_temp', 0), 0),
min(config.get('max_colour_temp', 10000), 10000),
steps)
return {e: Rule(b, t) for e, b, t in zip(elev, bri, ct)}
def make_light_map(config):
"""
Builds a mapping of light options and state.
:param config: configuration dictionary
:returns: a dictionary of light states
"""
def transform(in_):
options = {k: v for k, v in in_.items() if k != 'name'}
return {'options': options, 'on': False}
return {x['name']: transform(x) for x in config}
async def get_broker(config): async def get_broker(config):
""" """
Gets the mqtt broker address from an SRV record. Gets the mqtt broker address from an SRV record.
@ -223,8 +294,11 @@ async def init(config):
:param config: configuration dictionary :param config: configuration dictionary
""" """
state = {'rules': config.get('rules', []), state = {'lights': make_light_map(config.get('lights', [])),
'scene-name': config.get('scene-name', 'Circadian')} 'rules': make_rule_map(config.get('options', {})),
'scene_name': config.get('scene_name', 'Circadian'),
'options': config.get('options', {}),
'paused': False}
mutex = asyncio.Lock() mutex = asyncio.Lock()
tasks = set() tasks = set()
@ -234,22 +308,20 @@ async def init(config):
logging.info('Connected to mqtt broker') logging.info('Connected to mqtt broker')
topics = { topics = {
'circadian/reset': on_reset, 'circadian/paused/set': on_pause,
'sun/elevation': on_sun_event, 'light/+/on': on_light_event,
'sun/rising': on_sun_event} 'sun/state': on_solar_state_change}
for t, cb in topics.items(): for t, cb in topics.items():
manager = client.filtered_messages(t) manager = client.filtered_messages(t)
messages = await stack.enter_async_context(manager) messages = await stack.enter_async_context(manager)
task = asyncio.create_task(cb(state, mutex, messages)) task = asyncio.create_task(cb(client, state, mutex, messages))
tasks.add(task) tasks.add(task)
await client.subscribe('circadian/#') await client.subscribe('circadian/#')
await client.publish('circadian/paused', 0, retain=True)
await client.subscribe('light/#')
await client.subscribe('sun/#') await client.subscribe('sun/#')
task = asyncio.create_task(periodic_rule_check(client, state, mutex))
tasks.add(task)
await asyncio.gather(*tasks) await asyncio.gather(*tasks)

View File

@ -7,58 +7,30 @@ log:
# CRITICAL, ERROR, WARNING, INFO, DEBUG # CRITICAL, ERROR, WARNING, INFO, DEBUG
level: INFO level: INFO
scene-name: Circadian options:
# Automatically turn lights on at specific elevations
auto_on: true
# Number of changes to make between sunrise/sunset and noon
change_steps: 20
# Maxiumum brightness percentage (0-100)
max_brightness: 100
# Maxium colour temperature in kelvin (for Hue, 2000-6500)
max_colour_temp: 6000
# Minimum brightness percentage (0-100)
min_brightness: 60
# Minimum colour temperature in kelvin (for Hue, 2000-6500)
min_colour_temp: 3000
# Name of the scene to annouce on elevation change
scene_name: Circadian
rules: lights:
- name: Early Morning - name: Nightstand
trigger: - name: Floor Lamp
type: elevation # Offset in degrees to make lights brighter or dimmer
at: 0.0 adjust_brightness: 10
until: 6.0 # Offset in kelvin to make lights warmer or cooler
scene: adjust_colour_temp: -500
- topic: hue/light/5/set # Solar elevation for when this light should be turned on
data: auto_on_elevation: 30
'on': true # 1 for rising, 0 for setting
bri: 128 auto_on_direction: 0
ct_k: 4000
- name: Late Morning
trigger:
type: elevation
at: 18.0
until: 30.0
scene:
- topic: hue/light/5/bri
data: 255
- name: Lunch
trigger:
type: time
at: '12:00'
# minutes
for: 60
scene:
- topic: hue/light/8/set
data:
'on': true
bri: 224
ct_k: 3800
- name: Late Afternoon
trigger:
type: elevation
at: 18.0
until: 12.0
scene:
- topic: hue/light/5/set
data:
'on': true
bri: 255
ct_k: 3600
- topic: hue/group/6/set
data:
'on': true
bri: 255
ct_k: 3600
- topic: hue/light/8/on
data: 0

View File

@ -2,7 +2,6 @@
asyncio-mqtt==0.8.0 asyncio-mqtt==0.8.0
dnspython==2.1.0 dnspython==2.1.0
paho-mqtt==1.5.1 paho-mqtt==1.5.1
python-dateutil==2.8.1
PyYAML==5.3.1 PyYAML==5.3.1
six==1.15.0 six==1.15.0
u-msgpack-python==2.7.1 u-msgpack-python==2.7.1

View File

@ -2,7 +2,7 @@
from setuptools import setup, find_namespace_packages from setuptools import setup, find_namespace_packages
setup(name='cherry-circadian', setup(name='cherry-circadian',
version='0.4', version='0.5',
description='adapts lighting based on the sun movement', description='adapts lighting based on the sun movement',
author='Bob Carroll', author='Bob Carroll',
author_email='bob.carroll@alum.rit.edu', author_email='bob.carroll@alum.rit.edu',
@ -11,8 +11,7 @@ setup(name='cherry-circadian',
'pyyaml', 'pyyaml',
'dnspython', 'dnspython',
'asyncio_mqtt', 'asyncio_mqtt',
'u-msgpack-python', 'u-msgpack-python'],
'python-dateutil'],
classifiers=[ classifiers=[
'Development Status :: 2 - Pre-Alpha', 'Development Status :: 2 - Pre-Alpha',
'Environment :: Console', 'Environment :: Console',