Listen for preset changes

src/wled/wled.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import socket
+import time
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Self
 
@@ -30,6 +31,12 @@
     from .const import LiveDataOverride
 
 
+@dataclass
+class _PresetsVersion:
+    presets_modified_timestamp: int
+    boot_time: int
+
+
 @dataclass
 class WLED:
     """Main class for handling connections with WLED."""
@@ -41,6 +48,7 @@ class WLED:
     _client: aiohttp.ClientWebSocketResponse | None = None
     _close_session: bool = False
     _device: Device | None = None
+    _presets_version: _PresetsVersion | None = None
 
     @property
     def connected(self) -> bool:
@@ -119,7 +127,21 @@ async def listen(self, callback: Callable[[Device], None]) -> None:
 
             if message.type == aiohttp.WSMsgType.TEXT:
                 message_data = message.json()
+
+                presets_changed, new_presets_version = self._check_presets_version(
+                    message_data
+                )
+                if presets_changed:
+                    if not (presets := await self.request("/presets.json")):
+                        err_msg = (
+                            f"WLED device at {self.host} returned an empty API"
+                            " response on presets update",
+                        )
+                        raise WLEDConnectionError(WLEDEmptyResponseError(err_msg))
+                    message_data["presets"] = presets
+
                 device = self._device.update_from_dict(data=message_data)
+                self._presets_version = new_presets_version
                 callback(device)
 
             if message.type in (
@@ -256,19 +278,22 @@ async def update(self) -> Device:
             )
             raise WLEDEmptyResponseError(msg)
 
-        if not (presets := await self.request("/presets.json")):
-            msg = (
-                f"WLED device at {self.host} returned an empty API"
-                " response on presets update",
-            )
-            raise WLEDEmptyResponseError(msg)
-        data["presets"] = presets
+        presets_changed, new_presets_version = self._check_presets_version(data)
+        if presets_changed:
+            if not (presets := await self.request("/presets.json")):
+                msg = (
+                    f"WLED device at {self.host} returned an empty API"
+                    " response on presets update",
+                )
+                raise WLEDEmptyResponseError(msg)
+            data["presets"] = presets
 
         if not self._device:
             self._device = Device.from_dict(data)
         else:
             self._device.update_from_dict(data)
 
+        self._presets_version = new_presets_version
         return self._device
 
     async def master(
@@ -703,6 +728,59 @@ async def __aexit__(self, *_exc_info: object) -> None:
         """
         await self.close()
 
+    def _check_presets_version(
+        self, device_data: Any
+    ) -> tuple[bool, _PresetsVersion | None]:
+        """Check if the presets have possibly been changed since last check.
+
+        Returns
+        -------
+            Tuple where first element denotes if presets have possibly changed, and
+            the second element the new version tuple. If version cannot be parsed, the
+            version is None.
+
+        """
+        # pylint: disable=too-many-boolean-expressions
+        if (
+            not isinstance(device_data, dict)
+            or not (info := device_data.get("info"))
+            or not (uptime := info.get("uptime"))
+            or not (fs := info.get("fs"))
+            or not (pmt := fs.get("pmt"))
+        ):
+            return (True, None)
+
+        try:
+            presets_modified_timestamp = int(pmt)
+            uptime_seconds = int(uptime)
+            current_time_seconds = int(time.time())
+            boot_time_approx = current_time_seconds - uptime_seconds
+            new_version = _PresetsVersion(
+                presets_modified_timestamp, int(boot_time_approx)
+            )
+
+            # Presets are the same if the presets last modified timestamp has not been
+            # modified. Since the last modified timestamp is only stored in memory, it
+            # will be reset to 0 on device restart, and we might miss an update. Detect
+            # device restarts by tracking the boot time.
+            #
+            # Since we are approximating the time, allow 2 seconds changes for rounding
+            # errors and network delay.
+            #
+            # We could get a better approximation of the boot time by using the device
+            # time from info.time. The device time is however unreliable, especially
+            # around boot.
+            changed = (
+                self._presets_version is None
+                or self._presets_version.presets_modified_timestamp
+                != new_version.presets_modified_timestamp
+                or abs(self._presets_version.boot_time - new_version.boot_time) > 2
+            )
+        except ValueError:
+            # In case of a parse failure, assume presets might have changed
+            return (True, None)
+        return (changed, new_version)
+
 
 @dataclass
 class WLEDReleases:

tests/test_wled.py

@@ -1,6 +1,7 @@
 """Tests for `wled.WLED`."""
 
 import asyncio
+import string
 
 import aiohttp
 import pytest
@@ -161,3 +162,132 @@ async def test_http_error500(aresponses: ResponsesMockServer) -> None:
         wled = WLED("example.com", session=session)
         with pytest.raises(WLEDError):
             assert await wled.request("/")
+
+
+@pytest.mark.asyncio
+async def test_presets_cache(aresponses: ResponsesMockServer) -> None:
+    """Test presets is fetched only when needed."""
+
+    def make_json(uptime: int, time: str, pmt: int) -> str:
+        return string.Template(
+            """
+            {"info": {"uptime":$uptime,"time":"$time","fs":{"pmt": $pmt}},
+            "state":{"bri":127,"on":true,"transition":7,"ps":-1,"pl":-1,
+            "nl":{"on":false,"dur":60,"mode":1,"tbri":0,"rem":-1},
+            "udpn":{"send":false,"recv":true,"sgrp":1,"rgrp":1},"lor":0,
+            "seg":[{"id":0,"start":0,"stop":48,"startY":0,"stopY":19,"len":48,
+            "grp":1,"spc":0,"of":0,"on":true,"frz":false,"bri":255,"cct":127,
+            "set":0,"lc":1,"col":[[18,22,255],[0,0,0],[0,0,0]],"fx":174,"sx":0,
+            "ix":91,"pal":0,"c1":128,"c2":128,"c3":16,"sel":true,"rev":false,
+            "mi":false,"rY":false,"mY":false,"tp":false,"o1":false,"o2":false,
+            "o3":false,"si":0,"m12":0,"bm":0}]}}
+            """
+        ).substitute({"uptime": uptime, "time": time, "pmt": pmt})
+
+    # First poll
+    aresponses.add(
+        "example.com",
+        "/json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=make_json(uptime=17836, time="2026-1-4, 18:32:43", pmt=1767549790),
+        ),
+    )
+    aresponses.add(
+        "example.com",
+        "/presets.json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text='{"0": {}, "1": {"n": "My Preset" }}',
+        ),
+    )
+
+    # Second poll, timestamp doesn't change, no fetching
+    aresponses.add(
+        "example.com",
+        "/json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=make_json(uptime=17836, time="2026-1-4, 18:32:43", pmt=1767549790),
+        ),
+    )
+
+    # Third poll, user renames a preset, timestamp changes
+    aresponses.add(
+        "example.com",
+        "/json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=make_json(uptime=17836, time="2026-1-4, 18:32:43", pmt=1767554102),
+        ),
+    )
+    aresponses.add(
+        "example.com",
+        "/presets.json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text='{"0": {}, "1": {"n": "My New Preset" }}',
+        ),
+    )
+
+    # Fourth poll, timestamp doesn't change, no fetching
+    aresponses.add(
+        "example.com",
+        "/json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=make_json(uptime=17836, time="2026-1-4, 18:32:43", pmt=1767554102),
+        ),
+    )
+
+    # Fifth poll, wled restart
+    aresponses.add(
+        "example.com",
+        "/json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=make_json(uptime=3, time="2026-1-4, 21:16:51", pmt=0),
+        ),
+    )
+    aresponses.add(
+        "example.com",
+        "/presets.json",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text='{"0": {}, "1": {"n": "My New Preset" }}',
+        ),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        wled = WLED("example.com", session=session)
+        response = await wled.update()
+        assert response.presets[1].name == "My Preset"
+
+        response = await wled.update()
+        assert response.presets[1].name == "My Preset"
+
+        response = await wled.update()
+        assert response.presets[1].name == "My New Preset"
+
+        response = await wled.update()
+        assert response.presets[1].name == "My New Preset"
+
+        response = await wled.update()
+        assert response.presets[1].name == "My New Preset"
+    aresponses.assert_plan_strictly_followed()