Create silero_vad

core/silero_vad

@@ -0,0 +1,194 @@
+import torch
+import numpy as np
+from typing import Optional, Callable
+from config.settings import settings
+
+class SileroVAD:
+    def __init__(self):
+        self.model = None
+        self.sample_rate = 16000  # Silero VAD yêu cầu 16kHz
+        self.is_streaming = False
+        self.speech_callback = None
+        self.audio_buffer = []
+        self._initialize_model()
+
+    def _initialize_model(self):
+        """Khởi tạo Silero VAD model"""
+        try:
+            print("🔄 Đang tải Silero VAD model...")
+            torch.hub.download_url_to_file(
+                'https://raw.githubusercontent.com/snakers4/silero-vad/master/files/model.jit',
+                'silero_vad.jit'
+            )
+            self.model = torch.jit.load('silero_vad.jit')
+            self.model.eval()
+            print("✅ Đã tải Silero VAD model thành công")
+        except Exception as e:
+            print(f"❌ Lỗi tải Silero VAD model: {e}")
+            self.model = None
+
+    def start_stream(self, speech_callback: Callable):
+        """Bắt đầu stream với VAD"""
+        if self.model is None:
+            print("❌ Silero VAD model chưa được khởi tạo")
+            return False
+        
+        self.is_streaming = True
+        self.speech_callback = speech_callback
+        self.audio_buffer = []
+        print("🎙️ Bắt đầu Silero VAD streaming...")
+        return True
+
+    def stop_stream(self):
+        """Dừng stream"""
+        self.is_streaming = False
+        self.speech_callback = None
+        self.audio_buffer = []
+        print("🛑 Đã dừng Silero VAD streaming")
+
+    def process_stream(self, audio_chunk: np.ndarray, sample_rate: int):
+        """Xử lý audio chunk với Silero VAD"""
+        if not self.is_streaming or self.model is None:
+            return
+
+        try:
+            # Resample nếu cần (Silero yêu cầu 16kHz)
+            if sample_rate != self.sample_rate:
+                audio_chunk = self._resample_audio(audio_chunk, sample_rate, self.sample_rate)
+
+            # Thêm vào buffer
+            self.audio_buffer.extend(audio_chunk)
+
+            # Xử lý khi buffer đủ lớn (1 giây - Silero làm việc tốt với chunk nhỏ)
+            buffer_duration = len(self.audio_buffer) / self.sample_rate
+            if buffer_duration >= 1.0:  # Giảm từ 2.0 xuống 1.0 giây
+                self._process_buffer()
+
+        except Exception as e:
+            print(f"❌ Lỗi xử lý Silero VAD: {e}")
+
+    def _process_buffer(self):
+        """Xử lý buffer audio với Silero VAD"""
+        try:
+            # Silero VAD làm việc tốt với chunk 1 giây
+            chunk_size = self.sample_rate  # 1 giây
+            if len(self.audio_buffer) < chunk_size:
+                return
+
+            # Lấy chunk 1 giây
+            audio_chunk = np.array(self.audio_buffer[:chunk_size])
+            
+            # Chuẩn hóa audio cho Silero
+            if audio_chunk.dtype != np.float32:
+                audio_chunk = audio_chunk.astype(np.float32) / 32768.0  # Normalize to [-1, 1]
+            
+            # Chuyển thành tensor
+            audio_tensor = torch.from_numpy(audio_chunk).unsqueeze(0)
+            
+            # Phát hiện speech với Silero VAD
+            with torch.no_grad():
+                speech_prob = self.model(audio_tensor, self.sample_rate).item()
+            
+            print(f"🎯 Silero VAD speech probability: {speech_prob:.3f}")
+            
+            # Ngưỡng phát hiện speech (có thể điều chỉnh)
+            if speech_prob > settings.VAD_THRESHOLD:
+                print(f"🎯 Silero VAD phát hiện speech: {speech_prob:.3f}")
+                
+                # Gọi callback với speech segment
+                if self.speech_callback:
+                    self.speech_callback(audio_chunk, self.sample_rate)
+
+            # Giữ lại 0.3 giây cuối để overlap (Silero nhạy hơn)
+            keep_samples = int(self.sample_rate * 0.3)
+            if len(self.audio_buffer) > keep_samples:
+                self.audio_buffer = self.audio_buffer[-keep_samples:]
+            else:
+                self.audio_buffer = []
+
+        except Exception as e:
+            print(f"❌ Lỗi xử lý Silero VAD buffer: {e}")
+            self.audio_buffer = []
+
+    def _resample_audio(self, audio: np.ndarray, orig_sr: int, target_sr: int) -> np.ndarray:
+        """Resample audio nếu cần"""
+        if orig_sr == target_sr:
+            return audio
+        
+        try:
+            # Simple resampling bằng interpolation
+            orig_length = len(audio)
+            new_length = int(orig_length * target_sr / orig_sr)
+            
+            # Linear interpolation
+            x_old = np.linspace(0, 1, orig_length)
+            x_new = np.linspace(0, 1, new_length)
+            resampled_audio = np.interp(x_new, x_old, audio)
+            
+            return resampled_audio
+        except Exception as e:
+            print(f"⚠️ Lỗi resample: {e}")
+            return audio
+
+    def is_speech(self, audio_chunk: np.ndarray, sample_rate: int) -> bool:
+        """Kiểm tra xem audio chunk có phải là speech không"""
+        if self.model is None:
+            return True  # Fallback: luôn coi là speech
+            
+        try:
+            # Resample nếu cần
+            if sample_rate != self.sample_rate:
+                audio_chunk = self._resample_audio(audio_chunk, sample_rate, self.sample_rate)
+            
+            # Chuẩn hóa audio
+            if audio_chunk.dtype != np.float32:
+                audio_chunk = audio_chunk.astype(np.float32) / 32768.0
+            
+            # Đảm bảo độ dài phù hợp
+            if len(audio_chunk) < 512:  # Silero cần ít nhất 512 samples
+                padding = np.zeros(512 - len(audio_chunk))
+                audio_chunk = np.concatenate([audio_chunk, padding])
+            
+            # Chuyển thành tensor
+            audio_tensor = torch.from_numpy(audio_chunk).unsqueeze(0)
+            
+            # Phát hiện speech
+            with torch.no_grad():
+                speech_prob = self.model(audio_tensor, self.sample_rate).item()
+            
+            # Kiểm tra ngưỡng
+            return speech_prob > settings.VAD_THRESHOLD
+            
+        except Exception as e:
+            print(f"❌ Lỗi kiểm tra speech với Silero: {e}")
+            return True
+
+    def get_speech_probability(self, audio_chunk: np.ndarray, sample_rate: int) -> float:
+        """Lấy xác suất speech (dùng cho debugging)"""
+        if self.model is None:
+            return 0.0
+            
+        try:
+            # Resample nếu cần
+            if sample_rate != self.sample_rate:
+                audio_chunk = self._resample_audio(audio_chunk, sample_rate, self.sample_rate)
+            
+            # Chuẩn hóa audio
+            if audio_chunk.dtype != np.float32:
+                audio_chunk = audio_chunk.astype(np.float32) / 32768.0
+            
+            # Đảm bảo độ dài phù hợp
+            if len(audio_chunk) < 512:
+                padding = np.zeros(512 - len(audio_chunk))
+                audio_chunk = np.concatenate([audio_chunk, padding])
+            
+            # Chuyển thành tensor
+            audio_tensor = torch.from_numpy(audio_chunk).unsqueeze(0)
+            
+            # Phát hiện speech
+            with torch.no_grad():
+                return self.model(audio_tensor, self.sample_rate).item()
+            
+        except Exception as e:
+            print(f"❌ Lỗi lấy speech probability: {e}")
+            return 0.0