f17e77f7bb
- 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
29 lines
986 B
Python
29 lines
986 B
Python
from lightsync.devices.base import BaseDevice
|
|
from lightsync.models.device import DeviceConfig
|
|
from lightsync.protocol.animation_cmd import encode_animation_cmd as _encode_cmd
|
|
|
|
|
|
class WS2801Device(BaseDevice):
|
|
"""WS2801 RGB LED strip — 3 bytes per pixel (R, G, B)."""
|
|
|
|
def __init__(self, config: DeviceConfig):
|
|
super().__init__(config)
|
|
|
|
@property
|
|
def bytes_per_pixel(self) -> int:
|
|
return 3
|
|
|
|
def encode_frame(self, pixels: list[tuple]) -> bytes:
|
|
"""Encode pixel list to WS2801 frame bytes.
|
|
Each pixel: (r, g, b) tuple — 3 bytes per LED.
|
|
"""
|
|
buf = bytearray()
|
|
for pixel in pixels:
|
|
r, g, b = pixel[0], pixel[1], pixel[2]
|
|
buf.extend([r & 0xFF, g & 0xFF, b & 0xFF])
|
|
return bytes(buf)
|
|
|
|
def encode_animation_cmd(self, animation: str, params: dict) -> bytes:
|
|
"""Encode animation command as 16-byte 0xAC LightSync packet."""
|
|
return _encode_cmd(animation, params)
|