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

button and door converted

Browse Source
This commit is contained in:
John Lancaster
2024-07-27 14:35:07 -05:00
parent 9ce8432bba
commit 1c5edf6cc7
3 changed files with 70 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
src/room_control/button.py
View File

@@ -1,53 +1,28 @@
import json
from dataclasses import dataclass
from logging import Logger
from typing import TYPE_CHECKING, List
from typing import Any, Dict
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from . import console
from .model import ButtonConfig
if TYPE_CHECKING:
from room_control import RoomController
from appdaemon.adapi import ADAPI
@dataclass(init=False)
class Button(Mqtt):
button: str | List[str]
rich: bool = False
config: ButtonConfig
logger: Logger
@dataclass
class Button:
adapi: ADAPI
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.log = self.adapi.log
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.log(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']
@@ -58,19 +33,7 @@ class Button(Mqtt):
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.adapi.call_service(
f'{self.adapi.name}/toggle', namespace='controller', cause='button'
)

35
src/room_control/door.py
View File

@@ -1,26 +1,25 @@
from logging import Logger
from typing import TYPE_CHECKING
from dataclasses import dataclass
from logging import Logger, LoggerAdapter
from appdaemon.plugins.hass.hassapi import Hass
from . import console
if TYPE_CHECKING:
from room_control import RoomController
from appdaemon.adapi import ADAPI
class Door(Hass):
app: 'RoomController'
logger: Logger
@dataclass
class Door:
adapi: ADAPI
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 = LoggerAdapter(
self.adapi.logger.logger.getChild('door'), self.adapi.logger.extra
)
await self.listen_state(
self.app.activate_all_off,
entity_id=self.args['door'],
self.adapi.listen_state(
callback=lambda *args, **kwargs: self.call_service(
f'{self.adapi.name}/activate_all_off', namespace='controller'
),
entity_id=self.entity_id,
new='on',
cause='door open',
)
self.logger.debug(f'Initialized door for [room]{self.adapi.name}[/]')

102
src/room_control/room_control.py
View File

@@ -1,9 +1,7 @@
import datetime
import json
import logging
import logging.config
import traceback
from copy import deepcopy
from functools import wraps
from typing import Any, Dict, List, Set
@@ -12,6 +10,8 @@ from appdaemon.plugins.hass.hassapi import Hass
from astral.location import Location
from . import console
from .button import Button
from .door import Door
from .model import ControllerStateConfig, RoomControllerConfig
logger = logging.getLogger(__name__)
@@ -32,6 +32,7 @@ class RoomController(Hass):
- <name>/activate
- <name>/activate_all_off
- <name>/deactivate
- <name>/toggle
"""
@@ -69,13 +70,22 @@ class RoomController(Hass):
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)
if door := self.args.get('door'):
if isinstance(door, str):
self.log('door--')
self.door = Door(self, entity_id=door)
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')
@@ -152,13 +162,6 @@ class RoomController(Hass):
# 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]:
# 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)
@@ -173,12 +176,24 @@ class RoomController(Hass):
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():
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):
@@ -190,6 +205,15 @@ class RoomController(Hass):
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}
return states
@@ -224,16 +248,6 @@ class RoomController(Hass):
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
@@ -251,47 +265,3 @@ class RoomController(Hass):
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_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}')