"""EARS -> latent dataset builder (ADR-0007, ``nida data-ears``).
EARS (Expressive Anechoic Recordings of Speech, CC BY-NC 4.0) provides English emotional,
whispered and non-verbal speech. This module turns the selected files into a *latent dataset
directory* in exactly the format of :mod:`nida.train.data` (records carry the optional key
``"language": "en"``), so it can be trained together with Turkish datasets
(``train --data A --data B``).
What is used (rules and mappings live in ``configs/data/ears.yaml``):
* ``emo_<emotion>_sentences``: reading transcript, emotion mapped (unmapped -> ``null``);
* ``rainbow_NN_<style>`` / ``sentences_NN_<style>``: ``whisper`` -> the whole text wrapped in a
whisper span (rendered by :mod:`nida.text.tags`), the plain styles -> plain text;
* laughter/crying/screaming/coughing/throat-clearing files: just the event tag;
* everything else is skipped and counted by reason.
Per speaker (one remote zip of ~564 MB, read with HTTP Range requests, see
:mod:`nida.data.remote_zip`): list the archive -> decide on every member (skip / text
quarantine, before any download) -> extract the selected members, convert 48 kHz -> 16 kHz mono
-> measure. The speakers are handled in *waves* of ``speakers_per_wave``: one AudioVAE run
(the same ``vae_encode`` worker and environment as the pilot, so the model is loaded once per
wave, not once per speaker) encodes the wave's audio, then every speaker of the wave gets its
float16 shard, index records and ``done.json``. Layout::
<root>/plan.json seed, config digest, VAE pin, transcripts SHA-256
<root>/source/transcripts.json pinned reading transcripts
<root>/speakers/pNNN/records.jsonl index rows of the speaker
<root>/speakers/pNNN/quarantine.jsonl every quarantined item with its reason
<root>/speakers/pNNN/done.json written last: the speaker is finished
<root>/workers/vae/ worker requests/results of the current wave
<root>/dataset/ latent dataset: index.jsonl, meta.json, shard-pNNN.npy
<root>/report.md, <root>/stats.json
<cache>/... raw members, 16 kHz wavs and latent arrays (local disk)
Resumable and idempotent: a speaker with ``done.json`` is skipped without any network access, an
interrupted wave continues (worker results are kept, already converted audio is reused), and
the dataset directory is reassembled from all finished speakers at the end of every run, so
it is usable after any run. A run on a root made with another seed, config digest or VAE pin,
or with different transcripts, is refused.
"""
from __future__ import annotations
import hashlib
import io
import json
import logging
import math
import re
import shutil
import string
import time
from bisect import bisect_right
from collections import Counter
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Literal
import numpy as np
import numpy.typing as npt
# scipy ships no type stubs; the ignore is scoped to this single import (as in acoustic.py).
from scipy import signal # type: ignore[import-untyped]
from nida.bench import envs as bench_envs
from nida.bench import fetch
from nida.bench.b1 import MAX_ERROR_FRACTION, _stream_runner
from nida.bench.protocol import (
WorkRequest,
WorkResult,
drop_results,
load_results,
read_summary,
write_requests,
)
from nida.bench.registry import load_sources, require_approved
from nida.data import acoustic, envs
from nida.data.ingest import IngestConfig
from nida.data.pilot import DatasetConfig
from nida.data.quarantine import QuarantineLog
from nida.data.registry import DATA_MODELS
from nida.data.remote_zip import RemoteZip, ZipMember
from nida.data.splits import assign_splits
from nida.data.types import QuarantineRecord, segment_id
from nida.model.vocab import encode_text
from nida.text import tags
from nida.text.turkish import nfc
from nida.train.data import (
EMOTIONS,
REJECT_TEXT_TOO_LONG,
SPLITS,
DataFormatError,
load_latent_dataset,
)
from nida.utils.io import atomic_write_bytes, atomic_write_json, atomic_write_text
__all__ = [
"SOURCE_ID",
"AudioFacts",
"Ears",
"EarsConfig",
"EarsSettings",
"ItemRejected",
"PlannedItem",
"Selection",
"SelectionPlan",
"Skip",
"TextRejected",
"bucket_of",
"build_text",
"classify_stem",
"convert_to_target_rate",
"en_lower",
"load_settings",
"measure_audio",
"normalise_english",
"parse_speakers",
"rate_chars_per_s",
"resample",
"select_members",
"speaker_split",
]
log = logging.getLogger(__name__)
SOURCE_ID = "ears"
STAGE = "ears"
LANGUAGE = "en"
VAE_KEY = "voxcpm2_audiovae_v2"
DEFAULT_SEED = 20261009
FLOAT16_MAX_ABS = 6.0e4 # latents beyond this would overflow the float16 shards (as in pilot.py)
CONFIG_FILE = "ears.yaml"
INGEST_FILE = "ingest.yaml" # duration limits shared with the Turkish pilot
DATASET_FILE = "dataset.yaml" # fixed condition-bucket edges
_ASCII_LOWER = str.maketrans(string.ascii_uppercase, string.ascii_lowercase)
_SHA256_RE = re.compile(r"[0-9a-f]{64}")
_SPEAKER_RE = re.compile(r"p(\d{3})")
_WORD_RE = re.compile(r"[a-z]+")
_STEM_RE = re.compile(r"[a-z0-9_]+")
_EMO_RE = re.compile(r"emo_([a-z]+)_(sentences|freeform)")
_READ_RE = re.compile(r"([a-z]+)_(\d+)_([a-z]+)")
_TAG_CHARS = frozenset("[]<>")
Kind = Literal["emotion", "read", "whisper", "event"]
FloatArray = npt.NDArray[np.float64]
# --------------------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------------------
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 _non_negative_int(raw: Mapping[str, object], key: str) -> int:
value = raw[key]
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise ValueError(f"{key} must be a non-negative int, got {value!r}")
return value
def _positive_int(raw: Mapping[str, object], key: str) -> int:
value = raw[key]
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{key} must be a positive int, got {value!r}")
return value
def _fraction(raw: Mapping[str, object], key: str) -> float:
value = raw[key]
if isinstance(value, bool) or not isinstance(value, int | float):
raise ValueError(f"{key} must be a number, got {value!r}")
if not (math.isfinite(value) and 0.0 <= value < 1.0):
raise ValueError(f"{key} must be in [0, 1), got {value!r}")
return float(value)
def _words(raw: Mapping[str, object], key: str) -> tuple[str, ...]:
value = raw[key]
if (
not isinstance(value, list)
or not value
or not all(isinstance(x, str) and _WORD_RE.fullmatch(x) for x in value)
or len(set(value)) != len(value)
):
raise ValueError(f"{key} must be a non-empty list of distinct lowercase words")
return tuple(value)
def _str_map(raw: Mapping[str, object], key: str) -> dict[str, str]:
value = raw[key]
if not isinstance(value, dict) or not all(
isinstance(k, str) and isinstance(v, str) for k, v in value.items()
):
raise ValueError(f"{key} must be a mapping of strings")
return dict(value)
@dataclass(frozen=True)
class EarsConfig:
"""``configs/data/ears.yaml`` (all keys required; see the file for the meaning)."""
transcripts_url: str
transcripts_sha256: str | None
archive_url_template: str
n_speakers: int
expected_sample_rate_hz: int
target_sample_rate_hz: int
val_speaker_fraction: float
test_speaker_fraction: float
speakers_per_wave: int
read_prefixes: tuple[str, ...]
whisper_styles: tuple[str, ...]
plain_styles: tuple[str, ...]
max_read_per_speaker: int
emotion_map: dict[str, str] # plain dicts (not frozen views): ``asdict`` feeds the digest
event_map: dict[str, str]
char_map: dict[str, str]
def __post_init__(self) -> None:
for name in ("transcripts_url", "archive_url_template"):
if not getattr(self, name).lower().startswith(("http://", "https://")):
raise ValueError(f"{name} must be an http(s) URL")
if "{speaker}" not in self.archive_url_template:
raise ValueError("archive_url_template must contain {speaker}")
if self.transcripts_sha256 is not None and not _SHA256_RE.fullmatch(
self.transcripts_sha256
):
raise ValueError("transcripts_sha256 must be null or 64 lowercase hex characters")
if self.val_speaker_fraction + self.test_speaker_fraction >= 1.0:
raise ValueError("val and test speaker fractions must leave room for train")
if set(self.whisper_styles) & set(self.plain_styles):
raise ValueError("a style cannot be both whisper and plain")
for emotion, mapped in self.emotion_map.items():
if not _WORD_RE.fullmatch(emotion) or mapped not in EMOTIONS:
raise ValueError(f"emotion_map {emotion!r}: {mapped!r} is not one of {EMOTIONS}")
for stem, event in self.event_map.items():
if not _STEM_RE.fullmatch(stem) or event not in tags.EVENTS:
raise ValueError(f"event_map {stem!r}: {event!r} is not a tag-spec event id")
for char, replacement in self.char_map.items():
if len(char) != 1 or not replacement:
raise ValueError(f"char_map entries map one character to a string: {char!r}")
@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> EarsConfig:
expected = set(cls.__dataclass_fields__)
if set(raw) != expected:
raise ValueError(
f"ears config keys: missing={sorted(expected - set(raw))} "
f"unknown={sorted(set(raw) - expected)}"
)
for key in ("transcripts_url", "archive_url_template"):
if not isinstance(raw[key], str):
raise ValueError(f"{key} must be a string")
sha = raw["transcripts_sha256"]
if sha is not None and not isinstance(sha, str):
raise ValueError("transcripts_sha256 must be null or a string")
return cls(
transcripts_url=str(raw["transcripts_url"]),
transcripts_sha256=sha,
archive_url_template=str(raw["archive_url_template"]),
n_speakers=_positive_int(raw, "n_speakers"),
expected_sample_rate_hz=_positive_int(raw, "expected_sample_rate_hz"),
target_sample_rate_hz=_positive_int(raw, "target_sample_rate_hz"),
val_speaker_fraction=_fraction(raw, "val_speaker_fraction"),
test_speaker_fraction=_fraction(raw, "test_speaker_fraction"),
speakers_per_wave=_positive_int(raw, "speakers_per_wave"),
read_prefixes=_words(raw, "read_prefixes"),
whisper_styles=_words(raw, "whisper_styles"),
plain_styles=_words(raw, "plain_styles"),
max_read_per_speaker=_non_negative_int(raw, "max_read_per_speaker"),
emotion_map=_str_map(raw, "emotion_map"),
event_map=_str_map(raw, "event_map"),
char_map=_str_map(raw, "char_map"),
)
def archive_url(self, speaker: str) -> str:
return self.archive_url_template.replace("{speaker}", speaker)
@dataclass(frozen=True)
class EarsSettings:
"""Everything the builder reads from ``configs/data``."""
ears: EarsConfig
min_duration_ms: int
max_duration_ms: int
dataset: DatasetConfig
digest: str
def load_settings(config_dir: Path) -> EarsSettings:
"""Load ``ears.yaml`` plus the duration limits (``ingest.yaml``) and bucket edges
(``dataset.yaml``) it shares with the Turkish pilot.
``digest`` names everything a finished speaker depends on; it ignores
``transcripts_sha256`` so that pinning the hash in the config later is allowed.
"""
ears = EarsConfig.from_mapping(_load_yaml(config_dir / CONFIG_FILE))
ingest_cfg = IngestConfig.from_mapping(_load_yaml(config_dir / INGEST_FILE))
dataset = DatasetConfig.from_mapping(_load_yaml(config_dir / DATASET_FILE))
body = asdict(ears)
del body["transcripts_sha256"]
del body["speakers_per_wave"] # operational: does not change any output
canonical = json.dumps(
{
"ears": body,
"min_duration_ms": ingest_cfg.min_duration_ms,
"max_duration_ms": ingest_cfg.max_duration_ms,
"dataset": asdict(dataset),
},
sort_keys=True,
ensure_ascii=False,
)
return EarsSettings(
ears=ears,
min_duration_ms=ingest_cfg.min_duration_ms,
max_duration_ms=ingest_cfg.max_duration_ms,
dataset=dataset,
digest=hashlib.sha1(canonical.encode("utf-8")).hexdigest()[:12],
)
def parse_speakers(spec: str, n_speakers: int) -> list[str]:
"""``"all"`` or ``"p001,p007"`` -> speaker ids.
Raises:
ValueError: on an empty list, a malformed or out-of-range id, or a repeated id.
"""
if spec.strip() == "all":
return [f"p{i:03d}" for i in range(1, n_speakers + 1)]
ids = [part.strip() for part in spec.split(",")]
for speaker in ids:
match = _SPEAKER_RE.fullmatch(speaker)
if match is None or not 1 <= int(match.group(1)) <= n_speakers:
raise ValueError(f"invalid speaker {speaker!r}: use p001..p{n_speakers:03d} or 'all'")
if len(set(ids)) != len(ids):
raise ValueError(f"repeated speaker in {spec!r}")
return ids
# --------------------------------------------------------------------------------------
# Text
# --------------------------------------------------------------------------------------
class TextRejected(ValueError):
"""The text of an item cannot be used; ``reason`` is the quarantine reason."""
def __init__(self, reason: str, detail: str) -> None:
super().__init__(f"{reason}: {detail}")
self.reason = reason
self.detail = detail
def en_lower(text: str) -> str:
"""Lower-case ``A``-``Z`` only (English). Never use ``tr_lower`` (``I`` -> ``ı``) or
``str.lower`` (``İ`` -> ``i`` + combining dot) on English text: every other character is
returned unchanged."""
return text.translate(_ASCII_LOWER)
def normalise_english(text: str, char_map: Mapping[str, str]) -> str:
"""NFC, ``char_map`` replacements, ASCII lower-case and collapsed whitespace.
Returns ``""`` for text without any non-space character.
"""
mapped = nfc(text).translate({ord(k): v for k, v in char_map.items()})
return " ".join(en_lower(mapped).split())
@dataclass(frozen=True)
class BuiltText:
"""Model text (tag-spec surface) and the plain words the speaker says (may be empty)."""
tagged: str
plain: str
def build_text(selection: Selection, transcript: str | None, cfg: EarsConfig) -> BuiltText:
"""Build the training text of one selected file.
Raises:
TextRejected: with reason ``no_transcript``, ``empty_text``, ``digits``,
``tag_characters``, ``non_ascii_letter`` or ``text_encoding`` (the model
vocabulary cannot encode the result).
"""
if selection.kind == "event":
assert selection.event is not None # classify_stem always sets it for events
tagged = tags.render([tags.Event(selection.event)])
plain = ""
else:
if transcript is None:
raise TextRejected("no_transcript", f"no entry for {selection.stem!r}")
plain = normalise_english(transcript, cfg.char_map)
if not plain:
raise TextRejected("empty_text", "transcript has no characters")
if any(ch.isdigit() for ch in plain):
raise TextRejected("digits", plain[:80])
if any(ch in _TAG_CHARS for ch in plain):
raise TextRejected("tag_characters", plain[:80])
if any(ch.isalpha() and not ch.isascii() for ch in plain):
raise TextRejected("non_ascii_letter", plain[:80])
if selection.kind == "whisper":
try:
span = tags.Span("whisper", tuple(tags.Word(w) for w in plain.split()))
except ValueError as err:
raise TextRejected("text_encoding", str(err)) from err
tagged = tags.render([span])
else:
tagged = plain
try:
ids = encode_text(tagged)
except ValueError as err: # unknown character, bad tag
raise TextRejected("text_encoding", str(err)) from err
if not ids:
raise TextRejected("empty_text", "text encodes to no tokens")
return BuiltText(tagged=tagged, plain=plain)
# --------------------------------------------------------------------------------------
# Selection
# --------------------------------------------------------------------------------------
@dataclass(frozen=True)
class Selection:
"""A file that is used. ``emotion`` is set for ``emotion`` items (may be ``None``),
``event`` for ``event`` items."""
stem: str
kind: Kind
emotion: str | None = None
event: str | None = None
@property
def label(self) -> str:
"""Report category: ``emotion``, ``read``, ``whisper`` or ``event:<id>``."""
return f"event:{self.event}" if self.kind == "event" else self.kind
@dataclass(frozen=True)
class Skip:
"""A file that is not used, with the counted reason."""
stem: str
reason: str
def classify_stem(stem: str, cfg: EarsConfig) -> Selection | Skip:
"""Decide on one file stem (ADR-0007 table). Pure; thresholds and maps come from ``cfg``.
Skip reasons: ``freeform``, ``interjection``, ``melodic``, ``event_unmapped``,
``style_unsupported``, ``unknown_stem``.
"""
emo = _EMO_RE.fullmatch(stem)
if emo is not None:
if emo.group(2) != "sentences":
return Skip(stem, "freeform")
return Selection(stem, "emotion", emotion=cfg.emotion_map.get(emo.group(1)))
event = cfg.event_map.get(stem)
if event is not None:
return Selection(stem, "event", event=event)
read = _READ_RE.fullmatch(stem)
if read is not None and read.group(1) in cfg.read_prefixes:
style = read.group(3)
if style in cfg.whisper_styles:
return Selection(stem, "whisper")
if style in cfg.plain_styles:
return Selection(stem, "read")
return Skip(stem, "style_unsupported")
if stem.startswith("freeform_speech"):
return Skip(stem, "freeform")
for prefix, reason in (
("interjection_", "interjection"),
("melodic_", "melodic"),
("nonverbal_", "event_unmapped"),
("vegetative_", "event_unmapped"),
):
if stem.startswith(prefix):
return Skip(stem, reason)
return Skip(stem, "unknown_stem")
@dataclass(frozen=True)
class PlannedItem:
"""A selected archive member with its final text."""
stem: str
member: ZipMember
selection: Selection
text: BuiltText
@dataclass
class SelectionPlan:
"""Result of :func:`select_members`: nothing is lost, every member is in one of three places."""
items: list[PlannedItem] = field(default_factory=list)
skipped: Counter[str] = field(default_factory=Counter)
quarantine: list[QuarantineRecord] = field(default_factory=list)
def select_members(
speaker: str,
members: Sequence[ZipMember],
transcripts: Mapping[str, str],
cfg: EarsConfig,
) -> SelectionPlan:
"""Decide on every archive member before anything is downloaded.
``items`` come back in archive order (ascending ``header_offset``) so extraction profits
from the read-ahead window. Skipped members are counted by reason; text problems become
quarantine records (stage ``ears``, item id ``<speaker>/<stem>``).
"""
plan = SelectionPlan()
seen: set[str] = set()
for member in sorted(members, key=lambda m: m.header_offset):
if member.is_dir:
plan.skipped["directory"] += 1
continue
parts = member.name.split("/")
if len(parts) != 2 or parts[0] != speaker:
plan.skipped["outside_speaker_folder"] += 1
continue
if not parts[1].endswith(".wav"):
plan.skipped["not_wav"] += 1
continue
stem = parts[1][: -len(".wav")]
decision = classify_stem(stem, cfg)
if isinstance(decision, Skip):
plan.skipped[decision.reason] += 1
continue
item_id = f"{speaker}/{stem}"
if stem in seen:
plan.quarantine.append(
QuarantineRecord(item_id, STAGE, "duplicate_member", {"name": member.name})
)
continue
seen.add(stem)
try:
text = build_text(decision, transcripts.get(stem), cfg)
except TextRejected as err:
plan.quarantine.append(
QuarantineRecord(item_id, STAGE, err.reason, {"detail": err.detail})
)
continue
plan.items.append(PlannedItem(stem, member, decision, text))
_cap_reads(plan, speaker, cfg.max_read_per_speaker)
return plan
def _cap_reads(plan: SelectionPlan, speaker: str, cap: int) -> None:
"""Keep at most ``cap`` plain reading items per speaker (0 = no limit; ADR-0007: plain
reading is a low share). The kept ones are a deterministic hash choice of the stems;
the rest are counted as skipped ``read_cap``."""
reads = [item for item in plan.items if item.selection.kind == "read"]
if cap == 0 or len(reads) <= cap:
return
ranked = sorted(reads, key=lambda it: hashlib.sha1(f"{speaker}/{it.stem}".encode()).hexdigest())
dropped = {id(item) for item in ranked[cap:]}
plan.items = [item for item in plan.items if id(item) not in dropped]
plan.skipped["read_cap"] += len(dropped)
# --------------------------------------------------------------------------------------
# Audio
# --------------------------------------------------------------------------------------
class ItemRejected(Exception):
"""An audio file cannot be used; ``reason`` is the quarantine reason."""
def __init__(self, reason: str, **details: object) -> None:
super().__init__(f"{reason}: {details}")
self.reason = reason
self.details = details
@dataclass(frozen=True)
class AudioFacts:
"""Measurements of a converted file: ``duration_ms`` of the 16 kHz audio, RMS level."""
duration_ms: int
loudness_dbfs: float
def resample(audio: FloatArray, rate_hz: int, target_hz: int) -> FloatArray:
"""Polyphase resampling of a mono signal (anti-aliased) from ``rate_hz`` to ``target_hz``.
Args:
audio: ``(num_samples,)`` float signal.
rate_hz: Source rate.
target_hz: Target rate.
Returns:
``(ceil(num_samples * target_hz / rate_hz),)`` float64 signal.
"""
if rate_hz == target_hz:
return np.asarray(audio, dtype=np.float64)
divisor = math.gcd(rate_hz, target_hz)
out = signal.resample_poly(audio, target_hz // divisor, rate_hz // divisor)
return np.asarray(out, dtype=np.float64)
def convert_to_target_rate(raw: Path, dest: Path, cfg: EarsConfig) -> None:
"""Read a raw EARS wav, check it, resample to ``target_sample_rate_hz`` mono and write a
32-bit float wav atomically to ``dest``.
Raises:
ItemRejected: ``unreadable_audio``, ``not_mono``, ``unexpected_sample_rate``,
``empty_audio`` or ``non_finite_audio``.
"""
import soundfile # type: ignore[import-untyped] # bench extra, untyped
try:
audio, rate_hz = soundfile.read(str(raw), dtype="float64", always_2d=True)
except (RuntimeError, soundfile.LibsndfileError) as err:
raise ItemRejected("unreadable_audio", error=str(err)) from err
if audio.shape[1] != 1:
raise ItemRejected("not_mono", channels=int(audio.shape[1]))
if rate_hz != cfg.expected_sample_rate_hz:
raise ItemRejected("unexpected_sample_rate", sample_rate_hz=int(rate_hz))
mono = np.ascontiguousarray(audio[:, 0])
if mono.shape[0] == 0:
raise ItemRejected("empty_audio")
if not np.all(np.isfinite(mono)):
raise ItemRejected("non_finite_audio")
converted = resample(mono, int(rate_hz), cfg.target_sample_rate_hz)
buffer = io.BytesIO()
soundfile.write(buffer, converted, cfg.target_sample_rate_hz, format="WAV", subtype="FLOAT")
atomic_write_bytes(dest, buffer.getvalue())
def measure_audio(
path: Path, cfg: EarsConfig, min_duration_ms: int, max_duration_ms: int
) -> AudioFacts:
"""Duration and loudness of a converted wav; enforces the duration limits.
The loudness is :func:`nida.data.acoustic.loudness_dbfs`, the measure the pilot ingest
stores, here on the 16 kHz signal the VAE hears.
Raises:
ItemRejected: ``unreadable_audio``, ``too_short``, ``too_long`` or ``silent``.
ValueError: if the file is not at ``target_sample_rate_hz`` (the cache is inconsistent
with the config).
"""
import soundfile
try:
audio, rate_hz = soundfile.read(str(path), dtype="float64", always_2d=True)
except (RuntimeError, soundfile.LibsndfileError) as err:
raise ItemRejected("unreadable_audio", error=str(err)) from err
if rate_hz != cfg.target_sample_rate_hz or audio.shape[1] != 1:
raise ValueError(
f"{path}: expected {cfg.target_sample_rate_hz} Hz mono, got {rate_hz} Hz x "
f"{audio.shape[1]}; delete the cache folder"
)
mono = np.ascontiguousarray(audio[:, 0])
duration_ms = round(1000 * mono.shape[0] / rate_hz)
if duration_ms < min_duration_ms:
raise ItemRejected("too_short", duration_ms=duration_ms)
if duration_ms > max_duration_ms:
raise ItemRejected("too_long", duration_ms=duration_ms)
if float(np.max(np.abs(mono))) == 0.0:
raise ItemRejected("silent")
return AudioFacts(duration_ms=duration_ms, loudness_dbfs=float(acoustic.loudness_dbfs(mono)))
# --------------------------------------------------------------------------------------
# Buckets and splits
# --------------------------------------------------------------------------------------
def bucket_of(value: float, edges: Sequence[float]) -> int:
"""Bucket index of ``value``: the number of ``edges`` that are ``<= value``
(``numpy.searchsorted(edges, value, side="right")``, as the pilot dataset stage)."""
return bisect_right(edges, value)
def rate_chars_per_s(plain_text: str, duration_ms: int) -> float:
"""Alphanumeric characters of the plain text per second of audio (pilot formula)."""
return sum(ch.isalnum() for ch in plain_text) / (duration_ms / 1000)
def speaker_split(group: str, val_fraction: float, test_fraction: float, seed: int) -> str:
"""Split of a speaker group: a pure function of ``(seed, group)`` (:mod:`nida.data.splits`)."""
fractions = {
"train": 1.0 - val_fraction - test_fraction,
"val": val_fraction,
"test": test_fraction,
}
return assign_splits({"speaker": group}, fractions, str(seed))["speaker"]
# --------------------------------------------------------------------------------------
# The builder
# --------------------------------------------------------------------------------------
@dataclass(frozen=True)
class _Measured:
item: PlannedItem
segment_id: str
wav: Path
facts: AudioFacts
@dataclass
class _Prepared:
"""A speaker whose audio is ready for the VAE: nothing is written for it yet."""
plan: SelectionPlan
quarantine: list[QuarantineRecord]
measured: list[_Measured]
@dataclass
class Ears:
"""EARS -> latent dataset builder (see the module docstring).
Attributes:
root: Durable directory (Drive on Colab).
cache: Local scratch for raw members, 16 kHz audio and latent arrays.
envs_root: Where the worker virtual environments live.
repo_dir: Repository root (environment builds need it).
sources_path: ``configs/sources.yaml``.
config_dir: ``configs/data``.
seed: Split seed (recorded in ``plan.json``).
runner: Worker command runner (injectable).
opener: Network access (injectable).
sleep: Backoff sleep for retries (injectable).
"""
root: Path
cache: Path
envs_root: Path
repo_dir: Path
sources_path: Path
config_dir: Path
seed: int = DEFAULT_SEED
uv: str | None = None
runner: Callable[[Sequence[str]], int] = _stream_runner
opener: fetch.Opener = fetch.default_opener
sleep: Callable[[float], None] = time.sleep
_python: Path | None = field(default=None, init=False, repr=False)
# ---------------------------------------------------------------- paths
@property
def dataset_dir(self) -> Path:
return self.root / "dataset"
def _speaker_dir(self, speaker: str) -> Path:
return self.root / "speakers" / speaker
def _npy(self, segment: str) -> Path:
return self.cache / "arrays" / "vae" / f"{segment}.npy"
# ---------------------------------------------------------------- run
def run(self, speakers: Sequence[str]) -> dict[str, object]:
"""Build (or continue) the dataset for ``speakers``.
Finished speakers are skipped. The dataset directory, ``report.md`` and
``stats.json`` are rewritten from *all* finished speakers of the root at the end.
Returns:
Summary with ``dataset``, ``speakers_done``, ``segments``, ``hours``,
``skipped`` and ``quarantined`` counts, and ``failed_speakers``.
Raises:
SourceNotApprovedError: before any download, if a source is not approved.
ValueError: on a plan/config/transcripts mismatch with the root.
RuntimeError: if a worker fails, a speaker could not be fetched (after all other
speakers were processed and the dataset was assembled), or nothing was built.
"""
settings = load_settings(self.config_dir)
model = DATA_MODELS[VAE_KEY]
require_approved({SOURCE_ID, model.source_id}, load_sources(self.sources_path))
wanted = list(speakers)
for speaker in wanted:
parse_speakers(speaker, settings.ears.n_speakers)
transcripts, transcripts_sha = self._prepare_plan(settings)
failed: dict[str, str] = {}
pending: list[str] = []
for speaker in dict.fromkeys(wanted):
if (self._speaker_dir(speaker) / "done.json").is_file():
log.info("EARS %s already done", speaker)
else:
pending.append(speaker)
wave_size = settings.ears.speakers_per_wave
for start in range(0, len(pending), wave_size):
wave = pending[start : start + wave_size]
prepared: dict[str, _Prepared] = {}
for speaker in wave:
log.info("EARS %s (%d/%d)", speaker, pending.index(speaker) + 1, len(pending))
try:
prepared[speaker] = self._prepare_speaker(speaker, settings, transcripts)
except fetch.FetchError as err: # network or archive problem: others still run
log.error("EARS %s failed: %s", speaker, err)
failed[speaker] = str(err)
results = self._encode([m for entry in prepared.values() for m in entry.measured])
for speaker, entry in prepared.items():
self._finish_speaker(speaker, settings, entry, results)
summary = self._assemble(settings, transcripts_sha, failed)
if failed:
raise RuntimeError(
f"{len(failed)} speaker(s) failed and can be retried by running again: "
+ "; ".join(f"{s}: {e[:200]}" for s, e in sorted(failed.items()))
)
return summary
# ---------------------------------------------------------------- plan and transcripts
def _prepare_plan(self, settings: EarsSettings) -> tuple[dict[str, str], str]:
"""Check/write ``plan.json`` and fetch the pinned transcripts."""
model = DATA_MODELS[VAE_KEY]
core: dict[str, object] = {
"seed": self.seed,
"config_digest": settings.digest,
"archive_url_template": settings.ears.archive_url_template,
"transcripts_url": settings.ears.transcripts_url,
"vae": [model.ref.model_id, model.ref.revision],
}
plan_path = self.root / "plan.json"
previous: dict[str, object] | None = None
if plan_path.is_file():
loaded = json.loads(plan_path.read_text(encoding="utf-8"))
previous = dict(loaded)
recorded_sha = previous.pop("transcripts_sha256", None)
if previous != core:
raise ValueError(
f"{plan_path} was written for a different plan: {previous} != {core}"
)
if (
settings.ears.transcripts_sha256 is not None
and recorded_sha is not None
and settings.ears.transcripts_sha256 != recorded_sha
):
raise ValueError(
f"config pins transcripts {settings.ears.transcripts_sha256} but the root "
f"was built with {recorded_sha}"
)
expected = recorded_sha if isinstance(recorded_sha, str) else None
else:
expected = None
expected = settings.ears.transcripts_sha256 or expected
downloaded = fetch.download(
settings.ears.transcripts_url,
self.root / "source" / "transcripts.json",
expected_sha256=expected,
opener=self.opener,
sleep=self.sleep,
)
if previous is None:
atomic_write_json(plan_path, {**core, "transcripts_sha256": downloaded.sha256})
return _parse_transcripts(downloaded.path), downloaded.sha256
# ---------------------------------------------------------------- speaker
def _python_exe(self) -> Path:
if self._python is None:
if self.uv is None:
self.uv = bench_envs.find_uv(bench_envs.subprocess_runner)
spec = envs.data_env_for_worker(DATA_MODELS[VAE_KEY].worker)
self._python = bench_envs.ensure_env(spec, self.envs_root, self.repo_dir, uv=self.uv)
return self._python
def _prepare_speaker(
self, speaker: str, settings: EarsSettings, transcripts: Mapping[str, str]
) -> _Prepared:
"""List the archive, decide on every member, fetch + convert + measure the selected."""
cfg = settings.ears
QuarantineLog(self._speaker_dir(speaker)).path.unlink(missing_ok=True) # rebuilt below
archive = RemoteZip(cfg.archive_url(speaker), opener=self.opener, sleep=self.sleep)
members = archive.members()
plan = select_members(speaker, members, transcripts, cfg)
quarantine = list(plan.quarantine)
log.info(
"EARS %s: %d members, %d selected, %d quarantined, skipped %s",
speaker,
len(members),
len(plan.items),
len(quarantine),
dict(plan.skipped),
)
measured: list[_Measured] = []
for item in plan.items: # archive order: one read-ahead window serves neighbours
item_id = f"{speaker}/{item.stem}"
wav = self.cache / "audio16k" / speaker / f"{item.stem}.wav"
try:
if not wav.is_file():
raw = archive.extract(
item.member, self.cache / "raw" / speaker / f"{item.stem}.wav"
)
convert_to_target_rate(raw, wav, cfg)
raw.unlink()
facts = measure_audio(wav, cfg, settings.min_duration_ms, settings.max_duration_ms)
except ItemRejected as err:
quarantine.append(QuarantineRecord(item_id, STAGE, err.reason, dict(err.details)))
continue
sid = segment_id(SOURCE_ID, item_id, 0, facts.duration_ms)
measured.append(_Measured(item, sid, wav, facts))
return _Prepared(plan, quarantine, measured)
def _finish_speaker(
self,
speaker: str,
settings: EarsSettings,
prepared: _Prepared,
results: Mapping[str, WorkResult],
) -> None:
"""Write the speaker's outputs, then drop its scratch files."""
self._write_speaker(
speaker, settings, prepared.measured, results, prepared.plan, prepared.quarantine
)
for sub in ("raw", "audio16k"):
shutil.rmtree(self.cache / sub / speaker, ignore_errors=True)
for entry in prepared.measured:
self._npy(entry.segment_id).unlink(missing_ok=True)
def _encode(self, measured: Sequence[_Measured]) -> dict[str, WorkResult]:
"""Run the pilot's ``vae_encode`` worker once over the audio of a wave (resumable)."""
if not measured:
return {}
model = DATA_MODELS[VAE_KEY]
python = self._python_exe()
out_dir = self.root / "workers" / "vae"
results_path = out_dir / "results.jsonl"
wanted = {m.segment_id for m in measured}
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.ref.model_id, model.ref.revision):
raise RuntimeError(
f"existing latents come from {previous.model_id}@"
f"{previous.revision}, not {model.ref.model_id}@{model.ref.revision}; "
"move them away first"
)
if results_path.exists():
existing = load_results(results_path, repair=True)
stale = [
cid
for cid, res in existing.items()
if cid not in wanted
or res.output is None # failed before: try again (a stuck item fails again)
or not Path(str(res.output.get("out_path", ""))).is_file()
]
if stale:
log.info("dropping %d stray, failed or lost worker results", len(stale))
drop_results(results_path, stale)
requests = [
WorkRequest(m.segment_id, str(m.wav), {"out_path": str(self._npy(m.segment_id))})
for m in measured
]
write_requests(out_dir / "requests.jsonl", requests)
command = [
str(python),
"-m",
envs.worker_module_name(model.worker),
"--requests",
str(out_dir / "requests.jsonl"),
"--results",
str(results_path),
"--model-id",
model.ref.model_id,
"--revision",
model.ref.revision,
"--batch-size",
str(model.batch_size),
"--device",
"cuda",
]
code = self.runner(command)
if code != 0:
raise RuntimeError(f"vae_encode worker exited with code {code}")
summary = read_summary(summary_path)
total = summary.n_ok + summary.n_error
if total != len(requests):
raise RuntimeError(f"{total} latent results for {len(requests)} requests")
if summary.n_error > MAX_ERROR_FRACTION * total:
raise RuntimeError(f"{summary.n_error}/{total} items failed in the worker")
return load_results(results_path)
def _write_speaker(
self,
speaker: str,
settings: EarsSettings,
measured: Sequence[_Measured],
results: Mapping[str, WorkResult],
plan: SelectionPlan,
quarantine: list[QuarantineRecord],
) -> dict[str, object]:
"""Shard, records, quarantine log and finally the done marker of one speaker."""
cfg = settings.ears
group = f"ears-{speaker}"
split = speaker_split(group, cfg.val_speaker_fraction, cfg.test_speaker_fraction, self.seed)
rate_edges = list(settings.dataset.rate_bucket_edges_chars_per_s)
loud_edges = list(settings.dataset.loudness_bucket_edges_dbfs)
shard_name = f"shard-{speaker}.npy"
rows: list[dict[str, object]] = []
arrays: list[npt.NDArray[np.float32]] = []
kinds: dict[str, dict[str, int]] = {}
latent_dim: int | None = None
frame_hop_ms: float | None = None
frames = 0
for entry in sorted(measured, key=lambda m: m.item.stem):
item_id = f"{speaker}/{entry.item.stem}"
result = results.get(entry.segment_id)
if result is None:
raise RuntimeError(f"{speaker}: the worker returned no result for {item_id}")
if result.output is None:
quarantine.append(
QuarantineRecord(item_id, STAGE, "vae_error", {"error": str(result.error)})
)
continue
array = np.load(str(result.output["out_path"]), allow_pickle=False)
hop = float(str(result.output["frame_hop_ms"]))
if array.ndim != 2 or array.shape[0] < 1:
raise ValueError(f"{item_id}: latents must be 2-D with frames, got {array.shape}")
if frame_hop_ms is not None and hop != frame_hop_ms:
raise ValueError(f"{item_id}: frame hop {hop} != {frame_hop_ms}")
if latent_dim is not None and int(array.shape[1]) != latent_dim:
raise ValueError(f"{item_id}: latent_dim {array.shape[1]} != {latent_dim}")
frame_hop_ms, latent_dim = hop, int(array.shape[1])
if not np.all(np.isfinite(array)) or float(np.max(np.abs(array))) > FLOAT16_MAX_ABS:
quarantine.append(QuarantineRecord(item_id, STAGE, "latents_not_float16", {}))
continue
num_frames = int(array.shape[0])
n_tokens = len(encode_text(entry.item.text.tagged))
if n_tokens > num_frames:
quarantine.append(
QuarantineRecord(
item_id,
STAGE,
REJECT_TEXT_TOO_LONG,
{"text_tokens": n_tokens, "num_frames": num_frames},
)
)
continue
plain = entry.item.text.plain
rows.append(
{
"segment_id": entry.segment_id,
"shard": shard_name,
"offset": frames,
"num_frames": num_frames,
"text": entry.item.text.tagged,
"emotion": entry.item.selection.emotion,
"intensity": None,
"rate_bucket": (
bucket_of(rate_chars_per_s(plain, entry.facts.duration_ms), rate_edges)
if plain
else None
),
"loudness_bucket": bucket_of(entry.facts.loudness_dbfs, loud_edges),
"group": group,
"split": split,
"language": LANGUAGE,
}
)
arrays.append(array.astype(np.float32))
frames += num_frames
stat = kinds.setdefault(entry.item.selection.label, {"segments": 0, "audio_ms": 0})
stat["segments"] += 1
stat["audio_ms"] += entry.facts.duration_ms
spk_dir = self._speaker_dir(speaker)
if arrays:
buffer = io.BytesIO()
np.save(buffer, np.concatenate(arrays).astype(np.float16), allow_pickle=False)
atomic_write_bytes(self.dataset_dir / shard_name, buffer.getvalue())
atomic_write_text(
spk_dir / "records.jsonl",
"".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
)
QuarantineLog(spk_dir).append(quarantine)
marker: dict[str, object] = {
"speaker": speaker,
"group": group,
"split": split,
"segments": len(rows),
"frames": frames,
"audio_ms": sum(s["audio_ms"] for s in kinds.values()),
"latent_dim": latent_dim if rows else None,
"frame_hop_ms": frame_hop_ms if rows else None,
"shard": shard_name if rows else None,
"kinds": kinds,
"skipped": dict(sorted(plan.skipped.items())),
"quarantined": dict(sorted(Counter(q.reason for q in quarantine).items())),
}
atomic_write_json(spk_dir / "done.json", marker) # written last: the speaker is done
return marker
# ---------------------------------------------------------------- assembly
def _assemble(
self, settings: EarsSettings, transcripts_sha: str, failed: Mapping[str, str]
) -> dict[str, object]:
"""Index, meta, report and stats from every finished speaker of the root."""
markers: list[dict[str, object]] = []
speakers_dir = self.root / "speakers"
for done_path in sorted(speakers_dir.glob("p*/done.json")) if speakers_dir.is_dir() else []:
markers.append(json.loads(done_path.read_text(encoding="utf-8")))
with_data = [m for m in markers if m["segments"]]
if not with_data:
raise RuntimeError(
f"no EARS segment built (speakers done: {len(markers)}, failed: {sorted(failed)})"
)
hops = {float(str(m["frame_hop_ms"])) for m in with_data}
dims = {int(str(m["latent_dim"])) for m in with_data}
if len(hops) != 1 or len(dims) != 1:
raise ValueError(
f"speakers disagree on frame hop {sorted(hops)} or latent_dim {sorted(dims)}"
)
index_lines: list[str] = []
for marker in with_data:
lines = (
(self._speaker_dir(str(marker["speaker"])) / "records.jsonl")
.read_text(encoding="utf-8")
.splitlines(keepends=True)
)
if len(lines) != marker["segments"]:
raise ValueError(f"{marker['speaker']}: records.jsonl and done.json disagree")
if not (self.dataset_dir / str(marker["shard"])).is_file():
raise ValueError(f"{marker['speaker']}: shard {marker['shard']} is missing")
index_lines.extend(lines)
total_segments = len(index_lines)
total_frames = sum(int(str(m["frames"])) for m in with_data)
atomic_write_text(self.dataset_dir / "index.jsonl", "".join(index_lines))
source = load_sources(self.sources_path)[SOURCE_ID]
model = DATA_MODELS[VAE_KEY]
meta: dict[str, object] = {
"segments": total_segments,
"frames": total_frames,
"latent_dim": dims.pop(),
"frame_hop_ms": hops.pop(),
"rate_bucket_edges_chars_per_s": list(settings.dataset.rate_bucket_edges_chars_per_s),
"loudness_bucket_edges_dbfs": list(settings.dataset.loudness_bucket_edges_dbfs),
"source": SOURCE_ID,
"license": source.get("license"),
"language": LANGUAGE,
"transcripts_sha256": transcripts_sha,
"transcripts_url": settings.ears.transcripts_url,
"config_digest": settings.digest,
"seed": self.seed,
"vae": [model.ref.model_id, model.ref.revision],
"speakers": {
str(m["speaker"]): {"split": m["split"], "segments": m["segments"]} for m in markers
},
}
atomic_write_json(self.dataset_dir / "meta.json", meta)
loader = self._check_with_loader()
stats = _stats(markers, loader, settings, failed, meta)
atomic_write_json(self.root / "stats.json", stats)
atomic_write_text(self.root / "report.md", _render_report(stats))
rejected = {s: c for s, c in loader.items() if c["rejected"]}
if rejected:
raise RuntimeError(f"nida.train.data rejects records the builder accepted: {rejected}")
return {
"dataset": str(self.dataset_dir),
"speakers_done": len(markers),
"segments": total_segments,
"hours": stats["hours"],
"skipped": stats["skipped"],
"quarantined": stats["quarantined"],
"failed_speakers": dict(failed),
}
def _check_with_loader(self) -> dict[str, dict[str, object]]:
"""Load every split with the training loader; counts per split (must not reject)."""
out: dict[str, dict[str, object]] = {}
for split in SPLITS:
try:
dataset, report = load_latent_dataset(self.dataset_dir, split)
except DataFormatError as err:
raise RuntimeError(f"the assembled dataset is invalid: {err}") from err
dataset.close()
out[split] = {
"records": report.num_in_split,
"accepted": report.num_accepted,
"rejected": report.num_rejected,
"rejection_counts": dict(report.rejection_counts),
}
return out
# --------------------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------------------
def _parse_transcripts(path: Path) -> dict[str, str]:
obj = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(obj, dict) or not obj:
raise ValueError(f"{path}: expected a non-empty mapping stem -> text")
bad = [k for k, v in obj.items() if not isinstance(k, str) or not isinstance(v, str)]
if bad:
raise ValueError(f"{path}: entries must map strings to strings, e.g. {bad[:3]}")
return dict(obj)
def _sum_counts(markers: Sequence[Mapping[str, object]], key: str) -> dict[str, int]:
total: Counter[str] = Counter()
for marker in markers:
value = marker[key]
assert isinstance(value, dict)
total.update({str(k): int(v) for k, v in value.items()})
return dict(sorted(total.items(), key=lambda kv: (-kv[1], kv[0])))
def _stats(
markers: Sequence[Mapping[str, object]],
loader: Mapping[str, Mapping[str, object]],
settings: EarsSettings,
failed: Mapping[str, str],
meta: Mapping[str, object],
) -> dict[str, object]:
kinds: dict[str, dict[str, int]] = {}
for marker in markers:
value = marker["kinds"]
assert isinstance(value, dict)
for label, stat in value.items():
into = kinds.setdefault(label, {"segments": 0, "audio_ms": 0})
into["segments"] += int(stat["segments"])
into["audio_ms"] += int(stat["audio_ms"])
splits: dict[str, dict[str, int]] = {s: {"speakers": 0, "segments": 0} for s in SPLITS}
for marker in markers:
entry = splits[str(marker["split"])]
entry["speakers"] += 1
entry["segments"] += int(str(marker["segments"]))
audio_ms = sum(int(str(m["audio_ms"])) for m in markers)
return {
"speakers_done": len(markers),
"speakers_total": settings.ears.n_speakers,
"segments": meta["segments"],
"frames": meta["frames"],
"hours": round(audio_ms / 3_600_000, 3),
"kinds": dict(sorted(kinds.items())),
"splits": splits,
"skipped": _sum_counts(markers, "skipped"),
"quarantined": _sum_counts(markers, "quarantined"),
"loader": {k: dict(v) for k, v in loader.items()},
"failed_speakers": dict(failed),
"transcripts_sha256": meta["transcripts_sha256"],
"config_digest": settings.digest,
}
def _render_report(stats: Mapping[str, object]) -> str:
kinds = stats["kinds"]
splits = stats["splits"]
loader = stats["loader"]
assert isinstance(kinds, dict) and isinstance(splits, dict) and isinstance(loader, dict)
lines = [
"# EARS latent veri seti raporu",
"",
f"- Konuşmacı: {stats['speakers_done']}/{stats['speakers_total']} tamamlandı",
f"- Segment: {stats['segments']}, süre: {stats['hours']} sa, kare: {stats['frames']}",
f"- Yapılandırma özeti: `{stats['config_digest']}`; transkript SHA-256: "
f"`{stats['transcripts_sha256']}`",
"",
"## Seçilen türler",
"",
*(
f"- {label}: {v['segments']} segment, {v['audio_ms'] / 3_600_000:.3f} sa"
for label, v in kinds.items()
),
"",
"## Bölmeler (konuşmacı-ayrık)",
"",
*(
f"- {name}: {v['speakers']} konuşmacı, {v['segments']} segment"
for name, v in splits.items()
),
"",
"## Atlananlar (neden: sayı)",
"",
json.dumps(stats["skipped"], ensure_ascii=False),
"",
"## Karantina (neden: sayı)",
"",
json.dumps(stats["quarantined"], ensure_ascii=False),
"",
"## Eğitim yükleyicisi doğrulaması (`nida.train.data`)",
"",
*(
f"- {name}: {v['accepted']}/{v['records']} kabul, ret: {v['rejection_counts']}"
for name, v in loader.items()
),
]
failed = stats["failed_speakers"]
if failed:
lines += ["", "## Başarısız konuşmacılar (yeniden çalıştırın)", ""]
assert isinstance(failed, dict)
lines += [f"- {speaker}: {error[:300]}" for speaker, error in failed.items()]
return "\n".join(lines) + "\n"
"""EARS -> latent dataset builder (ADR-0007, ``nida data-ears``).
EARS (Expressive Anechoic Recordings of Speech, CC BY-NC 4.0) provides English emotional,
whispered and non-verbal speech. This module turns the selected files into a *latent dataset
directory* in exactly the format of :mod:`nida.train.data` (records carry the optional key
``"language": "en"``), so it can be trained together with Turkish datasets
(``train --data A --data B``).
What is used (rules and mappings live in ``configs/data/ears.yaml``):
* ``emo_<emotion>_sentences``: reading transcript, emotion mapped (unmapped -> ``null``);
* ``rainbow_NN_<style>`` / ``sentences_NN_<style>``: ``whisper`` -> the whole text wrapped in a
whisper span (rendered by :mod:`nida.text.tags`), the plain styles -> plain text;
* laughter/crying/screaming/coughing/throat-clearing files: just the event tag;
* everything else is skipped and counted by reason.
Per speaker (one remote zip of ~564 MB, read with HTTP Range requests, see
:mod:`nida.data.remote_zip`): list the archive -> decide on every member (skip / text
quarantine, before any download) -> extract the selected members, convert 48 kHz -> 16 kHz mono
-> measure. The speakers are handled in *waves* of ``speakers_per_wave``: one AudioVAE run
(the same ``vae_encode`` worker and environment as the pilot, so the model is loaded once per
wave, not once per speaker) encodes the wave's audio, then every speaker of the wave gets its
float16 shard, index records and ``done.json``. Layout::
<root>/plan.json seed, config digest, VAE pin, transcripts SHA-256
<root>/source/transcripts.json pinned reading transcripts
<root>/speakers/pNNN/records.jsonl index rows of the speaker
<root>/speakers/pNNN/quarantine.jsonl every quarantined item with its reason
<root>/speakers/pNNN/done.json written last: the speaker is finished
<root>/workers/vae/ worker requests/results of the current wave
<root>/dataset/ latent dataset: index.jsonl, meta.json, shard-pNNN.npy
<root>/report.md, <root>/stats.json
<cache>/... raw members, 16 kHz wavs and latent arrays (local disk)
Resumable and idempotent: a speaker with ``done.json`` is skipped without any network access, an
interrupted wave continues (worker results are kept, already converted audio is reused), and
the dataset directory is reassembled from all finished speakers at the end of every run, so
it is usable after any run. A run on a root made with another seed, config digest or VAE pin,
or with different transcripts, is refused.
"""
from __future__ import annotations
import hashlib
import io
import json
import logging
import math
import re
import shutil
import string
import time
from bisect import bisect_right
from collections import Counter
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Literal
import numpy as np
import numpy.typing as npt
# scipy ships no type stubs; the ignore is scoped to this single import (as in acoustic.py).
from scipy import signal # type: ignore[import-untyped]
from nida.bench import envs as bench_envs
from nida.bench import fetch
from nida.bench.b1 import MAX_ERROR_FRACTION, _stream_runner
from nida.bench.protocol import (
WorkRequest,
WorkResult,
drop_results,
load_results,
read_summary,
write_requests,
)
from nida.bench.registry import load_sources, require_approved
from nida.data import acoustic, envs
from nida.data.ingest import IngestConfig
from nida.data.pilot import DatasetConfig
from nida.data.quarantine import QuarantineLog
from nida.data.registry import DATA_MODELS
from nida.data.remote_zip import RemoteZip, ZipMember
from nida.data.splits import assign_splits
from nida.data.types import QuarantineRecord, segment_id
from nida.model.vocab import encode_text
from nida.text import tags
from nida.text.turkish import nfc
from nida.train.data import (
EMOTIONS,
REJECT_TEXT_TOO_LONG,
SPLITS,
DataFormatError,
load_latent_dataset,
)
from nida.utils.io import atomic_write_bytes, atomic_write_json, atomic_write_text
__all__ = [
"SOURCE_ID",
"AudioFacts",