stagedlight

apps/apps.yaml

@@ -4,8 +4,8 @@
 
 
 bar_lights:
-  module: motion
+  module: light
-  class: StagedMotionLight
+  class: StagedLight
   stages:
     - start: '06:00 am'
       scene:

apps/light.py

@@ -0,0 +1,81 @@
+from collections.abc import Generator
+from datetime import datetime
+from enum import Enum
+from itertools import count
+from itertools import cycle
+from itertools import pairwise
+from typing import Any
+
+from appdaemon.plugins.hass import Hass
+from pydantic import TypeAdapter
+from stages import Stage
+
+
+class StagedControlEvent(str, Enum):
+    ACTIVATE = 'activate'
+    DEACTIVATE = 'deactivate'
+
+
+class StagedLight(Hass):
+    def initialize(self):
+        self.set_log_level('DEBUG')
+        self._type_adapter = TypeAdapter(list[Stage])
+        self._stages = self._type_adapter.validate_python(self.args['stages'])
+
+        self.log(f'Initialized Motion Sensor with {len(self._stages)} stages')
+        self.listen_event(self.handle_event, 'stage_control')
+        self.activate()
+
+    ### Stages
+
+    def _stage_starts(self) -> Generator[datetime]:
+        for offset in count(start=-1):
+            for stage in self._stages:
+                dt = self.parse_datetime(stage.start, days_offset=offset, aware=True, today=True)
+                dt = dt.replace(microsecond=0)
+                yield dt
+
+    def start_pairs(self):
+        """Yield from an infinite progression of start and end times for the stages."""
+        yield from pairwise(self._stage_starts())
+
+    def current_stage(self) -> Stage:
+        for stage, (t1, t2) in zip(cycle(self._stages), self.start_pairs()):
+            start, end = sorted([t1, t2])
+            if self.now_is_between(start, end):
+                self.log(f'Current stage start time: {stage.start}', level='DEBUG')
+                stage.assign_start(start)
+                return stage
+        else:
+            raise ValueError
+
+    def current_scene(self):
+        return self.current_stage().scene_json()
+
+    ### Events
+
+    def handle_event(self, event_type: str, data: dict[str, Any], **kwargs: Any) -> None:
+        self.log(f'Event handler: {event_type}', level='DEBUG')
+        stage = self.current_stage()
+        start_time_str = stage.formatted_start('%I:%M %p')
+        match data:
+            case {'action': StagedControlEvent.ACTIVATE}:
+                self.log(f'Activating current stage: {start_time_str}')
+                self.activate(scene=stage.scene_json())
+            case {'action': StagedControlEvent.DEACTIVATE}:
+                self.log(f'Deactivating current stage: {start_time_str}')
+                self.deactivate()
+            case _:
+                self.log(str(data), level='DEBUG')
+                self.log(str(kwargs), level='DEBUG')
+
+    ### Actions
+
+    def activate(self, scene: dict | None = None, **kwargs):
+        scene = scene if scene is not None else self.current_scene()
+        return self.call_service('scene/apply', entities=scene)
+
+    def deactivate(self, stage: Stage | None = None, **kwargs):
+        stage = stage if stage is not None else self.current_stage()
+        for entity in stage.scene:
+            self.turn_off(entity)

apps/motion.py

@@ -1,55 +0,0 @@
-import json
-from collections.abc import Generator
-from datetime import datetime
-from itertools import count, cycle, pairwise
-from pathlib import Path
-
-from appdaemon.adapi import ADAPI
-from pydantic import BaseModel, TypeAdapter
-
-from stages import Stage
-
-adapter = TypeAdapter(list[Stage])
-
-
-class StagedMotionLight(ADAPI):
-    def initialize(self):
-        self.set_log_level("DEBUG")
-        self._stages = adapter.validate_python(self.args["stages"])
-
-        self.log(f"Initialized Motion Sensor with {len(self._stages)} stages")
-        self.activate()
-
-    def _stage_starts(self) -> Generator[datetime]:
-        for offset in count(start=-1):
-            for stage in self._stages:
-                dt = self.parse_datetime(
-                    stage.start,
-                    days_offset=offset,
-                    aware=True,
-                    today=True,
-                )
-                dt = dt.replace(microsecond=0)
-                yield dt
-
-    def start_pairs(self):
-        yield from pairwise(self._stage_starts())
-
-    def current_stage(self) -> Stage:
-        for stage, (t1, t2) in zip(cycle(self._stages), self.start_pairs()):
-            start, end = sorted([t1, t2])
-            if self.now_is_between(start, end):
-                self.log(f"Current stage start time: {stage.start}")
-                return stage
-        else:
-            raise ValueError
-
-    def current_scene(self):
-        return self.current_stage().model_dump(mode="json")["scene"]
-
-    def activate(self, **kwargs):
-        return self.call_service("scene/apply", entities=self.current_scene())
-
-    # @property
-    # def stages(self) -> GeneratorExit:
-    #     yield from pairwise(self._stage_starts)

apps/stages.py

@@ -1,19 +1,9 @@
-import functools
+from datetime import datetime
-from datetime import datetime, time
+from typing import Any
-from pathlib import Path
-from typing import Annotated, Any
 
-from appdaemon.adapi import ADAPI
+from pydantic import BaseModel
-from pydantic import (
+from pydantic import PrivateAttr
-    BaseModel,
+from pydantic import field_serializer
-    BeforeValidator,
-    Field,
-    PrivateAttr,
-    TypeAdapter,
-    field_serializer,
-    field_validator,
-)
-from rich import print as rprint
 
 
 class EntityState(BaseModel):
@@ -21,16 +11,25 @@ class EntityState(BaseModel):
     color_temp_kelvin: int
     brightness: int
 
-    @field_serializer("state")
+    @field_serializer('state')
     def convert_state(self, val: Any):
         if val:
-            return "on"
+            return 'on'
         else:
-            return "off"
+            return 'off'
 
 
 class Stage(BaseModel):
     # start: Annotated[time, BeforeValidator(lambda v: parser(v).time())]
     start: str
-    _start: time = PrivateAttr()
+    _start: datetime = PrivateAttr()
     scene: dict[str, EntityState]
+
+    def assign_start(self, dt: datetime):
+        self._start = dt
+
+    def formatted_start(self, fmt: str) -> str:
+        return self._start.strftime(fmt)
+
+    def scene_json(self):
+        return self.model_dump(mode='json')['scene']