Gitea-Admin/led2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(07): create phase 7 microcontroller firmware plans

Browse Source 
Three sequential plans for MicroPython ESP32 firmware:
- 07-01: Scaffold (UDP listener, packet discriminator, LED drivers, config)
- 07-02: Animation renderer (all 7 types ported from lightsync)
- 07-03: Frame mode (DRGB/DRGBW/DNRGB parsing, reassembly, timeout)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-04-07 12:22:49 +00:00
parent b7b0673b43
commit 9a3370e71c
3 changed files with 1591 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

526
.planning/phases/07-microcontroller-firmware/07-01-PLAN.md Normal file
View File

@@ -0,0 +1,526 @@
# Plan 07-01: Firmware Scaffold
**Phase**: 7 — Microcontroller Firmware
**Goal**: A MicroPython firmware skeleton that connects to WiFi, listens for UDP packets, discriminates packet types, and drives SK6812 or WS2801 LED strips via a unified driver abstraction.
## Prompt
Create the firmware directory at `firmware/` in the project root. This is standalone MicroPython code for ESP32 (primary) / Pico W (secondary). It does NOT import from `lightsync/` — all code is self-contained.
### File 1: `firmware/config.json`
Create the default device configuration:
```json
{
"ssid": "MyWiFi",
"password": "secret",
"led_count": 300,
"strip_type": "sk6812",
"led_pin": 5,
"spi_id": 1,
"spi_clk_pin": 18,
"spi_dat_pin": 23,
"udp_port": 21324
}
```
- `strip_type`: either `"sk6812"` (RGBW, neopixel) or `"ws2801"` (RGB, SPI)
- `led_pin`: GPIO pin for neopixel data line (SK6812 only)
- `spi_id`, `spi_clk_pin`, `spi_dat_pin`: SPI config for WS2801 only
- `udp_port`: UDP listen port (WLED default 21324)
### File 2: `firmware/led_driver.py`
Create the LED strip driver abstraction. Two classes, no ABC (MicroPython has no `abc` module in standard builds):
```python
class NeoPixelStrip:
"""SK6812 RGBW driver using built-in neopixel module."""
def __init__(self, pin_num, led_count):
import neopixel
from machine import Pin
self.np = neopixel.NeoPixel(Pin(pin_num), led_count, bpp=4, timing=1)
self.led_count = led_count
def write_rgb(self, pixels):
"""Write list of (R,G,B) tuples. W channel set to 0."""
for i, (r, g, b) in enumerate(pixels):
if i >= self.led_count:
break
self.np[i] = (r, g, b, 0)
self.np.write()
def write_rgbw(self, pixels):
"""Write list of (R,G,B,W) tuples."""
for i, (r, g, b, w) in enumerate(pixels):
if i >= self.led_count:
break
self.np[i] = (r, g, b, w)
self.np.write()
def fill(self, r, g, b, w=0):
"""Fill all pixels with one color."""
self.np.fill((r, g, b, w))
self.np.write()
def clear(self):
self.fill(0, 0, 0, 0)
class WS2801Strip:
"""WS2801 RGB driver using SPI — no external library needed."""
def __init__(self, spi_id, clk_pin, dat_pin, led_count):
from machine import SPI, Pin
self.spi = SPI(spi_id, baudrate=1000000, polarity=0, phase=0,
sck=Pin(clk_pin), mosi=Pin(dat_pin))
self.led_count = led_count
self._buf = bytearray(led_count * 3)
def write_rgb(self, pixels):
"""Write list of (R,G,B) tuples."""
for i, (r, g, b) in enumerate(pixels):
if i >= self.led_count:
break
off = i * 3
self._buf[off] = r
self._buf[off + 1] = g
self._buf[off + 2] = b
self.spi.write(self._buf)
def write_rgbw(self, pixels):
"""Write RGBW tuples — W channel is discarded (WS2801 is RGB only)."""
for i, (r, g, b, _w) in enumerate(pixels):
if i >= self.led_count:
break
off = i * 3
self._buf[off] = r
self._buf[off + 1] = g
self._buf[off + 2] = b
self.spi.write(self._buf)
def fill(self, r, g, b, w=0):
for i in range(self.led_count):
off = i * 3
self._buf[off] = r
self._buf[off + 1] = g
self._buf[off + 2] = b
self.spi.write(self._buf)
def clear(self):
self.fill(0, 0, 0)
def create_strip(config):
"""Factory: create the right strip driver from config dict."""
if config['strip_type'] == 'sk6812':
return NeoPixelStrip(config['led_pin'], config['led_count'])
elif config['strip_type'] == 'ws2801':
return WS2801Strip(config['spi_id'], config.get('spi_clk_pin', 18),
config.get('spi_dat_pin', 23), config['led_count'])
else:
raise ValueError('Unknown strip_type: ' + config['strip_type'])
```
Both classes MUST implement the same interface: `write_rgb(pixels)`, `write_rgbw(pixels)`, `fill(r,g,b,w)`, `clear()`. This is the contract that `main.py` and animation/frame code will use.
### File 3: `firmware/protocol.py`
Create packet parsers that decode the exact binary formats produced by `lightsync/protocol/drgb.py` and `lightsync/protocol/animation_cmd.py`. These must be byte-exact mirrors of the Python encoders.
```python
import struct
# Protocol type bytes (first byte of UDP packet)
PROTO_DRGB = 0x02
PROTO_DRGBW = 0x03
PROTO_DNRGB = 0x04
PROTO_ANIM_CMD = 0xAC
ANIMATION_NAMES = {
0: 'solid_color', 1: 'chase', 2: 'pulse',
3: 'rainbow', 4: 'strobe', 5: 'color_wipe', 6: 'fire'
}
def classify_packet(data):
"""Return packet type string from first byte, or None if unknown."""
if not data:
return None
b = data[0]
if b == PROTO_DRGB:
return 'drgb'
elif b == PROTO_DRGBW:
return 'drgbw'
elif b == PROTO_DNRGB:
return 'dnrgb'
elif b == PROTO_ANIM_CMD:
return 'anim_cmd'
return None
def parse_drgb(data):
"""Parse DRGB packet -> list of (R,G,B) tuples.
Format: [0x02][timeout][R0][G0][B0]...[Rn][Gn][Bn]
"""
n = (len(data) - 2) // 3
pixels = []
for i in range(n):
off = 2 + i * 3
pixels.append((data[off], data[off + 1], data[off + 2]))
return pixels
def parse_drgbw(data):
"""Parse DRGBW packet -> list of (R,G,B,W) tuples.
Format: [0x03][timeout][R0][G0][B0][W0]...[Rn][Gn][Bn][Wn]
"""
n = (len(data) - 2) // 4
pixels = []
for i in range(n):
off = 2 + i * 4
pixels.append((data[off], data[off + 1], data[off + 2], data[off + 3]))
return pixels
def parse_dnrgb(data):
"""Parse DNRGB packet -> (start_index, list of (R,G,B) tuples).
Format: [0x04][timeout][start_hi][start_lo][R0][G0][B0]...
Start index is big-endian uint16.
"""
start_index = struct.unpack('>H', data[2:4])[0]
n = (len(data) - 4) // 3
pixels = []
for i in range(n):
off = 4 + i * 3
pixels.append((data[off], data[off + 1], data[off + 2]))
return start_index, pixels
def parse_animation_cmd(data):
"""Parse animation command packet -> dict with animation name and params.
Format (16 bytes, big-endian):
Byte 0: 0xAC marker
Byte 1: version (1)
Byte 2: animation type ID (0-6)
Byte 3: flags (bit 0 = reverse)
Bytes 4-7: speed (float32 big-endian)
Bytes 8-10: primary color R, G, B
Bytes 11-13: background color R, G, B
Byte 14: white channel
Byte 15: density/size/cooling
"""
if len(data) < 16 or data[0] != PROTO_ANIM_CMD:
return None
anim_id = data[2]
flags = data[3]
speed = struct.unpack('>f', data[4:8])[0]
anim_name = ANIMATION_NAMES.get(anim_id)
if anim_name is None:
return None
return {
'animation': anim_name,
'speed': speed,
'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),
}
```
### File 4: `firmware/boot.py`
WiFi connection on startup. This runs before `main.py` automatically on MicroPython.
```python
import network
import time
import json
def load_config():
try:
with open('config.json') as f:
return json.load(f)
except OSError:
print('[boot] config.json not found, using defaults')
return {
'ssid': 'MyWiFi', 'password': 'secret',
'led_count': 300, 'strip_type': 'sk6812',
'led_pin': 5, 'spi_id': 1, 'udp_port': 21324
}
def connect_wifi(ssid, password, timeout=15):
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if wlan.isconnected():
print('[boot] Already connected:', wlan.ifconfig()[0])
return wlan
print('[boot] Connecting to', ssid, '...')
wlan.connect(ssid, password)
start = time.time()
while not wlan.isconnected():
if time.time() - start > timeout:
print('[boot] WiFi timeout after', timeout, 'seconds')
return wlan
time.sleep(0.5)
print('[boot] Connected:', wlan.ifconfig()[0])
return wlan
config = load_config()
wlan = connect_wifi(config['ssid'], config['password'])
```
### File 5: `firmware/main.py`
The main asyncio loop with two tasks: UDP listener and animation render loop. This is the scaffold only — animation rendering and frame passthrough will be added by plans 07-02 and 07-03.
```python
import asyncio
import socket
import select
import time
import json
# Import firmware modules
from protocol import classify_packet, parse_animation_cmd
from led_driver import create_strip
# --- Global state ---
config = None
strip = None
current_mode = 'idle' # 'idle', 'animation', 'frame'
current_animation = None # dict from parse_animation_cmd
animation_start_ms = 0
frame_pixels = None # set by frame packets
def load_config():
try:
with open('config.json') as f:
return json.load(f)
except OSError:
return {'led_count': 300, 'strip_type': 'sk6812', 'led_pin': 5,
'spi_id': 1, 'udp_port': 21324}
async def udp_listener_task(port):
"""Non-blocking UDP listener using select.poll(0) inside asyncio."""
global current_mode, current_animation, animation_start_ms, frame_pixels
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('0.0.0.0', port))
sock.setblocking(False)
poller = select.poll()
poller.register(sock, select.POLLIN)
print('[udp] Listening on port', port)
while True:
events = poller.poll(0)
if events:
try:
data, addr = sock.recvfrom(1500)
ptype = classify_packet(data)
if ptype == 'anim_cmd':
cmd = parse_animation_cmd(data)
if cmd:
current_animation = cmd
current_mode = 'animation'
animation_start_ms = time.ticks_ms()
print('[udp] Animation:', cmd['animation'])
elif ptype in ('drgb', 'drgbw', 'dnrgb'):
# Frame handling added by Plan 07-03
current_mode = 'frame'
print('[udp] Frame packet:', ptype, len(data), 'bytes')
else:
print('[udp] Unknown packet type:', data[0] if data else '?')
except OSError:
pass
await asyncio.sleep_ms(5)
async def render_task():
"""Animation render loop at ~60fps. Rendering logic added by Plans 07-02/07-03."""
global current_mode, frame_pixels
while True:
if current_mode == 'animation' and current_animation:
# Plan 07-02 adds: render_animation() call here
pass
elif current_mode == 'frame' and frame_pixels:
# Plan 07-03 adds: write frame_pixels to strip here
pass
await asyncio.sleep_ms(16) # ~60 fps
async def main():
global config, strip
config = load_config()
strip = create_strip(config)
strip.clear()
print('[main] LightSync firmware ready')
print('[main] Strip:', config['strip_type'], '/', config['led_count'], 'LEDs')
asyncio.create_task(udp_listener_task(config['udp_port']))
asyncio.create_task(render_task())
# Run forever
while True:
await asyncio.sleep(1)
# Entry point
asyncio.run(main())
```
**CRITICAL**: The `await asyncio.sleep_ms(5)` in the UDP listener is essential. Without it, the listener monopolizes MicroPython's cooperative scheduler and the render task never runs.
### File 6: `tests/test_firmware_protocol.py`
Create host-side tests that validate the firmware protocol parsers produce correct output for packets generated by the existing Python encoders. These tests run on the development machine (regular Python, not MicroPython), importing `firmware/protocol.py` directly since it uses only stdlib modules (`struct`).
```python
"""Host-side tests for firmware protocol parsers.
These tests verify that firmware/protocol.py correctly decodes packets
produced by lightsync/protocol/drgb.py and lightsync/protocol/animation_cmd.py.
Run with: python -m pytest tests/test_firmware_protocol.py -v
"""
import sys
import os
import pytest
# Add firmware/ to path so we can import protocol.py
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'firmware'))
from protocol import classify_packet, parse_drgb, parse_drgbw, parse_dnrgb, parse_animation_cmd
# Import the server-side encoders for round-trip testing
from lightsync.protocol.drgb import encode_drgb, encode_drgbw, encode_dnrgb_packets
from lightsync.protocol.animation_cmd import encode_animation_cmd
class TestClassifyPacket:
def test_drgb(self):
assert classify_packet(bytes([0x02, 0x00])) == 'drgb'
def test_drgbw(self):
assert classify_packet(bytes([0x03, 0x00])) == 'drgbw'
def test_dnrgb(self):
assert classify_packet(bytes([0x04, 0x00])) == 'dnrgb'
def test_anim_cmd(self):
assert classify_packet(bytes([0xAC, 0x00])) == 'anim_cmd'
def test_unknown(self):
assert classify_packet(bytes([0xFF])) is None
def test_empty(self):
assert classify_packet(b'') is None
class TestDRGBRoundTrip:
def test_simple_pixels(self):
pixels = [(255, 0, 0), (0, 255, 0), (0, 0, 255)]
packet = encode_drgb(pixels)
decoded = parse_drgb(packet)
assert decoded == pixels
def test_300_leds(self):
pixels = [(i % 256, (i * 2) % 256, (i * 3) % 256) for i in range(300)]
packet = encode_drgb(pixels)
decoded = parse_drgb(packet)
assert len(decoded) == 300
assert decoded[0] == pixels[0]
assert decoded[299] == pixels[299]
class TestDRGBWRoundTrip:
def test_simple_pixels(self):
pixels = [(255, 0, 0, 128), (0, 255, 0, 64)]
packet = encode_drgbw(pixels)
decoded = parse_drgbw(packet)
assert decoded == pixels
def test_300_leds(self):
pixels = [(i % 256, (i * 2) % 256, (i * 3) % 256, (i * 4) % 256) for i in range(300)]
packet = encode_drgbw(pixels)
decoded = parse_drgbw(packet)
assert len(decoded) == 300
assert decoded[0] == pixels[0]
class TestDNRGBRoundTrip:
def test_single_packet(self):
pixels = [(255, 0, 0)] * 100
packets = encode_dnrgb_packets(pixels)
assert len(packets) == 1
start_idx, decoded = parse_dnrgb(packets[0])
assert start_idx == 0
assert len(decoded) == 100
def test_multi_packet(self):
pixels = [(i % 256, 0, 0) for i in range(600)]
packets = encode_dnrgb_packets(pixels)
assert len(packets) == 2
start0, px0 = parse_dnrgb(packets[0])
start1, px1 = parse_dnrgb(packets[1])
assert start0 == 0
assert start1 == 489
assert len(px0) == 489
assert len(px1) == 111
class TestAnimationCmdRoundTrip:
@pytest.mark.parametrize("anim_name", [
'solid_color', 'chase', 'pulse', 'rainbow', 'strobe', 'color_wipe', 'fire'
])
def test_all_animations(self, anim_name):
params = {'color': [255, 128, 0], 'speed': 0.7, 'white': 32,
'bg_color': [10, 20, 30], 'reverse': True, 'density': 42}
packet = encode_animation_cmd(anim_name, params)
decoded = parse_animation_cmd(packet)
assert decoded is not None
assert decoded['animation'] == anim_name
assert decoded['color'] == (255, 128, 0)
assert decoded['bg_color'] == (10, 20, 30)
assert decoded['white'] == 32
assert decoded['reverse'] is True
assert abs(decoded['speed'] - 0.7) < 0.001
def test_invalid_marker(self):
assert parse_animation_cmd(bytes(16)) is None
def test_too_short(self):
assert parse_animation_cmd(bytes([0xAC, 0x01])) is None
```
These tests are the primary verification mechanism. They prove that firmware parsers correctly decode packets from the server-side encoders — the exact round-trip that must work in production.
## Verification
- [ ] `firmware/` directory exists with 5 files: `boot.py`, `main.py`, `config.json`, `led_driver.py`, `protocol.py`
- [ ] `python -m pytest tests/test_firmware_protocol.py -v` passes all tests (run from project root)
- [ ] `firmware/protocol.py` correctly round-trips all 4 packet types against server-side encoders
- [ ] `firmware/led_driver.py` has both `NeoPixelStrip` and `WS2801Strip` classes with identical interface (`write_rgb`, `write_rgbw`, `fill`, `clear`)
- [ ] `firmware/main.py` has two asyncio tasks (`udp_listener_task` and `render_task`) with proper `await asyncio.sleep_ms()` yields
- [ ] `firmware/boot.py` reads `config.json` and connects to WiFi
- [ ] `firmware/config.json` has all 9 fields: ssid, password, led_count, strip_type, led_pin, spi_id, spi_clk_pin, spi_dat_pin, udp_port
- [ ] Packet discriminator handles all 4 types: 0x02 (DRGB), 0x03 (DRGBW), 0x04 (DNRGB), 0xAC (animation cmd)
## Notes
- The `led_driver.py` module uses `machine` and `neopixel` imports which only exist on MicroPython. The host-side tests do NOT test `led_driver.py` — that is device-only verification. Tests focus on `protocol.py` which is pure Python and runs on both CPython and MicroPython.
- The `main.py` render_task and frame handling are intentionally stubbed. Plan 07-02 fills in animation rendering, Plan 07-03 fills in frame passthrough.
- WS2801 `write_rgbw` discards the W channel since WS2801 is RGB-only. This is correct — the protocol may send DRGBW packets to any device, and the driver silently drops W.
- Use `broad except OSError` for socket errors, not specific `EAGAIN` — MicroPython on ESP8266 raises `ETIMEDOUT` instead of `EAGAIN` for non-blocking recv.

608
.planning/phases/07-microcontroller-firmware/07-02-PLAN.md Normal file
View File

@@ -0,0 +1,608 @@
# Plan 07-02: Animation Command Renderer
**Phase**: 7 — Microcontroller Firmware
**Goal**: The firmware receives animation command packets (0xAC) and renders all 7 animation types locally on the LED strip at ~60fps.
## Prompt
This plan builds on 07-01 (complete). The firmware scaffold (`firmware/main.py`, `firmware/protocol.py`, `firmware/led_driver.py`) already exists with a working UDP listener and packet discriminator. Now add the animation rendering engine.
### File 1: `firmware/animations.py`
Create a single file containing all 7 animations ported from `lightsync/animations/*.py` to MicroPython. Key porting changes from the CPython originals:
1. **No imports from `lightsync`** — this is standalone MicroPython code
2. **No `abc` module** — use plain classes, no `AnimationBase` ABC
3. **No `colorsys`** — inline HSV-to-RGB for rainbow (MicroPython does not have `colorsys`)
4. **No type hints** — MicroPython handles them but they add no value on device
5. **Pre-allocate buffers** for fire animation to avoid GC pressure
6. **`time.ticks_ms()`** instead of `time.time()` for animation timing
Here is the exact code to create. Each animation is a direct port of the corresponding file in `lightsync/animations/`:
```python
"""All 7 LightSync animations for MicroPython firmware.
Ported from lightsync/animations/*.py — same math, MicroPython-compatible.
Each animation has render(t, led_count) -> list of (R,G,B) tuples.
"""
import math
import random
def _hsv_to_rgb(h, s, v):
"""HSV to RGB conversion — replaces colorsys.hsv_to_rgb (not in MicroPython)."""
if s == 0.0:
c = int(v * 255)
return c, c, c
i = int(h * 6.0)
f = (h * 6.0) - i
p = v * (1.0 - s)
q = v * (1.0 - s * f)
t = v * (1.0 - s * (1.0 - f))
i = i % 6
if i == 0:
return int(v * 255), int(t * 255), int(p * 255)
if i == 1:
return int(q * 255), int(v * 255), int(p * 255)
if i == 2:
return int(p * 255), int(v * 255), int(t * 255)
if i == 3:
return int(p * 255), int(q * 255), int(v * 255)
if i == 4:
return int(t * 255), int(p * 255), int(v * 255)
return int(v * 255), int(p * 255), int(q * 255)
class SolidColorAnimation:
def __init__(self, color, white=0):
self.color = tuple(color)
self.white = white
def render(self, t, led_count):
return [self.color] * led_count
class ChaseAnimation:
def __init__(self, color, bg_color, speed, size, spacing, reverse, white):
self.color = tuple(color)
self.bg_color = tuple(bg_color)
self.speed = speed
self.size = size
self.spacing = spacing
self.reverse = reverse
self.white = white
def render(self, t, led_count):
period = self.size + self.spacing
offset = int(t * self.speed * 60)
pixels = []
for i in range(led_count):
if self.reverse:
pos = (i + offset) % period
else:
pos = (i - offset) % period
if pos < self.size:
pixels.append(self.color)
else:
pixels.append(self.bg_color)
return pixels
class PulseAnimation:
def __init__(self, color, speed, min_brightness, max_brightness, white):
self.color = tuple(color)
self.speed = speed
self.min_brightness = min_brightness
self.max_brightness = max_brightness
self.white = white
def render(self, t, led_count):
period = 0.1 + (1.0 - self.speed) * 4.9
brightness = (
self.min_brightness
+ (self.max_brightness - self.min_brightness)
* (0.5 + 0.5 * math.sin(2 * math.pi * t / period))
)
scale = brightness / 255.0
r = min(255, max(0, int(self.color[0] * scale)))
g = min(255, max(0, int(self.color[1] * scale)))
b = min(255, max(0, int(self.color[2] * scale)))
pixel = (r, g, b)
return [pixel] * led_count
class RainbowAnimation:
def __init__(self, speed, period):
self.speed = speed
self.period = period
def render(self, t, led_count):
pixels = []
for i in range(led_count):
hue = (i / led_count / self.period + t * self.speed) % 1.0
r, g, b = _hsv_to_rgb(hue, 1.0, 1.0)
pixels.append((r, g, b))
return pixels
class StrobeAnimation:
def __init__(self, color, speed, duty_cycle, white):
self.color = tuple(color)
self.speed = speed
self.duty_cycle = duty_cycle
self.white = white
def render(self, t, led_count):
freq = 1.0 + self.speed * 19.0
phase = (t * freq) % 1.0
if phase < self.duty_cycle:
pixel = self.color
else:
pixel = (0, 0, 0)
return [pixel] * led_count
class ColorWipeAnimation:
def __init__(self, color, speed, reverse, white):
self.color = tuple(color)
self.speed = speed
self.reverse = reverse
self.white = white
def render(self, t, led_count):
fill_count = min(led_count, int(t * self.speed * 60))
black = (0, 0, 0)
if not self.reverse:
pixels = [self.color] * fill_count + [black] * (led_count - fill_count)
else:
pixels = [black] * (led_count - fill_count) + [self.color] * fill_count
return pixels
class FireAnimation:
def __init__(self, cooling, sparking, speed):
self.cooling = cooling
self.sparking = sparking
self.speed = speed
self._heat = None # bytearray, lazy-init
self._led_count = 0
def render(self, t, led_count):
# Lazy-init or resize heat array (bytearray to avoid GC pressure)
if self._heat is None or self._led_count != led_count:
self._heat = bytearray(led_count)
self._led_count = led_count
steps = max(1, int(self.speed * 3))
heat = self._heat
for _ in range(steps):
# Step 1: Cool every cell
for i in range(led_count):
cooldown = random.randint(0, (self.cooling * 10 // led_count) + 2)
v = heat[i] - cooldown
heat[i] = v if v > 0 else 0
# Step 2: Heat drifts upward
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 sparks near bottom
if random.randint(0, 255) < self.sparking:
y = random.randint(0, min(6, led_count - 1))
v = heat[y] + random.randint(160, 255)
heat[y] = v if v < 255 else 255
# Map heat to color
pixels = []
for i in range(led_count):
h = heat[i]
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
c = h2 * 3
pixels.append((255, 255, c if c < 255 else 255))
return pixels
# Animation factory — maps names to constructors with default params
def create_animation(cmd):
"""Create animation instance from parsed animation command dict.
Args:
cmd: dict from parse_animation_cmd() with keys:
animation, speed, color, bg_color, white, density, reverse
Returns:
Animation instance with render(t, led_count) method.
"""
name = cmd['animation']
color = cmd['color'] # (R, G, B) tuple
bg_color = cmd['bg_color'] # (R, G, B) tuple
speed = cmd['speed']
white = cmd['white']
density = cmd['density']
reverse = cmd['reverse']
if name == 'solid_color':
return SolidColorAnimation(color, white)
elif name == 'chase':
return ChaseAnimation(color, bg_color, speed, density or 3, 7, reverse, white)
elif name == 'pulse':
return PulseAnimation(color, speed, 0, 255, white)
elif name == 'rainbow':
return RainbowAnimation(speed, 1.0)
elif name == 'strobe':
return StrobeAnimation(color, speed, 0.1, white)
elif name == 'color_wipe':
return ColorWipeAnimation(color, speed, reverse, white)
elif name == 'fire':
return FireAnimation(density or 55, 120, speed)
else:
print('[anim] Unknown animation:', name)
return SolidColorAnimation(color, white) # fallback
```
**IMPORTANT**: The `chase` animation's `density` field maps to `size` (per the animation_cmd.py protocol where byte 15 is "density/size/cooling"). For `fire`, `density` maps to `cooling`. For `chase`, if density is 0 (default), use 3 as the default size.
### File 2: Update `firmware/main.py`
Modify `main.py` to wire the animation rendering into the render loop. Changes:
1. Add import: `from animations import create_animation`
2. Add a global: `current_anim_instance = None` (the animation object, not just the raw cmd dict)
3. In the `udp_listener_task`, when an animation command is received, create the animation instance:
```python
current_animation = cmd
current_anim_instance = create_animation(cmd)
current_mode = 'animation'
animation_start_ms = time.ticks_ms()
```
4. In `render_task`, fill in the animation branch:
```python
if current_mode == 'animation' and current_anim_instance:
t = time.ticks_diff(time.ticks_ms(), animation_start_ms) / 1000.0
pixels = current_anim_instance.render(t, config['led_count'])
strip.write_rgb(pixels)
```
The complete updated `main.py` should look like:
```python
import asyncio
import socket
import select
import time
import json
from protocol import classify_packet, parse_animation_cmd
from led_driver import create_strip
from animations import create_animation
# --- Global state ---
config = None
strip = None
current_mode = 'idle' # 'idle', 'animation', 'frame'
current_animation = None # dict from parse_animation_cmd
current_anim_instance = None # animation object with render()
animation_start_ms = 0
frame_pixels = None # set by frame packets
def load_config():
try:
with open('config.json') as f:
return json.load(f)
except OSError:
return {'led_count': 300, 'strip_type': 'sk6812', 'led_pin': 5,
'spi_id': 1, 'udp_port': 21324}
async def udp_listener_task(port):
"""Non-blocking UDP listener using select.poll(0) inside asyncio."""
global current_mode, current_animation, current_anim_instance
global animation_start_ms, frame_pixels
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('0.0.0.0', port))
sock.setblocking(False)
poller = select.poll()
poller.register(sock, select.POLLIN)
print('[udp] Listening on port', port)
while True:
events = poller.poll(0)
if events:
try:
data, addr = sock.recvfrom(1500)
ptype = classify_packet(data)
if ptype == 'anim_cmd':
cmd = parse_animation_cmd(data)
if cmd:
current_animation = cmd
current_anim_instance = create_animation(cmd)
current_mode = 'animation'
animation_start_ms = time.ticks_ms()
print('[udp] Animation:', cmd['animation'])
elif ptype in ('drgb', 'drgbw', 'dnrgb'):
# Frame handling added by Plan 07-03
current_mode = 'frame'
print('[udp] Frame packet:', ptype, len(data), 'bytes')
else:
print('[udp] Unknown packet type:', data[0] if data else '?')
except OSError:
pass
await asyncio.sleep_ms(5)
async def render_task():
"""Animation render loop at ~60fps."""
global current_mode, frame_pixels
while True:
if current_mode == 'animation' and current_anim_instance:
t = time.ticks_diff(time.ticks_ms(), animation_start_ms) / 1000.0
pixels = current_anim_instance.render(t, config['led_count'])
strip.write_rgb(pixels)
elif current_mode == 'frame' and frame_pixels:
# Plan 07-03 adds frame write logic here
pass
await asyncio.sleep_ms(16) # ~60 fps
async def main():
global config, strip
config = load_config()
strip = create_strip(config)
strip.clear()
print('[main] LightSync firmware ready')
print('[main] Strip:', config['strip_type'], '/', config['led_count'], 'LEDs')
asyncio.create_task(udp_listener_task(config['udp_port']))
asyncio.create_task(render_task())
while True:
await asyncio.sleep(1)
asyncio.run(main())
```
### File 3: `tests/test_firmware_animations.py`
Create host-side tests that verify the MicroPython animation code produces correct output. These tests import `firmware/animations.py` directly (it uses only `math` and `random`, both available in CPython).
```python
"""Host-side tests for firmware animation renderers.
Verifies that the MicroPython animation ports produce correct output.
Run with: python -m pytest tests/test_firmware_animations.py -v
"""
import sys
import os
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'firmware'))
from animations import (
SolidColorAnimation, ChaseAnimation, PulseAnimation, RainbowAnimation,
StrobeAnimation, ColorWipeAnimation, FireAnimation, create_animation,
_hsv_to_rgb
)
class TestHSVtoRGB:
def test_red(self):
r, g, b = _hsv_to_rgb(0.0, 1.0, 1.0)
assert (r, g, b) == (255, 0, 0)
def test_green(self):
r, g, b = _hsv_to_rgb(1/3, 1.0, 1.0)
assert r == 0 and g == 255 and b == 0
def test_blue(self):
r, g, b = _hsv_to_rgb(2/3, 1.0, 1.0)
assert r == 0 and g == 0 and b == 255
def test_white(self):
r, g, b = _hsv_to_rgb(0.0, 0.0, 1.0)
assert (r, g, b) == (255, 255, 255)
def test_black(self):
r, g, b = _hsv_to_rgb(0.0, 0.0, 0.0)
assert (r, g, b) == (0, 0, 0)
class TestSolidColor:
def test_uniform(self):
anim = SolidColorAnimation((255, 0, 0))
pixels = anim.render(0.0, 10)
assert len(pixels) == 10
assert all(p == (255, 0, 0) for p in pixels)
def test_time_independent(self):
anim = SolidColorAnimation((0, 255, 0))
assert anim.render(0.0, 5) == anim.render(100.0, 5)
class TestChase:
def test_length(self):
anim = ChaseAnimation((255, 0, 0), (0, 0, 0), 0.5, 3, 7, False, 0)
pixels = anim.render(0.0, 30)
assert len(pixels) == 30
def test_has_both_colors(self):
anim = ChaseAnimation((255, 0, 0), (0, 0, 0), 0.5, 3, 7, False, 0)
pixels = anim.render(0.0, 30)
assert (255, 0, 0) in pixels
assert (0, 0, 0) in pixels
def test_reverse_differs(self):
fwd = ChaseAnimation((255, 0, 0), (0, 0, 0), 0.5, 3, 7, False, 0)
rev = ChaseAnimation((255, 0, 0), (0, 0, 0), 0.5, 3, 7, True, 0)
# At t=0.3 (not a multiple of period), forward and reverse differ
assert fwd.render(0.3, 20) != rev.render(0.3, 20)
class TestPulse:
def test_length(self):
anim = PulseAnimation((255, 255, 255), 0.5, 0, 255, 0)
pixels = anim.render(0.0, 10)
assert len(pixels) == 10
def test_values_in_range(self):
anim = PulseAnimation((255, 128, 64), 0.5, 0, 255, 0)
for t in [0.0, 0.5, 1.0, 2.5]:
pixels = anim.render(t, 10)
for r, g, b in pixels:
assert 0 <= r <= 255
assert 0 <= g <= 255
assert 0 <= b <= 255
class TestRainbow:
def test_length(self):
anim = RainbowAnimation(0.5, 1.0)
pixels = anim.render(0.0, 10)
assert len(pixels) == 10
def test_distinct_hues(self):
anim = RainbowAnimation(0.5, 1.0)
pixels = anim.render(0.0, 10)
# All 10 pixels should have distinct colors
unique = set(pixels)
assert len(unique) == 10
def test_values_in_range(self):
anim = RainbowAnimation(0.5, 1.0)
pixels = anim.render(0.0, 300)
for r, g, b in pixels:
assert 0 <= r <= 255
assert 0 <= g <= 255
assert 0 <= b <= 255
class TestStrobe:
def test_length(self):
anim = StrobeAnimation((255, 255, 255), 0.5, 0.5, 0)
pixels = anim.render(0.0, 10)
assert len(pixels) == 10
def test_on_phase(self):
# At t=0.0, phase = 0.0 < duty_cycle=0.5, should be ON
anim = StrobeAnimation((255, 255, 255), 0.5, 0.5, 0)
pixels = anim.render(0.0, 5)
assert all(p == (255, 255, 255) for p in pixels)
class TestColorWipe:
def test_length(self):
anim = ColorWipeAnimation((255, 0, 0), 0.5, False, 0)
pixels = anim.render(0.0, 10)
assert len(pixels) == 10
def test_progressive(self):
anim = ColorWipeAnimation((255, 0, 0), 1.0, False, 0)
p1 = anim.render(0.5, 100)
p2 = anim.render(1.0, 100)
filled1 = sum(1 for p in p1 if p == (255, 0, 0))
filled2 = sum(1 for p in p2 if p == (255, 0, 0))
assert filled2 >= filled1
def test_at_zero(self):
anim = ColorWipeAnimation((255, 0, 0), 0.5, False, 0)
pixels = anim.render(0.0, 10)
assert all(p == (0, 0, 0) for p in pixels)
class TestFire:
def test_length(self):
anim = FireAnimation(55, 120, 0.5)
pixels = anim.render(0.0, 30)
assert len(pixels) == 30
def test_values_in_range(self):
anim = FireAnimation(55, 120, 0.5)
# Run several frames to build up heat
for i in range(10):
pixels = anim.render(i * 0.016, 30)
for r, g, b in pixels:
assert 0 <= r <= 255
assert 0 <= g <= 255
assert 0 <= b <= 255
def test_uses_bytearray(self):
anim = FireAnimation(55, 120, 0.5)
anim.render(0.0, 30)
assert isinstance(anim._heat, bytearray)
class TestCreateAnimation:
def test_all_types(self):
for name in ['solid_color', 'chase', 'pulse', 'rainbow', 'strobe', 'color_wipe', 'fire']:
cmd = {
'animation': name, 'speed': 0.5,
'color': (255, 0, 0), 'bg_color': (0, 0, 0),
'white': 0, 'density': 0, 'reverse': False
}
anim = create_animation(cmd)
pixels = anim.render(0.5, 10)
assert len(pixels) == 10
def test_unknown_falls_back(self):
cmd = {
'animation': 'nonexistent', 'speed': 0.5,
'color': (255, 0, 0), 'bg_color': (0, 0, 0),
'white': 0, 'density': 0, 'reverse': False
}
anim = create_animation(cmd)
# Falls back to SolidColor
assert isinstance(anim, SolidColorAnimation)
class TestRoundTripWithServerEncoders:
"""Verify firmware animations can be created from server-encoded packets."""
def test_encode_decode_create_render(self):
"""Full round-trip: server encode -> firmware decode -> create animation -> render."""
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from lightsync.protocol.animation_cmd import encode_animation_cmd
# Also need firmware protocol parser
from protocol import parse_animation_cmd as fw_parse
for anim_name in ['solid_color', 'chase', 'pulse', 'rainbow', 'strobe', 'color_wipe', 'fire']:
params = {'color': [255, 128, 0], 'speed': 0.7, 'white': 32}
packet = encode_animation_cmd(anim_name, params)
cmd = fw_parse(packet)
assert cmd is not None
anim = create_animation(cmd)
pixels = anim.render(0.5, 10)
assert len(pixels) == 10
```
## Verification
- [ ] `firmware/animations.py` exists with all 7 animation classes + `_hsv_to_rgb` + `create_animation` factory
- [ ] `python -m pytest tests/test_firmware_animations.py -v` passes all tests
- [ ] `_hsv_to_rgb` produces correct values for primary colors (red=0.0, green=0.333, blue=0.667)
- [ ] Fire animation uses `bytearray` for heat buffer (not `list`)
- [ ] Rainbow animation uses inline `_hsv_to_rgb` (no `colorsys` import)
- [ ] `create_animation()` handles all 7 animation names from parsed command dicts
- [ ] Full round-trip test passes: server `encode_animation_cmd` -> firmware `parse_animation_cmd` -> `create_animation` -> `render` -> correct pixel output
- [ ] `firmware/main.py` is updated to call `create_animation()` on anim_cmd packets and `render()` in the render loop
- [ ] No `colorsys`, no `abc`, no `lightsync` imports in any `firmware/` file
## Notes
- The animation math is identical to `lightsync/animations/*.py` — same formulas, same defaults. The only changes are: removing type hints, removing ABC inheritance, inlining HSV conversion, and using `bytearray` for fire heat.
- The `density` field from the protocol packet (byte 15) maps to different params per animation: `size` for chase, `cooling` for fire, ignored for others. The `create_animation()` factory handles this mapping.
- All animations output `(R,G,B)` tuples even for SK6812 RGBW strips. The `NeoPixelStrip.write_rgb()` driver method appends W=0 automatically. This is correct — animations operate in RGB space. The `white` field from the command is stored but not used in rendering (it's available for future RGBW-native animations).
- `time.ticks_ms()` and `time.ticks_diff()` are MicroPython-specific but `time.ticks_ms` is not available in CPython. The host tests only test the animation `render()` method which takes `t` as a float parameter — no `ticks_ms` dependency in the tested code path.

457
.planning/phases/07-microcontroller-firmware/07-03-PLAN.md Normal file
View File

@@ -0,0 +1,457 @@
# Plan 07-03: Frame Mode Renderer
**Phase**: 7 — Microcontroller Firmware
**Goal**: The firmware receives DRGB/DRGBW/DNRGB raw frame packets and outputs correct pixel values to the physical LED strip, with multi-packet reassembly for DNRGB and automatic fallback to animation mode when frames stop arriving.
## Prompt
This plan builds on 07-01 and 07-02 (both complete). The firmware has a working UDP listener, packet discriminator, animation renderer, and LED strip drivers. Now add raw frame mode: receiving pixel data from the server and writing it directly to the strip.
### Update 1: `firmware/protocol.py` — Add DNRGB reassembly helper
Add a `FrameAssembler` class at the bottom of `firmware/protocol.py` that handles multi-packet DNRGB reassembly. This is needed for strips > 489 LEDs (RGB) where one frame spans multiple UDP packets.
Add this code to the end of the existing `firmware/protocol.py`:
```python
class FrameAssembler:
"""Reassembles multi-packet DNRGB frames.
For 300-LED strips, frames fit in one packet so reassembly never triggers.
Implemented for completeness and future strip sizes.
"""
def __init__(self, expected_count):
self.expected_count = expected_count
self._buf = {} # start_index -> list of (R,G,B)
self._last_ms = 0
self._TIMEOUT_MS = 200
def feed(self, start_index, pixels, now_ms):
"""Feed a DNRGB chunk. Returns complete frame list or None."""
# Timeout: discard stale partial frames
if self._buf and (now_ms - self._last_ms) > self._TIMEOUT_MS:
self._buf.clear()
self._last_ms = now_ms
for i, px in enumerate(pixels):
self._buf[start_index + i] = px
if len(self._buf) >= self.expected_count:
frame = [self._buf.get(i, (0, 0, 0)) for i in range(self.expected_count)]
self._buf.clear()
return frame
return None
def reset(self):
self._buf.clear()
```
### Update 2: `firmware/main.py` — Wire frame mode into the render loop
Replace the entire `firmware/main.py` with the final version that handles all three modes: animation, DRGB/DRGBW direct frames, and DNRGB reassembled frames.
Key changes from 07-02 version:
1. Import frame parsers: `parse_drgb`, `parse_drgbw`, `parse_dnrgb`, `FrameAssembler`
2. Add `frame_timeout_ms = 2000` — if no frame packet arrives for 2 seconds, revert to last animation command
3. Add `last_frame_ms` tracking for timeout detection
4. Parse and store frame pixels in the UDP listener
5. Write frame pixels in the render loop using appropriate driver method (`write_rgb` for DRGB/DNRGB, `write_rgbw` for DRGBW)
Here is the complete final `firmware/main.py`:
```python
"""LightSync firmware — main entry point.
Two asyncio tasks:
1. udp_listener_task: non-blocking UDP socket, parses packets, updates state
2. render_task: 60fps loop, renders animations or writes frame data to strip
Modes:
- 'idle': strip is off, waiting for first packet
- 'animation': running a locally-rendered animation (from 0xAC command)
- 'frame': displaying raw pixel data (from DRGB/DRGBW/DNRGB packets)
Frame timeout: if no frame packet arrives for 2 seconds, reverts to last animation.
"""
import asyncio
import socket
import select
import time
import json
from protocol import (
classify_packet, parse_animation_cmd,
parse_drgb, parse_drgbw, parse_dnrgb, FrameAssembler
)
from led_driver import create_strip
from animations import create_animation
# --- Global state ---
config = None
strip = None
current_mode = 'idle' # 'idle', 'animation', 'frame'
current_animation = None # dict from parse_animation_cmd
current_anim_instance = None # animation object with render()
animation_start_ms = 0
frame_pixels = None # list of (R,G,B) or (R,G,B,W) tuples
frame_is_rgbw = False # True if last frame was DRGBW
last_frame_ms = 0 # ticks_ms of last frame packet
frame_assembler = None # FrameAssembler for DNRGB
FRAME_TIMEOUT_MS = 2000 # revert to animation after 2s without frames
def load_config():
try:
with open('config.json') as f:
return json.load(f)
except OSError:
return {'led_count': 300, 'strip_type': 'sk6812', 'led_pin': 5,
'spi_id': 1, 'udp_port': 21324}
async def udp_listener_task(port):
"""Non-blocking UDP listener using select.poll(0) inside asyncio."""
global current_mode, current_animation, current_anim_instance
global animation_start_ms, frame_pixels, frame_is_rgbw
global last_frame_ms, frame_assembler
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('0.0.0.0', port))
sock.setblocking(False)
poller = select.poll()
poller.register(sock, select.POLLIN)
print('[udp] Listening on port', port)
while True:
events = poller.poll(0)
if events:
try:
data, addr = sock.recvfrom(1500)
ptype = classify_packet(data)
if ptype == 'anim_cmd':
cmd = parse_animation_cmd(data)
if cmd:
current_animation = cmd
current_anim_instance = create_animation(cmd)
current_mode = 'animation'
animation_start_ms = time.ticks_ms()
print('[udp] Animation:', cmd['animation'])
elif ptype == 'drgb':
pixels = parse_drgb(data)
frame_pixels = pixels
frame_is_rgbw = False
current_mode = 'frame'
last_frame_ms = time.ticks_ms()
elif ptype == 'drgbw':
pixels = parse_drgbw(data)
frame_pixels = pixels
frame_is_rgbw = True
current_mode = 'frame'
last_frame_ms = time.ticks_ms()
elif ptype == 'dnrgb':
start_idx, pixels = parse_dnrgb(data)
now = time.ticks_ms()
complete = frame_assembler.feed(start_idx, pixels, now)
if complete:
frame_pixels = complete
frame_is_rgbw = False # DNRGB is always RGB
current_mode = 'frame'
last_frame_ms = now
else:
print('[udp] Unknown packet type:', data[0] if data else '?')
except OSError:
pass
await asyncio.sleep_ms(5)
async def render_task():
"""Animation/frame render loop at ~60fps."""
global current_mode, frame_pixels
while True:
if current_mode == 'frame' and frame_pixels:
# Check frame timeout — revert to animation if no packets for 2s
if time.ticks_diff(time.ticks_ms(), last_frame_ms) > FRAME_TIMEOUT_MS:
if current_anim_instance:
current_mode = 'animation'
animation_start_ms = time.ticks_ms()
print('[render] Frame timeout, reverting to animation')
else:
current_mode = 'idle'
strip.clear()
else:
# Write frame to strip
if frame_is_rgbw:
strip.write_rgbw(frame_pixels)
else:
strip.write_rgb(frame_pixels)
elif current_mode == 'animation' and current_anim_instance:
t = time.ticks_diff(time.ticks_ms(), animation_start_ms) / 1000.0
pixels = current_anim_instance.render(t, config['led_count'])
strip.write_rgb(pixels)
await asyncio.sleep_ms(16) # ~60 fps
async def main():
global config, strip, frame_assembler
config = load_config()
strip = create_strip(config)
strip.clear()
frame_assembler = FrameAssembler(config['led_count'])
print('[main] LightSync firmware ready')
print('[main] Strip:', config['strip_type'], '/', config['led_count'], 'LEDs')
print('[main] UDP port:', config['udp_port'])
asyncio.create_task(udp_listener_task(config['udp_port']))
asyncio.create_task(render_task())
while True:
await asyncio.sleep(1)
asyncio.run(main())
```
### File 3: `tests/test_firmware_frames.py`
Create host-side tests that verify end-to-end frame handling: server encodes pixels -> firmware parses -> correct pixel values extracted. Also test DNRGB reassembly and the frame assembler.
```python
"""Host-side tests for firmware frame mode.
Tests DRGB/DRGBW/DNRGB parsing and DNRGB multi-packet reassembly.
Run with: python -m pytest tests/test_firmware_frames.py -v
"""
import sys
import os
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'firmware'))
from protocol import parse_drgb, parse_drgbw, parse_dnrgb, FrameAssembler
from lightsync.protocol.drgb import encode_drgb, encode_drgbw, encode_dnrgb_packets
class TestDRGBFrames:
def test_single_pixel(self):
packet = encode_drgb([(255, 0, 0)])
pixels = parse_drgb(packet)
assert pixels == [(255, 0, 0)]
def test_300_leds(self):
src = [(i % 256, (i * 7) % 256, (i * 13) % 256) for i in range(300)]
packet = encode_drgb(src)
pixels = parse_drgb(packet)
assert len(pixels) == 300
assert pixels[0] == src[0]
assert pixels[149] == src[149]
assert pixels[299] == src[299]
def test_all_black(self):
src = [(0, 0, 0)] * 100
packet = encode_drgb(src)
pixels = parse_drgb(packet)
assert all(p == (0, 0, 0) for p in pixels)
def test_all_white(self):
src = [(255, 255, 255)] * 50
packet = encode_drgb(src)
pixels = parse_drgb(packet)
assert all(p == (255, 255, 255) for p in pixels)
class TestDRGBWFrames:
def test_single_pixel(self):
packet = encode_drgbw([(255, 0, 0, 128)])
pixels = parse_drgbw(packet)
assert pixels == [(255, 0, 0, 128)]
def test_300_leds_rgbw(self):
src = [(i % 256, (i * 3) % 256, (i * 5) % 256, (i * 7) % 256) for i in range(300)]
packet = encode_drgbw(src)
pixels = parse_drgbw(packet)
assert len(pixels) == 300
assert pixels[0] == src[0]
assert pixels[299] == src[299]
def test_w_channel_preserved(self):
"""Verify the W channel survives encode/decode round-trip."""
src = [(100, 100, 100, 200)]
packet = encode_drgbw(src)
pixels = parse_drgbw(packet)
assert pixels[0][3] == 200
class TestDNRGBFrames:
def test_single_packet_small(self):
src = [(255, 0, 0)] * 100
packets = encode_dnrgb_packets(src)
assert len(packets) == 1
start_idx, pixels = parse_dnrgb(packets[0])
assert start_idx == 0
assert len(pixels) == 100
assert pixels[0] == (255, 0, 0)
def test_multi_packet_600_leds(self):
"""600 LEDs = 2 DNRGB packets (489 + 111)."""
src = [(i % 256, 0, 0) for i in range(600)]
packets = encode_dnrgb_packets(src)
assert len(packets) == 2
s0, px0 = parse_dnrgb(packets[0])
s1, px1 = parse_dnrgb(packets[1])
assert s0 == 0
assert s1 == 489
assert len(px0) == 489
assert len(px1) == 111
# Verify pixel values survive round-trip
assert px0[0] == src[0]
assert px0[488] == src[488]
assert px1[0] == src[489]
assert px1[110] == src[599]
def test_start_index_offset(self):
"""Packets with non-zero start_index."""
src = [(0, 255, 0)] * 50
packets = encode_dnrgb_packets(src, start_index=100)
assert len(packets) == 1
start_idx, pixels = parse_dnrgb(packets[0])
assert start_idx == 100
assert len(pixels) == 50
class TestFrameAssembler:
def test_single_chunk_completes(self):
asm = FrameAssembler(100)
pixels = [(i, 0, 0) for i in range(100)]
result = asm.feed(0, pixels, 1000)
assert result is not None
assert len(result) == 100
assert result[0] == (0, 0, 0)
assert result[99] == (99, 0, 0)
def test_two_chunks(self):
asm = FrameAssembler(200)
chunk1 = [(i, 0, 0) for i in range(100)]
chunk2 = [(100 + i, 0, 0) for i in range(100)]
result1 = asm.feed(0, chunk1, 1000)
assert result1 is None # not complete yet
result2 = asm.feed(100, chunk2, 1010)
assert result2 is not None
assert len(result2) == 200
assert result2[0] == (0, 0, 0)
assert result2[100] == (100, 0, 0)
assert result2[199] == (199, 0, 0)
def test_timeout_clears_buffer(self):
asm = FrameAssembler(200)
chunk1 = [(i, 0, 0) for i in range(100)]
asm.feed(0, chunk1, 1000)
assert len(asm._buf) == 100
# Feed again after timeout (300ms > 200ms timeout)
chunk2 = [(50 + i, 0, 0) for i in range(50)]
asm.feed(0, chunk2, 1300)
# Buffer should have been cleared, only chunk2 remains
assert len(asm._buf) == 50
def test_reset(self):
asm = FrameAssembler(100)
asm.feed(0, [(0, 0, 0)] * 50, 1000)
asm.reset()
assert len(asm._buf) == 0
def test_full_round_trip_with_encoder(self):
"""Multi-packet DNRGB: encode 600 LEDs -> feed both packets -> get complete frame."""
asm = FrameAssembler(600)
src = [(i % 256, (i * 3) % 256, (i * 7) % 256) for i in range(600)]
packets = encode_dnrgb_packets(src)
assert len(packets) == 2
s0, px0 = parse_dnrgb(packets[0])
result = asm.feed(s0, px0, 1000)
assert result is None
s1, px1 = parse_dnrgb(packets[1])
result = asm.feed(s1, px1, 1005)
assert result is not None
assert len(result) == 600
assert result[0] == src[0]
assert result[599] == src[599]
class TestFrameModeIntegration:
"""Test that SK6812 (RGBW) and WS2801 (RGB) frame paths work correctly."""
def test_drgb_for_ws2801(self):
"""WS2801 is RGB — DRGB packets contain 3 bytes per pixel."""
src = [(255, 128, 64)] * 300
packet = encode_drgb(src)
pixels = parse_drgb(packet)
# All pixels should be RGB tuples
assert all(len(p) == 3 for p in pixels)
assert pixels[0] == (255, 128, 64)
def test_drgbw_for_sk6812(self):
"""SK6812 is RGBW — DRGBW packets contain 4 bytes per pixel."""
src = [(255, 128, 64, 200)] * 300
packet = encode_drgbw(src)
pixels = parse_drgbw(packet)
# All pixels should be RGBW tuples
assert all(len(p) == 4 for p in pixels)
assert pixels[0] == (255, 128, 64, 200)
def test_drgb_pixel_count_matches(self):
"""300 LED DRGB: 2 + 300*3 = 902 bytes, fits in one packet."""
src = [(0, 0, 0)] * 300
packet = encode_drgb(src)
assert len(packet) == 902
pixels = parse_drgb(packet)
assert len(pixels) == 300
def test_drgbw_pixel_count_matches(self):
"""300 LED DRGBW: 2 + 300*4 = 1202 bytes, fits in one packet."""
src = [(0, 0, 0, 0)] * 300
packet = encode_drgbw(src)
assert len(packet) == 1202
pixels = parse_drgbw(packet)
assert len(pixels) == 300
```
## Verification
- [ ] `python -m pytest tests/test_firmware_frames.py -v` passes all tests
- [ ] `firmware/protocol.py` contains `FrameAssembler` class with `feed()`, `reset()`, and 200ms timeout
- [ ] `firmware/main.py` handles all 3 frame types: DRGB (direct RGB write), DRGBW (direct RGBW write), DNRGB (via FrameAssembler)
- [ ] Frame mode uses `strip.write_rgbw()` for DRGBW packets and `strip.write_rgb()` for DRGB/DNRGB packets
- [ ] Frame timeout (2 seconds) reverts to last animation command mode
- [ ] DNRGB reassembly correctly handles 600-LED multi-packet frame (2 packets: 489 + 111 LEDs)
- [ ] 300-LED DRGB packet = 902 bytes, 300-LED DRGBW packet = 1202 bytes — both fit in single UDP packet
- [ ] `FrameAssembler` clears stale partial frames after 200ms timeout
- [ ] WS2801 strips receive RGB data via `write_rgb()` (W channel dropped by driver)
- [ ] SK6812 strips receive RGBW data via `write_rgbw()` when DRGBW packets arrive
## Notes
- For the 300-LED strips used in this project, DRGB (902 bytes) and DRGBW (1202 bytes) both fit in a single UDP packet. DNRGB reassembly will never trigger in practice but is implemented for future strip sizes.
- Frame mode takes priority over animation mode: when a raw frame arrives, the strip immediately shows it. After 2 seconds without frame packets, the firmware reverts to the last animation command (if any). This matches WLED's behavior.
- The `frame_is_rgbw` flag tracks whether to use `write_rgb` or `write_rgbw`. DRGB and DNRGB are always RGB (3 bytes/pixel). DRGBW is always RGBW (4 bytes/pixel). The strip driver handles the mismatch: `WS2801Strip.write_rgbw()` silently drops the W channel, `NeoPixelStrip.write_rgb()` sets W=0.
- The render task checks frame timeout BEFORE writing — this avoids re-writing stale frame data every 16ms when no new packets arrive. Once timeout triggers, it reverts to animation mode which produces fresh frames each cycle.