backend/voice/vad.py

# backend/voice/vad.py
# Real Voice Activity Detection using Silero VAD
import numpy as np

class SileroVAD:
    def __init__(self):
        import torch
        
        # Monkey patch torch.load for Silero VAD (moved from main.py to enforce lazy-loading)
        if not hasattr(torch, '_jarvis_patched'):
            _orig_load = torch.load
            def _safe_load(*args, **kwargs):
                kwargs["weights_only"] = False
                return _orig_load(*args, **kwargs)
            torch.load = _safe_load
            torch._jarvis_patched = True

        try:
            # Download real Silero VAD model on first run
            self.model, utils = torch.hub.load(
                repo_or_dir='snakers4/silero-vad',
                model='silero_vad',
                force_reload=False,
                trust_repo=True
            )
            (self.get_speech_timestamps, _, self.read_audio,
             self.VADIterator, self.collect_chunks) = utils
            self.iterator = self.VADIterator(self.model, sampling_rate=16000)
            self.available = True
        except Exception as e:
            import logging
            logging.error(f"SileroVAD failed to load from torch.hub: {e}")
            self.available = False
            self.iterator = None

    def process_chunk(self, audio_chunk: np.ndarray) -> dict:
        if not getattr(self, 'available', False) or not self.iterator:
            return {}
            
        import logging
        try:
            # Returns {"start": int, "end": int} if speech boundaries found, else {}
            return self.iterator(audio_chunk, return_seconds=False) or {}
        except Exception as e:
            logging.warning(
                "SileroVAD dropped corrupt audio chunk", 
                extra={"service": "silerovad", "error": str(e), "chunk_size": len(audio_chunk)}
            )
            return {}