Signal scope

cores/signal/signal_scope/tool/capture_plot.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+import argparse
+import sys
+import time
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+
+
+def _add_bridge_module_path() -> None:
+    here = Path(__file__).resolve()
+    bridge_tool = here.parents[3] / "wb" / "jtag_wb_bridge" / "tool"
+    sys.path.insert(0, str(bridge_tool))
+
+
+def _to_signed(value: int, width: int) -> int:
+    if width <= 0:
+        return value
+    sign_bit = 1 << (width - 1)
+    mask = (1 << width) - 1
+    value &= mask
+    return value - (1 << width) if (value & sign_bit) else value
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Arm signal_scope once, dump samples over JTAG WB, and plot them."
+    )
+    parser.add_argument("--port", type=int, default=0, help="Digilent device index")
+    parser.add_argument("--chain", type=int, default=1, help="JTAG USER chain")
+    parser.add_argument("--selector", type=str, default=None, help="Optional device selector string")
+    parser.add_argument(
+        "--depth",
+        type=int,
+        default=1024,
+        help="Number of scope frames to read (must match RTL depth)",
+    )
+    parser.add_argument(
+        "--wait-s",
+        type=float,
+        default=0.05,
+        help="Seconds to wait after arm/dearm before reading",
+    )
+    parser.add_argument(
+        "--unsigned",
+        action="store_true",
+        help="Plot samples as unsigned (default: signed two's complement)",
+    )
+    parser.add_argument("--out", type=str, default=None, help="Optional PNG output path")
+    parser.add_argument(
+        "--dump-csv",
+        type=str,
+        default=None,
+        help="Optional CSV output path with columns: index,value",
+    )
+    parser.add_argument(
+        "--interactive",
+        action="store_true",
+        help="Keep running: press Enter to recapture/replot in the same window",
+    )
+    return parser.parse_args()
+
+
+def capture_once(bridge, args: argparse.Namespace) -> list[tuple[int, int, int, int]]:
+    samples = []
+    frame_count = args.depth
+    print("[signal_scope] Arming scope (set_reset=1 -> 0)...")
+    bridge.set_reset(True)
+    bridge.set_reset(False)
+    if args.wait_s > 0:
+        print(f"[signal_scope] Waiting {args.wait_s:.3f}s for capture to complete...")
+        time.sleep(args.wait_s)
+
+    print(f"[signal_scope] Reading back {frame_count} frames...")
+    for idx in range(frame_count):
+        base = idx * 8
+        low = bridge.read32(base)
+        high = bridge.read32(base + 4)
+
+        ch_a = low & 0xFFFF
+        ch_b = (low >> 16) & 0xFFFF
+        ch_c = high & 0xFFFF
+        ch_d = (high >> 16) & 0xFFFF
+        if not args.unsigned:
+            ch_a = _to_signed(ch_a, 16)
+            ch_b = _to_signed(ch_b, 16)
+            ch_c = _to_signed(ch_c, 16)
+            ch_d = _to_signed(ch_d, 16)
+        samples.append((ch_a, ch_b, ch_c, ch_d))
+        if idx and (idx % max(1, frame_count // 10) == 0):
+            pct = (100 * idx) // frame_count
+            print(f"[signal_scope] Read progress: {pct}% ({idx}/{frame_count})")
+    print(f"[signal_scope] Read complete: {len(samples)} frames")
+    return samples
+
+
+def write_csv(samples: list[tuple[int, int, int, int]], csv_path: Path) -> None:
+    print(f"[signal_scope] Writing CSV to {csv_path}...")
+    with csv_path.open("w", encoding="utf-8") as f:
+        f.write("index,ch_a,ch_b,ch_c,ch_d\n")
+        for idx, values in enumerate(samples):
+            f.write(f"{idx},{values[0]},{values[1]},{values[2]},{values[3]}\n")
+    print(f"Wrote CSV: {csv_path}")
+
+
+def plot_samples(ax, samples: list[tuple[int, int, int, int]], args: argparse.Namespace, capture_idx: int) -> None:
+    series = [[], [], [], []]
+    for ch_a, ch_b, ch_c, ch_d in samples:
+        series[0].append(ch_a)
+        series[1].append(ch_b)
+        series[2].append(ch_c)
+        series[3].append(ch_d)
+
+    ax.cla()
+    ax.plot(series[0], linewidth=1, label="ch_a")
+    ax.plot(series[1], linewidth=1, label="ch_b")
+    ax.plot(series[2], linewidth=1, label="ch_c")
+    ax.plot(series[3], linewidth=1, label="ch_d")
+    ax.set_title(f"signal_scope_q15 capture #{capture_idx} (depth={args.depth}, chain={args.chain})")
+    ax.set_xlabel("Sample")
+    ax.set_ylabel("Value")
+    if not args.unsigned:
+        ax.set_ylim([-2**15, 2**15])
+    ax.grid(True, alpha=0.3)
+    ax.legend(loc="upper right")
+
+
+def main() -> int:
+    args = parse_args()
+
+    if args.depth <= 0:
+        raise ValueError("--depth must be > 0")
+
+    _add_bridge_module_path()
+    from libjtag_wb_bridge.jtag_bridge import JtagBridge  # pylint: disable=import-error
+
+    print(
+        f"[signal_scope] Starting capture: port={args.port}, chain={args.chain}, "
+        f"depth={args.depth}, selector={args.selector!r}"
+    )
+
+    with JtagBridge() as bridge:
+        print("[signal_scope] Opening JTAG bridge...")
+        if args.selector:
+            bridge.open_selector(args.selector, port=args.port, chain=args.chain)
+        else:
+            bridge.open(port=args.port, chain=args.chain)
+        print("[signal_scope] Bridge opened")
+
+        print("[signal_scope] Clearing bridge flags and sending ping...")
+        bridge.clear_flags()
+        bridge.ping()
+        print("[signal_scope] Bridge ready")
+
+        fig, ax = plt.subplots(figsize=(12, 4))
+        capture_idx = 1
+
+        while True:
+            print(f"[signal_scope] Capture cycle #{capture_idx}")
+            samples = capture_once(bridge, args)
+            plot_samples(ax, samples, args, capture_idx)
+            fig.tight_layout()
+            fig.canvas.draw_idle()
+            fig.canvas.flush_events()
+
+            if args.dump_csv:
+                write_csv(samples, Path(args.dump_csv))
+
+            if args.out:
+                out_path = Path(args.out)
+                print(f"[signal_scope] Saving plot to {out_path}...")
+                fig.savefig(out_path, dpi=150)
+                print(f"Wrote plot: {out_path}")
+
+            if not args.interactive:
+                break
+
+            plt.show(block=False)
+            answer = input("[signal_scope] Press Enter to recapture, or 'q' + Enter to quit: ")
+            if answer.strip().lower().startswith("q"):
+                break
+            capture_idx += 1
+
+        if not args.out:
+            print("[signal_scope] Showing plot window...")
+            plt.show()
+
+    print("[signal_scope] Done")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())