diff --git a/src/wled/wled.py b/src/wled/wled.py index aa3b53d2..d4d00fe3 100644 --- a/src/wled/wled.py +++ b/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: 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()