---
phase: 03-communication-protocol
plan: 03
type: execute
wave: 2
depends_on: ["03-01", "03-02"]
files_modified:
- lightsync/protocol/simulator.py
- lightsync/main.py
- tests/test_e2e_protocol.py
autonomous: true
requirements: [UDP-04]
must_haves:
truths:
- "Simulator receives and logs DRGB packets with correct LED count"
- "Simulator receives and logs DRGBW packets distinguishing 4-byte pixels"
- "Simulator receives and logs DNRGB multi-packet frames with correct start indices"
- "Simulator receives and logs animation command packets (0xAC) with decoded params"
- "UDPSender is started in FastAPI lifespan and available as app.state.udp_sender"
- "End-to-end test sends packets via UDPSender and simulator receives them correctly"
artifacts:
- path: "lightsync/protocol/simulator.py"
provides: "Async UDP simulator/receiver with packet logging"
exports: ["SimulatorProtocol", "run_simulator"]
contains: "class SimulatorProtocol"
- path: "tests/test_e2e_protocol.py"
provides: "End-to-end protocol validation tests"
min_lines: 60
key_links:
- from: "lightsync/protocol/simulator.py"
to: "lightsync/protocol/drgb.py"
via: "protocol type byte constants"
pattern: "PROTOCOL_DRGB|0x02|0x03|0x04|0xAC"
- from: "lightsync/main.py"
to: "lightsync/protocol/udp_sender.py"
via: "app.state.udp_sender"
pattern: "app.state.udp_sender"
- from: "tests/test_e2e_protocol.py"
to: "lightsync/protocol/udp_sender.py"
via: "UDPSender send()"
pattern: "UDPSender"
---
Build the software UDP simulator, integrate UDPSender into FastAPI lifespan, and run end-to-end protocol validation proving all packet types are sent and received correctly.
Purpose: UDP-04 requires a simulator for hardware-free protocol testing. This plan proves the full stack works: animation renders pixels -> encoder creates packets -> sender transmits -> simulator receives and validates.
Output: Working simulator, FastAPI integration, end-to-end test suite.
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/03-communication-protocol/03-RESEARCH.md
@.planning/phases/03-communication-protocol/03-01-SUMMARY.md
@.planning/phases/03-communication-protocol/03-02-SUMMARY.md
From lightsync/protocol/drgb.py:
```python
WLED_PORT = 21324
PROTOCOL_DRGB = 2
PROTOCOL_DRGBW = 3
PROTOCOL_DNRGB = 4
TIMEOUT_BYTE = 2
DRGB_MAX_PX = 490
DRGBW_MAX_PX = 367
DNRGB_MAX_PX = 489
def encode_drgb(pixels: list[tuple[int,int,int]]) -> bytes: ...
def encode_drgbw(pixels: list[tuple[int,int,int,int]]) -> bytes: ...
def encode_dnrgb_packets(pixels: list[tuple[int,int,int]], start_index: int = 0) -> list[bytes]: ...
```
From lightsync/protocol/animation_cmd.py:
```python
ANIM_CMD_MARKER = 0xAC
ANIMATION_IDS = {"solid_color": 0, "chase": 1, "pulse": 2, "rainbow": 3, "strobe": 4, "color_wipe": 5, "fire": 6}
def encode_animation_cmd(animation: str, params: dict) -> bytes: ...
def decode_animation_cmd(data: bytes) -> dict: ...
```
From lightsync/protocol/udp_sender.py:
```python
class UDPSender:
async def start(self) -> None: ...
def send(self, payload: bytes, ip: str, port: int) -> None: ...
async def stop(self) -> None: ...
@property
def is_running(self) -> bool: ...
```
From lightsync/animations/__init__.py:
```python
ANIMATION_REGISTRY: dict[str, type[AnimationBase]] = { ... } # 7 types
def create_animation(name: str, params: dict) -> AnimationBase: ...
```
From lightsync/animations/base.py:
```python
class AnimationBase(ABC):
def render(self, t: float, led_count: int) -> list[tuple]: ...
@classmethod
def from_params(cls, params: dict) -> "AnimationBase": ...
```
From lightsync/main.py (current lifespan):
```python
@asynccontextmanager
async def lifespan(app: FastAPI):
global registry, show_store
registry = DeviceRegistry(Path("devices.json"))
await registry.load()
show_store = ShowStore(Path("shows"))
show_store.ensure_dir()
app.state.beats = {}
yield
await registry.save()
```
Task 1: UDP simulator with packet logging
lightsync/protocol/simulator.py, tests/test_e2e_protocol.py
- lightsync/protocol/drgb.py (protocol constants PROTOCOL_DRGB, PROTOCOL_DRGBW, PROTOCOL_DNRGB, WLED_PORT)
- lightsync/protocol/animation_cmd.py (ANIM_CMD_MARKER, decode_animation_cmd)
- lightsync/protocol/udp_sender.py (UDPSender class for sending in tests)
- lightsync/animations/__init__.py (create_animation, ANIMATION_REGISTRY)
- .planning/phases/03-communication-protocol/03-RESEARCH.md (Simulator code example, Pitfall 5: port binding)
- SimulatorProtocol.datagram_received parses DRGB packets: extracts led_count = (len(data)-2)//3
- SimulatorProtocol.datagram_received parses DRGBW packets: extracts led_count = (len(data)-2)//4
- SimulatorProtocol.datagram_received parses DNRGB packets: extracts start index (big-endian uint16 at bytes 2-3) and led_count
- SimulatorProtocol.datagram_received parses animation command packets (0xAC): calls decode_animation_cmd
- SimulatorProtocol stores received packets in a log list for test assertions
- E2E test: encode_drgb 10 pixels -> send via UDPSender -> simulator receives with led_count=10
- E2E test: encode_drgbw 10 pixels -> send -> simulator receives with led_count=10 and protocol=DRGBW
- E2E test: encode_dnrgb_packets 600 pixels -> send all packets -> simulator receives 2 packets with correct start indices (0, 489)
- E2E test: encode_animation_cmd "chase" -> send -> simulator receives and decode matches original
- E2E test: create_animation("solid_color", {...}).render() -> encode_drgb -> send -> simulator receives correct pixel count
- run_simulator is runnable as __main__ standalone
Create `lightsync/protocol/simulator.py`:
```python
import asyncio
import struct
from dataclasses import dataclass, field
from lightsync.protocol.drgb import PROTOCOL_DRGB, PROTOCOL_DRGBW, PROTOCOL_DNRGB, WLED_PORT
from lightsync.protocol.animation_cmd import ANIM_CMD_MARKER, decode_animation_cmd
PROTOCOL_NAMES = {
PROTOCOL_DRGB: "DRGB",
PROTOCOL_DRGBW: "DRGBW",
PROTOCOL_DNRGB: "DNRGB",
ANIM_CMD_MARKER: "ANIM_CMD",
}
@dataclass
class PacketLog:
protocol: str
addr: tuple
led_count: int = 0
start_index: int = 0
raw: bytes = b""
decoded: dict | None = None
class SimulatorProtocol(asyncio.DatagramProtocol):
def __init__(self):
self.packets: list[PacketLog] = []
self.transport: asyncio.DatagramTransport | None = None
def connection_made(self, transport):
self.transport = transport
def datagram_received(self, data: bytes, addr: tuple) -> None:
if len(data) < 2:
return
proto_type = data[0]
name = PROTOCOL_NAMES.get(proto_type, f"UNKNOWN(0x{proto_type:02X})")
log = PacketLog(protocol=name, addr=addr, raw=data)
if proto_type == PROTOCOL_DRGB:
log.led_count = (len(data) - 2) // 3
elif proto_type == PROTOCOL_DRGBW:
log.led_count = (len(data) - 2) // 4
elif proto_type == PROTOCOL_DNRGB:
log.start_index = struct.unpack_from(">H", data, 2)[0]
log.led_count = (len(data) - 4) // 3
elif proto_type == ANIM_CMD_MARKER:
try:
log.decoded = decode_animation_cmd(data)
except Exception:
log.decoded = None
self.packets.append(log)
print(f"[SIM] {name} from {addr}: {log.led_count} LEDs"
+ (f" start={log.start_index}" if proto_type == PROTOCOL_DNRGB else "")
+ (f" cmd={log.decoded}" if log.decoded else ""))
async def run_simulator(host: str = "0.0.0.0", port: int = WLED_PORT) -> tuple[asyncio.DatagramTransport, SimulatorProtocol]:
loop = asyncio.get_running_loop()
transport, protocol = await loop.create_datagram_endpoint(
SimulatorProtocol,
local_addr=(host, port),
)
print(f"[SIM] Listening on {host}:{port}")
return transport, protocol
if __name__ == "__main__":
async def main():
transport, protocol = await run_simulator()
try:
await asyncio.sleep(float("inf"))
except KeyboardInterrupt:
pass
finally:
transport.close()
asyncio.run(main())
```
Create `tests/test_e2e_protocol.py` — end-to-end tests using asyncio:
Use `pytest` with `pytest-asyncio` (or manual `asyncio.run` in fixtures). Each test:
1. Start simulator on localhost with a random free port (use port=0, then read `.transport.get_extra_info("sockname")[1]` for actual port)
2. Start UDPSender
3. Encode packet, send to simulator's port on 127.0.0.1
4. `await asyncio.sleep(0.05)` for UDP delivery
5. Assert simulator.packets has expected entries
6. Cleanup: stop sender, close simulator transport
Write a reusable async fixture or helper function for setup/teardown.
Tests to write:
- `test_e2e_drgb_10_pixels`: encode_drgb with 10 red pixels, send, assert simulator got DRGB with led_count=10
- `test_e2e_drgbw_10_pixels`: encode_drgbw with 10 RGBW pixels, send, assert DRGBW with led_count=10
- `test_e2e_dnrgb_600_pixels`: encode_dnrgb_packets with 600 pixels, send all packets, assert 2 packets received with start_index 0 and 489
- `test_e2e_animation_cmd_chase`: encode_animation_cmd("chase", params), send, assert simulator decoded animation=="chase"
- `test_e2e_full_pipeline`: create_animation("solid_color", {"color":[255,0,0]}).render(0.0, 20) -> encode_drgb -> send -> simulator receives 20 LED DRGB frame
- `test_e2e_sk6812_rgbw_pipeline`: create SK6812 device, render solid_color with white=128, encode_drgbw, send, simulator receives DRGBW with 4 bytes/pixel
- `test_simulator_standalone_import`: `from lightsync.protocol.simulator import run_simulator` does not raise
If `pytest-asyncio` is not available, use this pattern instead:
```python
def test_e2e_something():
asyncio.run(_test_e2e_something())
async def _test_e2e_something():
# actual async test logic
```
cd /home/claude/led2 && python -m pytest tests/test_e2e_protocol.py -v
- lightsync/protocol/simulator.py contains "class SimulatorProtocol(asyncio.DatagramProtocol)"
- lightsync/protocol/simulator.py contains "class PacketLog"
- lightsync/protocol/simulator.py contains "async def run_simulator("
- lightsync/protocol/simulator.py contains 'if __name__ == "__main__"'
- lightsync/protocol/simulator.py contains "decode_animation_cmd"
- lightsync/protocol/simulator.py contains 'struct.unpack_from(">H"'
- tests/test_e2e_protocol.py exits 0 with all tests passing
- tests/test_e2e_protocol.py contains at least 5 test functions
- tests/test_e2e_protocol.py contains "UDPSender" and "SimulatorProtocol" and "encode_drgb"
Simulator receives and correctly parses all 4 packet types (DRGB, DRGBW, DNRGB, ANIM_CMD). End-to-end tests prove full pipeline: animation render -> encode -> send -> receive -> validate. Simulator runnable standalone via `python -m lightsync.protocol.simulator`.
Task 2: Integrate UDPSender into FastAPI lifespan
lightsync/main.py
- lightsync/main.py (current lifespan function — app.state.beats pattern)
- lightsync/protocol/udp_sender.py (UDPSender start/stop lifecycle)
Edit `lightsync/main.py` lifespan function to add UDPSender lifecycle:
1. Add import at top: `from lightsync.protocol.udp_sender import UDPSender`
2. In the lifespan function, AFTER the existing `app.state.beats = {}` line, add:
```python
# UDP sender for device communication (Phase 3)
udp_sender = UDPSender()
await udp_sender.start()
app.state.udp_sender = udp_sender
```
3. In the shutdown section (after `yield`), BEFORE `await registry.save()`, add:
```python
await app.state.udp_sender.stop()
```
Keep ALL existing lifespan code unchanged. Only add the UDPSender lines.
cd /home/claude/led2 && python -c "
import ast, sys
with open('lightsync/main.py') as f:
source = f.read()
tree = ast.parse(source)
# Check import exists
has_import = 'from lightsync.protocol.udp_sender import UDPSender' in source
# Check app.state.udp_sender assignment
has_assignment = 'app.state.udp_sender' in source
# Check stop call
has_stop = 'udp_sender.stop()' in source
# Check existing code preserved
has_beats = 'app.state.beats' in source
has_registry = 'await registry.save()' in source
results = [
('import UDPSender', has_import),
('app.state.udp_sender', has_assignment),
('udp_sender.stop()', has_stop),
('app.state.beats preserved', has_beats),
('registry.save() preserved', has_registry),
]
for name, ok in results:
status = 'OK' if ok else 'MISSING'
print(f'{name}: {status}')
all_ok = all(ok for _, ok in results)
sys.exit(0 if all_ok else 1)
"
- lightsync/main.py contains "from lightsync.protocol.udp_sender import UDPSender"
- lightsync/main.py contains "app.state.udp_sender = udp_sender" (or equivalent)
- lightsync/main.py contains "await udp_sender.stop()" (or "await app.state.udp_sender.stop()")
- lightsync/main.py still contains "app.state.beats = {}" (existing code preserved)
- lightsync/main.py still contains "await registry.save()" (existing shutdown preserved)
- lightsync/main.py still contains "registry = DeviceRegistry" (existing startup preserved)
UDPSender starts on app boot and stops on shutdown. Available to all routes via request.app.state.udp_sender. All existing lifespan behavior unchanged.
- `python -m pytest tests/test_e2e_protocol.py -v` — all end-to-end tests pass
- `python -c "from lightsync.protocol.simulator import run_simulator; print('simulator importable')"` — no import errors
- `python -m lightsync.protocol.simulator &` then `python -c "import asyncio; from lightsync.protocol.udp_sender import UDPSender; from lightsync.protocol.drgb import encode_drgb; asyncio.run((s := UDPSender()) or s.start()); s.send(encode_drgb([(255,0,0)]*10), '127.0.0.1', 21324)"` — simulator logs DRGB packet
- lightsync/main.py has app.state.udp_sender in lifespan
- Simulator correctly parses and logs all 4 packet types (DRGB, DRGBW, DNRGB, animation command)
- End-to-end tests prove: animation render -> encode -> UDP send -> simulator receive -> validate
- DNRGB multi-packet splitting verified: 600-LED frame splits into 2 packets with correct start indices
- UDPSender integrated into FastAPI lifespan (app.state.udp_sender)
- Simulator runnable standalone via `python -m lightsync.protocol.simulator`
- No hardware required for any validation