"""Phase 1 pilot runner behind ``nida data-pilot`` (YODAS, ~10 h, end to end on Colab).
Stages (each idempotent and resumable; ``all`` runs them in order in one session):
check → ingest → sed → diar → asr → ctc → emotion → annotate → report → enhance →
latents → dataset → panel
* ``ingest`` (CPU): YODAS listing → deterministic selection up to ``max_hours`` →
streaming extraction of the selected wavs into the local cache → S1/S2 checks
(:mod:`nida.data.ingest`) → ``work/ingest`` manifest + quarantine.
* ``sed`` / ``diar`` / ``ctc`` / ``emotion`` / ``asr`` (GPU): model workers in isolated
environments over the ingested segments. Array outputs live in the *local* cache; a
result whose array is gone (new Colab session) is dropped and recomputed.
* ``annotate`` (CPU): S2 music, S3 speakers, S4 dual ASR, S5 alignment, S6 events/styles/
emotion, S7 fusion → ``work/annotate-<config hash>`` manifest + quarantine. Changing any
threshold file gives a new directory, so tuning never overwrites earlier outputs.
* ``report``: S8 statistics, S10 video-disjoint splits, ``reports/pilot.md``.
* ``enhance`` (GPU, optional): speech enhancement of the accepted segments by the arm in
``configs/data/enhance.yaml`` (``uvr`` | ``sidon`` | ``dfn``, :mod:`nida.data.enhance`);
a no-op returning ``{"arm": "none"}`` for ``none`` (the baseline). Reads the original wav,
writes a 16 kHz mono PCM16 wav of the *same number of samples* to
``<cache>/audio-enh-<arm>/<same relpath>`` and records, per segment, the pre-fix length
difference, the peak and the loudness of the written samples. The worker only refuses absurd
length differences (``max_length_diff_ms``); the pilot quarantines by ``length_tolerance_ms``
and quarantines worker errors, into ``work/enhance-<arm>-<annotate hash>/quarantine.jsonl``,
excluding them from ``latents``, ``dataset``, ``timings`` and ``panel``. Only segments a later
stage still reads are re-enhanced in a new session (the wavs are local-cache only), and nothing
at all when none is needed. Enhancement changes neither labels nor splits, so ``annotate`` and
``report`` (and their hash) are shared with the baseline and always see the original audio
(tags were decided on it); ``report`` gains an enhance section.
* ``latents`` (GPU, S9): VAE latents of the accepted segments (frozen AudioVAE V2). With an
enhancement arm the *enhanced* wavs are encoded, into their own results folder and array
cache (``voxcpm2_audiovae_v2-enh-<arm>``), so baseline and enhanced latents never mix.
* ``dataset`` (CPU): the Phase 2 training dataset ``dataset-<config hash>/``: float16 latent
shards, ``index.jsonl`` (text, emotion, intensity, rate/loudness buckets, group, split)
and ``meta.json`` (written last; its presence marks the stage done). With an enhancement
arm the directory is ``dataset-<config hash>-enh-<arm>/``, the loudness buckets come from the
enhanced audio, and ``meta.json`` gains an ``enhance`` block (arm, model id, revision, limits,
excluded segments and their digest, clipping summary, ``loudness_source``, worker packages and
runtime); without an arm nothing about the dataset differs from the pre-enhancement pipeline.
* ``panel``: per-tag precision listening panel (:mod:`nida.data.tag_panel`); with an enhancement
arm it plays the enhanced audio and its id carries ``-enh-<arm>``.
* ``timings`` (extra stage, GPU: CTC only): the unit timeline of every dataset segment for the
aligned text path (E11, ADR-0008) -> ``dataset-<hash>/timings.jsonl``
(:mod:`nida.data.timings`). For an enhanced dataset the CTC runs on the enhanced audio.
``all`` normally runs the stages one after another. With ``parallel_workers > 1`` the
independent worker stages (``sed``, ``diar``, both ``asr`` systems, ``ctc``, ``emotion``)
run concurrently as separate subprocesses (see :meth:`Pilot._run_worker_group`).
Durable outputs (manifests, worker results, reports, panels) are on Drive under
``root``; audio and arrays are in ``cache`` (local disk) and are rebuilt when missing.
"""
from __future__ import annotations
import functools
import hashlib
import io
import json
import logging
import math
import shutil
from collections import Counter, defaultdict
from collections.abc import Callable, Iterable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict, dataclass, fields
from itertools import pairwise
from pathlib import Path
from typing import TYPE_CHECKING, Any
from nida.bench import envs as bench_envs
from nida.bench import fetch
from nida.bench.b1 import LOG_TAG, MAX_ERROR_FRACTION, _stream_runner
from nida.bench.clips import YodasUtt, yodas_video_id
from nida.bench.prepare import parse_yodas_listing
from nida.bench.protocol import (
WorkRequest,
WorkResult,
drop_results,
load_results,
read_summary,
write_requests,
)
from nida.bench.registry import ASR_SYSTEMS, DATA_PINS, System, load_sources, require_approved
from nida.bench.textmetrics import normalize_for_scoring
from nida.data import acoustic, enhance, envs, ingest, manifest, stats, tag_panel
from nida.data.align import align_words
from nida.data.asr_filter import AsrFilterConfig, check_alignment, check_transcripts
from nida.data.enhance import NO_ENHANCE, EnhanceConfig
from nida.data.events import EventConfig, detect_events
from nida.data.fillers_ctc import (
FillerCandidate,
FillerConfig,
gap_filler_candidates,
insert_fillers,
)
from nida.data.fusion import FusionConfig, emotion_condition, fuse_segment
from nida.data.quarantine import QuarantineLog
from nida.data.registry import DATA_MODELS, DataModel, enhance_model
from nida.data.speakers import SpeakerConfig, check_speakers
from nida.data.splits import assign_splits
from nida.data.timings import (
TIMINGS_FILENAME,
TimelineResult,
build_timings,
load_timings,
read_index,
write_timings,
)
from nida.data.types import QuarantineRecord, TimedWord, WordStyleScore
from nida.text.fillers import canonical_filler
from nida.text.numbers import verbalize_digits
from nida.text.turkish import nfc
from nida.utils.io import atomic_write_json, atomic_write_text
if TYPE_CHECKING:
import pyarrow as pa # type: ignore[import-untyped] # no stubs
__all__ = [
"ALL_STAGES",
"EXTRA_STAGES",
"Pilot",
"PilotPlan",
"QualityConfig",
"WorkerJob",
"enhanced_audio_root_for",
"load_enhance_config",
"run_worker_job",
]
log = logging.getLogger(__name__)
#: Stages outside ``all`` (measurements that never change the dataset).
EXTRA_STAGES = ("fillers", "fillers_whisper", "timings")
#: Most filler candidates put on the listening panel.
FILLER_PANEL_ITEMS = 40
#: ``timings``: suffix of the CTC re-run's results folder / array cache. Distinct from the plain
#: ``mpoyraz_ctc_tr`` results, which cover every ingested segment and which ``_run_worker``
#: would prune down to the dataset's segments.
TIMINGS_CTC_SUFFIX = "-timings"
#: ``fillers_whisper``: accepted segments sent to the prompted Whisper run (seeded rank order).
FILLER_WHISPER_SAMPLE = 300
#: ``fillers_whisper``: ASR system that is prompted (approved in configs/sources.yaml).
FILLER_WHISPER_SYSTEM = "whisper_large_v3"
#: ``fillers_whisper``: results folder; distinct from the plain ``asr-whisper_large_v3`` so
#: prompted and plain transcripts can never mix.
FILLER_WHISPER_WORKER = f"asr-{FILLER_WHISPER_SYSTEM}-fillerprompt"
#: Whisper decoder prompt for the ``fillers_whisper`` experiment. Whisper imitates the style of
#: its prompt, so a prompt that *contains* Turkish hesitation sounds (spelled the way the
#: canonical set spells them, with elongations) makes it more willing to write such sounds
#: instead of dropping them (docs/research/R10-dolgu-farkindalikli-asr.md). Changing this
#: string requires a new results folder (the stage refuses to reuse results of another prompt).
FILLER_PROMPT = "Iıı, şey, eee... hmm, yani, ımm, aaa."
#: A Whisper transcript holding this many consecutive prompt tokens is a prompt echo
#: (hallucinated copy of the prompt), not evidence of spoken fillers.
PROMPT_ECHO_MIN_TOKENS = 4
#: ``all`` with ``parallel_workers > 1``: the worker stages that depend only on the ingest
#: manifest and the extracted audio (not on each other), in their sequential order.
PARALLEL_GROUP = ("sed", "diar", "asr", "ctc", "emotion")
#: ASR roles, in run order (primary first).
ASR_ROLES = ("primary", "secondary")
ALL_STAGES = (
"check",
"ingest",
"sed",
"diar",
"asr",
"ctc",
"emotion",
"annotate",
"report",
"enhance",
"latents",
"dataset",
"panel",
)
#: Threshold files whose bytes (plus the model pins) name the ``annotate-<digest>`` directory.
#: ``enhance.yaml`` is deliberately not one of them (:func:`load_enhance_config`).
CONFIG_FILES = (
"acoustic.yaml",
"asr_filter.yaml",
"dataset.yaml",
"events.yaml",
"fillers.yaml",
"fusion.yaml",
"ingest.yaml",
"quality.yaml",
"speakers.yaml",
)
EXTRA_SED_CLASSES = ("Music", "Speech")
INGEST_CHUNK = 1000 # rows per manifest part / quarantine flush
#: tr000 shards reserved for the B1 layer-C pool (nida.bench.prepare). tr100 has no such
#: reservation: the benchmark never reads it.
BENCH_SHARDS = frozenset({"00000000", "00000001"})
#: YODAS Turkish subsets the pilot reads (``data/<subset>/{audio,text,duration}/<shard>``):
#: ``tr000`` has manual subtitles, ``tr100`` automatic ones. Both use the same file layout.
YODAS_SUBSETS = ("tr000", "tr100")
#: The subset of every plan that does not name one (and of ``pilot_plan.json`` files written
#: before the ``subset`` key existed).
DEFAULT_SUBSET = "tr000"
#: Number of tr000 shards: ``data/tr000/text/00000000.txt`` .. ``00000063.txt``. Counted from
#: the espnet/yodas Hugging Face tree on 2026-10-12 (64 shards of each of audio/text/duration);
#: ``00000064.txt`` answers 404 at the pinned revision. The de-duplication against tr000
#: downloads all of them, so a wrong count is not tolerated silently: a missing listing raises.
TR000_NUM_SHARDS = 64
TR000_SHARDS = tuple(f"{index:08d}" for index in range(TR000_NUM_SHARDS))
#: Counts key of the utterances dropped because their video is in tr000 (other subsets only).
DUPLICATE_VIDEO_KEY = "duplicate_video_in_tr000"
FLOAT16_MAX_ABS = 6.0e4 # latents beyond this would overflow the float16 dataset shards
DATASET_SHARD_FRAMES = 200_000 # ~2.2 h of 25 Hz latents per shard
DURATION_BIN_EDGES_MS = (1000, 2000, 4000, 6000, 8000, 10000, 15000, 20000)
LANGUAGE = "tr" # forced language of ASR systems run without language identification
#: ``enhance`` stage: the marker (written last) in ``work/enhance-<arm>-<annotate hash>/`` that
#: says the stage finished for exactly this accepted set; ``latents``/``dataset``/``timings``
#: of an enhanced dataset refuse to run without it.
ENHANCE_MARKER = "enhance.json"
#: ``enhance`` stage: quarantine ``stage`` field.
ENHANCE_STAGE = "enhance"
#: ``meta.json`` ``enhance.loudness_source``: the dataset's loudness buckets are measured on the
#: enhanced audio (the worker's ``loudness_dbfs``), not on the ingest audio.
ENHANCED_LOUDNESS_SOURCE = "enhanced"
#: Sentinels of the enhance section in ``reports/pilot.md``.
ENHANCE_REPORT_BEGIN = "<!-- enhance:begin -->"
ENHANCE_REPORT_END = "<!-- enhance:end -->"
@dataclass(frozen=True)
class PilotPlan:
"""Everything that determines *which* segments the pilot processes."""
seed: int = 20261008
shards: str = "00000002"
max_hours: float = 10.0
asr_primary: str = "omniasr_llm_7b"
asr_secondary: str = "qwen3_asr_1_7b" # different model family than the primary
yodas_revision: str = DATA_PINS["yodas_revision"]
panel_per_tag: int = 15
#: YODAS subset (``tr000`` or ``tr100``). Recorded in ``pilot_plan.json``; a plan file
#: without the key was written for ``tr000``.
subset: str = DEFAULT_SUBSET
def __post_init__(self) -> None:
if self.subset not in YODAS_SUBSETS:
raise ValueError(f"subset must be one of {YODAS_SUBSETS}, got {self.subset!r}")
self.shard_list()
if not (math.isfinite(self.max_hours) and self.max_hours > 0):
raise ValueError(f"max_hours must be a positive finite number, got {self.max_hours!r}")
if self.panel_per_tag < 1:
raise ValueError("panel_per_tag must be >= 1")
reserved = set(self.shard_list()) & BENCH_SHARDS if self.subset == "tr000" else set()
if reserved:
raise ValueError(f"shards {sorted(reserved)} hold the B1 layer-C pool (faz-1.md)")
def shard_list(self) -> list[str]:
shards = [s.strip() for s in self.shards.split(",")]
if not shards or any(len(s) != 8 or not s.isdigit() for s in shards):
raise ValueError(f"invalid shards {self.shards!r}")
if len(set(shards)) != len(shards):
raise ValueError(f"repeated shard in {self.shards!r}")
return shards
@dataclass(frozen=True)
class QualityConfig:
"""configs/data/quality.yaml."""
music_prob: float
max_music_ratio: float
def __post_init__(self) -> None:
for name in ("music_prob", "max_music_ratio"):
value = getattr(self, name)
if isinstance(value, bool) or not isinstance(value, int | float):
raise ValueError(f"{name} must be a number")
if not (math.isfinite(value) and 0.0 <= value <= 1.0):
raise ValueError(f"{name} must be in [0, 1], got {value!r}")
@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> QualityConfig:
expected = {f.name for f in fields(cls)}
if set(raw) != expected:
raise ValueError(f"quality config keys must be exactly {sorted(expected)}")
return cls(**raw) # type: ignore[arg-type] # validated in __post_init__
@dataclass(frozen=True)
class DatasetConfig:
"""``configs/data/dataset.yaml``: fixed condition-bucket edges (shared by all roots)."""
rate_bucket_edges_chars_per_s: tuple[float, ...]
loudness_bucket_edges_dbfs: tuple[float, ...]
@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> DatasetConfig:
from nida.train.data import N_LOUDNESS_BUCKETS, N_RATE_BUCKETS
expected = {"rate_bucket_edges_chars_per_s", "loudness_bucket_edges_dbfs"}
if set(raw) != expected:
raise ValueError(
f"dataset config keys: missing={sorted(expected - set(raw))} "
f"unknown={sorted(set(raw) - expected)}"
)
edges: dict[str, tuple[float, ...]] = {}
for key, n_buckets in (
("rate_bucket_edges_chars_per_s", N_RATE_BUCKETS),
("loudness_bucket_edges_dbfs", N_LOUDNESS_BUCKETS),
):
value = raw[key]
if not isinstance(value, list) or not all(
isinstance(x, int | float) and not isinstance(x, bool) and math.isfinite(x)
for x in value
):
raise ValueError(f"{key} must be a list of finite numbers")
if len(value) != n_buckets - 1 or any(b <= a for a, b in pairwise(value)):
raise ValueError(f"{key} must be {n_buckets - 1} strictly increasing edges")
edges[key] = tuple(float(x) for x in value)
return cls(edges["rate_bucket_edges_chars_per_s"], edges["loudness_bucket_edges_dbfs"])
@dataclass(frozen=True)
class _Configs:
ingest: ingest.IngestConfig
acoustic: acoustic.AcousticConfig
asr: AsrFilterConfig
dataset: DatasetConfig
events: EventConfig
fillers: FillerConfig
fusion: FusionConfig
quality: QualityConfig
speakers: SpeakerConfig
digest: str
#: Not part of ``digest`` (enhancement changes neither labels nor splits).
enhance: EnhanceConfig = NO_ENHANCE
def _load_yaml(path: Path) -> Mapping[str, object]:
import yaml # type: ignore[import-untyped] # PyYAML: bench extra, no stubs installed
obj = yaml.safe_load(path.read_text(encoding="utf-8"))
if not isinstance(obj, dict):
raise ValueError(f"{path}: expected a mapping")
return obj
def load_enhance_config(config_dir: Path) -> EnhanceConfig:
"""``enhance.yaml`` of ``config_dir``; a directory without the file means no enhancement.
The file is read separately from :data:`CONFIG_FILES` and never enters the annotate digest:
the arm must not rename ``annotate-<digest>`` (labels and splits do not depend on it).
Raises:
ValueError: if the file exists but is not a valid enhance config (the message names it).
"""
path = config_dir / enhance.CONFIG_FILENAME
if not path.exists():
return NO_ENHANCE
try:
return EnhanceConfig.from_mapping(_load_yaml(path))
except ValueError as err:
raise ValueError(f"{path}: {err}") from err
def load_configs(config_dir: Path) -> _Configs:
"""Load and cross-check every threshold file; ``digest`` names the annotate output."""
digest = hashlib.sha1()
for name in CONFIG_FILES:
digest.update(name.encode("utf-8") + b"\x00" + (config_dir / name).read_bytes())
# Model identity is part of what the outputs depend on (a new pin = new outputs).
refs = {k: [m.ref.model_id, m.ref.revision] for k, m in sorted(DATA_MODELS.items())}
digest.update(json.dumps(refs, sort_keys=True).encode("utf-8"))
fusion_cfg = FusionConfig.from_mapping(_load_yaml(config_dir / "fusion.yaml"))
event_cfg = EventConfig.from_mapping(_load_yaml(config_dir / "events.yaml"))
event_cfg.check_onset_against({k: b.low for k, b in fusion_cfg.bands.items()})
return _Configs(
ingest=ingest.IngestConfig.from_mapping(_load_yaml(config_dir / "ingest.yaml")),
acoustic=acoustic.load_acoustic_config(config_dir / "acoustic.yaml"),
asr=AsrFilterConfig.from_mapping(_load_yaml(config_dir / "asr_filter.yaml")),
dataset=DatasetConfig.from_mapping(_load_yaml(config_dir / "dataset.yaml")),
events=event_cfg,
fillers=FillerConfig.from_mapping(_load_yaml(config_dir / "fillers.yaml")),
fusion=fusion_cfg,
quality=QualityConfig.from_mapping(_load_yaml(config_dir / "quality.yaml")),
speakers=SpeakerConfig.from_mapping(_load_yaml(config_dir / "speakers.yaml")),
digest=digest.hexdigest()[:12],
enhance=load_enhance_config(config_dir),
)
#: Qwen3-ASR language names -> codes used in asr_filter.yaml ``allowed_languages``.
_QWEN_LANGUAGE_CODES = {"Turkish": "tr", "English": "en"}
def language_code(name: str) -> str:
"""Map a detected language name to a code; unknown names are kept (and so rejected)."""
return _QWEN_LANGUAGE_CODES.get(name.strip(), name.strip() or "unknown")
def clean_asr_text(text: str) -> str:
"""NFC + undo the ``i`` + U+0307 that some ASR models emit for a lower-cased ``İ``.
omniASR writes "İtalya" as "i̇talya" (i followed by COMBINING DOT ABOVE); NFC cannot
compose it, so the alignment vocabulary rejects it (pilot smoke run 2026-10-08).
"""
return nfc(text).replace("i\u0307", "i")
def normalise_transcript(text: str) -> str | None:
""":func:`clean_asr_text`, then digits to Turkish words (tag-spec §4: text is verbalized).
Both ASR transcripts go through this before the agreement check, so "3" and "üç" agree,
and the alignment and the stored text never contain digits. ``None`` when a digit token
is ambiguous or unsupported (``14:30``, ``5km``); the caller quarantines the segment.
"""
return verbalize_digits(clean_asr_text(text))
@dataclass(frozen=True)
class WhisperFiller:
"""One filler token in a Whisper transcript, located by its neighbouring words.
Attributes:
index: Position in the normalised token list of the transcript.
filler: Canonical filler (a member of :data:`nida.text.fillers.FILLERS`).
before: Nearest preceding non-filler word, ``None`` at the start of the text.
after: Nearest following non-filler word, ``None`` at the end of the text.
"""
index: int
filler: str
before: str | None
after: str | None
def filler_tokens(text: str) -> list[str]:
"""Tokens of an ASR transcript, fillers in canonical spelling.
NFC (:func:`clean_asr_text`), Turkish lower-case, punctuation split, and every filler
spelling variant mapped through ``canonical_filler`` (``"Ee..."`` becomes ``"ee"``).
"""
return normalize_for_scoring(clean_asr_text(text))
def find_whisper_fillers(tokens: Sequence[str]) -> list[WhisperFiller]:
"""Canonical fillers among ``tokens`` with their non-filler neighbours.
``tokens`` must come from :func:`filler_tokens`. Consecutive fillers share the same
neighbouring words.
"""
flags = [canonical_filler(t) for t in tokens]
found: list[WhisperFiller] = []
for i, canonical in enumerate(flags):
if canonical is None:
continue
before = next((tokens[j] for j in range(i - 1, -1, -1) if flags[j] is None), None)
after = next((tokens[j] for j in range(i + 1, len(tokens)) if flags[j] is None), None)
found.append(WhisperFiller(i, canonical, before, after))
return found
def mark_fillers(tokens: Sequence[str]) -> str:
"""``tokens`` joined by spaces with every filler shown as ``⟦ıı?⟧`` (listening panel)."""
return " ".join(f"⟦{t}?⟧" if canonical_filler(t) is not None else t for t in tokens)
#: Hesitation fillers (the ones the experiment is about); ``of``, ``ah``, ``vay`` ... are
#: interjections Whisper writes without any prompt and are reported separately.
HESITATION_FILLERS = frozenset({"ıı", "ee", "hmm", "ımm", "aa"})
def echoes_prompt(tokens: Sequence[str], prompt_tokens: Sequence[str]) -> bool:
"""True if ``tokens`` contain :data:`PROMPT_ECHO_MIN_TOKENS` consecutive prompt tokens,
or consist only of prompt words / fillers (a partial echo or a filler loop).
Whisper sometimes repeats its prompt instead of transcribing; the fillers of such a
transcript come from the prompt, not from the audio.
"""
vocabulary = set(prompt_tokens)
if tokens and all(t in vocabulary or canonical_filler(t) is not None for t in tokens):
return True
n = PROMPT_ECHO_MIN_TOKENS
windows = {tuple(prompt_tokens[i : i + n]) for i in range(len(prompt_tokens) - n + 1)}
return any(tuple(tokens[i : i + n]) in windows for i in range(len(tokens) - n + 1))
def _basename_in(member: str, *, names: frozenset[str]) -> bool:
"""Archive member selector: the member's file name (any folder) is one of ``names``."""
return member.rsplit("/", 1)[-1] in names
def _rank(seed: int, utt_id: str) -> str:
return hashlib.sha1(f"{seed}\x1f{utt_id}".encode()).hexdigest()
def select_utterances(
utts: Sequence[YodasUtt], plan: PilotPlan, cfg: ingest.IngestConfig
) -> list[YodasUtt]:
"""Deterministic pick: listing duration in range, hash order, up to ``max_hours``."""
budget_ms = plan.max_hours * 3_600_000
eligible = [u for u in utts if cfg.min_duration_ms <= u.duration_ms <= cfg.max_duration_ms]
eligible.sort(key=lambda u: _rank(plan.seed, u.utt_id))
chosen: list[YodasUtt] = []
total = 0
for utt in eligible:
if total + utt.duration_ms > budget_ms:
break
chosen.append(utt)
total += utt.duration_ms
return sorted(chosen, key=lambda u: u.utt_id)
def video_ids_in_text_listing(text_file: str, shard: str) -> set[str]:
"""Video ids of a YODAS ``text/<shard>.txt`` listing (lines ``<utt_id> <caption>``).
Raises:
ValueError: if a line's utterance id has the wrong shape, or the listing has no
utterance at all (a real shard is never empty). An empty time span is not an
error here: the line still names its video.
"""
videos: set[str] = set()
for number, line in enumerate(text_file.splitlines(), start=1):
if not line.strip():
continue
utt_id = line.split(" ", 1)[0]
try:
videos.add(yodas_video_id(utt_id))
except ValueError as err:
raise ValueError(f"YODAS text listing {shard} line {number}: {err}") from err
if not videos:
raise ValueError(f"YODAS text listing {shard} holds no utterance")
return videos
def drop_videos(
utts: Sequence[YodasUtt], blocked_videos: frozenset[str]
) -> tuple[list[YodasUtt], int]:
"""Split off the utterances whose video is in ``blocked_videos``.
Returns the kept utterances (order preserved) and the number dropped. An utterance
whose id has no parseable video part is kept: the ingest stage quarantines it as
``bad_utt_id``, so it is still accounted for.
"""
kept: list[YodasUtt] = []
for utt in utts:
try:
video_id = yodas_video_id(utt.utt_id)
except ValueError:
kept.append(utt)
continue
if video_id not in blocked_videos:
kept.append(utt)
return kept, len(utts) - len(kept)
def audio_root_for(cache: Path, subset: str) -> Path:
"""Folder holding ``<shard>/<utt_id>.wav`` of ``subset`` in the local cache.
tr000 keeps the original ``<cache>/audio`` (existing caches stay valid); every other
subset has its own ``<cache>/audio/<subset>`` because shard numbers repeat across subsets.
"""
return cache / "audio" if subset == DEFAULT_SUBSET else cache / "audio" / subset
def enhanced_audio_root_for(cache: Path, subset: str, arm: str) -> Path:
"""Folder holding ``<shard>/<utt_id>.wav`` of ``arm``'s enhanced audio of ``subset``.
Same layout as :func:`audio_root_for` with ``audio`` replaced by ``audio-enh-<arm>``, so
``<root>/<audio_relpath>`` is the enhanced twin of ``<audio_root>/<audio_relpath>``.
"""
base = cache / enhance.audio_dir_name(arm)
return base if subset == DEFAULT_SUBSET else base / subset
def _untagged_event_counts(rows: Sequence[Mapping[str, object]]) -> dict[str, int]:
"""Per-kind number of untagged (below-band) detections in accepted annotate rows.
Raises:
ValueError: a row without ``untagged_events_json`` (written before ADR-0010).
"""
counts: Counter[str] = Counter()
for row in rows:
raw = row.get("untagged_events_json")
if not isinstance(raw, str):
raise ValueError(f"annotate row {row.get('segment_id')!r} has no untagged_events_json")
counts.update(str(e["kind"]) for e in json.loads(raw))
return dict(sorted(counts.items()))
def _annotate_schema() -> pa.Schema:
import pyarrow as pa
return manifest.schema_from_fields(
[
("segment_id", pa.string()),
("utt_id", pa.string()),
("video_id", pa.string()),
("audio_relpath", pa.string()),
("duration_ms", pa.int64()),
("text", pa.string()),
("plain_transcript", pa.string()),
("tags_json", pa.string()),
("tag_ids_json", pa.string()),
("emotion", pa.string(), True),
("intensity", pa.float64(), True),
("asr_cer", pa.float64()),
("align_score", pa.float64()),
("shout_measured", pa.bool_()),
# ADR-0010: below-band event detections left untagged in an accepted segment
# (mostly uncertain breaths); [{kind, start_ms, end_ms, conf, source}].
("untagged_events_json", pa.string()),
]
)
@dataclass(frozen=True)
class WorkerJob:
"""One prepared worker invocation: everything :meth:`Pilot._run_worker` needs.
Attributes:
name: Results folder name under ``<root>/workers/`` (also the log tag).
python: Interpreter of the worker's isolated environment.
module: Worker module run with ``python -m``.
model_id: Pinned model id.
revision: Pinned model revision.
batch_size: Worker batch size.
requests: One request per segment.
array_outputs: True if results point to arrays in the local cache.
"""
name: str
python: Path
module: str
model_id: str
revision: str
batch_size: int
requests: tuple[WorkRequest, ...]
array_outputs: bool
def run_worker_job(
out_dir: Path,
name: str,
python: Path | Callable[[], Path],
module: str,
model_id: str,
revision: str,
batch_size: int,
requests: Sequence[WorkRequest],
*,
array_outputs: bool,
runner: Callable[[Sequence[str]], int],
retry_failed: bool = False,
skip_if_complete: bool = False,
) -> dict[str, WorkResult]:
"""Run one model worker over ``requests`` (resumable) and return its results.
``out_dir`` holds ``requests.jsonl``, ``results.jsonl`` and ``results.summary.json``. A
results file written by another model/revision is refused; results for clips that are
no longer requested, and results whose array output vanished from the local cache, are
dropped (and so recomputed) before the worker is started.
Args:
out_dir: Results folder of this worker (created on demand).
name: Log tag / error prefix.
python: Interpreter of the worker's environment, or a function returning it that
is called only if the worker is actually started (building an environment is
expensive and pointless when every result already exists).
module: Worker module (``python -m``).
model_id: Pinned model id.
revision: Pinned model revision.
batch_size: Worker batch size.
requests: One request per clip (must not be empty).
array_outputs: ``True`` if results point to ``.npy`` files in the local cache.
runner: Runs the command line, returns the exit code.
retry_failed: Pass ``--retry-failed`` to the worker.
skip_if_complete: Do not start the worker when every requested clip already has a
result (and, with ``retry_failed``, none of them failed). The failure-fraction
check still applies to the stored results.
Raises:
RuntimeError: no requests, results of another model, worker failure, a result count
that differs from the request count, or more than ``MAX_ERROR_FRACTION`` failures.
"""
if not requests:
raise RuntimeError(f"{name}: no requests; did the previous stage produce any segment?")
results_path = out_dir / "results.jsonl"
wanted = {r.clip_id for r in requests}
summary_path = out_dir / "results.summary.json"
if results_path.exists() and summary_path.exists():
previous = read_summary(summary_path)
if (previous.model_id, previous.revision) != (model_id, revision):
raise RuntimeError(
f"{name}: existing results come from {previous.model_id}@"
f"{previous.revision}, not {model_id}@{revision}; move them away first"
)
if results_path.exists():
existing = load_results(results_path, repair=True)
stray = [cid for cid in existing if cid not in wanted]
lost: list[str] = []
if array_outputs:
by_id = {r.clip_id: r for r in requests}
lost = [
cid
for cid, res in existing.items()
if cid in by_id
and res.output is not None
and not Path(str(res.output.get("out_path", ""))).exists()
]
if stray or lost:
log.warning("%s: dropping %d stray and %d lost results", name, len(stray), len(lost))
drop_results(results_path, [*stray, *lost])
if skip_if_complete:
kept = load_results(results_path)
failed = sum(1 for cid in wanted if cid in kept and kept[cid].output is None)
if wanted <= set(kept) and not (retry_failed and failed):
if failed > MAX_ERROR_FRACTION * len(wanted):
raise RuntimeError(
f"{name}: {failed}/{len(wanted)} failed; fix, then --retry-failed"
)
return kept
write_requests(out_dir / "requests.jsonl", requests)
interpreter = python() if callable(python) else python
cmd = [
str(interpreter),
"-m",
module,
"--requests",
str(out_dir / "requests.jsonl"),
"--results",
str(results_path),
"--model-id",
model_id,
"--revision",
revision,
"--batch-size",
str(batch_size),
"--device",
"cuda",
*(["--retry-failed"] if retry_failed else []),
]
code = runner(cmd)
if code != 0:
raise RuntimeError(f"{name}: worker exited with code {code}")
summary = read_summary(out_dir / "results.summary.json")
total = summary.n_ok + summary.n_error
if total != len(requests):
raise RuntimeError(f"{name}: {total} results for {len(requests)} requests")
if summary.n_error > MAX_ERROR_FRACTION * total:
raise RuntimeError(f"{name}: {summary.n_error}/{total} failed; fix, then --retry-failed")
return load_results(results_path)
@dataclass(frozen=True)
class _Enhanced:
"""What the pilot keeps of one successful enhancement result."""
length_diff_samples: int
peak: float
loudness_dbfs: float
output: Mapping[str, object]
def _ids_digest(ids: Iterable[str]) -> str:
"""SHA-256 of the sorted ids (identity of the excluded set, recorded in marker and meta)."""
return hashlib.sha256("\n".join(sorted(ids)).encode("utf-8")).hexdigest()
def _number(segment_id: str, output: Mapping[str, object], key: str) -> float:
value = output.get(key)
if isinstance(value, bool) or not isinstance(value, int | float) or not math.isfinite(value):
raise RuntimeError(f"{segment_id}: enhance result has no finite number {key!r}")
return float(value)
def _enhance_outcome(
segment_id: str,
result: WorkResult,
cfg: EnhanceConfig,
clipping_threshold: float,
wav_path: Path | None,
) -> _Enhanced | QuarantineRecord:
"""Judge one enhancement result: what to keep of it, or the quarantine record.
The worker only refuses absurd length differences; the tolerance and the clipping limit are
applied here, to the recorded ``length_diff_samples`` and ``clipping_ratio`` (so the config
governs the outcome of every run). A length difference is judged before clipping.
Args:
clipping_threshold: ``acoustic.yaml`` threshold the current run counts clipping with;
a result made with another one is an error (its ratio means something else).
wav_path: Where the segment's wav must be (checked when a later stage will read it);
``None`` = nobody needs it in this session, its absence is expected.
Raises:
RuntimeError: if a successful result lacks ``length_diff_samples``, ``peak``,
``loudness_dbfs`` or ``clipping_ratio``, was made with another clipping threshold,
or its wav is missing from ``wav_path`` (a worker that lies is a bug, not a bad
segment).
"""
if result.output is None:
reason = (
enhance.REASON_LENGTH
if enhance.is_length_error(result.error)
else enhance.REASON_FAILED
)
return QuarantineRecord(segment_id, ENHANCE_STAGE, reason, {"error": result.error})
diff = result.output.get("length_diff_samples")
if isinstance(diff, bool) or not isinstance(diff, int):
raise RuntimeError(f"{segment_id}: enhance result has no integer length_diff_samples")
peak = _number(segment_id, result.output, "peak")
loudness = _number(segment_id, result.output, "loudness_dbfs")
ratio = _number(segment_id, result.output, "clipping_ratio")
if _number(segment_id, result.output, "clipping_threshold") != clipping_threshold:
raise RuntimeError(
f"{segment_id}: the enhance result counts clipping at "
f"{result.output['clipping_threshold']}, acoustic.yaml says {clipping_threshold}; "
"move the enhance worker results away and rerun"
)
if wav_path is not None and not wav_path.is_file():
raise RuntimeError(f"{segment_id}: enhance result without a wav at {wav_path}")
if abs(diff) > cfg.length_tolerance_samples:
return QuarantineRecord(
segment_id,
ENHANCE_STAGE,
enhance.REASON_LENGTH,
{"length_diff_samples": diff, "tolerance_samples": cfg.length_tolerance_samples},
)
if ratio > cfg.max_enhanced_clipping_ratio:
return QuarantineRecord(
segment_id,
ENHANCE_STAGE,
enhance.REASON_CLIPPED,
{
"clipping_ratio": ratio,
"max_enhanced_clipping_ratio": cfg.max_enhanced_clipping_ratio,
"clipping_threshold": clipping_threshold,
"peak": peak,
},
)
return _Enhanced(diff, peak, loudness, result.output)
def _worker_runtime(outputs: Iterable[Mapping[str, object]]) -> dict[str, object]:
"""Facts the worker added to its payloads (:data:`nida.data.enhance.PAYLOAD_KEYS` aside).
One value per key; if the items disagree (a model file changed between sessions) the key
maps to the list of distinct values and a warning is logged.
"""
seen: dict[str, dict[str, object]] = {}
for output in outputs:
for key, value in output.items():
if key not in enhance.PAYLOAD_KEYS:
seen.setdefault(key, {})[json.dumps(value, sort_keys=True)] = value
runtime: dict[str, object] = {}
for key, values in sorted(seen.items()):
if len(values) > 1:
log.warning("enhance: the worker reported %d different values of %s", len(values), key)
runtime[key] = list(values.values())
else:
runtime[key] = next(iter(values.values()))
return runtime
def _enhance_section(marker: Mapping[str, Any]) -> str:
"""The ``reports/pilot.md`` section of an enhanced run (between sentinel comments)."""
length, clipping = marker["length_diff_samples"], marker["clipping"]
lines = [
ENHANCE_REPORT_BEGIN,
f"## Konuşma iyileştirme ({marker['arm']})",
"",
f"- Model: `{marker['model_id']}` @ `{marker['revision']}`",
f"- Kabul edilen segment: {marker['accepted']}; iyileştirilen: {marker['enhanced']}",
f"- Karantina: {json.dumps(marker['quarantined'], ensure_ascii=False)}",
f"- Uzunluk farkı: en çok {length['max_abs']} örnek; sıfırdan farklı "
f"{length['nonzero']} segment (tolerans {marker['length_tolerance_ms']} ms, işçi "
f"sınırı {marker['max_length_diff_ms']} ms)",
f"- Tam ölçeği aşan (tepe > 1,0; PCM16'ya kırpılır): "
f"{clipping['segments_over_full_scale']} segment; en yüksek tepe {clipping['max_peak']}",
f"- Kırpma oranı sınırı ({marker['max_enhanced_clipping_ratio']}, eşik "
f"{marker['clipping_threshold']}) nedeniyle karantinaya alınan: "
f"{clipping['quarantined']} segment; en yüksek oran {clipping['max_ratio']}",
"- Etiketler özgün ses üzerinde belirlendi (annotate); iyileştirilmiş sesle yeniden "
"doğrulanmadı.",
ENHANCE_REPORT_END,
]
return "\n".join(lines)
def _with_enhance_section(text: str, marker: Mapping[str, Any] | None) -> str:
"""``text`` with its enhance section replaced by (or, for ``None``, stripped of) the marker's.
Idempotent, and ``report`` and ``enhance`` produce the same bytes in either order.
"""
begin = text.find(ENHANCE_REPORT_BEGIN)
if begin >= 0:
end = text.index(ENHANCE_REPORT_END, begin) + len(ENHANCE_REPORT_END)
text = text[:begin].rstrip("\n") + "\n" + text[end:].lstrip("\n")
if marker is None:
return text
return text.rstrip("\n") + "\n\n" + _enhance_section(marker) + "\n"
@dataclass
class Pilot:
root: Path
cache: Path
envs_root: Path
repo_dir: Path
sources_path: Path
config_dir: Path
plan: PilotPlan = PilotPlan()
uv: str | None = None
retry_failed: bool = False
#: ASR batch size override (operational: the bench sizes are for a 22 GB L4; a 96 GB G4
#: takes far larger batches). Not part of the plan: it does not change the outputs.
asr_batch_size: int | None = None
#: Worker subprocesses ``run("all")`` may run at the same time (``sed``, ``diar``, both
#: ASR systems, ``ctc``, ``emotion``). 1 = the plain sequential order. Operational like
#: ``asr_batch_size``: it does not change the outputs.
parallel_workers: int = 1
runner: Callable[[Sequence[str]], int] = _stream_runner
opener: fetch.Opener = fetch.default_opener
def __post_init__(self) -> None:
if isinstance(self.parallel_workers, bool) or self.parallel_workers < 1:
raise ValueError(
f"parallel_workers must be an integer >= 1, got {self.parallel_workers!r}"
)
# ---------------------------------------------------------------- paths/helpers
@property
def work(self) -> Path:
return self.root / "work"
@property
def audio_root(self) -> Path:
return audio_root_for(self.cache, self.plan.subset)
def _enhanced_root(self, cfg: EnhanceConfig) -> Path:
return enhanced_audio_root_for(self.cache, self.plan.subset, cfg.arm)
def _dataset_dir(self, out_dir: Path, cfg: EnhanceConfig) -> Path:
"""``dataset-<annotate hash>``, plus ``-enh-<arm>`` for an enhancement arm."""
suffix = enhance.arm_suffix(cfg.arm) if cfg.enabled else ""
return self.root / f"dataset-{out_dir.name.removeprefix('annotate-')}{suffix}"
def _primary(self) -> System:
return ASR_SYSTEMS[self.plan.asr_primary]
def _secondary(self) -> System:
return ASR_SYSTEMS[self.plan.asr_secondary]
def _require(self) -> None:
if self.plan.asr_primary == self.plan.asr_secondary:
raise ValueError("primary and secondary ASR must differ")
if self._primary().worker == self._secondary().worker:
# the agreement filter needs uncorrelated errors (ADR-0001 rule 2, ADR-0011); "family"
# is approximated by the worker module (e.g. Whisper and BuzzASR share asr_hf)
raise ValueError(
"primary and secondary ASR must come from different model families, got "
f"{self.plan.asr_primary} and {self.plan.asr_secondary} "
f"(both {self._primary().worker})"
)
if self._language_role() is None:
raise ValueError(
"no ASR system of the plan identifies the language: the S4 language filter "
"would be silently disabled (docs/03; use a Qwen3-ASR system in either role)"
)
needed = {"yodas_tr", self._primary().source_id, self._secondary().source_id}
needed |= {m.source_id for m in DATA_MODELS.values()}
arm = load_enhance_config(self.config_dir)
if arm.enabled:
needed.add(enhance_model(arm.arm).source_id)
require_approved(needed, load_sources(self.sources_path))
def _uv(self) -> str:
if self.uv is None:
self.uv = bench_envs.find_uv(bench_envs.subprocess_runner)
return self.uv
def _plan_check(self) -> None:
path = self.root / "manifests" / "pilot_plan.json"
if path.exists():
existing = json.loads(path.read_text(encoding="utf-8"))
if isinstance(existing, dict) and "subset" not in existing:
existing = {**existing, "subset": DEFAULT_SUBSET} # written before the key existed
if existing != asdict(self.plan):
raise ValueError(
f"{path} was written for a different plan: {existing} != {asdict(self.plan)}"
)
else:
atomic_write_json(path, asdict(self.plan))
def _listing_file(self, subset: str, kind: str, shard: str) -> Path:
"""Download (cached) ``data/<subset>/<kind>/<shard>.txt``; ``kind``: text | duration.
tr000 keeps the original cache folder ``<cache>/yodas/<kind>``; other subsets use
``<cache>/yodas/<subset>/<kind>`` (shard numbers repeat across subsets).
"""
name = f"data/{subset}/{kind}/{shard}.txt"
url = fetch.hf_resolve_url("espnet/yodas", self.plan.yodas_revision, name, "dataset")
folder = self.cache / "yodas"
dest = (folder if subset == DEFAULT_SUBSET else folder / subset) / kind / f"{shard}.txt"
return fetch.download(url, dest, opener=self.opener).path
def _listing(self, shard: str) -> list[YodasUtt]:
subset = self.plan.subset
return parse_yodas_listing(
self._listing_file(subset, "text", shard).read_text(encoding="utf-8"),
self._listing_file(subset, "duration", shard).read_text(encoding="utf-8"),
shard,
)
def _tr000_video_ids(self) -> frozenset[str]:
"""Video ids of all tr000 shards (text listings only), to de-duplicate other subsets.
Raises:
nida.bench.fetch.FetchError: if a tr000 text listing cannot be downloaded.
"""
videos: set[str] = set()
for shard in TR000_SHARDS:
try:
path = self._listing_file(DEFAULT_SUBSET, "text", shard)
except fetch.FetchError as err:
raise fetch.FetchError(
f"tr000 text listing {shard} is unavailable at revision "
f"{self.plan.yodas_revision}; the de-duplication needs all "
f"{TR000_NUM_SHARDS} listings (TR000_SHARDS): {err}"
) from err
videos |= video_ids_in_text_listing(path.read_text(encoding="utf-8"), shard)
log.info("tr000 holds %d videos in %d shards", len(videos), TR000_NUM_SHARDS)
return frozenset(videos)
def _selection(self, cfgs: _Configs) -> tuple[list[YodasUtt], dict[str, int]]:
"""Selected utterances and listing counts (nothing leaves the listing uncounted).
``listed`` = ``listed_duration_out_of_range`` + ``beyond_hour_budget`` + ``selected``
(+ ``duplicate_video_in_tr000`` for a subset other than tr000). The duplicates are
removed first, so the duration counts describe the utterances that remain.
"""
listed = [u for shard in self.plan.shard_list() for u in self._listing(shard)]
ids = [u.utt_id for u in listed]
if len(set(ids)) != len(ids):
raise ValueError("YODAS utterance ids repeat across the pilot shards")
utts, duplicates = listed, 0
if self.plan.subset != DEFAULT_SUBSET:
utts, duplicates = drop_videos(listed, self._tr000_video_ids())
log.info(
"%s: dropped %d of %d listed utterances whose video is in tr000",
self.plan.subset,
duplicates,
len(listed),
)
chosen = select_utterances(utts, self.plan, cfgs.ingest)
in_range = sum(
1
for u in utts
if cfgs.ingest.min_duration_ms <= u.duration_ms <= cfgs.ingest.max_duration_ms
)
counts = {
"listed": len(listed),
"listed_duration_out_of_range": len(utts) - in_range,
"beyond_hour_budget": in_range - len(chosen),
"selected": len(chosen),
}
if self.plan.subset != DEFAULT_SUBSET:
counts[DUPLICATE_VIDEO_KEY] = duplicates
return chosen, counts
def _extract(self, utts: Sequence[YodasUtt]) -> None:
"""Make sure every given utterance's wav is in the cache (streams each shard once).
Members are matched by file name, not by their folder inside the archive: some
YODAS tr000 archives (00000008 and later) keep every wav under ``00000000/`` instead
of their own shard number. Extracted files go to a staging folder first and are
then moved to ``<audio_root>/<shard>/<utt_id>.wav``; a file name that occurs in two
archive folders is an error (the utterance would be ambiguous).
"""
by_shard: dict[str, set[str]] = defaultdict(set)
for utt in utts:
if not (self.audio_root / utt.shard / f"{utt.utt_id}.wav").exists():
by_shard[utt.shard].add(f"{utt.utt_id}.wav")
for shard, wanted in sorted(by_shard.items()):
url = fetch.hf_resolve_url(
"espnet/yodas",
self.plan.yodas_revision,
f"data/{self.plan.subset}/audio/{shard}.tar.gz",
"dataset",
)
log.info("extracting %d wavs from shard %s", len(wanted), shard)
staging = self.audio_root / ".staging" / shard
extracted = fetch.stream_extract(
url,
staging,
functools.partial(_basename_in, names=frozenset(wanted)),
opener=self.opener,
)
by_name: dict[str, Path] = {}
for path in extracted:
if path.name in by_name:
raise RuntimeError(
f"shard {shard}: {path.name} occurs twice in the archive "
f"({by_name[path.name].parent.name}/ and {path.parent.name}/)"
)
by_name[path.name] = path
elsewhere = sum(1 for p in by_name.values() if p.parent.name != shard)
if elsewhere:
log.warning("shard %s: %d wavs sit in another archive folder", shard, elsewhere)
target_dir = self.audio_root / shard
target_dir.mkdir(parents=True, exist_ok=True)
for name, path in by_name.items():
path.replace(target_dir / name)
shutil.rmtree(staging)
def _ingested(self) -> list[dict[str, object]]:
stage_dir = self.work / "ingest"
return manifest.read_manifest(stage_dir) if stage_dir.exists() else []
def _run_worker(
self,
name: str,
python: Path,
module: str,
model_id: str,
revision: str,
batch_size: int,
requests: Sequence[WorkRequest],
*,
array_outputs: bool,
) -> dict[str, WorkResult]:
"""Run one worker over ``requests`` (resumable) and return its results."""
return run_worker_job(
self.root / "workers" / name,
name,
python,
module,
model_id,
revision,
batch_size,
requests,
array_outputs=array_outputs,
runner=self.runner,
retry_failed=self.retry_failed,
)
def _pending(
self,
name: str,
rows: Sequence[Mapping[str, object]],
*,
array_outputs: bool,
include_failed: bool = True,
) -> list[Mapping[str, object]]:
"""Rows whose worker result is missing, failed or lost its array (need audio).
``include_failed=False`` leaves out rows with a recorded failure (a worker only retries
those under ``--retry-failed``, so their audio is not needed otherwise).
"""
existing = load_results(self.root / "workers" / name / "results.jsonl", repair=True)
pending: list[Mapping[str, object]] = []
for row in rows:
result = existing.get(str(row["segment_id"]))
lost = (
result is not None
and result.output is not None
and array_outputs
and not Path(str(result.output.get("out_path", ""))).is_file()
)
if result is None or (result.output is None and include_failed) or lost:
pending.append(row)
return pending
def _prepare_data_job(
self,
model: DataModel,
params: Callable[[str], dict[str, str]],
*,
name: str | None = None,
array_outputs: bool = True,
rows: Sequence[Mapping[str, object]] | None = None,
extract: bool = True,
audio_root: Path | None = None,
extract_failed: bool = True,
) -> WorkerJob:
"""Prepare a data-model worker: audio, environment and requests (main thread only).
``extract_failed=False`` does not extract the audio of rows whose result is a recorded
failure (the enhancement stage: the worker skips them unless ``--retry-failed``).
``extract=False`` leaves the audio to the caller (the parallel group extracts the
union of its jobs' pending rows once, before any thread starts).
``audio_root`` reads another audio root than the original one (the enhanced wavs of an
enhancement arm). Nothing is extracted for it: a pending row whose wav is missing is an
error naming the stage that makes it.
"""
rows = self._ingested() if rows is None else rows
key = name or model.key
if audio_root is not None:
self._require_audio(
audio_root, self._pending(key, rows, array_outputs=array_outputs), key
)
elif extract:
pending = self._pending(
key, rows, array_outputs=array_outputs, include_failed=extract_failed
)
self._extract(self._utts_of(pending))
spec = envs.data_env_for_worker(model.worker)
python = bench_envs.ensure_env(spec, self.envs_root, self.repo_dir, uv=self._uv())
root = self.audio_root if audio_root is None else audio_root
requests = tuple(
WorkRequest(
str(row["segment_id"]),
str(root / str(row["audio_relpath"])),
params(str(row["segment_id"])),
)
for row in rows
)
return WorkerJob(
key,
python,
envs.worker_module_name(model.worker),
model.ref.model_id,
model.ref.revision,
model.batch_size,
requests,
array_outputs,
)
@staticmethod
def _require_audio(root: Path, rows: Sequence[Mapping[str, object]], what: str) -> None:
"""Raise if a wav of ``rows`` is missing under ``root`` (never extracted)."""
missing = [
str(row["audio_relpath"])
for row in rows
if not (root / str(row["audio_relpath"])).is_file()
]
if missing:
raise RuntimeError(
f"{what}: {len(missing)} enhanced wavs are missing from {root} (e.g. "
f"{missing[:3]}); run the enhance stage first (a new Colab session loses them)"
)
def _execute_job(self, job: WorkerJob, *, tag_logs: bool = False) -> dict[str, WorkResult]:
"""Run a prepared job (blocks until its subprocess ends; safe to call from a thread).
With ``tag_logs`` every streamed worker line is prefixed with the job name.
"""
token = LOG_TAG.set(job.name) if tag_logs else None
try:
return self._run_worker(
job.name,
job.python,
job.module,
job.model_id,
job.revision,
job.batch_size,
job.requests,
array_outputs=job.array_outputs,
)
finally:
if token is not None:
LOG_TAG.reset(token)
def _data_worker(
self,
model: DataModel,
params: Callable[[str], dict[str, str]],
*,
name: str | None = None,
array_outputs: bool = True,
rows: Sequence[Mapping[str, object]] | None = None,
audio_root: Path | None = None,
) -> None:
self._execute_job(
self._prepare_data_job(
model,
params,
name=name,
array_outputs=array_outputs,
rows=rows,
audio_root=audio_root,
)
)
def _npy(self, model: DataModel, segment: str) -> str:
return str(self.cache / "arrays" / model.key / f"{segment}.npy")
@staticmethod
def _utts_of(rows: Sequence[Mapping[str, object]]) -> list[YodasUtt]:
"""Extraction keys from ingest *or* annotate rows (both carry utt_id,
duration_ms and audio_relpath = ``<shard>/<utt_id>.wav``; captions are not needed)."""
utts: list[YodasUtt] = []
for r in rows:
shard, _, name = str(r["audio_relpath"]).partition("/")
if name != f"{r['utt_id']}.wav" or not shard:
raise ValueError(f"unexpected audio_relpath {r['audio_relpath']!r}")
utts.append(YodasUtt(str(r["utt_id"]), "", int(str(r["duration_ms"])), shard))
return utts
# ------------------------------------------------------------------- stages
def run(self, stage: str) -> dict[str, object]:
if stage == "all":
out: dict[str, object] = {}
for name in ALL_STAGES:
if self.parallel_workers > 1 and name in PARALLEL_GROUP:
if name == PARALLEL_GROUP[0]:
out.update(self._run_worker_group())
continue
out[name] = self.run(name)
return out
if stage not in ALL_STAGES + EXTRA_STAGES:
known = ", ".join(ALL_STAGES + EXTRA_STAGES)
raise ValueError(f"unknown pilot stage {stage!r}; known: {known}")
log.info("=== pilot stage %s", stage)
self._require()
self._plan_check()
handler: Callable[[], dict[str, object]] = getattr(self, "_stage_" + stage)
return handler()
def _stage_check(self) -> dict[str, object]:
cfgs = load_configs(self.config_dir)
return {"configs": cfgs.digest, "shards": self.plan.shard_list()}
def _stage_ingest(self) -> dict[str, object]:
cfgs = load_configs(self.config_dir)
selection, listing = self._selection(cfgs)
stage_dir = self.work / "ingest"
qlog = QuarantineLog(stage_dir)
qlog.recover()
done = qlog.quarantined_ids()
if stage_dir.exists():
done |= {str(r["utt_id"]) for r in manifest.read_manifest(stage_dir, ["utt_id"])}
todo = [u for u in selection if u.utt_id not in done]
self._extract(todo)
rows: list[dict[str, object]] = []
drops: list[QuarantineRecord] = []
for index, utt in enumerate(todo, start=1):
result: dict[str, object] | QuarantineRecord
if not (self.audio_root / utt.shard / f"{utt.utt_id}.wav").exists():
result = QuarantineRecord(
utt.utt_id, ingest.STAGE, "missing_from_archive", {"shard": utt.shard}
)
else:
result = ingest.ingest_utterance(utt, self.audio_root, cfgs.ingest, cfgs.acoustic)
if isinstance(result, QuarantineRecord):
drops.append(result)
else:
rows.append(result)
if len(rows) >= INGEST_CHUNK or index == len(todo):
if drops:
qlog.append(drops)
drops = []
if rows:
manifest.write_part(stage_dir, rows, ingest.ingest_schema())
rows = []
log.info("ingest: %d/%d", index, len(todo))
kept = self._ingested()
hours = sum(int(str(r["duration_ms"])) for r in kept) / 3_600_000
return {
"listing": listing,
"selected": len(selection),
"kept": len(kept),
"kept_hours": round(hours, 3),
"quarantined": qlog.counts_by_reason(),
}
@staticmethod
def _sed_classes(cfgs: _Configs) -> list[str]:
classes = sorted({n for names in cfgs.events.classes.values() for n in names})
return classes + [c for c in EXTRA_SED_CLASSES if c not in classes]
def _sed_name(self, cfgs: _Configs) -> str:
"""SED results are keyed by the class list: a changed events.yaml recomputes them."""
digest = hashlib.sha1(json.dumps(self._sed_classes(cfgs)).encode("utf-8")).hexdigest()
return f"panns_cnn14_dlm-{digest[:8]}"
def _sed_job(self, cfgs: _Configs, *, extract: bool = True) -> WorkerJob:
model = DATA_MODELS["panns_cnn14_dlm"]
classes = self._sed_classes(cfgs)
name = self._sed_name(cfgs)
return self._prepare_data_job(
model,
lambda sid: {
"out_path": str(self.cache / "arrays" / name / f"{sid}.npy"),
"classes": json.dumps(classes),
},
name=name,
extract=extract,
)
def _sed_out(self, cfgs: _Configs) -> dict[str, object]:
return {"classes": self._sed_classes(cfgs), "results": self._sed_name(cfgs)}
def _diar_job(self, *, extract: bool = True) -> WorkerJob:
model = DATA_MODELS["sortformer_v2"]
return self._prepare_data_job(
model, lambda sid: {"out_path": self._npy(model, sid)}, extract=extract
)
def _ctc_job(self, *, extract: bool = True) -> WorkerJob:
model = DATA_MODELS["mpoyraz_ctc_tr"]
return self._prepare_data_job(
model, lambda sid: {"out_path": self._npy(model, sid)}, extract=extract
)
def _timings_ctc_job(
self, rows: Sequence[Mapping[str, object]], cfg: EnhanceConfig = NO_ENHANCE
) -> WorkerJob:
"""CTC over ``rows`` only, in its own results folder (never prunes the ``ctc`` stage's).
With an enhancement arm the CTC runs on the enhanced wavs, in a folder of its own
(``...-timings-enh-<arm>``) so original-audio and enhanced-audio emissions never mix.
"""
model = DATA_MODELS["mpoyraz_ctc_tr"]
name = model.key + TIMINGS_CTC_SUFFIX + (enhance.arm_suffix(cfg.arm) if cfg.enabled else "")
return self._prepare_data_job(
model,
lambda sid: {"out_path": str(self.cache / "arrays" / name / f"{sid}.npy")},
name=name,
rows=rows,
audio_root=self._enhanced_root(cfg) if cfg.enabled else None,
)
def _emotion_job(self, *, extract: bool = True) -> WorkerJob:
model = DATA_MODELS["emotion2vec_plus_large"]
return self._prepare_data_job(model, lambda sid: {}, array_outputs=False, extract=extract)
def _stage_sed(self) -> dict[str, object]:
cfgs = load_configs(self.config_dir)
self._execute_job(self._sed_job(cfgs))
return self._sed_out(cfgs)
def _stage_diar(self) -> dict[str, object]:
self._execute_job(self._diar_job())
return {}
def _stage_ctc(self) -> dict[str, object]:
self._execute_job(self._ctc_job())
return {}
def _stage_emotion(self) -> dict[str, object]:
self._execute_job(self._emotion_job())
return {}
@staticmethod
def _asr_language(system: System) -> str:
"""A Qwen system identifies the language itself, in either role (S4 language filter,
docs/03; ADR-0011 makes Qwen the primary): its detected language is checked in
annotate. Every other system is forced to Turkish."""
if system.worker == "asr_qwen":
return "auto"
return LANGUAGE
def _language_role(self) -> str | None:
"""The ASR role whose detected language drives the S4 language filter: the role run
with automatic language identification (the secondary when both are), else ``None``."""
for role in ("secondary", "primary"):
if self._asr_language(self._asr_system(role)) == "auto":
return role
return None
def _asr_system(self, role: str) -> System:
if role not in ASR_ROLES:
raise ValueError(f"unknown ASR role {role!r}; known: {ASR_ROLES}")
return self._primary() if role == "primary" else self._secondary()
def _asr_job(
self, role: str, rows: Sequence[Mapping[str, object]], *, extract: bool = True
) -> WorkerJob:
"""Prepare the ASR worker of ``role`` ("primary" or "secondary") over ``rows``."""
system = self._asr_system(role)
name = f"asr-{system.system_id}"
if extract:
self._extract(self._utts_of(self._pending(name, rows, array_outputs=False)))
spec = bench_envs.env_for_worker(system.worker)
python = bench_envs.ensure_env(spec, self.envs_root, self.repo_dir, uv=self._uv())
requests = tuple(
WorkRequest(
str(r["segment_id"]),
str(self.audio_root / str(r["audio_relpath"])),
{"language": self._asr_language(system)},
)
for r in rows
)
return WorkerJob(
name,
python,
f"nida.bench.workers.{system.worker}",
system.model_id,
system.revision,
self.asr_batch_size or system.batch_size,
requests,
array_outputs=False,
)
def _asr_out(self, role: str, results: Mapping[str, WorkResult]) -> dict[str, object]:
return {
"system": self._asr_system(role).system_id,
"ok": sum(r.ok for r in results.values()),
}
def _stage_asr(self) -> dict[str, object]:
rows = self._ingested()
out: dict[str, object] = {}
for role in ASR_ROLES:
out[role] = self._asr_out(role, self._execute_job(self._asr_job(role, rows)))
return out
# ------------------------------------------------------------ parallel worker group
def _run_worker_group(self) -> dict[str, object]:
"""Run ``sed``, ``diar``, ``asr`` (both systems), ``ctc`` and ``emotion`` concurrently.
Prepare (main thread, sequential): build every job (environments are created one
at a time; uv builds must not overlap), then extract the audio of the *union* of the
jobs' pending rows once, so no job touches the shared audio cache while the workers
run. Execute: each job's subprocess in a pool of ``parallel_workers`` threads. All
jobs are waited for (a failing job does not cancel the others, so their results are
saved); if any failed, one ``RuntimeError`` names every failed job.
Thread safety: a job only reads and writes ``<root>/workers/<job name>/`` and its
own array folder under ``<cache>/arrays/``; job names are distinct (checked below).
``self.runner`` is stateless (one subprocess per call) and the other ``Pilot``
fields are read-only while the threads run.
Returns:
The values the sequential stages return, keyed by stage, in ``ALL_STAGES`` order.
Raises:
RuntimeError: if one or more jobs failed (after all jobs have finished).
"""
log.info(
"=== pilot stages %s (parallel_workers=%d)",
", ".join(PARALLEL_GROUP),
self.parallel_workers,
)
self._require()
self._plan_check()
cfgs = load_configs(self.config_dir)
rows = self._ingested()
jobs = [
self._sed_job(cfgs, extract=False),
self._diar_job(extract=False),
self._asr_job("primary", rows, extract=False),
self._asr_job("secondary", rows, extract=False),
self._ctc_job(extract=False),
self._emotion_job(extract=False),
]
names = [job.name for job in jobs]
if len(set(names)) != len(names):
raise RuntimeError(f"parallel worker jobs share a results folder: {sorted(names)}")
pending: dict[str, Mapping[str, object]] = {}
for job in jobs:
for row in self._pending(job.name, rows, array_outputs=job.array_outputs):
pending[str(row["segment_id"])] = row
self._extract(self._utts_of(list(pending.values())))
results: dict[str, dict[str, WorkResult]] = {}
errors: dict[str, Exception] = {}
with ThreadPoolExecutor(
max_workers=self.parallel_workers, thread_name_prefix="pilot-worker"
) as pool:
futures = {job.name: pool.submit(self._execute_job, job, tag_logs=True) for job in jobs}
for name, future in futures.items(): # the pool has shut down: every future is done
try:
results[name] = future.result()
except Exception as err: # noqa: BLE001 - collected and raised together below
log.error("%s: worker job failed: %s: %s", name, type(err).__name__, err)
errors[name] = err
if errors:
detail = "; ".join(f"{n}: {type(e).__name__}: {e}" for n, e in errors.items())
raise RuntimeError(
f"{len(errors)} of {len(jobs)} parallel worker jobs failed: {detail}"
) from next(iter(errors.values()))
return {
"sed": self._sed_out(cfgs),
"diar": {},
"asr": {
role: self._asr_out(role, results[f"asr-{self._asr_system(role).system_id}"])
for role in ASR_ROLES
},
"ctc": {},
"emotion": {},
}
# ---------------------------------------------------------------- annotate
def _results(self, name: str) -> dict[str, WorkResult]:
return load_results(self.root / "workers" / name / "results.jsonl")
def _stage_annotate(self) -> dict[str, object]:
import numpy as np
import soundfile # type: ignore[import-untyped] # untyped
cfgs = load_configs(self.config_dir)
rows = self._ingested()
self._extract(self._utts_of(rows))
out_dir = self.work / f"annotate-{cfgs.digest}"
qlog = QuarantineLog(out_dir)
qlog.recover()
done = qlog.quarantined_ids()
if out_dir.exists():
done |= manifest.done_ids(out_dir, "segment_id")
res = {
"sed": self._results(self._sed_name(cfgs)),
"diar": self._results("sortformer_v2"),
"ctc": self._results("mpoyraz_ctc_tr"),
"emotion": self._results("emotion2vec_plus_large"),
"primary": self._results(f"asr-{self.plan.asr_primary}"),
"secondary": self._results(f"asr-{self.plan.asr_secondary}"),
}
todo_ids = {str(r["segment_id"]) for r in rows} - done
array_keys = ("sed", "diar", "ctc")
for key, results in res.items():
missing = [str(r["segment_id"]) for r in rows if str(r["segment_id"]) not in results]
if missing:
raise RuntimeError(
f"annotate: {key} has no result for {len(missing)} segments "
f"(e.g. {missing[:3]}); run its stage first"
)
if key in array_keys:
lost = [
cid
for cid, result in results.items()
if result.output is not None
and not Path(str(result.output["out_path"])).is_file()
]
if lost:
raise RuntimeError(
f"annotate: {len(lost)} {key} arrays are not in the local cache "
"(new Colab session?); rerun the sed/diar/ctc stages first"
)
deferred = sorted(
sid for sid in todo_ids if any(res[key][sid].output is None for key in res)
)
if deferred:
log.warning(
"annotate: %d segments have failed worker items; they are skipped (not "
"quarantined) until the worker is rerun with --retry-failed",
len(deferred),
)
todo_ids -= set(deferred)
# Pass 1 over ALL ingested segments (not only the ones left to do): speaker
# baselines must not depend on how far an interrupted run got. Only the
# segments still to do are staged / quarantined.
staged: list[tuple[dict[str, object], Any]] = []
drops: list[QuarantineRecord] = []
features_by_video: dict[str, list[acoustic.WordFeatures]] = defaultdict(list)
for row in rows:
sid = str(row["segment_id"])
if any(res[key][sid].output is None for key in res):
continue # deferred (worker error), see above
outcome = self._annotate_one(row, res, cfgs, np, soundfile)
if isinstance(outcome, QuarantineRecord):
if sid in todo_ids:
drops.append(outcome)
continue
features_by_video[str(row["video_id"])].extend(outcome["features"])
if sid in todo_ids:
staged.append((row, outcome))
if len(drops) >= INGEST_CHUNK:
qlog.append(drops)
drops = []
# Pass 2: speaker-relative shout scores (per video), fusion, emotion.
baselines: dict[str, acoustic.SpeakerBaseline | None] = {}
for video, feats in features_by_video.items():
try:
baselines[video] = acoustic.speaker_baseline(feats, cfgs.acoustic)
except acoustic.BaselineError:
baselines[video] = None # too few voiced words: shout not measurable
out_rows: list[dict[str, object]] = []
fillers_inserted: Counter[str] = Counter() # in accepted segments of this pass
fillers_skipped_event = 0 # candidates an event detector also heard (accepted segs)
for row, outcome in staged:
sid = str(row["segment_id"])
styles: list[WordStyleScore] = [acoustic.phrase_support(outcome["whisper"])]
baseline = baselines[str(row["video_id"])]
if baseline is not None:
shout = acoustic.shout_scores(outcome["features"], baseline, cfgs.acoustic)
styles.append(acoustic.phrase_support(shout))
fused = fuse_segment(
sid,
outcome["words"],
int(str(row["duration_ms"])),
outcome["events"],
styles,
cfgs.fusion,
)
if fused.quarantine is not None:
drops.append(fused.quarantine)
continue
emotion = emotion_condition(outcome["emotion_scores"], cfgs.fusion)
assert fused.text is not None
out_rows.append(
{
"segment_id": sid,
"utt_id": row["utt_id"],
"video_id": row["video_id"],
"audio_relpath": row["audio_relpath"],
"duration_ms": row["duration_ms"],
"text": fused.text,
"plain_transcript": nfc(str(outcome["transcript"])),
"tags_json": json.dumps([asdict(t) for t in fused.tags], ensure_ascii=False),
"tag_ids_json": json.dumps(sorted({t.tag for t in fused.tags})),
"emotion": emotion.emotion,
"intensity": emotion.intensity,
"asr_cer": outcome["asr_cer"],
"align_score": outcome["align_score"],
"shout_measured": baseline is not None,
"untagged_events_json": json.dumps(
[asdict(e) for e in fused.ignored_events], ensure_ascii=False
),
}
)
fillers_inserted.update(c.text for c in outcome["fillers"])
fillers_skipped_event += int(outcome["fillers_skipped_event"])
if len(out_rows) >= INGEST_CHUNK:
manifest.write_part(out_dir, out_rows, _annotate_schema())
out_rows = []
if out_rows:
manifest.write_part(out_dir, out_rows, _annotate_schema())
if drops:
qlog.append(drops)
accepted = manifest.read_manifest(out_dir) if out_dir.exists() else []
return {
"config_digest": cfgs.digest,
"accepted": len(accepted),
"accepted_hours": round(
sum(int(str(r["duration_ms"])) for r in accepted) / 3_600_000, 3
),
"quarantined": qlog.counts_by_reason(),
"videos_without_shout_baseline": sum(1 for b in baselines.values() if b is None),
"untagged_events": _untagged_event_counts(accepted),
"accepted_without_shout_measurement": sum(
1 for r in accepted if not r["shout_measured"]
),
"deferred_worker_errors": len(deferred),
"fillers_enabled": cfgs.fillers.enabled,
"fillers_inserted_this_pass": dict(sorted(fillers_inserted.items())),
"fillers_skipped_event_this_pass": fillers_skipped_event,
}
def _annotate_one(
self,
row: Mapping[str, object],
res: Mapping[str, Mapping[str, WorkResult]],
cfgs: _Configs,
np: Any, # Any: numpy module passed in (imported lazily by the caller)
soundfile: Any, # Any: untyped soundfile module
) -> dict[str, Any] | QuarantineRecord:
"""S2–S6 for one segment; returns the inputs of fusion or a quarantine record."""
from nida.data.asr_filter import asr_agreement
sid = str(row["segment_id"])
duration_ms = int(str(row["duration_ms"]))
def drop(reason: str, **details: object) -> QuarantineRecord:
return QuarantineRecord(sid, "annotate", reason, dict(details))
outputs: dict[str, Mapping[str, object]] = {}
for key, results in res.items():
output = results[sid].output
if output is None: # excluded by the caller (deferred), never quarantined
raise AssertionError(f"{sid}: {key} result has no output")
outputs[key] = output
# S2 music
sed = outputs["sed"]
sed_probs = np.load(str(sed["out_path"]), allow_pickle=False)
classes = [str(c) for c in sed["classes"]] # type: ignore[attr-defined]
music = sed_probs[:, classes.index("Music")]
music_ratio = float(np.mean(music >= cfgs.quality.music_prob))
if music_ratio > cfgs.quality.max_music_ratio:
return drop("music", music_ratio=music_ratio)
# S3 speakers
diar = outputs["diar"]
activity = np.load(str(diar["out_path"]), allow_pickle=False)
speakers = check_speakers(
activity,
frame_hop_ms=float(str(diar["frame_hop_ms"])),
duration_ms=duration_ms,
cfg=cfgs.speakers,
)
if speakers.reason is not None:
return drop(
speakers.reason, n_speakers=speakers.n_speakers, overlap_ms=speakers.overlap_ms
)
# S4 dual ASR
primary_raw = str(outputs["primary"]["text"])
secondary_raw = str(outputs["secondary"]["text"])
primary = normalise_transcript(primary_raw)
secondary = normalise_transcript(secondary_raw)
if primary is None or secondary is None:
return drop("unverbalizable_number", primary=primary_raw, secondary=secondary_raw)
language_role = self._language_role()
if language_role is None: # _require refuses such plans; kept for type narrowing
raise ValueError("no ASR system of the plan identifies the language")
# strict: a language-identifying worker always reports it (like "text" above)
language = language_code(str(outputs[language_role]["language"]))
reason = check_transcripts(primary, secondary, language, cfgs.asr)
if reason is not None:
return drop(reason, primary=primary, secondary=secondary)
# S5 alignment
ctc = outputs["ctc"]
vocab_list = [str(t) for t in ctc["vocab"]] # type: ignore[attr-defined]
vocab = {tok: i for i, tok in enumerate(vocab_list)}
emissions = np.load(str(ctc["out_path"]), allow_pickle=False)
aligned = align_words(
primary.split(),
emissions,
vocab,
blank_token=vocab_list[int(str(ctc["blank_id"]))],
word_delimiter="|",
frame_hop_ms=float(str(ctc["frame_hop_ms"])),
duration_ms=duration_ms,
)
if aligned.words is None:
return drop(f"align_{aligned.failure}", **dict(aligned.details))
words: tuple[TimedWord, ...] = aligned.words
reason2 = check_alignment(words, duration_ms, cfgs.asr)
if reason2 is not None:
# The measured values, so the provisional thresholds can be calibrated later.
measured = [w.align_score for w in words if w.align_score is not None]
return drop(
reason2,
mean_align_score=sum(measured) / len(measured) if measured else None,
n_words=len(words),
speech_ms=words[-1].end_ms - words[0].start_ms if words else 0,
)
scores = [w.align_score for w in words if w.align_score is not None]
# S6 events
frames = detect_events(
sed_probs,
classes,
frame_hop_ms=float(str(sed["frame_hop_ms"])),
duration_ms=duration_ms,
cfg=cfgs.events,
source="panns_cnn14_dlm",
)
for kind, conf in frames.too_short.items():
if conf >= cfgs.fusion.bands[kind].low:
return drop("short_event", kind=kind, conf=conf)
# Fillers (ADR-0001: the ASRs never write them; CTC gap scan, option A). A gap
# sound that an event detector also heard (a sigh read as "hmm") is the event's:
# the candidate is skipped, so it cannot turn the event into speech overlap.
fillers: list[FillerCandidate] = []
fillers_skipped_event = 0
if cfgs.fillers.enabled:
candidates = gap_filler_candidates(
words,
emissions,
vocab,
blank_token=vocab_list[int(str(ctc["blank_id"]))],
word_delimiter="|",
frame_hop_ms=float(str(ctc["frame_hop_ms"])),
duration_ms=duration_ms,
min_gap_ms=cfgs.fillers.min_gap_ms,
)
heard = [
d
for d in frames.detections
if d.kind in cfgs.fusion.bands and d.conf >= cfgs.fusion.bands[d.kind].low
]
free = [
c
for c in candidates
if not any(
d.start_ms < c.speech_end_ms and c.speech_start_ms < d.end_ms for d in heard
)
]
fillers_skipped_event = len(candidates) - len(free)
words, fillers = insert_fillers(words, free, cfgs.fillers)
# S6 acoustic styles
audio, rate = soundfile.read(
str(self.audio_root / str(row["audio_relpath"])), dtype="float64"
)
features = acoustic.word_features(audio, int(rate), words, cfgs.acoustic)
whisper = acoustic.whisper_scores(features, cfgs.acoustic)
emotion_scores = {
str(k): float(str(v))
for k, v in outputs["emotion"]["probabilities"].items() # type: ignore[attr-defined]
}
return {
"words": words,
"fillers": fillers,
"fillers_skipped_event": fillers_skipped_event,
"events": frames.detections,
"features": features,
"whisper": whisper,
"emotion_scores": emotion_scores,
"transcript": primary,
"asr_cer": asr_agreement(primary, secondary),
"align_score": sum(scores) / len(scores),
}
# ------------------------------------------------------------------ report
def _accepted(self) -> tuple[Path, list[dict[str, object]]]:
cfgs = load_configs(self.config_dir)
out_dir = self.work / f"annotate-{cfgs.digest}"
if not out_dir.exists():
raise RuntimeError(f"{out_dir} missing: run the annotate stage first")
return out_dir, manifest.read_manifest(out_dir)
def _stage_report(self) -> dict[str, object]:
out_dir, rows = self._accepted()
if not rows:
raise RuntimeError("annotate accepted no segment; see its quarantine counts")
groups = {str(r["segment_id"]): str(r["video_id"]) for r in rows}
splits = assign_splits(
groups, {"train": 0.9, "val": 0.05, "test": 0.05}, str(self.plan.seed)
)
atomic_write_json(out_dir / "splits.json", splits)
seg_stats = [
stats.SegmentStat(
duration_ms=int(str(r["duration_ms"])),
num_chars=sum(ch.isalnum() for ch in str(r["plain_transcript"])),
tags=tuple(str(t["tag"]) for t in json.loads(str(r["tags_json"]))),
emotion=None if r["emotion"] is None else str(r["emotion"]),
)
for r in rows
]
summary = stats.summarize(seg_stats, duration_bin_edges_ms=DURATION_BIN_EDGES_MS)
quarantine = {
"ingest": QuarantineLog(self.work / "ingest").counts_by_reason(),
"annotate": QuarantineLog(out_dir).counts_by_reason(),
}
report = [
"# Faz 1 pilot raporu",
"",
f"- Plan: `{json.dumps(asdict(self.plan), ensure_ascii=False)}`",
f"- Eşik dosyaları özeti: `{out_dir.name}`",
f"- Kabul edilen segment: {len(rows)}",
f"- Bölmeler (video-ayrık): {dict(Counter(splits.values()))}",
"",
"## Karantina nedenleri",
"",
*(f"- {stage}: {json.dumps(c, ensure_ascii=False)}" for stage, c in quarantine.items()),
"",
stats.render_stats_markdown(summary),
]
text = "\n".join(report) + "\n"
enh = load_enhance_config(self.config_dir)
if enh.enabled:
marker, _ = self._read_marker(enh, out_dir, rows)
if marker is not None: # the enhance stage ran before: its section belongs here
text = _with_enhance_section(text, marker)
atomic_write_text(self.root / "reports" / "pilot.md", text)
return {
"accepted": len(rows),
"splits": dict(Counter(splits.values())),
"quarantine": quarantine,
}
# ---------------------------------------------------------------- enhance
@staticmethod
def _enhance_worker_name(cfg: EnhanceConfig) -> str:
"""Results folder (and log tag) of the enhancement worker of ``cfg.arm``."""
return f"enhance-{cfg.arm}"
def _enhance_dir(self, out_dir: Path, cfg: EnhanceConfig) -> Path:
"""Durable folder of one arm over one accepted set: marker and quarantine."""
return self.work / f"enhance-{cfg.arm}-{out_dir.name.removeprefix('annotate-')}"
def _latents_name(self, cfg: EnhanceConfig) -> str:
"""Results folder / array cache name of the VAE latents (arm-specific when enhanced)."""
key = DATA_MODELS["voxcpm2_audiovae_v2"].key
return key + enhance.arm_suffix(cfg.arm) if cfg.enabled else key
def _enhance_needed(
self, cfg: EnhanceConfig, out_dir: Path, accepted: Sequence[dict[str, object]]
) -> set[str]:
"""Segments whose enhanced wav a later stage still has to read.
Enhanced wavs live in the local cache only, so a new session must not enhance what
nobody will read: while the enhanced dataset is not built, the segments whose latents
are still to compute (no result, a failed one, or a lost array); once it is built, the
dataset segments that have no timeline yet (``timings`` is the only later reader).
Segments the current config excludes anyway (a recorded length difference above the
tolerance, too much clipping, a recorded worker failure) are not needed: nothing reads
their wav. They are judged from the durable worker results with the *current* limits, not
from an earlier quarantine file, so loosening a limit brings them back into ``need``.
"""
dataset_dir = self._dataset_dir(out_dir, cfg)
if not (dataset_dir / "meta.json").exists():
rows = self._pending(self._latents_name(cfg), accepted, array_outputs=True)
need = {str(r["segment_id"]) for r in rows}
else:
done = (
set(load_timings(dataset_dir))
if (dataset_dir / TIMINGS_FILENAME).exists()
else set()
)
need = set(read_index(dataset_dir)) - done
results_path = self.root / "workers" / self._enhance_worker_name(cfg) / "results.jsonl"
if not need or not results_path.exists():
return need
results = load_results(results_path, repair=True)
threshold = self._clipping_threshold()
for sid in sorted(need):
result = results.get(sid)
if result is None or (result.output is None and self.retry_failed):
continue # not enhanced yet, or about to be retried
outcome = _enhance_outcome(sid, result, cfg, threshold, None)
if isinstance(outcome, QuarantineRecord):
need.discard(sid)
return need
def _clipping_threshold(self) -> float:
"""``acoustic.yaml`` ``clipping_threshold``: the S2 definition of a clipped sample."""
return load_configs(self.config_dir).acoustic.clipping_threshold
def _stage_enhance(self) -> dict[str, object]:
"""Enhance the accepted segments with the configured arm (``enhance.yaml``).
``arm: none`` is a no-op returning ``{"arm": "none"}``. Otherwise the arm's worker
(own isolated environment, results in ``workers/enhance-<arm>``) turns the original wav
of every accepted segment into ``<cache>/audio-enh-<arm>/<audio_relpath>`` (16 kHz mono
PCM16, the same number of samples); see :meth:`_enhance_run` for what is kept, recomputed
and quarantined. Segments that no later stage needs are not re-enhanced in a new session
(:meth:`_enhance_needed`), and when nothing is needed and the marker is current the stage
does nothing at all (``"reused": true``).
"""
cfg = load_enhance_config(self.config_dir)
if not cfg.enabled:
return {"arm": enhance.ARM_NONE}
out_dir, accepted = self._accepted()
if not accepted:
raise RuntimeError("annotate accepted no segment; nothing to enhance")
return self._enhance_run(
cfg, out_dir, accepted, self._enhance_needed(cfg, out_dir, accepted)
)
def _enhance_run(
self,
cfg: EnhanceConfig,
out_dir: Path,
accepted: Sequence[dict[str, object]],
need: set[str],
) -> dict[str, object]:
"""Run the enhancement worker so that every segment of ``need`` has its wav on disk.
The worker is asked about *all* accepted segments, so its durable results (length
difference, peak, loudness: they are read when the dataset is built) are never pruned;
it only does work for segments without a result, and a finished result whose wav is
missing from the cache is dropped, and so recomputed, only if the segment is in
``need``. The worker writes every wav whose length is within ``max_length_diff_ms``;
the pilot quarantines by ``length_tolerance_ms`` from the recorded difference (also
re-checked on finished results, so changing it needs no recomputation) and quarantines
worker errors (``enhance_failed``) and clipping beyond ``max_enhanced_clipping_ratio``
(``enhance_clipped``). Nothing is dropped silently: the quarantine file is
derived from the results on every run and ``enhanced + quarantined == accepted`` is
checked. The marker ``enhance.json`` is removed first and written last; ``latents``,
``dataset`` and ``timings`` refuse to run without one that matches the current accepted
set, arm, model pin, limits and excluded segments.
"""
model = enhance_model(cfg.arm)
name = self._enhance_worker_name(cfg)
root = self._enhanced_root(cfg)
qdir = self._enhance_dir(out_dir, cfg)
if not need and not self.retry_failed:
marker, _ = self._read_marker(cfg, out_dir, accepted)
if marker is not None:
return {**marker, "reused": True}
out_paths = {str(r["segment_id"]): str(root / str(r["audio_relpath"])) for r in accepted}
results_path = self.root / "workers" / name / "results.jsonl"
if results_path.exists():
existing = load_results(results_path, repair=True)
lost = sorted(
sid
for sid in need
if sid in existing
and existing[sid].output is not None
and not Path(out_paths[sid]).is_file()
)
if lost:
log.info(
"%s: %d enhanced wavs are gone from the cache; recomputing", name, len(lost)
)
drop_results(results_path, lost)
# Not "done" from here until the very end: a failed or interrupted run must not leave
# a marker that vouches for results it did not see.
(qdir / ENHANCE_MARKER).unlink(missing_ok=True)
max_diff_ms = str(cfg.max_length_diff_ms)
threshold = self._clipping_threshold()
if self._pending(name, accepted, array_outputs=False):
results = self._execute_job(
self._prepare_data_job(
model,
lambda sid: {
"out_path": out_paths[sid],
"max_length_diff_ms": max_diff_ms,
"clipping_threshold": repr(threshold),
},
name=name,
array_outputs=False,
rows=accepted,
# originals only for segments without a result (and recorded failures when
# they are being retried): the worker skips the other failures
extract_failed=self.retry_failed,
)
)
else:
results = self._results(name)
quarantined: list[QuarantineRecord] = []
kept: list[_Enhanced] = []
for row in sorted(accepted, key=lambda r: str(r["segment_id"])):
sid = str(row["segment_id"])
if sid not in results:
raise RuntimeError(f"{name}: no result for segment {sid}")
outcome = _enhance_outcome(
sid,
results[sid],
cfg,
threshold,
Path(out_paths[sid]) if sid in need else None,
)
if isinstance(outcome, QuarantineRecord):
quarantined.append(outcome)
else:
kept.append(outcome)
if len(kept) + len(quarantined) != len(accepted):
raise RuntimeError(
f"{name}: {len(kept)} enhanced + {len(quarantined)} quarantined != "
f"{len(accepted)} accepted segments"
)
qlog = QuarantineLog(qdir)
qlog.path.unlink(missing_ok=True) # derived from this run's results
qlog.append(quarantined)
counts = qlog.counts_by_reason()
if quarantined:
log.warning("%s: %d segments quarantined: %s", name, len(quarantined), counts)
peaks = [e.peak for e in kept]
outputs = [results[str(row["segment_id"])].output for row in accepted]
ratios = [float(str(o["clipping_ratio"])) for o in outputs if o is not None]
report: dict[str, object] = {
"arm": cfg.arm,
"source_id": model.source_id,
"model_id": model.ref.model_id,
"revision": model.ref.revision,
"annotate": out_dir.name,
"accepted": len(accepted),
"enhanced": len(kept),
"quarantined": counts,
"excluded_sha256": _ids_digest(q.item_id for q in quarantined),
"length_tolerance_ms": cfg.length_tolerance_ms,
"max_length_diff_ms": cfg.max_length_diff_ms,
"max_enhanced_clipping_ratio": cfg.max_enhanced_clipping_ratio,
"clipping_threshold": threshold,
"length_diff_samples": {
"max_abs": max((abs(e.length_diff_samples) for e in kept), default=0),
"nonzero": sum(1 for e in kept if e.length_diff_samples),
},
"clipping": {
"segments_over_full_scale": sum(1 for p in peaks if p > 1.0),
"max_peak": max(peaks, default=0.0),
"quarantined": counts.get(enhance.REASON_CLIPPED, 0),
"max_ratio": max(ratios, default=0.0),
},
"worker_runtime": _worker_runtime(e.output for e in kept),
"results": name,
"audio_root": str(root),
}
atomic_write_json(qdir / ENHANCE_MARKER, report)
self._write_enhance_report(report)
return {**report, "reused": False}
def _read_marker(
self, cfg: EnhanceConfig, out_dir: Path, accepted: Sequence[dict[str, object]]
) -> tuple[dict[str, Any] | None, str]:
"""The ``enhance.json`` marker if it vouches for the current run, else why not."""
qdir = self._enhance_dir(out_dir, cfg)
marker_path = qdir / ENHANCE_MARKER
if not marker_path.is_file():
return None, f"{marker_path} missing: run the enhance stage first"
marker = json.loads(marker_path.read_text(encoding="utf-8"))
if not isinstance(marker, dict):
return None, f"{marker_path} is not a JSON object; rerun the enhance stage"
model = enhance_model(cfg.arm)
expected: dict[str, object] = {
"arm": cfg.arm,
"model_id": model.ref.model_id,
"revision": model.ref.revision,
"annotate": out_dir.name,
"accepted": len(accepted),
"length_tolerance_ms": cfg.length_tolerance_ms,
"max_length_diff_ms": cfg.max_length_diff_ms,
"max_enhanced_clipping_ratio": cfg.max_enhanced_clipping_ratio,
"clipping_threshold": self._clipping_threshold(),
}
stale = {k: [marker.get(k), v] for k, v in expected.items() if marker.get(k) != v}
if stale:
return None, (
f"{marker_path} does not match the current run (marker, expected): {stale}; "
"rerun the enhance stage"
)
excluded = QuarantineLog(qdir).quarantined_ids()
if _ids_digest(excluded) != marker.get("excluded_sha256"):
return None, (
f"{qdir}: the quarantine file ({len(excluded)} segments) is not the one the "
f"marker recorded ({marker.get('quarantined')}); rerun the enhance stage"
)
return marker, ""
def _enhanced_view(
self, cfg: EnhanceConfig, out_dir: Path, accepted: Sequence[dict[str, object]]
) -> tuple[dict[str, Any], list[dict[str, object]]]:
"""``(marker, rows)``: the current marker and the accepted rows the enhance stage kept.
Raises:
RuntimeError: if the stage has not finished for this accepted set, arm, model pin
and limits, or the quarantine file is not the one the marker recorded.
"""
marker, problem = self._read_marker(cfg, out_dir, accepted)
if marker is None:
raise RuntimeError(problem)
excluded = QuarantineLog(self._enhance_dir(out_dir, cfg)).quarantined_ids()
return marker, [r for r in accepted if str(r["segment_id"]) not in excluded]
def _enhanced_rows(
self, cfg: EnhanceConfig, out_dir: Path, accepted: Sequence[dict[str, object]]
) -> list[dict[str, object]]:
"""The accepted rows the finished ``enhance`` stage kept (accepted minus quarantined)."""
return self._enhanced_view(cfg, out_dir, accepted)[1]
def _ensure_enhanced_audio(
self,
cfg: EnhanceConfig,
out_dir: Path,
accepted: Sequence[dict[str, object]],
ids: Sequence[str],
) -> None:
"""Make sure the enhanced wavs of ``ids`` are in the cache (a stage that reads a few)."""
by_id = {str(r["segment_id"]): str(r["audio_relpath"]) for r in accepted}
root = self._enhanced_root(cfg)
missing = {sid for sid in ids if not (root / by_id[sid]).is_file()}
if missing:
self._enhance_run(cfg, out_dir, accepted, missing)
def _enhance_meta(
self, cfg: EnhanceConfig, marker: Mapping[str, Any], num_segments: int
) -> dict[str, object]:
"""The ``enhance`` block of an enhanced dataset's ``meta.json``."""
summary_path = (
self.root / "workers" / self._enhance_worker_name(cfg) / "results.summary.json"
)
return {
"arm": cfg.arm,
"source_id": marker["source_id"],
"model_id": marker["model_id"],
"revision": marker["revision"],
"length_tolerance_ms": marker["length_tolerance_ms"],
"max_length_diff_ms": marker["max_length_diff_ms"],
"max_enhanced_clipping_ratio": marker["max_enhanced_clipping_ratio"],
"clipping_threshold": marker["clipping_threshold"],
"length_diff_samples": marker["length_diff_samples"],
"clipping": marker["clipping"],
"excluded": marker["quarantined"],
"excluded_sha256": marker["excluded_sha256"],
"loudness_source": ENHANCED_LOUDNESS_SOURCE,
"segments": num_segments,
"worker_packages": dict(read_summary(summary_path).packages),
"worker_runtime": marker["worker_runtime"],
}
def _write_enhance_report(self, marker: Mapping[str, Any]) -> None:
"""Put the enhance section into an existing ``reports/pilot.md`` (it is written by
``report``, which runs before ``enhance``; ``report`` adds the section itself when it
runs later)."""
path = self.root / "reports" / "pilot.md"
if path.exists():
text = _with_enhance_section(path.read_text(encoding="utf-8"), marker)
atomic_write_text(path, text)
def _stage_latents(self) -> dict[str, object]:
model = DATA_MODELS["voxcpm2_audiovae_v2"]
cfg = load_enhance_config(self.config_dir)
out_dir, accepted = self._accepted()
if not accepted:
raise RuntimeError("annotate accepted no segment; nothing to encode")
if not cfg.enabled:
self._data_worker(model, lambda sid: {"out_path": self._npy(model, sid)}, rows=accepted)
return {"encoded": len(accepted)}
rows = self._enhanced_rows(cfg, out_dir, accepted)
if not rows:
raise RuntimeError("the enhance stage quarantined every accepted segment")
name = self._latents_name(cfg)
if (self._dataset_dir(out_dir, cfg) / "meta.json").exists():
# The enhanced wavs and latent arrays live in the local cache only; once the dataset
# is built nothing reads them, so a new session must not enhance and encode again.
return {
"encoded": 0,
"arm": cfg.arm,
"results": name,
"skipped": "dataset already built",
}
self._data_worker(
model,
lambda sid: {"out_path": str(self.cache / "arrays" / name / f"{sid}.npy")},
name=name,
rows=rows,
audio_root=self._enhanced_root(cfg),
)
return {"encoded": len(rows), "arm": cfg.arm, "results": name}
def _stage_dataset(self) -> dict[str, object]:
"""Write the training dataset (format: nida.train.data docstring)."""
import numpy as np
from nida.utils.io import atomic_write_bytes
enh = load_enhance_config(self.config_dir)
out_dir, accepted = self._accepted()
dataset_dir = self._dataset_dir(out_dir, enh)
meta_path = dataset_dir / "meta.json"
if meta_path.exists():
meta = json.loads(meta_path.read_text(encoding="utf-8"))
if enh.enabled:
current, kept = self._enhanced_view(enh, out_dir, accepted)
recorded = meta.get("enhance", {}).get("excluded_sha256")
if meta["segments"] != len(kept) or recorded != current["excluded_sha256"]:
raise RuntimeError(
f"{dataset_dir} holds {meta['segments']} segments built from other "
"enhance results than the enhance stage keeps now; move the dataset "
"away and rerun the dataset stage"
)
return {"dataset": str(dataset_dir), "segments": meta["segments"], "reused": True}
splits_path = out_dir / "splits.json"
if not splits_path.exists():
raise RuntimeError("splits.json missing: run the report stage first")
splits = json.loads(splits_path.read_text(encoding="utf-8"))
marker: dict[str, Any] | None = None
if enh.enabled:
marker, accepted = self._enhanced_view(enh, out_dir, accepted)
latents = self._results(self._latents_name(enh))
if marker is None:
loudness = {
str(r["segment_id"]): float(str(r["loudness_dbfs"])) for r in self._ingested()
}
else: # the level of the audio the latents were made from, measured by the worker
enhanced_results = self._results(self._enhance_worker_name(enh))
loudness = {
str(r["segment_id"]): _number(
str(r["segment_id"]),
enhanced_results[str(r["segment_id"])].output or {},
"loudness_dbfs",
)
for r in accepted
}
def rate(row: Mapping[str, object]) -> float:
chars = sum(ch.isalnum() for ch in str(row["plain_transcript"]))
return chars / (int(str(row["duration_ms"])) / 1000)
train = [r for r in accepted if splits[str(r["segment_id"])] == "train"]
if not train:
raise RuntimeError("no training segment")
# Fixed edges (configs/data/dataset.yaml), not this root's quantiles: datasets of
# different roots are merged for training and must mean the same by a bucket.
cfgs = load_configs(self.config_dir)
rate_edges = list(cfgs.dataset.rate_bucket_edges_chars_per_s)
loud_edges = list(cfgs.dataset.loudness_bucket_edges_dbfs)
observed_rate = [
float(x) for x in np.quantile([rate(r) for r in train], [0.2, 0.4, 0.6, 0.8])
]
observed_loud = [
float(x)
for x in np.quantile([loudness[str(r["segment_id"])] for r in train], [1 / 3, 2 / 3])
]
index: list[dict[str, object]] = []
missing: list[str] = []
shard: list[Any] = []
shard_frames = 0
shard_no = 0
def flush() -> None:
nonlocal shard, shard_frames, shard_no
if not shard:
return
buffer = io.BytesIO()
np.save(buffer, np.concatenate(shard).astype(np.float16), allow_pickle=False)
atomic_write_bytes(dataset_dir / f"shard-{shard_no:05d}.npy", buffer.getvalue())
shard, shard_frames = [], 0
shard_no += 1
latent_dim: int | None = None
frame_hop_ms: float | None = None
for row in sorted(accepted, key=lambda r: str(r["segment_id"])):
sid = str(row["segment_id"])
result = latents.get(sid)
if result is None or result.output is None:
missing.append(sid)
continue
array = np.load(str(result.output["out_path"]), allow_pickle=False)
hop = float(str(result.output["frame_hop_ms"]))
if frame_hop_ms is None:
frame_hop_ms = hop
elif hop != frame_hop_ms:
raise ValueError(f"{sid}: frame hop {hop} != {frame_hop_ms}")
if latent_dim is None:
latent_dim = int(array.shape[1])
elif int(array.shape[1]) != latent_dim:
raise ValueError(f"{sid}: latent_dim {array.shape[1]} != {latent_dim}")
if not np.all(np.isfinite(array)) or float(np.max(np.abs(array))) > FLOAT16_MAX_ABS:
raise ValueError(f"{sid}: latents not representable in float16")
index.append(
{
"segment_id": sid,
"shard": f"shard-{shard_no:05d}.npy",
"offset": shard_frames,
"num_frames": int(array.shape[0]),
"text": str(row["text"]),
"emotion": row["emotion"],
"intensity": row["intensity"],
"rate_bucket": int(np.searchsorted(rate_edges, rate(row), side="right")),
"loudness_bucket": int(
np.searchsorted(loud_edges, loudness[sid], side="right")
),
"group": str(row["video_id"]),
"split": splits[sid],
}
)
shard.append(array)
shard_frames += int(array.shape[0])
if shard_frames >= DATASET_SHARD_FRAMES:
flush()
flush()
if missing:
raise RuntimeError(
f"{len(missing)} accepted segments have no latents (e.g. {missing[:3]}); "
"rerun the latents stage"
)
lines = "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in index)
atomic_write_text(dataset_dir / "index.jsonl", lines)
meta = {
"segments": len(index),
"frames": sum(int(str(i["num_frames"])) for i in index),
"latent_dim": latent_dim,
"frame_hop_ms": frame_hop_ms,
"rate_bucket_edges_chars_per_s": rate_edges,
"loudness_bucket_edges_dbfs": loud_edges,
"train_quantile_rate_edges": observed_rate,
"train_quantile_loudness_edges": observed_loud,
"annotate": out_dir.name,
"plan": asdict(self.plan),
}
if marker is not None:
meta["enhance"] = self._enhance_meta(enh, marker, len(accepted))
atomic_write_json(meta_path, meta)
return {"dataset": str(dataset_dir), "segments": len(index), "reused": False}
def _stage_timings(self) -> dict[str, object]:
"""Unit timelines of the dataset's segments (E11, ADR-0008) -> ``timings.jsonl``.
For every segment of ``dataset-<hash>/index.jsonl`` that has no valid line in
``timings.jsonl`` yet: re-run the CTC worker over those segments (own results folder
``workers/mpoyraz_ctc_tr-timings``; lost arrays are recomputed by the usual worker
logic, audio is extracted first), re-align the stored ``plain_transcript`` with
:func:`align_words` exactly as ``annotate`` did, re-find the CTC-gap fillers when
``fillers.yaml`` has them enabled, and build the timeline
(:func:`nida.data.timings.build_timings`). Word times are not stored by annotate,
so this is where they come from; the ASR results are not needed because
``plain_transcript`` is the aligned transcript.
Resumable: segments already in a valid ``timings.jsonl`` are kept and not touched
(an invalid file raises); the file is rewritten atomically with old and new lines.
Segments that get no timeline are written to ``<dataset>/timings_quarantine/
quarantine.jsonl`` (replaced on every run, so it always describes the segments
missing from ``timings.jsonl``). To recompute segments, delete their lines (or the
file); ``written + quarantined == segments`` is checked before returning.
"""
import numpy as np
cfgs = load_configs(self.config_dir)
out_dir, accepted = self._accepted()
dataset_dir = self._dataset_dir(out_dir, cfgs.enhance)
meta_path = dataset_dir / "meta.json"
if not meta_path.exists():
raise RuntimeError(f"{meta_path} missing: run the dataset stage first")
meta = json.loads(meta_path.read_text(encoding="utf-8"))
frame_hop_ms = float(meta["frame_hop_ms"])
index = read_index(dataset_dir)
timings_path = dataset_dir / TIMINGS_FILENAME
existing = load_timings(dataset_dir) if timings_path.exists() else {}
todo = sorted(set(index) - set(existing))
if cfgs.enhance.enabled:
# An enhanced dataset: the CTC reads the enhanced wavs of the todo segments, which a
# new session no longer has: make them here (like the panel does), then require
# that the timelines come from the very enhance results the dataset was built from
# and from segments the enhance stage kept.
known = {str(r["segment_id"]) for r in accepted}
self._ensure_enhanced_audio(
cfgs.enhance, out_dir, accepted, [sid for sid in todo if sid in known]
)
marker, accepted = self._enhanced_view(cfgs.enhance, out_dir, accepted)
if meta.get("enhance", {}).get("excluded_sha256") != marker["excluded_sha256"]:
raise RuntimeError(
f"{dataset_dir} was built from other enhance results than the enhance "
"stage keeps now; move the dataset away and rerun the dataset stage"
)
by_id = {str(r["segment_id"]): r for r in accepted}
unknown = sorted(set(index) - set(by_id))
if unknown:
raise RuntimeError(
f"{len(unknown)} dataset segments are not among the accepted (and enhanced) "
f"segments of {out_dir.name} (e.g. {unknown[:3]})"
)
log.info(
"timings: %d dataset segments, %d done, %d to do", len(index), len(existing), len(todo)
)
built = dict(existing)
quarantined: list[QuarantineRecord] = []
clamped_units = 0
if todo:
results = self._execute_job(
self._timings_ctc_job([by_id[sid] for sid in todo], cfgs.enhance)
)
for sid in todo:
output = results[sid].output
if output is None:
quarantined.append(
QuarantineRecord(
sid, "timings", "timings_ctc_failed", {"error": results[sid].error}
)
)
continue
outcome = self._timings_one(
by_id[sid],
index[sid].text,
index[sid].num_frames,
frame_hop_ms,
output,
cfgs.fillers,
np,
)
if outcome.timings is not None:
built[sid] = outcome.timings
clamped_units += outcome.clamped_units
else:
assert outcome.quarantine is not None
quarantined.append(outcome.quarantine)
if len(built) + len(quarantined) != len(index):
raise RuntimeError(
f"timings: {len(built)} timelines + {len(quarantined)} quarantined != "
f"{len(index)} dataset segments"
)
if todo:
write_timings(timings_path, built.values())
qlog = QuarantineLog(dataset_dir / "timings_quarantine")
qlog.path.unlink(missing_ok=True) # derived from this run: every missing segment is in it
qlog.append(quarantined)
counts = qlog.counts_by_reason()
if quarantined:
log.warning("timings: %d segments quarantined: %s", len(quarantined), counts)
return {
"dataset": str(dataset_dir),
"segments": len(index),
"written": len(built),
"written_this_run": len(built) - len(existing),
"reused": len(existing),
"quarantined": counts,
"units": sum(len(t.units) for t in built.values()),
"zero_length_units": sum(t.zero_length_units for t in built.values()),
"clamped_units_this_run": clamped_units,
"frame_hop_ms": frame_hop_ms,
}
@staticmethod
def _timings_one(
row: Mapping[str, object],
text: str,
num_frames: int,
frame_hop_ms: float,
ctc: Mapping[str, object],
fillers: FillerConfig,
np: Any, # Any: numpy module passed in (imported lazily by the caller)
) -> TimelineResult:
"""Re-align one dataset segment's transcript and build its timeline."""
sid = str(row["segment_id"])
vocab_list = [str(t) for t in ctc["vocab"]] # type: ignore[attr-defined]
vocab = {tok: i for i, tok in enumerate(vocab_list)}
blank = vocab_list[int(str(ctc["blank_id"]))]
ctc_hop_ms = float(str(ctc["frame_hop_ms"]))
duration_ms = int(str(row["duration_ms"]))
emissions = np.load(str(ctc["out_path"]), allow_pickle=False)
try:
aligned = align_words(
str(row["plain_transcript"]).split(),
emissions,
vocab,
blank_token=blank,
word_delimiter="|",
frame_hop_ms=ctc_hop_ms,
duration_ms=duration_ms,
)
if aligned.words is None:
record = QuarantineRecord(
sid, "timings", f"timings_align_{aligned.failure}", dict(aligned.details)
)
return TimelineResult(None, record)
words: tuple[TimedWord, ...] = aligned.words
if fillers.enabled:
# As annotate did, minus the SED filter (the SED arrays are not re-run): the
# candidates are optional words for the matcher, see timings.spoken_words.
candidates = gap_filler_candidates(
words,
emissions,
vocab,
blank_token=blank,
word_delimiter="|",
frame_hop_ms=ctc_hop_ms,
duration_ms=duration_ms,
min_gap_ms=fillers.min_gap_ms,
)
words, _ = insert_fillers(words, candidates, fillers)
except ValueError as err:
record = QuarantineRecord(sid, "timings", "timings_align_error", {"error": str(err)})
return TimelineResult(None, record)
return build_timings(
sid,
text,
str(row["tags_json"]),
words,
num_frames=num_frames,
frame_hop_ms=frame_hop_ms,
)
def _stage_panel(self) -> dict[str, object]:
import soundfile # untyped, see above
enh = load_enhance_config(self.config_dir)
out_dir, accepted = self._accepted()
# With an enhancement arm the panel plays what the dataset was trained on (the enhanced
# wavs of the segments the enhance stage kept) and carries the arm in its id; the tags
# themselves were decided on the original audio (annotate).
rows = self._enhanced_rows(enh, out_dir, accepted) if enh.enabled else accepted
audio_root = self._enhanced_root(enh) if enh.enabled else self.audio_root
candidates = [
tag_panel.PanelCandidate(
segment_id=str(r["segment_id"]),
text=str(r["text"]),
tags=tuple(json.loads(str(r["tag_ids_json"]))),
audio_path=audio_root / str(r["audio_relpath"]),
info="" if r["emotion"] is None else f"duygu: {r['emotion']}",
)
for r in rows
]
chosen = tag_panel.select_items(candidates, self.plan.panel_per_tag, str(self.plan.seed))
if not chosen:
log.warning("panel: no accepted segment carries a tag; no panel written")
return {"panel": None, "items": 0, "reason": "no accepted segment carries a tag"}
wanted = {cand.segment_id for _, cand in chosen}
if enh.enabled:
self._ensure_enhanced_audio(enh, out_dir, accepted, sorted(wanted))
else:
self._extract(self._utts_of([r for r in self._ingested() if r["segment_id"] in wanted]))
suffix = enhance.arm_suffix(enh.arm) if enh.enabled else ""
panel_id = f"pilot-tags-{out_dir.name.removeprefix('annotate-')}{suffix}"
flac_dir = self.cache / "panel_audio" / panel_id
def to_flac(cand: tag_panel.PanelCandidate) -> Path:
dest = flac_dir / f"{cand.segment_id}.flac"
if not dest.exists():
audio, rate = soundfile.read(str(cand.audio_path), dtype="float64")
dest.parent.mkdir(parents=True, exist_ok=True)
soundfile.write(str(dest), audio, int(rate), subtype="PCM_16")
return dest
spec, key = tag_panel.build_tag_panel(panel_id, chosen, to_flac)
html = tag_panel_write(spec, self.root / "panels" / panel_id)
atomic_write_json(self.root / "manifests" / f"{panel_id}-key.json", key)
return {"panel": str(html), "items": len(key)}
def _stage_fillers(self) -> dict[str, object]:
"""Filler candidates in alignment gaps (docs/research/dolgu-tespiti-oneri.md, A).
Measurement only: re-aligns each accepted segment's transcript, proposes CTC gap
fillers, counts them and writes a listening panel of up to
:data:`FILLER_PANEL_ITEMS` candidates. The dataset is not changed.
"""
import numpy as np
import soundfile # untyped, see above
from nida.bench.panel import AudioRef, PanelItem, PanelSpec, Question
from nida.data.fillers_ctc import gap_filler_candidates
out_dir, rows = self._accepted()
ctc_results = self._results(DATA_MODELS["mpoyraz_ctc_tr"].key)
skipped: Counter[str] = Counter()
found: list[tuple[Mapping[str, object], Any, list[str], list[Any]]] = []
for row in rows:
sid = str(row["segment_id"])
result = ctc_results.get(sid)
if result is None or result.output is None:
skipped["no_ctc_result"] += 1
continue
ctc = result.output
if not Path(str(ctc["out_path"])).is_file():
skipped["ctc_array_not_cached"] += 1
continue
vocab_list = [str(t) for t in ctc["vocab"]] # type: ignore[attr-defined]
vocab = {tok: i for i, tok in enumerate(vocab_list)}
emissions = np.load(str(ctc["out_path"]), allow_pickle=False)
blank = vocab_list[int(str(ctc["blank_id"]))]
hop = float(str(ctc["frame_hop_ms"]))
duration_ms = int(str(row["duration_ms"]))
tokens = str(row["plain_transcript"]).split()
aligned = align_words(
tokens,
emissions,
vocab,
blank_token=blank,
word_delimiter="|",
frame_hop_ms=hop,
duration_ms=duration_ms,
)
if aligned.words is None:
skipped[f"realign_{aligned.failure}"] += 1
continue
candidates = gap_filler_candidates(
aligned.words,
emissions,
vocab,
blank_token=blank,
word_delimiter="|",
frame_hop_ms=hop,
duration_ms=duration_ms,
)
if candidates:
found.append((row, aligned.words, tokens, candidates))
by_text = Counter(c.text for _, _, _, cands in found for c in cands)
stats: dict[str, object] = {
"segments": len(rows),
"segments_with_candidates": len(found),
"candidates": sum(by_text.values()),
"by_filler": dict(sorted(by_text.items())),
"skipped": dict(sorted(skipped.items())),
}
flat = [(row, words, cand) for row, words, _, cands in found for cand in cands]
flat.sort(key=lambda x: _rank(self.plan.seed, f"{x[0]['segment_id']}:{x[2].start_ms}"))
chosen = flat[:FILLER_PANEL_ITEMS]
panel_id = f"pilot-fillers-{out_dir.name.removeprefix('annotate-')}"
atomic_write_json(self.root / "manifests" / f"{panel_id}-stats.json", stats)
if not chosen:
return {**stats, "panel": None}
wanted = {str(row["segment_id"]) for row, _, _ in chosen}
self._extract(self._utts_of([r for r in self._ingested() if r["segment_id"] in wanted]))
flac_dir = self.cache / "panel_audio" / panel_id
items: list[PanelItem] = []
key: dict[str, dict[str, object]] = {}
for index, (chosen_row, words, cand) in enumerate(chosen):
sid = str(chosen_row["segment_id"])
dest = flac_dir / f"{sid}.flac"
if not dest.exists():
audio, rate = soundfile.read(
str(self.audio_root / str(chosen_row["audio_relpath"])), dtype="float64"
)
dest.parent.mkdir(parents=True, exist_ok=True)
soundfile.write(str(dest), audio, int(rate), subtype="PCM_16")
before = [w.text for w in words if w.end_ms <= cand.start_ms]
after = [w.text for w in words if w.start_ms >= cand.end_ms]
shown = " ".join([*before, f"⟦{cand.text}?⟧", *after])
item_id = f"f{index:04d}"
key[item_id] = {
"segment_id": sid,
"filler": cand.text,
"start_ms": cand.start_ms,
"end_ms": cand.end_ms,
"speech_start_ms": cand.speech_start_ms,
"speech_end_ms": cand.speech_end_ms,
"decoded": cand.decoded,
"conf": cand.conf,
}
items.append(
PanelItem(
item_id=item_id,
audios=(AudioRef("Ses", dest),),
text=shown,
info=(
f"Aday dolgu: {cand.text} · {cand.speech_start_ms / 1000:.2f}–"
f"{cand.speech_end_ms / 1000:.2f} s · CTC: {cand.decoded} "
f"({cand.conf:.2f})"
),
)
)
spec = PanelSpec(
panel_id=panel_id,
title=f"Nida dolgu adayları — {panel_id}",
instructions=(
"Her kartta bir ses ve metin var; ⟦…?⟧ işareti modelin dolgu sesi "
"(ıı, ee, hmm…) bulduğunu düşündüğü yeri ve saniye aralığını gösteriyor. "
"O aralıkta gerçekten böyle bir dolgu sesi varsa 'evet', yoksa (sessizlik, "
"nefes, kelimenin parçası…) 'hayır', karar veremiyorsan 'emin değilim'. "
"Dolgu var ama türü farklıysa (ör. ıı yerine ee) 'evet' de ve nota yaz. "
"Bitince 'CSV indir'."
),
questions=(
Question(
"correct",
"İşaretli yerde gerçekten bir dolgu sesi var mı?",
"choice",
(tag_panel.ANSWER_YES, tag_panel.ANSWER_NO, tag_panel.ANSWER_UNSURE),
),
),
items=tuple(items),
)
html = tag_panel_write(spec, self.root / "panels" / panel_id)
atomic_write_json(self.root / "manifests" / f"{panel_id}-key.json", key)
return {**stats, "panel": str(html), "items": len(items)}
def _stage_fillers_whisper(self) -> dict[str, object]:
"""Does prompted Whisper write real fillers? (experiment, dataset unchanged)
Runs Whisper large-v3 with the decoder prompt :data:`FILLER_PROMPT` over
:data:`FILLER_WHISPER_SAMPLE` accepted segments (seeded ``_rank`` order), reads the
fillers it wrote, counts them and writes a listening panel of up to
:data:`FILLER_PANEL_ITEMS` segments for the user to confirm them by ear. Results
live in ``workers/asr-whisper_large_v3-fillerprompt``. Segments whose transcript
repeats the prompt (:func:`echoes_prompt`) are counted as ``prompt_echo`` and neither
counted as filler finds nor put on the panel.
"""
import soundfile # untyped, see above
from nida.bench.panel import AudioRef, PanelItem, PanelSpec, Question
system = ASR_SYSTEMS[FILLER_WHISPER_SYSTEM]
require_approved({system.source_id}, load_sources(self.sources_path))
out_dir, rows = self._accepted()
if not rows:
raise RuntimeError("annotate accepted no segment; nothing to transcribe")
sample = sorted(rows, key=lambda r: _rank(self.plan.seed, str(r["utt_id"])))[
:FILLER_WHISPER_SAMPLE
]
name = FILLER_WHISPER_WORKER
prompt_file = self.root / "workers" / name / "prompt.txt"
if prompt_file.exists():
used = prompt_file.read_text(encoding="utf-8")
if used != FILLER_PROMPT:
raise RuntimeError(
f"{name}: existing results were made with prompt {used!r}, not "
f"{FILLER_PROMPT!r}; move them away first"
)
self._extract(self._utts_of(self._pending(name, sample, array_outputs=False)))
python = bench_envs.ensure_env(
bench_envs.env_for_worker(system.worker), self.envs_root, self.repo_dir, uv=self._uv()
)
requests = [
WorkRequest(
str(r["segment_id"]),
str(self.audio_root / str(r["audio_relpath"])),
{"language": LANGUAGE, "prompt": FILLER_PROMPT},
)
for r in sample
]
atomic_write_text(prompt_file, FILLER_PROMPT)
results = self._run_worker(
name,
python,
f"nida.bench.workers.{system.worker}",
system.model_id,
system.revision,
self.asr_batch_size or system.batch_size,
requests,
array_outputs=False,
)
prompt_tokens = filler_tokens(FILLER_PROMPT)
examined = failed = echoes = prefix_stripped = primary_with_fillers = 0
by_filler: Counter[str] = Counter()
hits: list[tuple[Mapping[str, object], list[str], list[WhisperFiller]]] = []
for row in sample:
sid = str(row["segment_id"])
result = results.get(sid)
if result is None or result.output is None:
failed += 1
continue
if "text" not in result.output:
raise RuntimeError(f"{name}: payload of {sid} has no 'text'")
examined += 1
prefix_stripped += result.output.get("prompt_stripped") is True
primary_tokens = filler_tokens(str(row["plain_transcript"]))
primary_with_fillers += any(canonical_filler(t) is not None for t in primary_tokens)
tokens = filler_tokens(str(result.output["text"]))
if echoes_prompt(tokens, prompt_tokens):
echoes += 1
continue
fillers = find_whisper_fillers(tokens)
by_filler.update(f.filler for f in fillers)
if any(f.filler in HESITATION_FILLERS for f in fillers):
hits.append((row, tokens, fillers))
summary: dict[str, object] = {
"prompt": FILLER_PROMPT,
"accepted_segments": len(rows),
"sampled": len(sample),
"examined": examined,
"failed": failed,
"prompt_echo": echoes,
"prompt_prefix_stripped": prefix_stripped,
"segments_with_hesitation_fillers": len(hits),
"hesitation_fillers": sum(n for f, n in by_filler.items() if f in HESITATION_FILLERS),
"note": (
"sample = accepted segments only (ASR disagreement drops, a typical place "
"for fillers, are excluded): a lower bound; no control items"
),
"whisper_fillers": sum(by_filler.values()),
"by_filler": dict(sorted(by_filler.items())),
"primary_transcripts_with_fillers": primary_with_fillers,
}
panel_id = f"pilot-whisperfillers-{out_dir.name.removeprefix('annotate-')}"
atomic_write_json(self.root / "manifests" / f"{panel_id}-stats.json", summary)
chosen = hits[:FILLER_PANEL_ITEMS]
if not chosen:
return {**summary, "panel": None}
wanted = {str(row["segment_id"]) for row, _, _ in chosen}
self._extract(self._utts_of([r for r in self._ingested() if r["segment_id"] in wanted]))
flac_dir = self.cache / "panel_audio" / panel_id
items: list[PanelItem] = []
key: dict[str, dict[str, object]] = {}
for index, (hit_row, tokens, fillers) in enumerate(chosen):
sid = str(hit_row["segment_id"])
dest = flac_dir / f"{sid}.flac"
if not dest.exists():
audio, rate = soundfile.read(
str(self.audio_root / str(hit_row["audio_relpath"])), dtype="float64"
)
dest.parent.mkdir(parents=True, exist_ok=True)
soundfile.write(str(dest), audio, int(rate), subtype="PCM_16")
item_id = f"w{index:04d}"
key[item_id] = {
"segment_id": sid,
"whisper_text": " ".join(tokens),
"primary_transcript": str(hit_row["plain_transcript"]),
"fillers": [asdict(f) for f in fillers],
}
items.append(
PanelItem(
item_id=item_id,
audios=(AudioRef("Ses", dest),),
text=mark_fillers(tokens),
info=(
f"Ana ASR: {hit_row['plain_transcript']} · "
f"süre {int(str(hit_row['duration_ms'])) / 1000:.1f} s"
),
)
)
spec = PanelSpec(
panel_id=panel_id,
title=f"Nida Whisper dolguları — {panel_id}",
instructions=(
"Her kartta bir ses ve Whisper'ın yazdığı metin var; ⟦…?⟧ işaretli sözcükler "
"Whisper'ın yazdığı dolgu sesleri (ıı, ee, hmm…). Altta karşılaştırma için "
"ana ASR'nin metni görünüyor. İşaretli dolgular seste yaklaşık o "
"yerde gerçekten duyuluyorsa 'evet', duyulmuyorsa 'hayır', karar "
"veremiyorsan 'emin değilim'. Bazıları var bazıları yoksa 'hayır' de ve "
"nota yaz. Bitince 'CSV indir'."
),
questions=(
Question(
"present",
"İşaretli dolgu sesleri gerçekten var mı (yaklaşık o yerde)?",
"choice",
(tag_panel.ANSWER_YES, tag_panel.ANSWER_NO, tag_panel.ANSWER_UNSURE),
),
),
items=tuple(items),
)
html = tag_panel_write(spec, self.root / "panels" / panel_id)
atomic_write_json(self.root / "manifests" / f"{panel_id}-key.json", key)
return {**summary, "panel": str(html), "items": len(items)}
def tag_panel_write(spec: Any, out_dir: Path) -> Path: # Any: PanelSpec (kept import-light)
from nida.bench.panel import write_panel
return write_panel(spec, out_dir)
"""Phase 1 pilot runner behind ``nida data-pilot`` (YODAS, ~10 h, end to end on Colab).
Stages (each idempotent and resumable; ``all`` runs them in order in one session):
check → ingest → sed → diar → asr → ctc → emotion → annotate → report → enhance →
latents → dataset → panel
* ``ingest`` (CPU): YODAS listing → deterministic selection up to ``max_hours`` →
streaming extraction of the selected wavs into the local cache → S1/S2 checks
(:mod:`nida.data.ingest`) → ``work/ingest`` manifest + quarantine.
* ``sed`` / ``diar`` / ``ctc`` / ``emotion`` / ``asr`` (GPU): model workers in isolated
environments over the ingested segments. Array outputs live in the *local* cache; a
result whose array is gone (new Colab session) is dropped and recomputed.
* ``annotate`` (CPU): S2 music, S3 speakers, S4 dual ASR, S5 alignment, S6 events/styles/
emotion, S7 fusion → ``work/annotate-<config hash>`` manifest + quarantine. Changing any
threshold file gives a new directory, so tuning never overwrites earlier outputs.
* ``report``: S8 statistics, S10 video-disjoint splits, ``reports/pilot.md``.
* ``enhance`` (GPU, optional): speech enhancement of the accepted segments by the arm in
``configs/data/enhance.yaml`` (``uvr`` | ``sidon`` | ``dfn``, :mod:`nida.data.enhance`);
a no-op returning ``{"arm": "none"}`` for ``none`` (the baseline). Reads the original wav,
writes a 16 kHz mono PCM16 wav of the *same number of samples* to
``<cache>/audio-enh-<arm>/<same relpath>`` and records, per segment, the pre-fix length
difference, the peak and the loudness of the written samples. The worker only refuses absurd
length differences (``max_length_diff_ms``); the pilot quarantines by ``length_tolerance_ms``
and quarantines worker errors, into ``work/enhance-<arm>-<annotate hash>/quarantine.jsonl``,
excluding them from ``latents``, ``dataset``, ``timings`` and ``panel``. Only segments a later
stage still reads are re-enhanced in a new session (the wavs are local-cache only), and nothing
at all when none is needed. Enhancement changes neither labels nor splits, so ``annotate`` and
``report`` (and their hash) are shared with the baseline and always see the original audio
(tags were decided on it); ``report`` gains an enhance section.
* ``latents`` (GPU, S9): VAE latents of the accepted segments (frozen AudioVAE V2). With an
enhancement arm the *enhanced* wavs are encoded, into their own results folder and array
cache (``voxcpm2_audiovae_v2-enh-<arm>``), so baseline and enhanced latents never mix.
* ``dataset`` (CPU): the Phase 2 training dataset ``dataset-<config hash>/``: float16 latent
shards, ``index.jsonl`` (text, emotion, intensity, rate/loudness buckets, group, split)
and ``meta.json`` (written last; its presence marks the stage done). With an enhancement
arm the directory is ``dataset-<config hash>-enh-<arm>/``, the loudness buckets come from the
enhanced audio, and ``meta.json`` gains an ``enhance`` block (arm, model id, revision, limits,
excluded segments and their digest, clipping summary, ``loudness_source``, worker packages and
runtime); without an arm nothing about the dataset differs from the pre-enhancement pipeline.
* ``panel``: per-tag precision listening panel (:mod:`nida.data.tag_panel`); with an enhancement
arm it plays the enhanced audio and its id carries ``-enh-<arm>``.
* ``timings`` (extra stage, GPU: CTC only): the unit timeline of every dataset segment for the
aligned text path (E11, ADR-0008) -> ``dataset-<hash>/timings.jsonl``
(:mod:`nida.data.timings`). For an enhanced dataset the CTC runs on the enhanced audio.
``all`` normally runs the stages one after another. With ``parallel_workers > 1`` the
independent worker stages (``sed``, ``diar``, both ``asr`` systems, ``ctc``, ``emotion``)
run concurrently as separate subprocesses (see :meth:`Pilot._run_worker_group`).
Durable outputs (manifests, worker results, reports, panels) are on Drive under
``root``; audio and arrays are in ``cache`` (local disk) and are rebuilt when missing.
"""
from __future__ import annotations
import functools
import hashlib
import io
import json
import logging
import math
import shutil
from collections import Counter, defaultdict
from collections.abc import Callable, Iterable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict, dataclass, fields
from itertools import pairwise
from pathlib import Path
from typing import TYPE_CHECKING, Any
from nida.bench import envs as bench_envs
from nida.bench import fetch
from nida.bench.b1 import LOG_TAG, MAX_ERROR_FRACTION, _stream_runner
from nida.bench.clips import YodasUtt, yodas_video_id
from nida.bench.prepare import parse_yodas_listing
from nida.bench.protocol import (
WorkRequest,
WorkResult,
drop_results,
load_results,
read_summary,
write_requests,
)
from nida.bench.registry import ASR_SYSTEMS, DATA_PINS, System, load_sources, require_approved
from nida.bench.textmetrics import normalize_for_scoring
from nida.data import acoustic, enhance, envs, ingest, manifest, stats, tag_panel
from nida.data.align import align_words
from nida.data.asr_filter import AsrFilterConfig, check_alignment, check_transcripts
from nida.data.enhance import NO_ENHANCE, EnhanceConfig
from nida.data.events import EventConfig, detect_events
from nida.data.fillers_ctc import (
FillerCandidate,
FillerConfig,
gap_filler_candidates,
insert_fillers,
)
from nida.data.fusion import FusionConfig, emotion_condition, fuse_segment
from nida.data.quarantine import QuarantineLog
from nida.data.registry import DATA_MODELS, DataModel, enhance_model