from copy import deepcopy from datetime import time, timedelta from typing import Dict, List, Tuple, Union from appdaemon.plugins.hass.hassapi import Hass from continuous import Continuous class MotionLight(Hass): """Class for linking an light with a motion sensor. - Separate the turning on and turning off functions. - Use the state change of the light to set up the event for changing to the other state - `handle_on` - `handle_off` - When the light comes on, check if it's attributes match what they should, given the time. """ def initialize(self): self.state_change_handle = self.listen_state(self.state_change, self.entity) self.sync_state() self.schedule_transitions() self.run_daily(callback=self.schedule_transitions, start='00:00:00') if (button := self.args.get('button')): if not isinstance(button, list): button = [button] for button in button: self.log(f'Setting up button: {button}') self.listen_event(self.handle_button, event='deconz_event', id=button) if (door := self.args.get('door')): self.door_entity = self.get_entity(door) self.log(f'Setting up door: {self.door_entity.friendly_name}') self.listen_state( callback=self.activate_all_off, entity_id=door, new='on' ) self.log(f'All off: {self.all_off}') @property def sensor(self) -> str: return self.args['sensor'] @property def entity(self) -> str: return self.args['entity'] @property def off_duration(self) -> timedelta: """Determines the time that the motion sensor has to be clear before deactivating Priority: - Value in scene definition - Default value - Normal - value in app definition - Sleep - 0 """ if self.sleeping_active: _, duration_str = self.sleep_scene() else: _, _, duration_str = self.current_setting() try: hours, minutes, seconds = map(int, duration_str.split(':')) return timedelta(hours=hours, minutes=minutes, seconds=seconds) except Exception: return timedelta() @property def entity_state(self): return self.get_state(self.entity) == 'on' @entity_state.setter def entity_state(self, new): if isinstance(new, str): if new == 'on': self.turn_on(self.entity) elif new == 'off': self.turn_on(self.entity) else: raise ValueError(f'Invalid value for entity state: {new}') elif isinstance(new, bool): if new: self.turn_on(self.entity) self.log(f'Turned on {self.friendly_name(self.entity)}') else: self.turn_off(self.entity) self.log(f'Turned off {self.friendly_name(self.entity)}') elif isinstance(new, dict): if any(isinstance(val, dict) for val in new.values()): # self.log(f'Setting scene with nested dict: {new}') for entity, state in new.items(): if state.pop('state', 'on') == 'on': # self.log(f'Setting {entity} state with: {state}') self.turn_on(entity_id=entity, **state) else: self.turn_off(entity) else: if new.pop('state', 'on') == 'on': self.turn_on(self.entity, **new) else: self.turn_off(self.entity) else: raise TypeError(f'Invalid type: {type(new)}: {new}') @property def is_stateful(self): return 'scene' in self.args and isinstance(self.args['scene'], (list, dict)) def settings(self) -> List[Tuple[time, Union[Dict, str], timedelta]]: """Gets the settings for all the scenes based on time of day "Settings" refers to `tuple` groups that consist of - Scene start time - Dictionary of states or scene entity name - Time for motion to be off Returns: List[Tuple[time, Union[Dict, str], timedelta]]: Sorted list of settings """ assert self.is_stateful return sorted( (( self.parse_time(s['time']), s['scene'], s.get('off_duration', self.args.get('off_duration', '00:00:00')) ) for s in self.args['scene']), ) def current_setting(self) -> Tuple[time, Dict, timedelta]: assert self.is_stateful for dt, scene, off_duration in self.settings()[::-1]: if dt <= self.time(): # self.log(f'Active scene: {str(self.time())[:8]} {str(dt)[:8]}, {scene}, {off_duration}') return dt, scene, off_duration else: self.log('Setting last scene') return self.settings()[-1] def current_scene(self) -> Union[str, Dict]: if self.sleeping_active: return self.sleep_scene()[0] else: if self.is_stateful: dt, scene, _ = self.current_setting() self.log(f'Current scene: {str(dt)[:8]}, {scene}') return scene else: return self.args['scene'] def sleep_scene(self) -> Tuple[Dict, timedelta]: if (scene := self.args.get('sleep_scene')): scene = deepcopy(scene) if isinstance(scene, dict): off_duration = scene.pop('off_duration', '00:00:00') else: off_duration = '00:00:00' return scene, off_duration else: return None, None @property def app_entities(self): def gen(): for settings in deepcopy(self.args['scene']): # dt = self.parse_time(settings.pop('time')) if (scene := settings.get('scene')): if isinstance(scene, str): yield from self.get_entity(scene).get_state('all')['attributes']['entity_id'] else: yield from scene.keys() else: yield self.args['entity'] return list(set(gen())) @property def all_off(self) -> bool: """"All off" is the logic opposite of "any on" Returns: bool: Whether all the lights associated with the app are off """ return all(self.get_state(entity) != 'on' for entity in self.app_entities) @property def any_on(self) -> bool: """"Any on" is the logic opposite of "all off" Returns: bool: Whether any of the lights associated with the app are on """ return any(self.get_state(entity) == 'on' for entity in self.app_entities) @property def delay(self) -> timedelta: try: hours, minutes, seconds = map(int, self.args['delay'].split(':')) return timedelta(hours=hours, minutes=minutes, seconds=seconds) except Exception: return timedelta() @property def sleeping_active(self) -> bool: if 'sleep' in self.args: return self.get_state(self.args['sleep']) == 'on' else: return False @property def sleep_bool(self) -> bool: if (sleep_var := self.args.get('sleep')): return self.get_state(sleep_var) == 'on' @sleep_bool.setter def sleep_bool(self, val) -> bool: if (sleep_var := self.args.get('sleep')): if isinstance(val, str): self.set_state(sleep_var, state=val) elif isinstance(val, bool): self.set_state(sleep_var, state='on' if val else 'off') else: raise ValueError('Sleep variable is undefined') def sync_state(self): """Synchronizes the callbacks with the state of the light. Essentially mimics the `state_change` callback based on the current state of the light. """ if self.entity_state: self.callback_light_on() else: self.callback_light_off() def listen_motion_on(self): self.log(f'Waiting for motion on {self.friendly_name(self.sensor)} to turn on {self.friendly_name(self.entity)}') self.motion_on_handle = self.listen_state( callback=self.activate, entity_id=self.sensor, new='on', oneshot=True ) def listen_motion_off(self, duration: timedelta): self.log(f'Waiting for motion to stop on {self.friendly_name(self.sensor)} for {duration} to turn off {self.friendly_name(self.entity)}') self.motion_off_handle = self.listen_state( callback=self.deactivate, entity_id=self.sensor, new='off', duration=duration.total_seconds(), oneshot=True ) def cancel_motion(self, handle_name): if hasattr(self, handle_name): handle = getattr(self, handle_name) try: self.log(f'{handle_name}: {self.info_listen_state(handle)}') except ValueError: self.log(f'Error getting {handle_name} info') else: self.cancel_listen_state(handle) self.log(f'Cancelled handle: {handle_name}') finally: delattr(self, handle_name) else: self.log(f'No attribute: {handle_name}') def state_change(self, entity=None, attribute=None, old=None, new=None, kwargs=None): """Callback attached to the state change of the light. """ if new == 'on': self.callback_light_on(entity, attribute, old, new, kwargs) elif new == 'off': self.callback_light_off(entity, attribute, old, new, kwargs) else: self.log(f'Unknown state: {new}') def callback_light_on(self, entity=None, attribute=None, old=None, new=None, kwargs=None): """Called when the light turns on """ self.log('Light on callback') self.cancel_motion('motion_on_handle') self.listen_motion_off(self.off_duration) # if self.is_stateful: # self.activate() def callback_light_off(self, entity=None, attribute=None, old=None, new=None, kwargs=None): """Called when the light turns on """ self.log('Light off callback') self.cancel_motion('motion_off_handle') self.listen_motion_on() def activate(self, *args, **kwargs): self.log('Activating') scene = self.current_scene() if isinstance(scene, str): self.turn_on(scene) self.log(f'Turned on scene: {scene}') elif isinstance(scene, dict): for entity, settings in scene.items(): if 'state' not in settings: scene[entity]['state'] = 'on' daylights = [ (entity, settings['state']) for entity, settings in scene.items() if isinstance(settings['state'], str) and settings['state'].startswith('daylight') ] for entity, app_name in daylights: # risky: scene.pop(entity) app: Continuous = self.get_app(app_name) self.log(f'Adjusting with app...') app.adjust() self.call_service('scene/apply', entities=scene, transition=0) self.log(f'Applied scene: {scene}') elif scene is None: self.log(f'No scene, ignoring...') # Need to act as if the light had just turned off to reset the motion (and maybe other things?) self.callback_light_off() else: self.log(f'ERROR: unknown scene: {scene}') def activate_all_off(self, *args, **kwargs): if self.all_off: self.activate() else: self.log(f'Skipped activating - everything is not off') def activate_any_on(self, kwargs): if self.any_on(): self.activate() else: self.log(f'Skipped activating - everything is not off') def deactivate(self, *args, **kwargs): self.log('Deactivating') for entity in self.app_entities: self.turn_off(entity) self.log(f'Turned off {entity}') def schedule_transitions(self, *args, **kwargs): # times, scenes, offs = zip(*self.settings()) for dt, scene, off_duration in self.settings(): dt = str(dt)[:8] self.log(f'Scheduling transition at: {dt}') try: self.run_at(callback=self.activate_any_on, start=dt) except ValueError: # happens when the callback time is in the past pass except Exception as e: self.log(f'Failed with {type(e)}: {e}') def handle_button(self, event_name, data, kwargs): # event 1002 is a single button press if data['event'] == 1002: self.log(f'{data["id"]} single click') if self.entity_state: self.deactivate() else: # use this instead of self.activate() to avoid it being called again by the state change event # ^ no longer relevant with self.activate() commented out in self.callback_light_on() # self.entity_state = True self.activate() # event 1001 is a long press start elif data['event'] == 1001: self.log(f'{data["id"]} long press down') if 'delay' in self.args and self.entity_state: self.cancel_motion('motion_off_handle') self.listen_motion_off(self.delay) self.turn_on(self.entity, brightness_pct=100) # event 1004 is a double click elif data['event'] == 1004: self.log(f'{data["id"]} double click') if 'sleep' in self.args: self.sleep_bool = not self.sleep_bool self.cancel_motion('motion_off_handle') self.callback_light_on() self.activate()