profanity/tests/bench/compare_baseline.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
#
# This file is part of CProof.
# See LICENSE for the full GPLv3 text and the special OpenSSL linking exception.

"""
compare_baseline.py — diff a fresh `current.csv` against a committed
`baseline.csv` and exit non-zero on regressions.

Aggregation rule: same (scenario, volume) → median wall_ms across rows.
A regression is a > THRESHOLD percent slowdown vs. the baseline; speedups
of any size are reported but never fail the run.

Usage:
    compare_baseline.py [--baseline=PATH] [--current=PATH]
                        [--threshold=PCT] [--quiet]

Exit codes:
    0 — no regressions
    1 — at least one regression
    2 — input parse error
"""

import argparse
import csv
import statistics
import sys
from pathlib import Path

DEFAULT_THRESHOLD = 25.0  # percent slower vs baseline = regression


def load(path: Path) -> dict[tuple[str, str], list[float]]:
    if not path.is_file():
        return {}
    rows: dict[tuple[str, str], list[float]] = {}
    with path.open("r", newline="") as f:
        rdr = csv.DictReader(f)
        if not rdr.fieldnames or "scenario" not in rdr.fieldnames:
            sys.exit(f"ERROR: {path} has no 'scenario' column")
        for row in rdr:
            try:
                wall = float(row["wall_ms"])
            except (KeyError, ValueError):
                continue
            key = (row.get("scenario", ""), row.get("volume", ""))
            rows.setdefault(key, []).append(wall)
    return rows


def median(values: list[float]) -> float:
    return statistics.median(values) if values else 0.0


def fmt_ms(ms: float) -> str:
    if ms < 1.0:
        return f"{ms*1000:.0f} us"
    if ms < 1000.0:
        return f"{ms:.2f} ms"
    if ms < 60000.0:
        return f"{ms/1000:.2f} s"
    return f"{ms/60000:.2f} min"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--baseline", default="tests/bench/baseline.csv")
    ap.add_argument("--current", default="tests/bench/current.csv")
    ap.add_argument("--threshold", type=float, default=DEFAULT_THRESHOLD,
                    help="regression threshold in percent (default 25)")
    ap.add_argument("--quiet", action="store_true",
                    help="only print regressions, hide unchanged/improved rows")
    args = ap.parse_args()

    base_p, cur_p = Path(args.baseline), Path(args.current)
    if not cur_p.is_file():
        sys.exit(f"ERROR: --current {cur_p} not found (run `make bench` first)")

    base = load(base_p)
    cur = load(cur_p)
    if not cur:
        sys.exit(f"ERROR: {cur_p} has no rows")

    keys = sorted(set(base) | set(cur))
    regressions: list[tuple[str, str, float, float, float]] = []
    improvements: list[tuple[str, str, float, float, float]] = []
    new_rows: list[tuple[str, str, float]] = []
    missing: list[tuple[str, str, float]] = []

    print(f"{'scenario':<30s} {'volume':<14s} {'baseline':>12s} {'current':>12s} {'delta':>10s}")
    print("-" * 84)

    for k in keys:
        sc, vol = k
        b = median(base.get(k, []))
        c = median(cur.get(k, []))
        if k not in base:
            new_rows.append((sc, vol, c))
            if not args.quiet:
                print(f"{sc:<30s} {vol:<14s} {'(new)':>12s} {fmt_ms(c):>12s} {'':>10s}")
            continue
        if k not in cur:
            missing.append((sc, vol, b))
            if not args.quiet:
                print(f"{sc:<30s} {vol:<14s} {fmt_ms(b):>12s} {'(gone)':>12s} {'':>10s}")
            continue
        if b == 0:
            continue
        pct = (c - b) / b * 100.0
        marker = ""
        if pct >= args.threshold:
            regressions.append((sc, vol, b, c, pct))
            marker = "  REGRESSION"
        elif pct <= -args.threshold:
            improvements.append((sc, vol, b, c, pct))
            marker = "  improved"
        if marker or not args.quiet:
            sign = "+" if pct >= 0 else ""
            print(f"{sc:<30s} {vol:<14s} {fmt_ms(b):>12s} {fmt_ms(c):>12s} {sign}{pct:>8.1f}%{marker}")

    print()
    print(f"summary: {len(regressions)} regressions, {len(improvements)} improvements, "
          f"{len(new_rows)} new rows, {len(missing)} removed rows "
          f"(threshold = ±{args.threshold:.0f} %)")

    if regressions:
        print("REGRESSED scenarios:")
        for sc, vol, b, c, pct in regressions:
            print(f"  - {sc} ({vol}): {fmt_ms(b)} → {fmt_ms(c)}  (+{pct:.1f}%)")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())