# Phase 3: Communication Protocol - Research **Researched:** 2026-04-06 **Domain:** UDP networking, LED protocol encoding, animation library design, asyncio datagram sockets **Confidence:** HIGH --- ## Summary Phase 3 builds the wire protocol and animation library that microcontrollers will implement in Phase 7. The target is WLED-compatible UDP — specifically DRGB (full RGB frame), DRGBW (full RGBW frame), and DNRGB (multi-packet variant for large strips). The protocol is well-documented by the WLED project and has a fixed, binary byte layout. Nothing needs to be invented here: we adopt the WLED spec wholesale. The existing codebase already has the scaffolding Phase 3 needs. `BaseDevice` ABC is defined in `lightsync/devices/base.py` with `encode_frame` and `encode_animation_cmd` stubs. SK6812Device and WS2801Device exist with working pixel serialization but stub animation encoding. The device registry can instantiate live device objects. Phase 3 completes the stubs and adds the UDP transport layer on top. The two implementation risks are (1) multi-packet framing for large strips — DNRGB must split 300-LED RGBW frames (1200 bytes payload) across multiple 489-LED-per-packet chunks — and (2) defining a custom animation command packet format (the WLED protocol doesn't specify this; it's only for raw frames). Animation commands are custom binary messages Phase 7 firmware will parse. **Primary recommendation:** Use a single shared asyncio UDP socket (one `DatagramTransport`) with `sendto(data, addr)` calls to each device. No socket pool needed — UDP is connectionless and a single socket can send to N addresses without overhead. Implement the simulator as an asyncio UDP server on localhost using the same port convention (21324). --- ## Standard Stack ### Core | Library | Version | Purpose | Why Standard | |---------|---------|---------|--------------| | asyncio (stdlib) | 3.11 | Async UDP transport | No dependency, DatagramTransport handles non-blocking sends | | struct (stdlib) | 3.11 | Binary packet encoding | Correct byte-order packing, fast, zero dependency | | dataclasses / Pydantic | already in project | Animation parameter models | Pydantic already a dependency (pyproject.toml) | ### Supporting | Library | Version | Purpose | When to Use | |---------|---------|---------|-------------| | asyncio.DatagramProtocol | stdlib | UDP send/receive base class | Simulator receiver, optional transport wrapper | | typing.TypedDict | stdlib | Structured animation params | Lightweight param typing without full Pydantic model | ### Alternatives Considered | Instead of | Could Use | Tradeoff | |------------|-----------|----------| | asyncio DatagramTransport | asyncudp 0.10 | asyncudp adds nicer await-based API but adds a dependency for minimal gain | | struct.pack | manual bytearray | struct is faster, clearer, handles endianness correctly | | Pydantic for anim params | plain dicts | Plain dicts are fine for params since show.py CueModel already uses `dict[str, Any]` | No new dependencies required. All encoding and transport fits cleanly in stdlib asyncio + struct. --- ## Architecture Patterns ### Recommended Project Structure ``` lightsync/ ├── devices/ │ ├── base.py # BaseDevice ABC (already exists — extend, don't replace) │ ├── registry.py # DeviceRegistry (already exists — add instantiate pool) │ ├── sk6812.py # SK6812Device (already exists — fill encode_animation_cmd) │ ├── ws2801.py # WS2801Device (already exists — fill encode_animation_cmd) │ └── udp_sender.py # NEW: UDPSender — shared socket, send_frame / send_cmd ├── protocol/ │ ├── __init__.py │ ├── drgb.py # NEW: encode_drgb, encode_drgbw, encode_dnrgb │ ├── animation_cmd.py # NEW: encode_animation_cmd, decode_animation_cmd │ └── simulator.py # NEW: asyncio UDP receiver + packet logger └── animations/ ├── __init__.py ├── base.py # NEW: AnimationBase ABC — render(t, led_count) -> list[tuple] ├── chase.py ├── pulse.py ├── rainbow.py ├── strobe.py ├── color_wipe.py ├── fire.py └── solid_color.py ``` ### Pattern 1: Single Socket UDP Sender **What:** One asyncio DatagramTransport opened at app startup, held on `app.state`. All device sends use `transport.sendto(payload, (ip, port))`. **When to use:** Sending animation commands and pixel frames to N devices from a single FastAPI process. ```python # Source: Python docs asyncio-protocol.html import asyncio class _NullProtocol(asyncio.DatagramProtocol): """Minimal protocol — we only send, never receive on this socket.""" def error_received(self, exc: Exception) -> None: pass # UDP send errors are non-fatal for LED strip delivery class UDPSender: def __init__(self): self._transport: asyncio.DatagramTransport | None = None async def start(self) -> None: loop = asyncio.get_running_loop() self._transport, _ = await loop.create_datagram_endpoint( _NullProtocol, local_addr=("0.0.0.0", 0), # OS assigns ephemeral port ) def send(self, payload: bytes, ip: str, port: int) -> None: if self._transport: self._transport.sendto(payload, (ip, port)) async def stop(self) -> None: if self._transport: self._transport.close() ``` ### Pattern 2: WLED DRGB/DNRGB Packet Encoding **What:** Binary struct packing per the WLED spec. Two-byte header (type, timeout), then pixel data. **When to use:** Sending full pixel frames to WLED-compatible or LightSync firmware devices. ```python # Source: 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) DRGB_MAX_PX = 490 # 2 header + 490*3 = 1472 DRGBW_MAX_PX = 367 # 2 header + 367*4 = 1470 DNRGB_MAX_PX = 489 # 4 header + 489*3 = 1471 def encode_drgb(pixels: list[tuple[int, int, int]]) -> bytes: """Full DRGB frame — up to 490 pixels.""" 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: """Full DRGBW frame — up to 367 pixels.""" 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 large RGB frame into DNRGB chunks, each under MTU.""" 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 ``` ### Pattern 3: Custom Animation Command Packet **What:** A custom binary packet format for sending "start animation X with params P" to a device. Not part of WLED spec — specific to LightSync firmware. **When to use:** Phase 5 live execution fires animation commands when cue timestamps are reached. Proposed format (all big-endian): ``` Byte 0: Protocol type = 0xAC (animation command marker, distinguishes from WLED) Byte 1: Protocol version = 1 Byte 2: Animation type ID (0-6, see table below) Byte 3: Flags (bit 0 = loop, bit 1 = reverse, bits 2-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 color R, G, B Byte 14: White channel (for SK6812, 0 for RGB-only strips) Byte 15: Density/length (0-255, animation-specific meaning) ``` Animation type IDs: ``` 0 = solid_color 1 = chase 2 = pulse 3 = rainbow 4 = strobe 5 = color_wipe 6 = fire ``` ### Pattern 4: Animation Base + Render Protocol **What:** Each animation is a class with a `render(t: float, led_count: int) -> list[tuple]` method. Pure computation, no I/O. **When to use:** Phase 3 generates frames for simulator validation; Phase 5 live execution calls render() in real-time. ```python from abc import ABC, abstractmethod class AnimationBase(ABC): """Pure animation renderer — no I/O, stateless or state-per-instance.""" @abstractmethod def render(self, t: float, led_count: int) -> list[tuple]: """ Render frame at time t (seconds since animation started). Returns list of (R, G, B) or (R, G, B, W) tuples, length == led_count. """ ... @classmethod @abstractmethod def from_params(cls, params: dict) -> "AnimationBase": """Construct from CueModel.params dict.""" ... ``` ### Anti-Patterns to Avoid - **Creating one UDP socket per device:** Unnecessary OS resource usage. One socket can sendto() N addresses. - **Using TCP for LED data:** Adds head-of-line blocking; a dropped LED frame is better than blocking. - **Blocking socket calls in async context:** Never call `socket.send()` directly in a coroutine. Always use DatagramTransport.sendto(). - **IP fragmentation of large frames:** Always split at the application layer using DNRGB multi-packet. Do not rely on OS fragmentation reassembly on WiFi hardware. - **Storing animation state in the packet spec:** Animation command packets are fire-and-forget commands, not streaming state machines. The microcontroller owns animation loop state after receiving the command. --- ## Don't Hand-Roll | Problem | Don't Build | Use Instead | Why | |---------|-------------|-------------|-----| | Binary packing | Manual bytearray bit manipulation | `struct.pack(">BBH", ...)` | struct handles endianness, sign, alignment | | Fire animation math | Custom heat diffusion | Port Fire2012 algorithm (cooling + sparking) | FastLED Fire2012 is battle-tested on thousands of strips | | Rainbow hue cycling | Custom HSV math | `colorsys.hsv_to_rgb()` (stdlib) | stdlib handles edge cases, no dependency | | MTU calculation | Custom fragmentation logic | WLED DNRGB spec (489 px/packet) | Max values already calculated for standard Ethernet MTU 1500 | **Key insight:** The WLED protocol spec eliminates all packet design work. The only custom protocol element is the animation command packet (0xAC header), which is simple enough to implement with struct in ~30 lines. --- ## WLED Protocol Specification (HIGH confidence) Source: [kno.wled.ge/interfaces/udp-realtime/](https://kno.wled.ge/interfaces/udp-realtime/) ### Full Byte Layout **Common header (all protocols):** - Byte 0: Protocol type (1=WARLS, 2=DRGB, 3=DRGBW, 4=DNRGB) - Byte 1: Timeout in seconds (recommended: 2; 255=never timeout) **DRGB (type=2, max 490 LEDs):** ``` [0x02][timeout][R0][G0][B0][R1][G1][B1]...[Rn][Gn][Bn] payload = 2 + led_count * 3 bytes ``` **DRGBW (type=3, max 367 LEDs):** ``` [0x03][timeout][R0][G0][B0][W0][R1][G1][B1][W1]...[Rn][Gn][Bn][Wn] payload = 2 + led_count * 4 bytes ``` **DNRGB (type=4, max 489 LEDs per packet — multi-packet):** ``` [0x04][timeout][start_hi][start_lo][R0][G0][B0]...[Rn][Gn][Bn] payload = 4 + chunk_led_count * 3 bytes start index is 16-bit big-endian (max 65535 LEDs addressable) ``` ### MTU Math (verified) Standard Ethernet MTU = 1500 bytes IP header = 20 bytes, UDP header = 8 bytes Max UDP payload = 1472 bytes - DRGB max: (1472 - 2) / 3 = 490 LEDs - DRGBW max: (1472 - 2) / 4 = 367 LEDs - DNRGB max: (1472 - 4) / 3 = 489 LEDs per packet **SK6812 300-LED strip in DRGBW mode requires ceil(300/367) = 1 packet** — fits in MTU. **SK6812 300-LED RGBW payload = 2 + 300*4 = 1202 bytes — fits in single packet.** However, to stay compliant with DNRGB multi-packet splitting for arbitrarily large strips, Phase 3 MUST implement the chunked sender regardless — the UDP-03 requirement mandates it. ### Default Port WLED listens on UDP port **21324** (configurable in WLED settings). LightSync should use the same default for firmware compatibility. --- ## Animation Parameter Shapes All 7 required animation types with their parameters (ANI-01, ANI-02): ### solid_color ```python params = { "color": [r, g, b], # Primary color RGB 0-255 "white": 0, # W channel for SK6812 (ANI-03) } ``` ### chase ```python params = { "color": [r, g, b], "white": 0, "bg_color": [0, 0, 0], # Background color "speed": 0.5, # 0.0-1.0 (maps to px/sec) "size": 3, # Lit pixel run length "spacing": 7, # Dark gap after each run "reverse": False, } ``` ### pulse ```python params = { "color": [r, g, b], "white": 0, "speed": 0.5, # Period of one breath cycle (0.0-1.0 → 0.1-5.0 sec) "min_brightness": 0, # Floor brightness 0-255 "max_brightness": 255, # Ceiling brightness 0-255 } ``` ### rainbow ```python params = { "speed": 0.5, # Hue shift speed (0.0-1.0) "period": 1.0, # How many full rainbow cycles fit on strip (1.0 = one cycle) } # No color param — rainbow defines its own color progression ``` ### strobe ```python params = { "color": [r, g, b], "white": 0, "speed": 0.5, # Flashes per second (maps to 1-20 Hz) "duty_cycle": 0.1, # Fraction of period that LEDs are on (0.05-0.5) } ``` ### color_wipe ```python params = { "color": [r, g, b], "white": 0, "speed": 0.5, # Fill rate (0.0-1.0 → px/sec) "reverse": False, # Direction: False = start→end, True = end→start } ``` ### fire ```python params = { "cooling": 55, # Heat loss per cycle — 20 (tall flames) to 100 (short flames) "sparking": 120, # Probability of new spark (0-255) — 50 (flicker) to 200 (roaring) "speed": 0.5, # Frame rate multiplier } # Uses Fire2012 algorithm: cool → drift up → spark → render heat as color # Color palette: black → red → yellow → white (heat intensity) ``` **Common fields across all animations:** - `speed` (float 0.0-1.0): Normalized speed that each animation maps to its native unit - `color` ([R,G,B]): Primary color for animations that use it - `white` (int 0-255): W channel, only meaningful for SK6812 (ANI-03 / ANI-04) - `reverse` (bool): Direction flag where applicable --- ## Common Pitfalls ### Pitfall 1: DNRGB Start Index Byte Order **What goes wrong:** Encoding start index as little-endian instead of big-endian — simulator and firmware parse wrong start position. **Why it happens:** Python `struct.pack("H", idx)` or `struct.pack(">BBH", type, timeout, idx)` for DNRGB header. **Warning signs:** Simulator logs show LED segments rendering at wrong strip positions. ### Pitfall 2: Single Socket Used Across Threads **What goes wrong:** DatagramTransport.sendto() called from a non-asyncio thread (e.g., from a sync endpoint) causes race conditions. **Why it happens:** asyncio transports are not thread-safe (documented in Python stdlib). **How to avoid:** Always call `sender.send()` from within the asyncio event loop using `loop.call_soon_threadsafe()` if bridging from sync code. **Warning signs:** Sporadic `RuntimeError: no running event loop` or corrupted packets. ### Pitfall 3: SK6812 Frame Mapped to DRGB Instead of DRGBW **What goes wrong:** SK6812 device sends DRGB packets — W channel silently dropped, strip shows wrong colors. **Why it happens:** `encode_frame()` returns correct bytes but the dispatch logic sends DRGB type byte (0x02) instead of DRGBW (0x03). **How to avoid:** Device dispatch must check `device.bytes_per_pixel` — 4 bytes → DRGBW, 3 bytes → DRGB. The type byte must match the data layout. **Warning signs:** Simulator shows 4 bytes per pixel in payload but type byte = 0x02. ### Pitfall 4: Animation Command Packet Conflicts with WLED Type Byte **What goes wrong:** If animation command type byte (0xAC) ever matches WARLS (0x01), DRGB (0x02), DRGBW (0x03), or DNRGB (0x04), WLED firmware would misparse it. **Why it happens:** Not reserving a clean namespace for the custom protocol. **How to avoid:** Use 0xAC (172) as the animation command marker — well clear of WLED's 0x01-0x04 range. ### Pitfall 5: Simulator Bound to Same Port as Sender **What goes wrong:** Simulator and the app both try to bind UDP port 21324 on the same machine — bind fails with `OSError: [Errno 98] Address already in use`. **Why it happens:** Simulator acts as receiver (needs to bind), app is sender (doesn't bind to destination port). **How to avoid:** Simulator binds to 0.0.0.0:21324. App sender socket uses `local_addr=("0.0.0.0", 0)` (OS assigns ephemeral source port). These don't conflict. ### Pitfall 6: Fire Animation Indexing Direction **What goes wrong:** Fire animation renders with flames at top instead of bottom, or runs at wrong speed. **Why it happens:** Python's natural list iteration goes index 0 → N; fire should rise from index 0 (bottom). **How to avoid:** Heat array index 0 = bottom of strip (floor/heat source), index N-1 = top. Sparks added at indices 0-2, heat drifts toward index N-1. --- ## Code Examples ### Asyncio UDP Simulator (Receiver) ```python # Source: Python docs asyncio-protocol.html (DatagramProtocol pattern) import asyncio import struct class SimulatorProtocol(asyncio.DatagramProtocol): PROTOCOL_NAMES = {1: "WARLS", 2: "DRGB", 3: "DRGBW", 4: "DNRGB", 0xAC: "ANIM_CMD"} def datagram_received(self, data: bytes, addr: tuple) -> None: if len(data) < 2: return proto_type = data[0] timeout = data[1] name = self.PROTOCOL_NAMES.get(proto_type, f"UNKNOWN(0x{proto_type:02X})") if proto_type == 2: # DRGB led_count = (len(data) - 2) // 3 print(f"[SIM] DRGB from {addr}: {led_count} LEDs, timeout={timeout}s") elif proto_type == 3: # DRGBW led_count = (len(data) - 2) // 4 print(f"[SIM] DRGBW from {addr}: {led_count} LEDs (RGBW), timeout={timeout}s") elif proto_type == 4: # DNRGB start = struct.unpack_from(">H", data, 2)[0] led_count = (len(data) - 4) // 3 print(f"[SIM] DNRGB from {addr}: {led_count} LEDs starting at {start}") elif proto_type == 0xAC: # Animation command print(f"[SIM] ANIM_CMD from {addr}: {data[2:].hex()}") async def run_simulator(host: str = "0.0.0.0", port: int = 21324) -> None: loop = asyncio.get_running_loop() transport, _ = await loop.create_datagram_endpoint( SimulatorProtocol, local_addr=(host, port), ) print(f"[SIM] Listening on {host}:{port}") try: await asyncio.sleep(float("inf")) finally: transport.close() ``` ### Fire2012 Animation (Python Port) ```python # Source: FastLED Fire2012 algorithm (fastled.io/docs) import random import colorsys def fire_frame( heat: list[int], # mutable heat array, length = led_count led_count: int, cooling: int = 55, sparking: int = 120, ) -> list[tuple[int, int, int]]: """One Fire2012 step. Mutates heat array. Returns RGB pixel list.""" # Step 1: Cool every cell for i in range(led_count): cooldown = random.randint(0, (cooling * 10 // led_count) + 2) heat[i] = max(0, heat[i] - cooldown) # Step 2: Heat drifts up (index 0 = bottom) for i in range(led_count - 1, 1, -1): heat[i] = (heat[i - 1] + heat[i - 2] + heat[i - 2]) // 3 # Step 3: Randomly ignite new sparks near bottom if random.randint(0, 255) < sparking: y = random.randint(0, 6) if y < led_count: heat[y] = min(255, heat[y] + random.randint(160, 255)) # Step 4: Map heat to color (black → red → yellow → white) pixels = [] for h in heat: if h < 85: pixels.append((h * 3, 0, 0)) elif h < 170: h2 = h - 85 pixels.append((255, h2 * 3, 0)) else: h2 = h - 170 pixels.append((255, 255, h2 * 3)) return pixels ``` --- ## Existing Code Integration (from Phase 1-2) Phase 3 builds directly on these existing files — do not replace them: | File | What Already Exists | Phase 3 Action | |------|---------------------|----------------| | `lightsync/devices/base.py` | `BaseDevice` ABC with `encode_frame`, `encode_animation_cmd`, `bytes_per_pixel` | Add `send_frame(sender)` method OR leave I/O in UDPSender | | `lightsync/devices/sk6812.py` | `encode_frame()` working, `encode_animation_cmd()` returns `b""` | Fill `encode_animation_cmd()` using struct | | `lightsync/devices/ws2801.py` | Same as SK6812 — frame works, cmd stub | Same — fill cmd stub | | `lightsync/devices/registry.py` | `instantiate(device_id)` creates live device | Use this in UDPSender to get device object for encoding | | `lightsync/models/show.py` | `CueModel` has `animation: str`, `params: dict[str, Any]` | Animation params dict shape must match ANI-02 param schemas | | `lightsync/models/device.py` | `StripType = Literal["sk6812", "ws2801", "generic"]` | No change needed — strip type determines DRGB vs DRGBW | **Important:** `encode_animation_cmd` stub in SK6812Device/WS2801Device currently returns `b""`. Phase 3 replaces these stubs with real struct packing using the 0xAC animation command format. The `app.state` pattern (established in Phase 2 for engine/beats) should be extended to hold `app.state.udp_sender: UDPSender` started in the FastAPI lifespan handler. --- ## State of the Art | Old Approach | Current Approach | When Changed | Impact | |--------------|------------------|--------------|--------| | One socket per device | Single socket, sendto() per device | Always correct for UDP | Simpler, fewer OS resources | | Custom UDP protocol | WLED-compatible DRGB/DRGBW | WLED established ~2020 | Reuse WLED ecosystem clients; firmware Phase 7 has reference implementations | | Polling-based animation loop | Event-driven render-on-demand | N/A — this is Phase 3's new code | Render() called at cue time, not in background | **Deprecated/outdated:** - python-mpv: Removed in Phase 2 redesign — browser audio handles playback. No impact on Phase 3. --- ## Open Questions 1. **Animation command: should direction be in the packet or only in params dict?** - What we know: CueModel.params is `dict[str, Any]` — flexible - What's unclear: Whether firmware should get `reverse` as a packet field or we just send a second animation ID for reversed variant - Recommendation: Include `flags` byte (bit 0 = reverse) in animation command packet — avoids doubling animation IDs 2. **Should UDPSender retry on OS error?** - What we know: UDP is fire-and-forget; DatagramProtocol.error_received() is called on OS-level send errors - What's unclear: Whether LED strip devices on WiFi need any retry logic for dropped packets - Recommendation: No retry in Phase 3 — log errors only. Phase 5 can add retry if needed. Cue timing correctness matters more than delivery guarantee. 3. **Simulator: standalone script or integrated FastAPI endpoint?** - What we know: Phase 3 requirement UDP-04 says "software UDP receiver/simulator for testing without physical hardware" - What's unclear: Whether it should be a background task in the app or a separate runnable - Recommendation: Implement as both — a background asyncio task startable via `lightsync/protocol/simulator.py` as a `__main__` runnable AND optionally started via FastAPI on a debug endpoint. --- ## Environment Availability No external dependencies beyond stdlib asyncio. All required packages are already in pyproject.toml (fastapi, pydantic, structlog). No new packages needed for Phase 3. | Dependency | Required By | Available | Version | Fallback | |------------|------------|-----------|---------|----------| | asyncio stdlib | UDP transport | Yes | Python 3.11 stdlib | — | | struct stdlib | Packet encoding | Yes | Python 3.11 stdlib | — | | colorsys stdlib | Rainbow hue math | Yes | Python 3.11 stdlib | — | | UDP port 21324 | Simulator receiver | Likely free | — | Use any open port in tests | **Missing dependencies with no fallback:** None. --- ## Phase Requirements | ID | Description | Research Support | |----|-------------|------------------| | UDP-01 | Animation command packet: device ID, animation type, params — sent when block starts | Custom 0xAC binary packet format specified; struct encoding pattern documented | | UDP-02 | Raw pixel frame packet: WLED DRGB/DNRGB/DRGBW protocol — used for AI-generated sequences | Full WLED spec documented with byte layout, confirmed from official kno.wled.ge docs | | UDP-03 | Multi-packet framing for frames >MTU (SK6812 300 LEDs RGBW = 1200 bytes, exceeds WiFi MTU) | DNRGB chunked encoding documented; MTU math confirmed (489 px per packet) | | UDP-04 | Software UDP receiver/simulator for testing without physical hardware | asyncio DatagramProtocol pattern documented; simulator code example provided | | ANI-01 | Built-in animation types: chase, pulse, rainbow, strobe, color wipe, fire, solid color | All 7 types specified with render approach and parameter shapes | | ANI-02 | Per-animation parameters: speed, primary color(s), secondary color(s), direction, length/density | Full param shapes documented for all 7 animation types | | ANI-03 | RGBW-aware color handling for SK6812 (separate W channel) | DRGBW packet uses 4 bytes/pixel (R,G,B,W); `white` param included in all animation param shapes | | ANI-04 | RGB color handling for WS2801 | DRGB packet uses 3 bytes/pixel; existing `WS2801Device.encode_frame()` already correct — needs only dispatch to use type=0x02 | --- ## Sources ### Primary (HIGH confidence) - [kno.wled.ge/interfaces/udp-realtime/](https://kno.wled.ge/interfaces/udp-realtime/) — Full WLED UDP protocol spec, byte layouts, protocol types, max LED counts - [docs.python.org/3/library/asyncio-protocol.html](https://docs.python.org/3/library/asyncio-protocol.html) — DatagramProtocol, DatagramTransport.sendto() API and thread safety notes - [fastled.io/docs — Fire2012 algorithm](https://fastled.io/docs/df/d28/_fire2012_8ino-example.html) — Canonical fire animation: cooling, sparking, heat diffusion ### Secondary (MEDIUM confidence) - [GitHub WLED wiki UDP-Realtime-Control](https://github.com/Aircoookie/WLED/wiki/UDP-Realtime-Control) — Secondary source confirming protocol byte values - [WLED discourse: DRGB/DNRGB/WARLS differences](https://wled.discourse.group/t/drgb-dnrgb-warls-differences-best-set-up/688) — Community clarification on protocol selection - [CircuitPython LED Animation API](https://docs.circuitpython.org/projects/led-animation/en/latest/api.html) — Reference for animation parameter conventions (chase, color wipe, pulse, rainbow) ### Tertiary (LOW confidence) - WebSearch results on asyncio socket pool patterns — confirmed LOW: single socket with sendto() is the correct approach, no pool needed --- ## Metadata **Confidence breakdown:** - UDP wire protocol (WLED DRGB/DNRGB/DRGBW): HIGH — verified against official kno.wled.ge documentation - Animation command packet format (0xAC): MEDIUM — custom design, no external reference, but straightforward struct layout - Animation parameter shapes: MEDIUM — informed by Adafruit library conventions and Fire2012 spec, adapted to project requirements - Asyncio UDP patterns: HIGH — verified against Python stdlib docs - Fire animation algorithm: HIGH — FastLED Fire2012 is canonical reference implementation **Research date:** 2026-04-06 **Valid until:** 2026-10-06 (WLED protocol is stable; Python asyncio API is stable)