Files
lighting-controller/test/send_json.py

159 lines
5.1 KiB
Python
Executable File

#!/usr/bin/env python3
"""
CLI to send JSON over SPI to the ESP32-C3 SPI slave.
Legacy led-bar format (current branch expects):
{
"d": { "t": "b|u", "pt": "wave", "br": 128, "dl": 50, "cl": [[255,0,0]], "n1": 0, "n2": 0, "n3": 0, "s": 0 },
"<deviceName>": { "pt": ..., "br": ..., "dl": ..., "cl": [[255,0,0]], "n1": ..., "n2": ..., "n3": ..., "s": ... }
}
Examples:
./send_json.py --type b --pattern wave --brightness 128 --delay 50
./send_json.py --type u --name led-1234 --colors ff0000,00ff00
./send_json.py --data '{"d":{"t":"b","pt":"off","br":100}}'
./send_json.py --file payload.json
"""
import argparse
import json
import os
import sys
import re
# Ensure we can import the local test helper
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if SCRIPT_DIR not in sys.path:
sys.path.insert(0, SCRIPT_DIR)
from spi_master_test import SPIMasterTest # noqa: E402
HEX6_RE = re.compile(r"^[0-9a-fA-F]{6}$")
def parse_hex6_to_rgb(value: str):
v = value.strip()
if v.startswith("0x") or v.startswith("0X"):
v = v[2:]
if v.startswith("#"):
v = v[1:]
if not HEX6_RE.match(v):
raise ValueError(f"Invalid hex color: {value}")
return [int(v[0:2], 16), int(v[2:4], 16), int(v[4:6], 16)]
def build_payload_from_args(args: argparse.Namespace) -> dict:
if args.data:
try:
return json.loads(args.data)
except Exception as exc:
print(f"Invalid JSON in --data: {exc}")
sys.exit(2)
if args.file:
try:
with open(args.file, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as exc:
print(f"Failed to read JSON file {args.file}: {exc}")
sys.exit(2)
defaults: dict = {"t": args.type}
if args.pattern:
defaults["pt"] = args.pattern
if args.brightness is not None:
defaults["br"] = int(args.brightness)
if args.delay is not None:
defaults["dl"] = int(args.delay)
if args.n1 is not None:
defaults["n1"] = int(args.n1)
if args.n2 is not None:
defaults["n2"] = int(args.n2)
if args.n3 is not None:
defaults["n3"] = int(args.n3)
if args.step is not None:
defaults["s"] = int(args.step)
# Colors as RGB triplets
try:
if args.colors:
raw_list = [c for part in args.colors.split(',') for c in [part.strip()] if c]
defaults["cl"] = [parse_hex6_to_rgb(c) for c in raw_list if c]
except ValueError as exc:
print(str(exc), file=sys.stderr)
sys.exit(2)
payload: dict = {"d": defaults}
if args.name:
override: dict = {}
for key in ("pt", "br", "dl", "cl", "n1", "n2", "n3", "s"):
if key in defaults:
override[key] = defaults[key]
payload[args.name] = override
if args.name_value:
try:
name, value = args.name_value
override_obj = json.loads(value)
if not isinstance(override_obj, dict):
raise ValueError("--name-value JSON must be an object")
payload[name] = override_obj
except Exception as exc:
print(f"Invalid --name-value: {exc}")
sys.exit(2)
if not defaults and args.name is None and args.name_value is None:
print("No data specified to send.", file=sys.stderr)
return {"d": {"t": args.type}}
return payload
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Send JSON over SPI to ESP32-C3 (led-bar legacy format)")
src = p.add_mutually_exclusive_group()
src.add_argument("--data", help="Raw JSON string to send (passthrough)")
src.add_argument("--file", help="Path to JSON file to send (passthrough)")
p.add_argument("--type", choices=["b", "u"], default="b", help="Message type: b=beat, u=update (default: b)")
p.add_argument("--pattern", help="Pattern name (maps to pt)")
p.add_argument("--brightness", type=int, help="Brightness (br)")
p.add_argument("--delay", type=int, help="Delay (dl)")
p.add_argument("--n1", type=int, help="n1 parameter")
p.add_argument("--n2", type=int, help="n2 parameter")
p.add_argument("--n3", type=int, help="n3 parameter")
p.add_argument("--step", type=int, help="Pattern step override (s)")
p.add_argument("--colors", help="Comma-separated list for cl as RGB (e.g. ff0000,00ff00,0000ff)")
p.add_argument("--name", help="Device name key for per-bar override (e.g. led-1234)")
p.add_argument("--name-value", nargs=2, metavar=("NAME", "JSON"), help="Inject JSON under NAME key")
p.add_argument("--bus", type=int, default=0, help="SPI bus (default 0)")
p.add_argument("--device", type=int, default=0, help="SPI device/CE (default 0)")
p.add_argument("--speed", type=int, default=1_000_000, help="SPI speed Hz (default 1MHz)")
return p.parse_args()
def main() -> int:
args = parse_args()
payload = build_payload_from_args(args)
spi = SPIMasterTest(bus=args.bus, device=args.device, max_speed_hz=args.speed)
try:
spi.send_json(payload)
finally:
spi.cleanup()
return 0
if __name__ == "__main__":
raise SystemExit(main())