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

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:

{
  "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):

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.

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.

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.

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

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