| """Audio helpers: WAV I/O, resampling, augmentation and synthetic noise. |
| |
| Kept dependency-light: only numpy is required (no ffmpeg / scipy), so it |
| runs anywhere including a minimal Hugging Face Space. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import io |
| import math |
| import random |
| import re |
| import wave |
| from pathlib import Path |
| from typing import Tuple |
|
|
| import numpy as np |
|
|
|
|
| |
| |
| |
|
|
| def slugify(text: str, max_len: int = 80) -> str: |
| text = str(text).strip().lower() |
| text = re.sub(r"[^a-z0-9]+", "_", text) |
| text = re.sub(r"_+", "_", text).strip("_") |
| return (text or "item")[:max_len] |
|
|
|
|
| def stable_hash(text: str, length: int = 12) -> str: |
| return hashlib.sha1(text.encode("utf-8")).hexdigest()[:length] |
|
|
|
|
| |
| |
| |
|
|
| def read_wav_bytes(wav_bytes: bytes) -> Tuple[np.ndarray, int]: |
| """Decode 16-bit PCM WAV bytes into mono float32 samples and sample rate.""" |
| with wave.open(io.BytesIO(wav_bytes), "rb") as wf: |
| channels = wf.getnchannels() |
| sample_width = wf.getsampwidth() |
| sample_rate = wf.getframerate() |
| frames = wf.readframes(wf.getnframes()) |
|
|
| if sample_width != 2: |
| raise ValueError(f"Expected 16-bit PCM WAV, got sample width {sample_width}") |
|
|
| audio = np.frombuffer(frames, dtype=np.int16).astype(np.float32) |
| if channels > 1: |
| audio = audio.reshape(-1, channels).mean(axis=1) |
| return audio, sample_rate |
|
|
|
|
| def read_wav_file(path: Path) -> Tuple[np.ndarray, int]: |
| return read_wav_bytes(Path(path).read_bytes()) |
|
|
|
|
| def write_wav_file(path: Path, audio: np.ndarray, sample_rate_hz: int) -> None: |
| path = Path(path) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| audio_i16 = np.clip(audio, -32768, 32767).astype(np.int16) |
| with wave.open(str(path), "wb") as wf: |
| wf.setnchannels(1) |
| wf.setsampwidth(2) |
| wf.setframerate(sample_rate_hz) |
| wf.writeframes(audio_i16.tobytes()) |
|
|
|
|
| def resample(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray: |
| """Linear-interpolation resample. Good enough for a bootstrap dataset.""" |
| if src_rate == dst_rate or len(audio) == 0: |
| return audio.astype(np.float32) |
| duration = len(audio) / float(src_rate) |
| dst_len = max(1, int(round(duration * dst_rate))) |
| src_idx = np.linspace(0.0, len(audio) - 1, num=dst_len) |
| return np.interp(src_idx, np.arange(len(audio)), audio).astype(np.float32) |
|
|
|
|
| |
| |
| |
|
|
| def normalize(audio: np.ndarray, peak: float = 28000.0) -> np.ndarray: |
| if len(audio) == 0: |
| return audio |
| m = float(np.max(np.abs(audio))) |
| if m < 1.0: |
| return audio |
| return (audio * (peak / m)).astype(np.float32) |
|
|
|
|
| def pad_or_trim(audio: np.ndarray, target_samples: int, random_crop: bool = False) -> np.ndarray: |
| current = len(audio) |
| if current == target_samples: |
| return audio |
| if current > target_samples: |
| start = ( |
| random.randint(0, current - target_samples) |
| if random_crop |
| else (current - target_samples) // 2 |
| ) |
| return audio[start:start + target_samples] |
| pad_total = target_samples - current |
| pad_left = pad_total // 2 |
| pad_right = pad_total - pad_left |
| return np.pad(audio, (pad_left, pad_right), mode="constant") |
|
|
|
|
| |
| |
| |
|
|
| def _gain(audio: np.ndarray, gain_db: float) -> np.ndarray: |
| return audio * (10.0 ** (gain_db / 20.0)) |
|
|
|
|
| def _time_shift(audio: np.ndarray, max_shift: int) -> np.ndarray: |
| return np.roll(audio, random.randint(-max_shift, max_shift)) |
|
|
|
|
| def _add_noise(audio: np.ndarray, snr_db: float) -> np.ndarray: |
| noise = np.random.normal(0.0, 1.0, len(audio)).astype(np.float32) |
| clean_power = float(np.mean(audio ** 2)) |
| noise_power = float(np.mean(noise ** 2)) |
| if clean_power < 1.0 or noise_power < 1e-9: |
| return audio |
| target_noise_power = clean_power / (10.0 ** (snr_db / 10.0)) |
| noise *= math.sqrt(target_noise_power / noise_power) |
| return audio + noise |
|
|
|
|
| def _echo(audio: np.ndarray, sr: int) -> np.ndarray: |
| delay = random.randint(int(0.03 * sr), int(0.12 * sr)) |
| decay = random.uniform(0.08, 0.25) |
| out = audio.copy() |
| if 0 < delay < len(audio): |
| out[delay:] += audio[:-delay] * decay |
| return out |
|
|
|
|
| def augment(audio: np.ndarray, sr: int) -> np.ndarray: |
| out = audio.copy() |
| out = _gain(out, random.uniform(-6.0, 3.0)) |
| out = _time_shift(out, max_shift=int(0.12 * sr)) |
| if random.random() < 0.75: |
| out = _add_noise(out, random.choice([30, 25, 20, 15, 10])) |
| if random.random() < 0.35: |
| out = _echo(out, sr) |
| return normalize(out, 28000.0) |
|
|
|
|
| |
| |
| |
|
|
| def _white(n: int) -> np.ndarray: |
| return np.random.normal(0.0, 1.0, n).astype(np.float32) |
|
|
|
|
| def _pink(n: int) -> np.ndarray: |
| white = _white(n) |
| out = np.zeros_like(white) |
| alpha = 0.985 |
| for i in range(1, n): |
| out[i] = alpha * out[i - 1] + (1.0 - alpha) * white[i] |
| return out.astype(np.float32) |
|
|
|
|
| def _brown(n: int) -> np.ndarray: |
| brown = np.cumsum(_white(n)) |
| brown = brown - np.mean(brown) |
| return normalize(brown.astype(np.float32), 1.0) |
|
|
|
|
| def _hum(n: int, sr: int) -> np.ndarray: |
| t = np.arange(n, dtype=np.float32) / float(sr) |
| hum = ( |
| np.sin(2.0 * math.pi * 50.0 * t) |
| + 0.5 * np.sin(2.0 * math.pi * 100.0 * t) |
| + 0.25 * np.sin(2.0 * math.pi * 150.0 * t) |
| ) |
| hum += 0.04 * _white(n) |
| return hum.astype(np.float32) |
|
|
|
|
| def _fan(n: int, sr: int) -> np.ndarray: |
| base = _pink(n) |
| t = np.arange(n, dtype=np.float32) / float(sr) |
| blade_rate = random.uniform(18.0, 45.0) |
| modulation = 0.65 + 0.35 * np.sin(2.0 * math.pi * blade_rate * t) |
| return (base * modulation).astype(np.float32) |
|
|
|
|
| def _cafe(n: int, sr: int) -> np.ndarray: |
| base = 0.55 * _pink(n) + 0.45 * _white(n) |
| transient_count = max(1, int((n / sr) * random.uniform(2.0, 6.0))) |
| for _ in range(transient_count): |
| pos = random.randint(0, max(0, n - 1)) |
| length = random.randint(max(1, int(0.008 * sr)), max(2, int(0.05 * sr))) |
| end = min(n, pos + length) |
| if end <= pos: |
| continue |
| click = np.hanning(end - pos).astype(np.float32) |
| base[pos:end] += click * random.uniform(0.5, 2.0) |
| return base.astype(np.float32) |
|
|
|
|
| def _street(n: int, sr: int) -> np.ndarray: |
| base = 0.7 * _brown(n) + 0.3 * _white(n) |
| t = np.arange(n, dtype=np.float32) / float(sr) |
| for _ in range(random.randint(1, 3)): |
| center = random.uniform(0.2, max(0.21, t[-1] - 0.2)) |
| width = random.uniform(0.2, 0.8) |
| envelope = np.exp(-0.5 * ((t - center) / width) ** 2) |
| freq = random.uniform(70.0, 180.0) |
| base += 0.35 * envelope * np.sin(2.0 * math.pi * freq * t) |
| return base.astype(np.float32) |
|
|
|
|
| def make_background_noise(noise_type: str, num_samples: int, sr: int) -> np.ndarray: |
| if noise_type == "white": |
| noise = _white(num_samples) |
| elif noise_type == "pink": |
| noise = _pink(num_samples) |
| elif noise_type == "brown": |
| noise = _brown(num_samples) |
| elif noise_type == "hum": |
| noise = _hum(num_samples, sr) |
| elif noise_type == "fan": |
| noise = _fan(num_samples, sr) |
| elif noise_type == "cafe": |
| noise = _cafe(num_samples, sr) |
| elif noise_type == "street": |
| noise = _street(num_samples, sr) |
| else: |
| raise ValueError(f"Unknown noise type: {noise_type}") |
| return normalize(noise, random.uniform(6000, 22000)) |
|
|
