feat(07-01): add firmware scaffold — boot.py, config.json, led_driver.py, protocol tests
- firmware/boot.py: WiFi connection on startup (MicroPython boot sequence) - firmware/config.json: default device config with all 9 fields (ssid, password, led_count, strip_type, led_pin, spi_id, spi_clk_pin, spi_dat_pin, udp_port) - firmware/led_driver.py: NeoPixelStrip (SK6812 RGBW) and WS2801Strip (RGB SPI) with unified interface (write_rgb, write_rgbw, fill, clear) - tests/test_firmware_protocol.py: 21 host-side tests for round-trip protocol parsing (DRGB, DRGBW, DNRGB, animation_cmd)
This commit is contained in:
38
firmware/boot.py
Normal file
38
firmware/boot.py
Normal file
@@ -0,0 +1,38 @@
|
||||
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'])
|
||||
11
firmware/config.json
Normal file
11
firmware/config.json
Normal file
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"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
|
||||
}
|
||||
87
firmware/led_driver.py
Normal file
87
firmware/led_driver.py
Normal file
@@ -0,0 +1,87 @@
|
||||
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'])
|
||||
114
tests/test_firmware_protocol.py
Normal file
114
tests/test_firmware_protocol.py
Normal file
@@ -0,0 +1,114 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user