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..84d0dd66 100644 --- a/src/wled/models.py +++ b/src/wled/models.py @@ -338,6 +338,84 @@ 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] + + +@dataclass(kw_only=True) +class SegmentUpdate: + """Patch payload for :meth:`wled.WLED.segment`. + + 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.""" + + 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. + """ + + 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..85bb6381 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,134 @@ 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, + 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 + + 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, + "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, + } + segment = {k: v for k, v in segment.items() if v is not None} + + if colors := self._build_color_list( + update.segment_id, + update.color_primary, + update.color_secondary, + update.color_tertiary, + ): + segment["col"] = colors + + if segment: + segment["id"] = update.segment_id + + return segment + + async def segment( + self, + updates: list[SegmentUpdate], *, - 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: - """Change state of a WLED Light segment. + """Change state of one or more WLED Light segments. 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. + 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. - cct: White spectrum color temperature. Raises: ------ @@ -428,81 +496,15 @@ 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, - } - - # 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: - segment["col"] = colors + segments = [ + payload + for update in updates + if (payload := self._build_segment_payload(update)) + ] - if segment: - segment["id"] = segment_id - state["seg"] = [segment] + state: dict[str, Any] = {} + if segments: + state["seg"] = segments if transition is not None: state["tt"] = transition @@ -632,7 +634,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..466acc44 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([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, 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, 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, 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, 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, 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, color_tertiary=(0, 0, 255)) + await wled.segment([SegmentUpdate(segment_id=0, color_tertiary=(0, 0, 255))]) assert_post_payload( responses,