From 9ce8432bbaa48a88fe9fee20843cda8162cdf0d2 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jul 2024 12:17:34 -0500 Subject: [PATCH] added some services --- src/room_control/room_control.py | 148 +++++++++++++++++++------------ 1 file changed, 93 insertions(+), 55 deletions(-) diff --git a/src/room_control/room_control.py b/src/room_control/room_control.py index 376baf2..c9ef22b 100755 --- a/src/room_control/room_control.py +++ b/src/room_control/room_control.py @@ -5,7 +5,7 @@ import logging.config import traceback from copy import deepcopy from functools import wraps -from typing import Any, Dict, List +from typing import Any, Dict, List, Set from appdaemon.entity import Entity from appdaemon.plugins.hass.hassapi import Hass @@ -25,6 +25,14 @@ class RoomController(Hass): - `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 + """ @property @@ -43,38 +51,50 @@ class RoomController(Hass): @property def state_entity(self) -> Entity: - return self.get_entity(f'{self.name}.state') + return self.get_entity(f'{self.name}.state', namespace='controller') def initialize(self): - self.set_namespace('controller') - self.logger = console.load_rich_config(self.name) self.set_log_level('DEBUG') - self.register_service(f'{self.name}/activate', self.service_activate) - self.register_service(f'{self.name}/deactivate', self.service_deactivate) - self.refresh_state_times() self.run_daily(callback=self.refresh_state_times, start='00:00:00') - self.app_entities = set(self.gather_app_entities()) + 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}/deactivate', self._service_deactivate, 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') self.log(f'Initialized [bold green]{type(self).__name__}[/]') + self.activate_all_off(test_kwarg='abc123') def terminate(self): self.log('[bold red]Terminating[/]', level='DEBUG') - def gather_app_entities(self): - """Yields 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""" - 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, namespace='default', attribute='entity_id') - yield from entities - else: - yield from state.scene.keys() + 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 from state.scene.keys() + + return set(gen()) def refresh_state_times(self, *args, **kwargs): """Resets the `self.states` attribute to a newly parsed version of the states. @@ -123,25 +143,50 @@ class RoomController(Hass): else: return ControllerStateConfig() else: - attrs = self.state_entity.get_state('all')['attributes'] - return ControllerStateConfig.model_validate(attrs) + 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 - def current_scene(self, transition: int = None) -> Dict[str, Any]: + # def current_scene(self, transition: int = None) -> Dict[str, Any]: + # state = self.current_state() + # if isinstance(state.scene, str): + # return state.scene + # elif isinstance(state.scene, dict): + # return state.to_apply_kwargs(transition) + + 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): - return state.scene + self.turn_on(state.scene) + # self.turn_on(state.scene, transition=0) elif isinstance(state.scene, dict): - return state.to_apply_kwargs(transition) + scene = state.to_apply_kwargs() + self.call_service('scene/apply', **scene) + # scene = state.to_apply_kwargs(transition=0) - def service_activate(self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any]): - scene = self.current_scene(transition=0) + 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) - if isinstance(scene, str): - self.turn_on(scene) - elif isinstance(scene, dict): - self.call_service('scene/apply', namespace='default', **scene) + def _service_activate_all_off(self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any]): + if self.all_off(): + self.activate(**kwargs) - def service_deactivate(self, namespace: str, domain: str, service: str, kwargs: Dict[str, Any]): + 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) @@ -207,36 +252,29 @@ class RoomController(Hass): 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' + # 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) + # 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}') + # 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 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') + # 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_any_on(self, *args, **kwargs): """Activate if any of the entities are on. Args and kwargs are passed directly to self.activate()"""