ADR-227: Add BackgroundNoiseAugmentor and MicrophoneNoiseAugmentor stages

ml/params.yaml

@@ -2,6 +2,7 @@ pipeline:
   input_phrases_path: scripts/intent_prediction/01_input_phrases.csv
   variations_per_phrase: 20
   subsample_rate: 200
+  sample_rate: 16000
 
 stages:
   generate_phrases:
@@ -20,3 +21,11 @@ stages:
     suffix_vary_probability: 0.333
     suffix_min_s: 0.1
     suffix_max_s: 0.5
+  add_background_noise:
+    vary_probability: 0.5
+    volume_min: 0.0
+    volume_max: 0.3
+  add_mic_noise:
+    vary_probability: 0.5
+    amplitude_min: 0.001
+    amplitude_max: 0.05

ml/pipeline/speech/_doc_speech.md

@@ -10,9 +10,10 @@ produces a derived `AudioSample` manifest with per-sample variation applied.
 |--------|-------------|-----------|
 | [`tts_stage.py`](tts_stage.py) | `TtsSampleGenerator` | `TextSample → AudioSample` via edge_tts |
 | [`delay_stage.py`](delay_stage.py) | `DelayAugmentor` | `AudioSample → AudioSample` (silence padding) |
+| [`background_noise_stage.py`](background_noise_stage.py) | `BackgroundNoiseAugmentor` | `AudioSample → AudioSample` (environmental noise mix) |
+| [`mic_noise_stage.py`](mic_noise_stage.py) | `MicrophoneNoiseAugmentor` | `AudioSample → AudioSample` (Gaussian mic noise) |
 
-Planned stages (not yet implemented): `background_noise_stage.py`,
-`mic_noise_stage.py`, `token_stage.py`, `spectrogram_stage.py`.
+Planned stages (not yet implemented): `token_stage.py`, `spectrogram_stage.py`.
 
 ## Key design decisions
 
@@ -36,3 +37,18 @@ prefix for the same reason.
 accepts `list[str]` voices directly. The entry-point runs `asyncio.run(edge_tts.list_voices())`
 and filters: `Gender == 'Female'`, `Locale == 'en-US'`, `':' not in ShortName`,
 `'DragonHD' not in ShortName`, `'Turbo' not in ShortName`.
+
+**`NoiseProvider` protocol as the noise-file seam for `BackgroundNoiseAugmentor`.**
+Consistent with `TtsProvider` in `tts_stage.py` — the protocol lives in the same
+module as the stage that uses it. Unit tests supply `_FakeNoiseProvider`; the
+entry-point supplies `_DirectoryNoiseProvider` (globbing `*.wav` from `--noise-dir`).
+
+**`noise_file` is always chosen (hash stability), `noise_volume` is 0.0 when not applied.**
+`VariationGenerator.choose()` runs on the sorted filename list before the `should_vary`
+check so the content hash does not change if `vary_probability` is toggled. All three
+keys (`noise_file`, `noise_start_s`, `noise_volume`) are always present in `applied_values`.
+`noise_start_s` is always 0.0 — noise is mixed from the beginning of the noise file.
+
+**Gaussian noise in `MicrophoneNoiseAugmentor` is seeded from `output_seed`.** Uses
+`np.random.default_rng(output_seed).normal(0, amplitude, len(samples))` for
+reproducibility. The amplitude is stored as 0.0 when not applied.

ml/pipeline/speech/background_noise_stage.py

@@ -0,0 +1,151 @@
+"""BackgroundNoiseAugmentor: mix environmental noise into WAV audio samples."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Protocol
+
+import numpy as np
+
+from pipeline.core.manifest import ManifestStore
+from pipeline.core.modifier_stage import ModifierStage
+from pipeline.core.randomization import MinMaxFilter, VariationGenerator
+from pipeline.core.sample import AudioSample
+from pipeline.io.audio_io import AudioData, AudioReader, AudioWriter
+from pipeline.stages import conventions
+from pipeline.stages.params import AddBackgroundNoiseParams
+
+
+class NoiseProvider(Protocol):
+    """Protocol for supplying pre-loaded noise audio data.
+
+    Implementations load and resample noise files at construction time so that
+    each call to list_files() is cheap and all returned audio uses the same
+    sample rate.
+
+    Returns a list of (filename, AudioData) pairs. The filename is used for
+    hash-stable variation selection; AudioData is used for mixing.
+    """
+
+    def list_files(self) -> list[tuple[str, AudioData]]: ...
+
+
+class BackgroundNoiseAugmentor(ModifierStage[AudioSample, AudioSample]):
+    """Augmentation stage that mixes environmental noise into WAV samples.
+
+    The noise file is always chosen (even when not applied) so that the
+    content hash remains stable across configuration changes. Only
+    noise_volume is set to 0.0 when the noise is not applied. noise_start_s
+    is always 0.0 (noise always mixed from the start of the noise file).
+
+    When noise_volume == 0.0, the input sample is returned unchanged — no
+    file is written and no I/O occurs.
+
+    Noise mixing (when applied): slice noise for len(audio) samples,
+    zero-pad if noise is shorter, multiply by noise_volume, add, clip to [-1.0, 1.0].
+    """
+
+    def __init__(
+        self,
+        output_dir: Path,
+        manifest_store: ManifestStore,
+        audio_reader: AudioReader,
+        audio_writer: AudioWriter,
+        input_dir: Path,
+        noise_provider: NoiseProvider,
+        params: AddBackgroundNoiseParams,
+    ) -> None:
+        super().__init__(output_dir, manifest_store)
+        self._audio_reader = audio_reader
+        self._audio_writer = audio_writer
+        self._input_dir = input_dir
+        self._noise_provider = noise_provider
+        self._vary_probability = params.vary_probability
+        self._volume_filter = MinMaxFilter(params.volume_min, params.volume_max, precision=2)
+
+    def _get_applied_values(
+        self, sample: AudioSample, generator: VariationGenerator
+    ) -> dict[str, Any]:
+        # Always choose a noise file for hash stability, even when not applied.
+        # list_files() returns pre-loaded (name, AudioData) pairs.
+        noise_items = self._noise_provider.list_files()
+        noise_file: str = generator.choose("noise_file", sorted([name for name, _ in noise_items]))
+
+        if generator.should_vary("noise", self._vary_probability):
+            noise_volume = generator.generate("noise_volume", self._volume_filter)
+        else:
+            noise_volume = 0.0
+
+        return {
+            "noise_file": noise_file,
+            "noise_start_s": 0.0,  # Always zero — noise is mixed from the start of the file.
+            "noise_volume": float(noise_volume),
+        }
+
+    def _derive_id(self, input_sample: AudioSample, applied_values: dict[str, Any]) -> str:
+        noise_file: str = applied_values["noise_file"]
+        noise_volume: float = applied_values["noise_volume"]
+        noise_filestem = Path(noise_file).stem
+        return f"{input_sample.id}_{noise_filestem}_v{int(noise_volume * 100)}"
+
+    async def _generate_output(
+        self,
+        input_sample: AudioSample,
+        output_id: str,
+        output_seed: int,
+        applied_values: dict[str, Any],
+        parent_content_hash: str,
+    ) -> AudioSample:
+        noise_volume: float = applied_values["noise_volume"]
+
+        # When not applied, return the input sample unchanged — no I/O.
+        if noise_volume == 0.0:
+            return input_sample
+
+        noise_file: str = applied_values["noise_file"]
+        noise_start_s: float = applied_values["noise_start_s"]
+
+        input_path = self._input_dir / input_sample.path
+        audio = await self._audio_reader.read(input_path)
+
+        # Look up the pre-loaded noise audio from the provider.
+        noise_items = self._noise_provider.list_files()
+        noise_audio = next(data for name, data in noise_items if name == noise_file)
+
+        if noise_audio.sample_rate != audio.sample_rate:
+            raise ValueError(
+                f"Noise file '{noise_file}' has sample_rate {noise_audio.sample_rate} Hz "
+                f"but input audio has sample_rate {audio.sample_rate} Hz. "
+                f"Configure _DirectoryNoiseProvider with the pipeline sample_rate so that "
+                f"noise files are resampled at load time."
+            )
+
+        start_sample = int(noise_start_s * noise_audio.sample_rate)
+        n_needed = len(audio.samples)
+        noise_slice = noise_audio.samples[start_sample: start_sample + n_needed]
+
+        # Zero-pad if noise slice is shorter than audio
+        if len(noise_slice) < n_needed:
+            noise_slice = np.pad(noise_slice, (0, n_needed - len(noise_slice)))
+
+        output_samples: np.ndarray = np.clip(
+            audio.samples + noise_volume * noise_slice, -1.0, 1.0
+        ).astype(np.float32)
+
+        self._output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = conventions.sample_file_path(self._output_dir, output_id, "wav")
+        await self._audio_writer.write(output_path, AudioData(samples=output_samples, sample_rate=audio.sample_rate))
+
+        content_hash = self._compute_content_hash(
+            parent_content_hash, output_seed, applied_values
+        )
+
+        return AudioSample(
+            id=output_id,
+            seed=output_seed,
+            content_hash=content_hash,
+            path=Path(f"{output_id}.wav"),
+            parent_content_hash=parent_content_hash,
+            transcript=input_sample.transcript,
+            applied_values=applied_values,
+        )

ml/pipeline/speech/mic_noise_stage.py

@@ -0,0 +1,109 @@
+"""MicrophoneNoiseAugmentor: add Gaussian microphone noise to WAV audio samples.
+
+mic_noise_amplitude is stored as 0.0 when should_vary returns False.
+When amplitude == 0.0, the input sample is returned unchanged — no file is written.
+Gaussian noise is seeded from output_seed for reproducibility.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from pipeline.core.manifest import ManifestStore
+from pipeline.core.modifier_stage import ModifierStage
+from pipeline.core.randomization import MinMaxFilter, VariationGenerator
+from pipeline.core.sample import AudioSample
+from pipeline.io.audio_io import AudioData, AudioReader, AudioWriter
+from pipeline.stages import conventions
+from pipeline.stages.params import AddMicNoiseParams
+
+
+class MicrophoneNoiseAugmentor(ModifierStage[AudioSample, AudioSample]):
+    """Augmentation stage that adds Gaussian microphone noise to WAV samples.
+
+    mic_noise_amplitude is stored as 0.0 when the noise is not applied. When
+    amplitude == 0.0, the input sample is returned unchanged — no file is
+    written and no I/O occurs.
+
+    Gaussian noise uses output_seed for reproducibility via numpy's default_rng.
+
+    input_dir is the directory containing the input WAV files referenced by
+    AudioSample.path. This is typically the output directory of the preceding stage.
+    """
+
+    def __init__(
+        self,
+        output_dir: Path,
+        manifest_store: ManifestStore,
+        audio_reader: AudioReader,
+        audio_writer: AudioWriter,
+        input_dir: Path,
+        params: AddMicNoiseParams,
+    ) -> None:
+        super().__init__(output_dir, manifest_store)
+        self._audio_reader = audio_reader
+        self._audio_writer = audio_writer
+        self._input_dir = input_dir
+        self._vary_probability = params.vary_probability
+        self._amplitude_filter = MinMaxFilter(params.amplitude_min, params.amplitude_max, precision=3)
+
+    def _get_applied_values(
+        self, sample: AudioSample, generator: VariationGenerator
+    ) -> dict[str, Any]:
+        if generator.should_vary("mic_noise_amplitude", self._vary_probability):
+            amplitude = generator.generate("mic_noise_amplitude", self._amplitude_filter)
+        else:
+            amplitude = 0.0
+
+        return {
+            "mic_noise_amplitude": float(amplitude),
+        }
+
+    def _derive_id(self, input_sample: AudioSample, applied_values: dict[str, Any]) -> str:
+        amplitude: float = applied_values["mic_noise_amplitude"]
+        if amplitude == 0.0:
+            # No modification — return the input id unchanged.
+            return input_sample.id
+        return f"{input_sample.id}_mic{int(amplitude * 1000)}"
+
+    async def _generate_output(
+        self,
+        input_sample: AudioSample,
+        output_id: str,
+        output_seed: int,
+        applied_values: dict[str, Any],
+        parent_content_hash: str,
+    ) -> AudioSample:
+        amplitude: float = applied_values["mic_noise_amplitude"]
+
+        # When not applied, return the input sample unchanged — no I/O.
+        if amplitude == 0.0:
+            return input_sample
+
+        input_path = self._input_dir / input_sample.path
+        audio = await self._audio_reader.read(input_path)
+
+        rng = np.random.default_rng(output_seed)
+        noise = rng.normal(0, amplitude, len(audio.samples)).astype(np.float32)
+        output_samples: np.ndarray = np.clip(audio.samples + noise, -1.0, 1.0).astype(np.float32)
+
+        self._output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = conventions.sample_file_path(self._output_dir, output_id, "wav")
+        await self._audio_writer.write(output_path, AudioData(samples=output_samples, sample_rate=audio.sample_rate))
+
+        content_hash = self._compute_content_hash(
+            parent_content_hash, output_seed, applied_values
+        )
+
+        return AudioSample(
+            id=output_id,
+            seed=output_seed,
+            content_hash=content_hash,
+            path=Path(f"{output_id}.wav"),
+            parent_content_hash=parent_content_hash,
+            transcript=input_sample.transcript,
+            applied_values=applied_values,
+        )

ml/pipeline/stages/params.py

@@ -37,13 +37,30 @@ class AddDelaysParams:
     suffix_max_s: float
 
 
+@dataclass
+class AddBackgroundNoiseParams:
+    vary_probability: float
+    volume_min: float
+    volume_max: float
+
+
+@dataclass
+class AddMicNoiseParams:
+    vary_probability: float
+    amplitude_min: float
+    amplitude_max: float
+
+
 @dataclass
 class PipelineParams:
     variations_per_phrase: int
     subsample_rate: int
+    sample_rate: int
     generate_phrases: GeneratePhraseParams
     generate_samples: GenerateSamplesParams
     add_delays: AddDelaysParams
+    add_background_noise: AddBackgroundNoiseParams
+    add_mic_noise: AddMicNoiseParams
 
     @classmethod
     def load(cls, path: Path) -> "PipelineParams":
@@ -54,10 +71,13 @@ def load(cls, path: Path) -> "PipelineParams":
         stage_phrases = raw["stages"]["generate_phrases"]
         stage_samples = raw["stages"]["generate_speech_samples"]
         stage_delays = raw["stages"]["add_delays"]
+        stage_bg_noise = raw["stages"]["add_background_noise"]
+        stage_mic_noise = raw["stages"]["add_mic_noise"]
 
         return cls(
             variations_per_phrase=int(pipeline["variations_per_phrase"]),
             subsample_rate=int(pipeline["subsample_rate"]),
+            sample_rate=int(pipeline["sample_rate"]),
             generate_phrases=GeneratePhraseParams(
                 repeat_modifier_chance=float(stage_phrases["repeat_modifier_chance"]),
                 pleasantry_chance=float(stage_phrases["pleasantry_chance"]),
@@ -77,4 +97,14 @@ def load(cls, path: Path) -> "PipelineParams":
                 suffix_min_s=float(stage_delays["suffix_min_s"]),
                 suffix_max_s=float(stage_delays["suffix_max_s"]),
             ),
+            add_background_noise=AddBackgroundNoiseParams(
+                vary_probability=float(stage_bg_noise["vary_probability"]),
+                volume_min=float(stage_bg_noise["volume_min"]),
+                volume_max=float(stage_bg_noise["volume_max"]),
+            ),
+            add_mic_noise=AddMicNoiseParams(
+                vary_probability=float(stage_mic_noise["vary_probability"]),
+                amplitude_min=float(stage_mic_noise["amplitude_min"]),
+                amplitude_max=float(stage_mic_noise["amplitude_max"]),
+            ),
         )