| """Records shared by the data-pipeline stages. |
| |
| All times are integer milliseconds relative to the start of the *segment* (``*_ms``) |
| unless a name says otherwise (``file_*_ms`` = relative to the source file). Confidences |
| are probabilities in [0, 1]. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import math |
| from dataclasses import dataclass, field |
|
|
| from nida.text.turkish import nfc |
|
|
| __all__ = [ |
| "EventDetection", |
| "QuarantineRecord", |
| "TimedWord", |
| "WordStyleScore", |
| "segment_id", |
| ] |
|
|
|
|
| def segment_id(source_id: str, file_id: str, start_ms: int, end_ms: int) -> str: |
| """Content-addressed segment id: ``sha1(source, file, start_ms, end_ms)[:16]``. |
| |
| Fields are joined with U+001F (unit separator) so that no field value can make two |
| different tuples collide. Ids are NFC-normalised first: the same file name in NFD |
| (e.g. from macOS) must not get a second id (train/test leakage). |
| |
| Raises: |
| ValueError: if a field contains U+001F, a time is not an int, or the times are not |
| ``0 <= start < end``. |
| """ |
| source_id, file_id = nfc(source_id), nfc(file_id) |
| for name, value in (("source_id", source_id), ("file_id", file_id)): |
| if not value or "\x1f" in value: |
| raise ValueError(f"invalid {name} {value!r}") |
| _check_ms(start_ms, "start_ms") |
| _check_ms(end_ms, "end_ms") |
| if not 0 <= start_ms < end_ms: |
| raise ValueError(f"invalid segment times {start_ms}..{end_ms} ms") |
| key = "\x1f".join((source_id, file_id, str(start_ms), str(end_ms))) |
| return hashlib.sha1(key.encode("utf-8")).hexdigest()[:16] |
|
|
|
|
| def _check_ms(value: object, what: str) -> None: |
| """Times are integer ms; a float (e.g. seconds from an aligner) is a caller bug.""" |
| if type(value) is not int: |
| raise ValueError(f"{what}: time must be an int number of ms, got {value!r}") |
|
|
|
|
| def _check_interval(start_ms: int, end_ms: int, what: str) -> None: |
| _check_ms(start_ms, what) |
| _check_ms(end_ms, what) |
| if start_ms < 0 or end_ms <= start_ms: |
| raise ValueError(f"{what}: invalid interval {start_ms}..{end_ms} ms") |
|
|
|
|
| def _check_prob(value: float, what: str) -> None: |
| if not (math.isfinite(value) and 0.0 <= value <= 1.0): |
| raise ValueError(f"{what}: probability {value!r} not in [0, 1]") |
|
|
|
|
| @dataclass(frozen=True) |
| class TimedWord: |
| """One word of the primary transcript with its forced-alignment interval (S5).""" |
|
|
| text: str |
| start_ms: int |
| end_ms: int |
| align_score: float | None = None |
|
|
| def __post_init__(self) -> None: |
| if not self.text or any(ch.isspace() for ch in self.text): |
| raise ValueError(f"word text must be non-empty without whitespace: {self.text!r}") |
| _check_interval(self.start_ms, self.end_ms, f"word {self.text!r}") |
|
|
| @property |
| def duration_ms(self) -> int: |
| return self.end_ms - self.start_ms |
|
|
|
|
| @dataclass(frozen=True) |
| class EventDetection: |
| """A detected non-verbal event (S6). ``kind`` is a tag-spec event id (``laugh``, …).""" |
|
|
| kind: str |
| start_ms: int |
| end_ms: int |
| conf: float |
| source: str |
|
|
| def __post_init__(self) -> None: |
| _check_interval(self.start_ms, self.end_ms, f"event {self.kind!r}") |
| _check_prob(self.conf, f"event {self.kind!r}") |
| if not self.source: |
| raise ValueError("event source must be non-empty") |
|
|
| @property |
| def duration_ms(self) -> int: |
| return self.end_ms - self.start_ms |
|
|
|
|
| @dataclass(frozen=True) |
| class WordStyleScore: |
| """Per-word probability that a word is spoken in span style ``style`` (S6). |
| |
| ``scores[i]`` belongs to word ``i`` of the segment; ``None`` = not measurable (e.g. |
| word too short for a pitch estimate), which counts as "not marked". |
| """ |
|
|
| style: str |
| scores: tuple[float | None, ...] |
| source: str |
|
|
| def __post_init__(self) -> None: |
| for score in self.scores: |
| if score is not None: |
| _check_prob(score, f"style {self.style!r}") |
|
|
|
|
| @dataclass(frozen=True) |
| class QuarantineRecord: |
| """An item dropped by a stage, with the reason (never dropped silently).""" |
|
|
| item_id: str |
| stage: str |
| reason: str |
| details: dict[str, object] = field(default_factory=dict) |
|
|
| def __post_init__(self) -> None: |
| if not (self.item_id and self.stage and self.reason): |
| raise ValueError("quarantine record needs item_id, stage and reason") |
|
|
