diff --git a/observe/transcribe/main.py b/observe/transcribe/main.py index e8592911d..9b468db8e 100644 --- a/observe/transcribe/main.py +++ b/observe/transcribe/main.py @@ -61,7 +61,7 @@ from observe.transcribe import ( from observe.transcribe import transcribe as stt_transcribe from observe.transcribe.utils import is_apple_silicon from observe.transcribe.whisper import DEFAULT_COMPUTE, DEFAULT_DEVICE, DEFAULT_MODEL -from observe.utils import SAMPLE_RATE, get_segment_key +from observe.utils import SAMPLE_RATE, get_segment_key, load_audio from observe.vad import ( AudioReduction, VadResult, @@ -409,6 +409,7 @@ def _statements_to_jsonl( def process_audio( raw_path: Path, + audio_buffer: np.ndarray, vad_result: VadResult, backend_config: dict, redo: bool = False, @@ -428,6 +429,7 @@ def process_audio( Args: raw_path: Path to audio file in journal segment directory (HHMMSS_LEN/) + audio_buffer: Full audio waveform (float32 mono at SAMPLE_RATE) vad_result: Pre-computed VAD result from run_vad() backend_config: Configuration for STT backend redo: If True, skip "already processed" check @@ -436,8 +438,6 @@ def process_audio( backend: STT backend name (default: "whisper") entity_names: Optional list of entity names for STT and enrichment context """ - from faster_whisper.audio import decode_audio - start_time = time.time() # Derive segment from path @@ -461,15 +461,15 @@ def process_audio( except json.JSONDecodeError: logging.warning(f"Invalid SEGMENT_META JSON: {segment_meta_str[:100]}") - # Prepare audio buffer for processing + # Use reduced audio for STT if available, otherwise full buffer if reduced_audio is not None: - audio_buffer = reduced_audio + stt_buffer = reduced_audio else: - audio_buffer = decode_audio(str(raw_path), sampling_rate=SAMPLE_RATE) + stt_buffer = audio_buffer try: # Dispatch to STT backend - statements = stt_transcribe(backend, audio_buffer, SAMPLE_RATE, backend_config) + statements = stt_transcribe(backend, stt_buffer, SAMPLE_RATE, backend_config) # Get model info for metadata (dynamic import based on backend) backend_module = get_backend(backend) @@ -526,13 +526,13 @@ def process_audio( from observe.enrich import enrich_transcript enrichment = enrich_transcript( - audio_buffer, SAMPLE_RATE, statements, entity_names=entity_names + stt_buffer, SAMPLE_RATE, statements, entity_names=entity_names ) # Generate embeddings before timestamp restoration # Use reduced audio buffer if available for consistent timestamps embeddings_data = _embed_statements( - audio_buffer, statements, SAMPLE_RATE, model_info.get("device", "cpu") + stt_buffer, statements, SAMPLE_RATE, model_info.get("device", "cpu") ) # Restore original timestamps if audio was reduced @@ -650,8 +650,11 @@ def main(): logging.info(f"Processing audio: {audio_path}") + # Load audio once - handles M4A multi-stream mixing + audio_buffer = load_audio(audio_path) + # Stage 1: Run VAD to detect speech (lightweight, before loading STT model) - vad_result = run_vad(audio_path, min_speech_seconds=min_speech_seconds) + vad_result = run_vad(audio_buffer, min_speech_seconds=min_speech_seconds) # Early exit if no speech detected (skip loading heavy STT model) if not vad_result.has_speech: @@ -678,7 +681,7 @@ def main(): return # Stage 2: Reduce audio by trimming long silence gaps (>2s) - reduced_audio, reduction = reduce_audio(audio_path, vad_result) + reduced_audio, reduction = reduce_audio(audio_buffer, vad_result) # Stage 3: Determine backend and build backend config # CLI --backend flag overrides config, otherwise use config or default @@ -751,6 +754,7 @@ def main(): # Stage 4: Process audio with STT backend process_audio( audio_path, + audio_buffer, vad_result, backend_config, redo=args.redo, diff --git a/observe/utils.py b/observe/utils.py index 026c4a5b2..c9061eb90 100644 --- a/observe/utils.py +++ b/observe/utils.py @@ -47,16 +47,12 @@ def audio_to_flac_bytes(audio: np.ndarray, sample_rate: int) -> bytes: return buf.getvalue() -def prepare_audio_file(raw_path: Path, sample_rate: int = SAMPLE_RATE) -> Path: - """Prepare audio file for processing, converting M4A if needed. +def load_audio(raw_path: Path, sample_rate: int = SAMPLE_RATE) -> np.ndarray: + """Load audio file into a numpy buffer, mixing M4A streams if needed. - Returns path to a file suitable for transcription/embedding (mono FLAC). - For M4A files, converts to temporary FLAC, mixing all audio streams. - Other formats (.flac, .ogg, .opus) are returned as-is since faster-whisper - can decode them directly via ffmpeg. - - M4A files from sck-cli contain two mono streams: track 0 = system audio, - track 1 = microphone. Both are decoded and mixed together. + For M4A files from sck-cli (which contain two mono streams: track 0 = + system audio, track 1 = microphone), all streams are decoded and mixed + together. Other formats (.flac, .ogg, .opus) are decoded via ffmpeg. Parameters ---------- @@ -67,22 +63,22 @@ def prepare_audio_file(raw_path: Path, sample_rate: int = SAMPLE_RATE) -> Path: Returns ------- - Path - Path to audio file ready for processing. For most formats, returns - the original path. For .m4a files, returns path to temporary .flac - file that caller should delete after use. + np.ndarray + Audio waveform as float32 mono at the target sample rate Raises ------ ValueError If no audio streams found in M4A file """ - import av - if raw_path.suffix.lower() != ".m4a": - return raw_path + from faster_whisper.audio import decode_audio - logger.info(f"Converting m4a to FLAC: {raw_path}") + return decode_audio(str(raw_path), sampling_rate=sample_rate) + + import av + + logger.info(f"Loading m4a with stream mixing: {raw_path}") # First pass: count streams container = av.open(str(raw_path)) @@ -136,12 +132,7 @@ def prepare_audio_file(raw_path: Path, sample_rate: int = SAMPLE_RATE) -> Path: mixed = np.mean(padded, axis=0) logger.info(f" Mixed {len(stream_data)} streams -> {len(mixed)} samples") - # Write to temporary FLAC in same directory - temp_path = raw_path.with_suffix(".tmp.flac") - audio_int16 = (np.clip(mixed, -1.0, 1.0) * 32767).astype(np.int16) - sf.write(temp_path, audio_int16, sample_rate, format="FLAC") - - return temp_path + return mixed.astype(np.float32) def get_segment_key(media_path: Path) -> str | None: diff --git a/observe/vad.py b/observe/vad.py index ae69ea870..31c7d03f9 100644 --- a/observe/vad.py +++ b/observe/vad.py @@ -14,7 +14,7 @@ Uses Silero VAD via faster-whisper's bundled implementation. Audio Reduction: When there are long gaps (>2s) between speech segments, the audio can be "reduced" by trimming those gaps to a maximum of 2s (1s buffer on each side). -This creates a shorter audio file that any STT backend can process more +This creates a shorter audio buffer that any STT backend can process more efficiently. A mapping is preserved to restore original timestamps after transcription. """ @@ -24,8 +24,6 @@ from __future__ import annotations import logging import time from dataclasses import dataclass, field -from pathlib import Path - import numpy as np from observe.utils import SAMPLE_RATE @@ -226,32 +224,28 @@ class VadResult: def run_vad( - audio_path: Path, + audio: np.ndarray, min_speech_seconds: float, ) -> VadResult: - """Run Voice Activity Detection on an audio file. + """Run Voice Activity Detection on an audio buffer. - Loads the audio and runs Silero VAD to identify speech segments. - This can be used to filter silent files before loading heavier - transcription models. Also computes RMS of non-speech regions for - background noise detection. + Runs Silero VAD to identify speech segments. This can be used to + filter silent files before loading heavier transcription models. + Also computes RMS of non-speech regions for background noise detection. Args: - audio_path: Path to audio file (any format supported by ffmpeg/PyAV) + audio: Audio waveform (float32 mono at SAMPLE_RATE) min_speech_seconds: Minimum speech duration to set has_speech=True Returns: VadResult with duration info, has_speech flag, speech segment boundaries, and non-speech RMS level for noise detection """ - from faster_whisper.audio import decode_audio from faster_whisper.vad import VadOptions, get_speech_timestamps - logging.info(f"Running VAD on {audio_path.name}...") + logging.info("Running VAD...") t0 = time.perf_counter() - # Load audio at 16kHz mono (Silero VAD requirement) - audio = decode_audio(str(audio_path), sampling_rate=SAMPLE_RATE) duration = len(audio) / SAMPLE_RATE # Run Silero VAD with default options @@ -293,7 +287,7 @@ def run_vad( def reduce_audio( - audio_path: Path, + audio: np.ndarray, vad_result: VadResult, ) -> tuple[np.ndarray | None, AudioReduction | None]: """Reduce audio by trimming long silence gaps. @@ -303,7 +297,7 @@ def reduce_audio( that any STT backend can process more efficiently. Args: - audio_path: Path to original audio file + audio: Audio waveform (float32 mono at SAMPLE_RATE) vad_result: VAD result with speech segment boundaries Returns: @@ -311,8 +305,6 @@ def reduce_audio( - If no reduction needed (no gaps > 2s), returns (None, None) - Otherwise returns the reduced audio numpy array and the mapping """ - from faster_whisper.audio import decode_audio - if not vad_result.speech_segments: return None, None @@ -347,9 +339,6 @@ def reduce_audio( logging.info(" No gaps > 2s to reduce") return None, None - # Load audio for reduction - audio = decode_audio(str(audio_path), sampling_rate=SAMPLE_RATE) - # Build reduced audio by copying segments and trimmed gaps reduced_chunks = [] reduction_segments = [] diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index 356547e38..12dd33a4d 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -22,7 +22,7 @@ from observe.transcribe import ( build_statement, build_statements_from_acoustic, ) -from observe.utils import prepare_audio_file +from observe.utils import load_audio class TestBuildStatementsFromAcoustic: @@ -236,11 +236,11 @@ class TestConstants: assert DEFAULT_MIN_SPEECH_SECONDS == 1.0 -class TestPrepareAudioFile: - """Test the shared prepare_audio_file utility.""" +class TestLoadAudio: + """Test the shared load_audio utility.""" - def test_flac_passthrough(self): - """FLAC files should be returned unchanged.""" + def test_flac_returns_numpy_array(self): + """FLAC files should return a numpy array.""" with tempfile.TemporaryDirectory() as tmpdir: flac_path = Path(tmpdir) / "test.flac" @@ -249,12 +249,14 @@ class TestPrepareAudioFile: data = np.zeros(sample_rate, dtype=np.float32) sf.write(flac_path, data, sample_rate, format="FLAC") - result = prepare_audio_file(flac_path) - assert result == flac_path + result = load_audio(flac_path) + assert isinstance(result, np.ndarray) + assert result.dtype == np.float32 + assert len(result) == sample_rate @pytest.mark.skipif(not shutil.which("ffmpeg"), reason="ffmpeg not installed") - def test_m4a_conversion(self): - """M4A files should be converted to temp FLAC.""" + def test_m4a_returns_numpy_array(self): + """M4A files should return a numpy array with audio content.""" import subprocess with tempfile.TemporaryDirectory() as tmpdir: @@ -284,24 +286,15 @@ class TestPrepareAudioFile: ) assert result.returncode == 0 - # Test conversion - temp_flac = prepare_audio_file(m4a_path) - try: - assert temp_flac.exists() - assert temp_flac.suffix == ".flac" - assert temp_flac != m4a_path - - # Verify audio was extracted - mixed_data, sr = sf.read(temp_flac, dtype="float32") - assert sr == 16000 - assert len(mixed_data) > 0 - finally: - if temp_flac.exists(): - temp_flac.unlink() + # Test loading returns numpy array + audio = load_audio(m4a_path) + assert isinstance(audio, np.ndarray) + assert audio.dtype == np.float32 + assert len(audio) > 0 @pytest.mark.skipif(not shutil.which("ffmpeg"), reason="ffmpeg not installed") - def test_multi_track_m4a(self): - """Test that prepare_audio_file mixes multiple M4A audio streams together.""" + def test_multi_track_m4a_mixes_streams(self): + """load_audio should mix multiple M4A audio streams together.""" import subprocess with tempfile.TemporaryDirectory() as tmpdir: @@ -346,25 +339,15 @@ class TestPrepareAudioFile: ) assert result.returncode == 0, f"ffmpeg failed: {result.stderr}" - temp_flac = prepare_audio_file(m4a_path) + audio = load_audio(m4a_path) - try: - assert temp_flac.exists() - assert temp_flac.suffix == ".flac" + assert isinstance(audio, np.ndarray) + assert audio.dtype == np.float32 - # Read the output and verify both streams were mixed - mixed_data, sr = sf.read(temp_flac, dtype="float32") - - # The mixed audio should have content from track 1 (the sine wave) - # AAC compression affects amplitude, so use loose threshold - rms = np.sqrt(np.mean(mixed_data**2)) - assert rms > 0.1, f"Mixed audio should contain signal, got RMS={rms}" - - # Verify sample rate matches expected - assert sr == 16000 - finally: - if temp_flac.exists(): - temp_flac.unlink() + # The mixed audio should have content from track 1 (the sine wave) + # AAC compression affects amplitude, so use loose threshold + rms = np.sqrt(np.mean(audio**2)) + assert rms > 0.1, f"Mixed audio should contain signal, got RMS={rms}" class TestEmbeddingsFormat: diff --git a/tests/test_vad.py b/tests/test_vad.py index ceff2345c..6a28bc22f 100644 --- a/tests/test_vad.py +++ b/tests/test_vad.py @@ -3,7 +3,6 @@ """Tests for observe.vad module.""" -from pathlib import Path from unittest.mock import patch import numpy as np @@ -271,27 +270,25 @@ class TestRunVad: """Test run_vad function.""" @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_silent_audio_returns_no_speech(self, mock_decode, mock_get_timestamps): + def test_silent_audio_returns_no_speech(self, mock_get_timestamps): """Silent audio should return has_speech=False.""" - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) mock_get_timestamps.return_value = [] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.duration == 5.0 assert result.speech_duration == 0.0 assert result.has_speech is False @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_speech_audio_returns_has_speech(self, mock_decode, mock_get_timestamps): + def test_speech_audio_returns_has_speech(self, mock_get_timestamps): """Audio with speech should return has_speech=True.""" - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) # Mock: 2 seconds of speech (samples 16000-48000) mock_get_timestamps.return_value = [{"start": 16000, "end": 48000}] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.duration == 5.0 assert result.speech_duration == 2.0 @@ -300,112 +297,89 @@ class TestRunVad: assert result.speech_segments == [(1.0, 3.0)] @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_speech_below_threshold(self, mock_decode, mock_get_timestamps): + def test_speech_below_threshold(self, mock_get_timestamps): """Speech below threshold should return has_speech=False.""" - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) # Mock: 0.5 seconds of speech (below 1.0s threshold) mock_get_timestamps.return_value = [{"start": 0, "end": 8000}] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.duration == 5.0 assert result.speech_duration == 0.5 assert result.has_speech is False @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_custom_min_speech_threshold(self, mock_decode, mock_get_timestamps): + def test_custom_min_speech_threshold(self, mock_get_timestamps): """Custom min_speech_seconds threshold should be respected.""" - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) # Mock: 0.5 seconds of speech mock_get_timestamps.return_value = [{"start": 0, "end": 8000}] # With 0.3s threshold, should have speech - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=0.3) + result = run_vad(audio, min_speech_seconds=0.3) assert result.has_speech is True # With 1.0s threshold, should not have speech - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.has_speech is False @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_multiple_speech_chunks(self, mock_decode, mock_get_timestamps): + def test_multiple_speech_chunks(self, mock_get_timestamps): """Multiple speech chunks should be summed correctly.""" - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) # Mock: Two 1-second speech segments mock_get_timestamps.return_value = [ {"start": 16000, "end": 32000}, # 1 second {"start": 48000, "end": 64000}, # 1 second ] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.duration == 5.0 assert result.speech_duration == 2.0 assert result.has_speech is True @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_calls_decode_audio_correctly(self, mock_decode, mock_get_timestamps): - """run_vad should call decode_audio with correct parameters.""" - mock_decode.return_value = np.zeros(3 * SAMPLE_RATE, dtype=np.float32) - mock_get_timestamps.return_value = [{"start": 0, "end": 32000}] - - run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) - - mock_decode.assert_called_once_with( - "/fake/audio.flac", sampling_rate=SAMPLE_RATE - ) - - @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_returns_rms_for_silent_background(self, mock_decode, mock_get_timestamps): + def test_returns_rms_for_silent_background(self, mock_get_timestamps): """run_vad should return low RMS for silent non-speech regions.""" # Silent audio (zeros) - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) # Speech from 1-3s, leaving non-speech at 0-1s and 3-5s mock_get_timestamps.return_value = [{"start": 16000, "end": 48000}] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.noisy_rms is not None assert result.noisy_rms < 0.001 # Effectively zero assert result.noisy_s == 3.0 # 1s leading + 2s trailing @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_returns_rms_for_noisy_background(self, mock_decode, mock_get_timestamps): + def test_returns_rms_for_noisy_background(self, mock_get_timestamps): """run_vad should return measurable RMS for noisy non-speech regions.""" # Noisy audio np.random.seed(42) - mock_decode.return_value = np.random.uniform(-0.1, 0.1, 5 * SAMPLE_RATE).astype( - np.float32 - ) + audio = np.random.uniform(-0.1, 0.1, 5 * SAMPLE_RATE).astype(np.float32) # Speech from 1-3s mock_get_timestamps.return_value = [{"start": 16000, "end": 48000}] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=1.0) + result = run_vad(audio, min_speech_seconds=1.0) assert result.noisy_rms is not None assert result.noisy_rms > 0.01 # Noisy threshold assert result.noisy_s == 3.0 @patch("faster_whisper.vad.get_speech_timestamps") - @patch("faster_whisper.audio.decode_audio") - def test_returns_none_rms_when_no_qualifying_segments( - self, mock_decode, mock_get_timestamps - ): + def test_returns_none_rms_when_no_qualifying_segments(self, mock_get_timestamps): """run_vad should return None RMS when no qualifying non-speech segments.""" - mock_decode.return_value = np.zeros(2 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(2 * SAMPLE_RATE, dtype=np.float32) # Speech fills most of audio, leaving only 0.2s gaps (below 0.5s threshold) mock_get_timestamps.return_value = [ {"start": 3200, "end": 12800}, # 0.2s to 0.8s {"start": 16000, "end": 28800}, # 1.0s to 1.8s ] - result = run_vad(Path("/fake/audio.flac"), min_speech_seconds=0.5) + result = run_vad(audio, min_speech_seconds=0.5) assert result.noisy_rms is None assert result.noisy_s == 0.0 @@ -716,11 +690,10 @@ class TestRestoreSegmentTimestamps: class TestReduceAudio: """Test reduce_audio function.""" - @patch("faster_whisper.audio.decode_audio") - def test_no_gaps_to_reduce(self, mock_decode): + def test_no_gaps_to_reduce(self): """Should return None when no gaps > 2s exist.""" # 5s audio with speech from 0.5-1.5s and 2.0-3.0s (gap = 0.5s < 2s) - mock_decode.return_value = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=5.0, @@ -729,14 +702,15 @@ class TestReduceAudio: speech_segments=[(0.5, 1.5), (2.0, 3.0)], ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is None assert reduction is None - @patch("faster_whisper.audio.decode_audio") - def test_no_speech_segments(self, mock_decode): + def test_no_speech_segments(self): """Should return None when no speech segments.""" + audio = np.zeros(5 * SAMPLE_RATE, dtype=np.float32) + vad_result = VadResult( duration=5.0, speech_duration=0.0, @@ -744,16 +718,15 @@ class TestReduceAudio: speech_segments=[], ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is None assert reduction is None - @patch("faster_whisper.audio.decode_audio") - def test_leading_gap_reduction(self, mock_decode): + def test_leading_gap_reduction(self): """Should trim leading gap > 2s to GAP_BUFFER.""" # 10s audio with speech starting at 5s (leading gap = 5s > 2s) - mock_decode.return_value = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=10.0, @@ -762,7 +735,7 @@ class TestReduceAudio: speech_segments=[(5.0, 8.0)], # Speech from 5-8s ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is not None assert reduction is not None @@ -778,11 +751,10 @@ class TestReduceAudio: assert reduction.segments[0].original_start == 5.0 assert reduction.segments[0].reduced_start == GAP_BUFFER - @patch("faster_whisper.audio.decode_audio") - def test_trailing_gap_reduction(self, mock_decode): + def test_trailing_gap_reduction(self): """Should trim trailing gap > 2s to GAP_BUFFER.""" # 10s audio with speech from 1-3s (trailing gap = 7s > 2s) - mock_decode.return_value = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=10.0, @@ -791,7 +763,7 @@ class TestReduceAudio: speech_segments=[(1.0, 3.0)], # Speech from 1-3s ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is not None assert reduction is not None @@ -801,11 +773,10 @@ class TestReduceAudio: actual_duration = len(reduced_audio) / SAMPLE_RATE assert abs(actual_duration - expected_duration) < 0.1 - @patch("faster_whisper.audio.decode_audio") - def test_middle_gap_reduction(self, mock_decode): + def test_middle_gap_reduction(self): """Should trim middle gap > 2s to 2*GAP_BUFFER.""" # 10s audio with speech at 0-2s and 7-9s (gap = 5s > 2s) - mock_decode.return_value = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=10.0, @@ -814,7 +785,7 @@ class TestReduceAudio: speech_segments=[(0.0, 2.0), (7.0, 9.0)], ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is not None assert reduction is not None @@ -832,11 +803,10 @@ class TestReduceAudio: # Second segment should start at: speech1_end + trimmed_gap = 2.0 + 2.0 = 4.0 assert abs(reduction.segments[1].reduced_start - 4.0) < 0.1 - @patch("faster_whisper.audio.decode_audio") - def test_multiple_gaps_reduction(self, mock_decode): + def test_multiple_gaps_reduction(self): """Should trim multiple gaps > 2s.""" # 20s audio with speech at 5-7, 12-14, and 19-20 (two big gaps) - mock_decode.return_value = np.zeros(20 * SAMPLE_RATE, dtype=np.float32) + audio = np.zeros(20 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=20.0, @@ -845,7 +815,7 @@ class TestReduceAudio: speech_segments=[(5.0, 7.0), (12.0, 14.0), (19.0, 20.0)], ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert reduced_audio is not None assert reduction is not None @@ -867,10 +837,9 @@ class TestReduceAudio: # Check we have 3 speech segments in mapping assert len(reduction.segments) == 3 - @patch("faster_whisper.audio.decode_audio") - def test_returns_numpy_array(self, mock_decode): - """Should return numpy array, not file path.""" - mock_decode.return_value = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) + def test_returns_numpy_array(self): + """Should return numpy array.""" + audio = np.zeros(10 * SAMPLE_RATE, dtype=np.float32) vad_result = VadResult( duration=10.0, @@ -879,7 +848,7 @@ class TestReduceAudio: speech_segments=[(5.0, 7.0)], # Leading gap > 2s ) - reduced_audio, reduction = reduce_audio(Path("/fake/audio.flac"), vad_result) + reduced_audio, reduction = reduce_audio(audio, vad_result) assert isinstance(reduced_audio, np.ndarray) assert reduced_audio.dtype == np.float32