From 21e143100111332b0cc3be8eaabb923ef28ad7f6 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jul 2024 17:36:33 -0500 Subject: [PATCH] big prune --- src/room_control/__init__.py | 6 +- src/room_control/motion.py | 156 +---------------------------------- 2 files changed, 4 insertions(+), 158 deletions(-) diff --git a/src/room_control/__init__.py b/src/room_control/__init__.py index 12ef16d..390c5fd 100644 --- a/src/room_control/__init__.py +++ b/src/room_control/__init__.py @@ -1,6 +1,6 @@ -from .room_control import RoomController -from .motion import Motion from .button import Button from .door import Door +from .motion import MotionSensor +from .room_control import RoomController -__all__ = ['RoomController', 'Motion', 'Button', 'Door'] +__all__ = ['RoomController', 'MotionSensor', 'Button', 'Door'] diff --git a/src/room_control/motion.py b/src/room_control/motion.py index 66f90ae..7185e4a 100644 --- a/src/room_control/motion.py +++ b/src/room_control/motion.py @@ -1,12 +1,9 @@ -import re from dataclasses import dataclass from datetime import timedelta -from logging import Logger from typing import TYPE_CHECKING, Literal, Optional from appdaemon.entity import Entity -from appdaemon.plugins.hass.hassapi import Hass -from pydantic import BaseModel, ValidationError +from pydantic import BaseModel from . import console @@ -72,7 +69,6 @@ class MotionSensor: if new['state'] == 'on': self.logger.debug(f'Detected {entity} turning on') duration = self.adapi.off_duration() - # self.logger.debug(f'Off duration: {duration}') self.listen_motion_off(duration) def callback_light_off(self, entity: str, attribute: str, old: str, new: str, kwargs: dict): @@ -115,153 +111,3 @@ class MotionSensor: self.logger.warning( f'Sensor [friendly_name]{self.sensor_entity.friendly_name}[/] is currently off', ) - - -class Motion(Hass): - logger: Logger - app: 'RoomController' - - @property - def sensor(self) -> Entity: - return self.get_entity(self.args['sensor']) - - @property - def sensor_state(self) -> bool: - return self.sensor.state == 'on' - - @property - def ref_entity(self) -> Entity: - return self.get_entity(self.args['ref_entity']) - - @property - def ref_entity_state(self) -> bool: - return self.ref_entity.get_state() == 'on' - - @property - def state_mismatch(self) -> bool: - return self.sensor_state != self.ref_entity_state - - def initialize(self): - self.app: 'RoomController' = self.get_app(self.args['app']) - self.logger = console.load_rich_config(self.app.name, type(self).__name__) - - assert self.entity_exists(self.args['sensor']) - assert self.entity_exists(self.args['ref_entity']) - - base_kwargs = dict( - entity_id=self.ref_entity.entity_id, - immediate=True, # avoids needing to sync the state - ) - - if self.state_mismatch: - self.log( - f'Sensor is {self.sensor_state} ' f'and light is {self.ref_entity_state}', - level='WARNING', - ) - if self.sensor_state: - self.app.activate(kwargs={'cause': f'Syncing state with {self.sensor.entity_id}'}) - - # don't need to await these because they'll already get turned into a task by the utils.sync_wrapper decorator - self.listen_state( - **base_kwargs, - attribute='brightness', - callback=self.callback_light_on, - ) - self.listen_state(**base_kwargs, new='off', callback=self.callback_light_off) - - for handle, cb in self.callbacks(): - self.log(f'Handle [yellow]{handle[:4]}[/]: {cb.function}', level='DEBUG') - - self.log(f'Initialized [bold green]{type(self).__name__}[/]') - - def callbacks(self): - """Returns a dictionary of validated CallbackEntry objects that are associated with this app""" - self_callbacks = self.get_callback_entries().get(self.name, {}) - for handle, cb_dict in self_callbacks.items(): - try: - yield handle, CallbackEntry.model_validate(cb_dict) - except ValidationError as e: - self.logger.error('Error parsing callback dictionary') - self.logger.error(e) - - def listen_motion_on(self): - """Sets up the motion on callback to activate the room""" - self.cancel_motion_callback() - self.listen_state( - callback=self.app.activate_all_off, - entity_id=self.sensor.entity_id, - new='on', - oneshot=True, - cause='motion on', - ) - self.log(f'Waiting for sensor motion on [friendly_name]{self.sensor.friendly_name}[/]') - if self.sensor_state: - self.log( - f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is already on', - level='WARNING', - ) - - def listen_motion_off(self, duration: timedelta): - """Sets up the motion off callback to deactivate the room""" - self.cancel_motion_callback() - self.listen_state( - callback=self.app.deactivate, - entity_id=self.sensor.entity_id, - new='off', - duration=duration.total_seconds(), - oneshot=True, - cause='motion off', - ) - self.log( - f'Waiting for sensor [friendly_name]{self.sensor.friendly_name}[/] to be clear for {duration}' - ) - - if not self.sensor_state: - self.log( - f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is currently off', - level='WARNING', - ) - - def callback_light_on(self, entity=None, attribute=None, old=None, new=None, kwargs=None): - """Called when the light turns on""" - if new is not None: - self.log(f'Detected {entity} turning on', level='DEBUG') - duration = self.app.off_duration() - self.listen_motion_off(duration) - - def callback_light_off(self, entity=None, attribute=None, old=None, new=None, kwargs=None): - """Called when the light turns off""" - self.log(f'Detected {entity} turning off', level='DEBUG') - self.listen_motion_on() - - def get_app_callbacks(self, name: str = None): - """Gets all the callbacks associated with the app""" - name = name or self.name - callbacks = { - handle: info - for app_name, callbacks in self.get_callback_entries().items() - for handle, info in callbacks.items() - if app_name == name - } - return callbacks - - def get_sensor_callbacks(self): - return { - handle: info - for handle, info in self.get_app_callbacks().items() - if info['entity'] == self.sensor.entity_id - } - - def cancel_motion_callback(self): - callbacks = self.get_sensor_callbacks() - # self.log(f'Found {len(callbacks)} callbacks for {self.sensor.entity_id}') - for handle, info in callbacks.items(): - entity = info['entity'] - kwargs = info['kwargs'] - if (m := re.match('new=(?P.*?)\s', kwargs)) is not None: - new = m.group('new') - self.cancel_listen_state(handle) - self.log( - f'cancelled callback for sensor {entity} turning {new}', - level='DEBUG', - )