diff --git a/flux_led/aiodevice.py b/flux_led/aiodevice.py index 89996e61..b806e160 100644 --- a/flux_led/aiodevice.py +++ b/flux_led/aiodevice.py @@ -34,7 +34,12 @@ STATE_GREEN, STATE_RED, STATE_WARM_WHITE, + ExtendedCustomEffectDirection, + ExtendedCustomEffectOption, + ExtendedCustomEffectPattern, MultiColorEffects, + ScribbleEffect, + ScribbleLED, ) from .protocol import ( POWER_RESTORE_BYTES_TO_POWER_RESTORE, @@ -422,6 +427,94 @@ async def async_set_custom_pattern( self._generate_custom_patterm(rgb_list, speed, transition_type) ) + async def async_set_extended_custom_effect( + self, + pattern_id: ExtendedCustomEffectPattern | int, + colors: list[tuple[int, int, int]], + speed: int = 50, + density: int = 50, + direction: ExtendedCustomEffectDirection | int = 0x01, + option: ExtendedCustomEffectOption | int = 0x00, + ) -> None: + """Set an extended custom effect on the device. + + Only supported on devices using the extended protocol (e.g., 0xB6). + + Args: + pattern_id: Pattern ID (1-22 or 101-102). See ExtendedCustomEffectPattern. + colors: List of 1-8 RGB color tuples + speed: Animation speed 0-100 (default 50) + density: Pattern density 0-100 (default 50) + direction: Animation direction (0x01=L->R, 0x02=R->L) + option: Pattern-specific option (default 0) + """ + if not self.supports_extended_custom_effects: + raise ValueError("device does not support extended custom effects") + await self._async_send_msg( + self._generate_extended_custom_effect( + pattern_id, colors, speed, density, direction, option + ) + ) + + async def async_set_custom_segment_colors( + self, + segments: list[tuple[int, int, int] | None], + ) -> None: + """Set custom colors for each segment on the device. + + Only supported on devices using the extended protocol (e.g., 0xB6). + Sets static HSV colors for each of 20 segments on the light strip. + + Args: + segments: List of up to 20 segment colors. Each is (R, G, B) or None for off. + """ + if not self.supports_extended_custom_effects: + raise ValueError("device does not support extended custom effects") + await self._async_send_msg(self._generate_custom_segment_colors(segments)) + + async def async_set_scribble( + self, + leds: list[ScribbleLED], + effect: ScribbleEffect | int = ScribbleEffect.STATIC, + direction: ExtendedCustomEffectDirection | int = ( + ExtendedCustomEffectDirection.LEFT_TO_RIGHT + ), + density: int = 80, + speed: int = 100, + enter_mode: bool = True, + ) -> None: + """Set a per-LED ('scribble') configuration on a 0xB6 device. + + Renders every LED via grouped E1 26 paints (one per color/blink group), + covering all N LEDs including an (0,0,0) group for off bulbs. Per-LED + blink/color come from each ScribbleLED. ``effect`` is a ScribbleEffect or + a raw int id 0x00-0x08 (ids 3,4,6,7 are valid effects with no UI name). + When enter_mode is True, first sends one all-zero E1 23 to guarantee + scribble-mode entry from another mode (E1 26 alone also works when + already in scribble). E1 23 does NOT render -- all colors render through + E1 26 (verified on hardware). + + Only supported on devices using the extended protocol (e.g., 0xB6). + """ + if not self.supports_scribble: + raise ValueError("device does not support scribble") + num_leds = self.led_count if self.led_count is not None else len(leds) + direction_val = ( + direction.value + if isinstance(direction, ExtendedCustomEffectDirection) + else int(direction) + ) + # Build (and validate) the paint messages up front so an invalid call + # raises before any network send -- otherwise the E1 23 init would leave + # the device in scribble-init mode with nothing rendered. + paint_msgs = self._scribble_paint_groups( + leds, effect, direction_val, density, speed, num_leds + ) + if enter_mode: + await self._async_send_msg(self._generate_scribble_init(num_leds)) + for msg in paint_msgs: + await self._async_send_msg(msg) + async def async_set_effect( self, effect: str, speed: int, brightness: int = 100 ) -> None: diff --git a/flux_led/base_device.py b/flux_led/base_device.py index f17e9208..efef22f2 100644 --- a/flux_led/base_device.py +++ b/flux_led/base_device.py @@ -49,6 +49,12 @@ STATE_RED, STATE_WARM_WHITE, STATIC_MODES, + ExtendedCustomEffectDirection, + ExtendedCustomEffectOption, + ExtendedCustomEffectPattern, + ScribbleBlinkMode, + ScribbleEffect, + ScribbleLED, WhiteChannelType, ) from .models_db import ( @@ -72,6 +78,7 @@ EFFECT_LIST, EFFECT_LIST_DIMMABLE, EFFECT_LIST_LEGACY_CCT, + EXTENDED_CUSTOM_EFFECT_ID_NAME, ORIGINAL_ADDRESSABLE_EFFECT_ID_NAME, ORIGINAL_ADDRESSABLE_EFFECT_NAME_ID, PresetPattern, @@ -634,6 +641,13 @@ def effect_list(self) -> list[str]: """Return the list of available effects.""" effects: Iterable[str] = [] protocol = self.protocol + # TODO: 0xB6 extended custom effects are reported by `effect` and are + # settable via the dedicated setExtendedCustomEffect API, but are not yet + # wired into the generic effect_list/set_effect path -- they require a + # color list, which set_effect(name, speed) cannot supply. Until a + # follow-up PR integrates them (set_effect -> setExtendedCustomEffect), a + # running extended scene may report an `effect` name not present here; + # discovery is available via `extended_custom_effect_pattern_list`. if protocol in OLD_EFFECTS_PROTOCOLS: effects = ORIGINAL_ADDRESSABLE_EFFECT_ID_NAME.values() elif protocol in NEW_EFFECTS_PROTOCOLS: @@ -653,6 +667,18 @@ def supports_extended_custom_effects(self) -> bool: """Return True if the device uses the extended custom protocol.""" return self.protocol == PROTOCOL_LEDENET_EXTENDED_CUSTOM + @property + def supports_scribble(self) -> bool: + """Return True if the device supports the per-LED scribble feature.""" + return self.protocol == PROTOCOL_LEDENET_EXTENDED_CUSTOM + + @property + def extended_custom_effect_pattern_list(self) -> list[str] | None: + """Return available extended custom effect patterns, or None if not supported.""" + if not self.supports_extended_custom_effects: + return None + return list(EXTENDED_CUSTOM_EFFECT_ID_NAME.values()) + @property def effect(self) -> str | None: """Return the current effect.""" @@ -667,6 +693,20 @@ def _named_effect(self) -> str | None: mode = self.raw_state.mode pattern_code = self.preset_pattern_num protocol = self.protocol + # Devices with extended custom effects use different pattern names + if self.supports_extended_custom_effects and pattern_code == 0x25: + # Preset 0x25 always means a scene is active, so an unmapped mode is + # still a real effect -- return a distinguishable placeholder rather + # than None (which would be indistinguishable from "no effect"). + name = EXTENDED_CUSTOM_EFFECT_ID_NAME.get(mode) + if name is None: + _LOGGER.debug("Unmapped extended custom effect mode: %s", mode) + return f"unknown ({mode:#04x})" + return name + # 0xB6 "Colorful" (solid or multi-segment) reports preset_pattern=0x24; it is a + # plain color, not an effect (verified on hardware), so report no effect. + if self.supports_extended_custom_effects and pattern_code == 0x24: + return None if protocol in OLD_EFFECTS_PROTOCOLS: effect_id = (pattern_code << 8) + mode - 99 return ORIGINAL_ADDRESSABLE_EFFECT_ID_NAME.get(effect_id) @@ -1325,6 +1365,258 @@ def _generate_custom_patterm( assert self._protocol is not None return self._protocol.construct_custom_effect(rgb_list, speed, transition_type) + def _generate_extended_custom_effect( + self, + pattern_id: ExtendedCustomEffectPattern | int, + colors: list[tuple[int, int, int]], + speed: int = 50, + density: int = 50, + direction: ExtendedCustomEffectDirection | int = 0x01, + option: ExtendedCustomEffectOption | int = 0x00, + ) -> bytearray: + """Generate the extended custom effect protocol bytes with validation. + + Only supported on devices using the extended protocol (e.g., 0xB6). + """ + # Accept the public enums (ExtendedCustomEffectPattern/Direction/Option) + # as well as plain ints by coercing enum members to their int value up + # front, before the int-literal validation below. + pattern_id = ( + pattern_id.value + if isinstance(pattern_id, ExtendedCustomEffectPattern) + else pattern_id + ) + direction = ( + direction.value + if isinstance(direction, ExtendedCustomEffectDirection) + else direction + ) + option = ( + option.value if isinstance(option, ExtendedCustomEffectOption) else option + ) + + # Validate pattern_id + valid_ids = set(range(1, 23)) | {101, 102} + if pattern_id not in valid_ids: + raise ValueError(f"Pattern ID must be 1-22 or 101-102, got {pattern_id}") + + # Reject more than 8 colors (the wire format supports at most 8) + if len(colors) > 8: + raise ValueError(f"at most 8 colors are supported, got {len(colors)}") + + # Require at least one color + if len(colors) == 0: + raise ValueError("Surplife pattern requires at least one color") + + # Validate color tuples + for idx, color in enumerate(colors): + if len(color) != 3: + raise ValueError(f"Color {idx} must be (R, G, B) tuple") + for c in color: + if not 0 <= c <= 255: + raise ValueError(f"Color values must be 0-255, got {c}") + + # General wire-byte bounds only. Per-effect applicability (which effects + # accept options/density/direction, and their valid option sets) is + # documented in the PR description and deliberately NOT enforced here yet. + if not 0 <= speed <= 100: + raise ValueError(f"speed must be 0-100, got {speed}") + if not 0 <= density <= 100: + raise ValueError(f"density must be 0-100, got {density}") + if direction not in (0x01, 0x02): + raise ValueError(f"direction must be 0x01 or 0x02, got {direction}") + if not 0 <= option <= 0x02: + raise ValueError(f"option must be 0-2, got {option}") + + assert self._protocol is not None + assert isinstance(self._protocol, ProtocolLEDENETExtendedCustom) + return self._protocol.construct_extended_custom_effect( + pattern_id, colors, speed, density, direction, option + ) + + def _generate_custom_segment_colors( + self, + segments: list[tuple[int, int, int] | None], + ) -> bytearray: + """Generate custom segment colors protocol bytes with validation. + + Only supported on devices using the extended protocol (e.g., 0xB6). + + Args: + segments: List of up to 20 segment colors. Each is (R, G, B) or None for off. + """ + # Reject more than 20 segments (the wire format supports at most 20) + if len(segments) > 20: + raise ValueError(f"at most 20 segments are supported, got {len(segments)}") + + # Validate color tuples + for idx, color in enumerate(segments): + if color is None: + continue + if len(color) != 3: + raise ValueError(f"Segment {idx} must be (R, G, B) tuple or None") + for c in color: + if not 0 <= c <= 255: + raise ValueError(f"Color values must be 0-255, got {c}") + + assert self._protocol is not None + assert isinstance(self._protocol, ProtocolLEDENETExtendedCustom) + return self._protocol.construct_custom_segment_colors(segments) + + def _generate_scribble_init(self, num_leds: int) -> bytearray: + """Generate the scribble-mode init (E1 23, non-rendering) bytes. + + E1 23 does not render -- it only guarantees scribble-mode entry. Colors + render through grouped E1 26 paints (see _scribble_paint_groups). + """ + # The E1 23 byte-8 count and the E1 26 bitmap sizing are one-byte / + # ceil(N/8) fields; N is unrepresentable above 255 on the wire. + if not 0 < num_leds <= 255: + raise ValueError(f"num_leds must be 1..255, got {num_leds}") + assert self._protocol is not None + assert isinstance(self._protocol, ProtocolLEDENETExtendedCustom) + return self._protocol.construct_scribble_init(num_leds) + + def _generate_scribble_paint( + self, + effect: int, + direction: int, + density: int, + speed: int, + color: tuple[int, int, int] | None, + white: int | None, + blink_mode: int, + blink_speed: int, + led_indices: list[int] | None, + num_leds: int, + ) -> bytearray: + """Generate a single scribble paint (E1 26) for one color/blink group. + + Color and white are mutually exclusive. Colors are converted to + (H/2, S, V) via the existing _rgb_to_hsv_bytes helper. + """ + if color is not None and white is not None: + raise ValueError("color and white are mutually exclusive") + assert self._protocol is not None + assert isinstance(self._protocol, ProtocolLEDENETExtendedCustom) + if color is not None: + h2, s, v = self._protocol._rgb_to_hsv_bytes(*color)[:3] + white_level = 0 + else: + h2 = s = v = 0 + white_level = white or 0 + if led_indices is not None: + for i in led_indices: + if not 0 <= i < num_leds: + raise ValueError(f"LED index {i} out of range for {num_leds} LEDs") + return self._protocol.construct_scribble_paint( + effect=effect, + direction=direction, + density=density, + speed=speed, + blink_mode=blink_mode, + h2=h2, + s=s, + v=v, + white=white_level, + blink_speed=blink_speed, + bitmap_leds=led_indices, + num_leds=num_leds, + ) + + def _scribble_paint_groups( + self, + leds: list[ScribbleLED], + effect: ScribbleEffect | int, + direction: int, + density: int, + speed: int, + num_leds: int, + ) -> list[bytearray]: + """Group per-LED settings and return one E1 26 paint per group. + + This is the primary render path for every scribble call (E1 23 does + not render). LEDs are grouped by (rgb, white, blink_mode, blink_speed) + in first-appearance order; an off LED (rgb None, white None) is painted + as color (0,0,0). The union of all group bitmaps covers exactly the N + LEDs, so the result is deterministic regardless of prior device state. + """ + if not leds: + raise ValueError("leds must not be empty") + if not 0 < num_leds <= 255: + raise ValueError(f"num_leds must be 1..255, got {num_leds}") + if self.led_count is not None and self.led_count != len(leds): + raise ValueError(f"Got {len(leds)} LEDs but device has {self.led_count}") + if num_leds != len(leds): + raise ValueError(f"num_leds {num_leds} does not match {len(leds)} LEDs") + + # effect may be a ScribbleEffect (the 5 named ones) or a raw int id. + # The device accepts effect ids 0x00-0x08 (verified on-device); ids 3,4, + # 6,7 are valid effects with no UI name, reachable only as raw ints. + effect_id = effect.value if isinstance(effect, ScribbleEffect) else int(effect) + if not 0x00 <= effect_id <= 0x08: + raise ValueError(f"scribble effect id must be 0x00-0x08, got {effect_id}") + if direction not in (0x01, 0x02): + raise ValueError(f"direction must be 0x01 or 0x02, got {direction}") + if not 0 <= density <= 100: + raise ValueError(f"density must be 0-100, got {density}") + if not 0 <= speed <= 100: + raise ValueError(f"speed must be 0-100, got {speed}") + + # Validate per-LED settings and bucket indices by group key. + groups: dict[ + tuple[tuple[int, int, int] | None, int | None, ScribbleBlinkMode, int], + list[int], + ] = {} + for idx, led in enumerate(leds): + if led.rgb is not None and led.white is not None: + raise ValueError(f"LED {idx}: rgb and white are mutually exclusive") + if led.rgb is not None: + if len(led.rgb) != 3 or any(not 0 <= c <= 255 for c in led.rgb): + raise ValueError(f"LED {idx}: rgb values must be 0-255") + if led.white is not None and not 0 <= led.white <= 100: + raise ValueError(f"LED {idx}: white must be 0-100") + if not 0 <= led.blink_speed <= 100: + raise ValueError(f"LED {idx}: blink_speed must be 0-100") + key = ( + led.rgb, + led.white, + led.blink_mode, + led.blink_speed if led.blink_mode is not ScribbleBlinkMode.NONE else 0, + ) + groups.setdefault(key, []).append(idx) + + messages: list[bytearray] = [] + for (rgb, white, blink_mode, _blink_speed), indices in groups.items(): + # Off LEDs (rgb None and white None) paint as color (0,0,0). + color: tuple[int, int, int] | None + paint_white: int | None + if rgb is None and white is None: + color = (0, 0, 0) + paint_white = None + else: + color = rgb + paint_white = white + # The group key normalizes blink_speed to 0 while blink is off, so + # take the emitted blink_speed from a representative LED to preserve + # the captured wire format (the device still carries a blink_speed + # byte even when not blinking; it is simply ignored there). + messages.append( + self._generate_scribble_paint( + effect=effect_id, + direction=direction, + density=density, + speed=speed, + color=color, + white=paint_white, + blink_mode=blink_mode.value, + blink_speed=leds[indices[0]].blink_speed, + led_indices=indices, + num_leds=num_leds, + ) + ) + return messages + def _effect_to_pattern(self, effect: str) -> int: """Convert an effect to a pattern code.""" protocol = self.protocol diff --git a/flux_led/const.py b/flux_led/const.py index ca122a66..3db23aea 100755 --- a/flux_led/const.py +++ b/flux_led/const.py @@ -4,7 +4,7 @@ from dataclasses import dataclass from enum import Enum -from typing import Final # pylint: disable=no-name-in-module +from typing import Final, NamedTuple # pylint: disable=no-name-in-module MIN_TEMP: Final = 2700 MAX_TEMP: Final = 6500 @@ -38,6 +38,101 @@ class MultiColorEffects(Enum): BREATHING = 0x05 +class ExtendedCustomEffectPattern(Enum): + """Pattern IDs for extended custom effects (e.g., 0xB6 device).""" + + WAVE = 0x01 + METEOR = 0x02 + STREAMER = 0x03 + BUILDING_BLOCKS = 0x04 + FLOWING_WATER = 0x05 + CHASE = 0x06 + HORSE_RACING = 0x07 + CYCLE = 0x08 + BREATHE = 0x09 + JUMP = 0x0A + STROBE = 0x0B + TWINKLING_STARS = 0x0C + STARS_WINK = 0x0D + WARNING = 0x0E + COLLISION = 0x0F + FIREWORKS = 0x10 + COMET = 0x11 + GRADIENT_METEOR = 0x12 + VOLCANO = 0x13 + SUPERLUMINAL = 0x14 + RAINBOW_BRIDGE = 0x15 + GRADIENT_OVERLAY = 0x16 + STATIC_GRADIENT = 0x65 + STATIC_FILL = 0x66 + + +class ExtendedCustomEffectDirection(Enum): + """Direction for extended custom effect animations.""" + + LEFT_TO_RIGHT = 0x01 + RIGHT_TO_LEFT = 0x02 + + +class ExtendedCustomEffectOption(Enum): + """Option values for extended custom effects. + + The meaning varies by pattern: + - Some patterns: DEFAULT=static color, VARIANT_1=color change + - Rainbow patterns: DEFAULT=strobe, VARIANT_1=twinkle, VARIANT_2=breathe + """ + + DEFAULT = 0x00 + VARIANT_1 = 0x01 + VARIANT_2 = 0x02 + + +class ScribbleEffect(Enum): + """Global effect id for the scribble feature (E1 26 byte 2). + + Distinct from ExtendedCustomEffectPattern (E1 21). Byte-2 gaps (0x04, + 0x06, 0x07, ...) imply more effects/variants than the UI exposes. + """ + + STATIC = 0x00 + FLOWING = 0x01 + TWINKLING_STARS = 0x02 + TWINKLING_STARS_VARIANT = 0x03 + STARS_WINK = 0x05 + ACCUMULATE = 0x08 + + +class ScribbleBlinkMode(Enum): + """Blink mode for the scribble global command (E1 26 byte 6).""" + + NONE = 0x00 + SLOW = 0x08 + FAST = 0x10 + + +class ScribbleLED(NamedTuple): + """One LED's setting in a scribble (per-LED) configuration. + + Color and white are mutually exclusive: + * rgb set, white None -> color mode; brightness is the RGB value itself + (the E1 26 V byte is derived from the RGB tuple) + * white set, rgb None -> white mode; the white level (0-100) is the + brightness (the E1 26 WLVL byte) + * both None -> LED off (painted as color (0,0,0)) + + Per-LED blink is achieved by grouping: LEDs sharing a + (blink_mode, blink_speed) are painted by one E1 26 and blink independently + per group. These are real, rendered fields that flow into the group key + and the E1 26 paint (the E1 23 init carries no color/blink and does not + render, so they never go into an E1 23 record). + """ + + rgb: tuple[int, int, int] | None = None # (R,G,B) 0-255, or None + white: int | None = None # warm-white level 0-100, or None + blink_mode: ScribbleBlinkMode = ScribbleBlinkMode.NONE # E1 26 byte 6 + blink_speed: int = 100 # 0-100, only meaningful when blinking + + DEFAULT_WHITE_CHANNEL_TYPE: Final = WhiteChannelType.WARM PRESET_MUSIC_MODE: Final = 0x62 diff --git a/flux_led/device.py b/flux_led/device.py index a5dae295..615f1889 100644 --- a/flux_led/device.py +++ b/flux_led/device.py @@ -22,6 +22,11 @@ STATE_GREEN, STATE_RED, STATE_WARM_WHITE, + ExtendedCustomEffectDirection, + ExtendedCustomEffectOption, + ExtendedCustomEffectPattern, + ScribbleEffect, + ScribbleLED, ) from .scanner import FluxLEDDiscovery from .sock import _socket_retry @@ -391,5 +396,107 @@ def setCustomPattern( retry=retry, ) + def setExtendedCustomEffect( + self, + pattern_id: ExtendedCustomEffectPattern | int, + colors: list[tuple[int, int, int]], + speed: int = 50, + density: int = 50, + direction: ExtendedCustomEffectDirection | int = 0x01, + option: ExtendedCustomEffectOption | int = 0x00, + retry: int = DEFAULT_RETRIES, + ) -> None: + """Set an extended custom effect on the device. + + Only supported on devices using the extended protocol (e.g., 0xB6). + + Args: + pattern_id: Pattern ID (1-22 or 101-102). See ExtendedCustomEffectPattern. + colors: List of 1-8 RGB color tuples, e.g., [(255, 0, 0), (0, 255, 0)] + speed: Animation speed 0-100 (default 50) + density: Pattern density 0-100 (default 50) + direction: Animation direction. See ExtendedCustomEffectDirection. + 0x01 = Left to Right (default) + 0x02 = Right to Left + option: Pattern-specific option (default 0) + retry: Number of retries on failure + """ + if not self.supports_extended_custom_effects: + raise ValueError("device does not support extended custom effects") + self._send_and_read_with_retry( + self._generate_extended_custom_effect( + pattern_id, colors, speed, density, direction, option + ), + 0, + retry=retry, + ) + + def setCustomSegmentColors( + self, + segments: list[tuple[int, int, int] | None], + retry: int = DEFAULT_RETRIES, + ) -> None: + """Set custom colors for each segment on the device. + + Only supported on devices using the extended protocol (e.g., 0xB6). + Sets static HSV colors for each of 20 segments on the light strip. + + Args: + segments: List of up to 20 segment colors. Each is (R, G, B) or None for off. + retry: Number of retries on failure + """ + if not self.supports_extended_custom_effects: + raise ValueError("device does not support extended custom effects") + self._send_and_read_with_retry( + self._generate_custom_segment_colors(segments), + 0, + retry=retry, + ) + + def setScribble( + self, + leds: list[ScribbleLED], + effect: ScribbleEffect | int = ScribbleEffect.STATIC, + direction: ExtendedCustomEffectDirection | int = ( + ExtendedCustomEffectDirection.LEFT_TO_RIGHT + ), + density: int = 80, + speed: int = 100, + enter_mode: bool = True, + retry: int = DEFAULT_RETRIES, + ) -> None: + """Set a per-LED ('scribble') configuration on a 0xB6 device. + + Renders every LED via grouped E1 26 paints (one per color/blink group), + covering all N LEDs including an (0,0,0) group for off bulbs. Per-LED + blink/color come from each ScribbleLED. ``effect`` is a ScribbleEffect or + a raw int id 0x00-0x08 (ids 3,4,6,7 are valid effects with no UI name). + When enter_mode is True, first sends one all-zero E1 23 to guarantee + scribble-mode entry. E1 23 does NOT render -- all colors render through + E1 26 (verified on hardware). + + Only supported on devices using the extended protocol (e.g., 0xB6). + """ + if not self.supports_scribble: + raise ValueError("device does not support scribble") + num_leds = self.led_count if self.led_count is not None else len(leds) + direction_val = ( + direction.value + if isinstance(direction, ExtendedCustomEffectDirection) + else int(direction) + ) + # Build (and validate) the paint messages up front so an invalid call + # raises before any network send -- otherwise the E1 23 init would leave + # the device in scribble-init mode with nothing rendered. + paint_msgs = self._scribble_paint_groups( + leds, effect, direction_val, density, speed, num_leds + ) + if enter_mode: + self._send_and_read_with_retry( + self._generate_scribble_init(num_leds), 0, retry=retry + ) + for msg in paint_msgs: + self._send_and_read_with_retry(msg, 0, retry=retry) + def refreshState(self) -> None: return self.update_state() diff --git a/flux_led/pattern.py b/flux_led/pattern.py index 28d7575a..e4602e85 100644 --- a/flux_led/pattern.py +++ b/flux_led/pattern.py @@ -201,6 +201,38 @@ v: k for k, v in ASSESSABLE_MULTI_COLOR_ID_NAME.items() } +# Extended custom effect pattern names for 0xB6 (Surplife) devices +# These map to ExtendedCustomEffectPattern enum values +EXTENDED_CUSTOM_EFFECT_ID_NAME = { + 0x01: "Wave", + 0x02: "Meteor", + 0x03: "Streamer", + 0x04: "Building Blocks", + 0x05: "Flowing Water", + 0x06: "Chase", + 0x07: "Horse Racing", + 0x08: "Cycle", + 0x09: "Breathe", + 0x0A: "Jump", + 0x0B: "Strobe", + 0x0C: "Twinkling Stars", + 0x0D: "Stars Wink", + 0x0E: "Warning", + 0x0F: "Collision", + 0x10: "Fireworks", + 0x11: "Comet", + 0x12: "Gradient Meteor", + 0x13: "Volcano", + 0x14: "Superluminal", + 0x15: "Rainbow Bridge", + 0x16: "Gradient Overlay", + 0x65: "Static Gradient", + 0x66: "Static Fill", +} +EXTENDED_CUSTOM_EFFECT_NAME_ID = { + v: k for k, v in EXTENDED_CUSTOM_EFFECT_ID_NAME.items() +} + ORIGINAL_ADDRESSABLE_EFFECT_ID_NAME = { 1: "Circulate all modes", 2: "7 colors change gradually", diff --git a/flux_led/protocol.py b/flux_led/protocol.py index e65a009e..b39fe723 100755 --- a/flux_led/protocol.py +++ b/flux_led/protocol.py @@ -7,6 +7,7 @@ import datetime import logging from abc import abstractmethod +from collections.abc import Iterable from dataclasses import dataclass from enum import Enum from typing import NamedTuple @@ -1680,6 +1681,311 @@ def extended_state_led_count(self, raw_state: bytes | bytearray) -> int | None: return None return raw_state[LEDENET_EXTENDED_STATE_LED_COUNT_POS] + def _rgb_to_hsv_bytes_rgbw( + self, r: int, g: int, b: int, white: int = 0 + ) -> list[int]: + """Convert RGBW (0-255) to 5-byte HSVW format for extended effect commands. + + Output format: + [H/2, S, V, 0x00, W] + 0 1 2 3 4 + | | | | white LED brightness (0-255) + | | | unused (always 0x00) + | | value/brightness (0-100) + | saturation (0-100) + hue divided by 2 (0-180) + + Args: + r: Red component (0-255) + g: Green component (0-255) + b: Blue component (0-255) + white: White LED brightness (0-255) + + Returns: + 5-byte list [H/2, S, V, 0x00, W] + """ + h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) + return [int(h * 180), int(s * 100), int(v * 100), 0x00, white] + + def construct_levels_change( + self, + persist: int, + red: int | None, + green: int | None, + blue: int | None, + warm_white: int | None, + cool_white: int | None, + write_mode: LevelWriteMode | int, + ) -> list[bytearray]: + """Construct a level change using a uniform 0xE1 0x22 fill command. + + Sending a solid color via the 0xE1 0x21 STATIC_FILL pattern puts the + device into a "Scene" state (preset_pattern=0x25, mode=0x66), so a + plain color-set reports back as the effect "Static Fill". The vendor + app instead sets a solid color with a UNIFORM 0xE1 0x22 command (all + 20 segments identical), which lands the device in preset_pattern=0x24 + ("Colorful" = a plain color) and reports effect=None with the correct + rgb/brightness. This is hardware-verified on the 0xB6 device. + + Inner message format (before wrapping): + pos 0 1 2 3 4 5 6 [ 5-byte segment ] x 20 + E1 22 00 00 00 00 14 [ H/2 S V 00 WW ] + | each of the 20 identical segments: + | color: [H/2, S, V, 0x00, 0x00] + | white: [0x00, 0x64, 0x00, 0x00, W] + segment count (0x14 = 20) + + The device is single-white RGB and the app sends EITHER a color OR a + white, never both. Warm and cool white are combined into a single + white level. + """ + # The device has a single white LED. The caller (_generate_levels_change) + # mirrors a single white value into BOTH warm and cool for non-CCT + # devices, so take the max (not the sum) to avoid double-counting it. + w = max(warm_white or 0, cool_white or 0) + + if w > 0 and not (red or green or blue): + # WHITE: scale 0-255 white to the device's 0-100 white level. + # S byte MUST be 0x64 (100); with S=0 the device silently ignores + # the frame (hardware-verified). + white_level = max(0, min(100, round(w * 100 / 255))) + segment = [0x00, 0x64, 0x00, 0x00, white_level] + else: + # COLOR: [H/2, S, V, 0x00, 0x00] + segment = self._rgb_to_hsv_bytes((red or 0), (green or 0), (blue or 0)) + + # Uniform E1 22 fill: header + segment repeated for all 20 segments + inner = bytearray([0xE1, 0x22, 0x00, 0x00, 0x00, 0x00, 0x14]) + for _ in range(20): + inner.extend(segment) + + return [ + self.construct_wrapped_message( + inner, inner_pre_constructed=True, version=0x02 + ) + ] + + def _rgb_to_hsv_bytes(self, r: int, g: int, b: int) -> list[int]: + """Convert RGB (0-255) to 5-byte HSV format [H/2, S, V, 0, 0]. + + This format is used by extended commands (0xE1 0x21, 0xE1 0x22). + + Equivalent to the RGBW variant with white=0, which yields the same + [H/2, S, V, 0x00, 0x00] layout, so delegate to avoid duplicating the + conversion logic. + """ + return self._rgb_to_hsv_bytes_rgbw(r, g, b, 0) + + def construct_extended_custom_effect( + self, + pattern_id: int, + colors: list[tuple[int, int, int]], + speed: int = 50, + density: int = 50, + direction: int = 0x01, + option: int = 0x00, + ) -> bytearray: + """Construct an extended custom effect command. + + Protocol format: + 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 ... + e1 21 01 64 PP OO DD NS SP 00 00 00 00 00 00 CC [H S V 00 00] x N + | | | | | | colors (5 bytes each) + | | | | | color count + | | | | speed (0-100) + | | | density (0-100) + | | direction (01=L->R, 02=R->L) + | option (00=default, 01=color change) + pattern_id (1-22 or 101-102) + + Args: + pattern_id: Pattern ID 1-22 or 101-102 + colors: List of 1-8 RGB color tuples (0-255 per channel) + speed: Animation speed 0-100 (default 50) + density: Pattern density 0-100 (default 50) + direction: 0x01=L->R, 0x02=R->L (default L->R) + option: Pattern-specific option (default 0) + + Returns: + Wrapped command bytearray + """ + # Clamp values to valid range + speed = max(0, min(100, speed)) + density = max(0, min(100, density)) + + # Convert Enum to value if needed + if hasattr(pattern_id, "value"): + pattern_id = pattern_id.value + if hasattr(option, "value"): + option = option.value + if hasattr(direction, "value"): + direction = direction.value + + # Build inner message + msg = bytearray( + [ + 0xE1, + 0x21, + 0x01, # Command type + 0x64, # Constant (100) + pattern_id, + option, + direction, + density, + speed, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, # Reserved (6 bytes) + len(colors), # Color count + ] + ) + + # Add colors as HSV (5 bytes each) + for r, g, b in colors: + msg.extend(self._rgb_to_hsv_bytes(r, g, b)) + + return self.construct_wrapped_message( + msg, inner_pre_constructed=True, version=0x02 + ) + + def construct_custom_segment_colors( + self, + segments: list[tuple[int, int, int] | None], + ) -> bytearray: + """Construct a custom segment colors command (0xE1 0x22). + + Sets static HSV colors for each of 20 segments on the light strip. + Used by devices like AK001-ZJ21413 (model 0xB6) under the "colorful" menu. + + Protocol format: + e1 22 00 00 00 00 14 [H/2 S V 00 00] x 20 + + Args: + segments: List of up to 20 segment colors. Each segment is either: + - None or (0, 0, 0) for off + - (R, G, B) tuple with values 0-255 + + Returns: + Wrapped command bytearray + """ + # Build inner message: header + 20 segments + msg = bytearray([0xE1, 0x22, 0x00, 0x00, 0x00, 0x00, 0x14]) + + for i in range(20): + segment = segments[i] if i < len(segments) else None + if segment and segment != (0, 0, 0): + msg.extend(self._rgb_to_hsv_bytes(*segment)) + else: + msg.extend([0x00, 0x00, 0x00, 0x00, 0x00]) # Off + + return self.construct_wrapped_message( + msg, inner_pre_constructed=True, version=0x02 + ) + + @staticmethod + def _scribble_bitmap(led_indices: Iterable[int], num_leds: int) -> bytearray: + """Build a ceil(num_leds/8)-byte MSB-first bitmap. + + LED i -> byte i//8, bit 7-(i%8). Bit set = 'apply this command to LED i'. + Trailing pad bits in the last byte are 0 (e.g. N=100 -> 13 bytes, last + byte 0xf0; N=80 -> 10 bytes, no pad). + """ + nbytes = (num_leds + 7) // 8 + bitmap = bytearray(nbytes) + for i in led_indices: + if not 0 <= i < num_leds: + raise ValueError(f"LED index {i} out of range for {num_leds} LEDs") + bitmap[i // 8] |= 0x80 >> (i % 8) + return bitmap + + def construct_scribble_init(self, num_leds: int) -> bytearray: + """Construct a scribble-mode init command (0xE1 0x23, non-rendering). + + Verified on hardware: E1 23 does NOT render -- it is only a + non-rendering buffer / scribble-mode-init (it flips state preset to + 0x66). Its per-LED color fields are vestigial for driving the display, + so this emits all-zero records. All colors render through grouped + E1 26 paints (see construct_scribble_paint). + + Inner byte layout: + e1 23 | 01 00 01 50 64 00 | N | + 0 1 2 3 4 5 6 7 8 9 ... + + Each 7-byte record is [H/2, S, V, pad, pad, WW, bright] = all-zero + color/white with brightness byte 0x64 (100). inner_len = 9 + 7*N. + """ + msg = bytearray([0xE1, 0x23, 0x01, 0x00, 0x01, 0x50, 0x64, 0x00, num_leds]) + msg.extend(bytes.fromhex("00000000000064") * num_leds) + return self.construct_wrapped_message( + msg, inner_pre_constructed=True, version=0x02 + ) + + def construct_scribble_paint( + self, + effect: int = 0x00, + direction: int = 0x01, + density: int = 0x50, + speed: int = 0x64, + blink_mode: int = 0x00, + h2: int = 0x00, + s: int = 0x00, + v: int = 0x00, + white: int = 0x00, + blink_speed: int = 0x64, + bitmap_leds: Iterable[int] | None = None, + num_leds: int = 0, + ) -> bytearray: + """Construct a scribble paint command (0xE1 0x26, the render path). + + Sets and displays per-LED color / white / blink / global-effect onto + the subset of LEDs selected by a bitmap. + + Inner byte layout: + e1 26 | EFFECT | DIR | DENSITY | SPEED | BLINK | H/2 | S | V | 00 | WLVL | BLINKSPD | + 0 1 2 3 4 5 6 7 8 9 10 11 12 13 ... + + 13-byte header then ceil(num_leds/8)-byte bitmap. If bitmap_leds is + None, all LEDs (0..num_leds-1) are selected. + + Color vs white are mutually exclusive (caller's responsibility): + color mode -> v set, white=0; white mode -> white set, v=0. + + All arguments are plain ints; enum->value conversion is done in the + generate/API layer where the type is a known enum (mypy-clean). + """ + density = max(0, min(100, density)) + speed = max(0, min(100, speed)) + blink_speed = max(0, min(100, blink_speed)) + s = max(0, min(100, s)) + v = max(0, min(100, v)) + white = max(0, min(100, white)) + h2 = max(0, min(180, h2)) + msg = bytearray( + [ + 0xE1, + 0x26, + effect, + direction, + density, + speed, + blink_mode, + h2, + s, + v, + 0x00, + white, + blink_speed, + ] + ) + indices = range(num_leds) if bitmap_leds is None else bitmap_leds + msg.extend(self._scribble_bitmap(indices, num_leds)) + return self.construct_wrapped_message( + msg, inner_pre_constructed=True, version=0x02 + ) + class ProtocolLEDENETAddressableBase(ProtocolLEDENET9Byte): """Base class for addressable protocols.""" diff --git a/tests/test_aio.py b/tests/test_aio.py index dd9ff006..454969fc 100644 --- a/tests/test_aio.py +++ b/tests/test_aio.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import colorsys import contextlib import datetime import json @@ -24,7 +25,13 @@ MAX_TEMP, MIN_TEMP, PUSH_UPDATE_INTERVAL, + ExtendedCustomEffectDirection, + ExtendedCustomEffectOption, + ExtendedCustomEffectPattern, MultiColorEffects, + ScribbleBlinkMode, + ScribbleEffect, + ScribbleLED, WhiteChannelType, ) from flux_led.protocol import ( @@ -4383,6 +4390,122 @@ def _updated_callback(*args, **kwargs): assert light.led_count == 100 +@pytest.mark.asyncio +async def test_0xB6_colorful_solid_red_full(mock_aio_protocol): + """0xB6 "Colorful" solid red at full brightness is a plain color, not an effect. + + Real device EA81 capture (app Colorful -> red): preset_pattern=0x24, mode=0x01 + -> reported as a color with no effect (verified on device). + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923240164f0b46464ff000500500000002002010003") + ) + await task + assert light.model_num == 0xB6 + assert light.effect is None + assert light.is_on is True + assert light.rgb == (255, 0, 0) + assert light.brightness == 255 + + +@pytest.mark.asyncio +async def test_0xB6_colorful_solid_red_dim(mock_aio_protocol): + """0xB6 "Colorful" solid red dimmed to 30% is a plain color, not an effect. + + Real device EA81 capture (app Colorful -> red @ 30%): preset_pattern=0x24, + mode=0x01, value byte 0x1e (30%). + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923240164f0b4641eff000500500000002002010003") + ) + await task + assert light.model_num == 0xB6 + assert light.effect is None + assert light.is_on is True + # Clearly dimmed (well below the full-brightness value of 255). + assert light.brightness == 76 + + +@pytest.mark.asyncio +async def test_0xB6_scene_wave(mock_aio_protocol): + """0xB6 "Scenes" animated effect reports preset_pattern=0x25, mode=effect id. + + Real device EA81 capture of the app's Scenes -> Wave (preset 0x25, mode 0x01). + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923250150f0b46464ff000500500000002002010003") + ) + await task + assert light.model_num == 0xB6 + assert light.effect == "Wave" + + +@pytest.mark.asyncio +async def test_0xB6_scene_static_fill(mock_aio_protocol): + """0xB6 "Scenes" Static Fill reports preset_pattern=0x25, mode=0x66. + + Real device EA81 capture of the app's Scenes -> Static Fill (preset 0x25, + mode 0x66, blue). + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923256632f0786464ff000500500000002002010003") + ) + await task + assert light.model_num == 0xB6 + assert light.effect == "Static Fill" + + +@pytest.mark.asyncio +async def test_0xB6_scene_unmapped_mode(mock_aio_protocol): + """An unmapped 0xB6 scene mode (preset 0x25) reports a distinguishable name. + + Preset 0x25 always means a scene is active, so an unmapped mode byte must + not report ``None`` (indistinguishable from no effect). + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + # Same frame as the Wave scene but with an unmapped mode byte (0x50). + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923255050f0b46464ff000500500000002002010003") + ) + await task + assert light.model_num == 0xB6 + assert light.effect == "unknown (0x50)" + + def test_protocol_extended_custom_state_response_length(): """ProtocolLEDENETExtendedCustom expects the 27-byte extended state.""" proto = ProtocolLEDENETExtendedCustom() @@ -4455,3 +4578,1736 @@ def test_extended_state_led_count(): # Too-short / invalid frame returns None assert proto.extended_state_led_count(b"\xea\x81\x01") is None assert proto.extended_state_led_count(b"\x00" * 27) is None + + +def test_extended_custom_effect_pattern_enum_values(): + """Test ExtendedCustomEffectPattern enum has expected values.""" + assert ExtendedCustomEffectPattern.WAVE.value == 0x01 + assert ExtendedCustomEffectPattern.METEOR.value == 0x02 + assert ExtendedCustomEffectPattern.STREAMER.value == 0x03 + assert ExtendedCustomEffectPattern.BUILDING_BLOCKS.value == 0x04 + assert ExtendedCustomEffectPattern.BREATHE.value == 0x09 + assert ExtendedCustomEffectPattern.STATIC_GRADIENT.value == 0x65 + assert ExtendedCustomEffectPattern.STATIC_FILL.value == 0x66 + + +def test_extended_custom_effect_direction_enum_values(): + """Test ExtendedCustomEffectDirection enum has expected values.""" + assert ExtendedCustomEffectDirection.LEFT_TO_RIGHT.value == 0x01 + assert ExtendedCustomEffectDirection.RIGHT_TO_LEFT.value == 0x02 + + +def test_extended_custom_effect_option_enum_values(): + """Test ExtendedCustomEffectOption enum has expected values.""" + assert ExtendedCustomEffectOption.DEFAULT.value == 0x00 + assert ExtendedCustomEffectOption.VARIANT_1.value == 0x01 + assert ExtendedCustomEffectOption.VARIANT_2.value == 0x02 + + +def test_construct_extended_custom_effect_single_color(): + """Test constructing an extended custom effect with single color.""" + proto = ProtocolLEDENETExtendedCustom() + + # Single red color, pattern Wave, default settings + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=[(255, 0, 0)], + speed=50, + density=50, + direction=0x01, + option=0x00, + ) + + # Result should be a wrapped message + assert isinstance(result, bytearray) + assert len(result) > 0 + + # Check the wrapper header (b0 b1 b2 b3) + assert result[0] == 0xB0 + assert result[1] == 0xB1 + assert result[2] == 0xB2 + assert result[3] == 0xB3 + + +def test_construct_extended_custom_effect_multiple_colors(): + """Test constructing an extended custom effect with multiple colors.""" + proto = ProtocolLEDENETExtendedCustom() + + # Three colors: red, green, blue + colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255)] + result = proto.construct_extended_custom_effect( + pattern_id=2, # Meteor + colors=colors, + speed=80, + density=100, + direction=0x02, # Right to Left + option=0x01, # Color change + ) + + assert isinstance(result, bytearray) + assert len(result) > 0 + + +def test_construct_extended_custom_effect_color_order(): + """Test that colors are stored in input order.""" + proto = ProtocolLEDENETExtendedCustom() + + # Two distinct colors + colors = [(255, 0, 0), (0, 0, 255)] # Red, Blue + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=colors, + speed=50, + density=50, + ) + + # The message structure after wrapper: + # Inner message starts after wrapper header + length bytes + # Find the color data section (after the 16-byte header) + # Colors are 5 bytes each (H, S, V, 0, 0) in input order + + # Red (255, 0, 0) in HSV: H=0, S=100, V=100 -> stored as (0, 100, 100) + # Blue (0, 0, 255) in HSV: H=240, S=100, V=100 -> stored as (120, 100, 100) + + # Red should come first in the message (same order as input) + assert isinstance(result, bytearray) + + +def test_construct_extended_custom_effect_hsv_conversion(): + """Test RGB to HSV conversion accuracy.""" + proto = ProtocolLEDENETExtendedCustom() + + # Test with a known color: pure green + colors = [(0, 255, 0)] + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=colors, + ) + + # Green in HSV: H=120, S=100%, V=100% + # Stored as: H/2=60, S=100, V=100 + # Verify the message was constructed + assert isinstance(result, bytearray) + assert len(result) > 0 + + +def test_construct_extended_custom_effect_speed_clamping(): + """Test that speed is clamped to 0-100.""" + proto = ProtocolLEDENETExtendedCustom() + + # Speed > 100 should be clamped + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=[(255, 0, 0)], + speed=150, + ) + assert isinstance(result, bytearray) + + # Speed < 0 should be clamped + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=[(255, 0, 0)], + speed=-10, + ) + assert isinstance(result, bytearray) + + +def test_construct_extended_custom_effect_density_clamping(): + """Test that density is clamped to 0-100.""" + proto = ProtocolLEDENETExtendedCustom() + + # Density > 100 should be clamped + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=[(255, 0, 0)], + density=200, + ) + assert isinstance(result, bytearray) + + +def test_construct_extended_custom_effect_max_colors(): + """Test constructing an effect with maximum 8 colors.""" + proto = ProtocolLEDENETExtendedCustom() + + # 8 colors (maximum) + colors = [ + (255, 0, 0), + (255, 128, 0), + (255, 255, 0), + (0, 255, 0), + (0, 255, 255), + (0, 0, 255), + (128, 0, 255), + (255, 0, 255), + ] + result = proto.construct_extended_custom_effect( + pattern_id=1, + colors=colors, + ) + + assert isinstance(result, bytearray) + # Each color is 5 bytes, 8 colors = 40 bytes for colors + # Plus 16 bytes header = 56 bytes inner message + # Plus wrapper overhead + + +def test_construct_extended_custom_effect_with_enums(): + """Test using enum values for parameters.""" + proto = ProtocolLEDENETExtendedCustom() + + result = proto.construct_extended_custom_effect( + pattern_id=ExtendedCustomEffectPattern.WAVE, + colors=[(255, 0, 0)], + speed=50, + density=50, + direction=ExtendedCustomEffectDirection.RIGHT_TO_LEFT, + option=ExtendedCustomEffectOption.VARIANT_1, + ) + + assert isinstance(result, bytearray) + + +def test_construct_extended_custom_effect_with_variant_2(): + """Test using VARIANT_2 option (e.g., breathe mode for rainbow patterns).""" + proto = ProtocolLEDENETExtendedCustom() + + # Rainbow colors with VARIANT_2 option + colors = [ + (255, 0, 0), # Red + (255, 255, 0), # Yellow + (0, 255, 0), # Green + (0, 255, 255), # Cyan + (0, 0, 255), # Blue + (255, 0, 255), # Magenta + ] + result = proto.construct_extended_custom_effect( + pattern_id=12, # Twinkling stars + colors=colors, + speed=60, + density=100, + direction=ExtendedCustomEffectDirection.LEFT_TO_RIGHT, + option=ExtendedCustomEffectOption.VARIANT_2, + ) + + assert isinstance(result, bytearray) + assert len(result) > 0 + + +def test_extended_custom_effect_hsv_values(): + """Test specific HSV value calculations.""" + # Test the HSV conversion formula used in the protocol + # RGB (255, 0, 0) -> HSV (0, 100%, 100%) -> stored as (0, 100, 100) + r, g, b = 255, 0, 0 + h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) + hsv_h = int(h * 180) + hsv_s = int(s * 100) + hsv_v = int(v * 100) + + assert hsv_h == 0 # Red hue + assert hsv_s == 100 # Full saturation + assert hsv_v == 100 # Full value + + # RGB (0, 0, 255) -> HSV (240, 100%, 100%) -> stored as (120, 100, 100) + r, g, b = 0, 0, 255 + h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) + hsv_h = int(h * 180) + hsv_s = int(s * 100) + hsv_v = int(v * 100) + + assert hsv_h == 120 # Blue hue (240/2) + assert hsv_s == 100 + assert hsv_v == 100 + + # RGB (0, 255, 0) -> HSV (120, 100%, 100%) -> stored as (60, 100, 100) + r, g, b = 0, 255, 0 + h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) + hsv_h = int(h * 180) + hsv_s = int(s * 100) + hsv_v = int(v * 100) + + assert hsv_h == 60 # Green hue (120/2) + assert hsv_s == 100 + assert hsv_v == 100 + + +# Tests for _generate_extended_custom_effect validation (base_device.py lines 1335-1360) + + +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_validation(mock_aio_protocol): + """Test validation in _generate_extended_custom_effect.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # Setup 0xB6 device with extended state + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + assert light.protocol == PROTOCOL_LEDENET_EXTENDED_CUSTOM + + # Test invalid pattern_id (0 is not valid) + with pytest.raises(ValueError, match="Pattern ID must be 1-22 or 101-102"): + light._generate_extended_custom_effect(0, [(255, 0, 0)]) + + # Test invalid pattern_id (23 is not valid -- max animated id is 22) + with pytest.raises(ValueError, match="Pattern ID must be 1-22 or 101-102"): + light._generate_extended_custom_effect(23, [(255, 0, 0)]) + + # Test invalid pattern_id (24 is not valid) + with pytest.raises(ValueError, match="Pattern ID must be 1-22 or 101-102"): + light._generate_extended_custom_effect(24, [(255, 0, 0)]) + + # Test invalid pattern_id (25 is not valid) + with pytest.raises(ValueError, match="Pattern ID must be 1-22 or 101-102"): + light._generate_extended_custom_effect(25, [(255, 0, 0)]) + + # Test empty colors list + with pytest.raises(ValueError, match="at least one color"): + light._generate_extended_custom_effect(1, []) + + # Test invalid color tuple (not 3 elements) + with pytest.raises(ValueError, match=r"must be .* tuple"): + light._generate_extended_custom_effect(1, [(255, 0)]) + + # Test color values out of range (> 255) + with pytest.raises(ValueError, match="must be 0-255"): + light._generate_extended_custom_effect(1, [(256, 0, 0)]) + + # Test color values out of range (< 0) + with pytest.raises(ValueError, match="must be 0-255"): + light._generate_extended_custom_effect(1, [(-1, 0, 0)]) + + # Test valid pattern_id 22 (max animated id) + result = light._generate_extended_custom_effect(22, [(255, 0, 0)]) + assert isinstance(result, bytearray) + + # Test valid pattern_id 101 (STATIC_GRADIENT) + result = light._generate_extended_custom_effect(101, [(255, 0, 0)]) + assert isinstance(result, bytearray) + + # Test valid pattern_id 102 (STATIC_FILL) + result = light._generate_extended_custom_effect(102, [(255, 0, 0)]) + assert isinstance(result, bytearray) + + +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_rejects_too_many_colors( + mock_aio_protocol, +): + """Test that too many colors (>8) raise ValueError instead of truncating.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + + # 9 colors (one over the 8 max) must raise; 8 is still accepted. + colors = [(i * 25, 0, 0) for i in range(9)] + with pytest.raises(ValueError, match="at most 8 colors are supported, got 9"): + light._generate_extended_custom_effect(1, colors) + + result = light._generate_extended_custom_effect(1, colors[:8]) + assert isinstance(result, bytearray) + + +@pytest.mark.parametrize( + "kwargs, match", + [ + ({"speed": 101}, "speed must be 0-100"), + ({"speed": -1}, "speed must be 0-100"), + ({"density": 101}, "density must be 0-100"), + ({"density": -1}, "density must be 0-100"), + ({"direction": 0x00}, "direction must be 0x01 or 0x02"), + ({"direction": 0x03}, "direction must be 0x01 or 0x02"), + ({"option": 3}, "option must be 0-2"), + ], +) +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_param_bounds( + mock_aio_protocol, kwargs, match +): + """General wire-byte bounds for speed/density/direction/option are enforced.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + with pytest.raises(ValueError, match=match): + light._generate_extended_custom_effect(1, [(255, 0, 0)], **kwargs) + + +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_param_bounds_valid(mock_aio_protocol): + """Boundary-valid animation params still succeed.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + result = light._generate_extended_custom_effect( + 1, [(255, 0, 0)], speed=100, density=0, direction=0x02, option=2 + ) + assert isinstance(result, bytearray) + + +@pytest.mark.asyncio +async def test_async_set_extended_custom_effect_accepts_enums(mock_aio_protocol): + """async_set_extended_custom_effect accepts the public enums, not just ints.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + # Passing enum members must not raise and must send a wrapped message. + await light.async_set_extended_custom_effect( + pattern_id=ExtendedCustomEffectPattern.WAVE, + colors=[(255, 0, 0), (0, 255, 0)], + speed=50, + density=50, + direction=ExtendedCustomEffectDirection.LEFT_TO_RIGHT, + option=ExtendedCustomEffectOption.VARIANT_1, + ) + assert len(sent) == 1 + assert sent[0][0] == 0xB0 + assert sent[0][1] == 0xB1 + + # The produced inner message must equal the equivalent all-int call. + # (The B0B1 wrapper carries an incrementing counter, so compare the + # unwrapped inner E1 21 payload.) + int_bytes = light._generate_extended_custom_effect( + ExtendedCustomEffectPattern.WAVE.value, + [(255, 0, 0), (0, 255, 0)], + 50, + 50, + ExtendedCustomEffectDirection.LEFT_TO_RIGHT.value, + ExtendedCustomEffectOption.VARIANT_1.value, + ) + enum_bytes = light._generate_extended_custom_effect( + ExtendedCustomEffectPattern.WAVE, + [(255, 0, 0), (0, 255, 0)], + 50, + 50, + ExtendedCustomEffectDirection.LEFT_TO_RIGHT, + ExtendedCustomEffectOption.VARIANT_1, + ) + assert _inner_of(enum_bytes) == _inner_of(int_bytes) + + +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_rejects_wrong_enum(mock_aio_protocol): + """A mismatched enum type must not be silently unwrapped into an int. + + ScribbleEffect.ACCUMULATE == 0x08 would coerce into an in-range pattern id + if any Enum were accepted. Coercion is narrowed to the expected enum type, + so the wrong enum falls through unchanged and fails validation. + """ + light, _t = await _setup_scribble_light(mock_aio_protocol) + + with pytest.raises(ValueError): + light._generate_extended_custom_effect( + pattern_id=ScribbleEffect.ACCUMULATE, colors=[(255, 0, 0)] + ) + + with pytest.raises(ValueError): + await light.async_set_extended_custom_effect( + pattern_id=ScribbleEffect.ACCUMULATE, colors=[(255, 0, 0)] + ) + + # The correct enum still works. + result = light._generate_extended_custom_effect( + pattern_id=ExtendedCustomEffectPattern.WAVE, colors=[(255, 0, 0)] + ) + assert result[0] == 0xB0 + + +# Tests for _generate_custom_segment_colors validation (base_device.py lines 1376-1392) + + +@pytest.mark.asyncio +async def test_generate_custom_segment_colors_validation(mock_aio_protocol): + """Test validation in _generate_custom_segment_colors.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + # Test invalid color tuple (not 3 elements) + with pytest.raises(ValueError, match=r"must be .* tuple"): + light._generate_custom_segment_colors([(255, 0)]) + + # Test color values out of range (> 255) + with pytest.raises(ValueError, match="must be 0-255"): + light._generate_custom_segment_colors([(256, 0, 0)]) + + # Test valid segments with None + result = light._generate_custom_segment_colors([None, (255, 0, 0), None]) + assert isinstance(result, bytearray) + + +@pytest.mark.asyncio +async def test_generate_custom_segment_colors_rejects_too_many_segments( + mock_aio_protocol, +): + """Test that too many segments (>20) raise ValueError instead of truncating.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + + # 21 segments (one over the 20 max) must raise; 20 is still accepted. + segments = [(i * 10, 0, 0) for i in range(21)] + with pytest.raises(ValueError, match="at most 20 segments are supported, got 21"): + light._generate_custom_segment_colors(segments) + + result = light._generate_custom_segment_colors(segments[:20]) + assert isinstance(result, bytearray) + + +# Tests for construct_levels_change (protocol.py lines 1826-1861) + + +def test_protocol_construct_levels_change_0xB6(): + """Test construct_levels_change sets a color via a uniform E1 22 fill. + + A solid color must land the device in preset 0x24 ("Colorful"), so it is + sent as a uniform E1 22 (all 20 segments identical), not an E1 21 Static + Fill (hardware-verified). + """ + proto = ProtocolLEDENETExtendedCustom() + + # Test with RGB values (pure red) + result = proto.construct_levels_change( + persist=1, + red=255, + green=0, + blue=0, + warm_white=0, + cool_white=0, + write_mode=0, + ) + + assert len(result) == 1 + msg = result[0] + assert isinstance(msg, bytearray) + # Check wrapper header + assert msg[0] == 0xB0 + assert msg[1] == 0xB1 + assert msg[2] == 0xB2 + assert msg[3] == 0xB3 + + # Pin the full inner E1 22 uniform frame. This is byte-identical to the real + # app "Colorful -> red" packet captured from the device (inner: + # e1 22 00 00 00 00 14 [00 64 64 00 00] x 20). + inner = _inner_of(msg) + header = _h("e1 22 00 00 00 00 14") + # (255,0,0): hue 0, sat 100, val 100 -> [00 64 64 00 00] + red_seg = _h("00 64 64 00 00") + expected = header + red_seg * 20 + assert inner == expected + assert len(inner) == 7 + 20 * 5 + + +def test_protocol_construct_levels_change_with_white(): + """Test construct_levels_change for a white-only set. + + A white-only set is sent as a uniform E1 22 with the literal white segment + [00 64 00 00 W], where W is the 0-100 white level (S byte MUST be 0x64). + """ + proto = ProtocolLEDENETExtendedCustom() + + # Test with white values: warm_white=255 -> W = round(255*100/255) = 100 + result = proto.construct_levels_change( + persist=1, + red=0, + green=0, + blue=0, + warm_white=255, + cool_white=0, + write_mode=0, + ) + + assert len(result) == 1 + msg = result[0] + assert isinstance(msg, bytearray) + + inner = _inner_of(msg) + header = _h("e1 22 00 00 00 00 14") + # W = 100 = 0x64; segment [00 64 00 00 64] + white_seg = _h("00 64 00 00 64") + expected = header + white_seg * 20 + assert inner == expected + assert len(inner) == 7 + 20 * 5 + + +def test_protocol_construct_levels_change_white_not_doubled(): + """A single white value mirrored into warm+cool must not be double-counted. + + The device has one white LED; _generate_levels_change mirrors a single white + value into BOTH warm and cool for non-CCT devices, so combining them by max + (not sum) is required. Regression test: warm=cool=128 (a mirrored single + white of 128) must yield W = round(128*100/255) = 50, not 100. + """ + proto = ProtocolLEDENETExtendedCustom() + + result = proto.construct_levels_change( + persist=1, + red=0, + green=0, + blue=0, + warm_white=128, + cool_white=128, # mirrored single white; max -> 128 -> W=50 (not 2x) + write_mode=0, + ) + + assert len(result) == 1 + assert isinstance(result[0], bytearray) + + inner = _inner_of(result[0]) + header = _h("e1 22 00 00 00 00 14") + white_seg = _h("00 64 00 00 32") # W = 50 = 0x32 (not doubled to 0x64) + expected = header + white_seg * 20 + assert inner == expected + + +def test_protocol_rgb_to_hsv_bytes_rgbw(): + """Test _rgb_to_hsv_bytes_rgbw conversion.""" + proto = ProtocolLEDENETExtendedCustom() + + # Test pure red with white + result = proto._rgb_to_hsv_bytes_rgbw(255, 0, 0, 100) + assert len(result) == 5 + assert result[0] == 0 # Hue (red = 0) + assert result[1] == 100 # Saturation + assert result[2] == 100 # Value + assert result[3] == 0x00 # Unused + assert result[4] == 100 # White + + # Test pure green + result = proto._rgb_to_hsv_bytes_rgbw(0, 255, 0, 0) + assert result[0] == 60 # Hue (green = 120/2) + + # Test pure blue + result = proto._rgb_to_hsv_bytes_rgbw(0, 0, 255, 255) + assert result[0] == 120 # Hue (blue = 240/2) + assert result[4] == 255 # White + + +# Tests for construct_custom_segment_colors (protocol.py lines 1971-1980) + + +def test_protocol_construct_custom_segment_colors(): + """Test construct_custom_segment_colors command format.""" + proto = ProtocolLEDENETExtendedCustom() + + # Test with a few segments + segments = [(255, 0, 0), None, (0, 0, 255)] + result = proto.construct_custom_segment_colors(segments) + + assert isinstance(result, bytearray) + # Check wrapper header + assert result[0] == 0xB0 + assert result[1] == 0xB1 + assert result[2] == 0xB2 + assert result[3] == 0xB3 + + # Pin the full inner E1 22 payload: header + 0x14 (20) count byte, then + # one 5-byte [H/2, S, V, 0x00, 0x00] record per segment, padded to 20. + inner = _inner_of(result) + header = _h("e1 22 00 00 00 00 14") + red = _h("00 64 64 00 00") # (255,0,0): hue 0, sat 100, val 100 + off = _h("00 00 00 00 00") # None / off segment + blue = _h("78 64 64 00 00") # (0,0,255): hue 120 -> byte 0x78, sat/val 100 + # 3 provided segments (red, off, blue) + 17 off segments = 20 total. + expected = header + red + off + blue + off * 17 + assert inner == expected + assert len(inner) == 7 + 20 * 5 + + +def test_protocol_construct_custom_segment_colors_all_off(): + """Test construct_custom_segment_colors with all segments off.""" + proto = ProtocolLEDENETExtendedCustom() + + # All None segments + segments = [None] * 10 + result = proto.construct_custom_segment_colors(segments) + + assert isinstance(result, bytearray) + + +def test_protocol_construct_custom_segment_colors_zero_tuple(): + """Test that (0,0,0) is treated as off.""" + proto = ProtocolLEDENETExtendedCustom() + + # Mix of None and (0,0,0) + segments = [None, (0, 0, 0), (255, 0, 0)] + result = proto.construct_custom_segment_colors(segments) + + assert isinstance(result, bytearray) + + +# Tests for async API methods + + +@pytest.mark.asyncio +async def test_async_set_extended_custom_effect_0xB6(mock_aio_protocol): + """Test async_set_extended_custom_effect sends correct bytes.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + transport, _protocol = await mock_aio_protocol() + + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + transport.reset_mock() + + await light.async_set_extended_custom_effect( + pattern_id=1, + colors=[(255, 0, 0), (0, 255, 0)], + speed=50, + density=50, + ) + + assert transport.write.called + written_data = transport.write.call_args[0][0] + # Verify it's a wrapped message + assert written_data[0] == 0xB0 + assert written_data[1] == 0xB1 + + +@pytest.mark.asyncio +async def test_async_set_custom_segment_colors_0xB6(mock_aio_protocol): + """Test async_set_custom_segment_colors sends correct bytes.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + transport, _protocol = await mock_aio_protocol() + + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + transport.reset_mock() + + await light.async_set_custom_segment_colors( + segments=[(255, 0, 0), None, (0, 0, 255)] + ) + + assert transport.write.called + written_data = transport.write.call_args[0][0] + # Verify it's a wrapped message + assert written_data[0] == 0xB0 + assert written_data[1] == 0xB1 + + +# Scribble (per-LED) feature tests (0xB6) + + +def _inner_of(wrapped: bytearray) -> bytes: + """Extract the inner message from a B0B1B2B3-wrapped result. + + wrapper = b0 b1 b2 b3 | 00 01 | ver | counter | len_hi len_lo | inner | cksum + """ + inner_len = (wrapped[8] << 8) | wrapped[9] + return bytes(wrapped[10 : 10 + inner_len]) + + +ALL_ON_100 = bytes.fromhex("ff" * 12 + "f0") # N=100, 13 bytes +ALL_ON_80 = bytes.fromhex("ff" * 10) # N=80, 10 bytes +OFF0_80 = bytes.fromhex("80" + "00" * 9) # N=80, only LED 0 +GROUP_40_79_80 = bytes.fromhex("00" * 5 + "ff" * 5) # N=80, LEDs 40-79 + + +def _h(s: str) -> bytes: + return bytes.fromhex(s.replace(" ", "")) + + +def test_scribble_bitmap_n100(): + proto = ProtocolLEDENETExtendedCustom() + bitmap = proto._scribble_bitmap(range(100), 100) + assert len(bitmap) == 13 + assert bitmap == ALL_ON_100 + assert bitmap[-1] == 0xF0 + + +def test_scribble_bitmap_n80(): + proto = ProtocolLEDENETExtendedCustom() + bitmap = proto._scribble_bitmap(range(80), 80) + assert len(bitmap) == 10 + assert bitmap == ALL_ON_80 + + +def test_scribble_bitmap_led0(): + proto = ProtocolLEDENETExtendedCustom() + bitmap = proto._scribble_bitmap([0], 80) + assert bitmap[0] == 0x80 + + +def test_scribble_bitmap_led7(): + proto = ProtocolLEDENETExtendedCustom() + bitmap = proto._scribble_bitmap([7], 80) + assert bitmap[0] == 0x01 + + +def test_scribble_bitmap_out_of_range(): + proto = ProtocolLEDENETExtendedCustom() + with pytest.raises(ValueError): + proto._scribble_bitmap([80], 80) + with pytest.raises(ValueError): + proto._scribble_bitmap([-1], 80) + + +def test_scribble_paint_all_green_static_n100(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + effect=0x00, + direction=0x01, + density=0x50, + speed=0x64, + blink_mode=0x00, + h2=0x3C, + s=0x64, + v=0x64, + white=0x00, + blink_speed=0x64, + num_leds=100, + ) + ) + assert inner[:2] == b"\xe1\x26" + assert inner == _h("e1 26 00 01 50 64 00 3c 64 64 00 00 64") + ALL_ON_100 + + +def test_scribble_paint_blue_50pct(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint(h2=0x78, s=0x64, v=0x32, num_leds=80) + ) + assert inner == _h("e1 26 00 01 50 64 00 78 64 32 00 00 64") + ALL_ON_80 + + +def test_scribble_paint_white_100(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint(h2=0x78, s=0x64, v=0x00, white=0x64, num_leds=80) + ) + assert inner == _h("e1 26 00 01 50 64 00 78 64 00 00 64 64") + ALL_ON_80 + + +def test_scribble_paint_fast_blink_50(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + blink_mode=0x10, h2=0x78, s=0x64, v=0x64, blink_speed=0x32, num_leds=80 + ) + ) + assert inner == _h("e1 26 00 01 50 64 10 78 64 64 00 00 32") + ALL_ON_80 + + +def test_scribble_paint_slow_blink_50(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + blink_mode=0x08, h2=0x78, s=0x64, v=0x64, blink_speed=0x32, num_leds=80 + ) + ) + assert inner == _h("e1 26 00 01 50 64 08 78 64 64 00 00 32") + ALL_ON_80 + + +def test_scribble_paint_flowing_r2l_speed50(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + effect=0x01, + direction=0x02, + speed=0x32, + h2=0x78, + s=0x64, + v=0x64, + num_leds=80, + ) + ) + assert inner == _h("e1 26 01 02 50 32 00 78 64 64 00 00 64") + ALL_ON_80 + + +def test_scribble_paint_twinkling_d50_s61(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + effect=0x03, + direction=0x02, + density=0x32, + speed=0x3D, + h2=0x78, + s=0x64, + v=0x64, + num_leds=80, + ) + ) + assert inner == _h("e1 26 03 02 32 3d 00 78 64 64 00 00 64") + ALL_ON_80 + + +def test_scribble_paint_off_bulb0_n80(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + effect=0x00, + density=0x50, + h2=0x00, + s=0x00, + v=0x00, + blink_speed=0x64, + bitmap_leds=[0], + num_leds=80, + ) + ) + assert inner == _h("e1 26 00 01 50 64 00 00 00 00 00 00 64") + OFF0_80 + + +def test_scribble_paint_flowing_blue_group_n80(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of( + proto.construct_scribble_paint( + effect=0x01, + direction=0x02, + density=0x50, + speed=0x26, + h2=0x78, + s=0x64, + v=0x64, + blink_speed=0x64, + bitmap_leds=list(range(40, 80)), + num_leds=80, + ) + ) + assert inner == _h("e1 26 01 02 50 26 00 78 64 64 00 00 64") + GROUP_40_79_80 + + +def test_scribble_init_n80(): + proto = ProtocolLEDENETExtendedCustom() + inner = _inner_of(proto.construct_scribble_init(80)) + assert inner[:9] == _h("e1 23 01 00 01 50 64 00 50") + assert inner[9:] == _h("00 00 00 00 00 00 64") * 80 + assert len(inner) == 569 # 9 + 7*80 + + +@pytest.mark.asyncio +async def test_generate_scribble_init_invalid(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + with pytest.raises(ValueError, match=r"num_leds must be 1\.\.255"): + light._generate_scribble_init(0) + # 256 exceeds the one-byte E1 23 count / bitmap addressing limit and must + # raise the descriptive error, not the generic "byte must be in range". + with pytest.raises(ValueError, match=r"num_leds must be 1\.\.255, got 256"): + light._generate_scribble_init(256) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_num_leds_over_255_raises(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + light._extended_led_count = None # bypass the led_count mismatch check + leds = [ScribbleLED(rgb=(255, 0, 0))] * 256 + with pytest.raises(ValueError, match=r"num_leds must be 1\.\.255, got 256"): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 256) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_rgb_and_white_raises(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(255, 0, 0), white=50)] + [ScribbleLED()] * 79 + with pytest.raises(ValueError): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_channel_out_of_range_raises(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(300, 0, 0))] + [ScribbleLED()] * 79 + with pytest.raises(ValueError): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_effect_int_and_enum(mock_aio_protocol): + """effect accepts a raw int id (e.g. 4, unnamed) or a ScribbleEffect; the + device-valid range is 0x00-0x08, anything else raises ValueError.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(255, 0, 0))] * 80 + # raw int id not exposed as a name (3,4,6,7 are valid on-device) + assert ( + _inner_of(light._scribble_paint_groups(leds, 4, 0x01, 0x50, 0x64, 80)[0])[2] + == 4 + ) + # enum still works + assert ( + _inner_of( + light._scribble_paint_groups( + leds, ScribbleEffect.FLOWING, 0x01, 0x50, 0x64, 80 + )[0] + )[2] + == 0x01 + ) + # out-of-range id (device accepts only 0x00-0x08) + with pytest.raises(ValueError): + light._scribble_paint_groups(leds, 9, 0x01, 0x50, 0x64, 80) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_wrong_count_raises(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + assert light.led_count == 80 + leds = [ScribbleLED(rgb=(255, 0, 0))] * 40 + with pytest.raises(ValueError): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 40) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_empty_raises(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + with pytest.raises(ValueError): + light._scribble_paint_groups([], 0x00, 0x01, 0x50, 0x64, 0) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_two_color_static(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(255, 0, 0))] * 40 + [ScribbleLED(rgb=(0, 0, 255))] * 40 + msgs = light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + assert len(msgs) == 2 # red group then blue group, first-appearance order + red = _inner_of(msgs[0]) + blue = _inner_of(msgs[1]) + # red occupies LEDs 0-39, blue 40-79 + assert red[13:] == _h("ff" * 5 + "00" * 5) + assert blue[13:] == GROUP_40_79_80 + # blue hue byte (h2) = 0x78 + assert blue[7] == 0x78 + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_blink_grouping(mock_aio_protocol): + """LEDs with different blink modes split into separate E1 26 groups.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(255, 0, 0), blink_mode=ScribbleBlinkMode.FAST)] * 40 + [ + ScribbleLED(rgb=(255, 0, 0), blink_mode=ScribbleBlinkMode.NONE) + ] * 40 + msgs = light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + assert len(msgs) == 2 # same color, different blink -> two groups + fast = _inner_of(msgs[0]) + steady = _inner_of(msgs[1]) + assert fast[6] == 0x10 # FAST blink byte + assert steady[6] == 0x00 # NONE blink byte + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_non_blinking_ignore_blink_speed( + mock_aio_protocol, +): + """Non-blinking LEDs with the same color coalesce regardless of blink_speed. + + blink_speed is only meaningful while blinking; with blink off it must not + split otherwise-identical LEDs into redundant E1 26 paint groups. + """ + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ + ScribbleLED(rgb=(255, 0, 0), blink_mode=ScribbleBlinkMode.NONE, blink_speed=50) + ] * 40 + [ + ScribbleLED(rgb=(255, 0, 0), blink_mode=ScribbleBlinkMode.NONE, blink_speed=100) + ] * 40 + msgs = light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + assert len(msgs) == 1 # same color, blink off -> single group + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_off_group_color_zero(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(255, 0, 0))] + [ScribbleLED()] * 79 + msgs = light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + assert len(msgs) == 2 + off = _inner_of(msgs[1]) + # off group: all-zero color, header matches the off golden + assert off[:13] == _h("e1 26 00 01 50 64 00 00 00 00 00 00 64") + assert off[13:] == _h("7f" + "ff" * 8 + "ff") # LEDs 1-79 set + + +async def _setup_scribble_light(mock_aio_protocol, led_count_byte=0x50): + """Set up a 0xB6 AIO light with a known led_count from state byte 18.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + transport, _protocol = await mock_aio_protocol() + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + led_count_byte, + 0x00, + 0x83, + ) + ) + ) + await task + transport.reset_mock() + return light, transport + + +@pytest.mark.asyncio +async def test_async_extended_custom_methods_raise_on_unsupported_device( + mock_aio_protocol, +): + """Async extended-custom methods raise ValueError on a non-0xB6 device.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + _transport, _protocol = await mock_aio_protocol() + # Standard RGBCW bulb (0x35) -- does not use the extended custom protocol. + light._aio_protocol.data_received( + b"\x81\x35\x23\x61\x05\x10\xb6\x00\x98\x19\x04\x25\x0f\xee" + ) + await task + assert light.model_num == 0x35 + assert not light.supports_extended_custom_effects + assert not light.supports_scribble + + with pytest.raises(ValueError): + await light.async_set_extended_custom_effect(1, [(255, 0, 0)]) + with pytest.raises(ValueError): + await light.async_set_custom_segment_colors([(255, 0, 0)]) + with pytest.raises(ValueError): + await light.async_set_scribble([ScribbleLED(rgb=(255, 0, 0))]) + + +@pytest.mark.asyncio +async def test_async_set_scribble_static_two_color(mock_aio_protocol): + """STATIC 2-color config sends E1 23 init + one E1 26 per color group.""" + light, _transport = await _setup_scribble_light(mock_aio_protocol) + assert light.led_count == 80 + + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + leds = [ScribbleLED(rgb=(255, 0, 0))] * 40 + [ScribbleLED(rgb=(0, 0, 255))] * 40 + await light.async_set_scribble(leds, effect=ScribbleEffect.STATIC) + + assert len(sent) == 3 + init = _inner_of(sent[0]) + assert init[:2] == b"\xe1\x23" + assert len(init) == 569 + red = _inner_of(sent[1]) + blue = _inner_of(sent[2]) + assert red[:2] == b"\xe1\x26" + assert red[13:] == _h("ff" * 5 + "00" * 5) + assert blue[13:] == GROUP_40_79_80 + assert blue[7] == 0x78 # blue hue + + +@pytest.mark.asyncio +async def test_async_set_scribble_no_enter_mode(mock_aio_protocol): + """enter_mode=False sends no E1 23, one E1 26 per group.""" + light, _transport = await _setup_scribble_light(mock_aio_protocol) + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + leds = [ScribbleLED(rgb=(255, 0, 0))] * 40 + [ScribbleLED(rgb=(0, 0, 255))] * 40 + await light.async_set_scribble(leds, enter_mode=False) + assert len(sent) == 2 + assert _inner_of(sent[0])[:2] == b"\xe1\x26" + assert _inner_of(sent[1])[:2] == b"\xe1\x26" + + +@pytest.mark.asyncio +async def test_async_set_scribble_invalid_input_sends_nothing(mock_aio_protocol): + """Invalid scribble input raises before any send (including the E1 23 init). + + An invalid call must not leave the device in scribble-init mode with nothing + rendered, so validation happens before the init is sent. + """ + light, _transport = await _setup_scribble_light(mock_aio_protocol) + assert light.led_count == 80 + + sent = [] + # 3 LEDs on an 80-LED device is a length mismatch -> ValueError. + with ( + patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)), + pytest.raises(ValueError), + ): + await light.async_set_scribble( + [ScribbleLED(rgb=(255, 0, 0))] * 3, enter_mode=True + ) + # Nothing must have been sent -- not even the E1 23 init. + assert sent == [] + + +@pytest.mark.asyncio +async def test_async_set_scribble_flowing_blue_group(mock_aio_protocol): + """FLOWING effect: blue group paint matches the captured golden.""" + light, _transport = await _setup_scribble_light(mock_aio_protocol) + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + leds = [ScribbleLED(rgb=(255, 0, 0))] * 40 + [ScribbleLED(rgb=(0, 0, 255))] * 40 + await light.async_set_scribble( + leds, + effect=ScribbleEffect.FLOWING, + direction=ExtendedCustomEffectDirection.RIGHT_TO_LEFT, + density=0x50, + speed=0x26, + ) + assert len(sent) == 3 + red = _inner_of(sent[1]) + blue = _inner_of(sent[2]) + # blue group matches the captured golden + assert blue == _h("e1 26 01 02 50 26 00 78 64 64 00 00 64") + GROUP_40_79_80 + # red group carries red color on LEDs 0-39 under the same effect + assert red[:7] == _h("e1 26 01 02 50 26 00") + assert red[13:] == _h("ff" * 5 + "00" * 5) + + +@pytest.mark.asyncio +async def test_async_set_scribble_off_group(mock_aio_protocol): + """Mixed lit + off config: off group paints all-zero color.""" + light, _transport = await _setup_scribble_light(mock_aio_protocol) + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + leds = [ScribbleLED(rgb=(255, 0, 0))] + [ScribbleLED()] * 79 + await light.async_set_scribble(leds, enter_mode=False) + assert len(sent) == 2 + red = _inner_of(sent[0]) + off = _inner_of(sent[1]) + assert red[13:] == OFF0_80 + assert off[:13] == _h("e1 26 00 01 50 64 00 00 00 00 00 00 64") + + +@pytest.mark.asyncio +async def test_async_set_scribble_raw_int_direction(mock_aio_protocol): + """A raw int direction (e.g. 0x02) is accepted, not just the enum.""" + light, _transport = await _setup_scribble_light(mock_aio_protocol) + sent = [] + with patch.object(light, "_async_send_msg", side_effect=lambda m: sent.append(m)): + leds = [ScribbleLED(rgb=(255, 0, 0))] * 80 + # Passing a raw int must not raise AttributeError on direction.value. + await light.async_set_scribble(leds, direction=0x02, enter_mode=False) + assert len(sent) == 1 + paint = _inner_of(sent[0]) + assert paint[:2] == b"\xe1\x26" + assert paint[3] == 0x02 # direction byte carried through as the raw int + + +@pytest.mark.parametrize( + "leds, num_leds, match", + [ + # empty leds + ([], 80, "leds must not be empty"), + # num_leds out of 1..255 (0) + ([ScribbleLED(rgb=(1, 2, 3))], 0, r"num_leds must be 1\.\.255"), + # num_leds out of 1..255 (256) + ([ScribbleLED(rgb=(1, 2, 3))] * 256, 256, r"num_leds must be 1\.\.255"), + # num_leds-vs-len mismatch (num_leds != len(leds)) + ([ScribbleLED(rgb=(1, 2, 3))] * 80, 79, r"does not match"), + ], +) +@pytest.mark.asyncio +async def test_scribble_paint_groups_count_guards( + mock_aio_protocol, leds, num_leds, match +): + """Cover the count/length ValueError guards in _scribble_paint_groups.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + light._extended_led_count = None # bypass led_count-vs-len check for these cases + with pytest.raises(ValueError, match=match): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, num_leds) + + +@pytest.mark.asyncio +async def test_scribble_paint_groups_led_count_mismatch_guard(mock_aio_protocol): + """led_count-vs-len mismatch raises (device reports 80, we pass 40).""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + assert light.led_count == 80 + leds = [ScribbleLED(rgb=(1, 2, 3))] * 40 + with pytest.raises(ValueError, match="device has 80"): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 40) + + +@pytest.mark.parametrize( + "led, match", + [ + # rgb AND white both set (mutually exclusive) + (ScribbleLED(rgb=(1, 2, 3), white=50), "mutually exclusive"), + # rgb channel out of 0-255 + (ScribbleLED(rgb=(256, 0, 0)), "rgb values must be 0-255"), + # white out of 0-100 + (ScribbleLED(white=101), "white must be 0-100"), + # blink_speed out of 0-100 + (ScribbleLED(rgb=(1, 2, 3), blink_speed=101), "blink_speed must be 0-100"), + ], +) +@pytest.mark.asyncio +async def test_scribble_paint_groups_per_led_guards(mock_aio_protocol, led, match): + """Cover the per-LED ValueError guards in _scribble_paint_groups.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [led] + [ScribbleLED()] * 79 + with pytest.raises(ValueError, match=match): + light._scribble_paint_groups(leds, 0x00, 0x01, 0x50, 0x64, 80) + + +@pytest.mark.parametrize( + "effect_id", + [-1, 9, 0x10], +) +@pytest.mark.asyncio +async def test_scribble_paint_groups_effect_id_out_of_range( + mock_aio_protocol, effect_id +): + """effect id outside 0x00-0x08 raises ValueError.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(1, 2, 3))] * 80 + with pytest.raises(ValueError, match=r"effect id must be 0x00-0x08"): + light._scribble_paint_groups(leds, effect_id, 0x01, 0x50, 0x64, 80) + + +@pytest.mark.parametrize( + "direction, density, speed, match", + [ + # direction outside {0x01, 0x02} + (0x00, 0x50, 0x64, "direction must be 0x01 or 0x02"), + (0x03, 0x50, 0x64, "direction must be 0x01 or 0x02"), + # density outside 0-100 + (0x01, 101, 0x64, "density must be 0-100"), + # speed outside 0-100 + (0x01, 0x50, 101, "speed must be 0-100"), + ], +) +@pytest.mark.asyncio +async def test_scribble_paint_groups_animation_param_bounds( + mock_aio_protocol, direction, density, speed, match +): + """Animation params (direction/density/speed) are bounds-checked once here.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + leds = [ScribbleLED(rgb=(1, 2, 3))] * 80 + with pytest.raises(ValueError, match=match): + light._scribble_paint_groups(leds, 0x00, direction, density, speed, 80) + + +@pytest.mark.parametrize( + "colors, match", + [ + # empty colors + ([], "at least one color"), + # color tuple wrong length + ([(255, 0)], r"must be .* tuple"), + # color channel out of 0-255 (high) + ([(256, 0, 0)], "must be 0-255"), + # color channel out of 0-255 (low) + ([(-1, 0, 0)], "must be 0-255"), + ], +) +@pytest.mark.asyncio +async def test_generate_extended_custom_effect_color_guards( + mock_aio_protocol, colors, match +): + """Cover the color/empty ValueError guards in _generate_extended_custom_effect.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + with pytest.raises(ValueError, match=match): + light._generate_extended_custom_effect(1, colors) + + +@pytest.mark.parametrize( + "segments, match", + [ + # color tuple wrong length + ([(255, 0)], r"must be .* tuple"), + # color channel out of 0-255 (high) + ([(256, 0, 0)], "must be 0-255"), + # color channel out of 0-255 (low) + ([(0, -1, 0)], "must be 0-255"), + ], +) +@pytest.mark.asyncio +async def test_generate_custom_segment_colors_color_guards( + mock_aio_protocol, segments, match +): + """Cover the color-tuple ValueError guards in _generate_custom_segment_colors.""" + light, _t = await _setup_scribble_light(mock_aio_protocol) + with pytest.raises(ValueError, match=match): + light._generate_custom_segment_colors(segments) + + +@pytest.mark.asyncio +async def test_supports_scribble_property(mock_aio_protocol): + light, _t = await _setup_scribble_light(mock_aio_protocol) + assert light.supports_scribble is True + + +# Tests for extended_custom_effect_pattern_list property (base_device.py line 658) + + +@pytest.mark.asyncio +async def test_extended_custom_effect_pattern_list_0xB6(mock_aio_protocol): + """Test extended_custom_effect_pattern_list returns list for 0xB6 device.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x61, + 0x24, + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + # Test extended_custom_effect_pattern_list returns a list + pattern_list = light.extended_custom_effect_pattern_list + assert pattern_list is not None + assert isinstance(pattern_list, list) + assert len(pattern_list) > 0 + # Check some expected patterns (Title Case, matching the reported effect names) + assert "Wave" in pattern_list + assert "Meteor" in pattern_list + assert "Breathe" in pattern_list + assert "Static Fill" in pattern_list + + +@pytest.mark.asyncio +async def test_extended_custom_effect_pattern_list_non_0xB6(mock_aio_protocol): + """Test extended_custom_effect_pattern_list returns None for non-0xB6 device.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # Standard 0x25 device (not extended custom) + light._aio_protocol.data_received( + b"\x81\x25\x23\x61\x05\x10\xb6\x00\x98\x19\x04\x25\x0f\xde" + ) + await task + + # Should return None for non-extended devices + assert light.extended_custom_effect_pattern_list is None + + +# Tests for _named_effect with extended custom effects (base_device.py line 676) + + +@pytest.mark.asyncio +async def test_named_effect_extended_custom_0xB6(mock_aio_protocol): + """Test _named_effect returns extended effect name for 0xB6 in custom mode.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # 0xB6 device with preset_pattern=0x25 (custom effect mode) and mode=0x01 (Wave) + # Extended state: preset_pattern at pos 7, mode at pos 8 + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, # Model + 0x01, # Version + 0x23, # Power on + 0x25, # preset_pattern = 0x25 (custom effect mode) + 0x01, # mode = Wave pattern ID + 0x64, # speed + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + # Effect should be "Wave" from EXTENDED_CUSTOM_EFFECT_ID_NAME + assert light.effect == "Wave" + + +@pytest.mark.asyncio +async def test_0xB6_extended_effect_reporting_and_discovery(mock_aio_protocol): + """0xB6 reports the running extended scene and exposes it for discovery. + + A running scene is reported by ``effect`` (e.g. "Wave") and is discoverable + via ``extended_custom_effect_pattern_list``. Extended effects are set via the + dedicated ``setExtendedCustomEffect`` API and are intentionally NOT yet wired + into the generic ``effect_list``/``set_effect`` path (deferred to a follow-up + PR; they require a color list) -- so the reported name is not expected in + ``effect_list`` here. + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # 0xB6 Wave scene frame (preset_pattern 0x25, mode 0x01 = Wave). + light._aio_protocol.data_received( + bytes.fromhex("ea810100b60923250150f0b46464ff000500500000002002010003") + ) + await task + + # Readback of the running scene works. + assert light.effect == "Wave" + # Discoverable via the dedicated list (Title Case, same source as `effect`). + assert "Wave" in light.extended_custom_effect_pattern_list + assert "Static Fill" in light.extended_custom_effect_pattern_list + # Deferred: extended names are not (yet) advertised in the generic effect_list. + assert "Wave" not in light.effect_list + + +@pytest.mark.asyncio +async def test_named_effect_extended_custom_meteor(mock_aio_protocol): + """Test _named_effect returns Meteor for mode=0x02.""" + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # 0xB6 device with mode=0x02 (Meteor) + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x25, # preset_pattern = 0x25 + 0x02, # mode = Meteor + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + assert light.effect == "Meteor" + + +@pytest.mark.asyncio +async def test_named_effect_extended_custom_unmapped_mode(mock_aio_protocol, caplog): + """An unmapped extended-custom mode returns a placeholder and is logged. + + Preset 0x25 always means a scene is active, so an unmapped mode reports a + distinguishable ``unknown (0x..)`` name rather than None. + """ + light = AIOWifiLedBulb("192.168.1.166") + + def _updated_callback(*args, **kwargs): + pass + + task = asyncio.create_task(light.async_setup(_updated_callback)) + await mock_aio_protocol() + + # 0xB6 device with preset_pattern=0x25 and an unmapped mode (0x30) + light._aio_protocol.data_received( + bytes( + ( + 0xEA, + 0x81, + 0x01, + 0x00, + 0xB6, + 0x01, + 0x23, + 0x25, # preset_pattern = 0x25 + 0x30, # mode = unmapped + 0x64, + 0x0F, + 0x00, + 0x00, + 0x00, + 0x64, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x83, + ) + ) + ) + await task + + with caplog.at_level(logging.DEBUG, logger="flux_led.base_device"): + assert light.effect == "unknown (0x30)" + assert "Unmapped extended custom effect mode" in caplog.text diff --git a/tests/test_sync.py b/tests/test_sync.py index 80a0e01f..e94f5884 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -19,7 +19,12 @@ STATE_RED, STATE_WARM_WHITE, TRANSITION_GRADUAL, + ExtendedCustomEffectDirection, + ExtendedCustomEffectOption, + ExtendedCustomEffectPattern, MultiColorEffects, + ScribbleEffect, + ScribbleLED, ) from flux_led.pattern import PresetPattern from flux_led.protocol import ( @@ -1484,6 +1489,304 @@ def read_data(expected): with self.assertRaisesRegex(Exception, "Cannot determine protocol"): flux_led.WifiLedBulb("192.168.1.199") + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setExtendedCustomEffect(self, mock_connect, mock_read, mock_send): + """Test sync setExtendedCustomEffect for 0xB6 device.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + + # Call setExtendedCustomEffect + light.setExtendedCustomEffect( + pattern_id=1, + colors=[(255, 0, 0), (0, 255, 0)], + speed=50, + density=50, + ) + + # Verify _send_msg was called + assert mock_send.called + sent_data = mock_send.call_args[0][0] + # Verify it's a wrapped message + assert sent_data[0] == 0xB0 + assert sent_data[1] == 0xB1 + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setExtendedCustomEffect_enums( + self, mock_connect, mock_read, mock_send + ): + """setExtendedCustomEffect accepts the public enums, not just ints.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + + # Passing the enum members must not raise and must send a wrapped message. + light.setExtendedCustomEffect( + pattern_id=ExtendedCustomEffectPattern.WAVE, + colors=[(255, 0, 0), (0, 255, 0)], + speed=50, + density=50, + direction=ExtendedCustomEffectDirection.LEFT_TO_RIGHT, + option=ExtendedCustomEffectOption.VARIANT_1, + ) + assert mock_send.called + enum_sent = mock_send.call_args[0][0] + assert enum_sent[0] == 0xB0 + assert enum_sent[1] == 0xB1 + + # The produced inner message must equal the equivalent all-int call. + # (The B0B1 wrapper carries an incrementing counter, so compare the + # unwrapped inner E1 21 payload.) + int_bytes = light._generate_extended_custom_effect( + ExtendedCustomEffectPattern.WAVE.value, + [(255, 0, 0), (0, 255, 0)], + 50, + 50, + ExtendedCustomEffectDirection.LEFT_TO_RIGHT.value, + ExtendedCustomEffectOption.VARIANT_1.value, + ) + enum_bytes = light._generate_extended_custom_effect( + ExtendedCustomEffectPattern.WAVE, + [(255, 0, 0), (0, 255, 0)], + 50, + 50, + ExtendedCustomEffectDirection.LEFT_TO_RIGHT, + ExtendedCustomEffectOption.VARIANT_1, + ) + assert self._inner_of(enum_bytes) == self._inner_of(int_bytes) + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setCustomSegmentColors(self, mock_connect, mock_read, mock_send): + """Test sync setCustomSegmentColors for 0xB6 device.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + + # Call setCustomSegmentColors + light.setCustomSegmentColors(segments=[(255, 0, 0), None, (0, 0, 255)]) + + # Verify _send_msg was called + assert mock_send.called + sent_data = mock_send.call_args[0][0] + # Verify it's a wrapped message + assert sent_data[0] == 0xB0 + assert sent_data[1] == 0xB1 + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setScribble(self, mock_connect, mock_read, mock_send): + """Test sync setScribble runs the E1 23 init + E1 26 paint loop.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + + # Ignore any sends performed during setup / state query. + mock_send.reset_mock() + + # Two color groups so the paint loop emits more than one E1 26 message. + num_leds = light.led_count + half = num_leds // 2 + leds = [ScribbleLED(rgb=(255, 0, 0))] * half + [ + ScribbleLED(rgb=(0, 0, 255)) + ] * (num_leds - half) + + light.setScribble(leds, effect=ScribbleEffect.STATIC) + + # enter_mode init (E1 23) + one E1 26 paint per color group = 3 sends, + # all B0 B1 wrapped, confirming the sync multi-message loop runs. + assert mock_send.call_count == 3 + sent = [call.args[0] for call in mock_send.call_args_list] + for msg in sent: + assert msg[0] == 0xB0 + assert msg[1] == 0xB1 + # First send is the scribble-mode init (E1 23); the rest are paints (E1 26). + assert self._inner_of(sent[0])[:2] == b"\xe1\x23" + assert self._inner_of(sent[1])[:2] == b"\xe1\x26" + assert self._inner_of(sent[2])[:2] == b"\xe1\x26" + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setScribble_raw_int_direction( + self, mock_connect, mock_read, mock_send + ): + """setScribble accepts a raw int direction, not just the enum.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + mock_send.reset_mock() + + leds = [ScribbleLED(rgb=(255, 0, 0))] * light.led_count + # A raw int direction (0x02) must not raise AttributeError on + # direction.value and must carry through as the wire direction byte. + light.setScribble(leds, direction=0x02, enter_mode=False) + + assert mock_send.call_count == 1 + paint = self._inner_of(mock_send.call_args_list[0].args[0]) + assert paint[:2] == b"\xe1\x26" + assert paint[3] == 0x02 # direction byte + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_setScribble_invalid_input_sends_nothing( + self, mock_connect, mock_read, mock_send + ): + """Invalid scribble input raises before any send (including the E1 23 init).""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\xea\x81") + if calls == 2: + self.assertEqual(expected, 25) + return bytearray( + b"\x01\x00\xb6\x09\x24\x66\x01\x64\xf0\x00\x00\x00\x00\x64\x05\x00\x64\x00\x00\x00\x20\x02\x01\x00\x03" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + self.assertEqual(light.protocol, PROTOCOL_LEDENET_EXTENDED_CUSTOM) + mock_send.reset_mock() + + # 3 LEDs on a 100-LED device is a length mismatch -> ValueError, and + # nothing (not even the E1 23 init) must be sent. + with self.assertRaises(ValueError): + light.setScribble([ScribbleLED(rgb=(255, 0, 0))] * 3, enter_mode=True) + assert mock_send.call_count == 0 + + @staticmethod + def _inner_of(wrapped: bytearray) -> bytes: + """Extract the inner message from a B0B1B2B3-wrapped result.""" + inner_len = (wrapped[8] << 8) | wrapped[9] + return bytes(wrapped[10 : 10 + inner_len]) + + @patch("flux_led.WifiLedBulb._send_msg") + @patch("flux_led.WifiLedBulb._read_msg") + @patch("flux_led.WifiLedBulb.connect") + def test_0xB6_methods_raise_on_unsupported_device( + self, mock_connect, mock_read, mock_send + ): + """Sync extended-custom methods raise ValueError on a non-0xB6 device.""" + calls = 0 + + def read_data(expected): + nonlocal calls + calls += 1 + if calls == 1: + self.assertEqual(expected, 2) + return bytearray(b"\x81\x35") + if calls == 2: + self.assertEqual(expected, 12) + return bytearray(b"\x23\x61\x05\x10\xb6\x00\x98\x00\x09\x00\xf0\x96") + if calls == 3: + self.assertEqual(expected, 14) + return bytearray( + b"\x81\x35\x23\x61\x05\x10\xb6\x00\x98\x19\x09\x25\x0f\xf3" + ) + if calls == 4: + self.assertEqual(expected, 14) + return bytearray( + b"\x81\x35\x23\x38\x05\x10\xb6\x00\x98\x19\x09\x25\x0f\xca" + ) + + mock_read.side_effect = read_data + light = flux_led.WifiLedBulb("192.168.1.164") + # Standard RGBCW bulb -- does not use the extended custom protocol. + self.assertEqual(light.protocol, PROTOCOL_LEDENET_9BYTE_DIMMABLE_EFFECTS) + assert not light.supports_extended_custom_effects + assert not light.supports_scribble + + with self.assertRaises(ValueError): + light.setExtendedCustomEffect(pattern_id=1, colors=[(255, 0, 0)]) + with self.assertRaises(ValueError): + light.setCustomSegmentColors(segments=[(255, 0, 0)]) + with self.assertRaises(ValueError): + light.setScribble([ScribbleLED(rgb=(255, 0, 0))]) + @patch("flux_led.WifiLedBulb._send_msg") @patch("flux_led.WifiLedBulb._read_msg") @patch("flux_led.WifiLedBulb.connect")