backend/whispercode/main.py

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