Merge pull request #43 from graphras-com/feature/scene-domain-rewrite

feat: reimplement Scene domain with create, apply, and delete support

Author: Faerkeren

docs/reference/domains/scene.md

@@ -0,0 +1,3 @@
+# Scene
+
+::: haclient.domains.scene

mkdocs.yml

@@ -35,6 +35,7 @@ nav:
           - Sensor: reference/domains/sensor.md
           - Binary Sensor: reference/domains/binary_sensor.md
           - Media Player: reference/domains/media_player.md
+          - Scene: reference/domains/scene.md
           - Timer: reference/domains/timer.md
 
 plugins:

src/haclient/client.py

@@ -429,6 +429,73 @@ def scene(self, name: str) -> Scene:
 
         return self._get_or_create("scene", name, _Scene)
 
+    async def create_scene(
+        self,
+        scene_id: str,
+        entities: dict[str, dict[str, Any]],
+        *,
+        snapshot_entities: list[str] | None = None,
+    ) -> Scene:
+        """Create a dynamic scene and return the `Scene` object.
+
+        This calls the ``scene.create`` service, which creates (or updates)
+        a scene at runtime.  The resulting scene can later be deleted with
+        `Scene.delete`.
+
+        Parameters
+        ----------
+        scene_id : str
+            The object-id for the new scene (e.g. ``"romantic"`` becomes
+            ``scene.romantic``).
+        entities : dict[str, dict[str, Any]]
+            Mapping of entity IDs to the state/attribute dicts that the
+            scene should apply.  For example::
+
+                {"light.ceiling": {"state": "on", "brightness": 120}}
+        snapshot_entities : list of str or None, optional
+            Entity IDs whose **current** state should be captured into
+            the scene instead of using explicit values.
+
+        Returns
+        -------
+        Scene
+            The newly created (or updated) `Scene` instance.
+        """
+        from .domains.scene import Scene as _Scene
+
+        data: dict[str, Any] = {
+            "scene_id": scene_id,
+            "entities": entities,
+        }
+        if snapshot_entities is not None:
+            data["snapshot_entities"] = snapshot_entities
+
+        await self._call_service("scene", "create", data)
+        return self._get_or_create("scene", scene_id, _Scene)
+
+    async def apply_scene(
+        self,
+        entities: dict[str, dict[str, Any]],
+        *,
+        transition: float | None = None,
+    ) -> None:
+        """Apply entity states without creating a persistent scene.
+
+        This calls the ``scene.apply`` service.  It works like activating
+        a scene, but the state combination is not saved.
+
+        Parameters
+        ----------
+        entities : dict[str, dict[str, Any]]
+            Mapping of entity IDs to desired state/attribute dicts.
+        transition : float or None, optional
+            Transition time in seconds for entities that support it.
+        """
+        data: dict[str, Any] = {"entities": entities}
+        if transition is not None:
+            data["transition"] = transition
+        await self._call_service("scene", "apply", data)
+
     def timer(self, name: str | None = None, *, persistent: bool = False) -> Timer:
         """Return a `Timer`, creating the Python object if needed.
 

src/haclient/domains/scene.py

@@ -1,4 +1,13 @@
-"""``scene`` domain implementation."""
+"""``scene`` domain implementation.
+
+Scenes apply a pre-defined set of entity states in one shot.  They are
+fire-and-forget: there is no ``turn_off`` counterpart.  The entity
+``state`` is the ISO-8601 timestamp of the last activation (or
+``"unavailable"`` / ``"unknown"`` when not applicable).
+
+Dynamic scenes (created via `HAClient.create_scene`) can also be
+deleted through the `Scene.delete` method.
+"""
 
 from __future__ import annotations
 
@@ -87,6 +96,20 @@ async def activate(self, *, transition: float | None = None) -> None:
             data = {"transition": transition}
         await self._call_service("turn_on", data)
 
+    async def delete(self) -> None:
+        """Delete this dynamically-created scene.
+
+        Only scenes created via ``HAClient.create_scene`` (i.e. the
+        ``scene.create`` service) can be deleted.  Attempting to delete
+        a YAML-defined scene will raise an error from Home Assistant.
+
+        Raises
+        ------
+        HAClientError
+            If Home Assistant rejects the deletion.
+        """
+        await self._call_service("delete")
+
     # -- Listener decorators --
 
     def on_activate(self, func: Any) -> Any:

src/haclient/sync.py

@@ -254,6 +254,51 @@ def scene(self, name: str) -> Any:
         """
         return _SyncProxy(self._client.scene(name), self._loop_thread)
 
+    def create_scene(
+        self,
+        scene_id: str,
+        entities: dict[str, dict[str, Any]],
+        *,
+        snapshot_entities: list[str] | None = None,
+    ) -> Any:
+        """Create a dynamic scene synchronously.
+
+        Parameters
+        ----------
+        scene_id : str
+            The object-id for the new scene.
+        entities : dict[str, dict[str, Any]]
+            Mapping of entity IDs to desired state/attribute dicts.
+        snapshot_entities : list of str or None, optional
+            Entity IDs whose current state should be captured.
+
+        Returns
+        -------
+        Any
+            A sync proxy wrapping the new `Scene`.
+        """
+        scene = self._loop_thread.submit(
+            self._client.create_scene(scene_id, entities, snapshot_entities=snapshot_entities)
+        )
+        return _SyncProxy(scene, self._loop_thread)
+
+    def apply_scene(
+        self,
+        entities: dict[str, dict[str, Any]],
+        *,
+        transition: float | None = None,
+    ) -> None:
+        """Apply entity states without creating a persistent scene.
+
+        Parameters
+        ----------
+        entities : dict[str, dict[str, Any]]
+            Mapping of entity IDs to desired state/attribute dicts.
+        transition : float or None, optional
+            Transition time in seconds.
+        """
+        self._loop_thread.submit(self._client.apply_scene(entities, transition=transition))
+
     def timer(self, name: str) -> Any:
         """Return a sync proxy wrapping the async `Timer`.
 

tests/test_client.py

@@ -115,6 +115,74 @@ async def test_call_service_via_rest_fallback(fake_ha: FakeHA) -> None:
     assert fake_ha.rest_service_calls == [("switch", "toggle", {"entity_id": "switch.x"})]
 
 
+async def test_create_scene(fake_ha: FakeHA) -> None:
+    ha = HAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+    try:
+        await ha.connect()
+        scene = await ha.create_scene(
+            "romantic",
+            {"light.ceiling": {"state": "on", "brightness": 80}},
+        )
+        assert scene.entity_id == "scene.romantic"
+        calls = fake_ha.ws_service_calls
+        assert len(calls) == 1
+        assert calls[0]["domain"] == "scene"
+        assert calls[0]["service"] == "create"
+        assert calls[0]["service_data"]["scene_id"] == "romantic"
+        assert calls[0]["service_data"]["entities"] == {
+            "light.ceiling": {"state": "on", "brightness": 80}
+        }
+    finally:
+        await ha.close()
+
+
+async def test_create_scene_with_snapshot(fake_ha: FakeHA) -> None:
+    ha = HAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+    try:
+        await ha.connect()
+        scene = await ha.create_scene(
+            "snapshot_test",
+            {"light.ceiling": {"state": "on"}},
+            snapshot_entities=["light.lamp"],
+        )
+        assert scene.entity_id == "scene.snapshot_test"
+        calls = fake_ha.ws_service_calls
+        assert calls[0]["service_data"]["snapshot_entities"] == ["light.lamp"]
+    finally:
+        await ha.close()
+
+
+async def test_apply_scene(fake_ha: FakeHA) -> None:
+    ha = HAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+    try:
+        await ha.connect()
+        await ha.apply_scene({"light.ceiling": {"state": "on", "brightness": 200}})
+        calls = fake_ha.ws_service_calls
+        assert len(calls) == 1
+        assert calls[0]["domain"] == "scene"
+        assert calls[0]["service"] == "apply"
+        assert calls[0]["service_data"]["entities"] == {
+            "light.ceiling": {"state": "on", "brightness": 200}
+        }
+        assert "transition" not in calls[0]["service_data"]
+    finally:
+        await ha.close()
+
+
+async def test_apply_scene_with_transition(fake_ha: FakeHA) -> None:
+    ha = HAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+    try:
+        await ha.connect()
+        await ha.apply_scene(
+            {"light.ceiling": {"state": "on"}},
+            transition=3.0,
+        )
+        calls = fake_ha.ws_service_calls
+        assert calls[0]["service_data"]["transition"] == 3.0
+    finally:
+        await ha.close()
+
+
 async def test_invalid_entity_id_direct_construction() -> None:
     ha = HAClient("http://x", "t")
     from haclient import Light

tests/test_domains.py

@@ -470,6 +470,16 @@ def _listener(old: Any, new: Any) -> None:
     assert fired[0][1] == "2024-06-15T20:30:00+00:00"
 
 
+async def test_scene_delete(client: HAClient, fake_ha: FakeHA) -> None:
+    sc = client.scene("romantic")
+    await sc.delete()
+    calls = fake_ha.ws_service_calls
+    assert len(calls) == 1
+    assert calls[0]["domain"] == "scene"
+    assert calls[0]["service"] == "delete"
+    assert calls[0]["service_data"]["entity_id"] == "scene.romantic"
+
+
 async def test_timer_time_remaining_active() -> None:
     """``time_remaining`` computes live seconds from ``finishes_at`` when active."""
     import datetime

tests/test_sync.py

@@ -122,3 +122,32 @@ def run() -> dict[str, str]:
     assert names["binary_sensor"] == "binary_sensor.b"
     assert names["scene"] == "scene.sc"
     assert names["timer"] == "timer.tm"
+
+
+async def test_sync_create_scene(fake_ha: FakeHA) -> None:
+    def run() -> str:
+        client = SyncHAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+        try:
+            client.connect()
+            scene = client.create_scene(
+                "romantic",
+                {"light.ceiling": {"state": "on", "brightness": 80}},
+            )
+            return scene.entity_id  # type: ignore[no-any-return]
+        finally:
+            client.close()
+
+    eid = await asyncio.get_running_loop().run_in_executor(None, run)
+    assert eid == "scene.romantic"
+
+
+async def test_sync_apply_scene(fake_ha: FakeHA) -> None:
+    def run() -> None:
+        client = SyncHAClient(fake_ha.base_url, fake_ha.token, ping_interval=0)
+        try:
+            client.connect()
+            client.apply_scene({"light.ceiling": {"state": "on"}})
+        finally:
+            client.close()
+
+    await asyncio.get_running_loop().run_in_executor(None, run)