import asyncio import numpy as np import sounddevice as sd from collections import deque import threading import time from queue import Queue import warnings warnings.filterwarnings("ignore") try: from faster_whisper import WhisperModel FASTER_WHISPER_AVAILABLE = True except ImportError: print("Installing faster-whisper: pip install faster-whisper") FASTER_WHISPER_AVAILABLE = False class RealtimeWhisperVAD: def __init__(self, whisper_model="base", #Can choose base, medium, small, etc sample_rate=16000, chunk_duration=0.5, # seconds vad_threshold=0.5, min_speech_duration=0.25, max_speech_duration=30, silence_duration=0.5): self.sample_rate = sample_rate self.chunk_size = int(sample_rate * chunk_duration) self.vad_threshold = vad_threshold self.min_speech_samples = int(sample_rate * min_speech_duration) self.max_speech_samples = int(sample_rate * max_speech_duration) self.silence_samples = int(sample_rate * silence_duration) if FASTER_WHISPER_AVAILABLE: self.whisper_model = WhisperModel( whisper_model, device="cpu", compute_type="float32" #float 16 if cuda ) self.use_faster_whisper = True else: print("Install faster whisper") return self.vad_model = None # Audio buffers self.audio_buffer = deque(maxlen=self.max_speech_samples) self.speech_buffer = [] self.silence_counter = 0 self.is_speech = False # Threading self.audio_queue = Queue() self.transcription_queue = Queue() self.running = False def energy_vad(self, audio_chunk): """Simple energy-based VAD fallback""" energy = np.mean(audio_chunk ** 2) return energy > 0.001 def transcribe_audio(self, audio_data): """Transcribe audio using Whisper""" try: if len(audio_data) < self.min_speech_samples: return "" # Normalize audio audio_normalized = audio_data.astype(np.float32) if np.max(np.abs(audio_normalized)) > 0: audio_normalized = audio_normalized / np.max(np.abs(audio_normalized)) segments, _ = self.whisper_model.transcribe( audio_normalized, language="en", # Specify language for better performance beam_size=1, # Faster inference best_of=1, temperature=0.0, vad_filter=True, # We're doing our own VAD(need to change later for now yes.) word_timestamps=False ) text = " ".join([segment.text for segment in segments]).strip() return text except Exception as e: print(f"Transcription error: {e}") return "" def audio_callback(self, indata, frames, time, status): """Audio input callback""" if status: print(f"Audio callback status: {status}") # Convert to mono if stereo if len(indata.shape) > 1: audio_chunk = indata[:, 0] else: audio_chunk = indata.flatten() self.audio_queue.put(audio_chunk.copy()) def process_audio(self): """Process audio chunks for VAD and transcription""" while self.running: try: if not self.audio_queue.empty(): audio_chunk = self.audio_queue.get(timeout=0.1) # VAD detection has_speech = self.energy_vad(audio_chunk) if has_speech: if not self.is_speech: print("šŸŽ¤ Speech detected") self.is_speech = True self.speech_buffer = [] # Add to speech buffer self.speech_buffer.extend(audio_chunk) self.silence_counter = 0 # Prevent buffer overflow if len(self.speech_buffer) > self.max_speech_samples: # Transcribe current buffer audio_array = np.array(self.speech_buffer[-self.max_speech_samples:]) self.transcribe_and_output(audio_array) # Keep some overlap self.speech_buffer = self.speech_buffer[-self.sample_rate:] else: # No speech if self.is_speech: self.silence_counter += len(audio_chunk) # Add silence to buffer (helps with word boundaries) if self.silence_counter < self.silence_samples: self.speech_buffer.extend(audio_chunk) # End of speech segment if self.silence_counter >= self.silence_samples: print("šŸ”‡ Speech ended") if len(self.speech_buffer) >= self.min_speech_samples: audio_array = np.array(self.speech_buffer) self.transcribe_and_output(audio_array) self.is_speech = False self.speech_buffer = [] self.silence_counter = 0 else: time.sleep(0.01) except Exception as e: print(f"Audio processing error: {e}") time.sleep(0.01) def transcribe_and_output(self, audio_array): """Transcribe audio and output result""" def transcribe(): text = self.transcribe_audio(audio_array) if text: timestamp = time.strftime("%H:%M:%S") print(f"[{timestamp}] {text}") # Run transcription in separate thread to avoid blocking threading.Thread(target=transcribe, daemon=True).start() def start_realtime(self): """Start real-time transcription""" print("šŸš€ Starting real-time Whisper transcription with Silero VAD") print(f"Model: {'faster-whisper' if self.use_faster_whisper else 'whisper'}") print(f"VAD: {'Silero VAD' if self.vad_model else 'Energy VAD'}") print(f"Sample rate: {self.sample_rate} Hz") print("Press Ctrl+C to stop\n") self.running = True # Start audio processing thread audio_thread = threading.Thread(target=self.process_audio, daemon=True) audio_thread.start() try: # Start audio stream with sd.InputStream( channels=1, samplerate=self.sample_rate, blocksize=self.chunk_size, callback=self.audio_callback, dtype=np.float32 ): print("šŸŽ§ Listening... Speak into your microphone") while self.running: time.sleep(0.1) except KeyboardInterrupt: print("\nā¹ļø Stopping transcription...") except Exception as e: print(f"Stream error: {e}") finally: self.running = False def main(): # Configuration config = { "whisper_model": "tiny", # tiny, base, small, medium, large "sample_rate": 16000, # 16kHz is optimal for Whisper "chunk_duration": 0.3, # Process every 300ms "vad_threshold": 0.5, # VAD sensitivity (0.0-1.0) "min_speech_duration": 0.3, # Minimum speech length to process "max_speech_duration": 5, # Maximum continuous speech "silence_duration": 0.8, # Silence before ending speech segment } # Initialize and start transcriber = RealtimeWhisperVAD(**config) transcriber.start_realtime() if __name__ == "__main__": main() # Installation requirements: # pip install sounddevice numpy # pip install faster-whisper