isolating lamp behaviour

h8536/serial_scenario.py

@@ -230,9 +230,13 @@ def _step_table_sweep(ctx: ScenarioContext, spec: dict[str, Any]) -> None:
     selectors = _selector_list(spec)
     gap = float(spec.get("gap", 0.080))
     ack = _ack_config(spec.get("ack_on", {}))
+    ack_note = (
+        "ack=disabled"
+        if not ack["enabled"]
+        else f"ack_targets={len(ack['targets'])} ack_frame={format_frame(ack['frame'])}"
+    )
     ctx.logger.event(
-        f"TABLE_SWEEP selectors={len(selectors)} gap={gap:.3f}s "
-        f"ack_targets={len(ack['targets'])} ack_frame={format_frame(ack['frame'])}"
+        f"TABLE_SWEEP selectors={len(selectors)} gap={gap:.3f}s {ack_note}"
     )
     for selector in selectors:
         if ctx.abort_requested:
@@ -246,7 +250,10 @@ def _step_table_sweep(ctx: ScenarioContext, spec: dict[str, Any]) -> None:
 
 def _ack_config(raw: Any) -> dict[str, Any]:
     spec = raw if isinstance(raw, dict) else {}
+    enabled = bool(spec.get("enabled", True))
     targets = _parse_frame_list(spec.get("frames", spec.get("frame", DEFAULT_ACK_TARGET)))
+    if not enabled:
+        targets = set()
     return {
         "targets": set(targets),
         "frame": _parse_optional_frame(spec.get("ack_frame"), DEFAULT_ACK_FRAME),
@@ -254,7 +261,7 @@ def _ack_config(raw: Any) -> dict[str, Any]:
         "poll_interval": float(spec.get("poll_interval", 0.005)),
         "post_read": float(spec.get("post_ack_read", 0.250)),
         "once_per_selector": bool(spec.get("once_per_selector", True)),
-        "enabled": bool(spec.get("enabled", True)),
+        "enabled": enabled,
         "max_acks": _optional_int(spec.get("max_acks")),
         "max_target_hits": _optional_int(spec.get("max_target_hits")),
         "abort_on_limit": bool(spec.get("abort_on_limit", True)),
@@ -452,23 +459,52 @@ def _print_dry_run(args: argparse.Namespace, scenario: dict[str, Any], log_path:
     for index, step in enumerate(_scenario_steps(scenario), start=1):
         action, spec = _normalize_step(step)
         print(f"step[{index}]={action}", file=stdout)
-        if action == "send":
-            frame = _parse_required_frame(spec.get("frame"))
-            print(f"  frame={format_frame(frame)} checksum_ok={int(frame_checksum_ok(frame))}", file=stdout)
-        elif action == "table_sweep":
-            selectors = _selector_list(spec)
-            ack = _ack_config(spec.get("ack_on", {}))
-            if selectors:
-                first = selectors[0]
-                last = selectors[-1]
-                print(f"  selectors={len(selectors)} first=0x{first:03X} last=0x{last:03X}", file=stdout)
-            else:
-                print("  selectors=0", file=stdout)
-            print(f"  gap={float(spec.get('gap', 0.080)):.3f}", file=stdout)
+        _print_step_dry_run(action, spec, stdout)
+
+
+def _print_step_dry_run(action: str, spec: dict[str, Any], stdout: TextIO, *, indent: str = "  ") -> None:
+    if action == "send":
+        frame = _parse_required_frame(spec.get("frame"))
+        print(f"{indent}frame={format_frame(frame)} checksum_ok={int(frame_checksum_ok(frame))}", file=stdout)
+        if float(spec.get("listen", 0.0)) > 0:
+            print(f"{indent}listen={float(spec.get('listen', 0.0)):.3f}s", file=stdout)
+    elif action in {"drain", "listen", "wait"}:
+        print(f"{indent}seconds={float(spec.get('seconds', spec.get('value', 0.0))):.3f}", file=stdout)
+    elif action == "wait_ready":
+        print(
+            f"{indent}heartbeats={int(spec.get('heartbeats', 2))} "
+            f"timeout={float(spec.get('timeout', 10.0)):.3f}s "
+            f"require={int(bool(spec.get('require', False)))}",
+            file=stdout,
+        )
+    elif action == "table_sweep":
+        selectors = _selector_list(spec)
+        ack = _ack_config(spec.get("ack_on", {}))
+        if selectors:
+            first = selectors[0]
+            last = selectors[-1]
+            print(f"{indent}selectors={len(selectors)} first=0x{first:03X} last=0x{last:03X}", file=stdout)
+        else:
+            print(f"{indent}selectors=0", file=stdout)
+        print(f"{indent}gap={float(spec.get('gap', 0.080)):.3f}", file=stdout)
+        if not ack["enabled"]:
+            print(f"{indent}ack=disabled", file=stdout)
+        else:
             for target in sorted(ack["targets"]):
-                print(f"  ack_target={format_frame(target)}", file=stdout)
-            print(f"  ack_frame={format_frame(ack['frame'])}", file=stdout)
-            print(f"  max_acks={ack['max_acks']} max_target_hits={ack['max_target_hits']}", file=stdout)
+                print(f"{indent}ack_target={format_frame(target)}", file=stdout)
+            print(f"{indent}ack_frame={format_frame(ack['frame'])}", file=stdout)
+            print(f"{indent}max_acks={ack['max_acks']} max_target_hits={ack['max_target_hits']}", file=stdout)
+    elif action == "repeat":
+        count = max(0, int(spec.get("count", 1)))
+        steps = spec.get("steps", [])
+        step_count = len(steps) if isinstance(steps, list) else 0
+        print(f"{indent}count={count} steps={step_count}", file=stdout)
+        if not isinstance(steps, list):
+            return
+        for child_index, child in enumerate(steps, start=1):
+            child_action, child_spec = _normalize_step(child)
+            print(f"{indent}child[{child_index}]={child_action}", file=stdout)
+            _print_step_dry_run(child_action, child_spec, stdout, indent=indent + "  ")
 
 
 def _emit_summary(ctx: ScenarioContext, logger: BenchLogger) -> None: