File size: 8,019 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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 | # 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
"""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
|