From d4fa84d4829fb1e3ed207404ccf12dfa0f46cac2 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 09:07:13 +0200 Subject: [PATCH] test(e2e-chain): report plot generation Co-Authored-By: Claude Opus 4.6 --- Test/E2E/chain/plots.py | 174 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 Test/E2E/chain/plots.py diff --git a/Test/E2E/chain/plots.py b/Test/E2E/chain/plots.py new file mode 100644 index 0000000..e04b67c --- /dev/null +++ b/Test/E2E/chain/plots.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +""" +plots.py — Report figures for the streaming-chain E2E suite. + +For a scenario it reads ``received_.bin`` (RCV1), ``metrics_.json`` and +``checks_.json`` from the artifact dir and writes: + + * ``wave_.png`` — received waveform(s) vs analytic sinusoid fit (truth) + * ``trig_.png`` — trigger signal with threshold + fired-trigger markers + * ``zoom_.png`` — full received signal with the requested zoom spans shaded + +Missing inputs are handled gracefully (a placeholder note is drawn) so the +orchestrator never aborts on a partial scenario. +""" +import argparse +import json +import os +import sys + +import matplotlib +matplotlib.use("Agg") +import matplotlib.pyplot as plt # noqa: E402 +import numpy as np # noqa: E402 + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import scenarios as S # noqa: E402 +import gen_data as G # noqa: E402 +import validate_waveform as V # noqa: E402 + + +def _placeholder(path, msg): + fig, ax = plt.subplots(figsize=(10, 3)) + ax.axis("off") + ax.text(0.5, 0.5, msg, ha="center", va="center", fontsize=12) + fig.savefig(path, dpi=110, bbox_inches="tight") + plt.close(fig) + + +def _load(scenario, d): + sid = scenario["id"] + recv_p = os.path.join(d, f"received_{sid}.bin") + checks_p = os.path.join(d, f"checks_{sid}.json") + recv = V.read_received(recv_p) if os.path.exists(recv_p) else {} + checks = json.load(open(checks_p)) if os.path.exists(checks_p) else {} + return recv, checks + + +def _data_keys(recv, gt): + """Non-time signal keys present in received, base-matched to ground truth.""" + out = [] + for k in sorted(recv): + base = k.split("[")[0] + g = gt.get(base) + if g is None or g.get("is_time"): + continue + out.append((k, base, g)) + return out + + +def plot_wave(scenario, recv, gt, path): + keys = _data_keys(recv, gt) + if not keys: + _placeholder(path, f"{scenario['id']}: no received signal data") + return path + keys = keys[:4] + fig, axes = plt.subplots(len(keys), 1, figsize=(12, 3.0 * len(keys)), + squeeze=False) + for ax, (k, base, g) in zip(axes[:, 0], keys): + t, v = recv[k] + if t.size == 0: + ax.text(0.5, 0.5, f"{k}: empty", ha="center"); continue + t0 = t[0] + ax.plot(t - t0, v, ".", ms=3, label="received", color="tab:blue") + if g["formula"] == "sine" and g["freq"] and t.size >= 8: + corr, nrmse, amp = V.sine_shape(t, v, g["freq"]) + w = 2 * np.pi * g["freq"] + A = np.column_stack([np.sin(w * t), np.cos(w * t), np.ones_like(t)]) + coef, *_ = np.linalg.lstsq(A, v, rcond=None) + ts = np.linspace(t.min(), t.max(), 2000) + As = np.column_stack([np.sin(w * ts), np.cos(w * ts), np.ones_like(ts)]) + ax.plot(ts - t0, As @ coef, "-", lw=1, color="tab:orange", + label=f"fit f={g['freq']}Hz corr={corr:.4f}") + ax.set_title(f"{k} ({g['type']}, quant={g['quant']}, {g['formula']})") + ax.set_xlabel("t − t0 (s)") + ax.legend(loc="upper right", fontsize=8) + ax.grid(alpha=0.3) + fig.suptitle(f"{scenario['id']} — received waveform vs truth") + fig.tight_layout() + fig.savefig(path, dpi=120, bbox_inches="tight") + plt.close(fig) + return path + + +def plot_trigger(scenario, recv, checks, gt, path): + trig = checks.get("trigger", []) + if not trig: + _placeholder(path, f"{scenario['id']}: no trigger checks") + return path + key = trig[0].get("key", "") + if key not in recv or recv[key][0].size == 0: + _placeholder(path, f"{scenario['id']}: trigger signal {key} not recorded") + return path + t, v = recv[key] + t0 = t[0] + thr = float(np.mean(v)) + fig, ax = plt.subplots(figsize=(12, 4)) + ax.plot(t - t0, v, "-", lw=0.7, color="tab:blue", label=key) + ax.axhline(thr, color="gray", ls="--", lw=1, label=f"~threshold {thr:.3g}") + seen = set() + for tr in trig: + if not tr.get("fired"): + continue + tt = tr["trigTime"] - t0 + lbl = f"{tr['edge']}/{tr['mode']}" + ax.axvline(tt, color="tab:red", lw=1, alpha=0.6, + label=("fired" if "fired" not in seen else None)) + seen.add("fired") + ax.set_title(f"{scenario['id']} — trigger captures ({len(trig)} combos)") + ax.set_xlabel("t − t0 (s)") + ax.legend(loc="upper right", fontsize=8) + ax.grid(alpha=0.3) + fig.tight_layout() + fig.savefig(path, dpi=120, bbox_inches="tight") + plt.close(fig) + return path + + +def plot_zoom(scenario, recv, checks, gt, path): + zooms = checks.get("zoom", []) + keys = _data_keys(recv, gt) + if not keys: + _placeholder(path, f"{scenario['id']}: no received data for zoom") + return path + k, base, g = keys[0] + t, v = recv[k] + t0 = t[0] + fig, ax = plt.subplots(figsize=(12, 4)) + ax.plot(t - t0, v, "-", lw=0.6, color="tab:blue", label=k) + for zc in zooms: + rg = zc.get("range", [0, 0]) + ax.axvspan(rg[0] - t0, rg[1] - t0, alpha=0.15, color="tab:green") + ax.set_title(f"{scenario['id']} — zoom spans (n={len(zooms)})") + ax.set_xlabel("t − t0 (s)") + ax.legend(loc="upper right", fontsize=8) + ax.grid(alpha=0.3) + fig.tight_layout() + fig.savefig(path, dpi=120, bbox_inches="tight") + plt.close(fig) + return path + + +def main(): + p = argparse.ArgumentParser(description="Generate E2E chain report plots") + p.add_argument("--scenario", required=True) + p.add_argument("--dir", required=True) + args = p.parse_args() + sc = next((s for s in S.SCENARIOS if s["id"] == args.scenario), None) + if sc is None: + print(f"unknown scenario {args.scenario}", file=sys.stderr) + sys.exit(2) + gt = G.build_ground_truth(sc) + recv, checks = _load(sc, args.dir) + w = plot_wave(sc, recv, gt, os.path.join(args.dir, f"wave_{sc['id']}.png")) + tr = plot_trigger(sc, recv, checks, gt, + os.path.join(args.dir, f"trig_{sc['id']}.png")) + z = plot_zoom(sc, recv, checks, gt, + os.path.join(args.dir, f"zoom_{sc['id']}.png")) + print(w) + print(tr) + print(z) + + +if __name__ == "__main__": + main()