Spaces:
Running
Running
| """DEPRECATED: Real-time VAD-based speech segmentation using sounddevice and silero-vad. | |
| NOTE: This module is currently unused. The Streamlit app uses silero-vad | |
| directly in audio_preprocess.py (get_speech_timestamps / collect_chunks) | |
| instead of this class-based sounddevice integration. Keep for future use | |
| if real-time microphone streaming is needed.""" | |
| from __future__ import annotations | |
| import io | |
| import threading | |
| import time | |
| from collections import deque | |
| from typing import Callable | |
| import numpy as np | |
| import sounddevice as sd | |
| class VADSpeechProcessor: | |
| """Continuously listen to microphone, detect speech segments via VAD.""" | |
| def __init__( | |
| self, | |
| sample_rate: int = 16000, | |
| chunk_duration: float = 0.03, # 30ms chunks for VAD | |
| speech_pad_duration: float = 0.3, # pad speech segments | |
| min_speech_duration: float = 0.5, # minimum speech to process | |
| max_silence_duration: float = 1.0, # silence to end segment | |
| ) -> None: | |
| self.sample_rate = sample_rate | |
| self.chunk_samples = int(sample_rate * chunk_duration) | |
| self.speech_pad_samples = int(sample_rate * speech_pad_duration) | |
| self.min_speech_samples = int(sample_rate * min_speech_duration) | |
| self.max_silence_samples = int(sample_rate * max_silence_duration) | |
| self._vad_model = None | |
| self._buffer = deque() | |
| self._speech_segments = deque() | |
| self._is_listening = False | |
| self._stream = None | |
| self._lock = threading.Lock() | |
| self._process_thread = None | |
| def _get_vad_model(self): | |
| if self._vad_model is None: | |
| from silero_vad import load_silero_vad | |
| self._vad_model = load_silero_vad() | |
| return self._vad_model | |
| def _audio_callback(self, indata, frames, time_info, status): | |
| if status: | |
| print(f"Audio callback status: {status}") | |
| audio_chunk = indata[:, 0].astype(np.float32) # mono | |
| with self._lock: | |
| self._buffer.append(audio_chunk) | |
| def start_listening(self): | |
| if self._is_listening: | |
| return | |
| self._is_listening = True | |
| self._stream = sd.InputStream( | |
| samplerate=self.sample_rate, | |
| channels=1, | |
| dtype=np.float32, | |
| blocksize=self.chunk_samples, | |
| callback=self._audio_callback, | |
| ) | |
| self._stream.start() | |
| # Start processing thread | |
| self._process_thread = threading.Thread(target=self._process_loop, daemon=True) | |
| self._process_thread.start() | |
| def stop_listening(self): | |
| self._is_listening = False | |
| if self._stream: | |
| self._stream.stop() | |
| self._stream.close() | |
| self._stream = None | |
| def _process_loop(self): | |
| """Process audio buffer, detect speech segments.""" | |
| vad = self._get_vad_model() | |
| audio_buffer = np.array([], dtype=np.float32) | |
| is_speech = False | |
| speech_start = 0 | |
| silence_samples = 0 | |
| while self._is_listening: | |
| with self._lock: | |
| chunks = list(self._buffer) | |
| self._buffer.clear() | |
| if not chunks: | |
| time.sleep(0.01) | |
| continue | |
| for chunk in chunks: | |
| audio_buffer = np.concatenate([audio_buffer, chunk]) | |
| # Run VAD on chunk | |
| speech_prob = vad(chunk, self.sample_rate) | |
| chunk_is_speech = speech_prob > 0.5 | |
| if chunk_is_speech and not is_speech: | |
| # Speech started | |
| is_speech = True | |
| speech_start = max(0, len(audio_buffer) - len(chunk) - self.speech_pad_samples) | |
| silence_samples = 0 | |
| elif not chunk_is_speech and is_speech: | |
| silence_samples += len(chunk) | |
| if silence_samples > self.max_silence_samples: | |
| # Speech ended | |
| speech_end = len(audio_buffer) - silence_samples + self.speech_pad_samples | |
| segment = audio_buffer[speech_start:speech_end] | |
| if len(segment) >= self.min_speech_samples: | |
| self._speech_segments.append(segment) | |
| # Reset buffer to keep only recent audio | |
| audio_buffer = audio_buffer[speech_end:] | |
| is_speech = False | |
| silence_samples = 0 | |
| def get_speech_segments(self) -> list[np.ndarray]: | |
| """Get detected speech segments and clear queue.""" | |
| segments = [] | |
| while self._speech_segments: | |
| segments.append(self._speech_segments.popleft()) | |
| return segments | |
| def is_listening(self) -> bool: | |
| return self._is_listening | |