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
2026-04-06 21:40:52 +00:00
parent ffc4ef6dc6
commit 5e13da3967
2 changed files with 217 additions and 0 deletions
--- a/lightsync/protocol/animation_cmd.py
+++ 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,
+        },
+    }
--- a/lightsync/protocol/drgb.py
+++ 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