run 13

scripts/serial_button_response_test.py

@@ -9,6 +9,8 @@ This helper can:
 4. Optionally mirror CALL high/low events with matching host responses.
 5. Optionally transmit follow-up frames when a watched response frame appears.
 6. Optionally transmit startup frames without waiting for a button event.
+7. Optionally repeat the startup-frame group to test retrigger/latch behavior.
+8. Optionally read after each startup-frame group before sending the next one.
 
 Known RCP-origin button frames:
 
@@ -128,6 +130,25 @@ def emit_known_counts(emit, label: str, data: bytes) -> None:
         emit(f"{label} raw {hex_preview(data)}")
 
 
+def emit_rx_chunk(emit, args, totals, watch_totals, data: bytes) -> None:
+    stamp = dt.datetime.now().strftime("%H:%M:%S.%f")[:-3]
+    emit(f"{stamp}  RX {len(data):03d} bytes  {hex_preview(data)}")
+    if args.ascii:
+        emit(f"{'':14} ASCII      {ascii_preview(data)}")
+
+    for name, pattern in KNOWN_PATTERNS.items():
+        count = data.count(pattern)
+        if count:
+            totals[name] += count
+            emit(f"{stamp}  DETECT {name} x{count}")
+    for frame in args.watch_frame or []:
+        count = data.count(frame)
+        if count:
+            key = hex_preview(frame)
+            watch_totals[key] += count
+            emit(f"{stamp}  DETECT watch-frame {key} x{count}")
+
+
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description="Listen for CAM POWER/CALL frames and optionally respond to CAM POWER."
@@ -189,6 +210,24 @@ def parse_args() -> argparse.Namespace:
         default=0.05,
         help="delay between multiple startup frames",
     )
+    parser.add_argument(
+        "--startup-repeat",
+        type=int,
+        default=1,
+        help="how many times to send the full startup-frame group",
+    )
+    parser.add_argument(
+        "--startup-repeat-interval",
+        type=float,
+        default=1.0,
+        help="delay between repeated startup-frame groups",
+    )
+    parser.add_argument(
+        "--startup-read-after-group",
+        type=float,
+        default=0.0,
+        help="seconds to read and log RX after each startup-frame group before the next repeat",
+    )
     parser.add_argument("--response-delay", type=float, default=0.05)
     parser.add_argument(
         "--response-frame-interval",
@@ -268,10 +307,18 @@ def main() -> int:
             buffer = bytearray()
             if args.startup_frame:
                 time.sleep(args.startup_delay)
-                for frame_index, frame in enumerate(args.startup_frame):
-                    if frame_index:
-                        time.sleep(args.startup_frame_interval)
-                    send_frame(ser, emit, "startup", frame)
+                for repeat_index in range(args.startup_repeat):
+                    for frame_index, frame in enumerate(args.startup_frame):
+                        if frame_index:
+                            time.sleep(args.startup_frame_interval)
+                        send_frame(ser, emit, "startup", frame)
+                    if args.startup_read_after_group:
+                        group_data = read_window(ser, args.startup_read_after_group)
+                        if group_data:
+                            emit_rx_chunk(emit, args, totals, watch_totals, group_data)
+                            buffer.extend(group_data)
+                    if repeat_index + 1 < args.startup_repeat:
+                        time.sleep(args.startup_repeat_interval)
 
             while time.monotonic() < stop_at:
                 data = ser.read(args.chunk_size)
@@ -282,7 +329,6 @@ def main() -> int:
                 emit(f"{stamp}  RX {len(data):03d} bytes  {hex_preview(data)}")
                 if args.ascii:
                     emit(f"{'':14} ASCII      {ascii_preview(data)}")
-
                 buffer.extend(data)
                 for name, pattern in KNOWN_PATTERNS.items():
                     count = data.count(pattern)