File size: 4,405 Bytes
c0bce59 | 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 | """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")
|