john/room_control
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added some services

Browse Source
This commit is contained in:
John Lancaster
2024-07-27 12:17:34 -05:00
parent 7f68c8cad2
commit 9ce8432bba
1 changed files with 93 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
src/room_control/room_control.py
View File

@@ -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
- <name>/activate
- <name>/activate_all_off
- <name>/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()"""