281 lines
9.7 KiB
Python
281 lines
9.7 KiB
Python
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta
|
|
from typing import List
|
|
|
|
from appdaemon.entity import Entity
|
|
from appdaemon.plugins.hass.hassapi import Hass
|
|
from daylight_adjuster import DaylightAdjuster
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerEntities(Hass):
|
|
entities: List[Entity]
|
|
|
|
def initialize(self):
|
|
# assign fields
|
|
for arg, val in self.args.items():
|
|
if arg not in ['class', 'module']:
|
|
setattr(self, arg, val)
|
|
# self.log(f'Set {arg} to {val}')
|
|
|
|
for entity in self.entities:
|
|
assert self.entity_exists(entity), f'{entity} does not exist'
|
|
self.entities = [self.get_entity(e) for e in self.entities]
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerRoomLights(ControllerEntities):
|
|
sleep: str
|
|
|
|
def initialize(self):
|
|
super().initialize()
|
|
self.log(f'Initialized light controller for {[e.friendly_name for e in self.entities]}')
|
|
self.register_service(f'{self.name}/activate', self.activate)
|
|
self.register_service(f'{self.name}/deactivate', self.deactivate)
|
|
|
|
def activate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
|
|
# if self.is_sleeping:
|
|
# self.log(f'Sleep mode is on, returning early')
|
|
# return
|
|
|
|
for entity in self.entities:
|
|
self.log(f'Turning on {entity.name}')
|
|
entity.turn_on()
|
|
|
|
def deactivate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
|
|
self.log(self.entities)
|
|
for entity in self.entities:
|
|
self.log(f'Turning off {entity.name}')
|
|
entity.turn_off()
|
|
|
|
@property
|
|
def state(self) -> bool:
|
|
return any([e.get_state() == 'on' for e in self.entities])
|
|
|
|
@property
|
|
def is_sleeping(self) -> bool:
|
|
if 'sleep' in self.args:
|
|
return self.get_entity(self.args['sleep']).is_state('on')
|
|
else:
|
|
return False
|
|
|
|
@is_sleeping.setter
|
|
def is_sleeping(self, val: bool):
|
|
if 'sleep' in self.args:
|
|
self.get_entity(self.args['sleep']).set_state(state='on' if val else 'off')
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerMotion(ControllerEntities):
|
|
room: ControllerRoomLights
|
|
off_duration: timedelta
|
|
|
|
def initialize(self):
|
|
super().initialize()
|
|
# self.log('Motion Controller init')
|
|
|
|
# convert room to App
|
|
self.room: ControllerRoomLights = self.get_app(self.room)
|
|
|
|
# convert off_duration
|
|
try:
|
|
hours, minutes, seconds = map(int, self.args['off_duration'].split(':'))
|
|
self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
|
|
except Exception:
|
|
self.off_duration = timedelta()
|
|
|
|
if self.current_state:
|
|
self.room.activate()
|
|
|
|
self.listen_motion_off()
|
|
self.listen_motion_on()
|
|
|
|
@property
|
|
def current_state(self) -> bool:
|
|
return any(e.get_state() == 'on' for e in self.entities)
|
|
|
|
def listen_motion_on(self):
|
|
self.listen_state(
|
|
callback=self.callback_motion_on,
|
|
entity_id=[e.entity_id for e in self.entities],
|
|
new='on',
|
|
)
|
|
self.log(f'Waiting for motion on {[e.friendly_name for e in self.entities]} to turn on room {self.room.name}')
|
|
|
|
def listen_motion_off(self):
|
|
self.listen_state(
|
|
callback=self.callback_motion_off,
|
|
entity_id=[e.entity_id for e in self.entities],
|
|
new='off',
|
|
duration=self.off_duration.total_seconds(),
|
|
)
|
|
self.log(f'Waiting for motion off {[e.friendly_name for e in self.entities]} for {self.off_duration}')
|
|
|
|
def callback_motion_on(self, entity, attribute, old, new, kwargs):
|
|
self.log(f'Motion detected on {self.friendly_name(entity)}')
|
|
if not self.room.is_sleeping:
|
|
self.room.activate()
|
|
|
|
def callback_motion_off(self, entity, attribute, old, new, kwargs):
|
|
self.log(f'Motion stopped on {self.friendly_name(entity)} for {self.off_duration}')
|
|
self.room.deactivate()
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerButton(Hass):
|
|
room: ControllerRoomLights
|
|
buttons: List[str]
|
|
|
|
def initialize(self):
|
|
self.buttons = self.args['buttons']
|
|
|
|
# convert room to App
|
|
self.room: ControllerRoomLights = self.get_app(self.args['room'])
|
|
|
|
for button in self.buttons:
|
|
self.listen_event(
|
|
self.callback_button,
|
|
event='deconz_event',
|
|
id=button,
|
|
)
|
|
self.log(f'Listening to presses on button ID={button}')
|
|
|
|
def callback_button(self, event_name, data, kwargs):
|
|
# single press
|
|
if data['event'] == 1002:
|
|
self.log(f"Single press: {data['id']}")
|
|
if self.room.state:
|
|
self.room.deactivate()
|
|
else:
|
|
self.room.activate()
|
|
|
|
# double click
|
|
elif data['event'] == 1004:
|
|
self.log(f'{data["id"]} double click')
|
|
self.room.is_sleeping = not self.room.is_sleeping
|
|
if not self.room.is_sleeping:
|
|
self.room.activate()
|
|
|
|
|
|
@dataclass(init=False)
|
|
class ControllerDaylight(Hass):
|
|
room: ControllerRoomLights
|
|
entities: List[str]
|
|
latitude: float
|
|
longitude: float
|
|
|
|
def initialize(self):
|
|
# convert room to App
|
|
self.room: ControllerRoomLights = self.get_app(self.args['room'])
|
|
|
|
# convert entities
|
|
for entity in self.args['entities']:
|
|
assert self.entity_exists(entity), f'{entity} does not exist'
|
|
self.entities = [self.get_entity(e) for e in self.args['entities']]
|
|
# self.log(self.entities)
|
|
|
|
# create Adjuster
|
|
self.adjuster = DaylightAdjuster(
|
|
latitude=self.args['latitude'],
|
|
longitude=self.args['longitude'],
|
|
periods=self.args['periods'],
|
|
resolution=500
|
|
)
|
|
# self.log(self.adjuster)
|
|
|
|
self.listen_state(callback=self.handle_state_change,
|
|
entity_id=[e.entity_id for e in self.entities])
|
|
# self.listen_state(callback=self.handle_brightness_change,
|
|
# entity_id=[e.entity_id for e in self.entities],
|
|
# attribute='brightness')
|
|
# self.listen_state(callback=self.handle_brightness_change,
|
|
# entity_id=[e.entity_id for e in self.entities],
|
|
# attribute='color_temp')
|
|
ents = [e.friendly_name for e in self.entities]
|
|
if len(ents) > 1:
|
|
ents[-1] = f'and {ents[-1]}'
|
|
delim = ', ' if len(ents) >= 3 else ' '
|
|
self.log(f'Listening for state changes on {delim.join(ents)}')
|
|
|
|
self.run_every(callback=self.update_sensors, start='now', interval=5.0)
|
|
|
|
for entity in self.entities:
|
|
if entity.get_state() == 'on':
|
|
self.start_adjustments(entity.entity_id)
|
|
|
|
def handle_state_change(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
|
|
self.log(f'{entity}: {old} -> {new}')
|
|
if new == 'on':
|
|
if old == 'off':
|
|
self.start_adjustments(entity)
|
|
else:
|
|
self.stop_adjustments()
|
|
|
|
def handle_brightness_change(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
|
|
if not self.matching_state(entity):
|
|
self.log(f'{entity}.{attribute}: {old} -> {new}')
|
|
self.log(f'State does not match adjuster settings, disabling adjustments')
|
|
self.stop_adjustments()
|
|
|
|
def matching_state(self, entity_id: str) -> bool:
|
|
"""Checks whether the current state of the light matches the settings from the DaylightAdjuster
|
|
|
|
Args:
|
|
entity_id (str): full entity ID
|
|
|
|
Returns:
|
|
bool
|
|
"""
|
|
state = self.get_state(entity_id=entity_id, attribute='all')['attributes']
|
|
settings = self.adjuster.current_settings
|
|
try:
|
|
state = {s: state[s] for s in settings.keys()}
|
|
except KeyError:
|
|
for s in settings.keys():
|
|
if s not in state:
|
|
self.log(f'{s} not in {state}')
|
|
return False
|
|
else:
|
|
valid = all((state[s] == val) for s, val in settings.items())
|
|
if not valid:
|
|
for s, val in settings.items():
|
|
if state[s] != val:
|
|
self.log(f'{entity_id}.{s}: {state[s]} != {val}')
|
|
return valid
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return hasattr(self.adjustment_handle)
|
|
|
|
def start_adjustments(self, entity: str):
|
|
self.adjustment_handle = self.run_every(
|
|
callback=self.ongoing_adjustment,
|
|
start='now',
|
|
interval=10,
|
|
entity=entity
|
|
)
|
|
self.log(f'Started adjustments')
|
|
|
|
def stop_adjustments(self):
|
|
if hasattr(self, 'adjustment_handle'):
|
|
self.cancel_timer(self.adjustment_handle)
|
|
del self.adjustment_handle
|
|
self.log(f'Cancelled adjustments')
|
|
|
|
def ongoing_adjustment(self, kwargs):
|
|
self.log(f'Ongoing adjustment')
|
|
settings = self.adjuster.current_settings
|
|
matching = self.matching_state(entity_id=kwargs['entity'])
|
|
if not matching and not self.room.is_sleeping:
|
|
self.turn_on(entity_id=kwargs['entity'], **settings)
|
|
self.log(f'Adjusted {self.friendly_name(kwargs["entity"])} with {settings}')
|
|
|
|
def update_sensors(self, kwargs):
|
|
for key, val in self.adjuster.current_settings.items():
|
|
id = f'sensor.{self.name}_{key}'
|
|
self.set_state(
|
|
entity_id=id, state=val,
|
|
attributes={
|
|
'friendly_name': f'Daylight, {key}, {self.name}',
|
|
'state_class': 'measurement'})
|