from dataclasses import dataclass, field from datetime import datetime, timedelta import logging 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_off, entity_id=[e.entity_id for e in self.entities], new='off') self.listen_state(callback=self.handle_state_change, entity_id=[e.entity_id for e in self.entities], attribute='brightness') self.listen_state(callback=self.handle_state_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 ' ' ents = delim.join(ents) self.log(f'Listening for state changes on {ents}') interval = self.args.get('interval', 5) self.run_every( callback=self.update_sensors, start='now', interval=interval ) self.log(f'Updating sensors every {timedelta(seconds=interval)}') for entity in self.entities: self.run_every( callback=self.ongoing_adjustment, start='now', interval=interval, entity=entity.entity_id ) self.log(f'Updating sensors every {timedelta(seconds=interval)}') if (entity_name := self.args.get('enable')) is not None: self.enable_entity: Entity = self.get_entity(entity_name) self.log(f'enabled by {self.enable_entity.friendly_name}[{entity_name}]') self.listen_state( callback=lambda entity, attribute, old, new, kwargs: self.ongoing_adjustment({'entity': entity}), entity_id=entity_name, new='on' ) def handle_state_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.enabled = False 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: if hasattr(self, 'enable_entity'): return self.enable_entity.is_state('on') else: return True @enabled.setter def enabled(self, new: bool) -> bool: self.enable_entity.set_state(state='on' if new else 'off') def ongoing_adjustment(self, kwargs=None): # self.log('Ongoing adjustment') entity: Entity = self.get_entity(kwargs['entity']) if self.enabled: if entity.get_state() == 'on': self.log(f'Ongoing adjustment for {entity.friendly_name}') settings = self.adjuster.current_settings if not self.matching_state(entity_id=entity.entity_id): if not self.room.is_sleeping: self.turn_on(entity_id=entity.entity_id, **settings) self.log(f'Adjusted {entity.friendly_name} with {settings}') else: self.log(f'Sleeping mode active') else: self.log(f'{entity.friendly_name} settings already match') else: self.log(f'{entity.friendly_name} is off - no adjustment') else: self.log(f'App disabled by {self.enable_entity.friendly_name}') 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'}) def handle_off(self, entity, attribute, old, new, kwargs): self.log('Off handle') self.run_in( callback=lambda kwargs: self.enable_entity.set_state(state='on'), delay=1.0 ) self.log('Re-enabled')