| |
| |
| |
| |
| |
| |
| |
|
|
| """Define immutable backend-neutral alignment requests, results, and diagnostics. |
| |
| The records in this module are deliberately free of I/O and vendor types. Tuple-based |
| key/value collections keep context and diagnostic values immutable while remaining |
| straightforward to serialize at application boundaries. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from enum import StrEnum |
| from pathlib import Path |
|
|
| type JsonScalar = str | int | float | bool | None |
| type JsonValue = JsonScalar | tuple[JsonValue, ...] | tuple[tuple[str, JsonValue], ...] |
| type Attributes = tuple[tuple[str, JsonValue], ...] |
|
|
|
|
| class DiagnosticSeverity(StrEnum): |
| """Classify how strongly a diagnostic affects an alignment outcome.""" |
|
|
| INFO = "info" |
| WARNING = "warning" |
| ERROR = "error" |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class Diagnostic: |
| """Describe a stable, serializable event observed during alignment. |
| |
| Parameters |
| ---------- |
| code: Stable machine-facing diagnostic identifier. |
| severity: Importance of the diagnostic. |
| message: Safe human-facing explanation without secrets. |
| stage: Pipeline stage that emitted the diagnostic. |
| details: Immutable JSON-safe detail pairs. |
| |
| """ |
|
|
| code: str |
| severity: DiagnosticSeverity |
| message: str |
| stage: str |
| details: Attributes = () |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class AlignmentRequest: |
| """Carry one authoritative transcript and its already-probed WAV metadata. |
| |
| Parameters |
| ---------- |
| request_id: Caller-defined unique identity; metadata services use row ordinal. |
| audio_path: Resolved local WAV path. |
| transcript: Exact authoritative transcript, never an ASR replacement. |
| audio_duration_s: Positive finite WAV duration in seconds. |
| context: Immutable JSON-safe caller identifiers. |
| |
| """ |
|
|
| request_id: str |
| audio_path: Path |
| transcript: str |
| audio_duration_s: float |
| context: Attributes = () |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class LexicalWord: |
| """Identify one original-facing lexical word in a transcript. |
| |
| Parameters |
| ---------- |
| index: Zero-based lexical position. |
| text: Exact surface substring from the authoritative transcript. |
| start_char: Inclusive character offset in the transcript. |
| end_char: Exclusive character offset in the transcript. |
| |
| """ |
|
|
| index: int |
| text: str |
| start_char: int |
| end_char: int |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class WordAlignment: |
| """Associate one original lexical word with an acoustic interval. |
| |
| Parameters |
| ---------- |
| lexical_word: Original-facing word identity and source span. |
| start_s: Acoustic start in seconds. |
| end_s: Acoustic end in seconds. |
| confidence: Optional finite backend confidence. |
| attributes: Immutable backend diagnostic attributes. |
| |
| """ |
|
|
| lexical_word: LexicalWord |
| start_s: float |
| end_s: float |
| confidence: float | None = None |
| attributes: Attributes = () |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class AlignmentProvenance: |
| """Record backend, model, runtime, and artifact identity for an alignment. |
| |
| Parameters |
| ---------- |
| backend_name: Stable backend name. |
| backend_version: Installed backend package or executable version. |
| adapter_version: Project adapter version. |
| model_id: Model/checkpoint identifier. |
| model_revision: Immutable model revision when available. |
| model_sha256: Local model/checkpoint hash when available. |
| dictionary_id: Dictionary identifier when applicable. |
| dictionary_revision: Immutable dictionary revision when available. |
| dictionary_sha256: Local dictionary hash when available. |
| python_version: Runtime interpreter version. |
| torch_version: PyTorch version when applicable. |
| cuda_version: CUDA version when applicable. |
| device: CPU/GPU device description when applicable. |
| config_sha256: Canonical redacted configuration fingerprint. |
| native_command_sha256: Invoked native script/command fingerprint. |
| artifacts: Immutable retained artifact references. |
| attributes: Additional immutable JSON-safe provenance fields. |
| |
| """ |
|
|
| backend_name: str |
| backend_version: str | None = None |
| adapter_version: str | None = None |
| model_id: str | None = None |
| model_revision: str | None = None |
| model_sha256: str | None = None |
| dictionary_id: str | None = None |
| dictionary_revision: str | None = None |
| dictionary_sha256: str | None = None |
| python_version: str | None = None |
| torch_version: str | None = None |
| cuda_version: str | None = None |
| device: str | None = None |
| config_sha256: str | None = None |
| native_command_sha256: str | None = None |
| artifacts: tuple[str, ...] = () |
| attributes: Attributes = () |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class AlignmentResult: |
| """Represent a complete successful word alignment for one request. |
| |
| Parameters |
| ---------- |
| request_id: Identity copied from the corresponding request. |
| words: Complete ordered alignment tuple, one per lexical word. |
| provenance: Backend and runtime identity. |
| diagnostics: Non-fatal diagnostics retained with the result. |
| runtime_s: Measured backend runtime in seconds. |
| |
| """ |
|
|
| request_id: str |
| words: tuple[WordAlignment, ...] |
| provenance: AlignmentProvenance |
| diagnostics: tuple[Diagnostic, ...] = () |
| runtime_s: float = 0.0 |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class AlignmentFailure: |
| """Represent a structured failure without fabricated timestamps. |
| |
| Parameters |
| ---------- |
| request_id: Identity copied from the failed request. |
| code: Stable machine-facing failure code. |
| category: Stable broad failure category. |
| message: Safe human-facing explanation. |
| retryable: Whether retrying under unchanged inputs may reasonably succeed. |
| diagnostics: Structured evidence collected before failure. |
| provenance: Partial provenance when it was available. |
| |
| """ |
|
|
| request_id: str |
| code: str |
| category: str |
| message: str |
| retryable: bool = False |
| diagnostics: tuple[Diagnostic, ...] = () |
| provenance: AlignmentProvenance | None = None |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class AlignmentOutcome: |
| """Contain exactly one successful result or structured failure. |
| |
| Parameters |
| ---------- |
| result: Successful alignment, mutually exclusive with ``failure``. |
| failure: Failed alignment, mutually exclusive with ``result``. |
| |
| Raises |
| ------ |
| ValueError: If neither or both variants are supplied. |
| |
| """ |
|
|
| result: AlignmentResult | None = field(default=None) |
| failure: AlignmentFailure | None = field(default=None) |
|
|
| def __post_init__(self) -> None: |
| """Enforce the exactly-one variant invariant. |
| |
| Raises: |
| ValueError: If the outcome is empty or contradictory. |
| |
| """ |
| if (self.result is None) == (self.failure is None): |
| raise ValueError("an alignment outcome must contain exactly one variant") |
|
|
| @property |
| def request_id(self) -> str: |
| """Return the identity of the contained result or failure. |
| |
| Returns: |
| Stable request identifier. |
| |
| """ |
| variant = self.result if self.result is not None else self.failure |
| assert variant is not None |
| return variant.request_id |
|
|
| @property |
| def succeeded(self) -> bool: |
| """Report whether this outcome contains a result. |
| |
| Returns: |
| ``True`` for success and ``False`` for failure. |
| |
| """ |
| return self.result is not None |
|
|
