File size: 5,866 Bytes
51e84d2 fc1dcbc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | # 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
"""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:
# samefile requires both paths to exist; canonical targets remain
# authoritative when the supplemental identity check is unavailable.
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
|