| |
| |
| |
| |
| |
| |
| |
|
|
| """Assemble and deterministically render backend-neutral benchmark reports.""" |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import io |
| import json |
| import os |
| import tempfile |
| from collections.abc import Mapping |
| from dataclasses import asdict, dataclass |
| from pathlib import Path |
| from typing import Literal |
|
|
| from speech_text_data_aligners.benchmarking.metrics import ( |
| BenchmarkObservation, |
| GoldMetrics, |
| PairwiseMetrics, |
| ) |
| from speech_text_data_aligners.core.errors import BenchmarkError |
| from speech_text_data_aligners.core.validation import find_path_aliases |
|
|
| ReportFormat = Literal["json", "csv", "markdown"] |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class BackendReport: |
| """Bundle one backend's ordered observations and supported gold metrics. |
| |
| Parameters |
| ---------- |
| backend_name: Stable backend report name. |
| observations: Ordered neutral row observations. |
| gold_metrics: Aggregate supported reference/runtime metrics. |
| |
| """ |
|
|
| backend_name: str |
| observations: tuple[BenchmarkObservation, ...] |
| gold_metrics: GoldMetrics |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class BenchmarkReport: |
| """Bundle deterministic per-backend and pairwise benchmark results. |
| |
| Parameters |
| ---------- |
| backend_reports: Ordered backend reports. |
| pairwise_metrics: Deterministically ordered backend-pair comparisons. |
| |
| """ |
|
|
| backend_reports: tuple[BackendReport, ...] |
| pairwise_metrics: tuple[PairwiseMetrics, ...] = () |
|
|
|
|
| def _metric_rows(report: BenchmarkReport) -> list[tuple[str, str, str]]: |
| """Flatten headline metrics for CSV and Markdown renderers. |
| |
| Parameters |
| ---------- |
| report: Typed benchmark report. |
| |
| Returns |
| ------- |
| Ordered ``scope, metric, value`` rows. |
| |
| """ |
| rows: list[tuple[str, str, str]] = [] |
| for backend in report.backend_reports: |
| metrics = backend.gold_metrics |
| values: tuple[tuple[str, object], ...] = ( |
| ("success_count", metrics.success_count), |
| ("failure_count", metrics.failure_count), |
| ("referenced_rows", metrics.referenced_rows), |
| ("missing_reference_rows", metrics.missing_reference_rows), |
| ("text_reference_count", metrics.text_reference_count), |
| ("missing_text_reference_count", metrics.missing_text_reference_count), |
| ("exact_text_match_count", metrics.exact_text_match_count), |
| ("exact_text_match_rate", metrics.exact_text_match_rate), |
| ("first_end_reference_count", metrics.first_end_reference_count), |
| ( |
| "missing_first_end_reference_count", |
| metrics.missing_first_end_reference_count, |
| ), |
| ("first_end_scored_count", metrics.first_end_errors.count), |
| ("first_end_mae_s", metrics.first_end_errors.mae), |
| ("first_end_median_s", metrics.first_end_errors.median), |
| ("first_end_max_s", metrics.first_end_errors.maximum), |
| ("total_runtime_s", metrics.total_runtime_s), |
| ("mean_runtime_s", metrics.mean_runtime_s), |
| ) |
| rows.extend( |
| (backend.backend_name, name, "" if value is None else str(value)) |
| for name, value in values |
| ) |
| artifacts = dict.fromkeys( |
| artifact |
| for observation in backend.observations |
| for artifact in observation.native_artifacts |
| ) |
| rows.extend( |
| (backend.backend_name, f"native_artifact_{index}", artifact) |
| for index, artifact in enumerate(artifacts, start=1) |
| ) |
| for pairwise in report.pairwise_metrics: |
| scope = f"{pairwise.backend_a} vs {pairwise.backend_b}" |
| rows.append((scope, "comparable_rows", str(pairwise.comparable_rows))) |
| for name, summary in ( |
| ("first_start", pairwise.first_start), |
| ("first_end", pairwise.first_end), |
| ("last_start", pairwise.last_start), |
| ("last_end", pairwise.last_end), |
| ): |
| rows.append( |
| ( |
| scope, |
| f"{name}_mae_s", |
| "" if summary.mae is None else str(summary.mae), |
| ) |
| ) |
| return rows |
|
|
|
|
| def render_report(report: BenchmarkReport, output_format: ReportFormat) -> str: |
| """Render a typed report as canonical JSON, CSV, or concise Markdown. |
| |
| Parameters |
| ---------- |
| report: Backend-neutral benchmark report. |
| output_format: One of ``json``, ``csv``, or ``markdown``. |
| |
| Returns |
| ------- |
| Deterministically ordered newline-terminated report text. |
| |
| Raises |
| ------ |
| BenchmarkError: If an unsupported format is supplied at runtime. |
| |
| """ |
| if output_format == "json": |
| return ( |
| json.dumps( |
| asdict(report), |
| sort_keys=True, |
| separators=(",", ":"), |
| ensure_ascii=False, |
| ) |
| + "\n" |
| ) |
| rows = _metric_rows(report) |
| if output_format == "csv": |
| stream = io.StringIO(newline="") |
| writer = csv.writer(stream, lineterminator="\n") |
| writer.writerow(("scope", "metric", "value")) |
| writer.writerows(rows) |
| return stream.getvalue() |
| if output_format == "markdown": |
| lines = ["| Scope | Metric | Value |", "|---|---|---:|"] |
| lines.extend( |
| f"| {scope} | {metric} | {value} |" for scope, metric, value in rows |
| ) |
| return "\n".join(lines) + "\n" |
| raise BenchmarkError(f"unsupported report format: {output_format}") |
|
|
|
|
| def _write_text(path: Path, content: str, *, overwrite: bool) -> None: |
| """Atomically publish one report text artifact. |
| |
| Parameters |
| ---------- |
| path: Destination path. |
| content: Fully rendered text. |
| overwrite: Explicit replacement authorization. |
| |
| Raises |
| ------ |
| BenchmarkError: If the destination exists or publication fails. |
| |
| """ |
| if path.exists() and not overwrite: |
| raise BenchmarkError(f"report exists and overwrite was not authorized: {path}") |
| path.parent.mkdir(parents=True, exist_ok=True) |
| temporary_path: Path | None = None |
| try: |
| with tempfile.NamedTemporaryFile( |
| mode="w", |
| encoding="utf-8", |
| newline="\n", |
| dir=path.parent, |
| prefix=f".{path.name}.", |
| suffix=".tmp", |
| delete=False, |
| ) as stream: |
| temporary_path = Path(stream.name) |
| stream.write(content) |
| stream.flush() |
| os.fsync(stream.fileno()) |
| os.replace(temporary_path, path) |
| except OSError as error: |
| if temporary_path is not None: |
| temporary_path.unlink(missing_ok=True) |
| raise BenchmarkError(f"cannot publish benchmark report: {path}") from error |
|
|
|
|
| def report_bundle_paths(base_path: Path) -> tuple[Path, Path, Path]: |
| """Derive the three deterministic report targets from one file prefix. |
| |
| Parameters |
| ---------- |
| base_path: File prefix whose existing suffix is replaced per format. |
| |
| Returns |
| ------- |
| JSON, CSV, and Markdown target paths in that order. |
| |
| """ |
| return ( |
| base_path.with_suffix(".json"), |
| base_path.with_suffix(".csv"), |
| base_path.with_suffix(".md"), |
| ) |
|
|
|
|
| def preflight_report_bundle( |
| base_path: Path, |
| *, |
| metadata_csv: Path, |
| reference_csv: Path, |
| overwrite: bool = False, |
| ) -> tuple[Path, Path, Path]: |
| """Validate all report targets against immutable benchmark inputs and each other. |
| |
| Parameters |
| ---------- |
| base_path: File prefix used to derive the report destinations. |
| metadata_csv: Immutable benchmark metadata input. |
| reference_csv: Immutable sparse reference input. |
| overwrite: Whether existing, non-input report targets may be replaced. |
| |
| Returns |
| ------- |
| Validated JSON, CSV, and Markdown target paths in that order. |
| |
| Raises |
| ------ |
| BenchmarkError: If a target aliases an input/another target or exists without |
| overwrite authorization. |
| |
| Side Effects: |
| Reads filesystem identity/existence metadata only; no report is published. |
| |
| """ |
| targets = report_bundle_paths(base_path) |
| named_paths: Mapping[str, Path] = { |
| "metadata CSV": metadata_csv, |
| "reference CSV": reference_csv, |
| "report JSON": targets[0], |
| "report CSV": targets[1], |
| "report Markdown": targets[2], |
| } |
| aliases = tuple( |
| pair |
| for pair in find_path_aliases(named_paths) |
| if not all(name in {"metadata CSV", "reference CSV"} for name in pair) |
| ) |
| if aliases: |
| left, right = aliases[0] |
| raise BenchmarkError(f"benchmark artifact paths alias: {left} and {right}") |
| if not overwrite and any(path.exists() for path in targets): |
| raise BenchmarkError("one or more report targets already exist") |
| return targets |
|
|
|
|
| def write_report_bundle( |
| report: BenchmarkReport, |
| base_path: Path, |
| *, |
| metadata_csv: Path, |
| reference_csv: Path, |
| overwrite: bool = False, |
| ) -> tuple[Path, Path, Path]: |
| """Write matching JSON, CSV, and Markdown reports from one typed result. |
| |
| Parameters |
| ---------- |
| report: Typed benchmark result. |
| base_path: File prefix; existing suffix is replaced for each format. |
| metadata_csv: Immutable metadata input that targets must not alias. |
| reference_csv: Immutable reference input that targets must not alias. |
| overwrite: Explicitly permit replacement of existing report files. |
| |
| Returns |
| ------- |
| JSON, CSV, and Markdown output paths in that order. |
| |
| Raises |
| ------ |
| BenchmarkError: If targets collide, exist without authorization, or publication |
| fails. |
| |
| Side Effects: |
| Atomically writes three small report artifacts. Targets are preflighted before |
| the first write so unauthorized overwrite cannot create a partial bundle. |
| |
| """ |
| targets = preflight_report_bundle( |
| base_path, |
| metadata_csv=metadata_csv, |
| reference_csv=reference_csv, |
| overwrite=overwrite, |
| ) |
| formats: tuple[ReportFormat, ...] = ("json", "csv", "markdown") |
| for path, output_format in zip(targets, formats, strict=True): |
| _write_text(path, render_report(report, output_format), overwrite=overwrite) |
| return targets |
|
|
