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): 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): self.log(self.entities) 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]) @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)}') 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() @dataclass(init=False) class ControllerDaylight(ControllerEntities): latitude: float longitude: float def initialize(self): super().initialize() 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.log(f'Listening for state {[e.friendly_name for e in self.entities]}') self.run_every(callback=self.update_sensors, start='now', interval=5.0) def handle_state_change(self, entity=None, attribute=None, old=None, new=None, kwargs=None): if new == 'on': self.adjustment_handle = self.run_every( callback=self.ongoing_adjustment, start='now', interval=10, entity=entity ) self.log(f'Started adjustments') else: if hasattr(self, 'adjustment_handle'): self.cancel_timer(self.adjustment_handle) self.log(f'Cancelled adjustments') def matching_state(self, entity_id: str): 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: return False else: valid = all((state[s] == val) for s, val in settings.items()) return valid def ongoing_adjustment(self, kwargs): self.log(f'Ongoing adjustment') settings = self.adjuster.current_settings valid = self.matching_state(entity_id=kwargs['entity']) if not valid: 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'})