Update core/silero_vad.py

core/silero_vad.py

@@ -1,243 +1,4 @@
-# import torch
-# import numpy as np
-# from typing import Optional, Callable
-# from config.settings import settings
-# import os
-
-# class SileroVAD:
-#     def __init__(self):
-#         self.model = None
-#         self.sample_rate = 16000
-#         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 sử dụng torch.hub"""
-#         try:
-#             print("🔄 Đang tải Silero VAD model từ torch.hub...")
-            
-#             # Sử dụng torch.hub để load model (cách chính thức)
-#             self.model = torch.hub.load(
-#                 repo_or_dir=settings.VAD_MODEL,
-#                 model='silero_vad',
-#                 force_reload=False,  # Sử dụng cache nếu có
-#                 trust_repo=True
-#             )
-            
-#             print("✅ Đã tải Silero VAD model thành công")
-            
-#         except Exception as e:
-#             print(f"❌ Lỗi tải Silero VAD model: {e}")
-#             print("🔄 Đang thử cách tải thay thế...")
-#             self._initialize_model_fallback()
-
-#     def _initialize_model_fallback(self):
-#         """Fallback method nếu cách chính thức không hoạt động"""
-#         try:
-#             # Cách 2: Sử dụng direct download
-#             model_urls = {
-#                 'silero_vad.jit': 'https://github.com/snakers4/silero-vad/raw/master/files/silero_vad.jit'
-#             }
-            
-#             # Tạo thư mục cache
-#             os.makedirs('./models', exist_ok=True)
-#             model_path = './models/silero_vad.jit'
-            
-#             if not os.path.exists(model_path):
-#                 print("📥 Đang download Silero VAD model...")
-#                 torch.hub.download_url_to_file(
-#                     model_urls['silero_vad.jit'],
-#                     model_path
-#                 )
-            
-#             # Load model
-#             self.model = torch.jit.load(model_path)
-#             self.model.eval()
-#             print("✅ Đã tải Silero VAD model thành công (fallback)")
-            
-#         except Exception as e:
-#             print(f"❌ Lỗi tải Silero VAD model fallback: {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
-#             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)
-#             buffer_duration = len(self.audio_buffer) / self.sample_rate
-#             if buffer_duration >= 1.0:
-#                 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:
-#             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)
-#                 if np.max(np.abs(audio_chunk)) > 1.0:
-#                     audio_chunk = audio_chunk / 32768.0  # Normalize từ int16
-            
-#             # Đảm bảo audio trong range [-1, 1]
-#             audio_chunk = np.clip(audio_chunk, -1.0, 1.0)
-            
-#             # Chuyển thành tensor
-#             audio_tensor = torch.from_numpy(audio_chunk).float().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
-#             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
-#             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)
-#                 if np.max(np.abs(audio_chunk)) > 1.0:
-#                     audio_chunk = audio_chunk / 32768.0
-            
-#             audio_chunk = np.clip(audio_chunk, -1.0, 1.0)
-            
-#             # Đảm bảo độ dài phù hợp
-#             if len(audio_chunk) < 512:
-#                 padding = np.zeros(512 - len(audio_chunk), dtype=np.float32)
-#                 audio_chunk = np.concatenate([audio_chunk, padding])
-            
-#             # Chuyển thành tensor
-#             audio_tensor = torch.from_numpy(audio_chunk).float().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"""
-#         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)
-#                 if np.max(np.abs(audio_chunk)) > 1.0:
-#                     audio_chunk = audio_chunk / 32768.0
-            
-#             audio_chunk = np.clip(audio_chunk, -1.0, 1.0)
-            
-#             # Đảm bảo độ dài phù hợp
-#             if len(audio_chunk) < 512:
-#                 padding = np.zeros(512 - len(audio_chunk), dtype=np.float32)
-#                 audio_chunk = np.concatenate([audio_chunk, padding])
-            
-#             # Chuyển thành tensor
-#             audio_tensor = torch.from_numpy(audio_chunk).float().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.0import torch
+
 import torch
 import numpy as np
 from typing import Callable