button and door converted

src/room_control/button.py

@@ -1,53 +1,28 @@
 import json
 from dataclasses import dataclass
-from logging import Logger
+from typing import Any, Dict
-from typing import TYPE_CHECKING, List
 
-from appdaemon.plugins.mqtt.mqttapi import Mqtt
+from appdaemon.adapi import ADAPI
-
-from . import console
-from .model import ButtonConfig
-
-if TYPE_CHECKING:
-    from room_control import RoomController
 
 
-@dataclass(init=False)
+@dataclass
-class Button(Mqtt):
+class Button:
-    button: str | List[str]
+    adapi: ADAPI
-    rich: bool = False
+    button_name: str
-    config: ButtonConfig
-    logger: Logger
 
-    async def initialize(self):
+    def __post_init__(self):
-        self.app: 'RoomController' = await self.get_app(self.args['app'])
+        self.log = self.adapi.log
-        self.logger = console.load_rich_config(self.app.name, type(self).__name__)
+        topic = f'zigbee2mqtt/{self.button_name}'
-        self.config = ButtonConfig(**self.args)
+        self.adapi.listen_event(
-        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(
             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'])
+                    self.adapi.call_service(
-            kwargs = {'kwargs': {'cause': f'button single click: toggle while {state}'}}
+                        f'{self.adapi.name}/toggle', namespace='controller', cause='button'
-
+                    )
-            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

src/room_control/door.py

@@ -1,26 +1,25 @@
-from logging import Logger
+from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from logging import Logger, LoggerAdapter
 
-from appdaemon.plugins.hass.hassapi import Hass
+from appdaemon.adapi import ADAPI
-
-from . import console
-
-if TYPE_CHECKING:
-    from room_control import RoomController
 
 
-class Door(Hass):
+@dataclass
-    app: 'RoomController'
+class Door:
-    logger: Logger
+    adapi: ADAPI
+    entity_id: str
 
-    async def initialize(self):
+    def __post_init__(self):
-        self.app: 'RoomController' = await self.get_app(self.args['app'])
+        self.logger = LoggerAdapter(
-        self.logger = console.load_rich_config(room=self.app.name, component=type(self).__name__)
+            self.adapi.logger.logger.getChild('door'), self.adapi.logger.extra
-        self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
+        )
 
-        await self.listen_state(
+        self.adapi.listen_state(
-            self.app.activate_all_off,
+            callback=lambda *args, **kwargs: self.call_service(
-            entity_id=self.args['door'],
+                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}[/]')

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, root_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,23 +45,23 @@ 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:
+        # if time is not None and elevation is not None:
-            raise PydanticCustomError('bad_time_spec', 'Only one of time or elevation can be set.')
+        #     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 elevation is not None and 'direction' not in values:
             raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation')
         return values
 
-    def to_apply_kwargs(self, **kwargs):
+    def to_apply_kwargs(self, transition: int = None):
-        return ApplyKwargs(entities=self.scene, **kwargs).model_dump(exclude_none=True)
+        return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True)
 
 
 class RoomControllerConfig(BaseModel):
@@ -102,11 +102,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 +114,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:

src/room_control/room_control.py

@@ -1,21 +1,23 @@
 import datetime
-import json
 import logging
 import logging.config
-from copy import deepcopy
+import traceback
-from typing import Dict, List
+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
 
 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 +25,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
+
+    - <name>/activate
+    - <name>/activate_all_off
+    - <name>/deactivate
+    - <name>/toggle
+
     """
 
     @property
@@ -34,39 +45,66 @@ 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.set_log_level('DEBUG')
-        # self.log(f'entities: {self.app_entities}')
+
         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}/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__}[/]')
 
     def terminate(self):
         self.log('[bold red]Terminating[/]', level='DEBUG')
 
-    def gather_app_entities(self) -> List[str]:
+    def get_app_entities(self) -> Set[str]:
-        """Returns a list of all the entities involved in any of the states"""
+        """Gets a set of all the entities referenced by any of the state definitions"""
 
-        def generator():
+        def gen():
-            for settings in deepcopy(self.args['states']):
+            for state in self._room_config.states:
-                if scene := settings.get('scene'):
+                if isinstance(state.scene, str):
-                    if isinstance(scene, str):
+                    assert state.scene.startswith(
-                        assert scene.startswith(
                         'scene.'
-                        ), f"Scene definition must start with 'scene.' for app {self.name}"
+                    ), "Scene definition must start with 'scene.'"
-                        entity: Entity = self.get_entity(scene)
+                    entities = self.get_state(state.scene, attribute='entity_id')
-                        entity_state = entity.get_state('all')
+                    yield from entities
-                        attributes = entity_state['attributes']
-                        for entity in attributes['entity_id']:
-                            yield entity
                 else:
-                        for key in scene.keys():
+                    yield from state.scene.keys()
-                            yield key
-                else:
-                    yield self.args['entity']
 
-        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 +120,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,39 +128,92 @@ 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,
+                    callback=lambda cb_args: self.set_controller_scene(cb_args['state']),
-                    start=t.strftime('%H:%M:%S'),
+                    start=state.time.strftime('%H:%M:%S'),
-                    cause='scheduled transition',
+                    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:
+    def set_controller_scene(self, state: ControllerStateConfig):
+        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)
+            try:
-            self.log(f'Getting state for {now.strftime("%I:%M:%S %p")}', level='DEBUG')
+                attrs = self.state_entity.get_state('all')['attributes']
-
+                state = ControllerStateConfig.model_validate(attrs)
-            state = self._room_config.current_state(now)
+            except Exception:
-            self.log(f'Current state: {state.time}', level='DEBUG')
+                state = ControllerStateConfig()
+            finally:
+                # self.log(f'Current state: {state.model_dump(exclude_none=True)}', 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}
         return states
@@ -157,16 +248,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
 
@@ -184,54 +265,3 @@ class RoomController(Hass, Mqtt):
         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}')