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	Mapping[str, object]	Raw pytest-benchmark timing statistics.
extra_info	Mapping[str, object]	Custom metadata from benchmark.extra_info.
derived	Mapping[str, object]	Derived metric-specific statistics computed from JSON output.
samples	tuple[float, ...]	Raw per-round timing samples in seconds.

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.
        samples: Raw per-round timing samples in seconds.
    """

    benchmark_name: str
    metric_name: MetricName
    implementation_name: str
    case_name: str
    stats: Mapping[str, object]
    extra_info: Mapping[str, object]
    derived: Mapping[str, object]
    samples: tuple[float, ...] = ()

BenchmarkRun dataclass

One parsed pytest-benchmark run containing a benchmark matrix.

Attributes:

Name	Type	Description
rows	tuple[ParsedBenchmarkRow, ...]	Benchmatrix rows in their source-file order.
metadata	Mapping[str, object]	Top-level pytest-benchmark metadata excluding benchmarks.
source	Path | None	Source JSON path, when the run was loaded from a file.

Source code in src/benchmatrix/bench_results.py

@dataclass(frozen=True, slots=True)
class BenchmarkRun:
    """One parsed pytest-benchmark run containing a benchmark matrix.

    Attributes:
        rows: Benchmatrix rows in their source-file order.
        metadata: Top-level pytest-benchmark metadata excluding ``benchmarks``.
        source: Source JSON path, when the run was loaded from a file.
    """

    rows: tuple[ParsedBenchmarkRow, ...]
    metadata: Mapping[str, object]
    source: Path | None = None

    def __post_init__(self) -> None:
        """Normalize run containers and reject duplicate matrix cells."""
        rows = tuple(self.rows)
        metadata = MappingProxyType(dict(self.metadata))
        seen: set[tuple[str, str, MetricName]] = set()

        if not rows:
            raise BenchmarkJsonError("Benchmark run must contain at least one benchmatrix row.")

        for row in rows:
            if not row.implementation_name or not row.case_name:
                raise BenchmarkJsonError("Benchmark run matrix identifiers must not be empty.")
            if row.metric_name not in KNOWN_METRICS:
                raise BenchmarkJsonError(f"Unsupported benchmatrix metric in benchmark run: {row.metric_name!r}.")

            key = (row.implementation_name, row.case_name, row.metric_name)
            if key in seen:
                implementation_name, case_name, metric_name = key
                message = (
                    "Duplicate benchmark matrix cell for "
                    + f"implementation={implementation_name!r}, case={case_name!r}, metric={metric_name!r}."
                )
                raise BenchmarkJsonError(message)
            seen.add(key)

        object.__setattr__(self, "rows", rows)
        object.__setattr__(self, "metadata", metadata)
        if self.source is not None:
            object.__setattr__(self, "source", Path(self.source))

    @property
    def implementations(self) -> tuple[str, ...]:
        """Return sorted implementation names represented in this run."""
        return tuple(sorted({row.implementation_name for row in self.rows}))

    @property
    def cases(self) -> tuple[str, ...]:
        """Return sorted case names represented in this run."""
        return tuple(sorted({row.case_name for row in self.rows}))

    @property
    def metrics(self) -> tuple[MetricName, ...]:
        """Return sorted metric names represented in this run."""
        return tuple(sorted({row.metric_name for row in self.rows}))

    def compare_to(
        self,
        candidate: BenchmarkRun,
        *,
        compatibility_policy: RunCompatibilityPolicy | None = None,
        regression_policy: RegressionPolicy | None = None,
        inference_policy: InferencePolicy | None = None,
    ) -> BenchmarkRunComparison:
        """Compare this baseline run with a candidate run.

        Args:
            candidate: Run whose values should be compared with this baseline.
            compatibility_policy: Environment checks to apply.
            regression_policy: Thresholds used to classify cell changes.
            inference_policy: Statistical inference and multiplicity controls.

        Returns:
            A matrix-aware comparison containing matched, missing, and
            incompatible cells.
        """
        from .bench_compare import compare_benchmark_runs

        return compare_benchmark_runs(
            self,
            candidate,
            compatibility_policy=compatibility_policy,
            regression_policy=regression_policy,
            inference_policy=inference_policy,
        )

implementations property

implementations: tuple[str, ...]

Return sorted implementation names represented in this run.

cases property

cases: tuple[str, ...]

Return sorted case names represented in this run.

metrics property

metrics: tuple[MetricName, ...]

Return sorted metric names represented in this run.

__post_init__

__post_init__() -> None

Normalize run containers and reject duplicate matrix cells.

Source code in src/benchmatrix/bench_results.py

def __post_init__(self) -> None:
    """Normalize run containers and reject duplicate matrix cells."""
    rows = tuple(self.rows)
    metadata = MappingProxyType(dict(self.metadata))
    seen: set[tuple[str, str, MetricName]] = set()

    if not rows:
        raise BenchmarkJsonError("Benchmark run must contain at least one benchmatrix row.")

    for row in rows:
        if not row.implementation_name or not row.case_name:
            raise BenchmarkJsonError("Benchmark run matrix identifiers must not be empty.")
        if row.metric_name not in KNOWN_METRICS:
            raise BenchmarkJsonError(f"Unsupported benchmatrix metric in benchmark run: {row.metric_name!r}.")

        key = (row.implementation_name, row.case_name, row.metric_name)
        if key in seen:
            implementation_name, case_name, metric_name = key
            message = (
                "Duplicate benchmark matrix cell for "
                + f"implementation={implementation_name!r}, case={case_name!r}, metric={metric_name!r}."
            )
            raise BenchmarkJsonError(message)
        seen.add(key)

    object.__setattr__(self, "rows", rows)
    object.__setattr__(self, "metadata", metadata)
    if self.source is not None:
        object.__setattr__(self, "source", Path(self.source))

compare_to

compare_to(
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
) -> BenchmarkRunComparison

Compare this baseline run with a candidate run.

Parameters:

Name	Type	Description	Default
candidate	BenchmarkRun	Run whose values should be compared with this baseline.	required
compatibility_policy	RunCompatibilityPolicy | None	Environment checks to apply.	None
regression_policy	RegressionPolicy | None	Thresholds used to classify cell changes.	None
inference_policy	InferencePolicy | None	Statistical inference and multiplicity controls.	None

Returns:

Type	Description
BenchmarkRunComparison	A matrix-aware comparison containing matched, missing, and
BenchmarkRunComparison	incompatible cells.

Source code in src/benchmatrix/bench_results.py

def compare_to(
    self,
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare this baseline run with a candidate run.

    Args:
        candidate: Run whose values should be compared with this baseline.
        compatibility_policy: Environment checks to apply.
        regression_policy: Thresholds used to classify cell changes.
        inference_policy: Statistical inference and multiplicity controls.

    Returns:
        A matrix-aware comparison containing matched, missing, and
        incompatible cells.
    """
    from .bench_compare import compare_benchmark_runs

    return compare_benchmark_runs(
        self,
        candidate,
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        inference_policy=inference_policy,
    )

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.
    """
    return list(load_benchmark_run(path).rows)

load_benchmark_run

load_benchmark_run(path: str | Path) -> BenchmarkRun

Load a benchmatrix run from pytest-benchmark JSON.

Parameters:

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

Returns:

Type	Description
BenchmarkRun	A first-class run containing matrix rows and top-level run metadata.

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_run(path: str | Path) -> BenchmarkRun:
    """Load a benchmatrix run from pytest-benchmark JSON.

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

    Returns:
        A first-class run containing matrix rows and top-level run metadata.

    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}")
        _ = _require_bool(
            extra_info.get(KEY_CASE_FRESH_INPUTS),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_CASE_FRESH_INPUTS}",
        )

        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}",
        )
        if metric_name == METRIC_TAIL_LATENCY:
            _validate_tail_metadata(
                extra_info,
                path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}",
            )
        data = _extract_benchmark_data(entry, stats, metric_name, path=entry_path)
        stats_path = f"{entry_path}.{JSON_KEY_STATS}"
        extra_info_path = f"{entry_path}.{JSON_KEY_EXTRA_INFO}"

        rows.append(
            ParsedBenchmarkRow(
                benchmark_name=_benchmark_name(entry, path=entry_path),
                metric_name=metric_name,
                implementation_name=_require_non_empty_string(
                    extra_info.get(KEY_IMPLEMENTATION_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_IMPLEMENTATION_NAME}",
                ),
                case_name=_require_non_empty_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,
                    stats_path=stats_path,
                    extra_info_path=extra_info_path,
                ),
                samples=tuple(data),
            )
        )

    metadata = {key: value for key, value in payload_mapping.items() if key != JSON_KEY_BENCHMARKS}
    _validate_run_metadata(metadata)
    return BenchmarkRun(rows=tuple(rows), metadata=metadata, source=path_obj)

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,
    )