#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="Generate one multi-RIR replay case report") p.add_argument("--rir", required=True) p.add_argument("--snapshot-meta", required=True) p.add_argument("--snapshot-compare", required=True) p.add_argument("--delta-meta", required=True) p.add_argument("--delta-compare", required=True) p.add_argument("--routinator-base-seconds", required=True, type=float) p.add_argument("--routinator-delta-seconds", required=True, type=float) p.add_argument("--out-md", required=True) p.add_argument("--out-json", required=True) return p.parse_args() def read_json(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def parse_compare_md(path: str) -> dict: lines = Path(path).read_text(encoding="utf-8").splitlines() out = {} for line in lines: if not line.startswith("| "): continue parts = [p.strip() for p in line.strip("|").split("|")] if len(parts) != 2: continue key, value = parts if key in {"metric", "---"}: continue try: out[key] = int(value) except ValueError: pass return out def ratio(ours: float, baseline: float) -> float | None: if baseline <= 0: return None return ours / baseline def build_report(args: argparse.Namespace) -> dict: snapshot_meta = read_json(args.snapshot_meta) delta_meta = read_json(args.delta_meta) snapshot_compare = parse_compare_md(args.snapshot_compare) delta_compare = parse_compare_md(args.delta_compare) snapshot_ours = float(snapshot_meta["durations_secs"]["rpki_run"]) delta_ours = float(delta_meta["durations_secs"]["rpki_run"]) report = { "rir": args.rir, "snapshot": { "meta_json": str(Path(args.snapshot_meta).resolve()), "compare_md": str(Path(args.snapshot_compare).resolve()), "ours_seconds": snapshot_ours, "routinator_seconds": args.routinator_base_seconds, "ratio": ratio(snapshot_ours, args.routinator_base_seconds), "compare": snapshot_compare, "match": snapshot_compare.get("only_in_ours", -1) == 0 and snapshot_compare.get("only_in_record", -1) == 0, "counts": snapshot_meta.get("counts", {}), }, "delta": { "meta_json": str(Path(args.delta_meta).resolve()), "compare_md": str(Path(args.delta_compare).resolve()), "ours_seconds": delta_ours, "routinator_seconds": args.routinator_delta_seconds, "ratio": ratio(delta_ours, args.routinator_delta_seconds), "compare": delta_compare, "match": delta_compare.get("only_in_ours", -1) == 0 and delta_compare.get("only_in_record", -1) == 0, "counts": delta_meta.get("counts", {}), }, } return report def write_md(path: Path, report: dict) -> None: snapshot = report["snapshot"] delta = report["delta"] lines = [] lines.append(f"# {report['rir'].upper()} Replay Report\n\n") lines.append("## Summary\n\n") lines.append("| mode | match | ours_s | routinator_s | ratio | only_in_ours | only_in_record |\n") lines.append("|---|---|---:|---:|---:|---:|---:|\n") lines.append( f"| snapshot | {str(snapshot['match']).lower()} | {snapshot['ours_seconds']:.3f} | {snapshot['routinator_seconds']:.3f} | {snapshot['ratio']:.3f} | {snapshot['compare'].get('only_in_ours', 0)} | {snapshot['compare'].get('only_in_record', 0)} |\n" ) lines.append( f"| delta | {str(delta['match']).lower()} | {delta['ours_seconds']:.3f} | {delta['routinator_seconds']:.3f} | {delta['ratio']:.3f} | {delta['compare'].get('only_in_ours', 0)} | {delta['compare'].get('only_in_record', 0)} |\n" ) lines.append("\n## Snapshot Inputs\n\n") lines.append(f"- meta_json: `{snapshot['meta_json']}`\n") lines.append(f"- compare_md: `{snapshot['compare_md']}`\n") lines.append("\n## Delta Inputs\n\n") lines.append(f"- meta_json: `{delta['meta_json']}`\n") lines.append(f"- compare_md: `{delta['compare_md']}`\n") lines.append("\n## Counts\n\n") lines.append("### Snapshot\n\n") for k, v in sorted(snapshot.get("counts", {}).items()): lines.append(f"- {k}: `{v}`\n") lines.append("\n### Delta\n\n") for k, v in sorted(delta.get("counts", {}).items()): lines.append(f"- {k}: `{v}`\n") path.write_text("".join(lines), encoding="utf-8") def main() -> int: args = parse_args() report = build_report(args) out_json = Path(args.out_json) out_md = Path(args.out_md) out_json.parent.mkdir(parents=True, exist_ok=True) out_md.parent.mkdir(parents=True, exist_ok=True) out_json.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") write_md(out_md, report) print(out_md) return 0 if __name__ == "__main__": raise SystemExit(main())