feat(03-01): implement WLED protocol encoders and animation command packet
- encode_drgb/encode_drgbw: 2-byte header + RGB/RGBW body, truncate at MTU max - encode_dnrgb_packets: splits large frames into 489-LED chunks with big-endian start index - encode_animation_cmd: 16-byte 0xAC packet with speed (float32), color, flags, density - decode_animation_cmd: full inverse round-trip for all 7 animation types - All 27 tests pass
This commit is contained in:
146
lightsync/protocol/animation_cmd.py
Normal file
146
lightsync/protocol/animation_cmd.py
Normal file
@@ -0,0 +1,146 @@
|
||||
"""Custom animation command packet encoder/decoder for LightSync firmware.
|
||||
|
||||
Packet format (16 bytes, all big-endian):
|
||||
Byte 0: Protocol marker = 0xAC (distinguishes from WLED types 0x01-0x04)
|
||||
Byte 1: Protocol version = 1
|
||||
Byte 2: Animation type ID (0-6, see ANIMATION_IDS)
|
||||
Byte 3: Flags (bit 0 = reverse, bits 1-7 reserved)
|
||||
Bytes 4-7: Speed (float32 big-endian, range 0.0-1.0)
|
||||
Bytes 8-10: Primary color R, G, B
|
||||
Bytes 11-13: Secondary (background) color R, G, B
|
||||
Byte 14: White channel (SK6812 W value, 0 for RGB-only strips)
|
||||
Byte 15: Density/length (animation-specific: size, cooling, etc.)
|
||||
"""
|
||||
import struct
|
||||
|
||||
ANIM_CMD_MARKER = 0xAC
|
||||
ANIM_CMD_VERSION = 1
|
||||
|
||||
ANIMATION_IDS: dict[str, int] = {
|
||||
"solid_color": 0,
|
||||
"chase": 1,
|
||||
"pulse": 2,
|
||||
"rainbow": 3,
|
||||
"strobe": 4,
|
||||
"color_wipe": 5,
|
||||
"fire": 6,
|
||||
}
|
||||
|
||||
ANIMATION_NAMES: dict[int, str] = {v: k for k, v in ANIMATION_IDS.items()}
|
||||
|
||||
_PACKET_SIZE = 16
|
||||
|
||||
|
||||
def encode_animation_cmd(animation: str, params: dict) -> bytes:
|
||||
"""Encode an animation command into a 16-byte LightSync packet.
|
||||
|
||||
Args:
|
||||
animation: Animation name (must be a key in ANIMATION_IDS).
|
||||
params: Animation parameters dict. Recognized keys:
|
||||
- color ([R, G, B]): Primary color, default [0, 0, 0]
|
||||
- bg_color ([R, G, B]): Secondary/background color, default [0, 0, 0]
|
||||
- speed (float 0.0-1.0): Normalized speed, default 0.5
|
||||
- white (int 0-255): W channel for SK6812, default 0
|
||||
- reverse (bool): Direction flag, stored in flags bit 0, default False
|
||||
- density/size/cooling (int 0-255): Byte 15, animation-specific meaning
|
||||
|
||||
Returns:
|
||||
16-byte binary packet.
|
||||
|
||||
Raises:
|
||||
ValueError: If animation name is not in ANIMATION_IDS.
|
||||
"""
|
||||
if animation not in ANIMATION_IDS:
|
||||
raise ValueError(
|
||||
f"Unknown animation type: {animation!r}. "
|
||||
f"Valid types: {sorted(ANIMATION_IDS.keys())}"
|
||||
)
|
||||
|
||||
anim_id = ANIMATION_IDS[animation]
|
||||
|
||||
# Flags byte: bit 0 = reverse
|
||||
reverse = bool(params.get("reverse", False))
|
||||
flags = 0x01 if reverse else 0x00
|
||||
|
||||
speed = float(params.get("speed", 0.5))
|
||||
color = params.get("color", [0, 0, 0])
|
||||
bg_color = params.get("bg_color", [0, 0, 0])
|
||||
white = int(params.get("white", 0)) & 0xFF
|
||||
|
||||
# Byte 15: density falls back to size, then cooling, then 0
|
||||
density = int(
|
||||
params.get("density", params.get("size", params.get("cooling", 0)))
|
||||
) & 0xFF
|
||||
|
||||
packet = struct.pack(
|
||||
">BBBBf",
|
||||
ANIM_CMD_MARKER, # 0: marker
|
||||
ANIM_CMD_VERSION, # 1: version
|
||||
anim_id, # 2: animation type ID
|
||||
flags, # 3: flags
|
||||
speed, # 4-7: speed float32
|
||||
)
|
||||
packet += bytes([
|
||||
color[0] & 0xFF, # 8: R primary
|
||||
color[1] & 0xFF, # 9: G primary
|
||||
color[2] & 0xFF, # 10: B primary
|
||||
bg_color[0] & 0xFF, # 11: R secondary
|
||||
bg_color[1] & 0xFF, # 12: G secondary
|
||||
bg_color[2] & 0xFF, # 13: B secondary
|
||||
white, # 14: white channel
|
||||
density, # 15: density/size/cooling
|
||||
])
|
||||
|
||||
assert len(packet) == _PACKET_SIZE, f"Packet size {len(packet)} != {_PACKET_SIZE}"
|
||||
return packet
|
||||
|
||||
|
||||
def decode_animation_cmd(data: bytes) -> dict:
|
||||
"""Decode a LightSync animation command packet.
|
||||
|
||||
Args:
|
||||
data: 16-byte binary packet produced by encode_animation_cmd.
|
||||
|
||||
Returns:
|
||||
Dict with keys:
|
||||
- animation (str): Animation name
|
||||
- params (dict): Decoded parameters (speed, color, bg_color, white,
|
||||
density, reverse, flags)
|
||||
|
||||
Raises:
|
||||
ValueError: If packet is too short, marker is wrong, or animation ID unknown.
|
||||
"""
|
||||
if len(data) < _PACKET_SIZE:
|
||||
raise ValueError(f"Packet too short: {len(data)} bytes, expected {_PACKET_SIZE}")
|
||||
|
||||
marker = data[0]
|
||||
if marker != ANIM_CMD_MARKER:
|
||||
raise ValueError(f"Invalid marker: 0x{marker:02X}, expected 0x{ANIM_CMD_MARKER:02X}")
|
||||
|
||||
_version = data[1]
|
||||
anim_id = data[2]
|
||||
flags = data[3]
|
||||
(speed,) = struct.unpack_from(">f", data, 4)
|
||||
|
||||
if anim_id not in ANIMATION_NAMES:
|
||||
raise ValueError(f"Unknown animation ID: {anim_id}")
|
||||
|
||||
animation = ANIMATION_NAMES[anim_id]
|
||||
color = [data[8], data[9], data[10]]
|
||||
bg_color = [data[11], data[12], data[13]]
|
||||
white = data[14]
|
||||
density = data[15]
|
||||
reverse = bool(flags & 0x01)
|
||||
|
||||
return {
|
||||
"animation": animation,
|
||||
"params": {
|
||||
"speed": speed,
|
||||
"color": color,
|
||||
"bg_color": bg_color,
|
||||
"white": white,
|
||||
"density": density,
|
||||
"reverse": reverse,
|
||||
"flags": flags,
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user