Jimmy
4fc3f46866
Replace serial/Wi-Fi driver transport paths with WebSocket bridge client, binary espnow_wire delivery, device announce registry, and restructured espnow-sender (AP + broadcast passthrough). Includes docs and tests. Co-authored-by: Cursor <cursoragent@cursor.com>
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""Build binary ESP-NOW CMD / GROUP_CMD packets from preset/select data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from util.binary_envelope import pack_binary_envelope_v2
|
|
from util.espnow_wire import MAX_ESPNOW_PAYLOAD, pack_cmd, pack_group_cmd
|
|
|
|
|
|
def v1_dict_to_cmd_packet(body: Dict[str, Any]) -> bytes:
|
|
save = bool(body.get("save"))
|
|
kw: Dict[str, Any] = {}
|
|
if "presets" in body:
|
|
kw["presets"] = body["presets"]
|
|
if "select" in body:
|
|
kw["select"] = body["select"]
|
|
if "default" in body:
|
|
kw["default"] = body["default"]
|
|
kw["default_targets"] = body.get("targets")
|
|
if "b" in body:
|
|
kw["brightness_0_255"] = int(body["b"])
|
|
return pack_cmd(pack_binary_envelope_v2(**kw), save=save)
|
|
|
|
|
|
def build_preset_cmd_chunks(
|
|
presets_by_name: Dict[str, Any],
|
|
*,
|
|
save: bool = False,
|
|
default: Optional[str] = None,
|
|
max_payload: int = MAX_ESPNOW_PAYLOAD,
|
|
) -> List[bytes]:
|
|
"""Chunk presets into CMD packets each ≤ max_payload bytes."""
|
|
entries = list(presets_by_name.items())
|
|
chunks: List[bytes] = []
|
|
batch: Dict[str, Any] = {}
|
|
|
|
def _packet_for(presets_map: Dict[str, Any], *, final_save: bool, def_id: Optional[str]):
|
|
kw: Dict[str, Any] = {"presets": presets_map}
|
|
if def_id is not None:
|
|
kw["default"] = def_id
|
|
return pack_cmd(pack_binary_envelope_v2(**kw), save=final_save)
|
|
|
|
for name, preset_obj in entries:
|
|
trial = dict(batch)
|
|
trial[name] = preset_obj
|
|
try:
|
|
pkt = _packet_for(trial, final_save=False, def_id=None)
|
|
except ValueError:
|
|
pkt = b"\xff\xff"
|
|
if len(pkt) <= max_payload or not batch:
|
|
batch = trial
|
|
else:
|
|
chunks.append(_packet_for(batch, final_save=False, def_id=None))
|
|
batch = {name: preset_obj}
|
|
|
|
if batch:
|
|
chunks.append(
|
|
_packet_for(batch, final_save=save, def_id=str(default) if default else None),
|
|
)
|
|
|
|
return [c for c in chunks if c and c[0] == 0x4C]
|