"""Voice emotion recognition using HuBERT (OPTIMIZED VERSION)""" import time import threading from collections import deque, defaultdict import numpy as np import sounddevice as sd import webrtcvad from transformers import pipeline from ..config import (AUDIO_SR, AUDIO_BLOCK, CLIP_SECONDS, VAD_AGGRESSIVENESS, VOICE_EMOTION_MODEL, FUSE4) class VoiceEmotionWorker: """Real-time voice emotion detection with pause capability and silence optimization""" def __init__(self, whisper_worker=None, device=None, model_name=VOICE_EMOTION_MODEL): print(f"\n[VoiceEmotion] πŸš€ Initializing...") print(f"[VoiceEmotion] πŸ“¦ Loading model: {model_name}") # Load emotion recognition model try: self.ser = pipeline( "audio-classification", model=model_name, device=0 if device is None else device ) print(f"[VoiceEmotion] βœ… Model loaded successfully") except Exception as e: print(f"[VoiceEmotion] ❌ Failed to load model: {e}") raise # Initialize VAD try: self.vad = webrtcvad.Vad(VAD_AGGRESSIVENESS) print(f"[VoiceEmotion] βœ… VAD initialized (aggressiveness: {VAD_AGGRESSIVENESS})") except Exception as e: print(f"[VoiceEmotion] ❌ Failed to initialize VAD: {e}") raise # Audio buffer self.ring = deque(maxlen=int(CLIP_SECONDS / AUDIO_BLOCK)) self.lock = threading.Lock() # Emotion state self.current_probs = np.full(len(FUSE4), 0.25, dtype=np.float32) self.current_label = "Neutral" # Device and worker references self.device = device self.whisper_worker = whisper_worker # Control flags self.running = False self.paused = False self.pause_lock = threading.Lock() # ⭐ NEW: Speech activity tracking self.speech_chunks_count = 0 # Count of speech chunks in buffer self.last_speech_time = 0 # Timestamp of last speech detected self.silence_threshold = 2.0 # Seconds of silence before going idle # Statistics self.inference_count = 0 self.skipped_inferences = 0 # ⭐ NEW: Track skipped inferences self.total_audio_processed = 0.0 self.audio_chunks_received = 0 # ⭐ NEW: Error tracking (prevent spam) self.last_error_message = "" self.last_error_time = 0 # Check audio device try: devices = sd.query_devices() if self.device is not None: device_info = sd.query_devices(self.device) print(f"[VoiceEmotion] 🎀 Using device: {device_info['name']}") else: default_device = sd.query_devices(kind='input') print(f"[VoiceEmotion] 🎀 Using default input: {default_device['name']}") except Exception as e: print(f"[VoiceEmotion] ⚠️ Could not query audio device: {e}") print(f"[VoiceEmotion] βš™οΈ Config:") print(f"[VoiceEmotion] - Sample rate: {AUDIO_SR} Hz") print(f"[VoiceEmotion] - Audio block: {AUDIO_BLOCK}s") print(f"[VoiceEmotion] - Clip length: {CLIP_SECONDS}s") print(f"[VoiceEmotion] - Ring buffer: {int(CLIP_SECONDS / AUDIO_BLOCK)} chunks") print(f"[VoiceEmotion] - Silence threshold: {self.silence_threshold}s") print(f"[VoiceEmotion] - Emotions tracked: {', '.join(FUSE4)}") if whisper_worker: print(f"[VoiceEmotion] βœ… Linked to Whisper worker (shared audio)") else: print(f"[VoiceEmotion] ⚠️ No Whisper worker linked") print("[VoiceEmotion] βœ… Ready\n") def pause_listening(self): """Pause audio capture INSTANTLY when TTS starts""" with self.pause_lock: was_paused = self.paused self.paused = True if not was_paused: print("[VoiceEmotion] ⏸️ PAUSED (TTS speaking)") def resume_listening(self): """Resume audio capture INSTANTLY when TTS finishes""" # Clear buffers FIRST whisper_cleared = 0 if self.whisper_worker: try: with self.whisper_worker.lock: whisper_cleared = len(self.whisper_worker.audio_buffer) self.whisper_worker.audio_buffer = [] except Exception as e: self._log_error(f"Error clearing Whisper buffer: {e}") # Clear own buffer emotion_cleared = len(self.ring) self.ring.clear() self.speech_chunks_count = 0 # ⭐ Reset speech counter self.last_speech_time = 0 # ⭐ Reset speech timer # Then unpause with self.pause_lock: self.paused = False total_cleared = whisper_cleared + emotion_cleared print(f"[VoiceEmotion] ▢️ RESUMED (cleared {total_cleared} chunks: " f"{whisper_cleared} whisper + {emotion_cleared} emotion)") def _log_error(self, message): """Log errors with rate limiting to prevent spam""" current_time = time.time() # Only log if message changed or 5 seconds passed if message != self.last_error_message or current_time - self.last_error_time > 5.0: print(f"[VoiceEmotion] ⚠️ {message}") self.last_error_message = message self.last_error_time = current_time def _frames(self, indata, frames, time_, status): """Audio callback - called by sounddevice for each audio block""" # Report any audio issues (only once per issue type) if status: self._log_error(f"Audio status: {status}") # Skip if paused with self.pause_lock: if self.paused: return # Track received audio self.audio_chunks_received += 1 # Convert to mono if needed mono = indata[:, 0] if indata.ndim > 1 else indata # Share with Whisper worker if self.whisper_worker is not None: try: self.whisper_worker.add_audio(mono.copy()) except Exception as e: self._log_error(f"Error sending to Whisper: {e}") # Process for emotion detection hop = int(AUDIO_SR * AUDIO_BLOCK) for i in range(0, len(mono) - hop + 1, hop): chunk = mono[i:i+hop] # VAD check is_speech = False try: pcm16 = np.clip(chunk * 32768, -32768, 32767).astype(np.int16).tobytes() is_speech = self.vad.is_speech(pcm16, sample_rate=AUDIO_SR) except Exception as e: self._log_error(f"VAD error: {e}") # ⭐ NEW: Update speech tracking if is_speech: self.speech_chunks_count += 1 self.last_speech_time = time.time() # Add to ring buffer self.ring.append((chunk.copy(), is_speech)) def _is_speech_active(self): """⭐ NEW: Check if there's recent speech activity""" # Check if we have recent speech time_since_speech = time.time() - self.last_speech_time # If no speech for silence_threshold seconds, go idle if self.last_speech_time > 0 and time_since_speech > self.silence_threshold: return False # Check if we have enough speech chunks in buffer return self.speech_chunks_count >= 3 def _get_speech_chunks(self): """⭐ NEW: Get speech chunks from buffer efficiently""" if len(self.ring) == 0: return None # Collect speech chunks chunks = [c for (c, sp) in self.ring if sp] # Need at least 3 speech chunks if len(chunks) < 3: return None return chunks def _infer_loop(self): """Background thread for emotion inference (OPTIMIZED)""" last_t = 0 loop_count = 0 idle_count = 0 # ⭐ Track consecutive idle loops print("[VoiceEmotion] πŸ”„ Inference loop running...") while self.running: loop_count += 1 t = time.time() # ⭐ OPTIMIZED: Variable rate limiting based on activity min_interval = 0.5 if self._is_speech_active() else 1.0 # Slower when no speech if t - last_t < min_interval: time.sleep(0.05) continue last_t = t # Heartbeat every 200 loops (~2 minutes with optimizations) if loop_count % 200 == 0: with self.lock: emotion = self.current_label with self.pause_lock: paused = self.paused efficiency = (self.inference_count / (self.inference_count + self.skipped_inferences) * 100) if (self.inference_count + self.skipped_inferences) > 0 else 0 print(f"[VoiceEmotion] πŸ’“ Heartbeat: paused={paused}, emotion={emotion}, " f"inferences={self.inference_count}, skipped={self.skipped_inferences} ({efficiency:.1f}% efficiency), " f"chunks_received={self.audio_chunks_received}") # Skip if paused with self.pause_lock: if self.paused: time.sleep(0.1) continue # ⭐ NEW: Skip if no recent speech activity if not self._is_speech_active(): self.skipped_inferences += 1 idle_count += 1 # Log when going idle (only once) if idle_count == 1: print(f"[VoiceEmotion] 😴 Idle (no speech detected for {self.silence_threshold}s)") # Sleep longer during silence time.sleep(0.2) continue # Reset idle counter when active if idle_count > 0: print(f"[VoiceEmotion] 🎀 Active (speech detected)") idle_count = 0 # Get speech chunks chunks = self._get_speech_chunks() if chunks is None: self.skipped_inferences += 1 continue # Reset speech chunk counter periodically self.speech_chunks_count = max(0, self.speech_chunks_count - 1) # Prepare audio clip try: clip = np.concatenate(chunks, axis=0) max_len = int(AUDIO_SR * CLIP_SECONDS) if len(clip) > max_len: clip = clip[-max_len:] except Exception as e: self._log_error(f"Concatenation error: {e}") continue # Run emotion inference start_time = time.time() try: out = self.ser(clip, sampling_rate=AUDIO_SR, top_k=None) inference_time = time.time() - start_time # Map model outputs to our emotions probs = defaultdict(float) total = 0.0 for d in out: lab = d["label"].lower() score = float(d["score"]) total += score # Map to FUSE4 emotions if "ang" in lab: probs["Angry"] += score elif "hap" in lab: probs["Happy"] += score elif "sad" in lab: probs["Sad"] += score elif "neu" in lab: probs["Neutral"] += score elif "fear" in lab: probs["Sad"] += score elif "disg" in lab: probs["Angry"] += score elif "surp" in lab: probs["Neutral"] += score # Normalize probabilities if total <= 0: vec = np.ones(len(FUSE4), dtype=np.float32) / len(FUSE4) else: vec = np.array([probs[c]/total for c in FUSE4], dtype=np.float32) # ⭐ OPTIMIZED: Stronger smoothing to reduce jitter with self.lock: old_label = self.current_label # Increased smoothing: 0.7 old + 0.3 new (was 0.5/0.5) self.current_probs = 0.7 * self.current_probs + 0.3 * vec self.current_label = FUSE4[int(self.current_probs.argmax())] new_label = self.current_label self.inference_count += 1 self.total_audio_processed += len(clip) / AUDIO_SR # Log emotion changes if new_label != old_label: print(f"[VoiceEmotion] 😊 Emotion changed: {old_label} β†’ {new_label} " f"(inference #{self.inference_count}, took {inference_time:.3f}s)") # Log if inference is slow (increased threshold) if inference_time > 1.0: # Was 0.5s, now 1.0s print(f"[VoiceEmotion] ⚠️ Slow inference: {inference_time:.3f}s") except Exception as e: self._log_error(f"Inference error: {e}") print("[VoiceEmotion] πŸ”„ Inference loop exited") def start(self): """Start audio capture and inference""" if self.running: print("[VoiceEmotion] ⚠️ Already running!") return print("[VoiceEmotion] ▢️ Starting audio capture...") self.running = True try: # Create audio input stream self.stream = sd.InputStream( samplerate=AUDIO_SR, channels=1, dtype='float32', blocksize=int(AUDIO_SR * AUDIO_BLOCK), callback=self._frames, device=self.device ) self.stream.start() print("[VoiceEmotion] βœ… Audio stream started") except Exception as e: print(f"[VoiceEmotion] ❌ Failed to start audio stream: {e}") self.running = False raise # Start inference thread try: self.th = threading.Thread(target=self._infer_loop, daemon=True) self.th.start() print("[VoiceEmotion] βœ… Inference thread started") except Exception as e: print(f"[VoiceEmotion] ❌ Failed to start inference thread: {e}") self.stream.stop() self.stream.close() self.running = False raise print("[VoiceEmotion] 🎀 Listening for emotions...") def stop(self): """Stop audio capture and inference""" if not self.running: print("[VoiceEmotion] ⚠️ Already stopped!") return print("[VoiceEmotion] ⏹️ Stopping...") self.running = False # Stop audio stream try: self.stream.stop() self.stream.close() print("[VoiceEmotion] βœ… Audio stream stopped") except Exception as e: print(f"[VoiceEmotion] ⚠️ Error stopping stream: {e}") # Print statistics total_loops = self.inference_count + self.skipped_inferences efficiency = (self.inference_count / total_loops * 100) if total_loops > 0 else 0 print(f"[VoiceEmotion] πŸ“Š Statistics:") print(f"[VoiceEmotion] - Inferences: {self.inference_count}") print(f"[VoiceEmotion] - Skipped: {self.skipped_inferences}") print(f"[VoiceEmotion] - Efficiency: {efficiency:.1f}% (only processed during speech)") print(f"[VoiceEmotion] - Audio processed: {self.total_audio_processed:.1f}s") print(f"[VoiceEmotion] - Chunks received: {self.audio_chunks_received}") print(f"[VoiceEmotion] - Final emotion: {self.current_label}") def get_probs(self): """Get current emotion probabilities and label (thread-safe)""" with self.lock: return self.current_probs.copy(), self.current_label def get_state(self): """Debug: get current state""" with self.lock: probs = self.current_probs.copy() label = self.current_label with self.pause_lock: paused = self.paused is_active = self._is_speech_active() return { 'paused': paused, 'running': self.running, 'speech_active': is_active, # ⭐ NEW 'current_emotion': label, 'emotion_probs': {FUSE4[i]: float(probs[i]) for i in range(len(FUSE4))}, 'ring_buffer_len': len(self.ring), 'speech_chunks': self.speech_chunks_count, # ⭐ NEW 'inference_count': self.inference_count, 'skipped_inferences': self.skipped_inferences, # ⭐ NEW 'chunks_received': self.audio_chunks_received }