| """S7 — deterministic fusion of transcript, alignment and paralinguistic detections. |
| |
| Turns one segment's aligned words (S5), event detections and per-word style scores (S6) |
| into tagged text following docs/02-tag-spec.md, or into a quarantine record with a |
| machine-readable reason. No LLM, no randomness: the same inputs always give the same |
| output. Rules (docs/03-data-pipeline.md S7, tag-spec §7): |
| |
| 1. Bands: a detection below ``low`` is ignored (and counted); at or above ``high`` it is |
| used; in the middle band ``[low, high)`` it drops the whole segment |
| (``ambiguous_event`` / ``ambiguous_style``) — unless every millisecond of it is |
| covered by confident detections of the same family (then the event is there anyway). |
| 2. Families: ``laugh`` and ``laugh_long`` are one family, every other kind its own. |
| Confident detections of one family that overlap or are at most ``merge_gap_ms`` apart |
| are merged. |
| 3. Speech overlap: a laugh/cry detection covering more than ``style_word_overlap`` of a |
| word's duration marks the word ``laughing`` / ``crying``. If its longest non-speech |
| part is at least ``min_residual_event_ms`` it is *also* placed as an event. Any other |
| detection whose speech overlap exceeds ``max_event_speech_overlap`` of its length |
| drops the segment (``event_overlaps_speech``: false positive or overlapped audio). |
| 4. Placement: an event is placed at the word gap nearest the middle of its longest |
| non-speech part. A laugh becomes ``laugh_long`` if a detector said so or that placed |
| part lasts at least ``laugh_long_min_ms``. Placed events of different kinds whose |
| parts overlap by more than ``max_cross_kind_overlap`` of the shorter one drop the |
| segment (``conflicting_events``): taggers often fire related classes on one sound. |
| 5. Spans: consecutive words marked with the same style form a span if they are at least |
| ``min_span_words`` words or ``min_span_ms`` long; a shorter run drops the segment |
| (``short_style_run``: audible but untaggable). A word with two styles drops the |
| segment (``conflicting_styles``). Per-word score tracks of an ``ignored_styles`` |
| style are discarded before any of this: the detector is known to be unreliable, so |
| it neither tags nor drops anything (unlike ``disabled_tags``, which drop the segment |
| whenever the detector fires at or above ``low``). |
| 6. Pauses: measured on the longest part of an inter-word gap not covered by a confident |
| detection. Longer than ``pause_long_ms`` → ``pause_long``; at least ``pause_short_ms`` |
| → ``pause_short`` unless the previous word ends in punctuation. None at segment edges. |
| 7. Text: words are NFC; standalone fillers are canonicalised to their lowercase canonical |
| form (tag-spec §4); words with digits drop the segment (``unnormalised_number``) until |
| number normalisation exists; ``[ ] < >`` in a word drops it |
| (``reserved_char_in_text``). The output is validated by round-tripping through |
| :func:`nida.text.tags.parse` (``invalid_output``). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import math |
| import unicodedata |
| from collections.abc import Mapping, Sequence |
| from dataclasses import dataclass, field |
|
|
| from nida.data.types import EventDetection, QuarantineRecord, TimedWord, WordStyleScore |
| from nida.text.fillers import canonical_filler |
| from nida.text.tags import EVENTS, SPANS, Event, Item, Span, TagSyntaxError, Word, parse, render |
| from nida.text.turkish import nfc, tr_lower |
|
|
| __all__ = [ |
| "EMOTION_LABELS", |
| "STAGE", |
| "STYLE_FROM_EVENT", |
| "Band", |
| "EmotionCondition", |
| "FusionConfig", |
| "FusionResult", |
| "TagRecord", |
| "emotion_condition", |
| "fuse_segment", |
| "normalise_word", |
| ] |
|
|
| STAGE = "fusion" |
|
|
| |
| STYLE_FROM_EVENT: Mapping[str, str] = {"laugh": "laughing", "cry": "crying"} |
|
|
| |
| EMOTION_LABELS: frozenset[str] = frozenset( |
| {"angry", "disgusted", "fearful", "happy", "neutral", "other", "sad", "surprised", "unknown"} |
| ) |
| _NULL_EMOTIONS = frozenset({"other", "unknown"}) |
| _PAUSE_EVENTS = frozenset({"pause_short", "pause_long"}) |
| _RESERVED_CHARS = frozenset("[]<>") |
|
|
|
|
| def _family(kind: str) -> str: |
| return "laugh" if kind in {"laugh", "laugh_long"} else kind |
|
|
|
|
| @dataclass(frozen=True) |
| class Band: |
| """Confidence band: ``< low`` ignored, ``[low, high)`` ambiguous, ``>= high`` used.""" |
|
|
| low: float |
| high: float |
|
|
| def __post_init__(self) -> None: |
| if not (0.0 <= self.low <= self.high <= 1.0): |
| raise ValueError(f"invalid band low={self.low} high={self.high}") |
|
|
|
|
| _INT_FIELDS = ( |
| "min_span_words", |
| "min_span_ms", |
| "min_residual_event_ms", |
| "laugh_long_min_ms", |
| "merge_gap_ms", |
| "pause_short_ms", |
| "pause_long_ms", |
| ) |
| _FRACTION_FIELDS = ( |
| "style_word_overlap", |
| "max_event_speech_overlap", |
| "max_cross_kind_overlap", |
| "emotion_min_conf", |
| ) |
|
|
|
|
| @dataclass(frozen=True) |
| class FusionConfig: |
| """Fusion thresholds (configs/data/fusion.yaml). All fields are required.""" |
|
|
| bands: Mapping[str, Band] |
| disabled_tags: frozenset[str] |
| ignored_styles: frozenset[str] |
| min_span_words: int |
| min_span_ms: int |
| style_word_overlap: float |
| max_event_speech_overlap: float |
| max_cross_kind_overlap: float |
| min_residual_event_ms: int |
| laugh_long_min_ms: int |
| merge_gap_ms: int |
| pause_short_ms: int |
| pause_long_ms: int |
| emotion_min_conf: float |
|
|
| def __post_init__(self) -> None: |
| required = (set(EVENTS) - _PAUSE_EVENTS) | set(SPANS) |
| if set(self.bands) != required: |
| missing = sorted(required - set(self.bands)) |
| extra = sorted(set(self.bands) - required) |
| raise ValueError(f"bands must cover exactly the detectable tags: {missing=} {extra=}") |
| unknown = self.disabled_tags - (set(EVENTS) | set(SPANS)) |
| if unknown: |
| raise ValueError(f"unknown disabled tags {sorted(unknown)}") |
| |
| |
| scored = set(SPANS) - set(STYLE_FROM_EVENT.values()) |
| not_styles = self.ignored_styles - scored |
| if not_styles: |
| raise ValueError( |
| f"ignored_styles must be span styles with score tracks {sorted(scored)}, " |
| f"got {sorted(not_styles)}" |
| ) |
| both = self.ignored_styles & self.disabled_tags |
| if both: |
| raise ValueError(f"tags both disabled and ignored: {sorted(both)}") |
| for name in _INT_FIELDS: |
| value = getattr(self, name) |
| if isinstance(value, bool) or not isinstance(value, int) or value < 0: |
| raise ValueError(f"{name} must be a non-negative int, got {value!r}") |
| if self.min_span_words < 1: |
| raise ValueError("min_span_words must be >= 1") |
| if self.min_residual_event_ms < 1: |
| raise ValueError("min_residual_event_ms must be >= 1") |
| for name in _FRACTION_FIELDS: |
| value = getattr(self, name) |
| if isinstance(value, bool) or not isinstance(value, int | float): |
| raise ValueError(f"{name} must be a number, got {value!r}") |
| if not (math.isfinite(value) and 0.0 <= value <= 1.0): |
| raise ValueError(f"{name} must be in [0, 1], got {value!r}") |
| if not 0 < self.pause_short_ms <= self.pause_long_ms: |
| raise ValueError("need 0 < pause_short_ms <= pause_long_ms") |
|
|
| @classmethod |
| def from_mapping(cls, raw: Mapping[str, object]) -> FusionConfig: |
| """Build from a parsed YAML mapping; unknown or missing keys are errors.""" |
| expected = {f.name for f in cls.__dataclass_fields__.values()} |
| if set(raw) != expected: |
| raise ValueError( |
| f"fusion config keys: missing={sorted(expected - set(raw))} " |
| f"unknown={sorted(set(raw) - expected)}" |
| ) |
| bands_raw = raw["bands"] |
| if not isinstance(bands_raw, Mapping): |
| raise ValueError("bands must be a mapping") |
| bands: dict[str, Band] = {} |
| for kind, band in bands_raw.items(): |
| if not isinstance(band, Mapping) or set(band) != {"low", "high"}: |
| raise ValueError(f"band {kind!r} must be a mapping with exactly low/high") |
| low, high = band["low"], band["high"] |
| if not all(isinstance(v, int | float) and not isinstance(v, bool) for v in (low, high)): |
| raise ValueError(f"band {kind!r} values must be numbers") |
| bands[str(kind)] = Band(float(low), float(high)) |
| disabled = raw["disabled_tags"] |
| if not isinstance(disabled, list) or not all(isinstance(t, str) for t in disabled): |
| raise ValueError("disabled_tags must be a list of tag ids") |
| ignored = raw["ignored_styles"] |
| if not isinstance(ignored, list) or not all(isinstance(t, str) for t in ignored): |
| raise ValueError("ignored_styles must be a list of style ids") |
| lists = {"bands", "disabled_tags", "ignored_styles"} |
| values = {k: v for k, v in raw.items() if k not in lists} |
| |
| return cls( |
| bands=bands, |
| disabled_tags=frozenset(disabled), |
| ignored_styles=frozenset(ignored), |
| **values, |
| ) |
|
|
|
|
| @dataclass(frozen=True) |
| class TagRecord: |
| """Provenance of one emitted tag (tag-spec §7.3): ``{id, start, end, conf, source}``. |
| |
| ``start_ms``/``end_ms`` is what the tag stands for (the placed non-speech part of an |
| event, the words of a span, the silence of a pause); ``det_start_ms``/``det_end_ms`` |
| is the merged detector interval it came from, so it can be checked against the |
| detector output. |
| """ |
|
|
| tag: str |
| start_ms: int |
| end_ms: int |
| conf: float |
| source: str |
| det_start_ms: int |
| det_end_ms: int |
|
|
|
|
| @dataclass(frozen=True) |
| class FusionResult: |
| """Either ``items``/``text`` (accepted) or ``quarantine`` (dropped), never both.""" |
|
|
| segment_id: str |
| items: tuple[Item, ...] | None |
| text: str | None |
| tags: tuple[TagRecord, ...] |
| quarantine: QuarantineRecord | None |
| ignored_detections: int = 0 |
| |
| |
| ignored_events: tuple[EventDetection, ...] = () |
|
|
| def __post_init__(self) -> None: |
| accepted = self.items is not None and self.text is not None |
| dropped = self.items is None and self.text is None and not self.tags |
| if self.quarantine is None and not accepted: |
| raise ValueError("accepted result needs items and text") |
| if self.quarantine is not None and not dropped: |
| raise ValueError("quarantined result must not carry items, text or tags") |
|
|
| @property |
| def accepted(self) -> bool: |
| return self.quarantine is None |
|
|
|
|
| @dataclass(frozen=True) |
| class EmotionCondition: |
| emotion: str | None |
| intensity: float | None |
|
|
|
|
| @dataclass |
| class _Merged: |
| family: str |
| kinds: set[str] |
| start_ms: int |
| end_ms: int |
| conf: float |
| sources: set[str] = field(default_factory=set) |
|
|
| @property |
| def source(self) -> str: |
| return "+".join(sorted(self.sources)) |
|
|
|
|
| @dataclass(frozen=True) |
| class _Placed: |
| gap: int |
| kind: str |
| start_ms: int |
| end_ms: int |
| event: _Merged |
|
|
|
|
| class _Drop(Exception): |
| """Internal: the segment is quarantined with ``reason``.""" |
|
|
| def __init__(self, reason: str, **details: object) -> None: |
| super().__init__(reason) |
| self.reason = reason |
| self.details = details |
|
|
|
|
| def _overlap_ms(a_start: int, a_end: int, b_start: int, b_end: int) -> int: |
| return max(0, min(a_end, b_end) - max(a_start, b_start)) |
|
|
|
|
| def _union(intervals: Sequence[tuple[int, int]]) -> list[tuple[int, int]]: |
| out: list[tuple[int, int]] = [] |
| for start, end in sorted(intervals): |
| if out and start <= out[-1][1]: |
| out[-1] = (out[-1][0], max(out[-1][1], end)) |
| else: |
| out.append((start, end)) |
| return out |
|
|
|
|
| def _validate_inputs( |
| words: Sequence[TimedWord], |
| duration_ms: int, |
| events: Sequence[EventDetection], |
| styles: Sequence[WordStyleScore], |
| ) -> None: |
| if type(duration_ms) is not int or duration_ms <= 0: |
| raise ValueError(f"duration_ms must be a positive int, got {duration_ms!r}") |
| previous_end = 0 |
| for word in words: |
| if word.start_ms < previous_end: |
| raise ValueError(f"words overlap or are unsorted at {word.text!r} ({word.start_ms} ms)") |
| if word.end_ms > duration_ms: |
| raise ValueError(f"word {word.text!r} ends after the segment ({word.end_ms} ms)") |
| previous_end = word.end_ms |
| for event in events: |
| if event.kind not in EVENTS or event.kind in _PAUSE_EVENTS: |
| raise ValueError(f"not a detectable event kind: {event.kind!r}") |
| if event.end_ms > duration_ms: |
| raise ValueError(f"event {event.kind!r} ends after the segment") |
| seen: set[str] = set() |
| for style in styles: |
| if style.style not in SPANS: |
| raise ValueError(f"unknown style {style.style!r}") |
| if style.style in seen: |
| raise ValueError(f"duplicate style scores for {style.style!r}") |
| seen.add(style.style) |
| if len(style.scores) != len(words): |
| raise ValueError( |
| f"style {style.style!r} has {len(style.scores)} scores for {len(words)} words" |
| ) |
|
|
|
|
| def _split_by_band( |
| events: Sequence[EventDetection], cfg: FusionConfig |
| ) -> tuple[list[EventDetection], list[EventDetection], list[EventDetection]]: |
| """(confident, middle band, ignored = below ``low``).""" |
| confident: list[EventDetection] = [] |
| middle: list[EventDetection] = [] |
| ignored: list[EventDetection] = [] |
| for event in events: |
| band = cfg.bands[event.kind] |
| if event.conf < band.low: |
| ignored.append(event) |
| elif event.conf < band.high: |
| middle.append(event) |
| else: |
| confident.append(event) |
| return confident, middle, ignored |
|
|
|
|
| def _merge_confident( |
| confident: Sequence[EventDetection], middle: Sequence[EventDetection], cfg: FusionConfig |
| ) -> list[_Merged]: |
| """Rules 1–2: ambiguity check against confident cover, then per-family merging.""" |
| cover: dict[str, list[tuple[int, int]]] = {} |
| for event in confident: |
| cover.setdefault(_family(event.kind), []).append((event.start_ms, event.end_ms)) |
| union = {family: _union(spans) for family, spans in cover.items()} |
| for event in sorted(middle, key=lambda e: (e.start_ms, e.end_ms, e.kind, e.source, e.conf)): |
| if event.kind in cfg.disabled_tags: |
| raise _Drop("disabled_tag", tag=event.kind, start_ms=event.start_ms, conf=event.conf) |
| covered = sum( |
| _overlap_ms(event.start_ms, event.end_ms, a, b) |
| for a, b in union.get(_family(event.kind), []) |
| ) |
| if covered < event.duration_ms: |
| raise _Drop( |
| "ambiguous_event", |
| kind=event.kind, |
| conf=event.conf, |
| start_ms=event.start_ms, |
| end_ms=event.end_ms, |
| source=event.source, |
| ) |
| for event in sorted(confident, key=lambda e: (e.start_ms, e.end_ms, e.kind, e.source)): |
| if event.kind in cfg.disabled_tags: |
| |
| |
| raise _Drop("disabled_tag", tag=event.kind, start_ms=event.start_ms, conf=event.conf) |
| merged: list[_Merged] = [] |
| ordered = sorted( |
| confident, key=lambda e: (_family(e.kind), e.start_ms, e.end_ms, e.kind, e.source) |
| ) |
| for event in ordered: |
| family = _family(event.kind) |
| last = merged[-1] if merged else None |
| if ( |
| last is not None |
| and last.family == family |
| and event.start_ms <= last.end_ms + cfg.merge_gap_ms |
| ): |
| last.end_ms = max(last.end_ms, event.end_ms) |
| last.conf = max(last.conf, event.conf) |
| last.kinds.add(event.kind) |
| last.sources.add(event.source) |
| else: |
| merged.append( |
| _Merged( |
| family, {event.kind}, event.start_ms, event.end_ms, event.conf, {event.source} |
| ) |
| ) |
| return sorted(merged, key=lambda m: (m.start_ms, m.end_ms, m.family)) |
|
|
|
|
| def _non_speech_pieces( |
| start_ms: int, end_ms: int, words: Sequence[TimedWord] |
| ) -> list[tuple[int, int]]: |
| """Parts of ``[start_ms, end_ms)`` not covered by any word (words are sorted, disjoint).""" |
| pieces: list[tuple[int, int]] = [] |
| cursor = start_ms |
| for word in words: |
| if word.end_ms <= cursor: |
| continue |
| if word.start_ms >= end_ms: |
| break |
| if word.start_ms > cursor: |
| pieces.append((cursor, word.start_ms)) |
| cursor = max(cursor, word.end_ms) |
| if cursor < end_ms: |
| pieces.append((cursor, end_ms)) |
| return pieces |
|
|
|
|
| def _nearest_gap(time_ms: float, words: Sequence[TimedWord]) -> int: |
| """Gap ``g`` lies before word ``g`` (``g == len(words)``: after the last word).""" |
| for index, word in enumerate(words): |
| if time_ms <= word.start_ms: |
| return index |
| if time_ms < word.end_ms: |
| |
| |
| return index if time_ms - word.start_ms < word.end_ms - time_ms else index + 1 |
| return len(words) |
|
|
|
|
| def _event_kind(event: _Merged, placed_ms: int, cfg: FusionConfig) -> str: |
| if event.family != "laugh": |
| return event.family |
| if "laugh_long" in event.kinds or placed_ms >= cfg.laugh_long_min_ms: |
| return "laugh_long" |
| return "laugh" |
|
|
|
|
| def _place_events( |
| merged: Sequence[_Merged], words: Sequence[TimedWord], cfg: FusionConfig |
| ) -> tuple[list[_Placed], list[tuple[str, list[int], _Merged]]]: |
| """Rules 3–4. Returns (placed events, style evidence ``(style, word indices, event)``).""" |
| placed: list[_Placed] = [] |
| style_events: list[tuple[str, list[int], _Merged]] = [] |
| for event in merged: |
| length = event.end_ms - event.start_ms |
| style_id = STYLE_FROM_EVENT.get(event.family) |
| styled_words = [ |
| i |
| for i, w in enumerate(words) |
| if style_id is not None |
| and _overlap_ms(event.start_ms, event.end_ms, w.start_ms, w.end_ms) |
| > cfg.style_word_overlap * w.duration_ms |
| ] |
| pieces = _non_speech_pieces(event.start_ms, event.end_ms, words) |
| longest = max(pieces, key=lambda p: (p[1] - p[0], -p[0])) if pieces else None |
| longest_ms = longest[1] - longest[0] if longest is not None else 0 |
| if styled_words: |
| assert style_id is not None |
| |
| |
| |
| unmarked_ms = sum( |
| _overlap_ms(event.start_ms, event.end_ms, w.start_ms, w.end_ms) |
| for i, w in enumerate(words) |
| if i not in styled_words |
| ) |
| if unmarked_ms > cfg.max_event_speech_overlap * length: |
| raise _Drop( |
| "event_overlaps_speech", |
| kind=sorted(event.kinds), |
| start_ms=event.start_ms, |
| end_ms=event.end_ms, |
| speech_ms=unmarked_ms, |
| ) |
| style_events.append((style_id, styled_words, event)) |
| if longest_ms < cfg.min_residual_event_ms: |
| continue |
| else: |
| speech_ms = length - sum(b - a for a, b in pieces) |
| if longest is None or speech_ms > cfg.max_event_speech_overlap * length: |
| raise _Drop( |
| "event_overlaps_speech", |
| kind=sorted(event.kinds), |
| start_ms=event.start_ms, |
| end_ms=event.end_ms, |
| speech_ms=speech_ms, |
| ) |
| assert longest is not None |
| kind = _event_kind(event, longest_ms, cfg) |
| if kind in cfg.disabled_tags: |
| raise _Drop("disabled_tag", tag=kind, start_ms=event.start_ms, conf=event.conf) |
| placed.append( |
| _Placed(_nearest_gap((longest[0] + longest[1]) / 2, words), kind, *longest, event) |
| ) |
| for i, first in enumerate(placed): |
| for second in placed[i + 1 :]: |
| if first.kind == second.kind: |
| continue |
| shorter = min(first.end_ms - first.start_ms, second.end_ms - second.start_ms) |
| shared = _overlap_ms(first.start_ms, first.end_ms, second.start_ms, second.end_ms) |
| if shared > cfg.max_cross_kind_overlap * shorter: |
| raise _Drop( |
| "conflicting_events", |
| kinds=sorted({first.kind, second.kind}), |
| start_ms=max(first.start_ms, second.start_ms), |
| shared_ms=shared, |
| ) |
| return placed, style_events |
|
|
|
|
| def _style_marks( |
| words: Sequence[TimedWord], |
| styles: Sequence[WordStyleScore], |
| style_events: Sequence[tuple[str, list[int], _Merged]], |
| cfg: FusionConfig, |
| ) -> tuple[list[str | None], dict[int, tuple[float, str, int, int]]]: |
| """Per-word style (or None) and, per marked word, its evidence |
| ``(conf, source, det_start_ms, det_end_ms)`` (the word itself for acoustic scores).""" |
| marks: list[set[str]] = [set() for _ in words] |
| evidence: list[dict[str, tuple[float, str, int, int]]] = [{} for _ in words] |
| for style in sorted(styles, key=lambda s: s.style): |
| if style.style in cfg.ignored_styles: |
| continue |
| band = cfg.bands[style.style] |
| for index, score in enumerate(style.scores): |
| if score is None or score < band.low: |
| continue |
| if score < band.high: |
| raise _Drop( |
| "ambiguous_style", |
| style=style.style, |
| word_index=index, |
| word=words[index].text, |
| score=score, |
| source=style.source, |
| ) |
| if style.style in cfg.disabled_tags: |
| raise _Drop("disabled_tag", tag=style.style, word_index=index, score=score) |
| marks[index].add(style.style) |
| word = words[index] |
| evidence[index][style.style] = (score, style.source, word.start_ms, word.end_ms) |
| for style_id, indices, event in style_events: |
| if style_id in cfg.disabled_tags: |
| raise _Drop("disabled_tag", tag=style_id, start_ms=event.start_ms, conf=event.conf) |
| for index in indices: |
| marks[index].add(style_id) |
| previous = evidence[index].get(style_id) |
| if previous is None or event.conf > previous[0]: |
| evidence[index][style_id] = ( |
| event.conf, |
| event.source, |
| event.start_ms, |
| event.end_ms, |
| ) |
| resolved: list[str | None] = [] |
| for index, word_marks in enumerate(marks): |
| if len(word_marks) > 1: |
| raise _Drop( |
| "conflicting_styles", |
| word_index=index, |
| word=words[index].text, |
| styles=sorted(word_marks), |
| ) |
| resolved.append(next(iter(word_marks)) if word_marks else None) |
| chosen = { |
| index: evidence[index][style_id] |
| for index, style_id in enumerate(resolved) |
| if style_id is not None |
| } |
| return resolved, chosen |
|
|
|
|
| def _style_runs( |
| words: Sequence[TimedWord], marks: Sequence[str | None], cfg: FusionConfig |
| ) -> list[tuple[str, int, int]]: |
| """Rule 5: maximal runs ``(style, first, last)`` of equally marked consecutive words.""" |
| runs: list[tuple[str, int, int]] = [] |
| index = 0 |
| while index < len(marks): |
| style_id = marks[index] |
| if style_id is None: |
| index += 1 |
| continue |
| last = index |
| while last + 1 < len(marks) and marks[last + 1] == style_id: |
| last += 1 |
| count = last - index + 1 |
| span_ms = words[last].end_ms - words[index].start_ms |
| if count < cfg.min_span_words and span_ms < cfg.min_span_ms: |
| raise _Drop( |
| "short_style_run", |
| style=style_id, |
| words=[w.text for w in words[index : last + 1]], |
| span_ms=span_ms, |
| ) |
| runs.append((style_id, index, last)) |
| index = last + 1 |
| return runs |
|
|
|
|
| def _normalise_word(text: str) -> str: |
| """Rule 7 for one word.""" |
| text = nfc(text) |
| if any(ch in _RESERVED_CHARS for ch in text): |
| raise _Drop("reserved_char_in_text", word=text) |
| if any(unicodedata.category(ch).startswith("N") for ch in text): |
| raise _Drop("unnormalised_number", word=text) |
| start, end = 0, len(text) |
| while start < end and unicodedata.category(text[start]).startswith("P"): |
| start += 1 |
| while end > start and unicodedata.category(text[end - 1]).startswith("P"): |
| end -= 1 |
| core = text[start:end] |
| if not core: |
| return text |
| canonical = canonical_filler(tr_lower(core)) |
| return text if canonical is None else text[:start] + canonical + text[end:] |
|
|
|
|
| def normalise_word(text: str) -> str: |
| """Rule 7 for one word, as :func:`fuse_segment` applies it to every aligned word. |
| |
| Lets later stages (``timings``) compare the words of a stored tagged text with the |
| aligned words they came from. |
| |
| Raises: |
| ValueError: if fusion would drop the segment for this word (reserved tag character |
| or a digit-like character), with fusion's reason in the message. |
| """ |
| try: |
| return _normalise_word(text) |
| except _Drop as drop: |
| raise ValueError(f"{drop.reason}: {drop.details}") from drop |
|
|
|
|
| def _ends_with_punctuation(text: str) -> bool: |
| return unicodedata.category(text[-1]).startswith("P") |
|
|
|
|
| def _pauses( |
| words: Sequence[TimedWord], |
| texts: Sequence[str], |
| merged: Sequence[_Merged], |
| cfg: FusionConfig, |
| ) -> list[_Placed]: |
| """Rule 6, on the longest part of each gap that no confident detection covers. |
| |
| A short breath inside a 5 s silence must not hide the pause; the pause tag record |
| spans that uncovered part, and tags in one gap are later ordered by time. |
| """ |
| out: list[_Placed] = [] |
| for gap in range(1, len(words)): |
| gap_start, gap_end = words[gap - 1].end_ms, words[gap].start_ms |
| covered = _union( |
| [ |
| (max(gap_start, e.start_ms), min(gap_end, e.end_ms)) |
| for e in merged |
| if _overlap_ms(gap_start, gap_end, e.start_ms, e.end_ms) > 0 |
| ] |
| ) |
| free: list[tuple[int, int]] = [] |
| cursor = gap_start |
| for a, b in covered: |
| if a > cursor: |
| free.append((cursor, a)) |
| cursor = max(cursor, b) |
| if cursor < gap_end: |
| free.append((cursor, gap_end)) |
| if not free: |
| continue |
| start, end = max(free, key=lambda p: (p[1] - p[0], -p[0])) |
| silence = end - start |
| kind: str | None = None |
| if silence > cfg.pause_long_ms: |
| kind = "pause_long" |
| elif silence >= cfg.pause_short_ms and not _ends_with_punctuation(texts[gap - 1]): |
| kind = "pause_short" |
| if kind is not None: |
| if kind in cfg.disabled_tags: |
| raise _Drop("disabled_tag", tag=kind, start_ms=start) |
| silence_event = _Merged(kind, {kind}, start, end, 1.0, {"alignment"}) |
| out.append(_Placed(gap, kind, start, end, silence_event)) |
| return out |
|
|
|
|
| def _build_items( |
| texts: Sequence[str], |
| events_by_gap: Mapping[int, Sequence[_Placed]], |
| runs: Sequence[tuple[str, int, int]], |
| ) -> tuple[Item, ...]: |
| run_at = {first: (style_id, last) for style_id, first, last in runs} |
| items: list[Item] = [] |
| index = 0 |
| while True: |
| items.extend(Event(p.kind) for p in events_by_gap.get(index, ())) |
| if index == len(texts): |
| break |
| if index in run_at: |
| style_id, last = run_at[index] |
| children: list[Word | Event] = [] |
| for inner in range(index, last + 1): |
| if inner > index: |
| children.extend(Event(p.kind) for p in events_by_gap.get(inner, ())) |
| children.append(Word(texts[inner])) |
| items.append(Span(style_id, tuple(children))) |
| index = last + 1 |
| else: |
| items.append(Word(texts[index])) |
| index += 1 |
| return tuple(items) |
|
|
|
|
| def fuse_segment( |
| segment_id: str, |
| words: Sequence[TimedWord], |
| duration_ms: int, |
| events: Sequence[EventDetection], |
| styles: Sequence[WordStyleScore], |
| cfg: FusionConfig, |
| ) -> FusionResult: |
| """Fuse one segment's annotations into tagged text (see module docstring). |
| |
| Args: |
| segment_id: Content-addressed id (:func:`nida.data.types.segment_id`). |
| words: Aligned words, sorted and non-overlapping, segment-relative ms. |
| duration_ms: Segment length in ms. |
| events: Raw event detections (any confidence; bands are applied here). |
| styles: At most one per-word score track per span style. |
| cfg: Thresholds. |
| |
| Returns: |
| A :class:`FusionResult`; when ``quarantine`` is set, ``items``/``text`` are None. |
| ``ignored_detections`` counts below-band detections in both cases and |
| ``ignored_events`` holds them. |
| |
| Raises: |
| ValueError: on inconsistent inputs (a caller bug, not a data property). |
| """ |
| _validate_inputs(words, duration_ms, events, styles) |
| confident, middle, ignored = _split_by_band(events, cfg) |
| try: |
| if not words: |
| raise _Drop("no_words") |
| texts = [_normalise_word(w.text) for w in words] |
| merged = _merge_confident(confident, middle, cfg) |
| placed, style_events = _place_events(merged, words, cfg) |
| marks, evidence = _style_marks(words, styles, style_events, cfg) |
| runs = _style_runs(words, marks, cfg) |
| placed += _pauses(words, texts, merged, cfg) |
|
|
| events_by_gap: dict[int, list[_Placed]] = {} |
| tags: list[TagRecord] = [] |
| for p in sorted(placed, key=lambda p: (p.start_ms, p.end_ms, p.kind)): |
| events_by_gap.setdefault(p.gap, []).append(p) |
| tags.append( |
| TagRecord( |
| p.kind, |
| p.start_ms, |
| p.end_ms, |
| p.event.conf, |
| p.event.source, |
| p.event.start_ms, |
| p.event.end_ms, |
| ) |
| ) |
| for style_id, first, last in runs: |
| run_evidence = [evidence[i] for i in range(first, last + 1)] |
| tags.append( |
| TagRecord( |
| style_id, |
| words[first].start_ms, |
| words[last].end_ms, |
| min(e[0] for e in run_evidence), |
| "+".join(sorted({e[1] for e in run_evidence})), |
| min(e[2] for e in run_evidence), |
| max(e[3] for e in run_evidence), |
| ) |
| ) |
| try: |
| items = _build_items(texts, events_by_gap, runs) |
| text = render(items) |
| if parse(text) != items: |
| raise _Drop("invalid_output", text=text, detail="round-trip mismatch") |
| except (TagSyntaxError, ValueError) as err: |
| raise _Drop("invalid_output", error=str(err)) from err |
| except _Drop as drop: |
| return FusionResult( |
| segment_id=segment_id, |
| items=None, |
| text=None, |
| tags=(), |
| quarantine=QuarantineRecord(segment_id, STAGE, drop.reason, dict(drop.details)), |
| ignored_detections=len(ignored), |
| ignored_events=tuple(ignored), |
| ) |
| return FusionResult( |
| segment_id=segment_id, |
| items=items, |
| text=text, |
| tags=tuple(sorted(tags, key=lambda t: (t.start_ms, t.end_ms, t.tag))), |
| quarantine=None, |
| ignored_detections=len(ignored), |
| ignored_events=tuple(ignored), |
| ) |
|
|
|
|
| def emotion_condition(scores: Mapping[str, float], cfg: FusionConfig) -> EmotionCondition: |
| """Utterance-level emotion condition from emotion-model class probabilities. |
| |
| The top class becomes the condition with its probability as intensity. ``other`` / |
| ``unknown`` as top class, or a top probability below ``emotion_min_conf``, give a |
| null condition (not "neutral"; tag-spec §5, ADR-0003). Ties resolve to the |
| alphabetically first label so the result is deterministic. |
| |
| Raises: |
| ValueError: unknown labels, empty scores, or probabilities outside [0, 1]. |
| """ |
| if not scores: |
| raise ValueError("empty emotion scores") |
| unknown = set(scores) - EMOTION_LABELS |
| if unknown: |
| raise ValueError(f"unknown emotion labels {sorted(unknown)}") |
| for label, prob in scores.items(): |
| if not (math.isfinite(prob) and 0.0 <= prob <= 1.0): |
| raise ValueError(f"emotion {label!r} probability {prob!r} not in [0, 1]") |
| label, prob = min(scores.items(), key=lambda kv: (-kv[1], kv[0])) |
| if label in _NULL_EMOTIONS or prob < cfg.emotion_min_conf: |
| return EmotionCondition(None, None) |
| return EmotionCondition(label, prob) |
|
|
