From c6432fa57c236a6ff872cadbaaf18bc5870e7329 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jarkko=20P=C3=B6yry?= Date: Mon, 5 Jan 2026 02:09:58 +0200 Subject: [PATCH 1/3] In poll and listen mode, fetch presets.json when needed. --- src/wled/wled.py | 84 ++++++++++++++++++++++++++--- tests/test_wled.py | 130 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 207 insertions(+), 7 deletions(-) diff --git a/src/wled/wled.py b/src/wled/wled.py index aa3b53d2..3d1ffe63 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -5,6 +5,7 @@ import asyncio import socket from dataclasses import dataclass +from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Self import aiohttp @@ -41,6 +42,7 @@ class WLED: _client: aiohttp.ClientWebSocketResponse | None = None _close_session: bool = False _device: Device | None = None + _presets_version: tuple[int, int] | None = None @property def connected(self) -> bool: @@ -119,7 +121,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 +272,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 +722,57 @@ async def __aexit__(self, *_exc_info: object) -> None: """ await self.close() + def _check_presets_version( + self, device_data: Any + ) -> tuple[bool, tuple[int, int] | 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 (time := info.get("time")) + 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 = ( + datetime.strptime(time, "%Y-%m-%d, %H:%M:%S") + .replace(tzinfo=timezone.utc) + .timestamp() + ) + boot_time_approx = current_time_seconds - uptime_seconds + new_version = (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 1 second changes for rounding + # errors. + changed = ( + self._presets_version is None + or self._presets_version[0] != new_version[0] + or abs(self._presets_version[1] - new_version[1]) > 1 + ) + except ValueError: + # In case of a parse failure, assume presets might have changed + return (True, None) + return (changed, new_version) + @dataclass class WLEDReleases: diff --git a/tests/test_wled.py b/tests/test_wled.py index 12b306bc..b35571c4 100644 --- a/tests/test_wled.py +++ b/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() From f0e4b4505d714d448ac1d5095e3829d0ecd15cba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jarkko=20P=C3=B6yry?= Date: Mon, 5 Jan 2026 14:58:54 +0200 Subject: [PATCH 2/3] Don't rely on device's own clock. --- src/wled/wled.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/src/wled/wled.py b/src/wled/wled.py index 3d1ffe63..00d4dde0 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -4,8 +4,8 @@ import asyncio import socket +import time from dataclasses import dataclass -from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Self import aiohttp @@ -739,7 +739,6 @@ def _check_presets_version( not isinstance(device_data, dict) or not (info := device_data.get("info")) or not (uptime := info.get("uptime")) - or not (time := info.get("time")) or not (fs := info.get("fs")) or not (pmt := fs.get("pmt")) ): @@ -748,11 +747,7 @@ def _check_presets_version( try: presets_modified_timestamp = int(pmt) uptime_seconds = int(uptime) - current_time_seconds = ( - datetime.strptime(time, "%Y-%m-%d, %H:%M:%S") - .replace(tzinfo=timezone.utc) - .timestamp() - ) + current_time_seconds = int(time.time()) boot_time_approx = current_time_seconds - uptime_seconds new_version = (presets_modified_timestamp, int(boot_time_approx)) @@ -761,12 +756,16 @@ def _check_presets_version( # 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 1 second changes for rounding - # errors. + # 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[0] != new_version[0] - or abs(self._presets_version[1] - new_version[1]) > 1 + or abs(self._presets_version[1] - new_version[1]) > 2 ) except ValueError: # In case of a parse failure, assume presets might have changed From 67171ab843d4af76ddc923a66dae049fb02d673d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jarkko=20P=C3=B6yry?= Date: Mon, 5 Jan 2026 15:42:38 +0200 Subject: [PATCH 3/3] Use nicer types for the presets version. --- src/wled/wled.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/wled/wled.py b/src/wled/wled.py index 00d4dde0..d4d00fe3 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -31,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.""" @@ -42,7 +48,7 @@ class WLED: _client: aiohttp.ClientWebSocketResponse | None = None _close_session: bool = False _device: Device | None = None - _presets_version: tuple[int, int] | None = None + _presets_version: _PresetsVersion | None = None @property def connected(self) -> bool: @@ -724,7 +730,7 @@ async def __aexit__(self, *_exc_info: object) -> None: def _check_presets_version( self, device_data: Any - ) -> tuple[bool, tuple[int, int] | None]: + ) -> tuple[bool, _PresetsVersion | None]: """Check if the presets have possibly been changed since last check. Returns @@ -749,7 +755,9 @@ def _check_presets_version( uptime_seconds = int(uptime) current_time_seconds = int(time.time()) boot_time_approx = current_time_seconds - uptime_seconds - new_version = (presets_modified_timestamp, int(boot_time_approx)) + 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 @@ -764,8 +772,9 @@ def _check_presets_version( # around boot. changed = ( self._presets_version is None - or self._presets_version[0] != new_version[0] - or abs(self._presets_version[1] - new_version[1]) > 2 + 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