GeezerVison

"""Micro-benchmark for csv.reader field-append hot path.

Generates an in-memory CSV with many fields per row, then times how long it
takes to iterate the reader fully. Repeats N trials, reports min/median/mean
so noise is visible. Writes per-trial timings to results/.
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import statistics
import sys
import time
from pathlib import Path


def make_csv(rows: int, cols: int) -> str:
    buf = io.StringIO()
    w = csv.writer(buf)
    for r in range(rows):
        w.writerow([f"v{r}_{c}" for c in range(cols)])
    return buf.getvalue()


def time_reader(data: str) -> float:
    t0 = time.perf_counter_ns()
    for _ in csv.reader(io.StringIO(data)):
        pass
    return (time.perf_counter_ns() - t0) / 1e9


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--rows", type=int, default=20_000)
    ap.add_argument("--cols", type=int, default=20)
    ap.add_argument("--trials", type=int, default=15)
    ap.add_argument("--warmup", type=int, default=3)
    ap.add_argument("--label", default="run")
    ap.add_argument("--outdir", default="results/gh-150942")
    args = ap.parse_args()

    data = make_csv(args.rows, args.cols)
    size_bytes = len(data.encode())
    field_count = args.rows * args.cols

    for _ in range(args.warmup):
        time_reader(data)

    trials = [time_reader(data) for _ in range(args.trials)]

    outdir = Path(args.outdir)
    outdir.mkdir(parents=True, exist_ok=True)
    summary = {
        "label": args.label,
        "python": sys.executable,
        "rows": args.rows,
        "cols": args.cols,
        "trials": args.trials,
        "warmup": args.warmup,
        "field_count": field_count,
        "input_bytes": size_bytes,
        "seconds": {
            "min": min(trials),
            "median": statistics.median(trials),
            "mean": statistics.fmean(trials),
            "stdev": statistics.pstdev(trials),
            "all": trials,
        },
        "fields_per_sec_at_min": field_count / min(trials),
    }
    out = outdir / f"bench_{args.label}.json"
    out.write_text(json.dumps(summary, indent=2))

    s = summary["seconds"]
    print(
        f"{args.label}: min={s['min']*1000:.2f}ms median={s['median']*1000:.2f}ms "
        f"mean={s['mean']*1000:.2f}ms stdev={s['stdev']*1000:.2f}ms "
        f"fields/s@min={summary['fields_per_sec_at_min']:.2e}"
    )
    print(f"wrote {out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())