pygfried benchmark (issue #25)

bench.sh

#!/usr/bin/env bash
set -euo pipefail

INPUT_DIR="${INPUT_DIR:-pictures}"
RUNS="${RUNS:-30}"
WARMUP="${WARMUP:-3}"
WORKERS="${WORKERS:-1,2,4,8,16}"
RESULTS_DIR="${RESULTS_DIR:-bench-results}"
PYGFRIED_SPEC="${PYGFRIED_SPEC:-pygfried @ git+https://github.com/artefactual-labs/pygfried.git@dev/issue-25-many-apis}"

need() {
  if ! command -v "$1" >/dev/null 2>&1; then
    printf 'Missing required command: %s\n' "$1" >&2
    exit 1
  fi
}

need uv
need hyperfine

shell_join() {
  local quoted
  local -a parts=()

  for arg in "$@"; do
    printf -v quoted '%q' "$arg"
    parts+=("$quoted")
  done

  local IFS=' '
  printf '%s' "${parts[*]}"
}

if [[ ! -d "$INPUT_DIR" ]]; then
  printf 'Configured input directory does not exist\n' >&2
  exit 1
fi

if ! [[ "$RUNS" =~ ^[1-9][0-9]*$ ]]; then
  printf 'RUNS must be a positive integer, got: %s\n' "$RUNS" >&2
  exit 1
fi

if ! [[ "$WARMUP" =~ ^[0-9]+$ ]]; then
  printf 'WARMUP must be a non-negative integer, got: %s\n' "$WARMUP" >&2
  exit 1
fi

IFS=',' read -r -a WORKER_VALUES <<< "$WORKERS"
if [[ "${#WORKER_VALUES[@]}" -eq 0 ]]; then
  printf 'WORKERS must contain at least one worker count\n' >&2
  exit 1
fi

for worker in "${WORKER_VALUES[@]}"; do
  if ! [[ "$worker" =~ ^[1-9][0-9]*$ ]]; then
    printf 'Each WORKERS entry must be a positive integer, got: %s\n' "$worker" >&2
    exit 1
  fi
done

mkdir -p "$RESULTS_DIR"

tmpdir="$(mktemp -d)"
trap 'rm -rf "$tmpdir"' EXIT

harness="$tmpdir/bench_pygfried.py"
cat > "$harness" <<'PY'
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import pygfried


def collect_files(root: Path) -> list[str]:
    return [str(path) for path in sorted(root.rglob("*")) if path.is_file()]


def expect_count(result: object, expected: int, label: str) -> None:
    if not isinstance(result, dict):
        raise TypeError(f"{label} returned {type(result).__name__}, expected dict")

    files = result.get("files")
    if not isinstance(files, list):
        raise TypeError(f"{label} result has no files list")

    if len(files) != expected:
        raise RuntimeError(
            f"{label} identified {len(files)} files, expected {expected}"
        )


def run_baseline(root: Path) -> int:
    count = 0
    for path in collect_files(root):
        result = pygfried.identify(path, detailed=True)
        expect_count(result, 1, "identify(..., detailed=True)")
        count += 1
    return count


def run_identify_many(root: Path, workers: int) -> int:
    files = collect_files(root)
    result = pygfried.identify_many(files, workers=workers)
    expect_count(result, len(files), f"identify_many(workers={workers})")
    return len(files)


def run_identify_dir(root: Path, workers: int) -> int:
    expected = len(collect_files(root))
    result = pygfried.identify_dir(
        str(root),
        recursive=True,
        workers=workers,
        follow_symlinks=False,
    )
    expect_count(result, expected, f"identify_dir(workers={workers})")
    return expected


def cmd_preflight(args: argparse.Namespace) -> None:
    root = args.input_dir
    files = collect_files(root)
    if not files:
        raise RuntimeError("no files found below configured directory")

    print(f"pygfried siegfried version: {pygfried.version()}")
    print(f"file count: {len(files)}")
    print(f"workers: {args.workers}")

    first = files[0]
    expect_count(pygfried.identify(first, detailed=True), 1, "identify preflight")
    expect_count(
        pygfried.identify_many([first], workers=1),
        1,
        "identify_many preflight",
    )
    expect_count(
        pygfried.identify_dir(
            str(Path(first).parent),
            recursive=False,
            workers=1,
            follow_symlinks=False,
        ),
        len([path for path in Path(first).parent.iterdir() if path.is_file()]),
        "identify_dir preflight",
    )


def cmd_run(args: argparse.Namespace) -> None:
    root = args.input_dir
    if args.case == "identify-loop":
        count = run_baseline(root)
    elif args.case == "identify-many":
        count = run_identify_many(root, args.workers)
    elif args.case == "identify-dir":
        count = run_identify_dir(root, args.workers)
    else:
        raise ValueError(f"unknown case: {args.case}")

    if args.verbose:
        print(f"{args.case} identified {count} files")


def cmd_analyze(args: argparse.Namespace) -> None:
    with args.json_output.open() as fp:
        data = json.load(fp)

    results = data.get("results", [])
    if not results:
        raise RuntimeError(f"no benchmark results found in {args.json_output}")

    by_name = {result["command"]: result for result in results}
    baseline = by_name.get("identify loop detailed")
    fastest = min(results, key=lambda result: result["mean"])

    baseline_mean = baseline["mean"] if baseline else fastest["mean"]
    fastest_mean = fastest["mean"]

    print("\nAnalysis")
    print("--------")
    print(f"Fastest: {fastest['command']} ({fastest_mean * 1000:.1f} ms mean)")

    if baseline:
        print(
            "Baseline: identify loop detailed "
            f"({baseline_mean * 1000:.1f} ms mean)"
        )

    print()
    print(
        f"{'Command':32} {'Mean ms':>9} {'Std ms':>8} "
        f"{'CV %':>7} {'vs base':>8} {'vs best':>8}"
    )
    print("-" * 78)

    for result in results:
        mean = result["mean"]
        stddev = result.get("stddev") or 0
        cv = (stddev / mean * 100) if mean else 0
        vs_base = baseline_mean / mean
        vs_best = mean / fastest_mean
        print(
            f"{result['command'][:32]:32} "
            f"{mean * 1000:9.1f} "
            f"{stddev * 1000:8.1f} "
            f"{cv:7.2f} "
            f"{vs_base:8.2f} "
            f"{vs_best:8.2f}"
        )

    print("\nNotes")
    for result in results:
        stddev = result.get("stddev") or 0
        cv = (stddev / result["mean"] * 100) if result["mean"] else 0
        if cv >= 8:
            print(
                "- High variance: "
                f"{result['command']} has CV {cv:.1f}%."
            )

    for prefix in ("identify_many", "identify_dir"):
        matching = [
            result for result in results if result["command"].startswith(prefix)
        ]
        if not matching:
            continue

        best = min(matching, key=lambda result: result["mean"])
        worst = max(matching, key=lambda result: result["mean"])
        if worst["mean"] >= best["mean"] * 1.5:
            print(
                "- Worker sensitivity: "
                f"{worst['command']} is {worst['mean'] / best['mean']:.2f}x "
                f"slower than {best['command']}."
            )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input-dir", type=Path, required=True)
    parser.add_argument("--workers", type=int, default=1)
    parser.add_argument("--verbose", action="store_true")

    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("preflight")

    run_parser = subparsers.add_parser("run")
    run_parser.add_argument(
        "--case",
        required=True,
        choices=("identify-loop", "identify-many", "identify-dir"),
    )

    analyze_parser = subparsers.add_parser("analyze")
    analyze_parser.add_argument("--json-output", type=Path, required=True)

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if not args.input_dir.is_dir():
        parser.error("configured input directory is not a directory")

    if args.workers < 1:
        parser.error("--workers must be a positive integer")

    if args.command == "preflight":
        cmd_preflight(args)
    elif args.command == "run":
        cmd_run(args)
    elif args.command == "analyze":
        cmd_analyze(args)
    else:
        parser.error(f"unknown command: {args.command}")

    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)
PY

timestamp="$(date +%Y%m%d-%H%M%S)"
json_output="$RESULTS_DIR/pygfried-$timestamp.json"
markdown_output="$RESULTS_DIR/pygfried-$timestamp.md"
csv_output="$RESULTS_DIR/pygfried-$timestamp.csv"

printf 'Warming uv and pygfried before benchmarking...\n'
bench_venv="$tmpdir/.venv"
bench_python="$bench_venv/bin/python"
uv venv --quiet "$bench_venv"
uv pip install --quiet --python "$bench_python" "$PYGFRIED_SPEC"
base_command=("$bench_python" "$harness" --input-dir "$INPUT_DIR")
"${base_command[@]}" --workers 1 preflight

names=(-n 'identify loop detailed')
commands=("$(shell_join "${base_command[@]}" run --case identify-loop)")

for worker in "${WORKER_VALUES[@]}"; do
  names+=(-n "identify_many workers=$worker")
  commands+=("$(shell_join "${base_command[@]}" --workers "$worker" run --case identify-many)")

  names+=(-n "identify_dir workers=$worker")
  commands+=("$(shell_join "${base_command[@]}" --workers "$worker" run --case identify-dir)")
done

printf '\nRunning benchmarks with %s measured run(s), %s warmup run(s)...\n' "$RUNS" "$WARMUP"
printf 'Results will be written to:\n'
printf '  %s\n' "$json_output"
printf '  %s\n' "$markdown_output"
printf '  %s\n\n' "$csv_output"

hyperfine \
  --warmup "$WARMUP" \
  --runs "$RUNS" \
  --sort command \
  --time-unit millisecond \
  --export-json "$json_output" \
  --export-markdown "$markdown_output" \
  --export-csv "$csv_output" \
  "${names[@]}" \
  "${commands[@]}"

"${base_command[@]}" analyze --json-output "$json_output"

printf '\nBenchmark complete.\n'
printf 'JSON: %s\n' "$json_output"
printf 'Markdown: %s\n' "$markdown_output"
printf 'CSV: %s\n' "$csv_output"