Fetch effects from /json/effects to handle ESP8266 buffer truncation

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,32 @@ 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:
+            if not (effects := await self.request("/json/effects")):
+                msg = (
+                    f"WLED device at {self.host} returned an empty API"
+                    " response on effects update"
+                )
+                raise WLEDEmptyResponseError(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 +847,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 or self._device.effects 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:

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(

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,124 @@ 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"]
+
+    with patch("wled.wled.time.time", return_value=1_000_000.0):
+        # 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["info"]["fxcount"] += 1
+    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"]
+    initial_effect_count = len(device1.effects)
+
+    device2 = await wled.update()
+    assert device2.info.effect_count == restarted_data["info"]["fxcount"]
+    # "Post Restart Effect" added — re-fetch after restart brought it in
+    assert len(device2.effects) == initial_effect_count + 1
+
+
+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 in-place — 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: