| """S1 — ingestion of YODAS utterances: listing, audio checks, S1/S2 signal measurements. |
| |
| YODAS ships one 16 kHz wav per subtitle cue (``<video>-<seg>-<start_cs>-<end_cs>``), so a |
| YODAS utterance *is* a segment candidate: no long-file segmentation is needed. Every |
| utterance becomes either a manifest row or a quarantine record with a reason. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import math |
| from collections.abc import Mapping |
| from dataclasses import dataclass, fields |
| from pathlib import Path |
| from typing import TYPE_CHECKING |
|
|
| from nida.bench.clips import YodasUtt, parse_yodas_utt_id |
| from nida.data import acoustic |
| from nida.data.types import QuarantineRecord, segment_id |
| from nida.text.turkish import nfc |
|
|
| if TYPE_CHECKING: |
| import pyarrow as pa |
|
|
| __all__ = ["STAGE", "IngestConfig", "ingest_schema", "ingest_utterance"] |
|
|
| STAGE = "ingest" |
| SOURCE_ID = "yodas_tr" |
|
|
|
|
| @dataclass(frozen=True) |
| class IngestConfig: |
| """configs/data/ingest.yaml (all required).""" |
|
|
| expected_sample_rate_hz: int |
| duration_tolerance_ms: int |
| min_duration_ms: int |
| max_duration_ms: int |
| max_clipping_ratio: float |
| max_silence_ratio: float |
|
|
| def __post_init__(self) -> None: |
| for name in ( |
| "expected_sample_rate_hz", |
| "duration_tolerance_ms", |
| "min_duration_ms", |
| "max_duration_ms", |
| ): |
| value = getattr(self, name) |
| if isinstance(value, bool) or not isinstance(value, int) or value < 0: |
| raise ValueError(f"{name} must be a non-negative int, got {value!r}") |
| if self.expected_sample_rate_hz == 0: |
| raise ValueError("expected_sample_rate_hz must be positive") |
| if not 0 < self.min_duration_ms < self.max_duration_ms: |
| raise ValueError("need 0 < min_duration_ms < max_duration_ms") |
| for name in ("max_clipping_ratio", "max_silence_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]) -> IngestConfig: |
| expected = {f.name for f in fields(cls)} |
| if set(raw) != expected: |
| raise ValueError( |
| f"ingest config keys: missing={sorted(expected - set(raw))} " |
| f"unknown={sorted(set(raw) - expected)}" |
| ) |
| return cls(**raw) |
|
|
|
|
| def ingest_schema() -> pa.Schema: |
| import pyarrow as pa |
|
|
| from nida.data.manifest import schema_from_fields |
|
|
| return schema_from_fields( |
| [ |
| ("segment_id", pa.string()), |
| ("source_id", pa.string()), |
| ("utt_id", pa.string()), |
| ("video_id", pa.string()), |
| ("shard", pa.string()), |
| ("caption", pa.string()), |
| ("audio_relpath", pa.string()), |
| ("duration_ms", pa.int64()), |
| ("sample_rate_hz", pa.int64()), |
| ("bandwidth_hz", pa.float64()), |
| ("loudness_dbfs", pa.float64()), |
| ("clipping_ratio", pa.float64()), |
| ("silence_ratio", pa.float64()), |
| ] |
| ) |
|
|
|
|
| def _drop(utt_id: str, reason: str, **details: object) -> QuarantineRecord: |
| return QuarantineRecord(utt_id, STAGE, reason, dict(details)) |
|
|
|
|
| def ingest_utterance( |
| utt: YodasUtt, |
| audio_root: Path, |
| cfg: IngestConfig, |
| acfg: acoustic.AcousticConfig, |
| ) -> dict[str, object] | QuarantineRecord: |
| """Read ``<audio_root>/<shard>/<utt_id>.wav``, check it and measure S1/S2 signals. |
| |
| Quarantine reasons: ``bad_utt_id``, ``unreadable_audio``, ``not_mono``, |
| ``unexpected_sample_rate``, ``empty_audio``, ``non_finite_audio``, ``duration_mismatch``, |
| ``too_short``, ``too_long``, ``silent``, ``clipping``, ``mostly_silence``. |
| The quarantine ``item_id`` is the YODAS utterance id (the segment id needs a valid |
| duration, which a broken file does not have). |
| """ |
| import numpy as np |
| import soundfile |
|
|
| if any(bad in utt.utt_id for bad in ("/", "\\", "..", chr(0))): |
| return _drop(utt.utt_id, "bad_utt_id", error="path characters in utterance id") |
| try: |
| video_id, start_ms, end_ms = parse_yodas_utt_id(utt.utt_id) |
| except ValueError as err: |
| return _drop(utt.utt_id, "bad_utt_id", error=str(err)) |
| if not utt.shard: |
| raise ValueError(f"utterance {utt.utt_id} has no shard") |
| relpath = f"{utt.shard}/{utt.utt_id}.wav" |
| try: |
| audio, rate_hz = soundfile.read(str(audio_root / relpath), dtype="float64", always_2d=True) |
| except (RuntimeError, soundfile.LibsndfileError) as err: |
| return _drop(utt.utt_id, "unreadable_audio", error=str(err)) |
| if audio.shape[1] != 1: |
| return _drop(utt.utt_id, "not_mono", channels=int(audio.shape[1])) |
| if rate_hz != cfg.expected_sample_rate_hz: |
| return _drop(utt.utt_id, "unexpected_sample_rate", sample_rate_hz=int(rate_hz)) |
| mono = np.ascontiguousarray(audio[:, 0]) |
| if mono.shape[0] == 0: |
| return _drop(utt.utt_id, "empty_audio") |
| if not np.all(np.isfinite(mono)): |
| return _drop(utt.utt_id, "non_finite_audio") |
| duration_ms = round(1000 * mono.shape[0] / rate_hz) |
| listed = [utt.duration_ms, end_ms - start_ms] |
| if any(abs(duration_ms - d) > cfg.duration_tolerance_ms for d in listed): |
| return _drop(utt.utt_id, "duration_mismatch", audio_ms=duration_ms, listed_ms=listed) |
| if duration_ms < cfg.min_duration_ms: |
| return _drop(utt.utt_id, "too_short", duration_ms=duration_ms) |
| if duration_ms > cfg.max_duration_ms: |
| return _drop(utt.utt_id, "too_long", duration_ms=duration_ms) |
| try: |
| bandwidth_hz = acoustic.estimate_bandwidth_hz( |
| mono, int(rate_hz), floor_db=acfg.bandwidth_floor_db, params=acfg.bandwidth |
| ) |
| except acoustic.SilentAudioError: |
| return _drop(utt.utt_id, "silent") |
| clipping = acoustic.clipping_ratio(mono, threshold=acfg.clipping_threshold) |
| if clipping > cfg.max_clipping_ratio: |
| return _drop(utt.utt_id, "clipping", clipping_ratio=clipping) |
| silence = acoustic.silence_ratio( |
| mono, int(rate_hz), frame_ms=acfg.silence_frame_ms, silence_dbfs=acfg.silence_dbfs |
| ) |
| if silence > cfg.max_silence_ratio: |
| return _drop(utt.utt_id, "mostly_silence", silence_ratio=silence) |
| return { |
| "segment_id": segment_id(SOURCE_ID, utt.utt_id, 0, duration_ms), |
| "source_id": SOURCE_ID, |
| "utt_id": utt.utt_id, |
| "video_id": video_id, |
| "shard": utt.shard, |
| "caption": nfc(utt.caption), |
| "audio_relpath": relpath, |
| "duration_ms": duration_ms, |
| "sample_rate_hz": int(rate_hz), |
| "bandwidth_hz": float(bandwidth_hz), |
| "loudness_dbfs": float(acoustic.loudness_dbfs(mono)), |
| "clipping_ratio": float(clipping), |
| "silence_ratio": float(silence), |
| } |
|
|
