feat(03-01): UDP sender and device stub completion

- UDPSender: single shared asyncio DatagramTransport, start/stop lifecycle, fire-and-forget send()
- _NullProtocol: swallows OS send errors (UDP delivery is best-effort for LEDs)
- SK6812Device.encode_animation_cmd: replaces b'' stub with real 16-byte 0xAC packet
- WS2801Device.encode_animation_cmd: replaces b'' stub with real 16-byte 0xAC packet
- Both devices delegate to encode_animation_cmd from lightsync.protocol.animation_cmd
This commit is contained in:
Claude
2026-04-06 21:42:29 +00:00
parent 5e13da3967
commit f17e77f7bb
3 changed files with 69 additions and 4 deletions

View File

@@ -1,5 +1,6 @@
from lightsync.devices.base import BaseDevice from lightsync.devices.base import BaseDevice
from lightsync.models.device import DeviceConfig from lightsync.models.device import DeviceConfig
from lightsync.protocol.animation_cmd import encode_animation_cmd as _encode_cmd
class SK6812Device(BaseDevice): class SK6812Device(BaseDevice):
@@ -24,5 +25,5 @@ class SK6812Device(BaseDevice):
return bytes(buf) return bytes(buf)
def encode_animation_cmd(self, animation: str, params: dict) -> bytes: def encode_animation_cmd(self, animation: str, params: dict) -> bytes:
"""Stub — Phase 3 implements real animation command encoding.""" """Encode animation command as 16-byte 0xAC LightSync packet."""
return b"" return _encode_cmd(animation, params)

View File

@@ -1,5 +1,6 @@
from lightsync.devices.base import BaseDevice from lightsync.devices.base import BaseDevice
from lightsync.models.device import DeviceConfig from lightsync.models.device import DeviceConfig
from lightsync.protocol.animation_cmd import encode_animation_cmd as _encode_cmd
class WS2801Device(BaseDevice): class WS2801Device(BaseDevice):
@@ -23,5 +24,5 @@ class WS2801Device(BaseDevice):
return bytes(buf) return bytes(buf)
def encode_animation_cmd(self, animation: str, params: dict) -> bytes: def encode_animation_cmd(self, animation: str, params: dict) -> bytes:
"""Stub — Phase 3 implements real animation command encoding.""" """Encode animation command as 16-byte 0xAC LightSync packet."""
return b"" return _encode_cmd(animation, params)

View File

@@ -0,0 +1,63 @@
"""Async UDP transport wrapper for sending LED packets to devices.
Provides a single shared socket that can sendto() multiple device addresses.
One socket handles all devices — UDP is connectionless, no socket pool needed.
Usage:
sender = UDPSender()
await sender.start() # Open socket at app startup
sender.send(payload, ip, port) # Fire-and-forget from async context
await sender.stop() # Close socket at app shutdown
"""
import asyncio
class _NullProtocol(asyncio.DatagramProtocol):
"""Minimal datagram protocol — we only send, never receive on this socket."""
def error_received(self, exc: Exception) -> None:
# UDP send errors are non-fatal for LED strip delivery.
# Network hiccups should not crash the show.
pass
class UDPSender:
"""Single shared UDP socket for sending LED frames and animation commands.
Thread safety: sendto() is NOT thread-safe across OS thread boundaries.
Always call send() from within the asyncio event loop. If bridging from
sync code, use loop.call_soon_threadsafe().
"""
def __init__(self) -> None:
self._transport: asyncio.DatagramTransport | None = None
async def start(self) -> None:
"""Open the UDP socket. Call once during FastAPI lifespan startup."""
loop = asyncio.get_running_loop()
self._transport, _ = await loop.create_datagram_endpoint(
_NullProtocol,
local_addr=("0.0.0.0", 0), # OS assigns ephemeral source port
)
def send(self, payload: bytes, ip: str, port: int) -> None:
"""Send payload to (ip, port). Fire-and-forget — no acknowledgement.
Args:
payload: Raw bytes to send (DRGB/DRGBW/DNRGB/animation command).
ip: Destination IP address string.
port: Destination UDP port (WLED default: 21324).
"""
if self._transport:
self._transport.sendto(payload, (ip, port))
async def stop(self) -> None:
"""Close the UDP socket. Call during FastAPI lifespan shutdown."""
if self._transport:
self._transport.close()
self._transport = None
@property
def is_running(self) -> bool:
"""True if the socket is open and ready to send."""
return self._transport is not None