Bench Results

benchmatrix.bench_results

Parse and display benchmatrix-tagged pytest-benchmark JSON output.

This module does not calculate timings itself. It validates benchmatrix metadata embedded in pytest-benchmark JSON and derives metric-specific views from the statistics and raw samples recorded by pytest-benchmark.

ParsedBenchmarkRow dataclass

One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

Attributes:

Name	Type	Description
benchmark_name	str	Name assigned by pytest-benchmark to this benchmark.
metric_name	MetricName	Benchmatrix metric name from extra_info.
implementation_name	str	Implementation name from extra_info.
case_name	str	Case name from extra_info.
stats	_BenchmarkStats	Raw pytest-benchmark timing statistics.
extra_info	_BenchmarkStats	Custom metadata from benchmark.extra_info.
derived	_BenchmarkStats	Derived metric-specific statistics computed from JSON output.

Source code in src/benchmatrix/bench_results.py

@dataclass(frozen=True, slots=True)
class ParsedBenchmarkRow:
    """One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

    Attributes:
        benchmark_name: Name assigned by pytest-benchmark to this benchmark.
        metric_name: Benchmatrix metric name from ``extra_info``.
        implementation_name: Implementation name from ``extra_info``.
        case_name: Case name from ``extra_info``.
        stats: Raw pytest-benchmark timing statistics.
        extra_info: Custom metadata from ``benchmark.extra_info``.
        derived: Derived metric-specific statistics computed from JSON output.
    """

    benchmark_name: str
    metric_name: MetricName
    implementation_name: str
    case_name: str
    stats: _BenchmarkStats
    extra_info: _BenchmarkStats
    derived: _BenchmarkStats

load_benchmark_json

load_benchmark_json(
    path: str | Path,
) -> list[ParsedBenchmarkRow]

Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a JSON file created with --benchmark-json.	required

Returns:

Type	Description
list[ParsedBenchmarkRow]	Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
list[ParsedBenchmarkRow]	metric-specific fields. Non-benchmatrix rows are rejected.

Raises:

Type	Description
BenchmarkJsonError	If the JSON does not have the expected pytest-benchmark and benchmatrix structure.

Source code in src/benchmatrix/bench_results.py

def load_benchmark_json(path: str | Path) -> list[ParsedBenchmarkRow]:
    """Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

    Args:
        path: Path to a JSON file created with ``--benchmark-json``.

    Returns:
        Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
        metric-specific fields. Non-benchmatrix rows are rejected.

    Raises:
        BenchmarkJsonError: If the JSON does not have the expected
            pytest-benchmark and benchmatrix structure.
    """
    path_obj = Path(path)

    try:
        payload = _load_json(path_obj)
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read benchmark JSON: {path_obj}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in benchmark file: {path_obj}") from exc

    payload_mapping = _require_mapping(payload, path="root")
    benchmarks = _require_list(
        payload_mapping.get(JSON_KEY_BENCHMARKS),
        path=f"root.{JSON_KEY_BENCHMARKS}",
    )

    rows: list[ParsedBenchmarkRow] = []
    for index, benchmark_entry in enumerate(benchmarks):
        entry_path = f"root.{JSON_KEY_BENCHMARKS}[{index}]"
        entry = _require_mapping(benchmark_entry, path=entry_path)
        extra_info = _require_mapping(
            entry.get(JSON_KEY_EXTRA_INFO),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}",
        )
        _require_benchmatrix_schema(extra_info, path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}")

        stats = _require_mapping(
            entry.get(JSON_KEY_STATS),
            path=f"{entry_path}.{JSON_KEY_STATS}",
        )

        metric_name = _require_metric_name(
            extra_info.get(KEY_METRIC_NAME),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_METRIC_NAME}",
        )
        data = _extract_benchmark_data(entry, stats, metric_name, path=entry_path)

        rows.append(
            ParsedBenchmarkRow(
                benchmark_name=_benchmark_name(entry),
                metric_name=metric_name,
                implementation_name=_require_string(
                    extra_info.get(KEY_IMPLEMENTATION_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_IMPLEMENTATION_NAME}",
                ),
                case_name=_require_string(
                    extra_info.get(KEY_CASE_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_CASE_NAME}",
                ),
                stats=stats,
                extra_info=extra_info,
                derived=_derive_stats(metric_name, stats, extra_info, data),
            )
        )

    return rows

display_benchmark_rows

display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None

Print concise metric-aware summaries of parsed benchmark rows.

Parameters:

Name	Type	Description	Default
rows	Iterable[ParsedBenchmarkRow]	Parsed benchmark rows.	required
stream	TextIO | None	Output stream. Defaults to sys.stdout.	None

Source code in src/benchmatrix/bench_results.py

def display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None:
    """Print concise metric-aware summaries of parsed benchmark rows.

    Args:
        rows: Parsed benchmark rows.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    for row in rows:
        display_benchmark_row(row, stream=stream)

display_benchmark_row

display_benchmark_row(
    row: ParsedBenchmarkRow, stream: TextIO | None = None
) -> None

Print one metric-aware summary of a parsed benchmark row.

Parameters:

Name	Type	Description	Default
row	ParsedBenchmarkRow	Parsed benchmark row to display.	required
stream	TextIO | None	Output stream. Defaults to sys.stdout.	None

Source code in src/benchmatrix/bench_results.py

def display_benchmark_row(
    row: ParsedBenchmarkRow,
    stream: TextIO | None = None,
) -> None:
    """Print one metric-aware summary of a parsed benchmark row.

    Args:
        row: Parsed benchmark row to display.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    output = sys.stdout if stream is None else stream
    prefix = f"[{row.metric_name}] implementation={row.implementation_name} case={row.case_name}"

    if row.metric_name == METRIC_SINGLE_CALL_LATENCY:
        message = (
            f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))} "
            + f"median={_format_seconds(row.stats.get(STAT_MEDIAN))} "
            + f"min={_format_seconds(row.stats.get(STAT_MIN))}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_BATCH_THROUGHPUT:
        message = (
            f"{prefix} "
            + f"throughput_mean={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEAN))} "
            + f"throughput_median={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEDIAN))} "
            + f"unit={row.derived.get(DERIVED_THROUGHPUT_UNIT_LABEL)}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_TAIL_LATENCY:
        message = (
            f"{prefix} p50={_format_seconds(row.derived.get(DERIVED_P50))} "
            + f"p95={_format_seconds(row.derived.get(DERIVED_P95))} "
            + f"p99={_format_seconds(row.derived.get(DERIVED_P99))} "
            + f"max={_format_seconds(row.derived.get(DERIVED_MAX))}"
        )
        print(
            message,
            file=output,
        )
        return

    print(
        f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))}",
        file=output,
    )