From 5e13da39671ead8f24cef5a69e4f0fa2d2fbba7d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Apr 2026 21:40:52 +0000 Subject: [PATCH] feat(03-01): implement WLED protocol encoders and animation command packet - encode_drgb/encode_drgbw: 2-byte header + RGB/RGBW body, truncate at MTU max - encode_dnrgb_packets: splits large frames into 489-LED chunks with big-endian start index - encode_animation_cmd: 16-byte 0xAC packet with speed (float32), color, flags, density - decode_animation_cmd: full inverse round-trip for all 7 animation types - All 27 tests pass --- lightsync/protocol/animation_cmd.py | 146 ++++++++++++++++++++++++++++ lightsync/protocol/drgb.py | 71 ++++++++++++++ 2 files changed, 217 insertions(+) create mode 100644 lightsync/protocol/animation_cmd.py create mode 100644 lightsync/protocol/drgb.py diff --git a/lightsync/protocol/animation_cmd.py b/lightsync/protocol/animation_cmd.py new file mode 100644 index 0000000..589032a --- /dev/null +++ b/lightsync/protocol/animation_cmd.py @@ -0,0 +1,146 @@ +"""Custom animation command packet encoder/decoder for LightSync firmware. + +Packet format (16 bytes, all big-endian): + Byte 0: Protocol marker = 0xAC (distinguishes from WLED types 0x01-0x04) + Byte 1: Protocol version = 1 + Byte 2: Animation type ID (0-6, see ANIMATION_IDS) + Byte 3: Flags (bit 0 = reverse, bits 1-7 reserved) + Bytes 4-7: Speed (float32 big-endian, range 0.0-1.0) + Bytes 8-10: Primary color R, G, B + Bytes 11-13: Secondary (background) color R, G, B + Byte 14: White channel (SK6812 W value, 0 for RGB-only strips) + Byte 15: Density/length (animation-specific: size, cooling, etc.) +""" +import struct + +ANIM_CMD_MARKER = 0xAC +ANIM_CMD_VERSION = 1 + +ANIMATION_IDS: dict[str, int] = { + "solid_color": 0, + "chase": 1, + "pulse": 2, + "rainbow": 3, + "strobe": 4, + "color_wipe": 5, + "fire": 6, +} + +ANIMATION_NAMES: dict[int, str] = {v: k for k, v in ANIMATION_IDS.items()} + +_PACKET_SIZE = 16 + + +def encode_animation_cmd(animation: str, params: dict) -> bytes: + """Encode an animation command into a 16-byte LightSync packet. + + Args: + animation: Animation name (must be a key in ANIMATION_IDS). + params: Animation parameters dict. Recognized keys: + - color ([R, G, B]): Primary color, default [0, 0, 0] + - bg_color ([R, G, B]): Secondary/background color, default [0, 0, 0] + - speed (float 0.0-1.0): Normalized speed, default 0.5 + - white (int 0-255): W channel for SK6812, default 0 + - reverse (bool): Direction flag, stored in flags bit 0, default False + - density/size/cooling (int 0-255): Byte 15, animation-specific meaning + + Returns: + 16-byte binary packet. + + Raises: + ValueError: If animation name is not in ANIMATION_IDS. + """ + if animation not in ANIMATION_IDS: + raise ValueError( + f"Unknown animation type: {animation!r}. " + f"Valid types: {sorted(ANIMATION_IDS.keys())}" + ) + + anim_id = ANIMATION_IDS[animation] + + # Flags byte: bit 0 = reverse + reverse = bool(params.get("reverse", False)) + flags = 0x01 if reverse else 0x00 + + speed = float(params.get("speed", 0.5)) + color = params.get("color", [0, 0, 0]) + bg_color = params.get("bg_color", [0, 0, 0]) + white = int(params.get("white", 0)) & 0xFF + + # Byte 15: density falls back to size, then cooling, then 0 + density = int( + params.get("density", params.get("size", params.get("cooling", 0))) + ) & 0xFF + + packet = struct.pack( + ">BBBBf", + ANIM_CMD_MARKER, # 0: marker + ANIM_CMD_VERSION, # 1: version + anim_id, # 2: animation type ID + flags, # 3: flags + speed, # 4-7: speed float32 + ) + packet += bytes([ + color[0] & 0xFF, # 8: R primary + color[1] & 0xFF, # 9: G primary + color[2] & 0xFF, # 10: B primary + bg_color[0] & 0xFF, # 11: R secondary + bg_color[1] & 0xFF, # 12: G secondary + bg_color[2] & 0xFF, # 13: B secondary + white, # 14: white channel + density, # 15: density/size/cooling + ]) + + assert len(packet) == _PACKET_SIZE, f"Packet size {len(packet)} != {_PACKET_SIZE}" + return packet + + +def decode_animation_cmd(data: bytes) -> dict: + """Decode a LightSync animation command packet. + + Args: + data: 16-byte binary packet produced by encode_animation_cmd. + + Returns: + Dict with keys: + - animation (str): Animation name + - params (dict): Decoded parameters (speed, color, bg_color, white, + density, reverse, flags) + + Raises: + ValueError: If packet is too short, marker is wrong, or animation ID unknown. + """ + if len(data) < _PACKET_SIZE: + raise ValueError(f"Packet too short: {len(data)} bytes, expected {_PACKET_SIZE}") + + marker = data[0] + if marker != ANIM_CMD_MARKER: + raise ValueError(f"Invalid marker: 0x{marker:02X}, expected 0x{ANIM_CMD_MARKER:02X}") + + _version = data[1] + anim_id = data[2] + flags = data[3] + (speed,) = struct.unpack_from(">f", data, 4) + + if anim_id not in ANIMATION_NAMES: + raise ValueError(f"Unknown animation ID: {anim_id}") + + animation = ANIMATION_NAMES[anim_id] + color = [data[8], data[9], data[10]] + bg_color = [data[11], data[12], data[13]] + white = data[14] + density = data[15] + reverse = bool(flags & 0x01) + + return { + "animation": animation, + "params": { + "speed": speed, + "color": color, + "bg_color": bg_color, + "white": white, + "density": density, + "reverse": reverse, + "flags": flags, + }, + } diff --git a/lightsync/protocol/drgb.py b/lightsync/protocol/drgb.py new file mode 100644 index 0000000..e0c0d8f --- /dev/null +++ b/lightsync/protocol/drgb.py @@ -0,0 +1,71 @@ +"""WLED-compatible UDP packet encoders — DRGB, DRGBW, DNRGB. + +Source: https://kno.wled.ge/interfaces/udp-realtime/ +""" +import struct + +WLED_PORT = 21324 +PROTOCOL_DRGB = 2 +PROTOCOL_DRGBW = 3 +PROTOCOL_DNRGB = 4 +TIMEOUT_BYTE = 2 # seconds before device returns to normal mode + +# Max pixels per packet (stays under 1472-byte UDP payload limit) +# Standard Ethernet MTU = 1500; IP header = 20; UDP header = 8 → max payload = 1472 +DRGB_MAX_PX = 490 # (1472 - 2) // 3 +DRGBW_MAX_PX = 367 # (1472 - 2) // 4 +DNRGB_MAX_PX = 489 # (1472 - 4) // 3 + + +def encode_drgb(pixels: list[tuple[int, int, int]]) -> bytes: + """Encode up to 490 RGB pixels into a DRGB UDP packet. + + Format: [0x02][timeout][R0][G0][B0]...[Rn][Gn][Bn] + Truncates silently to DRGB_MAX_PX if more pixels are provided. + """ + header = bytes([PROTOCOL_DRGB, TIMEOUT_BYTE]) + body = bytearray() + for r, g, b in pixels[:DRGB_MAX_PX]: + body.extend([r & 0xFF, g & 0xFF, b & 0xFF]) + return header + bytes(body) + + +def encode_drgbw(pixels: list[tuple[int, int, int, int]]) -> bytes: + """Encode up to 367 RGBW pixels into a DRGBW UDP packet. + + Format: [0x03][timeout][R0][G0][B0][W0]...[Rn][Gn][Bn][Wn] + Truncates silently to DRGBW_MAX_PX if more pixels are provided. + """ + header = bytes([PROTOCOL_DRGBW, TIMEOUT_BYTE]) + body = bytearray() + for r, g, b, w in pixels[:DRGBW_MAX_PX]: + body.extend([r & 0xFF, g & 0xFF, b & 0xFF, w & 0xFF]) + return header + bytes(body) + + +def encode_dnrgb_packets( + pixels: list[tuple[int, int, int]], + start_index: int = 0, +) -> list[bytes]: + """Split a large RGB pixel array into DNRGB UDP packets, each under MTU. + + Format per packet: [0x04][timeout][start_hi][start_lo][R0][G0][B0]...[Rn][Gn][Bn] + Start index is 16-bit big-endian per WLED spec. + + Args: + pixels: Full list of (R, G, B) pixel tuples. + start_index: LED strip offset for the first pixel (default 0). + + Returns: + List of byte strings, one per chunk of up to DNRGB_MAX_PX pixels. + """ + packets = [] + for offset in range(0, len(pixels), DNRGB_MAX_PX): + chunk = pixels[offset:offset + DNRGB_MAX_PX] + idx = start_index + offset + header = struct.pack(">BBH", PROTOCOL_DNRGB, TIMEOUT_BYTE, idx) + body = bytearray() + for r, g, b in chunk: + body.extend([r & 0xFF, g & 0xFF, b & 0xFF]) + packets.append(header + bytes(body)) + return packets