"""MrrrMe Smart Mirror - OPTIMIZED EVENT-DRIVEN ARCHITECTURE (OLLAMA-READY)""" import time import cv2 import numpy as np import torch from .config import * from .audio.voice_assistant import VoiceAssistant from .audio.whisper_transcription import WhisperTranscriptionWorker from .audio.voice_emotion import VoiceEmotionWorker from .nlp.text_sentiment import TextSentimentAnalyzer from .nlp.llm_generator_groq import LLMResponseGenerator from .vision.face_processor import FaceProcessor from .vision.async_face_processor import SmartFaceIntegration # ========== OPTIMIZED FUSION ENGINE ========== class IntelligentFusionEngine: """ ⭐ OPTIMIZED: Event-driven fusion (only recalculates when needed) """ def __init__(self): self.ema_alpha = 0.35 self.last_intensity = 0.5 self.last_masking_state = False self.last_conflicts = [] # ⭐ NEW: Caching for efficiency self.cached_result = ( np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32), # fused_probs "Neutral", # fused_top 0.5, # smooth_intensity False # is_masking ) self.last_update_time = 0 def calculate_reliability_weights(self, face_quality, face_confidence, voice_confidence, text_length): """Dynamic weighting based on signal quality""" face_weight = FUSE_ALPHA_FACE if face_quality < 0.5: face_weight *= 0.5 if face_confidence < 0.5: face_weight *= 0.7 voice_weight = FUSE_ALPHA_VOICE text_weight = FUSE_ALPHA_TEXT if text_length < 10: text_weight *= 0.7 total = face_weight + voice_weight + text_weight return { 'face': face_weight / total, 'voice': voice_weight / total, 'text': text_weight / total } def detect_conflicts(self, face_probs, voice_probs, text_probs): """Detect when modalities strongly disagree""" face_top_idx = np.argmax(face_probs) voice_top_idx = np.argmax(voice_probs) text_top_idx = np.argmax(text_probs) face_top = FUSE4[face_top_idx] voice_top = FUSE4[voice_top_idx] text_top = FUSE4[text_top_idx] positive_emotions = {'Happy'} negative_emotions = {'Sad', 'Angry'} conflicts = [] if face_top in positive_emotions and voice_top in negative_emotions: if voice_probs[voice_top_idx] > 0.3: conflicts.append(('face_voice', face_top, voice_top)) if face_top in positive_emotions and text_top in negative_emotions: if text_probs[text_top_idx] > 0.3: conflicts.append(('face_text', face_top, text_top)) return conflicts def fuse(self, async_face, voice_probs, text_probs, text_length, force=False): """ ⭐ OPTIMIZED: Only recalculate when forced (on user speech) During main loop, returns cached result for efficiency """ # ⭐ If not forced, return cached result (saves 600x calculations!) if not force: return self.cached_result # ⭐ Only recalculate when forced (user finished speaking) face_probs = async_face.get_emotion_probs() try: face_quality = async_face.face_processor.get_last_quality() except (AttributeError, Exception): face_quality = 0.5 try: face_confidence = async_face.face_processor.get_last_confidence() except (AttributeError, Exception): face_confidence = 0.5 try: is_masking = async_face.face_processor.is_masking_emotion() except (AttributeError, Exception): is_masking = False weights = self.calculate_reliability_weights( face_quality, face_confidence, 1.0, text_length ) conflicts = self.detect_conflicts(face_probs, voice_probs, text_probs) # Only print on changes if conflicts != self.last_conflicts: if conflicts: print(f"[Fusion] ⚠️ Conflicts: {conflicts}") elif self.last_conflicts: print(f"[Fusion] ✅ Conflicts resolved") self.last_conflicts = conflicts if is_masking != self.last_masking_state: if is_masking: print(f"[Fusion] 🎭 MASKING DETECTED") else: print(f"[Fusion] ✅ Genuine emotion") self.last_masking_state = is_masking # Weighted fusion fused = ( weights['face'] * face_probs + weights['voice'] * voice_probs + weights['text'] * text_probs ) fused = fused / (np.sum(fused) + 1e-8) fused_idx = int(np.argmax(fused)) fused_top = FUSE4[fused_idx] raw_intensity = float(np.max(fused)) if is_masking: raw_intensity *= 0.7 smooth_intensity = self.ema_alpha * raw_intensity + (1 - self.ema_alpha) * self.last_intensity self.last_intensity = smooth_intensity # ⭐ Cache the result self.cached_result = (fused, fused_top, smooth_intensity, is_masking) self.last_update_time = time.time() print(f"[Fusion] ✅ Calculated: {fused_top} (intensity={smooth_intensity:.2f})") return self.cached_result def main(): print("\n" + "="*70) print("🌟 MrrrMe Smart Mirror - OPTIMIZED MODE (LLAMA 3.1 8B) 🌟") print("="*70) print("[MrrrMe] 🚀 Initializing optimized emotion AI...") # ==================== PHASE 1: Initialize ==================== print("\n[Phase 1/4] 🔧 Loading AI models...") # ⭐ AVATAR MODE CONFIGURATION USE_AVATAR = True # Set to False to use voice assistant face_processor = FaceProcessor() text_analyzer = TextSentimentAnalyzer() whisper_worker = WhisperTranscriptionWorker(text_analyzer) voice_worker = VoiceEmotionWorker(whisper_worker=whisper_worker) # ⭐ CHANGED: Ollama-based LLM (no use_local param) llm_generator = LLMResponseGenerator(api_key="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") # ⭐ AVATAR OR VOICE MODE if USE_AVATAR: print("\n[MrrrMe] 🎭 AVATAR MODE ENABLED") from .avatar.avatar_controller import AvatarController voice_assistant = AvatarController() else: print("\n[MrrrMe] 🎤 VOICE MODE ENABLED") from .audio.voice_assistant import VoiceAssistant voice_assistant = VoiceAssistant() fusion_engine = IntelligentFusionEngine() # ==================== PHASE 2: Integration ==================== print("\n[Phase 2/4] 🔗 Setting up coordination...") smart_face = SmartFaceIntegration( face_processor=face_processor, whisper_worker=whisper_worker, voice_assistant=voice_assistant, sample_rate=1.0 ) # Register workers for BOTH modes (so they pause during speech) voice_assistant.register_audio_worker(voice_worker) voice_assistant.register_audio_worker(whisper_worker) print(f"[MrrrMe] ✅ Registered {len(voice_assistant.audio_workers)} workers with TTS") voice_worker.paused = False whisper_worker.paused = False print("[MrrrMe] ✅ Reset pause states") if hasattr(voice_worker, "set_barge_in_callback"): voice_worker.set_barge_in_callback( lambda: voice_assistant.stop() if voice_assistant.get_is_speaking() else None ) last_auto_response_time = [0] # ==================== PHASE 3: Response Handler ==================== def on_user_finished_speaking(transcribed_text): """Callback when user finishes speaking (WITH DETAILED TIMING)""" t_start = time.time() print(f"\n{'='*70}") print(f"[{time.strftime('%H:%M:%S')}] 🎤 USER FINISHED SPEAKING") print(f"{'='*70}") print(f"[00.000s] Transcribed: '{transcribed_text}'") if time.time() - last_auto_response_time[0] < AUTO_RESPONSE_COOLDOWN: print(f"[{time.time()-t_start:.3f}s] ❌ Cooldown active, skipping") return # Get emotions t1 = time.time() voice_probs, voice_top = voice_worker.get_probs() print(f"[{t1-t_start:.3f}s] ✅ Got voice emotion: {voice_top}") t2 = time.time() text_probs, text_content = text_analyzer.get_probs() print(f"[{t2-t_start:.3f}s] ✅ Got text sentiment") # Force fusion t3 = time.time() fused_probs, fused_top, smooth_intensity, is_masking = fusion_engine.fuse( smart_face.async_face, voice_probs, text_probs, len(transcribed_text), force=True ) print(f"[{t3-t_start:.3f}s] ✅ Emotion fusion complete: {fused_top} ({smooth_intensity:.2f})") t3b = time.time() face_top = smart_face.async_face.face_processor.get_last_emotion() text_top = FUSE4[int(text_probs.argmax())] print(f"[{t3b-t_start:.3f}s] Face: {face_top}, Voice: {voice_top}, Text: {text_top} → Fused: {fused_top}") # Filtering (use values directly, no import) min_length = 2 # Or MIN_CHARS if you imported it at the top if len(transcribed_text) < min_length: print(f"[{time.time()-t_start:.3f}s] ❌ Too short: {len(transcribed_text)} < {min_length}") return hallucinations = ["thank you", "thanks", "okay", "ok", "you", "thank you."] confidence_threshold = 0.35 if smooth_intensity < confidence_threshold: text_lower = transcribed_text.lower().strip() if text_lower in hallucinations or len(text_lower.split()) <= 2: print(f"[{time.time()-t_start:.3f}s] 🔇 Low confidence → ignoring") return t4 = time.time() print(f"[{t4-t_start:.3f}s] 🧠 Starting LLM generation...") response = llm_generator.generate_response( fused_top, face_top, voice_top, transcribed_text, force=True, intensity=smooth_intensity, is_masking=is_masking ) t5 = time.time() print(f"[{t5-t_start:.3f}s] ✅ LLM response generated ({t5-t4:.3f}s) ⭐") print(f"[{t5-t_start:.3f}s] Response: '{response}'") t6 = time.time() print(f"[{t6-t_start:.3f}s] 🎭 Sending to avatar backend...") voice_assistant.apply_emotion_voice(fused_top, smooth_intensity) voice_assistant.speak_async(response) t7 = time.time() print(f"[{t7-t_start:.3f}s] ✅ Avatar request sent ({t7-t6:.3f}s)") last_auto_response_time[0] = time.time() # Summary print(f"\n{'='*70}") print(f"⏱️ TIMING BREAKDOWN:") print(f"{'='*70}") print(f" Get emotions: {t2-t_start:.3f}s") print(f" Fusion: {t3-t2:.3f}s") print(f" LLM generation: {t5-t4:.3f}s ⭐ BOTTLENECK?") print(f" Avatar initiate: {t7-t6:.3f}s") print(f" TOTAL (no wait): {t7-t_start:.3f}s") print(f"{'='*70}") print(f"Note: Avatar TTS+Rhubarb runs async in background") print(f"{'='*70}\n") # ==================== PHASE 4: Start Systems ==================== print("\n[Phase 3/4] ▶️ Starting subsystems...") whisper_worker.set_response_callback(on_user_finished_speaking) whisper_worker.start() voice_worker.start() smart_face.start() print("\n[Phase 4/4] 📹 Initializing webcam...") cap = cv2.VideoCapture(0, cv2.CAP_DSHOW) if not cap.isOpened(): cap = cv2.VideoCapture(1, cv2.CAP_DSHOW) if not cap.isOpened(): raise RuntimeError("Webcam not found") time.sleep(2) test_ok, test_frame = cap.read() if not test_ok: cap.release() raise RuntimeError("Cannot capture frames") print("[Webcam] ✅ Ready!") print("\n" + "="*70) print("🎉 MrrrMe OPTIMIZED MODE READY!") print("="*70) print("✅ Event-Driven Fusion (600x more efficient)") print("✅ AU-Based Emotion Detection") print("✅ Intelligent Conflict Resolution") print("✅ Masking Detection") print("✅ Natural Conversation with Llama 3.1 8B") # ⭐ UPDATED print("✅ FIXED: Less aggressive response filters") print("="*70) print("\n💡 Controls: ESC=Quit | SPACE=Test | S=Stats | C=GPU Clear") print("🎤 Speak naturally!\n") # ==================== MAIN LOOP ==================== fps_counter = 0 fps_start = time.time() fps = 0.0 last_gpu_cleanup = time.time() try: print("[Main Loop] 🎬 Started!\n") while True: ok, frame = cap.read() if not ok: break # Process frame frame, face_emotion = smart_face.process_frame(frame) # ⭐ Get current emotions (for UI display only) voice_probs, voice_top = voice_worker.get_probs() text_probs, text_content = text_analyzer.get_probs() text_top = FUSE4[int(text_probs.argmax())] # ⭐ Use CACHED fusion result (no recalculation!) fused_probs, fused_top, smooth_intensity, is_masking = fusion_engine.fuse( smart_face.async_face, voice_probs, text_probs, len(text_content or ""), force=False # ← Use cache! ) # GPU cleanup if time.time() - last_gpu_cleanup > 30: if torch.cuda.is_available(): torch.cuda.empty_cache() last_gpu_cleanup = time.time() # Display UI H, W = frame.shape[:2] if voice_worker.paused: cv2.putText(frame, "AI SPEAKING", (10, H-120), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 165, 255), 2) if smart_face.gpu_coord.has_critical_tasks(): cv2.putText(frame, "GPU: BUSY", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) else: cv2.putText(frame, "GPU: IDLE", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) cv2.putText(frame, f"Voice: {voice_top}", (10, H-94), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 0), 2) cv2.putText(frame, f"Text: {text_top}", (10, H-64), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 165, 0), 2) masking_marker = " 🎭" if is_masking else "" cv2.putText(frame, f"Fused: {fused_top}{masking_marker}", (10, H-36), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2) cv2.putText(frame, f"Int: {smooth_intensity:.2f}", (W - 150, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (180, 255, 180), 2) if text_content: text_display = text_content[:50] + "..." if len(text_content) > 50 else text_content cv2.putText(frame, f"Said: {text_display}", (10, 120), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (200, 200, 200), 1) llm_response = llm_generator.get_last_response() if llm_response: words = llm_response.split() lines, current_line = [], "" for word in words: if len(current_line + word) < 45: current_line += word + " " else: lines.append(current_line) current_line = word + " " if current_line: lines.append(current_line) for i, line in enumerate(lines[:2]): cv2.putText(frame, line, (W - 450, H - 80 + i*25), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (100, 255, 100), 2) # FPS fps_counter += 1 if time.time() - fps_start >= 1.0: fps = fps_counter / (time.time() - fps_start) fps_start = time.time() fps_counter = 0 cv2.putText(frame, f"FPS: {fps:.1f}", (10, H-10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2) cv2.imshow("MrrrMe", frame) key = cv2.waitKey(1) & 0xFF if key == 27: # ESC break elif key == 32: # SPACE print("\n[MANUAL TRIGGER]") text_probs, text_content = text_analyzer.get_probs() # Force fusion _, fused_top, smooth_intensity, is_masking = fusion_engine.fuse( smart_face.async_face, voice_probs, text_probs, len(text_content or ""), force=True ) response = llm_generator.generate_response( fused_top, face_emotion, voice_top, text_content or "Hi", force=True, intensity=smooth_intensity, is_masking=is_masking ) voice_assistant.apply_emotion_voice(fused_top, smooth_intensity) voice_assistant.speak_async(response) elif key == ord('s') or key == ord('S'): print("\n" + "="*60) print("📊 SYSTEM STATISTICS") print("="*60) face_stats = smart_face.get_stats() print(f"Face: {face_stats['frames_processed']} processed, " f"{face_stats['frames_dropped']} dropped") if torch.cuda.is_available(): gpu_allocated = torch.cuda.memory_allocated(0) / 1024**3 print(f"GPU: {gpu_allocated:.2f} GB allocated") print("="*60 + "\n") elif key == ord('c') or key == ord('C'): if torch.cuda.is_available(): torch.cuda.empty_cache() print("[GPU] 🧹 Cleared!") last_gpu_cleanup = time.time() except Exception as e: print(f"\n[Error] {e}") import traceback traceback.print_exc() finally: print(f"\n[Shutdown] Stopping...") voice_worker.stop() whisper_worker.stop() smart_face.stop() cap.release() cv2.destroyAllWindows() if torch.cuda.is_available(): torch.cuda.empty_cache() print("[Shutdown] Complete ✅") if __name__ == "__main__": main()