GitHub-faithful implementation - 32kHz, 2048 FFT, per-model delays, 80ms gaps

app.py

@@ -201,15 +201,30 @@ class OptimizedEPANNs:
             if len(audio.shape) > 1:
                 audio = audio.mean(axis=1)
             
+            # Convert audio to target sample rate for E-PANNs
             if LIBROSA_AVAILABLE:
-                mel_spec = librosa.feature.melspectrogram(y=audio, sr=self.sample_rate, n_mels=64)
+                # Resample to E-PANNs sample rate if needed
+                audio_resampled = librosa.resample(audio.astype(float), 
+                                                 orig_sr=16000, 
+                                                 target_sr=self.sample_rate)
+                
+                mel_spec = librosa.feature.melspectrogram(y=audio_resampled, sr=self.sample_rate, n_mels=64)
                 energy = np.mean(librosa.power_to_db(mel_spec, ref=np.max))
-                spectral_centroid = np.mean(librosa.feature.spectral_centroid(y=audio, sr=self.sample_rate))
-                speech_score = (energy + 100) / 50 + spectral_centroid / 10000
+                spectral_centroid = np.mean(librosa.feature.spectral_centroid(y=audio_resampled, sr=self.sample_rate))
+                
+                # Better speech detection using multiple features
+                mfcc = librosa.feature.mfcc(y=audio_resampled, sr=self.sample_rate, n_mfcc=13)
+                mfcc_var = np.var(mfcc, axis=1).mean()
+                
+                # Combine features for better speech detection
+                speech_score = ((energy + 80) / 40) * 0.4 + (spectral_centroid / 5000) * 0.3 + (mfcc_var / 100) * 0.3
             else:
                 from scipy import signal
-                f, t, Sxx = signal.spectrogram(audio, self.sample_rate)
+                # Basic fallback without librosa
+                f, t, Sxx = signal.spectrogram(audio, 16000)  # Use original sample rate
                 energy = np.mean(10 * np.log10(Sxx + 1e-10))
+                
+                # Simple energy-based detection as fallback
                 speech_score = (energy + 100) / 50
             
             probability = np.clip(speech_score, 0, 1)
@@ -227,7 +242,6 @@ class OptimizedPANNs:
         self.sample_rate = 32000
         self.model = None
         self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-        self.cached_clip_prob = None
         self.load_model()
     
     def load_model(self):
@@ -243,11 +257,6 @@ class OptimizedPANNs:
             self.model = None
     
     def predict(self, audio: np.ndarray, timestamp: float = 0.0) -> VADResult:
-        if timestamp > 0 and self.cached_clip_prob is not None:
-            return VADResult(self.cached_clip_prob,
-                             self.cached_clip_prob > 0.5,
-                             self.model_name, 0.0, timestamp)
-
         start_time = time.time()
         
         if self.model is None or len(audio) == 0:
@@ -265,19 +274,45 @@ class OptimizedPANNs:
             if len(audio.shape) > 1:
                 audio = audio.mean(axis=1)
 
-            clip_probs, _ = self.model.inference(audio[np.newaxis, :],
+            # Convert audio to PANNs sample rate
+            if LIBROSA_AVAILABLE:
+                audio_resampled = librosa.resample(audio.astype(float), 
+                                                 orig_sr=16000, 
+                                                 target_sr=self.sample_rate)
+            else:
+                # Simple resampling fallback
+                resample_factor = self.sample_rate / 16000
+                audio_resampled = np.interp(
+                    np.linspace(0, len(audio) - 1, int(len(audio) * resample_factor)),
+                    np.arange(len(audio)),
+                    audio
+                )
+
+            # Ensure minimum length for PANNs (need at least 1 second)
+            min_samples = self.sample_rate  # 1 second
+            if len(audio_resampled) < min_samples:
+                audio_resampled = np.pad(audio_resampled, (0, min_samples - len(audio_resampled)), 'constant')
+            
+            clip_probs, _ = self.model.inference(audio_resampled[np.newaxis, :],
                                                  input_sr=self.sample_rate)
 
-            speech_idx = [i for i, lbl in enumerate(labels)
-                          if 'speech' in lbl.lower() or 'voice' in lbl.lower()]
-            if not speech_idx:
-                speech_idx = [labels.index('Speech')]
+            # Find speech-related indices
+            speech_indices = []
+            for i, lbl in enumerate(labels):
+                if any(word in lbl.lower() for word in ['speech', 'voice', 'talk', 'conversation', 'speaking']):
+                    speech_indices.append(i)
+            
+            if not speech_indices:
+                # Fallback to a known speech index if available
+                try:
+                    speech_indices = [labels.index('Speech')]
+                except ValueError:
+                    # If 'Speech' label doesn't exist, use first 10 indices as approximation
+                    speech_indices = list(range(min(10, len(labels))))
 
-            speech_prob = clip_probs[0, speech_idx].mean().item()
-            self.cached_clip_prob = float(speech_prob)
-            return VADResult(self.cached_clip_prob,
-                             self.cached_clip_prob > 0.5,
-                             self.model_name, time.time()-start_time, timestamp)
+            speech_prob = clip_probs[0, speech_indices].mean().item()
+            
+            return VADResult(float(speech_prob), speech_prob > 0.5, self.model_name, time.time()-start_time, timestamp)
             
         except Exception as e:
             print(f"Error in {self.model_name}: {e}")
@@ -298,7 +333,6 @@ class OptimizedAST:
         self.model = None
         self.feature_extractor = None
         self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-        self.cached_clip_prob = None
         self.load_model()
     
     def load_model(self):
@@ -318,11 +352,6 @@ class OptimizedAST:
             self.model = None
     
     def predict(self, audio: np.ndarray, timestamp: float = 0.0) -> VADResult:
-        if timestamp > 0 and self.cached_clip_prob is not None:
-            return VADResult(self.cached_clip_prob,
-                             self.cached_clip_prob > 0.5,
-                             self.model_name, 0.0, timestamp)
-
         start_time = time.time()
         
         if self.model is None or len(audio) == 0:
@@ -344,6 +373,11 @@ class OptimizedAST:
             if len(audio.shape) > 1:
                 audio = audio.mean(axis=1)
             
+            # Ensure minimum length for AST (typically needs longer sequences)
+            min_samples = self.sample_rate  # 1 second minimum
+            if len(audio) < min_samples:
+                audio = np.pad(audio, (0, min_samples - len(audio)), 'constant')
+            
             inputs = self.feature_extractor(audio, sampling_rate=self.sample_rate, return_tensors="pt")
             inputs = {k: v.to(self.device) for k, v in inputs.items()}
             
@@ -353,13 +387,18 @@ class OptimizedAST:
                 probs = torch.sigmoid(logits)
             
             label2id = self.model.config.label2id
-            speech_idx = [idx for lbl, idx in label2id.items()
-                          if 'speech' in lbl.lower() or 'voice' in lbl.lower()]
-            speech_prob = probs[0, speech_idx].mean().item()
-            self.cached_clip_prob = float(speech_prob)
-            return VADResult(self.cached_clip_prob,
-                             self.cached_clip_prob > 0.5,
-                             self.model_name, time.time()-start_time, timestamp)
+            speech_indices = []
+            for lbl, idx in label2id.items():
+                if any(word in lbl.lower() for word in ['speech', 'voice', 'talk', 'conversation', 'speaking', 'human']):
+                    speech_indices.append(idx)
+            
+            if speech_indices:
+                speech_prob = probs[0, speech_indices].mean().item()
+            else:
+                # Fallback: use average of first few probabilities
+                speech_prob = probs[0, :10].mean().item()
+            
+            return VADResult(float(speech_prob), speech_prob > 0.5, self.model_name, time.time()-start_time, timestamp)
             
         except Exception as e:
             print(f"Error in {self.model_name}: {e}")
@@ -833,50 +872,21 @@ class VADDemo:
             if len(processed_audio) == 0:
                 return None, "🎵 Processing audio...", "No audio data processed"
 
-            panns_prob = None
-            ast_prob = None
-            selected_models = list(set([model_a, model_b]))
-
-            if 'PANNs' in selected_models:
-                panns_model = self.models['PANNs']
-                # Reset cache for new audio clip
-                panns_model.cached_clip_prob = None
-                if LIBROSA_AVAILABLE:
-                    audio_32k = librosa.resample(processed_audio,
-                                                 orig_sr=self.processor.sample_rate,
-                                                 target_sr=panns_model.sample_rate)
-                    panns_prob = panns_model.predict(audio_32k, 0.0).probability
-                else:
-                    panns_prob = 0.0
-
-            if 'AST' in selected_models:
-                ast_model = self.models['AST']
-                # Reset cache for new audio clip
-                ast_model.cached_clip_prob = None
-                ast_prob = ast_model.predict(processed_audio, 0.0).probability
-
             window_samples = int(self.processor.sample_rate * self.processor.window_size)
             hop_samples = int(self.processor.sample_rate * self.processor.hop_size)
             vad_results = []
 
+            selected_models = list(set([model_a, model_b]))
+
+            # Process each window individually for all models
             for i in range(0, len(processed_audio) - window_samples, hop_samples):
                 timestamp = i / self.processor.sample_rate
+                chunk = processed_audio[i:i + window_samples]
                 
                 for model_name in selected_models:
-                    result = None
-                    if model_name == 'PANNs':
-                        if panns_prob is not None:
-                           result = VADResult(panns_prob, panns_prob > threshold, 'PANNs', 0.0, timestamp)
-                    elif model_name == 'AST':
-                         if ast_prob is not None:
-                           result = VADResult(ast_prob, ast_prob > threshold, 'AST', 0.0, timestamp)
-                    else:
-                        chunk = processed_audio[i:i + window_samples]
-                        if model_name in self.models:
-                            result = self.models[model_name].predict(chunk, timestamp)
-                            result.is_speech = result.probability > threshold
-                    
-                    if result:
+                    if model_name in self.models:
+                        result = self.models[model_name].predict(chunk, timestamp)
+                        result.is_speech = result.probability > threshold
                         vad_results.append(result)
 
             delay_compensation = self.processor.estimate_delay_compensation(processed_audio, vad_results)
@@ -987,7 +997,7 @@ def create_interface():
     with gr.Blocks(title="VAD Demo - Real-time Speech Detection", theme=gr.themes.Soft()) as interface:
         
         gr.Markdown("""
-        # 🎤 VAD Demo: Real-time Speech Detection Framework v2
+        # 🎤 VAD Demo: Real-time Speech Detection Framework v3
         
         **Multi-Model Voice Activity Detection with Advanced Onset/Offset Detection**