#!/usr/bin/env python3 """ report_build.py — Consolidate the E2E run into report_data.json (+ trend plots). Inputs (paths via flags): * results.json — per-scenario status + waveform metrics (orchestrator) * perf__*.json — per-scenario CPU/peak-RSS snapshots (proc_perf.py) * unit_tests.json — GTest/Go/Python suite results (collect.py) * coverage.json — per-language coverage (collect.py) Outputs (into --out): * report_data.json — everything the Typst template renders, including a ``regression`` block that diffs this run's headline metrics against the previous entry in history.jsonl (progression ▲ / regression ▼). * history.jsonl — appended one line of headline metrics per run. * trend_*.png — pass-rate / coverage / fidelity / memory over runs. Throughput is derived as recorded-samples / recording-duration. Memory is the peak resident set (VmHWM). All inputs are optional: a missing artifact degrades to nulls so a partial run still produces a report. """ import argparse import datetime import json import os import subprocess import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 REC_DUR_S = 4.0 # client -dur; samples/sec denominator def _load(path, default=None): if path and os.path.exists(path): try: return json.load(open(path)) except (ValueError, OSError): return default return default def _git_sha(repo): try: return subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=repo, capture_output=True, text=True, timeout=10).stdout.strip() except (subprocess.SubprocessError, OSError): return "unknown" def _scenario_perf(work, sid): out = {} for role in ("hub", "marte"): rec = _load(os.path.join(work, f"perf_{sid}_{role}.json"), {}) if rec and rec.get("avail"): out[role] = {"cpu_s": round(rec.get("cpu_s", 0.0), 3), "peak_rss_mb": round(rec.get("peak_rss_kb", 0) / 1024.0, 1), "threads": rec.get("threads")} return out def _scenario_descs(): """id -> human description, imported from the scenario matrix (best effort).""" try: from scenarios import SCENARIOS return {s["id"]: s.get("desc") for s in SCENARIOS} except Exception: return {} def build_e2e(results, work): descs = _scenario_descs() scen = [] corrs, rss_vals, cpu_vals, tput_vals = [], [], [], [] for r in results.get("scenarios", []): sid = r["id"] metrics = r.get("metrics", {}) sigs = [] nrecv_total = 0 for key, m in (metrics.get("signals", {}) or {}).items(): nrecv_total += int(m.get("n_recv", 0) or 0) if "corr" in m: corrs.append(m["corr"]) sigs.append({ "key": key, "pass": m.get("pass"), "type": m.get("type"), "quant": m.get("quant"), "max_abs_err": m.get("max_abs_err"), "corr": m.get("corr"), "nrmse": m.get("nrmse"), "fidelity_ok": m.get("fidelity_ok"), "shape_ok": m.get("shape_ok"), "n_recv": m.get("n_recv"), }) perf = _scenario_perf(work, sid) for role in perf.values(): if role.get("peak_rss_mb"): rss_vals.append(role["peak_rss_mb"]) if role.get("cpu_s"): cpu_vals.append(role["cpu_s"]) tput = round(nrecv_total / REC_DUR_S, 1) if nrecv_total else 0.0 if tput: tput_vals.append(tput) client = metrics.get("client", {}) or {} # waveform overview image (plots.py writes it into --work); record the # basename only when present so the Typst template can embed it without # tripping over a missing file (Typst read() throws on absence). wave_img = f"wave_{sid}.png" has_wave = os.path.exists(os.path.join(work, wave_img)) scen.append({ "id": sid, "status": r.get("status"), "desc": descs.get(sid), "known_issue": r.get("known_issue"), "signals": sigs, "perf": perf, "throughput_sps": tput, "live_frames": (client.get("live", {}) or {}).get("frames"), "rollup": client.get("_rollup", {}), # detailed client behavioural checks (chain-client checks_.json), # surfaced so the report can show real zoom ranges + trigger captures # rather than only the pass/fail rollup booleans. "zoom": client.get("zoom", []) or [], "window": client.get("window", {}) or {}, "trigger": client.get("trigger", []) or [], "wave_img": wave_img if has_wave else None, }) agg = { "mean_corr": round(sum(corrs) / len(corrs), 4) if corrs else None, "mean_peak_rss_mb": round(sum(rss_vals) / len(rss_vals), 1) if rss_vals else None, "mean_cpu_s": round(sum(cpu_vals) / len(cpu_vals), 3) if cpu_vals else None, "mean_throughput_sps": round(sum(tput_vals) / len(tput_vals), 1) if tput_vals else None, } npass = sum(1 for s in scen if s["status"] == "PASS") nfail = sum(1 for s in scen if s["status"] == "FAIL") nskip = sum(1 for s in scen if s["status"] == "SKIP") nxfail = sum(1 for s in scen if s["status"] == "XFAIL") nxpass = sum(1 for s in scen if s["status"] == "XPASS") return { "overall": results.get("overall", "FAIL"), "n_pass": npass, "n_fail": nfail, "n_skip": nskip, "n_xfail": nxfail, "n_xpass": nxpass, "scenarios": scen, "agg": agg, } def headline(e2e, ut, cov): cov_by = {c["name"]: c.get("pct") for c in cov.get("languages", [])} t = ut.get("totals", {}) return { "e2e_pass": e2e["n_pass"], "e2e_fail": e2e["n_fail"], "e2e_xfail": e2e.get("n_xfail", 0), "e2e_xpass": e2e.get("n_xpass", 0), "e2e_total": (e2e["n_pass"] + e2e["n_fail"] + e2e["n_skip"] + e2e.get("n_xfail", 0) + e2e.get("n_xpass", 0)), "unit_pass": t.get("passed", 0), "unit_fail": t.get("failed", 0), "unit_total": t.get("total", 0), "cov_python": cov_by.get("Python"), "cov_go": cov_by.get("Go"), "cov_cpp": cov_by.get("C++"), "mean_corr": e2e["agg"]["mean_corr"], "mean_peak_rss_mb": e2e["agg"]["mean_peak_rss_mb"], "mean_cpu_s": e2e["agg"]["mean_cpu_s"], "mean_throughput_sps": e2e["agg"]["mean_throughput_sps"], } # field → "higher is better" (True), "lower is better" (False) _DIRECTION = { "e2e_pass": True, "e2e_fail": False, "unit_pass": True, "unit_fail": False, "cov_python": True, "cov_go": True, "cov_cpp": True, "mean_corr": True, "mean_peak_rss_mb": False, "mean_cpu_s": False, "mean_throughput_sps": True, } _LABELS = { "e2e_pass": "E2E scenarios passed", "e2e_fail": "E2E scenarios failed", "unit_pass": "Unit tests passed", "unit_fail": "Unit tests failed", "cov_python": "Python coverage %", "cov_go": "Go coverage %", "cov_cpp": "C++ coverage %", "mean_corr": "Mean sine corr", "mean_peak_rss_mb": "Mean peak RSS (MB)", "mean_cpu_s": "Mean CPU (s)", "mean_throughput_sps": "Mean throughput (samp/s)", } def regression(curr, prev): rows = [] for k, label in _LABELS.items(): c = curr.get(k) p = prev.get(k) if prev else None better = None delta = None if isinstance(c, (int, float)) and isinstance(p, (int, float)): delta = round(c - p, 4) if delta == 0: better = None else: better = (delta > 0) == _DIRECTION[k] rows.append({"name": label, "key": k, "current": c, "previous": p, "delta": delta, "better": better, "higher_better": _DIRECTION[k]}) return rows def trend_plots(history, out): if not history: return [] xs = list(range(len(history))) labels = [h.get("ts_short", str(i)) for i, h in enumerate(history)] made = [] def _plot(fname, series, title, ylabel): ys = [[h.get(k) for h in history] for _, k in series] if all(all(v is None for v in y) for y in ys): return fig, ax = plt.subplots(figsize=(7, 3)) for (lbl, _), y in zip(series, ys): xp = [x for x, v in zip(xs, y) if v is not None] yp = [v for v in y if v is not None] if yp: ax.plot(xp, yp, "o-", label=lbl) ax.set_title(title) ax.set_ylabel(ylabel) ax.set_xticks(xs) ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=7) ax.grid(alpha=0.3) ax.legend(fontsize=8) fig.tight_layout() p = os.path.join(out, fname) fig.savefig(p, dpi=110) plt.close(fig) made.append(p) _plot("trend_tests.png", [("E2E pass", "e2e_pass"), ("Unit pass", "unit_pass")], "Passing tests over runs", "count") _plot("trend_coverage.png", [("Python", "cov_python"), ("Go", "cov_go"), ("C++", "cov_cpp")], "Code coverage over runs", "% covered") _plot("trend_fidelity.png", [("Mean sine corr", "mean_corr")], "Waveform fidelity over runs", "correlation") _plot("trend_perf.png", [("Peak RSS (MB)", "mean_peak_rss_mb"), ("CPU (s)", "mean_cpu_s")], "Resource use over runs", "value") return made def main(): ap = argparse.ArgumentParser(description="Build E2E report_data.json") ap.add_argument("--repo", required=True) ap.add_argument("--results", required=True) ap.add_argument("--work", required=True) ap.add_argument("--out", required=True) args = ap.parse_args() os.makedirs(args.out, exist_ok=True) results = _load(args.results, {"overall": "FAIL", "scenarios": []}) ut = _load(os.path.join(args.out, "unit_tests.json"), {"suites": [], "totals": {}}) cov = _load(os.path.join(args.out, "coverage.json"), {"languages": []}) e2e = build_e2e(results, args.work) now = datetime.datetime.now() meta = {"timestamp": now.isoformat(timespec="seconds"), "ts_short": now.strftime("%m-%d %H:%M"), "git_sha": _git_sha(args.repo), "target": "x86-linux"} hl = headline(e2e, ut, cov) # history: read previous, then append current hist_path = os.path.join(args.out, "history.jsonl") history = [] if os.path.exists(hist_path): for line in open(hist_path): line = line.strip() if line: try: history.append(json.loads(line)) except ValueError: pass prev = history[-1] if history else None reg = regression(hl, prev) entry = dict(hl) entry["timestamp"] = meta["timestamp"] entry["ts_short"] = meta["ts_short"] entry["git_sha"] = meta["git_sha"] entry["overall"] = e2e["overall"] with open(hist_path, "a") as f: f.write(json.dumps(entry) + "\n") history.append(entry) plots = [os.path.basename(p) for p in trend_plots(history, args.out)] doc = { "meta": meta, "e2e": e2e, "unit_tests": ut, "coverage": cov, "regression": reg, "headline": hl, "trend_plots": plots, "history_len": len(history), "is_first_run": prev is None, } with open(os.path.join(args.out, "report_data.json"), "w") as f: json.dump(doc, f, indent=2) print(f"report_data.json: e2e {e2e['n_pass']}/{e2e['n_pass']+e2e['n_fail']+e2e['n_skip']}" f" pass, units {hl['unit_pass']}/{hl['unit_total']}, " f"cov py={hl['cov_python']} go={hl['cov_go']} cpp={hl['cov_cpp']}") if prev: ups = sum(1 for r in reg if r["better"] is True) downs = sum(1 for r in reg if r["better"] is False) print(f"regression vs previous run: {ups} improved, {downs} regressed") else: print("regression: first run (baseline established)") if __name__ == "__main__": main()