fix: handle non-16kHz audio correctly in VideoClipper

funclip/videoclipper.py

@@ -134,12 +134,16 @@ def recog(self, audio_input, sd_switch='no', state=None, hotwords="", output_dir
         # Convert to float64 consistently (includes data type checking)
         data = convert_pcm_to_float(data)
 
-        # assert sr == 16000, "16kHz sample rate required, {} given.".format(sr)
-        if sr != 16000: # resample with librosa
-            data = librosa.resample(data, orig_sr=sr, target_sr=16000)
+        # Gradio stereo audio is shaped (samples, channels). Select the
+        # supported first channel before resampling so librosa operates on the
+        # sample axis rather than the channel axis.
         if len(data.shape) == 2:  # multi-channel wav input
             logging.warning("Input wav shape: {}, only first channel reserved.".format(data.shape))
             data = data[:,0]
+
+        if sr != 16000: # resample with librosa
+            data = librosa.resample(data, orig_sr=sr, target_sr=16000)
+            sr = 16000
         state['audio_input'] = (sr, data)
         if sd_switch == 'Yes':
             rec_result = self.funasr_model.generate(data, 
@@ -567,4 +571,4 @@ def main(cmd=None):
 
 
 if __name__ == '__main__':
-    main()
+    main()

tests/test_audio_normalization.py

@@ -0,0 +1,74 @@
+import sys
+from pathlib import Path
+
+import librosa
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "funclip"))
+
+from videoclipper import VideoClipper
+
+
+class CaptureASR:
+    def generate(self, data, **kwargs):
+        self.data = np.array(data, copy=True)
+        return [{
+            "text": "test",
+            "raw_text": "test",
+            "timestamp": [[0, 500]],
+            "sentence_info": [],
+        }]
+
+
+def _make_audio(rate, channels, dtype):
+    t = np.arange(rate) / rate
+    left = 0.2 * np.sin(2 * np.pi * 440 * t)
+    right = 0.3 * np.sin(2 * np.pi * 880 * t)
+    audio = left if channels == 1 else np.stack([left, right], axis=1)
+    if dtype == np.int16:
+        audio = np.rint(audio * 32767).astype(np.int16)
+    else:
+        audio = audio.astype(dtype)
+    return audio
+
+
+def _expected_first_channel(audio, rate):
+    data = audio[:, 0] if audio.ndim == 2 else audio
+    if data.dtype == np.int16:
+        data = data.astype(np.float64) / 32768.0
+    else:
+        data = data.astype(np.float64)
+    if rate != 16000:
+        data = librosa.resample(data, orig_sr=rate, target_sr=16000)
+    return data
+
+
+def test_recog_normalizes_supported_sample_rates_channels_and_dtypes():
+    for rate in (8000, 16000, 44100, 48000):
+        for channels in (1, 2):
+            for dtype in (np.float32, np.int16):
+                audio = _make_audio(rate, channels, dtype)
+                original = audio.copy()
+                expected = _expected_first_channel(audio, rate)
+                model = CaptureASR()
+                clipper = VideoClipper(model)
+                clipper.lang = "en"
+
+                _, _, state = clipper.recog((rate, audio))
+
+                stored_rate, stored_audio = state["audio_input"]
+                assert stored_rate == 16000
+                assert stored_audio.ndim == 1
+                assert len(stored_audio) == 16000
+                assert len(model.data) == 16000
+                np.testing.assert_array_equal(audio, original)
+                np.testing.assert_allclose(model.data, expected, rtol=1e-7, atol=1e-7)
+                np.testing.assert_allclose(stored_audio, expected, rtol=1e-7, atol=1e-7)
+
+                (out_rate, out), _, _ = clipper.clip(
+                    "", 0, 0, state, timestamp_list=[[0, 8000]]
+                )
+                assert out_rate == 16000
+                assert len(out) == 8000
+                assert len(out) / out_rate == 0.5
+                np.testing.assert_allclose(out, expected[:8000], rtol=1e-7, atol=1e-7)