#!/usr/bin/env python3 """Build a JSON serial_scenario for broad visible panel sweeps.""" from __future__ import annotations import argparse import json from pathlib import Path from typing import Iterable CHECKSUM_SEED = 0x5A CONNECT_OK_FRAME = "00 00 00 80 00 DA" SLAVE_POSITIVE_CONTROL = "00 00 13 80 00 C9" SLAVE_CLEAR = "00 00 13 00 00 49" def main() -> int: args = build_arg_parser().parse_args() values = parse_int_list(args.values) skips = set(parse_int_list(args.skip)) if args.skip else set() selectors = [ selector for selector in range(parse_int(args.start), parse_int(args.end) + 1) if selector not in skips and (args.include_selector_zero or selector != 0) ] if not selectors: raise SystemExit("selector range is empty after skips") scenario = build_scenario(args, selectors, values) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(scenario, indent=2) + "\n", encoding="utf-8") candidate_images = len(selectors) * len(values) windows = window_count(len(selectors), args.power_cycle_every) print(f"wrote {args.output}") print(f"selectors={len(selectors)} values={len(values)} candidate_snapshots={candidate_images}") print(f"power_cycle_windows={windows} estimated_candidate_time={candidate_images * args.listen / 60:.1f}min") return 0 def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Generate a serial_scenario JSON file for a broad panel-output webcam sweep." ) parser.add_argument("output", type=Path, help="scenario JSON path to write") parser.add_argument("--start", default="0x0001", help="first logical selector, inclusive") parser.add_argument("--end", default="0x017F", help="last logical selector, inclusive") parser.add_argument( "--values", default="0x8000,0x4000,0x2000,0x1000,0x0800", help="comma-separated 16-bit values to try for each selector", ) parser.add_argument( "--skip", default="", help="comma-separated logical selectors to skip, e.g. 0x006C,0x006D", ) parser.add_argument( "--include-selector-zero", action="store_true", help="include selector 0 in the sweep; omitted by default because it controls CONNECT OK state", ) parser.add_argument("--listen", type=float, default=0.65, help="seconds to listen after each candidate send") parser.add_argument("--clear-listen", type=float, default=0.15, help="seconds after selector clear writes") parser.add_argument("--ok-listen", type=float, default=0.30, help="seconds after CONNECT OK refresh writes") parser.add_argument( "--ok-every", type=int, default=8, help="refresh CONNECT OK every N selectors; use 0 to disable periodic refresh", ) parser.add_argument( "--power-cycle-every", type=int, default=32, help="power-cycle every N selectors to limit latch contamination; use 0 for one long session", ) parser.add_argument("--off-seconds", type=float, default=1.5, help="relay power-off time per window") parser.add_argument("--ready-timeout", type=float, default=10.0, help="wait_ready timeout") parser.add_argument("--drain", type=float, default=0.5, help="seconds to drain after ready") parser.add_argument("--final-listen", type=float, default=1.0, help="seconds to listen at the end") return parser def build_scenario(args: argparse.Namespace, selectors: list[int], values: list[int]) -> dict[str, object]: start = selectors[0] end = selectors[-1] value_text = ",".join(f"0x{value:04X}" for value in values) scenario: dict[str, object] = { "name": f"panel-atlas-big-visual-sweep-{start:03X}-{end:03X}", "notes": [ "Broad visible-output sweep generated by scripts/build_panel_visual_sweep.py.", "Candidate selector/value sends have webcam snapshots enabled; CONNECT OK refreshes and clears do not.", "Use image filenames candidate_XXXX_YYYY to refine any visible trigger into a smaller follow-up scenario.", f"Values: {value_text}", ], "steps": [], } steps: list[dict[str, object]] = scenario["steps"] # type: ignore[assignment] for window_index, window in enumerate(selector_windows(selectors, args.power_cycle_every), start=1): add_session_prelude(args, steps, window_index) if window_index == 1: steps.append(send_step("positive_control_0013_8000_slave_on", SLAVE_POSITIVE_CONTROL, args.listen)) steps.append(send_step("clear_0013_after_slave", SLAVE_CLEAR, args.clear_listen, snapshot=False)) for selector_index, selector in enumerate(window, start=1): if args.ok_every > 0 and (selector_index - 1) % args.ok_every == 0: steps.append( send_step( f"ok_refresh_before_{selector:04X}", CONNECT_OK_FRAME, args.ok_listen, snapshot=False, ) ) for value in values: steps.append( send_step( f"candidate_{selector:04X}_{value:04X}", frame_hex(0x00, selector, value), args.listen, ) ) steps.append( send_step( f"clear_{selector:04X}", frame_hex(0x00, selector, 0), args.clear_listen, snapshot=False, ) ) steps.append({"action": "listen", "seconds": args.final_listen}) return scenario def add_session_prelude(args: argparse.Namespace, steps: list[dict[str, object]], window_index: int) -> None: steps.extend( [ {"action": "power_cycle", "off_seconds": args.off_seconds}, { "action": "wait_ready", "heartbeats": 2, "timeout": args.ready_timeout, "require": True, }, {"action": "drain", "seconds": args.drain}, send_step(f"window_{window_index:03d}_ok_seed_1", CONNECT_OK_FRAME, args.ok_listen, snapshot=False), send_step(f"window_{window_index:03d}_ok_seed_2", CONNECT_OK_FRAME, args.ok_listen, snapshot=False), ] ) def send_step(label: str, frame: str, listen: float, *, snapshot: bool = True) -> dict[str, object]: step: dict[str, object] = { "action": "send", "label": label, "frame": frame, "listen": listen, } if not snapshot: step["snapshot"] = False return step def selector_windows(selectors: list[int], size: int) -> Iterable[list[int]]: if size <= 0: yield selectors return for index in range(0, len(selectors), size): yield selectors[index : index + size] def window_count(selector_count: int, size: int) -> int: if size <= 0: return 1 return (selector_count + size - 1) // size def frame_hex(command: int, selector: int, value: int) -> str: selector_hi, selector_lo = selector_bytes(selector) data = bytes( [ command & 0xFF, selector_hi, selector_lo, (value >> 8) & 0xFF, value & 0xFF, ] ) return " ".join(f"{byte:02X}" for byte in data + bytes([frame_checksum(data)])) def selector_bytes(selector: int) -> tuple[int, int]: selector &= 0x01FF if selector <= 0x007F: return 0x00, selector if selector <= 0x017F: return 0x01, selector - 0x0080 return 0x02, selector - 0x0180 def frame_checksum(data: bytes) -> int: checksum = CHECKSUM_SEED for value in data[:5]: checksum ^= value return checksum & 0xFF def parse_int_list(text: str) -> list[int]: values = [] for part in text.split(","): item = part.strip() if item: values.append(parse_int(item)) return values def parse_int(text: str) -> int: return int(str(text).strip(), 0) if __name__ == "__main__": raise SystemExit(main())