Plan 3: 24 kHz->16 kHz downsample helper for the wakeword feed

client/audio.py

@@ -0,0 +1,26 @@
+"""Audio helpers for the Pi client.
+
+This module is split into two layers:
+
+* Deterministic helpers (no hardware) — `resample_24k_to_16k`. Tested.
+* `AudioStreams` — owns the persistent PortAudio InputStream / OutputStream.
+  Filled in by Task 7. Smoke-tested by `main.py` running on the Pi.
+"""
+import numpy as np
+from scipy.signal import resample_poly
+
+
+def resample_24k_to_16k(frame: np.ndarray) -> np.ndarray:
+    """Downsample a 24 kHz int16 mono frame to 16 kHz int16 mono (2:3 poly).
+
+    Input length is preserved as `len(frame) * 2 // 3`. For the 1920-sample
+    frames produced by our 80 ms blocksize at 24 kHz, this yields 1280 samples
+    (80 ms at 16 kHz), which is exactly what openwakeword expects.
+    """
+    if frame.dtype != np.int16:
+        raise TypeError(f"expected int16, got {frame.dtype}")
+    if frame.size == 0:
+        raise ValueError("empty input")
+    floats = frame.astype(np.float32)
+    resampled = resample_poly(floats, up=2, down=3)
+    return np.clip(resampled, -32768, 32767).astype(np.int16)

client/tests/test_audio_resample.py

@@ -0,0 +1,47 @@
+import numpy as np
+import pytest
+
+scipy_signal = pytest.importorskip("scipy.signal")
+
+from client.audio import resample_24k_to_16k
+
+
+def test_length_and_dtype():
+    frame = np.zeros(1920, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    assert out.shape == (1280,)
+    assert out.dtype == np.int16
+
+
+def test_constant_signal_stays_constant():
+    frame = np.full(1920, 1000, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    # poly resample of a constant is the same constant up to filter ringing
+    assert np.abs(out.astype(int).mean() - 1000) < 5
+
+
+def test_silence_in_silence_out():
+    frame = np.zeros(1920, dtype=np.int16)
+    out = resample_24k_to_16k(frame)
+    assert int(np.max(np.abs(out))) == 0
+
+
+def test_sine_peak_frequency_preserved():
+    # 1 kHz sine at 24 kHz: 80 ms = 80 cycles.
+    t = np.arange(1920) / 24_000.0
+    sig = (8000 * np.sin(2 * np.pi * 1000 * t)).astype(np.int16)
+    out = resample_24k_to_16k(sig)
+    # 1 kHz in a 1280-sample 16 kHz window is bin (1000 / 16000) * 1280 = 80.
+    spec = np.abs(np.fft.rfft(out.astype(np.float32)))
+    peak = int(np.argmax(spec))
+    assert abs(peak - 80) <= 1
+
+
+def test_zero_length_input_raises():
+    with pytest.raises(ValueError):
+        resample_24k_to_16k(np.zeros(0, dtype=np.int16))
+
+
+def test_wrong_dtype_rejected():
+    with pytest.raises(TypeError):
+        resample_24k_to_16k(np.zeros(1920, dtype=np.float32))