Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/wled/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
Preset,
Releases,
Segment,
SegmentColor,
SegmentUpdate,
State,
UDPSync,
Wifi,
Expand All @@ -55,6 +57,8 @@
"Preset",
"Releases",
"Segment",
"SegmentColor",
"SegmentUpdate",
"SoundSimulationType",
"State",
"SyncGroup",
Expand Down
78 changes: 78 additions & 0 deletions src/wled/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
254 changes: 128 additions & 126 deletions src/wled/wled.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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(
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method should be called "segments," and then the "segment" method should look something like this:

def segment(self, segment_id: int, *, transition: int | None = None, **updates):
    return self.segments(
        SegmentUpdate(
            segment_id=segment_id, 
            **updates
        ),
        transition
    )

What do you think about this? This way we do not have code duplication, but at the same time we have support for batch updating of segment. In the case of animations and others, it sometimes matters whether segments are changed as two requests one after the other or as one action. Related discussion: https://github.com/orgs/home-assistant/discussions/1719

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:
------
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading