Benchmarks suite tool

.github/workflows/python-bench.yml

@@ -74,15 +74,15 @@ jobs:
 
       - name: Run parse benchmark
         run: |
-          poetry run python -m tests.benchmarks.bench_parse \
+          poetry run python -m pathable.benchmarks.bench_parse \
             --output "reports/bench-parse.${{ inputs.suffix }}.json" \
             ${{ inputs.quick && '--quick' || '' }} \
             --repeats "${{ inputs.repeats }}" \
             --warmup-loops "${{ inputs.warmup_loops }}"
 
       - name: Run lookup benchmark
         run: |
-          poetry run python -m tests.benchmarks.bench_lookup \
+          poetry run python -m pathable.benchmarks.bench_lookup \
             --output "reports/bench-lookup.${{ inputs.suffix }}.json" \
             ${{ inputs.quick && '--quick' || '' }} \
             --repeats "${{ inputs.repeats }}" \

README.md

@@ -144,32 +144,49 @@ pip install -e git+https://github.com/p1c2u/pathable.git#egg=pathable
 
 ## Benchmarks
 
-Benchmarks live in `tests/benchmarks/` and produce JSON reports.
+Benchmark tooling is shipped in the package and exposed as `pathable-bench`.
 
-Local run (recommended as modules):
+Install (core package):
 
 ```console
-poetry run python -m tests.benchmarks.bench_parse --output reports/bench-parse.json
-poetry run python -m tests.benchmarks.bench_lookup --output reports/bench-lookup.json
+pip install pathable
+```
+
+Optional benchmark extra (reserved for benchmark-specific deps):
+
+```console
+pip install "pathable[bench]"
+```
+
+Run all benchmark scenarios for an implementation:
+
+```console
+poetry run pathable-bench run --impl pathable.LookupPath --output reports/bench-lookup.json
 ```
 
 Quick sanity run:
 
 ```console
-poetry run python -m tests.benchmarks.bench_parse --quick --output reports/bench-parse.quick.json
-poetry run python -m tests.benchmarks.bench_lookup --quick --output reports/bench-lookup.quick.json
+poetry run pathable-bench run --impl pathable.LookupPath --quick --output reports/bench-lookup.quick.json
 ```
 
-Compare two results (fails if candidate is >20% slower in any scenario):
+Compare two results (compares overlapping scenarios only; fails if candidate is >20% slower in any compared scenario):
 
 ```console
-poetry run python -m tests.benchmarks.compare_results \
+poetry run pathable-bench compare \
     --baseline reports/bench-before.json \
     --candidate reports/bench-after.json \
     --tolerance 0.20
 ```
 
+Deprecated compatibility wrappers still exist for now:
+
+```console
+poetry run python -m tests.benchmarks.bench_lookup --output reports/bench-lookup.json
+poetry run python -m tests.benchmarks.bench_parse --output reports/bench-parse.json
+poetry run python -m tests.benchmarks.compare_results --baseline a.json --candidate b.json
+```
+
 CI (on-demand):
 
 - GitHub Actions workflow `Benchmarks` runs via `workflow_dispatch` and uploads the JSON artifacts.
-

pathable/benchmarks/__init__.py

@@ -0,0 +1 @@
+"""Public benchmark toolkit for pathable."""

pathable/benchmarks/bench_lookup.py

@@ -0,0 +1,49 @@
+"""Standalone lookup benchmark command."""
+
+import argparse
+from typing import Iterable
+from typing import cast
+
+from pathable.benchmarks.core import add_common_args
+from pathable.benchmarks.core import default_meta
+from pathable.benchmarks.core import results_to_json
+from pathable.benchmarks.core import write_json
+from pathable.benchmarks.registry import resolve_impl
+from pathable.benchmarks.scenarios.lookup import run_lookup_scenarios
+from pathable.paths import AccessorPath
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    add_common_args(parser)
+    parser.add_argument(
+        "--impl",
+        default="pathable.LookupPath",
+        help="AccessorPath implementation target (default: pathable.LookupPath).",
+    )
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    impl = resolve_impl(args.impl)
+    if not issubclass(impl, AccessorPath):
+        raise TypeError(
+            "lookup benchmark requires an AccessorPath implementation"
+        )
+    accessor_impl = cast(type[AccessorPath[object, object, object]], impl)
+
+    results = run_lookup_scenarios(
+        accessor_impl,
+        quick=args.quick,
+        repeats=args.repeats,
+        warmup_loops=args.warmup_loops,
+    )
+
+    meta = default_meta()
+    meta["impl"] = f"{impl.__module__}.{impl.__qualname__}"
+
+    payload = results_to_json(results=results, meta=meta)
+    write_json(args.output, payload)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pathable/benchmarks/bench_parse.py

@@ -0,0 +1,29 @@
+"""Standalone parse benchmark command."""
+
+import argparse
+from typing import Iterable
+
+from pathable.benchmarks.core import add_common_args
+from pathable.benchmarks.core import results_to_json
+from pathable.benchmarks.core import write_json
+from pathable.benchmarks.scenarios.parse import run_parse_scenarios
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    add_common_args(parser)
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    results = run_parse_scenarios(
+        quick=args.quick,
+        repeats=args.repeats,
+        warmup_loops=args.warmup_loops,
+    )
+
+    payload = results_to_json(results=results)
+    write_json(args.output, payload)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pathable/benchmarks/cli.py

@@ -0,0 +1,80 @@
+"""CLI entrypoint for pathable benchmarks."""
+
+import argparse
+from typing import Iterable
+
+from pathable.benchmarks.compare import main as compare_main
+from pathable.benchmarks.core import write_json
+from pathable.benchmarks.run import run_all
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="pathable-bench")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    run_parser = subparsers.add_parser("run", help="Run benchmark scenarios")
+    run_parser.add_argument(
+        "--impl",
+        required=True,
+        help="Implementation target, e.g. pathable.LookupPath",
+    )
+    run_parser.add_argument("--output", required=True)
+    run_parser.add_argument("--quick", action="store_true")
+    run_parser.add_argument("--repeats", type=int, default=5)
+    run_parser.add_argument("--warmup-loops", type=int, default=1)
+    run_parser.add_argument(
+        "--scenario",
+        action="append",
+        choices=["parse", "lookup"],
+        help=(
+            "Run only selected scenario groups. Repeat flag to select multiple; "
+            "defaults to both parse and lookup."
+        ),
+    )
+
+    compare_parser = subparsers.add_parser(
+        "compare",
+        help="Compare benchmark JSON reports",
+    )
+    compare_parser.add_argument("--baseline", required=True)
+    compare_parser.add_argument("--candidate", required=True)
+    compare_parser.add_argument("--tolerance", type=float, default=0.20)
+    return parser
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    parser = _build_parser()
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    if args.command == "run":
+        scenarios = (
+            tuple(args.scenario) if args.scenario else ("parse", "lookup")
+        )
+        payload = run_all(
+            impl_target=args.impl,
+            quick=args.quick,
+            repeats=args.repeats,
+            warmup_loops=args.warmup_loops,
+            scenarios=scenarios,
+        )
+        write_json(args.output, payload)
+        return 0
+
+    if args.command == "compare":
+        return compare_main(
+            [
+                "--baseline",
+                args.baseline,
+                "--candidate",
+                args.candidate,
+                "--tolerance",
+                str(args.tolerance),
+            ]
+        )
+
+    parser.error("unknown command")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pathable/benchmarks/compare.py

@@ -0,0 +1,149 @@
+"""Compare benchmark JSON results."""
+
+import argparse
+import json
+from dataclasses import dataclass
+from typing import Any
+from typing import Iterable
+from typing import Mapping
+from typing import cast
+
+
+@dataclass(frozen=True)
+class ScenarioComparison:
+    name: str
+    baseline_ops: float
+    candidate_ops: float
+    ratio: float
+
+
+@dataclass(frozen=True)
+class CompareResult:
+    comparisons: list[ScenarioComparison]
+    regressions: list[ScenarioComparison]
+    baseline_only: list[str]
+    candidate_only: list[str]
+
+
+def _load(path: str) -> Mapping[str, Any]:
+    with open(path, "r", encoding="utf-8") as f:
+        data_any = json.load(f)
+    if not isinstance(data_any, dict):
+        raise ValueError("Invalid report: expected top-level JSON object")
+    return cast(dict[str, Any], data_any)
+
+
+def _extract_ops(report: Mapping[str, Any]) -> dict[str, float]:
+    benchmarks = report.get("benchmarks")
+    if not isinstance(benchmarks, dict):
+        raise ValueError("Invalid report: missing 'benchmarks' dict")
+
+    benchmarks_d = cast(dict[str, Any], benchmarks)
+
+    out: dict[str, float] = {}
+    for name, payload in benchmarks_d.items():
+        if not isinstance(payload, dict):
+            continue
+        payload_d = cast(dict[str, Any], payload)
+        ops_any = payload_d.get("median_ops_per_sec")
+        ops = ops_any if isinstance(ops_any, (int, float)) else None
+        if ops is not None:
+            out[name] = float(ops)
+    return out
+
+
+def compare(
+    *,
+    baseline: Mapping[str, Any],
+    candidate: Mapping[str, Any],
+    tolerance: float,
+) -> CompareResult:
+    if tolerance < 0:
+        raise ValueError("tolerance must be >= 0")
+
+    b = _extract_ops(baseline)
+    c = _extract_ops(candidate)
+
+    b_names = set(b)
+    c_names = set(c)
+    common_names = sorted(b_names & c_names)
+
+    comparisons: list[ScenarioComparison] = []
+    for name in common_names:
+        bops = b[name]
+        cops = c[name]
+        ratio = cops / bops if bops > 0 else float("inf")
+        comparisons.append(
+            ScenarioComparison(
+                name=name,
+                baseline_ops=bops,
+                candidate_ops=cops,
+                ratio=ratio,
+            )
+        )
+
+    floor_ratio = 1.0 - tolerance
+    regressions = [x for x in comparisons if x.ratio < floor_ratio]
+
+    return CompareResult(
+        comparisons=comparisons,
+        regressions=regressions,
+        baseline_only=sorted(b_names - c_names),
+        candidate_only=sorted(c_names - b_names),
+    )
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--baseline", required=True)
+    parser.add_argument("--candidate", required=True)
+    parser.add_argument(
+        "--tolerance",
+        type=float,
+        default=0.20,
+        help="Allowed slowdown (e.g. 0.20 means 20% slower allowed).",
+    )
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    baseline = _load(args.baseline)
+    candidate = _load(args.candidate)
+
+    result = compare(
+        baseline=baseline,
+        candidate=candidate,
+        tolerance=args.tolerance,
+    )
+
+    common_count = len(result.comparisons)
+    print(
+        "scenarios: "
+        f"common={common_count} "
+        f"baseline_only={len(result.baseline_only)} "
+        f"candidate_only={len(result.candidate_only)}"
+    )
+
+    if common_count == 0:
+        print("ERROR: no overlapping scenarios between reports")
+        return 1
+
+    print("scenario\tbaseline_ops/s\tcandidate_ops/s\tratio")
+    for row in result.comparisons:
+        print(
+            f"{row.name}\t{row.baseline_ops:.2f}\t{row.candidate_ops:.2f}\t{row.ratio:.3f}"
+        )
+
+    if result.regressions:
+        print("\nREGRESSIONS:")
+        for row in result.regressions:
+            print(
+                f"- {row.name}: {row.ratio:.3f}x "
+                f"(baseline {row.baseline_ops:.2f} ops/s, "
+                f"candidate {row.candidate_ops:.2f} ops/s)"
+            )
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())