273 lines
8.0 KiB
Python
273 lines
8.0 KiB
Python
# cherry-circadian - adapts lighting based on the sun movement
|
|
# Copyright (C) 2021 Bob Carroll <bob.carroll@alum.rit.edu>
|
|
#
|
|
# This program is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU General Public License
|
|
# as published by the Free Software Foundation; either version 2
|
|
# of the License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software
|
|
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
|
|
|
import sys
|
|
import asyncio
|
|
from contextlib import AsyncExitStack
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
|
|
import yaml
|
|
from dns.asyncresolver import resolve
|
|
from asyncio_mqtt import Client
|
|
import umsgpack
|
|
from dateutil.parser import parse
|
|
|
|
|
|
async def apply_rule(rule, client):
|
|
"""
|
|
Sets the scene from a rule.
|
|
|
|
:param rule: a rule dictionary
|
|
:param client: mqtt client
|
|
"""
|
|
name = rule.get('name')
|
|
logging.debug(f'Begin applying rule: {name}')
|
|
|
|
for s in rule.get('scene', []):
|
|
topic = s.get('topic')
|
|
data = s.get('data')
|
|
retain = s.get('retain', False)
|
|
|
|
logging.debug(f'Publishing to {topic}: {data}')
|
|
if topic is None or data is None:
|
|
continue
|
|
elif isinstance(data, dict):
|
|
data = umsgpack.packb(data)
|
|
|
|
await client.publish(topic, data, retain=retain)
|
|
|
|
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 state: shared state dictionary
|
|
:param mutex: lock for making changes to shared state
|
|
"""
|
|
async with mutex:
|
|
rules = state['rules']
|
|
|
|
while True:
|
|
logging.debug('Begin rule check')
|
|
await mutex.acquire()
|
|
active = state.get('active', -1)
|
|
|
|
for i, r in enumerate(rules):
|
|
if active > -1 and i != active:
|
|
logging.debug('Skipping, another rule is active')
|
|
continue
|
|
|
|
try:
|
|
logging.debug(f'Checking rule: {r.get("name")}')
|
|
matched = await match_trigger(r.get('trigger', {}), state)
|
|
|
|
if matched and active == i:
|
|
logging.debug('Rule has already been triggered')
|
|
break
|
|
elif matched:
|
|
logging.info(f'Rule triggered: {r.get("name")}')
|
|
await apply_rule(r, client)
|
|
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
|
|
mutex.release()
|
|
logging.debug('End rule check')
|
|
await asyncio.sleep(30)
|
|
|
|
|
|
async def on_sun_event(state, mutex, messages):
|
|
"""
|
|
Event handler for sun state changes.
|
|
|
|
: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:
|
|
_, event = m.topic.split('/')
|
|
|
|
async with mutex:
|
|
if event == 'elevation':
|
|
state['elevation'] = round(float(m.payload), 1)
|
|
logging.debug(f'Received sun elevation: {state["elevation"]}')
|
|
elif event == 'rising':
|
|
state['rising'] = bool(int(m.payload))
|
|
logging.debug(f'Received sun rising state: {state["rising"]}')
|
|
except Exception as ex:
|
|
logging.error(str(ex))
|
|
|
|
|
|
async def on_reset(state, mutex, messages):
|
|
"""
|
|
Handler for resetting the active scene.
|
|
|
|
: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:
|
|
if int(m.payload) == 1:
|
|
logging.debug('Resetting active scene')
|
|
async with mutex:
|
|
state['active'] = -1
|
|
except Exception as ex:
|
|
logging.error(str(ex))
|
|
|
|
|
|
async def get_broker(config):
|
|
"""
|
|
Gets the mqtt broker address from an SRV record.
|
|
|
|
:param config: configuration dictionary
|
|
:returns: the broker address
|
|
"""
|
|
broker = config.get('mqtt', {}).get('broker')
|
|
if broker is not None:
|
|
logging.debug(f'Using pre-defined broker: {broker}')
|
|
return broker
|
|
|
|
answer = await resolve('_mqtt._tcp', 'SRV', search=True)
|
|
broker = next((x.target.to_text() for x in answer))
|
|
logging.debug(f'Found SRV record: {broker}')
|
|
return broker
|
|
|
|
|
|
async def init(config):
|
|
"""
|
|
Initializes the circadian lighting agent.
|
|
|
|
:param config: configuration dictionary
|
|
"""
|
|
state = {'rules': config.get('rules', [])}
|
|
mutex = asyncio.Lock()
|
|
tasks = set()
|
|
|
|
async with AsyncExitStack() as stack:
|
|
client = Client(await get_broker(config), client_id='cherry-circadian')
|
|
await stack.enter_async_context(client)
|
|
logging.info('Connected to mqtt broker')
|
|
|
|
topics = {
|
|
'circadian/reset': on_reset,
|
|
'sun/elevation': on_sun_event,
|
|
'sun/rising': on_sun_event}
|
|
|
|
for t, cb in topics.items():
|
|
manager = client.filtered_messages(t)
|
|
messages = await stack.enter_async_context(manager)
|
|
task = asyncio.create_task(cb(state, mutex, messages))
|
|
tasks.add(task)
|
|
|
|
await client.subscribe('circadian/#')
|
|
await client.subscribe('sun/#')
|
|
|
|
task = asyncio.create_task(periodic_rule_check(client, state, mutex))
|
|
tasks.add(task)
|
|
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
def main():
|
|
"""
|
|
CLI entry point.
|
|
"""
|
|
if len(sys.argv) != 2:
|
|
print('USAGE: cherry-circadian <config file>')
|
|
sys.exit(1)
|
|
|
|
with open(sys.argv[1], 'r') as f:
|
|
config = yaml.safe_load(f)
|
|
|
|
log = config.get('log', {})
|
|
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s',
|
|
level=log.get('level', logging.ERROR))
|
|
|
|
try:
|
|
asyncio.run(init(config))
|
|
except KeyboardInterrupt:
|
|
pass
|