| |
| |
| |
| |
| |
| |
| |
|
|
| """Benchmark identical ordered metadata requests through injected neutral backends.""" |
|
|
| from __future__ import annotations |
|
|
| import itertools |
| import math |
| from collections.abc import Mapping |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from speech_text_data_aligners.application.annotate import ( |
| _execute_backend, |
| _prepare_rows, |
| ) |
| from speech_text_data_aligners.benchmarking.metrics import ( |
| BenchmarkObservation, |
| GoldReference, |
| RowKey, |
| calculate_gold_metrics, |
| calculate_pairwise_metrics, |
| ) |
| from speech_text_data_aligners.benchmarking.reports import ( |
| BackendReport, |
| BenchmarkReport, |
| ) |
| from speech_text_data_aligners.core.errors import BenchmarkError |
| from speech_text_data_aligners.core.extraction import extract_boundary_fields |
| from speech_text_data_aligners.core.protocols import AlignmentBackend |
| from speech_text_data_aligners.data.metadata import MetadataTable, read_metadata |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class BenchmarkOptions: |
| """Control benchmark row-failure behavior. |
| |
| Parameters |
| ---------- |
| strict: Abort on the first preflight, backend, mapping, or validation failure. |
| |
| """ |
|
|
| strict: bool = False |
|
|
|
|
| _DEFAULT_BENCHMARK_OPTIONS = BenchmarkOptions() |
|
|
|
|
| def _parse_optional_seconds(value: str, *, row_ordinal: int) -> float | None: |
| """Parse one optional finite, non-negative gold timestamp. |
| |
| Parameters |
| ---------- |
| value: Exact CSV cell; empty means missing reference. |
| row_ordinal: Zero-based reference row ordinal used in diagnostics. |
| |
| Returns |
| ------- |
| Parsed seconds or ``None`` for an empty cell. |
| |
| Raises |
| ------ |
| BenchmarkError: If a non-empty value is invalid, negative, NaN, or infinite. |
| |
| """ |
| if value == "": |
| return None |
| try: |
| seconds = float(value) |
| except ValueError as error: |
| raise BenchmarkError( |
| f"reference row {row_ordinal} has an invalid first_word_end_s" |
| ) from error |
| if not math.isfinite(seconds) or seconds < 0: |
| raise BenchmarkError( |
| f"reference row {row_ordinal} has an invalid first_word_end_s" |
| ) |
| return seconds |
|
|
|
|
| def _read_references(table: MetadataTable) -> dict[RowKey, GoldReference]: |
| """Parse the supported sparse reference fields with unambiguous joins. |
| |
| Parameters |
| ---------- |
| table: Reference table preserving exact CSV cells. |
| |
| Returns |
| ------- |
| References keyed by ``(line_id, audio_id)``. |
| |
| Raises |
| ------ |
| BenchmarkError: If fields are missing, keys repeat, or timestamps are invalid. |
| |
| """ |
| required = ("first_word_text", "first_word_end_s") |
| if any(table.fieldnames.count(name) != 1 for name in required): |
| raise BenchmarkError("reference CSV lacks the supported first-word fields") |
| references: dict[RowKey, GoldReference] = {} |
| for ordinal in range(len(table.rows)): |
| row = table.row_dict(ordinal) |
| reference = GoldReference( |
| line_id=row["line_id"], |
| audio_id=row["audio_id"], |
| first_word_text=row["first_word_text"] or None, |
| first_word_end_s=_parse_optional_seconds( |
| row["first_word_end_s"], row_ordinal=ordinal |
| ), |
| ) |
| if reference.key in references: |
| raise BenchmarkError("reference CSV contains a duplicate join key") |
| references[reference.key] = reference |
| return references |
|
|
|
|
| def benchmark_backends( |
| *, |
| backends: Mapping[str, AlignmentBackend], |
| metadata_csv: Path, |
| reference_csv: Path, |
| audio_dir: Path, |
| options: BenchmarkOptions = _DEFAULT_BENCHMARK_OPTIONS, |
| ) -> BenchmarkReport: |
| """Run identical ordered requests through each injected backend and score results. |
| |
| Parameters |
| ---------- |
| backends: Insertion-ordered report name to protocol instance mapping. |
| metadata_csv: Metadata rows to align. |
| reference_csv: Sparse first-word gold/reference CSV. |
| audio_dir: Directory containing WAVs resolved from ``audio_id``. |
| options: Strict or continue-on-error policy. |
| |
| Returns |
| ------- |
| Typed per-backend gold/runtime and pairwise-disagreement report. |
| |
| Raises |
| ------ |
| BenchmarkError: If backend names/reference joins are invalid. |
| SpeechTextAlignerError: If strict preparation/alignment fails. |
| |
| Side Effects: |
| Probes source WAVs and calls each backend exactly once with the same request |
| tuple; it does not write reports or mutate metadata. |
| |
| """ |
| if not backends: |
| raise BenchmarkError("at least one backend is required") |
| if any(not name for name in backends): |
| raise BenchmarkError("benchmark backend report names cannot be empty") |
| prepared = _prepare_rows(metadata_csv, audio_dir, strict=options.strict) |
| references = _read_references(read_metadata(reference_csv)) |
| metadata_keys = { |
| ( |
| prepared.table.row_dict(index)["line_id"], |
| prepared.table.row_dict(index)["audio_id"], |
| ) |
| for index in range(len(prepared.table.rows)) |
| } |
| if set(references).difference(metadata_keys): |
| raise BenchmarkError("reference CSV contains rows absent from metadata") |
|
|
| request_by_id = {request.request_id: request for request in prepared.requests} |
| backend_reports: list[BackendReport] = [] |
| for report_name, backend in backends.items(): |
| outcomes = _execute_backend(prepared, backend, strict=options.strict) |
| observations: list[BenchmarkObservation] = [] |
| for ordinal, outcome in enumerate(outcomes): |
| row = prepared.table.row_dict(ordinal) |
| boundaries = None |
| runtime = None |
| failure_code = None |
| if outcome.result is not None: |
| request = request_by_id[outcome.request_id] |
| boundaries = extract_boundary_fields(request, outcome.result) |
| runtime = outcome.result.runtime_s |
| native_artifacts = outcome.result.provenance.artifacts |
| else: |
| assert outcome.failure is not None |
| failure_code = outcome.failure.code |
| native_artifacts = ( |
| outcome.failure.provenance.artifacts |
| if outcome.failure.provenance is not None |
| else () |
| ) |
| observations.append( |
| BenchmarkObservation( |
| backend_name=report_name, |
| line_id=row["line_id"], |
| audio_id=row["audio_id"], |
| boundaries=boundaries, |
| runtime_s=runtime, |
| failure_code=failure_code, |
| native_artifacts=native_artifacts, |
| ) |
| ) |
| observations_tuple = tuple(observations) |
| backend_reports.append( |
| BackendReport( |
| backend_name=report_name, |
| observations=observations_tuple, |
| gold_metrics=calculate_gold_metrics(observations_tuple, references), |
| ) |
| ) |
|
|
| pairwise = tuple( |
| calculate_pairwise_metrics(left.observations, right.observations) |
| for left, right in itertools.combinations(backend_reports, 2) |
| ) |
| return BenchmarkReport(tuple(backend_reports), pairwise) |
|
|
|
|
| def benchmark_backend( |
| *, |
| backend: AlignmentBackend, |
| metadata_csv: Path, |
| reference_csv: Path, |
| audio_dir: Path, |
| options: BenchmarkOptions = _DEFAULT_BENCHMARK_OPTIONS, |
| ) -> BenchmarkReport: |
| """Benchmark one injected backend as a convenience wrapper. |
| |
| Parameters |
| ---------- |
| backend: Backend-neutral protocol instance. |
| metadata_csv: Metadata rows to align. |
| reference_csv: Sparse first-word gold/reference CSV. |
| audio_dir: Directory containing WAVs resolved from ``audio_id``. |
| options: Strict or continue-on-error policy. |
| |
| Returns |
| ------- |
| A typed report containing one backend and no pairwise comparisons. |
| |
| Raises |
| ------ |
| SpeechTextAlignerError: If validation/preparation fails as documented by |
| :func:`benchmark_backends`. |
| |
| """ |
| return benchmark_backends( |
| backends={backend.name: backend}, |
| metadata_csv=metadata_csv, |
| reference_csv=reference_csv, |
| audio_dir=audio_dir, |
| options=options, |
| ) |
|
|
