led2/.planning/phases/07-microcontroller-firmware/07-03-PLAN.md

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:

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:

"""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.

"""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.