From 7cde3c75d8e0aa4bc452ece32b61e46274cb1503 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jul 2024 17:37:33 -0500 Subject: [PATCH] changed to use services --- src/room_control/__init__.py | 6 +- src/room_control/button.py | 72 +++----- src/room_control/door.py | 27 ++- src/room_control/model.py | 37 ++-- src/room_control/motion.py | 170 ++++++------------- src/room_control/room_control.py | 283 +++++++++++++++++++------------ 6 files changed, 274 insertions(+), 321 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/button.py b/src/room_control/button.py index df8c5ec..f2be4bc 100644 --- a/src/room_control/button.py +++ b/src/room_control/button.py @@ -1,76 +1,42 @@ import json from dataclasses import dataclass -from logging import Logger -from typing import TYPE_CHECKING, List - -from appdaemon.plugins.mqtt.mqttapi import Mqtt +from typing import TYPE_CHECKING, Any, Dict from . import console -from .model import ButtonConfig if TYPE_CHECKING: - from room_control import RoomController + from .room_control import RoomController -@dataclass(init=False) -class Button(Mqtt): - button: str | List[str] - rich: bool = False - config: ButtonConfig - logger: Logger +@dataclass +class Button: + adapi: 'RoomController' + button_name: str - async def initialize(self): - self.app: 'RoomController' = await self.get_app(self.args['app']) - self.logger = console.load_rich_config(self.app.name, type(self).__name__) - self.config = ButtonConfig(**self.args) - self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG') - - self.button = self.config.button - self.setup_buttons(self.button) - - def setup_buttons(self, buttons): - if isinstance(buttons, list): - for button in buttons: - self.setup_button(button) - else: - self.setup_button(buttons) - - def setup_button(self, name: str): - topic = f'zigbee2mqtt/{name}' - # self.mqtt_subscribe(topic, namespace='mqtt') - self.listen_event( + def __post_init__(self): + self.logger = console.load_rich_config(self.adapi.name, 'Button', 'DEBUG') + topic = f'zigbee2mqtt/{self.button_name}' + self.adapi.listen_event( self.handle_button, 'MQTT_MESSAGE', topic=topic, namespace='mqtt', - button=name, + button=self.button_name, ) - self.log(f'MQTT topic [topic]{topic}[/] controls app [room]{self.app.name}[/]') + self.logger.info(f'MQTT topic [topic]{topic}[/] controls [room]{self.adapi.name}[/]') - def handle_button(self, event_name, data, kwargs): + 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.handle_action(action) - - def handle_action(self, action: str): - if action == 'single': - state = self.get_state(self.args['ref_entity']) - kwargs = {'kwargs': {'cause': f'button single click: toggle while {state}'}} - - if manual_entity := self.args.get('manual_mode'): - self.set_state(entity_id=manual_entity, state='off') - - if state == 'on': - self.app.deactivate(**kwargs) - else: - self.app.activate(**kwargs) - else: - pass + 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 6b15f81..b2ed97c 100644 --- a/src/room_control/door.py +++ b/src/room_control/door.py @@ -1,26 +1,23 @@ -from logging import Logger +from dataclasses import dataclass from typing import TYPE_CHECKING -from appdaemon.plugins.hass.hassapi import Hass - from . import console if TYPE_CHECKING: - from room_control import RoomController + from .room_control import RoomController -class Door(Hass): - app: 'RoomController' - logger: Logger +@dataclass +class Door: + adapi: 'RoomController' + entity_id: str - async def initialize(self): - self.app: 'RoomController' = await self.get_app(self.args['app']) - self.logger = console.load_rich_config(room=self.app.name, component=type(self).__name__) - self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG') + def __post_init__(self): + self.logger = console.load_rich_config(self.adapi.name, 'Door', 'DEBUG') - await self.listen_state( - self.app.activate_all_off, - entity_id=self.args['door'], + self.adapi.listen_state( + lambda *args, **kwargs: self.adapi.activate_all_off(cause='door open'), + entity_id=self.entity_id, new='on', - cause='door open', ) + self.logger.debug(f'Initialized door for [room]{self.adapi.name}[/]') diff --git a/src/room_control/model.py b/src/room_control/model.py index ef3b1fd..0e74a1e 100644 --- a/src/room_control/model.py +++ b/src/room_control/model.py @@ -1,21 +1,21 @@ -from datetime import datetime, time, timedelta +import datetime from pathlib import Path from typing import Annotated, Dict, List, Optional, Self import yaml from astral import SunDirection -from pydantic import BaseModel, BeforeValidator, Field, root_validator +from pydantic import BaseModel, BeforeValidator, Field, model_validator from pydantic_core import PydanticCustomError from rich.console import Console, ConsoleOptions, RenderResult from rich.table import Column, Table -def str_to_timedelta(input_str: str) -> timedelta: +def str_to_timedelta(input_str: str) -> datetime.timedelta: try: hours, minutes, seconds = map(int, input_str.split(':')) - return timedelta(hours=hours, minutes=minutes, seconds=seconds) + return datetime.timedelta(hours=hours, minutes=minutes, seconds=seconds) except Exception: - return timedelta() + return datetime.timedelta() def str_to_direction(input_str: str) -> SunDirection: @@ -27,7 +27,7 @@ def str_to_direction(input_str: str) -> SunDirection: ) -OffDuration = Annotated[timedelta, BeforeValidator(str_to_timedelta)] +OffDuration = Annotated[datetime.timedelta, BeforeValidator(str_to_timedelta)] class State(BaseModel): @@ -45,28 +45,25 @@ class ApplyKwargs(BaseModel): class ControllerStateConfig(BaseModel): - time: Optional[str | datetime] = None + time: Optional[str | datetime.time | datetime.datetime] = None elevation: Optional[float] = None direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = None off_duration: Optional[OffDuration] = None - scene: dict[str, State] | str + scene: dict[str, State] | str = Field(default_factory=dict) - @root_validator(pre=True) + @model_validator(mode='before') def check_args(cls, values): - time, elevation = values.get('time'), values.get('elevation') - if time is not None and elevation is not None: - raise PydanticCustomError('bad_time_spec', 'Only one of time or elevation can be set.') - elif elevation is not None and 'direction' not in values: + if values.get('elevation') is not None and values.get('direction') is None: raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation') return values - def to_apply_kwargs(self, **kwargs): - return ApplyKwargs(entities=self.scene, **kwargs).model_dump(exclude_none=True) + def to_apply_kwargs(self, transition: int = None): + return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True) 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 @@ -102,11 +99,11 @@ class RoomControllerConfig(BaseModel): def sort_states(self): """Should only be called after all the times have been resolved""" assert all( - isinstance(state.time, time) for state in self.states + isinstance(state.time, datetime.time) for state in self.states ), 'Times have not all been resolved yet' self.states = sorted(self.states, key=lambda s: s.time, reverse=True) - def current_state(self, now: time) -> ControllerStateConfig: + def current_state(self, now: datetime.time) -> ControllerStateConfig: self.sort_states() for state in self.states: if state.time <= now: @@ -114,11 +111,11 @@ class RoomControllerConfig(BaseModel): else: return self.states[0] - def current_scene(self, now: time) -> Dict: + def current_scene(self, now: datetime.time) -> Dict: state = self.current_state(now) return state.scene - def current_off_duration(self, now: time) -> timedelta: + def current_off_duration(self, now: datetime.time) -> datetime.timedelta: state = self.current_state(now) if state.off_duration is None: if self.off_duration is None: diff --git a/src/room_control/motion.py b/src/room_control/motion.py index 19bdedf..7185e4a 100644 --- a/src/room_control/motion.py +++ b/src/room_control/motion.py @@ -1,11 +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 @@ -27,151 +25,89 @@ class CallbackEntry(BaseModel): Callbacks = dict[str, dict[str, CallbackEntry]] -class Motion(Hass): - logger: Logger - app: 'RoomController' +@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(self) -> Entity: - return self.get_entity(self.args['sensor']) + def sensor_entity(self) -> Entity: + return self.adapi.get_entity(self.sensor_entity_id) @property def sensor_state(self) -> bool: - return self.sensor.state == 'on' + return self.sensor_entity.get_state() == 'on' @property def ref_entity(self) -> Entity: - return self.get_entity(self.args['ref_entity']) + return self.adapi.get_entity(self.ref_entity_id) @property - def ref_entity_state(self) -> bool: + def ref_state(self) -> bool: return self.ref_entity.get_state() == 'on' @property def state_mismatch(self) -> bool: - return self.sensor_state != self.ref_entity_state + return self.sensor_state != self.ref_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__) + 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.listen_motion_off(duration) - 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 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.listen_state( - callback=self.app.activate_all_off, - entity_id=self.sensor.entity_id, + # 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, - cause='motion on', ) - self.log(f'Waiting for sensor motion on [friendly_name]{self.sensor.friendly_name}[/]') + self.logger.info( + f'Waiting for sensor motion on [friendly_name]{self.sensor_entity.friendly_name}[/]' + ) if self.sensor_state: - self.log( - f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is already on', - level='WARNING', + 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.listen_state( - callback=self.app.deactivate, - entity_id=self.sensor.entity_id, + # 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, - cause='motion off', ) - self.log( - f'Waiting for sensor [friendly_name]{self.sensor.friendly_name}[/] to be clear for {duration}' + 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.log( - f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is currently off', - level='WARNING', + self.logger.warning( + f'Sensor [friendly_name]{self.sensor_entity.friendly_name}[/] is currently off', ) - - 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', - ) diff --git a/src/room_control/room_control.py b/src/room_control/room_control.py index df14104..326c605 100755 --- a/src/room_control/room_control.py +++ b/src/room_control/room_control.py @@ -1,21 +1,24 @@ import datetime -import json import logging import logging.config -from copy import deepcopy -from typing import Dict, List +import traceback +from functools import wraps +from typing import Any, Dict, List, Set from appdaemon.entity import Entity from appdaemon.plugins.hass.hassapi import Hass -from appdaemon.plugins.mqtt.mqttapi import Mqtt +from astral.location import Location from . import console +from .button import Button +from .door import Door from .model import ControllerStateConfig, RoomControllerConfig +from .motion import MotionSensor logger = logging.getLogger(__name__) -class RoomController(Hass, Mqtt): +class RoomController(Hass): """Class for linking room's lights with a motion sensor. - Separate the turning on and turning off functions. @@ -23,6 +26,15 @@ class RoomController(Hass, Mqtt): - `handle_on` - `handle_off` - When the light comes on, check if it's attributes match what they should, given the time. + + + ## Services + + - /activate + - /activate_all_off + - /deactivate + - /toggle + """ @property @@ -34,39 +46,86 @@ class RoomController(Hass, Mqtt): assert all(isinstance(s, ControllerStateConfig) for s in new), f'Invalid: {new}' self._room_config.states = new + @property + @wraps(Location.time_at_elevation) + def time_at_elevation(self): + return self.AD.sched.location.time_at_elevation + + @property + def state_entity(self) -> Entity: + return self.get_entity(f'{self.name}.state', namespace='controller') + def initialize(self): self.logger = console.load_rich_config(self.name) - self.app_entities = self.gather_app_entities() - # self.log(f'entities: {self.app_entities}') + self.set_log_level('DEBUG') + self.refresh_state_times() self.run_daily(callback=self.refresh_state_times, start='00:00:00') + + self.register_service( + f'{self.name}/activate', self._service_activate, namespace='controller' + ) + self.register_service( + f'{self.name}/activate_all_off', self._service_activate_all_off, namespace='controller' + ) + self.register_service( + f'{self.name}/activate_any_on', self._service_activate_any_on, namespace='controller' + ) + self.register_service( + f'{self.name}/deactivate', self._service_deactivate, namespace='controller' + ) + self.register_service(f'{self.name}/toggle', self._service_toggle, namespace='controller') + + # This needs to come after this first call of refresh_state_times + self.app_entities = self.get_app_entities() + self.log(f'entities: {self.app_entities}', level='DEBUG') + + if button := self.args.get('button'): + if isinstance(button, str): + self.button = Button(self, button_name=button) + elif isinstance(button, list) and all(isinstance(b, str) for b in button): + self.button = [Button(self, button_name=b) for b in button] + + if door := self.args.get('door'): + if isinstance(door, str): + 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() + else: + t = state.time + if t < self.get_now().time(): + self.log(f'Initial state: {state.time}', level='DEBUG') + self.set_controller_scene(state) + break + self.log(f'Initialized [bold green]{type(self).__name__}[/]') def terminate(self): self.log('[bold red]Terminating[/]', level='DEBUG') - def gather_app_entities(self) -> List[str]: - """Returns a list of all the entities involved in any of the states""" + def get_app_entities(self) -> Set[str]: + """Gets a set of all the entities referenced by any of the state definitions""" - def generator(): - for settings in deepcopy(self.args['states']): - if scene := settings.get('scene'): - if isinstance(scene, str): - assert scene.startswith( - 'scene.' - ), f"Scene definition must start with 'scene.' for app {self.name}" - entity: Entity = self.get_entity(scene) - entity_state = entity.get_state('all') - attributes = entity_state['attributes'] - for entity in attributes['entity_id']: - yield entity - else: - for key in scene.keys(): - yield key + def gen(): + for state in self._room_config.states: + if isinstance(state.scene, str): + assert state.scene.startswith( + 'scene.' + ), "Scene definition must start with 'scene.'" + entities = self.get_state(state.scene, attribute='entity_id') + yield from entities else: - yield self.args['entity'] + yield from state.scene.keys() - return set(list(generator())) + return set(gen()) def refresh_state_times(self, *args, **kwargs): """Resets the `self.states` attribute to a newly parsed version of the states. @@ -82,7 +141,7 @@ class RoomController(Hass, Mqtt): for state in self._room_config.states: if state.time is None and state.elevation is not None: - state.time = self.AD.sched.location.time_at_elevation( + state.time = self.time_at_elevation( elevation=state.elevation, direction=state.direction ).time() elif isinstance(state.time, str): @@ -90,38 +149,99 @@ class RoomController(Hass, Mqtt): assert isinstance(state.time, datetime.time), f'Invalid time: {state.time}' - self.states = sorted(self.states, key=lambda s: s.time, reverse=True) - - # schedule the transitions - for state in self.states[::-1]: - # t: datetime.time = state['time'] - t: datetime.time = state.time try: self.run_at( - callback=self.activate_any_on, - start=t.strftime('%H:%M:%S'), - cause='scheduled transition', + callback=lambda cb_args: self.set_controller_scene(cb_args['state']), + start=state.time.strftime('%H:%M:%S'), + state=state, ) - 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 current_state(self, now: datetime.time = None) -> ControllerStateConfig: + self.state_entity.listen_state( + lambda *args, **kwargs: self.call_service( + f'{self.name}/activate_any_on', namespace='controller', cause='state transition' + ), + attribute='all', + ) + + def set_controller_scene(self, state: ControllerStateConfig): + """Sets the internal state for the app. Only used by the scheduled transition""" + try: + self.state_entity.set_state(attributes=state.model_dump()) + except Exception: + self.logger.error(traceback.format_exc()) + else: + self.log(f'Set controller state of {self.name}: {state.model_dump()}', level='DEBUG') + + def current_state(self) -> ControllerStateConfig: if self.sleep_bool(): - self.log('sleep: active') + self.log('sleep: active', level='DEBUG') if state := self.args.get('sleep_state'): return ControllerStateConfig(**state) else: - return ControllerStateConfig(scene={}) + return ControllerStateConfig() else: - now = now or self.get_now().time().replace(microsecond=0) - self.log(f'Getting state for {now.strftime("%I:%M:%S %p")}', level='DEBUG') + try: + attrs = self.state_entity.get_state('all')['attributes'] + state = ControllerStateConfig.model_validate(attrs) + except Exception: + state = ControllerStateConfig() + finally: + # self.log(f'Current state: {state.model_dump(exclude_none=True)}', level='DEBUG') + return state - state = self._room_config.current_state(now) - self.log(f'Current state: {state.time}', level='DEBUG') - return state + def activate(self, **kwargs): + self.call_service(f'{self.name}/activate', namespace='controller', **kwargs) + + def _service_activate(self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any]): + # self.log(f'Custom kwargs: {kwargs}', level='DEBUG') + state = self.current_state() + if isinstance(state.scene, str): + self.turn_on(state.scene) + # self.turn_on(state.scene, transition=0) + elif isinstance(state.scene, dict): + scene = state.to_apply_kwargs() + self.call_service('scene/apply', **scene) + # scene = state.to_apply_kwargs(transition=0) + + def activate_any_on(self, **kwargs): + """Activate if any of the entities are on. Args and kwargs are passed directly to self.activate()""" + self.call_service(f'{self.name}/activate_any_on', namespace='controller', **kwargs) + + def _service_activate_any_on( + self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any] + ): + if self.any_on() and not self.manual_mode(): + self.activate(**kwargs) + + def activate_all_off(self, **kwargs): + """Activate if all of the entities are off. Args and kwargs are passed directly to self.activate()""" + self.call_service(f'{self.name}/activate_all_off', namespace='controller', **kwargs) + + def _service_activate_all_off( + self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any] + ): + if self.all_off() and not self.manual_mode(): + self.activate(**kwargs) + + def deactivate(self, **kwargs): + self.call_service(f'{self.name}/deactivate', namespace='controller', **kwargs) + + def _service_deactivate( + self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any] + ): + for e in self.app_entities: + self.turn_off(e) + + def toggle(self, **kwargs): + self.call_service(f'{self.name}/toggle', namespace='controller', **kwargs) + + def _service_toggle(self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any]): + if self.any_on(): + self.deactivate(**kwargs) + else: + self.activate(**kwargs) def app_entity_states(self) -> Dict[str, str]: states = {entity: self.get_state(entity) for entity in self.app_entities} @@ -157,16 +277,6 @@ class RoomController(Hass, Mqtt): else: return False - # @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 off_duration(self, now: datetime.time = None) -> datetime.timedelta: """Determines the time that the motion sensor has to be clear before deactivating @@ -177,61 +287,8 @@ class RoomController(Hass, Mqtt): - 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) - - def activate(self, entity=None, attribute=None, old=None, new=None, kwargs=None): - if kwargs is not None: - cause = kwargs.get('cause', 'unknown') - else: - cause = 'unknown' - - self.log(f'Activating: {cause}') - scene_kwargs = self.current_state().to_apply_kwargs(transition=0) - - if isinstance(scene_kwargs, str): - self.turn_on(scene_kwargs) - self.log(f'Turned on scene: {scene_kwargs}') - - elif isinstance(scene_kwargs, dict): - self.call_service('scene/apply', **scene_kwargs) - self.log(f'Applied scene:\n{json.dumps(scene_kwargs, indent=2)}', level='DEBUG') - - elif scene_kwargs is None: - self.log('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_kwargs}') - - def activate_all_off(self, *args, **kwargs): - """Activate if all of the entities are off. Args and kwargs are passed directly to self.activate()""" - if self.all_off(): - self.activate(*args, **kwargs) - else: - self.log('Skipped activating - everything is not off') - - def activate_any_on(self, *args, **kwargs): - """Activate if any of the entities are on. Args and kwargs are passed directly to self.activate()""" - if self.any_on() and not self.manual_mode(): - self.activate(*args, **kwargs) - else: - self.log('Skipped activating - everything is off') - - def toggle_activate(self, *args, **kwargs): - if self.any_on(): - self.deactivate(*args, **kwargs) - else: - self.activate(*args, **kwargs) - - def deactivate(self, entity=None, attribute=None, old=None, new=None, kwargs=None): - cause = kwargs.get('cause', 'unknown') - self.log(f'Deactivating: {cause}') - for e in self.app_entities: - self.turn_off(e) - self.log(f'Turned off {e}') + return self.current_state().off_duration or self._room_config.off_duration