16 KiB
16 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | must_haves | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 03-communication-protocol | 03 | execute | 2 |
|
|
true |
|
|
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.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@.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.mdFrom lightsync/protocol/drgb.py:
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:
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:
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:
ANIMATION_REGISTRY: dict[str, type[AnimationBase]] = { ... } # 7 types
def create_animation(name: str, params: dict) -> AnimationBase: ...
From lightsync/animations/base.py:
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):
@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()
```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
```
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.
<success_criteria>
- 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 </success_criteria>