| |
| |
| |
| |
| |
| |
| |
|
|
| """Validate alignment results and filesystem publication-path relationships.""" |
|
|
| from __future__ import annotations |
|
|
| import math |
| import os |
| from collections.abc import Mapping |
| from contextlib import suppress |
| from pathlib import Path |
|
|
| from speech_text_data_aligners.core.errors import AlignmentValidationError |
| from speech_text_data_aligners.core.models import AlignmentRequest, AlignmentResult |
| from speech_text_data_aligners.core.normalization import lexicalize |
|
|
|
|
| def _canonical_path(path: Path) -> Path: |
| """Return a normalized absolute target without requiring the leaf to exist. |
| |
| Parameters |
| ---------- |
| path: Filesystem spelling to canonicalize. |
| |
| Returns |
| ------- |
| Absolute, normalized path with resolvable symlink components followed. |
| |
| Notes |
| ----- |
| Resolution failures fall back to the normalized absolute spelling so path |
| comparison remains available for not-yet-created or inaccessible targets. |
| |
| """ |
| absolute = Path(os.path.abspath(path)) |
| try: |
| resolved = absolute.resolve(strict=False) |
| except (OSError, RuntimeError): |
| resolved = absolute |
| return Path(os.path.normcase(resolved)) |
|
|
|
|
| def find_path_aliases(paths: Mapping[str, Path]) -> tuple[tuple[str, str], ...]: |
| """Find deterministic pairwise aliases in a named publication-path set. |
| |
| Canonical target comparison detects relative/absolute, ``.``/``..``, and |
| resolvable symlink spellings without requiring a destination leaf to exist. |
| ``Path.samefile`` additionally detects existing hard links and other same-file |
| aliases. Filesystem lookup errors merely make that supplemental check unavailable. |
| |
| Parameters |
| ---------- |
| paths: Insertion-ordered unique labels mapped to filesystem paths. |
| |
| Returns |
| ------- |
| Ordered pairs of labels whose paths identify the same target. |
| |
| Side Effects: |
| May stat existing paths for same-file comparison; it never mutates them. |
| |
| """ |
| items = tuple(paths.items()) |
| canonical = tuple(_canonical_path(path) for _, path in items) |
| aliases: list[tuple[str, str]] = [] |
| for left_index, (left_name, _) in enumerate(items): |
| for right_index in range(left_index + 1, len(items)): |
| right_name, _ = items[right_index] |
| same_target = canonical[left_index] == canonical[right_index] |
| if not same_target: |
| |
| |
| with suppress(OSError, ValueError): |
| same_target = canonical[left_index].samefile(canonical[right_index]) |
| if same_target: |
| aliases.append((left_name, right_name)) |
| return tuple(aliases) |
|
|
|
|
| def validate_alignment_result( |
| request: AlignmentRequest, |
| result: AlignmentResult, |
| ) -> AlignmentResult: |
| """Validate identity, completeness, ordering, timestamps, and audio bounds. |
| |
| Parameters |
| ---------- |
| request: Authoritative transcript and probed positive audio duration. |
| result: Candidate backend-neutral successful result. |
| |
| Returns |
| ------- |
| The unchanged result when every invariant holds. |
| |
| Raises |
| ------ |
| AlignmentValidationError: If any request/result invariant is violated. |
| TranscriptError: If the authoritative transcript contains no lexical words. |
| |
| """ |
| if result.request_id != request.request_id: |
| raise AlignmentValidationError("result request identity does not match request") |
| if not math.isfinite(request.audio_duration_s) or request.audio_duration_s <= 0: |
| raise AlignmentValidationError("audio duration must be finite and positive") |
| if not math.isfinite(result.runtime_s) or result.runtime_s < 0: |
| raise AlignmentValidationError( |
| "backend runtime must be finite and non-negative" |
| ) |
|
|
| expected = lexicalize(request.transcript) |
| if len(result.words) != len(expected): |
| raise AlignmentValidationError( |
| "alignment must contain exactly one record per lexical word" |
| ) |
|
|
| previous_end = 0.0 |
| seen_indices: set[int] = set() |
| for position, (alignment, lexical_word) in enumerate( |
| zip(result.words, expected, strict=True) |
| ): |
| if alignment.lexical_word != lexical_word: |
| raise AlignmentValidationError( |
| f"alignment at position {position} does not map to the authoritative word" |
| ) |
| if lexical_word.index in seen_indices: |
| raise AlignmentValidationError( |
| "alignment contains a duplicate lexical index" |
| ) |
| seen_indices.add(lexical_word.index) |
| if not math.isfinite(alignment.start_s) or not math.isfinite(alignment.end_s): |
| raise AlignmentValidationError("alignment timestamps must be finite") |
| if alignment.start_s < 0 or alignment.start_s > alignment.end_s: |
| raise AlignmentValidationError("alignment interval is negative or reversed") |
| if alignment.end_s > request.audio_duration_s: |
| raise AlignmentValidationError("alignment extends beyond the WAV duration") |
| if alignment.start_s < previous_end: |
| raise AlignmentValidationError( |
| "word alignments overlap or are non-monotonic" |
| ) |
| if alignment.confidence is not None and not math.isfinite(alignment.confidence): |
| raise AlignmentValidationError("alignment confidence must be finite") |
| previous_end = alignment.end_s |
| return result |
|
|
