From ffb2bd1ec9615728bedc5dd6173c24ae7111922f Mon Sep 17 00:00:00 2001 From: Frenckatron Date: Wed, 20 May 2026 22:36:21 +0200 Subject: [PATCH 1/3] Refactor WLED.segment() to accept a SegmentUpdate dataclass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapse the 19-parameter `segment()` signature into a single `SegmentUpdate` dataclass and extract three private helpers (`_resolve_effect`, `_resolve_palette`, `_build_color_list`) for the name-lookup and color-merge logic. The four-tag lint suppression block (`pylint: disable=too-many-locals, too-many-branches, too-many-arguments` plus `noqa: PLR0912, PLR0913`) is no longer needed and has been removed. `upgrade()` carries its own `too-many-branches` disable now — previously it was implicitly covered by the leaky scope of segment()'s disable comment. This is a breaking change: callers must now pass a `SegmentUpdate(...)` instance instead of keyword arguments to `WLED.segment(segment_id, ...)`. --- src/wled/__init__.py | 4 + src/wled/models.py | 81 +++++++++++++++ src/wled/wled.py | 233 ++++++++++++++++++++----------------------- tests/test_wled.py | 16 +-- 4 files changed, 201 insertions(+), 133 deletions(-) diff --git a/src/wled/__init__.py b/src/wled/__init__.py index d9ff9387..a6ee1df3 100644 --- a/src/wled/__init__.py +++ b/src/wled/__init__.py @@ -31,6 +31,8 @@ Preset, Releases, Segment, + SegmentColor, + SegmentUpdate, State, UDPSync, Wifi, @@ -55,6 +57,8 @@ "Preset", "Releases", "Segment", + "SegmentColor", + "SegmentUpdate", "SoundSimulationType", "State", "SyncGroup", diff --git a/src/wled/models.py b/src/wled/models.py index c068c7d8..d1f5e205 100644 --- a/src/wled/models.py +++ b/src/wled/models.py @@ -338,6 +338,87 @@ class Segment(BaseModel): """Transposes the segment, swapping X and Y dimensions (2D only).""" +SegmentColor = tuple[int, int, int, int] | tuple[int, int, int] +"""RGB or RGBW color tuple used when updating a segment.""" + + +@dataclass(kw_only=True) +class SegmentUpdate: + """Patch payload for :meth:`wled.WLED.segment`. + + Every field is optional; only fields that are not ``None`` are sent to the + WLED device. For ``name``, an empty string clears the name while ``None`` + leaves it unchanged. + """ + + brightness: int | None = None + """Brightness of the segment, between 0 and 255.""" + + clones: int | None = None + """Deprecated.""" + + color_primary: SegmentColor | None = None + """Primary color of the segment.""" + + color_secondary: SegmentColor | None = None + """Secondary color of the segment.""" + + color_tertiary: SegmentColor | None = None + """Tertiary color of the segment.""" + + effect: int | str | None = None + """Effect ID (or name) to use on this segment.""" + + freeze: bool | None = None + """Freeze the current segment state.""" + + individual: list[int | list[int] | SegmentColor] | None = None + """List of colors to use for each LED in the segment.""" + + intensity: int | None = None + """Effect intensity to use on this segment.""" + + length: int | None = None + """Length of this segment.""" + + name: str | None = None + """Name of the segment. Pass an empty string to clear it.""" + + on: bool | None = None + """Turn this segment on or off.""" + + palette: int | str | None = None + """Palette ID (or name) to use on this segment.""" + + reverse: bool | None = None + """Flip the segment, causing animations to change direction.""" + + selected: bool | None = None + """Whether the segment is selected for state updates by non-segment APIs.""" + + speed: int | None = None + """Relative effect speed, between 0 and 255.""" + + start: int | None = None + """LED the segment starts at.""" + + stop: int | None = None + """LED the segment stops at, not included in range. + + Setting this to a value at or below ``start`` (``0`` is recommended) + invalidates and deletes the segment. + """ + + transition: int | None = None + """Duration of the crossfade. + + One unit is 100ms, so a value of 4 results in a 400ms transition. + """ + + cct: int | None = None + """White spectrum color temperature.""" + + @dataclass(kw_only=True) class Leds: """Object holding leds info from WLED.""" diff --git a/src/wled/wled.py b/src/wled/wled.py index 65b88ef6..9cb8d4be 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -24,10 +24,10 @@ WLEDInvalidResponseError, WLEDUpgradeError, ) -from .models import Device, Playlist, Preset, Releases +from .models import Device, Playlist, Preset, Releases, SegmentColor, SegmentUpdate if TYPE_CHECKING: - from collections.abc import Callable, Sequence + from collections.abc import Callable from awesomeversion import AwesomeVersion @@ -355,66 +355,92 @@ async def master( await self.request("/json/state", method="POST", data=state) - # pylint: disable=too-many-locals, too-many-branches, too-many-arguments - async def segment( # noqa: PLR0912, PLR0913 + def _resolve_effect(self, effect: int | str | None) -> int | str | None: + """Resolve an effect name to its numeric ID. + + Returns the original value if it is already numeric or ``None``. When + ``effect`` is a string, the matching effect ID is looked up on the + cached device; an unknown name resolves to ``None``. + """ + if effect is None or not isinstance(effect, str): + return effect + + assert self._device is not None # noqa: S101 — guaranteed by caller + return next( + ( + item.effect_id + for item in self._device.effects.values() + if item.name.lower() == effect.lower() + ), + None, + ) + + def _resolve_palette(self, palette: int | str | None) -> int | str | None: + """Resolve a palette name to its numeric ID. + + Returns the original value if it is already numeric or ``None``. When + ``palette`` is a string, the matching palette ID is looked up on the + cached device; an unknown name resolves to ``None``. + """ + if palette is None or not isinstance(palette, str): + return palette + + assert self._device is not None # noqa: S101 — guaranteed by caller + return next( + ( + item.palette_id + for item in self._device.palettes.values() + if item.name.lower() == palette.lower() + ), + None, + ) + + def _build_color_list( self, segment_id: int, - *, - brightness: int | None = None, - clones: int | None = None, - color_primary: tuple[int, int, int, int] | tuple[int, int, int] | None = None, - color_secondary: tuple[int, int, int, int] | tuple[int, int, int] | None = None, - color_tertiary: tuple[int, int, int, int] | tuple[int, int, int] | None = None, - effect: int | str | None = None, - freeze: bool | None = None, - individual: Sequence[ - int | Sequence[int] | tuple[int, int, int, int] | tuple[int, int, int] - ] - | None = None, - intensity: int | None = None, - length: int | None = None, - name: str | None = None, - on: bool | None = None, - palette: int | str | None = None, - reverse: bool | None = None, - selected: bool | None = None, - speed: int | None = None, - start: int | None = None, - stop: int | None = None, - transition: int | None = None, - cct: int | None = None, - ) -> None: + primary: SegmentColor | None, + secondary: SegmentColor | None, + tertiary: SegmentColor | None, + ) -> list[SegmentColor]: + """Build the ``col`` array for a segment update. + + WLED's JSON API takes colors as an ordered list. When the caller only + supplies a higher-tier color (secondary/tertiary), the lower tiers are + filled in from the cached segment state, falling back to black when no + previous value is known. + """ + if primary is None and secondary is None and tertiary is None: + return [] + + assert self._device is not None # noqa: S101 — guaranteed by caller + current = self._device.state.segments[segment_id].color + colors: list[SegmentColor] = [] + + if primary is not None: + colors.append(primary) + elif secondary is not None or tertiary is not None: + colors.append(current.primary if current else (0, 0, 0)) + + if secondary is not None: + colors.append(secondary) + elif tertiary is not None: + colors.append( + current.secondary if current and current.secondary else (0, 0, 0) + ) + + if tertiary is not None: + colors.append(tertiary) + + return colors + + async def segment(self, segment_id: int, update: SegmentUpdate) -> None: """Change state of a WLED Light segment. Args: ---- segment_id: The ID of the segment to adjust. - brightness: The brightness of the segment, between 0 and 255. - clones: Deprecated. - color_primary: The primary color of this segment. - color_secondary: The secondary color of this segment. - color_tertiary: The tertiary color of this segment. - effect: The effect number (or name) to use on this segment. - freeze: Freeze the current segment state. - individual: A list of colors to use for each LED in the segment. - intensity: The effect intensity to use on this segment. - length: The length of this segment. - name: The name of the segment. Pass an empty string to clear the - name. None leaves the name unchanged. - on: A boolean, true to turn this segment on, false otherwise. - palette: The palette number or name to use on this segment. - reverse: Flips the segment, causing animations to change direction. - selected: Selected segments will have their state (color/FX) updated - by APIs that don't support segments. - speed: The relative effect speed, between 0 and 255. - start: LED the segment starts at. - stop: LED the segment stops at, not included in range. If stop is - set to a lower or equal value than start (setting to 0 is - recommended), the segment is invalidated and deleted. - transition: Duration of the crossfade between different - colors/brightness levels. One unit is 100ms, so a value of 4 - results in a transition of 400ms. - cct: White spectrum color temperature. + update: The patch describing which fields to change. Only fields + set to a non-``None`` value are sent to the device. Raises: ------ @@ -428,84 +454,41 @@ async def segment( # noqa: PLR0912, PLR0913 msg = "Unable to communicate with WLED to get the current state" raise WLEDError(msg) - state = {} # type: ignore[var-annotated] - segment = { - "bri": brightness, - "cln": clones, - "frz": freeze, - "fx": effect, - "i": individual, - "ix": intensity, - "len": length, - "n": name, - "on": on, - "pal": palette, - "rev": reverse, - "sel": selected, - "start": start, - "stop": stop, - "sx": speed, - "cct": cct, + segment: dict[str, Any] = { + "bri": update.brightness, + "cln": update.clones, + "frz": update.freeze, + "fx": self._resolve_effect(update.effect), + "i": update.individual, + "ix": update.intensity, + "len": update.length, + "n": update.name, + "on": update.on, + "pal": self._resolve_palette(update.palette), + "rev": update.reverse, + "sel": update.selected, + "start": update.start, + "stop": update.stop, + "sx": update.speed, + "cct": update.cct, } - - # Find effect if it was based on a name - if effect is not None and isinstance(effect, str): - segment["fx"] = next( - ( - item.effect_id - for item in self._device.effects.values() - if item.name.lower() == effect.lower() - ), - None, - ) - - # Find palette if it was based on a name - if palette is not None and isinstance(palette, str): - segment["pal"] = next( - ( - item.palette_id - for item in self._device.palettes.values() - if item.name.lower() == palette.lower() - ), - None, - ) - - # Filter out not set values - state = {k: v for k, v in state.items() if v is not None} segment = {k: v for k, v in segment.items() if v is not None} - # Determine color set - colors = [] - if color_primary is not None: - colors.append(color_primary) - elif color_secondary is not None or color_tertiary is not None: - if clrs := self._device.state.segments[segment_id].color: - colors.append(clrs.primary) - else: - colors.append((0, 0, 0)) - - if color_secondary is not None: - colors.append(color_secondary) - elif color_tertiary is not None: - if ( - clrs := self._device.state.segments[segment_id].color - ) and clrs.secondary: - colors.append(clrs.secondary) - else: - colors.append((0, 0, 0)) - - if color_tertiary is not None: - colors.append(color_tertiary) - - if colors: + if colors := self._build_color_list( + segment_id, + update.color_primary, + update.color_secondary, + update.color_tertiary, + ): segment["col"] = colors + state: dict[str, Any] = {} if segment: segment["id"] = segment_id state["seg"] = [segment] - if transition is not None: - state["tt"] = transition + if update.transition is not None: + state["tt"] = update.transition await self.request("/json/state", method="POST", data=state) @@ -632,7 +615,7 @@ async def nightlight( nightlight = {k: v for k, v in nightlight.items() if v is not None} await self.request("/json/state", method="POST", data={"nl": nightlight}) - async def upgrade( # noqa: PLR0912 + async def upgrade( # noqa: PLR0912 # pylint: disable=too-many-branches self, *, version: str | AwesomeVersion, diff --git a/tests/test_wled.py b/tests/test_wled.py index 1a4649fc..b1882e2d 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -11,7 +11,7 @@ from aioresponses import aioresponses from yarl import URL -from wled import WLED, Device, Releases +from wled import WLED, Device, Releases, SegmentUpdate from wled.const import LiveDataOverride from wled.exceptions import ( WLEDConnectionClosedError, @@ -525,7 +525,7 @@ async def test_segment( content_type="application/json", ) - await wled.segment(0, **kwargs) + await wled.segment(0, SegmentUpdate(**kwargs)) assert_post_payload( responses, @@ -544,7 +544,7 @@ async def test_segment_with_transition(responses: aioresponses, wled: WLED) -> N content_type="application/json", ) - await wled.segment(0, brightness=100, transition=5) + await wled.segment(0, SegmentUpdate(brightness=100, transition=5)) assert_post_payload( responses, @@ -571,7 +571,7 @@ async def test_segment_calls_update_when_no_device( content_type="application/json", ) - await wled.segment(0, on=True) + await wled.segment(0, SegmentUpdate(on=True)) assert_post_payload( responses, @@ -592,7 +592,7 @@ async def test_segment_no_device_raises(wled: WLED) -> None: patch.object(wled, "update", new_callable=AsyncMock), pytest.raises(WLEDError, match="Unable to communicate"), ): - await wled.segment(0, on=True) + await wled.segment(0, SegmentUpdate(on=True)) async def test_segment_color_tertiary_no_secondary_in_state( @@ -610,7 +610,7 @@ async def test_segment_color_tertiary_no_secondary_in_state( content_type="application/json", ) - await wled.segment(0, color_tertiary=(0, 0, 255)) + await wled.segment(0, SegmentUpdate(color_tertiary=(0, 0, 255))) assert_post_payload( responses, @@ -640,7 +640,7 @@ async def test_segment_secondary_no_color_in_state( ) # color is None, so it should use (0,0,0) fallback - await wled.segment(0, color_secondary=(0, 255, 0)) + await wled.segment(0, SegmentUpdate(color_secondary=(0, 255, 0))) assert_post_payload( responses, @@ -669,7 +669,7 @@ async def test_segment_tertiary_no_color_in_state( content_type="application/json", ) - await wled.segment(0, color_tertiary=(0, 0, 255)) + await wled.segment(0, SegmentUpdate(color_tertiary=(0, 0, 255))) assert_post_payload( responses, From 909db25d6ea5ab433acad6cc0d7b3134adcf8b11 Mon Sep 17 00:00:00 2001 From: Frenckatron Date: Thu, 21 May 2026 09:14:06 +0200 Subject: [PATCH 2/3] fix: replace SegmentColor docstring with comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The free-floating string after the SegmentColor type alias was flagged by the check-docstring-first hook as a second module docstring. Convert it to a regular comment — it had no runtime semantics anyway. --- src/wled/models.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/wled/models.py b/src/wled/models.py index d1f5e205..2b058490 100644 --- a/src/wled/models.py +++ b/src/wled/models.py @@ -338,8 +338,8 @@ class Segment(BaseModel): """Transposes the segment, swapping X and Y dimensions (2D only).""" +# RGB or RGBW color tuple used when updating a segment. SegmentColor = tuple[int, int, int, int] | tuple[int, int, int] -"""RGB or RGBW color tuple used when updating a segment.""" @dataclass(kw_only=True) From a17cbf1800ab07d3bbaacdeffb65b46feaf804de Mon Sep 17 00:00:00 2001 From: Frenckatron Date: Thu, 21 May 2026 09:38:28 +0200 Subject: [PATCH 3/3] pr-review: address feedback on #2077 --- src/wled/models.py | 15 ++++------ src/wled/wled.py | 73 +++++++++++++++++++++++++++++----------------- tests/test_wled.py | 46 ++++++++++++++++++++++++----- 3 files changed, 91 insertions(+), 43 deletions(-) diff --git a/src/wled/models.py b/src/wled/models.py index 2b058490..84d0dd66 100644 --- a/src/wled/models.py +++ b/src/wled/models.py @@ -346,11 +346,14 @@ class Segment(BaseModel): class SegmentUpdate: """Patch payload for :meth:`wled.WLED.segment`. - Every field is optional; only fields that are not ``None`` are sent to the - WLED device. For ``name``, an empty string clears the name while ``None`` - leaves it unchanged. + Every field except ``segment_id`` is optional; only fields that are not + ``None`` are sent to the WLED device. For ``name``, an empty string clears + the name while ``None`` leaves it unchanged. """ + segment_id: int + """ID of the segment to update.""" + brightness: int | None = None """Brightness of the segment, between 0 and 255.""" @@ -409,12 +412,6 @@ class SegmentUpdate: invalidates and deletes the segment. """ - transition: int | None = None - """Duration of the crossfade. - - One unit is 100ms, so a value of 4 results in a 400ms transition. - """ - cct: int | None = None """White spectrum color temperature.""" diff --git a/src/wled/wled.py b/src/wled/wled.py index 9cb8d4be..85bb6381 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -433,27 +433,8 @@ def _build_color_list( return colors - async def segment(self, segment_id: int, update: SegmentUpdate) -> None: - """Change state of a WLED Light segment. - - Args: - ---- - segment_id: The ID of the segment to adjust. - update: The patch describing which fields to change. Only fields - set to a non-``None`` value are sent to the device. - - Raises: - ------ - WLEDError: Something went wrong setting the segment state. - - """ - if self._device is None: - await self.update() - - if self._device is None: - msg = "Unable to communicate with WLED to get the current state" - raise WLEDError(msg) - + def _build_segment_payload(self, update: SegmentUpdate) -> dict[str, Any]: + """Build the JSON payload for a single segment update.""" segment: dict[str, Any] = { "bri": update.brightness, "cln": update.clones, @@ -475,20 +456,58 @@ async def segment(self, segment_id: int, update: SegmentUpdate) -> None: segment = {k: v for k, v in segment.items() if v is not None} if colors := self._build_color_list( - segment_id, + update.segment_id, update.color_primary, update.color_secondary, update.color_tertiary, ): segment["col"] = colors - state: dict[str, Any] = {} if segment: - segment["id"] = segment_id - state["seg"] = [segment] + segment["id"] = update.segment_id + + return segment + + async def segment( + self, + updates: list[SegmentUpdate], + *, + transition: int | None = None, + ) -> None: + """Change state of one or more WLED Light segments. - if update.transition is not None: - state["tt"] = update.transition + Args: + ---- + updates: One :class:`SegmentUpdate` per segment to adjust. Only + fields set to a non-``None`` value are sent to the device. + transition: Duration of the crossfade between different + colors/brightness levels. One unit is 100ms, so a value of 4 + results in a transition of 400ms. + + Raises: + ------ + WLEDError: Something went wrong setting the segment state. + + """ + if self._device is None: + await self.update() + + if self._device is None: + msg = "Unable to communicate with WLED to get the current state" + raise WLEDError(msg) + + segments = [ + payload + for update in updates + if (payload := self._build_segment_payload(update)) + ] + + state: dict[str, Any] = {} + if segments: + state["seg"] = segments + + if transition is not None: + state["tt"] = transition await self.request("/json/state", method="POST", data=state) diff --git a/tests/test_wled.py b/tests/test_wled.py index b1882e2d..466acc44 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -525,7 +525,7 @@ async def test_segment( content_type="application/json", ) - await wled.segment(0, SegmentUpdate(**kwargs)) + await wled.segment([SegmentUpdate(segment_id=0, **kwargs)]) assert_post_payload( responses, @@ -544,7 +544,7 @@ async def test_segment_with_transition(responses: aioresponses, wled: WLED) -> N content_type="application/json", ) - await wled.segment(0, SegmentUpdate(brightness=100, transition=5)) + await wled.segment([SegmentUpdate(segment_id=0, brightness=100)], transition=5) assert_post_payload( responses, @@ -559,6 +559,38 @@ async def test_segment_with_transition(responses: aioresponses, wled: WLED) -> N ) +async def test_segment_multiple_updates(responses: aioresponses, wled: WLED) -> None: + """Test updating multiple segments in a single call.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment( + [ + SegmentUpdate(segment_id=0, brightness=100), + SegmentUpdate(segment_id=1, on=True), + ], + transition=10, + ) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"bri": 100, "id": 0}, + {"on": True, "id": 1}, + ], + "tt": 10, + "v": True, + }, + ) + + async def test_segment_calls_update_when_no_device( responses: aioresponses, wled: WLED ) -> None: @@ -571,7 +603,7 @@ async def test_segment_calls_update_when_no_device( content_type="application/json", ) - await wled.segment(0, SegmentUpdate(on=True)) + await wled.segment([SegmentUpdate(segment_id=0, on=True)]) assert_post_payload( responses, @@ -592,7 +624,7 @@ async def test_segment_no_device_raises(wled: WLED) -> None: patch.object(wled, "update", new_callable=AsyncMock), pytest.raises(WLEDError, match="Unable to communicate"), ): - await wled.segment(0, SegmentUpdate(on=True)) + await wled.segment([SegmentUpdate(segment_id=0, on=True)]) async def test_segment_color_tertiary_no_secondary_in_state( @@ -610,7 +642,7 @@ async def test_segment_color_tertiary_no_secondary_in_state( content_type="application/json", ) - await wled.segment(0, SegmentUpdate(color_tertiary=(0, 0, 255))) + await wled.segment([SegmentUpdate(segment_id=0, color_tertiary=(0, 0, 255))]) assert_post_payload( responses, @@ -640,7 +672,7 @@ async def test_segment_secondary_no_color_in_state( ) # color is None, so it should use (0,0,0) fallback - await wled.segment(0, SegmentUpdate(color_secondary=(0, 255, 0))) + await wled.segment([SegmentUpdate(segment_id=0, color_secondary=(0, 255, 0))]) assert_post_payload( responses, @@ -669,7 +701,7 @@ async def test_segment_tertiary_no_color_in_state( content_type="application/json", ) - await wled.segment(0, SegmentUpdate(color_tertiary=(0, 0, 255))) + await wled.segment([SegmentUpdate(segment_id=0, color_tertiary=(0, 0, 255))]) assert_post_payload( responses,