compare.py

#!/usr/bin/env python3
from __future__ import annotations

import csv
import sys
from pathlib import Path


def read_one_row(path: Path) -> dict[str, str]:
    with path.open(newline="") as handle:
        rows = list(csv.DictReader(handle))
    if len(rows) != 1:
        raise RuntimeError(f"expected one data row in {path}")
    return rows[0]


def read_guard(path: Path) -> dict[int, dict[str, str]]:
    with path.open(newline="") as handle:
        return {int(row["inner_loops"]): row for row in csv.DictReader(handle)}


def emit_row(writer: csv.writer, section: str, key: str, hip_us: float, fastpt_us: float) -> None:
    delta = fastpt_us - hip_us
    writer.writerow(
        [
            section,
            key,
            f"{hip_us:.9f}",
            f"{fastpt_us:.9f}",
            f"{delta:.9f}",
            f"{delta / hip_us * 100.0:.6f}" if hip_us else "nan",
            f"{fastpt_us / hip_us:.6f}" if hip_us else "nan",
        ]
    )


def main() -> int:
    root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
    hip = root / "hip"
    fastpt = root / "fastpt-C"

    writer = csv.writer(sys.stdout)
    writer.writerow(["section", "case", "hip_us", "fastpt_c_us", "delta_us", "delta_pct", "ratio"])

    hip_query = read_one_row(hip / "device_query.csv")
    fastpt_query = read_one_row(fastpt / "device_query.csv")
    emit_row(
        writer,
        "device_query",
        f"{hip_query['api']} vs {fastpt_query['api']}",
        float(hip_query["median_us"]),
        float(fastpt_query["median_us"]),
    )

    hip_guard = read_guard(hip / "guard_loop.csv")
    fastpt_guard = read_guard(fastpt / "guard_loop.csv")
    for inner_loops in sorted(set(hip_guard) & set(fastpt_guard)):
        emit_row(
            writer,
            "guard_loop",
            str(inner_loops),
            float(hip_guard[inner_loops]["median_step_us"]),
            float(fastpt_guard[inner_loops]["median_step_us"]),
        )
    return 0


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