Config-based plan

hueplanner/dsl.py

@@ -0,0 +1,57 @@
+from typing import Any
+
+import yaml
+
+from .planner import (
+    ACTION_CLASSES,
+    TRIGGER_CLASSES,
+    PlanAction,
+    PlanActionSequence,
+    PlanEntry,
+    PlanTrigger,
+)
+
+
+def load_plan(path: str, encoding: str | None = None) -> list[PlanEntry]:
+    with open(path, "r", encoding=encoding) as f:
+        master_config = yaml.safe_load(f)
+        plan_entries = master_config.get("plan", [])
+        plan = [load_plan_entry(entry) for entry in plan_entries]
+        return plan
+
+
+def generate_mappings(classes, prefix):
+    return {cls.__name__.replace(prefix, ""): cls for cls in classes}
+
+
+action_map = generate_mappings(ACTION_CLASSES, "PlanAction")
+trigger_map = generate_mappings(TRIGGER_CLASSES, "PlanTrigger")
+
+
+def _load_action(action_data: dict[str, Any]) -> PlanAction:
+    type, args = action_data["type"], action_data["args"]
+    action_class = action_map[type]
+    if action_class is PlanActionSequence:
+        action = _load_sequence(args)
+    else:
+        action = action_class.loads(args)
+    return action
+
+
+def _load_trigger(trigger_data: dict[str, Any]) -> PlanTrigger:
+    type, args = trigger_data["type"], trigger_data["args"]
+    trigger_class = trigger_map[type]
+    return trigger_class.loads(args)
+
+
+def _load_sequence(entries: list[dict[str, Any]]) -> PlanActionSequence:
+    items: list[PlanAction] = []
+    for entry in entries:
+        items.append(_load_action(entry))
+    return PlanActionSequence(*items)
+
+
+def load_plan_entry(entry: dict[str, Any]) -> PlanEntry:
+    action = _load_action(entry["action"])
+    trigger = _load_trigger(entry["trigger"])
+    return PlanEntry(trigger=trigger, action=action)

hueplanner/hue/v1/client.py

@@ -86,18 +86,6 @@ async def get_groups(self) -> list[Group]:
             items.append(model)
         return items
 
-    # logic-related api
-    # async def switch_light(self, group_id: int | str, on: bool = True, scene: str | None = None) -> Group:
-    #     body: dict[str, Any] = {"on": on}
-    #     if scene is not None:
-    #         body["scene"] = scene
-    #     resp = await self.session.put(
-    #         self._api_url / f"groups/{group_id}/action",
-    #         json=body,
-    #     )
-    #     resp.raise_for_status()
-    #     return await resp.json()
-
     async def send_group_action(self, group_id: int | str, action: dict[str, Any]):
         resp = await self.session.put(
             self._api_url / f"groups/{group_id}/action",
@@ -106,10 +94,6 @@ async def send_group_action(self, group_id: int | str, action: dict[str, Any]):
         resp.raise_for_status()
         return await resp.json()
 
-    async def toggle_light(self, group_id: int | str) -> Group:
-        group = await self.get_group(group_id)
-        return await self.switch_light(group_id, not group.action.on)
-
     async def activate_scene(self, group_id: int | str, scene_id: str, transition_time: int | None = None):
         body: dict[str, Any] = {
             "scene": scene_id,

hueplanner/hue/v1/models/light.py

@@ -17,7 +17,7 @@ class LightState(BaseModel):
     reachable: bool
 
 
-class Swupdate(BaseModel):
+class SwUpdate(BaseModel):
     state: str
     lastinstall: str
 
@@ -56,7 +56,7 @@ class Config(BaseModel):
 class Light(BaseModel):
     id: int = None  # type: ignore
     state: LightState
-    swupdate: Swupdate
+    swupdate: SwUpdate
     type: str
     name: str
     modelid: str

hueplanner/hue/v2/client.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import json
-
 import aiohttp
 import structlog
 import yarl

hueplanner/planner/__init__.py

@@ -17,3 +17,21 @@
     PlanTriggerOnHueButtonEvent,
     PlanTriggerPeriodic,
 )
+
+ACTION_CLASSES = [
+    PlanActionStoreSceneById,
+    PlanActionStoreSceneByName,
+    PlanActionSequence,
+    PlanActionToggleStoredScene,
+    ## This cannot be parsed from config yet
+    # PlanActionCallback,
+    # PlanActionWithEvaluationCondition,
+    # PlanActionWithRuntimeCondition,
+]
+
+TRIGGER_CLASSES = [
+    PlanTriggerImmediately,
+    PlanTriggerOnce,
+    PlanTriggerOnHueButtonEvent,
+    PlanTriggerPeriodic,
+]

hueplanner/planner/actions.py

@@ -2,14 +2,16 @@
 
 import inspect
 from dataclasses import dataclass
-from typing import Awaitable, Callable, Protocol
 from datetime import datetime
-from zoneinfo import ZoneInfo
+from typing import Any, Awaitable, Callable, Protocol
 
 import structlog
+from pydantic import BaseModel
+from zoneinfo import ZoneInfo
 
 from ..hue.v1.models import Scene
 from .context import Context
+from .serializable import Serializable
 
 logger = structlog.getLogger(__name__)
 
@@ -43,6 +45,9 @@ def __init__(self, *actions) -> None:
         super().__init__()
         self._actions: tuple[PlanAction, ...] = tuple(item for a in actions for item in self._unpack_nested_sequence(a))
 
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(items=[" + ", ".join(repr(i) for i in self._actions) + "])"
+
     def _unpack_nested_sequence(self, action: PlanAction) -> tuple[PlanAction, ...]:
         if isinstance(action, PlanActionSequence):
             return action._actions
@@ -164,10 +169,15 @@ def match_scene(self, scene: Scene) -> bool: ...
 
 
 @dataclass
-class PlanActionStoreSceneByName(PlanActionStoreScene):
+class PlanActionStoreSceneByName(PlanActionStoreScene, Serializable):
     name: str
     group: int | None = None
 
+    class _Model(BaseModel):
+        store_as: str
+        name: str
+        group: int | None = None
+
     def match_scene(self, scene: Scene) -> bool:
         if scene.name == self.name:
             if not self.group:
@@ -178,9 +188,13 @@ def match_scene(self, scene: Scene) -> bool:
 
 
 @dataclass
-class PlanActionStoreSceneById(PlanActionStoreScene):  # type: ignore
+class PlanActionStoreSceneById(PlanActionStoreScene, Serializable):
     id: str
 
+    class _Model(BaseModel):
+        store_as: str
+        id: str
+
     def match_scene(self, scene: Scene) -> bool:
         return scene.id == self.id
 
@@ -189,9 +203,13 @@ def match_scene(self, scene: Scene) -> bool:
 
 
 @dataclass
-class PlanActionToggleStoredScene(PlanAction):
+class PlanActionToggleStoredScene(PlanAction, Serializable):
     stored_scene: str
-    fallback_run_job_tag: str | None = None
+    fallback_nearest_scheduler_job_tag: str | None = None
+
+    class _Model(BaseModel):
+        stored_scene: str
+        fallback_nearest_scheduler_job_tag: str | None = None
 
     @staticmethod
     async def run_nearest_scheduled_job(context: Context, tag: str):
@@ -220,11 +238,12 @@ async def define_action(self, context: Context) -> EvaluatedAction:
         async def toggle_current_scene():
             scene = await context.scenes.get(self.stored_scene)
             if not scene:
-                logger.debug(
-                    "No current scene stored, performing fallback procedure", stored_scene_id=self.stored_scene
-                )
-                if self.fallback_run_job_tag:
-                    await self.run_nearest_scheduled_job(context=context, tag=self.fallback_run_job_tag)
+                if self.fallback_nearest_scheduler_job_tag:
+                    logger.debug(
+                        "No current scene stored, performing fallback procedure",
+                        fallback_nearest_scheduler_job_tag=self.fallback_nearest_scheduler_job_tag,
+                    )
+                    await self.run_nearest_scheduled_job(context=context, tag=self.fallback_nearest_scheduler_job_tag)
                 scene = await context.scenes.get(self.stored_scene)
             if not scene:
                 logger.error("Can't toggle scene, because it was not set yet")
@@ -237,7 +256,7 @@ async def toggle_current_scene():
             group = await context.hue_client_v1.get_group(scene.group)
             logger.debug("Current group state", group_id=group.id, group_state=group.state)
 
-            # TODO: use models, not dict
+            # TODO: Better typing - use models, not dict
             if group.state.all_on:
                 action = {"on": False}
                 logger.info("Turning light off", group=scene.group)
@@ -251,5 +270,4 @@ async def toggle_current_scene():
                 action = {"on": True, "scene": scene.id}
             result = await context.hue_client_v1.send_group_action(scene.group, action)
             logger.debug("Scene toggled", result=result)
-
         return toggle_current_scene

hueplanner/planner/context.py

@@ -53,3 +53,9 @@ async def reset(self):
     def add_task_to_pool(self, task: asyncio.Task):
         self.task_pool.add(task)
         task.add_done_callback(self.task_pool.discard)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.shutdown()

hueplanner/planner/planner.py

@@ -9,7 +9,6 @@
 class PlanEntry:
     trigger: PlanTrigger
     action: PlanAction
-    tag: str | None = None
 
     async def apply(self, context: Context):
         action = await self.action.define_action(context)

hueplanner/planner/serializable.py

@@ -0,0 +1,15 @@
+from typing import Any, Protocol
+
+from pydantic import BaseModel
+
+
+class Serializable(Protocol):
+    @property
+    def _Model(self) -> BaseModel: ...
+
+    @classmethod
+    def loads(cls, data: dict[str, Any]):
+        return cls(**cls._Model.model_validate(data).model_dump())  # type: ignore
+
+    def dumps(self) -> dict[str, Any]:
+        return self._Model.model_validate(self, from_attributes=True).model_dump()

hueplanner/planner/triggers.py

@@ -1,34 +1,43 @@
+from __future__ import annotations
+
 import asyncio
 from contextlib import suppress
 from dataclasses import dataclass
-from datetime import datetime, time, timedelta
+from datetime import time, timedelta
 from typing import Protocol
 
 import structlog
+from pydantic import BaseModel
 
 from ..hue.v2.models import HueEvent
 from .actions import EvaluatedAction
 from .context import Context
+from .serializable import Serializable
 
 logger = structlog.getLogger(__name__)
 
 # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 
 
 class PlanTrigger(Protocol):
-    async def apply_trigger(self, context: Context, action: EvaluatedAction):
-        pass
+
+    async def apply_trigger(self, context: Context, action: EvaluatedAction): ...
 
 
 # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 
 
 @dataclass
-class PlanTriggerOnce(PlanTrigger):
+class PlanTriggerOnce(PlanTrigger, Serializable):
     act_on: str
     alias: str | None = None
     scheduler_tag: str | None = None
 
+    class _Model(BaseModel):
+        act_on: str
+        alias: str | None = None
+        scheduler_tag: str | None = None
+
     async def apply_trigger(self, context: Context, action: EvaluatedAction):
         logger.debug("Applying once trigger", act_on=str(self.act_on))
         alias = self.alias if self.alias is not None else self.act_on
@@ -42,11 +51,16 @@ async def apply_trigger(self, context: Context, action: EvaluatedAction):
 
 
 @dataclass
-class PlanTriggerPeriodic(PlanTrigger):
+class PlanTriggerPeriodic(PlanTrigger, Serializable):
     interval: timedelta
     first_run_time: time | None = None
     alias: str | None = None
 
+    class _Model(BaseModel):
+        interval: timedelta
+        first_run_time: time | None = None
+        alias: str | None = None
+
     async def apply_trigger(self, context: Context, action: EvaluatedAction):
         logger.debug("Applying periodic trigger", interval=str(self.interval), first_run_time=str(self.first_run_time))
         await context.scheduler.cyclic(
@@ -55,7 +69,10 @@ async def apply_trigger(self, context: Context, action: EvaluatedAction):
 
 
 @dataclass
-class PlanTriggerImmediately(PlanTrigger):
+class PlanTriggerImmediately(PlanTrigger, Serializable):
+    class _Model(BaseModel):
+        pass
+
     async def apply_trigger(self, context: Context, action: EvaluatedAction):
         logger.info(f"Executing action immediately: {action}")
         await action()
@@ -70,8 +87,7 @@ async def apply_trigger(self, context: Context, action: EvaluatedAction):
         self._action = action
         context.add_task_to_pool(listen_task)
 
-    async def _handle_event(self, hevent: HueEvent):
-        ...
+    async def _handle_event(self, hevent: HueEvent): ...
 
     async def _listener(self, context: Context):
         retry_counter = 0  # Initialize the retry counter
@@ -112,10 +128,14 @@ async def terminate():
 
 
 @dataclass
-class PlanTriggerOnHueButtonEvent(PlanTriggerOnHueEvent):
+class PlanTriggerOnHueButtonEvent(PlanTriggerOnHueEvent, Serializable):
     resource_id: str = ""
     action: str = ""
 
+    class _Model(BaseModel):
+        resource_id: str
+        action: str
+
     def __post_init__(self):
         if self.resource_id == "" or self.action == "":
             raise ValueError("Fields 'resource_id' and 'action' cannot be empty")