From 2cd02627d193f507fdb5d464c4d5222f2740bad6 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jul 2024 16:32:30 -0500 Subject: [PATCH] started motion redo --- src/room_control/button.py | 17 +++--- src/room_control/door.py | 17 +++--- src/room_control/model.py | 2 +- src/room_control/motion.py | 90 ++++++++++++++++++++++++++++++++ src/room_control/room_control.py | 14 +++-- 5 files changed, 118 insertions(+), 22 deletions(-) diff --git a/src/room_control/button.py b/src/room_control/button.py index f80e94d..f2be4bc 100644 --- a/src/room_control/button.py +++ b/src/room_control/button.py @@ -1,17 +1,20 @@ import json from dataclasses import dataclass -from typing import Any, Dict +from typing import TYPE_CHECKING, Any, Dict -from appdaemon.adapi import ADAPI +from . import console + +if TYPE_CHECKING: + from .room_control import RoomController @dataclass class Button: - adapi: ADAPI + adapi: 'RoomController' button_name: str def __post_init__(self): - self.log = self.adapi.log + self.logger = console.load_rich_config(self.adapi.name, 'Button', 'DEBUG') topic = f'zigbee2mqtt/{self.button_name}' self.adapi.listen_event( self.handle_button, @@ -20,19 +23,19 @@ class Button: namespace='mqtt', button=self.button_name, ) - self.log(f'MQTT topic [topic]{topic}[/] controls [room]{self.adapi.name}[/]') + self.logger.info(f'MQTT topic [topic]{topic}[/] controls [room]{self.adapi.name}[/]') def handle_button(self, event_name: str, data: Dict[str, Any], kwargs: Dict[str, Any]): try: payload = json.loads(data['payload']) action = payload['action'] except json.JSONDecodeError: - self.log(f'Error decoding JSON from {data["payload"]}', level='ERROR') + self.logger.error(f'Error decoding JSON from {data["payload"]}') except KeyError: return else: if isinstance(action, str) and action != '': - self.log(f'Action: [yellow]{action}[/]') + self.logger.info(f'Action: [yellow]{action}[/]') if action == 'single': self.adapi.call_service( f'{self.adapi.name}/toggle', namespace='controller', cause='button' diff --git a/src/room_control/door.py b/src/room_control/door.py index ec66bb1..b2ed97c 100644 --- a/src/room_control/door.py +++ b/src/room_control/door.py @@ -1,23 +1,22 @@ from dataclasses import dataclass -from logging import LoggerAdapter +from typing import TYPE_CHECKING -from appdaemon.adapi import ADAPI +from . import console + +if TYPE_CHECKING: + from .room_control import RoomController @dataclass class Door: - adapi: ADAPI + adapi: 'RoomController' entity_id: str def __post_init__(self): - self.logger = LoggerAdapter( - self.adapi.logger.logger.getChild('door'), self.adapi.logger.extra - ) + self.logger = console.load_rich_config(self.adapi.name, 'Door', 'DEBUG') self.adapi.listen_state( - callback=lambda *args, **kwargs: self.adapi.call_service( - f'{self.adapi.name}/activate_all_off', namespace='controller', cause='door open' - ), + lambda *args, **kwargs: self.adapi.activate_all_off(cause='door open'), entity_id=self.entity_id, new='on', ) diff --git a/src/room_control/model.py b/src/room_control/model.py index e20357d..0e74a1e 100644 --- a/src/room_control/model.py +++ b/src/room_control/model.py @@ -63,7 +63,7 @@ class ControllerStateConfig(BaseModel): class RoomControllerConfig(BaseModel): states: List[ControllerStateConfig] = Field(default_factory=list) - off_duration: Optional[OffDuration] = None + off_duration: Optional[OffDuration] = Field(default_factory=datetime.timedelta) sleep_state: Optional[ControllerStateConfig] = None rich: Optional[str] = None manual_mode: Optional[str] = None diff --git a/src/room_control/motion.py b/src/room_control/motion.py index 19bdedf..66f90ae 100644 --- a/src/room_control/motion.py +++ b/src/room_control/motion.py @@ -1,4 +1,5 @@ import re +from dataclasses import dataclass from datetime import timedelta from logging import Logger from typing import TYPE_CHECKING, Literal, Optional @@ -27,6 +28,95 @@ class CallbackEntry(BaseModel): Callbacks = dict[str, dict[str, CallbackEntry]] +@dataclass +class MotionSensor: + adapi: 'RoomController' + sensor_entity_id: str + ref_entity_id: str + + def __post_init__(self): + self.logger = console.load_rich_config(self.adapi.name, 'Motion', 'DEBUG') + + assert self.sensor_entity.exists() + assert self.ref_entity.exists() + + base_kwargs = dict( + entity_id=self.ref_entity_id, + immediate=True, # avoids needing to sync the state + ) + self.ref_entity.listen_state(self.callback_light_on, attribute='all', **base_kwargs) + self.ref_entity.listen_state(self.callback_light_off, new='off', **base_kwargs) + + @property + def sensor_entity(self) -> Entity: + return self.adapi.get_entity(self.sensor_entity_id) + + @property + def sensor_state(self) -> bool: + return self.sensor_entity.get_state() == 'on' + + @property + def ref_entity(self) -> Entity: + return self.adapi.get_entity(self.ref_entity_id) + + @property + def ref_state(self) -> bool: + return self.ref_entity.get_state() == 'on' + + @property + def state_mismatch(self) -> bool: + return self.sensor_state != self.ref_state + + def callback_light_on(self, entity: str, attribute: str, old: str, new: str, kwargs: dict): + """Called when the light turns on""" + 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): + """Called when the light turns off""" + self.logger.debug(f'Detected {entity} turning off') + self.listen_motion_on() + + def listen_motion_on(self): + """Sets up the motion on callback to activate the room""" + # self.cancel_motion_callback() + self.adapi.listen_state( + lambda *args, **kwargs: self.adapi.activate_all_off(cause='motion on'), + entity_id=self.sensor_entity_id, + new='on', + oneshot=True, + ) + self.logger.info( + f'Waiting for sensor motion on [friendly_name]{self.sensor_entity.friendly_name}[/]' + ) + if self.sensor_state: + self.logger.warning( + f'Sensor [friendly_name]{self.sensor_entity.friendly_name}[/] is already on', + ) + + def listen_motion_off(self, duration: timedelta): + """Sets up the motion off callback to deactivate the room""" + # self.cancel_motion_callback() + self.adapi.listen_state( + lambda *args, **kwargs: self.adapi.deactivate(cause='motion off'), + entity_id=self.sensor_entity_id, + new='off', + duration=duration.total_seconds(), + oneshot=True, + ) + self.logger.debug( + f'Waiting for sensor [friendly_name]{self.sensor_entity.friendly_name}[/] to be clear for {duration}' + ) + + if not self.sensor_state: + self.logger.warning( + f'Sensor [friendly_name]{self.sensor_entity.friendly_name}[/] is currently off', + ) + + class Motion(Hass): logger: Logger app: 'RoomController' diff --git a/src/room_control/room_control.py b/src/room_control/room_control.py index 75b7103..326c605 100755 --- a/src/room_control/room_control.py +++ b/src/room_control/room_control.py @@ -13,6 +13,7 @@ from . import console from .button import Button from .door import Door from .model import ControllerStateConfig, RoomControllerConfig +from .motion import MotionSensor logger = logging.getLogger(__name__) @@ -90,6 +91,11 @@ class RoomController(Hass): self.log('door--') self.door = Door(self, entity_id=door) + if motion := self.args.get('motion'): + self.motion = MotionSensor( + self, sensor_entity_id=motion['sensor'], ref_entity_id=motion['ref_entity'] + ) + for state in sorted(self._room_config.states, key=lambda s: s.time, reverse=True): if isinstance(state.time, datetime.datetime): t = state.time.time() @@ -281,10 +287,8 @@ class RoomController(Hass): - Sleep - 0 """ - sleep_mode_active = self.sleep_bool() - if sleep_mode_active: - self.log(f'Sleeping mode active: {sleep_mode_active}') + if sleep_active := self.sleep_bool(): + self.log(f'Sleeping mode active: {sleep_active}', level='DEBUG') return datetime.timedelta() else: - now = now or self.get_now().time() - return self._room_config.current_off_duration(now) + return self.current_state().off_duration or self._room_config.off_duration