# License: Apache-2.0 License # Created by: Patrick Lumbantobing, VertoX-AI # Copyright (c) 2026 VertoX-AI. All rights reserved. # # This work is licensed under the Apache-2.0 License. # To view a copy of this license, visit # https://www.apache.org/licenses/LICENSE-2.0 """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, )