#!/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 build_by_kind(results, kind): """Filter raw scenario records (as written by run_e2e.sh's results.json aggregation) down to a single scenario `kind` (direct/recorder/debug/ tcplogger), with a small pass/fail rollup for the report's headline KPIs and per-kind section.""" scenarios = [s for s in results.get("scenarios", []) if s.get("kind") == kind] n_pass = sum(1 for s in scenarios if s.get("status") == "PASS") n_fail = sum(1 for s in scenarios if s.get("status") == "FAIL") return { "kind": kind, "n_pass": n_pass, "n_fail": n_fail, "n_total": len(scenarios), "scenarios": scenarios, } 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 # ── stress (Test/E2E/suite/stress.py + stress_run.py) ─────────────────────── _STRESS_LABELS = { "stress_pass": "Stress cases passed", "stress_fail": "Stress cases failed", "stress_max_hub_rss_mb": "Stress max hub RSS (MB)", "stress_max_marte_rss_mb": "Stress max MARTe RSS (MB)", "stress_max_zoom_p95_ms": "Stress max zoom p95 (ms)", } _STRESS_DIRECTION = { "stress_pass": True, "stress_fail": False, "stress_max_hub_rss_mb": False, "stress_max_marte_rss_mb": False, "stress_max_zoom_p95_ms": False, } _DIRECTION.update({ "direct_pass": True, "direct_fail": False, "recorder_pass": True, "recorder_fail": False, "debug_pass": True, "debug_fail": False, "tcplogger_pass": True, "tcplogger_fail": False, }) _DIRECTION.update(_STRESS_DIRECTION) _LABELS.update({ "direct_pass": "Direct scenarios passed", "direct_fail": "Direct scenarios failed", "recorder_pass": "Recorder scenarios passed", "recorder_fail": "Recorder scenarios failed", "debug_pass": "Debug scenarios passed", "debug_fail": "Debug scenarios failed", "tcplogger_pass": "TCPLogger scenarios passed", "tcplogger_fail": "TCPLogger scenarios failed", }) _LABELS.update(_STRESS_LABELS) def build_stress(sr): """Shape stress_results.json into the report's stress block (+ by_axis).""" cases = sr.get("cases", []) or [] by_axis = {} for c in cases: by_axis.setdefault(c.get("axis", "?"), []).append(c) for axis in by_axis: by_axis[axis].sort(key=lambda c: c.get("level", 0)) return {"overall": sr.get("overall", "FAIL"), "cases": cases, "by_axis": by_axis} def stress_headline(stress): cases = stress.get("cases", []) or [] return { "stress_pass": sum(1 for c in cases if c.get("status") == "PASS"), "stress_fail": sum(1 for c in cases if c.get("status") == "FAIL"), "stress_max_hub_rss_mb": max((c.get("hub_rss_mb", 0) or 0 for c in cases), default=0.0), "stress_max_marte_rss_mb": max((c.get("marte_rss_mb", 0) or 0 for c in cases), default=0.0), "stress_max_zoom_p95_ms": max((c.get("zoom_p95_ms", 0) or 0 for c in cases), default=0.0), } _STRESS_AXIS_METRICS = { "ds_signal_elements": [("MARTe RSS (MB)", "marte_rss_mb"), ("hub RSS (MB)", "hub_rss_mb")], "hub_signal_elements": [("hub RSS (MB)", "hub_rss_mb"), ("hub CPU (s)", "hub_cpu_s")], "ds_signal_count": [("MARTe RSS (MB)", "marte_rss_mb"), ("MARTe CPU (s)", "marte_cpu_s")], "hub_source_count": [("hub RSS (MB)", "hub_rss_mb"), ("MARTe RSS (MB)", "marte_rss_mb")], "hub_ws_clients": [("hub RSS (MB)", "hub_rss_mb"), ("hub CPU (s)", "hub_cpu_s")], "ds_subscriber_hubs": [("hub RSS (MB)", "hub_rss_mb"), ("MARTe CPU (s)", "marte_cpu_s")], "hub_zoom_reqrate_hz": [("zoom p95 (ms)", "zoom_p95_ms"), ("zoom p50 (ms)", "zoom_p50_ms")], } def stress_plots(by_axis, out): """One scaling-curve PNG per axis: level (x) vs the axis's metrics (y).""" made = [] for axis, cases in by_axis.items(): series = _STRESS_AXIS_METRICS.get( axis, [("hub RSS (MB)", "hub_rss_mb"), ("MARTe RSS (MB)", "marte_rss_mb")]) xs = [c.get("level") for c in cases] fig, ax = plt.subplots(figsize=(7, 3)) plotted = False for lbl, field in series: ys = [c.get(field) for c in cases] if all((v is None or v == 0) for v in ys): continue ax.plot(xs, ys, "o-", label=lbl) plotted = True if not plotted: plt.close(fig) continue ax.set_title(f"Scaling: {axis}") ax.set_xlabel("load level") ax.set_ylabel("value") ax.grid(alpha=0.3) ax.legend(fontsize=8) fig.tight_layout() p = os.path.join(out, f"stress_{axis}.png") fig.savefig(p, dpi=110) plt.close(fig) made.append(p) 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) ap.add_argument("--stress-results", default=None) 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) direct = build_by_kind(results, "direct") recorder = build_by_kind(results, "recorder") debug = build_by_kind(results, "debug") tcplogger = build_by_kind(results, "tcplogger") stress = None stress_plot_paths = [] if args.stress_results and os.path.exists(args.stress_results): sr = _load(args.stress_results) if sr: stress = build_stress(sr) stress_plot_paths = stress_plots(stress["by_axis"], args.out) 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) hl.update({ "direct_pass": direct["n_pass"], "direct_fail": direct["n_fail"], "recorder_pass": recorder["n_pass"], "recorder_fail": recorder["n_fail"], "debug_pass": debug["n_pass"], "debug_fail": debug["n_fail"], "tcplogger_pass": tcplogger["n_pass"], "tcplogger_fail": tcplogger["n_fail"], }) if stress: hl.update(stress_headline(stress)) # 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, "direct": direct, "recorder": recorder, "debug": debug, "tcplogger": tcplogger, "stress": stress, "stress_plots": [os.path.basename(p) for p in stress_plot_paths], "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()