docs(03): create phase plan — protocol encoders, animation library, simulator
This commit is contained in:
Claude
322
.planning/phases/03-communication-protocol/03-01-PLAN.md
Normal file
322
.planning/phases/03-communication-protocol/03-01-PLAN.md
Normal file
@@ -0,0 +1,322 @@
|
||||
---
|
||||
phase: 03-communication-protocol
|
||||
plan: 01
|
||||
type: execute
|
||||
wave: 1
|
||||
depends_on: []
|
||||
files_modified:
|
||||
- lightsync/protocol/__init__.py
|
||||
- lightsync/protocol/drgb.py
|
||||
- lightsync/protocol/animation_cmd.py
|
||||
- lightsync/protocol/udp_sender.py
|
||||
- lightsync/devices/sk6812.py
|
||||
- lightsync/devices/ws2801.py
|
||||
- tests/test_protocol.py
|
||||
autonomous: true
|
||||
requirements: [UDP-01, UDP-02, UDP-03, ANI-03, ANI-04]
|
||||
|
||||
must_haves:
|
||||
truths:
|
||||
- "DRGB encoder produces correct 2-byte header + 3-byte-per-pixel body"
|
||||
- "DRGBW encoder produces correct 2-byte header + 4-byte-per-pixel body"
|
||||
- "DNRGB encoder splits 300+ LED frames into multiple packets under 1472 bytes each"
|
||||
- "Animation command packet uses 0xAC marker and encodes all 7 animation type IDs"
|
||||
- "SK6812 encode_animation_cmd returns real bytes, not empty stub"
|
||||
- "WS2801 encode_animation_cmd returns real bytes, not empty stub"
|
||||
- "UDPSender can send to arbitrary IP:port via asyncio DatagramTransport"
|
||||
artifacts:
|
||||
- path: "lightsync/protocol/drgb.py"
|
||||
provides: "DRGB, DRGBW, DNRGB encoders"
|
||||
exports: ["encode_drgb", "encode_drgbw", "encode_dnrgb_packets"]
|
||||
- path: "lightsync/protocol/animation_cmd.py"
|
||||
provides: "Animation command packet encoder/decoder"
|
||||
exports: ["encode_animation_cmd", "decode_animation_cmd", "ANIMATION_IDS"]
|
||||
- path: "lightsync/protocol/udp_sender.py"
|
||||
provides: "Async UDP transport wrapper"
|
||||
exports: ["UDPSender"]
|
||||
- path: "tests/test_protocol.py"
|
||||
provides: "Protocol encoding tests"
|
||||
min_lines: 80
|
||||
key_links:
|
||||
- from: "lightsync/devices/sk6812.py"
|
||||
to: "lightsync/protocol/animation_cmd.py"
|
||||
via: "import encode_animation_cmd"
|
||||
pattern: "from lightsync.protocol.animation_cmd import encode_animation_cmd"
|
||||
- from: "lightsync/protocol/drgb.py"
|
||||
to: "WLED spec"
|
||||
via: "byte layout constants"
|
||||
pattern: "PROTOCOL_DRGB.*=.*2"
|
||||
---
|
||||
|
||||
<objective>
|
||||
Build the UDP packet protocol layer: WLED-compatible frame encoders (DRGB/DRGBW/DNRGB), custom animation command packet format, async UDP sender, and fill device encode_animation_cmd stubs.
|
||||
|
||||
Purpose: This is the wire contract that microcontrollers (Phase 7) will implement. All packet formats must be correct and tested before building anything on top.
|
||||
Output: lightsync/protocol/ package with encoders, UDP sender, and passing tests.
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@$HOME/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/03-communication-protocol/03-RESEARCH.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Existing contracts the executor must work with -->
|
||||
|
||||
From lightsync/devices/base.py:
|
||||
```python
|
||||
class BaseDevice(ABC):
|
||||
def __init__(self, config: DeviceConfig):
|
||||
self.config = config
|
||||
|
||||
@property
|
||||
def id(self): return self.config.id
|
||||
|
||||
@property
|
||||
def led_count(self) -> int: return self.config.led_count
|
||||
|
||||
@abstractmethod
|
||||
def encode_frame(self, pixels: list[tuple]) -> bytes: ...
|
||||
|
||||
@abstractmethod
|
||||
def encode_animation_cmd(self, animation: str, params: dict) -> bytes: ...
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def bytes_per_pixel(self) -> int: ...
|
||||
```
|
||||
|
||||
From lightsync/models/show.py:
|
||||
```python
|
||||
class CueModel(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
timestamp: float
|
||||
mode: Literal["animation", "frame_sequence"] = "animation"
|
||||
animation: str | None = None
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
```
|
||||
|
||||
From lightsync/models/device.py:
|
||||
```python
|
||||
StripType = Literal["sk6812", "ws2801", "generic"]
|
||||
|
||||
class DeviceConfig(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
name: str
|
||||
strip_type: StripType
|
||||
led_count: int = Field(gt=0, le=1000)
|
||||
ip: str
|
||||
port: int = Field(ge=1, le=65535)
|
||||
enabled: bool = True
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: Protocol encoders — DRGB/DRGBW/DNRGB + animation command</name>
|
||||
<files>lightsync/protocol/__init__.py, lightsync/protocol/drgb.py, lightsync/protocol/animation_cmd.py, tests/test_protocol.py</files>
|
||||
<read_first>
|
||||
- lightsync/devices/base.py (BaseDevice ABC — encode_animation_cmd signature)
|
||||
- lightsync/models/show.py (CueModel — animation and params fields)
|
||||
- .planning/phases/03-communication-protocol/03-RESEARCH.md (WLED byte layout spec, animation command format)
|
||||
</read_first>
|
||||
<behavior>
|
||||
- encode_drgb([(255,0,0),(0,255,0)]) returns bytes([0x02, 2, 255, 0, 0, 0, 255, 0])
|
||||
- encode_drgbw([(255,0,0,128)]) returns bytes([0x03, 2, 255, 0, 0, 128])
|
||||
- encode_drgb with 490 pixels returns exactly 1472 bytes (2 header + 490*3)
|
||||
- encode_drgbw with 367 pixels returns exactly 1470 bytes (2 header + 367*4)
|
||||
- encode_drgb with 491 pixels truncates to 490 pixels
|
||||
- encode_dnrgb_packets with 600 pixels returns 2 packets (489 + 111 LEDs)
|
||||
- encode_dnrgb_packets second packet has start_index=489 in big-endian at bytes 2-3
|
||||
- Each DNRGB packet is <= 1472 bytes
|
||||
- encode_animation_cmd("chase", {"color":[255,0,0],"white":0,"speed":0.5,"reverse":False}) returns 16-byte packet starting with 0xAC, 0x01
|
||||
- encode_animation_cmd animation type IDs: solid_color=0, chase=1, pulse=2, rainbow=3, strobe=4, color_wipe=5, fire=6
|
||||
- decode_animation_cmd round-trips with encode_animation_cmd for all 7 types
|
||||
- Unknown animation type raises ValueError
|
||||
</behavior>
|
||||
<action>
|
||||
Create `lightsync/protocol/__init__.py` (empty, makes it a package).
|
||||
|
||||
Create `lightsync/protocol/drgb.py` with these exact constants and functions:
|
||||
```python
|
||||
WLED_PORT = 21324
|
||||
PROTOCOL_DRGB = 2
|
||||
PROTOCOL_DRGBW = 3
|
||||
PROTOCOL_DNRGB = 4
|
||||
TIMEOUT_BYTE = 2
|
||||
|
||||
DRGB_MAX_PX = 490 # (1472 - 2) // 3
|
||||
DRGBW_MAX_PX = 367 # (1472 - 2) // 4
|
||||
DNRGB_MAX_PX = 489 # (1472 - 4) // 3
|
||||
```
|
||||
|
||||
Functions:
|
||||
- `encode_drgb(pixels: list[tuple[int,int,int]]) -> bytes` — header [PROTOCOL_DRGB, TIMEOUT_BYTE] + RGB body, truncate at DRGB_MAX_PX
|
||||
- `encode_drgbw(pixels: list[tuple[int,int,int,int]]) -> bytes` — header [PROTOCOL_DRGBW, TIMEOUT_BYTE] + RGBW body, truncate at DRGBW_MAX_PX
|
||||
- `encode_dnrgb_packets(pixels: list[tuple[int,int,int]], start_index: int = 0) -> list[bytes]` — split into chunks of DNRGB_MAX_PX, each with 4-byte header using `struct.pack(">BBH", PROTOCOL_DNRGB, TIMEOUT_BYTE, idx)`. Use big-endian for start index per WLED spec.
|
||||
|
||||
Create `lightsync/protocol/animation_cmd.py` with:
|
||||
```python
|
||||
ANIM_CMD_MARKER = 0xAC
|
||||
ANIM_CMD_VERSION = 1
|
||||
|
||||
ANIMATION_IDS = {
|
||||
"solid_color": 0, "chase": 1, "pulse": 2, "rainbow": 3,
|
||||
"strobe": 4, "color_wipe": 5, "fire": 6,
|
||||
}
|
||||
ANIMATION_NAMES = {v: k for k, v in ANIMATION_IDS.items()}
|
||||
```
|
||||
|
||||
Functions:
|
||||
- `encode_animation_cmd(animation: str, params: dict) -> bytes` — 16-byte packet:
|
||||
Byte 0: 0xAC, Byte 1: version=1, Byte 2: animation type ID (from ANIMATION_IDS, raise ValueError if unknown),
|
||||
Byte 3: flags (bit 0 = params.get("reverse", False)),
|
||||
Bytes 4-7: struct.pack(">f", params.get("speed", 0.5)),
|
||||
Bytes 8-10: primary color RGB from params.get("color", [0,0,0]),
|
||||
Bytes 11-13: secondary color RGB from params.get("bg_color", [0,0,0]),
|
||||
Byte 14: params.get("white", 0),
|
||||
Byte 15: params.get("density", params.get("size", params.get("cooling", 0))) & 0xFF
|
||||
|
||||
- `decode_animation_cmd(data: bytes) -> dict` — inverse of encode. Returns {"animation": str, "params": dict} with speed, color, bg_color, white, density, reverse, flags.
|
||||
|
||||
Create `tests/test_protocol.py` with pytest tests covering all behaviors listed above.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>cd /home/claude/led2 && python -m pytest tests/test_protocol.py -v</automated>
|
||||
</verify>
|
||||
<acceptance_criteria>
|
||||
- lightsync/protocol/__init__.py exists
|
||||
- lightsync/protocol/drgb.py contains "PROTOCOL_DRGB = 2" and "PROTOCOL_DRGBW = 3" and "PROTOCOL_DNRGB = 4"
|
||||
- lightsync/protocol/drgb.py contains "def encode_drgb(" and "def encode_drgbw(" and "def encode_dnrgb_packets("
|
||||
- lightsync/protocol/drgb.py contains 'struct.pack(">BBH"'
|
||||
- lightsync/protocol/animation_cmd.py contains "ANIM_CMD_MARKER = 0xAC"
|
||||
- lightsync/protocol/animation_cmd.py contains "def encode_animation_cmd(" and "def decode_animation_cmd("
|
||||
- lightsync/protocol/animation_cmd.py contains '"solid_color": 0' and '"chase": 1' and '"fire": 6'
|
||||
- tests/test_protocol.py exits 0 with all tests passing
|
||||
- tests/test_protocol.py contains at least 10 test functions (test_encode_drgb, test_encode_drgbw, test_dnrgb_split, test_dnrgb_start_index, test_animation_cmd_chase, test_animation_cmd_roundtrip, test_animation_cmd_unknown, etc.)
|
||||
</acceptance_criteria>
|
||||
<done>All WLED frame encoders produce spec-compliant packets. Animation command encoder/decoder round-trips correctly for all 7 types. All tests pass.</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 2: UDP sender + device stub completion</name>
|
||||
<files>lightsync/protocol/udp_sender.py, lightsync/devices/sk6812.py, lightsync/devices/ws2801.py</files>
|
||||
<read_first>
|
||||
- lightsync/devices/sk6812.py (current encode_animation_cmd stub returning b"")
|
||||
- lightsync/devices/ws2801.py (current encode_animation_cmd stub returning b"")
|
||||
- lightsync/devices/base.py (BaseDevice ABC contract)
|
||||
- lightsync/protocol/animation_cmd.py (just created in Task 1 — encode_animation_cmd function)
|
||||
- lightsync/main.py (lifespan pattern — app.state usage)
|
||||
- .planning/phases/03-communication-protocol/03-RESEARCH.md (UDPSender pattern, single socket design)
|
||||
</read_first>
|
||||
<action>
|
||||
Create `lightsync/protocol/udp_sender.py`:
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
class _NullProtocol(asyncio.DatagramProtocol):
|
||||
def error_received(self, exc: Exception) -> None:
|
||||
pass # UDP send errors non-fatal for LED delivery
|
||||
|
||||
class UDPSender:
|
||||
def __init__(self):
|
||||
self._transport: asyncio.DatagramTransport | None = None
|
||||
|
||||
async def start(self) -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
self._transport, _ = await loop.create_datagram_endpoint(
|
||||
_NullProtocol,
|
||||
local_addr=("0.0.0.0", 0),
|
||||
)
|
||||
|
||||
def send(self, payload: bytes, ip: str, port: int) -> None:
|
||||
if self._transport:
|
||||
self._transport.sendto(payload, (ip, port))
|
||||
|
||||
async def stop(self) -> None:
|
||||
if self._transport:
|
||||
self._transport.close()
|
||||
self._transport = None
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return self._transport is not None
|
||||
```
|
||||
|
||||
Update `lightsync/devices/sk6812.py` — replace the encode_animation_cmd stub:
|
||||
- Add import: `from lightsync.protocol.animation_cmd import encode_animation_cmd as _encode_cmd`
|
||||
- Replace `encode_animation_cmd` method body from `return b""` to `return _encode_cmd(animation, params)`
|
||||
- Keep the existing `encode_frame` method unchanged.
|
||||
|
||||
Update `lightsync/devices/ws2801.py` — identical change:
|
||||
- Add import: `from lightsync.protocol.animation_cmd import encode_animation_cmd as _encode_cmd`
|
||||
- Replace `encode_animation_cmd` method body from `return b""` to `return _encode_cmd(animation, params)`
|
||||
- Keep the existing `encode_frame` method unchanged.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>cd /home/claude/led2 && python -c "
|
||||
from lightsync.protocol.udp_sender import UDPSender
|
||||
s = UDPSender()
|
||||
assert not s.is_running
|
||||
print('UDPSender: OK')
|
||||
|
||||
from lightsync.devices.sk6812 import SK6812Device
|
||||
from lightsync.devices.ws2801 import WS2801Device
|
||||
from lightsync.models.device import DeviceConfig
|
||||
cfg = DeviceConfig(name='test', strip_type='sk6812', led_count=10, ip='127.0.0.1', port=21324)
|
||||
sk = SK6812Device(cfg)
|
||||
result = sk.encode_animation_cmd('solid_color', {'color': [255,0,0], 'white': 128})
|
||||
assert len(result) == 16 and result[0] == 0xAC, f'SK6812 cmd failed: {result.hex()}'
|
||||
print(f'SK6812 encode_animation_cmd: {result.hex()} OK')
|
||||
|
||||
cfg2 = DeviceConfig(name='test2', strip_type='ws2801', led_count=10, ip='127.0.0.1', port=21324)
|
||||
ws = WS2801Device(cfg2)
|
||||
result2 = ws.encode_animation_cmd('chase', {'color': [0,255,0], 'speed': 0.7})
|
||||
assert len(result2) == 16 and result2[0] == 0xAC, f'WS2801 cmd failed: {result2.hex()}'
|
||||
print(f'WS2801 encode_animation_cmd: {result2.hex()} OK')
|
||||
print('ALL OK')
|
||||
"</automated>
|
||||
</verify>
|
||||
<acceptance_criteria>
|
||||
- lightsync/protocol/udp_sender.py contains "class UDPSender" and "class _NullProtocol"
|
||||
- lightsync/protocol/udp_sender.py contains "def send(self, payload: bytes, ip: str, port: int)"
|
||||
- lightsync/protocol/udp_sender.py contains 'local_addr=("0.0.0.0", 0)'
|
||||
- lightsync/devices/sk6812.py contains "from lightsync.protocol.animation_cmd import encode_animation_cmd"
|
||||
- lightsync/devices/sk6812.py does NOT contain 'return b""'
|
||||
- lightsync/devices/ws2801.py contains "from lightsync.protocol.animation_cmd import encode_animation_cmd"
|
||||
- lightsync/devices/ws2801.py does NOT contain 'return b""'
|
||||
- SK6812Device.encode_animation_cmd("solid_color", {"color":[255,0,0]}) returns 16 bytes starting with 0xAC
|
||||
- WS2801Device.encode_animation_cmd("chase", {"color":[0,255,0]}) returns 16 bytes starting with 0xAC
|
||||
</acceptance_criteria>
|
||||
<done>UDPSender class ready for integration into FastAPI lifespan. Both SK6812Device and WS2801Device produce real 16-byte animation command packets instead of empty stubs.</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- `python -m pytest tests/test_protocol.py -v` — all protocol encoding tests pass
|
||||
- `python -c "from lightsync.protocol.drgb import encode_drgb, encode_drgbw, encode_dnrgb_packets; print('imports OK')"` — no import errors
|
||||
- `python -c "from lightsync.protocol.udp_sender import UDPSender; print('UDPSender OK')"` — no import errors
|
||||
- SK6812 and WS2801 encode_animation_cmd return 16-byte packets, not b""
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- WLED DRGB/DRGBW/DNRGB packets match spec byte layout
|
||||
- DNRGB splits frames exceeding MTU into correct chunks with big-endian start indices
|
||||
- Animation command 0xAC packets encode/decode round-trip for all 7 animation types
|
||||
- UDPSender provides async start/stop lifecycle and synchronous send()
|
||||
- Device stubs replaced with real encoding
|
||||
- All tests pass
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/03-communication-protocol/03-01-SUMMARY.md`
|
||||
</output>
|
||||
253
.planning/phases/03-communication-protocol/03-02-PLAN.md
Normal file
253
.planning/phases/03-communication-protocol/03-02-PLAN.md
Normal file
@@ -0,0 +1,253 @@
|
||||
---
|
||||
phase: 03-communication-protocol
|
||||
plan: 02
|
||||
type: execute
|
||||
wave: 1
|
||||
depends_on: []
|
||||
files_modified:
|
||||
- lightsync/animations/__init__.py
|
||||
- lightsync/animations/base.py
|
||||
- lightsync/animations/solid_color.py
|
||||
- lightsync/animations/chase.py
|
||||
- lightsync/animations/pulse.py
|
||||
- lightsync/animations/rainbow.py
|
||||
- lightsync/animations/strobe.py
|
||||
- lightsync/animations/color_wipe.py
|
||||
- lightsync/animations/fire.py
|
||||
- tests/test_animations.py
|
||||
autonomous: true
|
||||
requirements: [ANI-01, ANI-02]
|
||||
|
||||
must_haves:
|
||||
truths:
|
||||
- "All 7 animation types render frames as list of RGB tuples with length == led_count"
|
||||
- "Each animation accepts speed, color, and type-specific params from a dict"
|
||||
- "Fire animation uses Fire2012 algorithm with cooling/sparking params"
|
||||
- "Rainbow animation produces distinct hues across the strip"
|
||||
- "Strobe alternates between on (color) and off (black) based on duty_cycle"
|
||||
- "Chase moves a lit segment along the strip over time"
|
||||
- "All animations can be constructed from a params dict via from_params()"
|
||||
artifacts:
|
||||
- path: "lightsync/animations/base.py"
|
||||
provides: "AnimationBase ABC"
|
||||
exports: ["AnimationBase"]
|
||||
- path: "lightsync/animations/__init__.py"
|
||||
provides: "Animation registry and factory"
|
||||
exports: ["ANIMATION_REGISTRY", "create_animation"]
|
||||
- path: "lightsync/animations/fire.py"
|
||||
provides: "Fire2012 animation"
|
||||
contains: "class FireAnimation"
|
||||
- path: "tests/test_animations.py"
|
||||
provides: "Animation render tests"
|
||||
min_lines: 60
|
||||
key_links:
|
||||
- from: "lightsync/animations/__init__.py"
|
||||
to: "lightsync/animations/*.py"
|
||||
via: "ANIMATION_REGISTRY dict"
|
||||
pattern: "ANIMATION_REGISTRY.*=.*{.*solid_color.*chase.*fire"
|
||||
- from: "lightsync/animations/base.py"
|
||||
to: "all animation subclasses"
|
||||
via: "ABC inheritance"
|
||||
pattern: "class AnimationBase"
|
||||
---
|
||||
|
||||
<objective>
|
||||
Build the animation library with all 7 built-in types. Each animation is a pure renderer: given time t and led_count, it returns a pixel frame. No I/O, no networking — just computation.
|
||||
|
||||
Purpose: These animations generate the pixel data that the protocol layer (Plan 01) encodes and sends. Phase 4 timeline editor will reference these by name. Phase 5 live execution will call render() in real-time.
|
||||
Output: lightsync/animations/ package with 7 animation classes, a registry/factory, and passing tests.
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@$HOME/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/03-communication-protocol/03-RESEARCH.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Animation parameter shapes from RESEARCH.md that each class must accept -->
|
||||
|
||||
solid_color: {"color": [r,g,b], "white": 0}
|
||||
chase: {"color": [r,g,b], "white": 0, "bg_color": [0,0,0], "speed": 0.5, "size": 3, "spacing": 7, "reverse": False}
|
||||
pulse: {"color": [r,g,b], "white": 0, "speed": 0.5, "min_brightness": 0, "max_brightness": 255}
|
||||
rainbow: {"speed": 0.5, "period": 1.0}
|
||||
strobe: {"color": [r,g,b], "white": 0, "speed": 0.5, "duty_cycle": 0.1}
|
||||
color_wipe: {"color": [r,g,b], "white": 0, "speed": 0.5, "reverse": False}
|
||||
fire: {"cooling": 55, "sparking": 120, "speed": 0.5}
|
||||
|
||||
From lightsync/models/show.py:
|
||||
```python
|
||||
class CueModel(BaseModel):
|
||||
animation: str | None = None
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: AnimationBase ABC + 7 animation implementations</name>
|
||||
<files>lightsync/animations/__init__.py, lightsync/animations/base.py, lightsync/animations/solid_color.py, lightsync/animations/chase.py, lightsync/animations/pulse.py, lightsync/animations/rainbow.py, lightsync/animations/strobe.py, lightsync/animations/color_wipe.py, lightsync/animations/fire.py, tests/test_animations.py</files>
|
||||
<read_first>
|
||||
- .planning/phases/03-communication-protocol/03-RESEARCH.md (Animation Parameter Shapes section, Fire2012 algorithm, Pattern 4: Animation Base)
|
||||
- lightsync/models/show.py (CueModel.params shape)
|
||||
</read_first>
|
||||
<behavior>
|
||||
- AnimationBase is ABC with abstract render(t, led_count) and from_params(params)
|
||||
- SolidColorAnimation.render(0.0, 10) returns 10 identical (r,g,b) tuples matching params["color"]
|
||||
- ChaseAnimation.render(t, 10) returns list of 10 tuples, some matching color and some matching bg_color
|
||||
- ChaseAnimation with reverse=True moves opposite direction vs reverse=False at same t
|
||||
- PulseAnimation.render(0.0, 5) returns 5 tuples with brightness based on sin wave at t=0
|
||||
- RainbowAnimation.render(0.0, 10) returns 10 distinct RGB tuples (hue spread across strip)
|
||||
- StrobeAnimation with duty_cycle=0.5 at t=0.0 returns color pixels (on-phase), at t where off returns black
|
||||
- ColorWipeAnimation.render(t, 10) returns mix of color and black pixels depending on t
|
||||
- FireAnimation.render(t, 30) returns 30 RGB tuples, all values in 0-255 range
|
||||
- All render() return lists with length == led_count
|
||||
- All from_params(params_dict) construct valid animation instances
|
||||
- create_animation("chase", {...}) returns a ChaseAnimation instance
|
||||
- create_animation("unknown_type", {}) raises ValueError
|
||||
</behavior>
|
||||
<action>
|
||||
Create `lightsync/animations/base.py`:
|
||||
```python
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
class AnimationBase(ABC):
|
||||
@abstractmethod
|
||||
def render(self, t: float, led_count: int) -> list[tuple]:
|
||||
"""Render frame at time t. Returns list of (R,G,B) tuples, len == led_count."""
|
||||
...
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def from_params(cls, params: dict) -> "AnimationBase":
|
||||
"""Construct from CueModel.params dict."""
|
||||
...
|
||||
```
|
||||
|
||||
Create `lightsync/animations/solid_color.py` — SolidColorAnimation:
|
||||
- `__init__(self, color: tuple[int,int,int], white: int = 0)`
|
||||
- `render(t, led_count)` returns `[self.color] * led_count` (time-independent)
|
||||
- `from_params(params)` reads `params.get("color", [0,0,0])` and `params.get("white", 0)`
|
||||
|
||||
Create `lightsync/animations/chase.py` — ChaseAnimation:
|
||||
- `__init__(self, color, bg_color, speed, size, spacing, reverse, white)`
|
||||
- `render(t, led_count)`: Calculate offset = `int(t * speed * 60)` (60 px/sec at speed=1.0). Pattern repeats every `size + spacing` pixels. For each LED index i, compute `pos = (i - offset) % period` if not reverse, `(i + offset) % period` if reverse. If `pos < size`, pixel = color, else bg_color.
|
||||
- `from_params` reads all chase params with defaults: color=[255,255,255], bg_color=[0,0,0], speed=0.5, size=3, spacing=7, reverse=False, white=0
|
||||
|
||||
Create `lightsync/animations/pulse.py` — PulseAnimation:
|
||||
- `__init__(self, color, speed, min_brightness, max_brightness, white)`
|
||||
- `render(t, led_count)`: speed maps to period: `period = 0.1 + (1.0 - speed) * 4.9` seconds (speed=1.0 -> 0.1s fast, speed=0.0 -> 5.0s slow). Brightness = `min_b + (max_b - min_b) * (0.5 + 0.5 * math.sin(2 * math.pi * t / period))`. Scale each channel of color by brightness/255. Return uniform list.
|
||||
- `from_params` reads pulse params with defaults
|
||||
|
||||
Create `lightsync/animations/rainbow.py` — RainbowAnimation:
|
||||
- `__init__(self, speed, period)`
|
||||
- `render(t, led_count)`: For each LED i, hue = `(i / led_count / period + t * speed) % 1.0`. Convert to RGB via `colorsys.hsv_to_rgb(hue, 1.0, 1.0)`, scale to 0-255 int.
|
||||
- `from_params` reads speed=0.5, period=1.0
|
||||
|
||||
Create `lightsync/animations/strobe.py` — StrobeAnimation:
|
||||
- `__init__(self, color, speed, duty_cycle, white)`
|
||||
- `render(t, led_count)`: `freq = 1 + speed * 19` (1-20 Hz). `phase = (t * freq) % 1.0`. If `phase < duty_cycle` -> color, else -> (0,0,0).
|
||||
- `from_params` reads strobe params
|
||||
|
||||
Create `lightsync/animations/color_wipe.py` — ColorWipeAnimation:
|
||||
- `__init__(self, color, speed, reverse, white)`
|
||||
- `render(t, led_count)`: `fill_count = min(led_count, int(t * speed * 60))`. If not reverse: first fill_count pixels = color, rest = black. If reverse: last fill_count pixels = color, rest = black.
|
||||
- `from_params` reads color_wipe params
|
||||
|
||||
Create `lightsync/animations/fire.py` — FireAnimation using Fire2012:
|
||||
- `__init__(self, cooling, sparking, speed)` — also init `self._heat: list[int] = []` (lazy-init per led_count)
|
||||
- `render(t, led_count)`: Lazy-init heat array if empty or wrong length. Compute `steps = max(1, int(speed * 3))` steps per render call (speed multiplier). For each step, run Fire2012: cool, drift up, spark. Map heat to color: heat<85 -> (h*3, 0, 0), heat<170 -> (255, (h-85)*3, 0), else (255, 255, (h-170)*3). Clamp all to 0-255.
|
||||
- Use `random` module (import at top). Index 0 = bottom (heat source).
|
||||
- `from_params` reads cooling=55, sparking=120, speed=0.5
|
||||
|
||||
Create `lightsync/animations/__init__.py`:
|
||||
```python
|
||||
from lightsync.animations.base import AnimationBase
|
||||
from lightsync.animations.solid_color import SolidColorAnimation
|
||||
from lightsync.animations.chase import ChaseAnimation
|
||||
from lightsync.animations.pulse import PulseAnimation
|
||||
from lightsync.animations.rainbow import RainbowAnimation
|
||||
from lightsync.animations.strobe import StrobeAnimation
|
||||
from lightsync.animations.color_wipe import ColorWipeAnimation
|
||||
from lightsync.animations.fire import FireAnimation
|
||||
|
||||
ANIMATION_REGISTRY: dict[str, type[AnimationBase]] = {
|
||||
"solid_color": SolidColorAnimation,
|
||||
"chase": ChaseAnimation,
|
||||
"pulse": PulseAnimation,
|
||||
"rainbow": RainbowAnimation,
|
||||
"strobe": StrobeAnimation,
|
||||
"color_wipe": ColorWipeAnimation,
|
||||
"fire": FireAnimation,
|
||||
}
|
||||
|
||||
def create_animation(name: str, params: dict) -> AnimationBase:
|
||||
cls = ANIMATION_REGISTRY.get(name)
|
||||
if cls is None:
|
||||
raise ValueError(f"Unknown animation: {name}")
|
||||
return cls.from_params(params)
|
||||
```
|
||||
|
||||
Create `tests/test_animations.py` with pytest tests for all behaviors above. Key tests:
|
||||
- test_solid_color_uniform: all pixels same color
|
||||
- test_chase_pattern: some pixels match color, some bg_color
|
||||
- test_chase_reverse: different from non-reverse at same t
|
||||
- test_pulse_brightness_range: pixel values between min and max brightness scaled
|
||||
- test_rainbow_distinct_hues: 10-LED strip has distinct pixel values
|
||||
- test_strobe_on_off: at certain t returns color, at other t returns black
|
||||
- test_color_wipe_progressive: more pixels filled at larger t
|
||||
- test_fire_output_range: all R,G,B values in 0-255
|
||||
- test_fire_length: output length == led_count
|
||||
- test_all_render_length: every animation returns list of length == led_count
|
||||
- test_create_animation_factory: create_animation returns correct type
|
||||
- test_create_animation_unknown: raises ValueError
|
||||
- test_from_params_all_types: from_params works for all 7 types with default-ish params
|
||||
</action>
|
||||
<verify>
|
||||
<automated>cd /home/claude/led2 && python -m pytest tests/test_animations.py -v</automated>
|
||||
</verify>
|
||||
<acceptance_criteria>
|
||||
- lightsync/animations/__init__.py contains "ANIMATION_REGISTRY" and "def create_animation("
|
||||
- lightsync/animations/__init__.py contains all 7 names: "solid_color", "chase", "pulse", "rainbow", "strobe", "color_wipe", "fire"
|
||||
- lightsync/animations/base.py contains "class AnimationBase(ABC)" and "def render(" and "def from_params("
|
||||
- lightsync/animations/solid_color.py contains "class SolidColorAnimation(AnimationBase)"
|
||||
- lightsync/animations/chase.py contains "class ChaseAnimation(AnimationBase)"
|
||||
- lightsync/animations/pulse.py contains "class PulseAnimation(AnimationBase)" and "math.sin"
|
||||
- lightsync/animations/rainbow.py contains "class RainbowAnimation(AnimationBase)" and "colorsys.hsv_to_rgb"
|
||||
- lightsync/animations/strobe.py contains "class StrobeAnimation(AnimationBase)"
|
||||
- lightsync/animations/color_wipe.py contains "class ColorWipeAnimation(AnimationBase)"
|
||||
- lightsync/animations/fire.py contains "class FireAnimation(AnimationBase)" and "cooling" and "sparking"
|
||||
- tests/test_animations.py exits 0 with all tests passing
|
||||
- tests/test_animations.py contains at least 10 test functions
|
||||
</acceptance_criteria>
|
||||
<done>All 7 animation types render correctly. Factory function creates animations by name. All RGB values in valid 0-255 range. Output list length always matches led_count. All tests pass.</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- `python -m pytest tests/test_animations.py -v` — all animation tests pass
|
||||
- `python -c "from lightsync.animations import create_animation, ANIMATION_REGISTRY; print(len(ANIMATION_REGISTRY), 'animations')"` outputs "7 animations"
|
||||
- `python -c "from lightsync.animations import create_animation; a = create_animation('fire', {'cooling': 55, 'sparking': 120}); f = a.render(1.0, 30); assert len(f) == 30; print('fire OK')"` passes
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- All 7 animation types implemented with correct parameter handling
|
||||
- AnimationBase ABC enforces render() and from_params() contract
|
||||
- ANIMATION_REGISTRY maps all 7 names to classes
|
||||
- create_animation() factory works for all types, raises ValueError for unknown
|
||||
- Fire2012 algorithm produces heat-mapped colors
|
||||
- All render() outputs have correct length and valid 0-255 RGB values
|
||||
- All tests pass
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/03-communication-protocol/03-02-SUMMARY.md`
|
||||
</output>
|
||||
374
.planning/phases/03-communication-protocol/03-03-PLAN.md
Normal file
374
.planning/phases/03-communication-protocol/03-03-PLAN.md
Normal file
@@ -0,0 +1,374 @@
|
||||
---
|
||||
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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
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.
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@$HOME/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<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.md
|
||||
|
||||
<interfaces>
|
||||
<!-- From Plan 01 outputs -->
|
||||
|
||||
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 Plan 02 outputs -->
|
||||
|
||||
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()
|
||||
```
|
||||
</interfaces>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: UDP simulator with packet logging</name>
|
||||
<files>lightsync/protocol/simulator.py, tests/test_e2e_protocol.py</files>
|
||||
<read_first>
|
||||
- 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)
|
||||
</read_first>
|
||||
<behavior>
|
||||
- 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
|
||||
</behavior>
|
||||
<action>
|
||||
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
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>cd /home/claude/led2 && python -m pytest tests/test_e2e_protocol.py -v</automated>
|
||||
</verify>
|
||||
<acceptance_criteria>
|
||||
- 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"
|
||||
</acceptance_criteria>
|
||||
<done>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`.</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 2: Integrate UDPSender into FastAPI lifespan</name>
|
||||
<files>lightsync/main.py</files>
|
||||
<read_first>
|
||||
- lightsync/main.py (current lifespan function — app.state.beats pattern)
|
||||
- lightsync/protocol/udp_sender.py (UDPSender start/stop lifecycle)
|
||||
</read_first>
|
||||
<action>
|
||||
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.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>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)
|
||||
"</automated>
|
||||
</verify>
|
||||
<acceptance_criteria>
|
||||
- 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)
|
||||
</acceptance_criteria>
|
||||
<done>UDPSender starts on app boot and stops on shutdown. Available to all routes via request.app.state.udp_sender. All existing lifespan behavior unchanged.</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- `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
|
||||
</verification>
|
||||
|
||||
<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>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/03-communication-protocol/03-03-SUMMARY.md`
|
||||
</output>
|
||||
Reference in New Issue
Block a user