diff --git a/src/wled/wled.py b/src/wled/wled.py index 65b88ef6..8c9a47fe 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -42,6 +42,14 @@ class _PresetsVersion: boot_time: int +@dataclass +class _EffectsVersion: + """Tracks effects state to avoid unnecessary fetches.""" + + effect_count: int + boot_time: int + + @dataclass class WLED: """Main class for handling connections with WLED.""" @@ -54,6 +62,7 @@ class WLED: _close_session: bool = False _device: Device | None = None _presets_version: _PresetsVersion | None = None + _effects_version: _EffectsVersion | None = None @property def connected(self) -> bool: @@ -138,7 +147,7 @@ async def listen(self, callback: Callable[[Device], None]) -> None: if not (presets := await self.request("/presets.json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on presets update", + " response on presets update" ) raise WLEDEmptyResponseError(msg) message_data["presets"] = presets @@ -302,7 +311,7 @@ async def update(self) -> Device: if not (data := await self.request("/json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on full update", + " response on full update" ) raise WLEDEmptyResponseError(msg) @@ -311,17 +320,33 @@ async def update(self) -> Device: if not (presets := await self.request("/presets.json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on presets update", + " response on presets update" ) raise WLEDEmptyResponseError(msg) data["presets"] = presets + changed_effects, new_effects_version = self._check_effects_changed(data) + if changed_effects: + effects = await self.request("/json/effects") + if not isinstance(effects, list): + msg = ( + f"WLED device at {self.host} returned an invalid" + " response on effects update" + ) + raise WLEDInvalidResponseError(msg) + data["effects"] = effects + else: + # Drop the possibly-truncated effects list from /json so that + # update_from_dict() keeps the cached full list from /json/effects. + data.pop("effects", None) + if not self._device: self._device = Device.from_dict(data) else: self._device.update_from_dict(data) self._presets_version = new_version + self._effects_version = new_effects_version return self._device async def master( @@ -823,6 +848,59 @@ def _check_presets_changed( ) return (changed, new_version) + def _check_effects_changed( + self, data: dict[str, Any] + ) -> tuple[bool, _EffectsVersion | None]: + """Check if effects have changed since the last check. + + Compares the effect count and approximate boot time to detect changes. + A significant shift in boot_time (> 2 s) signals a device restart. + + On ESP8266 devices the /json response may return a truncated effects + list due to a limited output buffer (WLED issue #5674). The initial + load therefore always fetches the complete list from /json/effects, + which is unaffected by that limitation. + + Returns + ------- + A tuple of (changed, new_version). If the version cannot be + determined from the data, returns (True, None) to trigger a + safe refetch. + + """ + if not isinstance(data, dict) or "info" not in data: + # No info in message (e.g. state-only WebSocket update), + # effects can't have changed. + return (False, self._effects_version) + + info = data["info"] + if (uptime := info.get("uptime")) is None or ( + fxcount := info.get("fxcount") + ) is None: + return (True, None) + + try: + new_version = _EffectsVersion( + effect_count=int(fxcount), + boot_time=int(time.time()) - int(uptime), + ) + except (ValueError, TypeError): + return (True, None) + + # For initial load, always fetch effects as /json may not include + # all effect information. + if self._device is None: + return (True, new_version) + + if self._effects_version is None: + return (True, new_version) + + changed = ( + self._effects_version.effect_count != new_version.effect_count + or abs(self._effects_version.boot_time - new_version.boot_time) > 2 + ) + return (changed, new_version) + @dataclass class WLEDReleases: diff --git a/tests/conftest.py b/tests/conftest.py index d62f32d8..4d602e7a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -43,7 +43,7 @@ def mock_json_and_presets( wled_data: dict[str, Any] | None = None, presets_data: dict[str, Any] | None = None, ) -> None: - """Register the two GET endpoints that WLED.update() calls.""" + """Register the three GET endpoints that WLED.update() calls.""" if wled_data is None: wled_data = load_fixture_json("wled") mocked.get( @@ -52,6 +52,12 @@ def mock_json_and_presets( body=json.dumps(wled_data), content_type="application/json", ) + mocked.get( + "http://example.com/json/effects", + status=200, + body=json.dumps(wled_data["effects"]), + content_type="application/json", + ) if presets_data is None: presets_data = load_fixture_json("presets") mocked.get( diff --git a/tests/test_wled.py b/tests/test_wled.py index 1a4649fc..7dd3b7cc 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -279,13 +279,19 @@ async def test_update_skips_presets_when_unchanged( """Test update() skips fetching presets.json when presets haven't changed.""" wled_data = load_fixture_json("wled") - # First update: fetches both /json and /presets.json + # First update: fetches /json, /json/effects, /presets.json responses.get( "http://example.com/json", status=200, body=json.dumps(wled_data), content_type="application/json", ) + responses.get( + "http://example.com/json/effects", + status=200, + body=json.dumps(wled_data["effects"]), + content_type="application/json", + ) responses.get( "http://example.com/presets.json", status=200, @@ -345,6 +351,120 @@ async def test_update_refetches_presets_when_info_incomplete( await wled.update() +async def test_update_skips_effects_when_unchanged( + responses: aioresponses, wled: WLED +) -> None: + """Test update() skips /json/effects when effect count and boot time unchanged.""" + wled_data = load_fixture_json("wled") + changed_data = json.loads(json.dumps(wled_data)) + changed_data["info"]["fxcount"] += 1 + changed_data["effects"] = wled_data["effects"] + ["New Effect"] + + # First update: fetches /json, /json/effects, /presets.json + mock_json_and_presets(responses, wled_data) + # Second update: same fxcount and boot_time — only /json fetched + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + # Third update: fxcount increased — /json/effects refetched with extra effect + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(changed_data), + content_type="application/json", + ) + responses.get( + "http://example.com/json/effects", + status=200, + body=json.dumps(changed_data["effects"]), + content_type="application/json", + ) + + device1 = await wled.update() + assert device1.info.effect_count == wled_data["info"]["fxcount"] + initial_effect_count = len(device1.effects) + + device2 = await wled.update() + assert device2.info.effect_count == wled_data["info"]["fxcount"] + assert len(device2.effects) == initial_effect_count # no re-fetch, unchanged + + device3 = await wled.update() + assert device3.info.effect_count == changed_data["info"]["fxcount"] + # "New Effect" added after fxcount bump — re-fetch brought it in + assert len(device3.effects) == initial_effect_count + 1 + + +async def test_update_refetches_effects_after_device_restart( + responses: aioresponses, wled: WLED +) -> None: + """Test update() refetches effects when a device restart is detected.""" + wled_data = load_fixture_json("wled") + restarted_data = json.loads(json.dumps(wled_data)) + restarted_data["info"]["uptime"] = 5 # uptime reset — device just booted + restarted_data["effects"] = wled_data["effects"] + ["Post Restart Effect"] + + mock_json_and_presets(responses, wled_data) + # After restart uptime drops from 32489 → 5, so boot_time shifts by ~32484s + mock_json_and_presets(responses, restarted_data) + + device1 = await wled.update() + assert device1.info.effect_count == wled_data["info"]["fxcount"] + + device2 = await wled.update() + # Refetch was triggered by boot_time shift, not fxcount — verify by content + assert any(e.name == "Post Restart Effect" for e in device2.effects.values()) + + +async def test_update_uses_effects_endpoint_for_full_list( + responses: aioresponses, wled: WLED +) -> None: + """Test update() uses /json/effects to get the complete effects list. + + Simulates the ESP8266 /json buffer overflow (WLED issue #5674): /json + returns a truncated effects list while /json/effects returns the full one. + """ + wled_data = load_fixture_json("wled") + full_effects = wled_data["effects"] + # Truncate list — simulates ESP8266 /json buffer overflow + wled_data["effects"] = full_effects[:1] + + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + responses.get( + "http://example.com/json/effects", + status=200, + body=json.dumps(full_effects), + content_type="application/json", + ) + responses.get( + "http://example.com/presets.json", + status=200, + body=json.dumps(load_fixture_json("presets")), + content_type="application/json", + ) + # Second update: /json still truncated, fxcount unchanged — no /json/effects stub. + # The cached full list must survive and not be overwritten by the truncated payload. + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + + device = await wled.update() + assert len(device.effects) == 3 # full list from /json/effects + + device = await wled.update() + assert len(device.effects) == 3 # still full — truncated /json did not overwrite + + async def test_listen_preset_change_via_websocket( responses: aioresponses, wled: WLED ) -> None: