Update app.py

app.py

@@ -1,4 +1,6 @@
-import gradio as gr
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 import torch
 import torch.nn as nn
 import torch.nn.functional as F
@@ -14,14 +16,45 @@ import librosa
 from PIL import Image
 import tempfile
 import os
-from typing import Tuple, Dict, Any
+import shutil
+from typing import Dict, Any, Optional
 import json
 import warnings
+import logging
+import asyncio
+from contextlib import asynccontextmanager
 
-print(f"PyTorch version: {torch.__version__}")
-print(f"CUDA available: {torch.cuda.is_available()}")
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
-# Actual MirrorMindModel architecture (copied from mirrormind.py)
+# Global model instance
+model_instance = None
+
+# Response models
+class EmotionScores(BaseModel):
+    Anger: float
+    Disgust: float
+    Fear: float
+    Happy: float
+    Neutral: float
+    Sad: float
+
+class AnalysisResult(BaseModel):
+    neuroticism: float
+    neuroticism_level: str
+    emotions: EmotionScores
+    dominant_emotion: str
+    frames_processed: int
+    audio_features_extracted: bool
+    model_used: str
+    confidence: str
+
+class ErrorResponse(BaseModel):
+    error: str
+    message: str
+
+# MirrorMind Model Architecture (same as your original)
 class GradientReverseFn(torch.autograd.Function):
     """Gradient reversal function for adversarial training"""
     @staticmethod
@@ -93,13 +126,13 @@ class MirrorMindModel(nn.Module):
                         if any(x in name for x in ['encoder.layers.10', 'encoder.layers.11']):
                             param.requires_grad = True
                 self.audio_pool = nn.AdaptiveAvgPool1d(1)
-                print("Using Wav2Vec2 audio encoder")
+                logger.info("Using Wav2Vec2 audio encoder")
             except Exception as e:
-                print(f"Warning: Could not load Wav2Vec2, using CNN: {e}")
+                logger.warning(f"Could not load Wav2Vec2, using CNN: {e}")
                 self._create_improved_audio_encoder()
         else:
             self._create_improved_audio_encoder()
-            print("Using CNN audio encoder")
+            logger.info("Using CNN audio encoder")
 
         self.audio_proj = nn.Sequential(
             nn.Linear(self.audio_feat_dim, hidden_dim),
@@ -231,7 +264,7 @@ class MirrorMindModel(nn.Module):
             vid_feat = self._process_video_temporal_attention(vid_feat_bt, B, T)
             vid_feat = self.video_proj(vid_feat)
         except Exception as e:
-            print(f"Video processing error: {e}")
+            logger.error(f"Video processing error: {e}")
             vid_feat = torch.zeros((B, self.hidden_dim), device=device)
         try:
             if audio is None or torch.all(audio == 0):
@@ -252,7 +285,7 @@ class MirrorMindModel(nn.Module):
                     aud_feat = x
                 aud_feat = self.audio_proj(aud_feat)
         except Exception as e:
-            print(f"Audio processing error: {e}")
+            logger.error(f"Audio processing error: {e}")
             aud_feat = torch.zeros((B, self.hidden_dim), device=device)
         fused = torch.cat([vid_feat, aud_feat], dim=1)
         fused_final = self.fusion_proj(fused)
@@ -265,17 +298,16 @@ class MirrorMindModel(nn.Module):
                 domain_logits = self.domain_head(rev)
         return neuroticism_pred, emotion_logits, domain_logits
 
-# Inference wrapper (renamed to avoid conflict)
 class MirrorMindInference:
     def __init__(self):
         self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        print(f"Using device: {self.device}")
+        logger.info(f"Using device: {self.device}")
         
         model_path = "mirror_model.pth"
-        print(f"Loading model from {model_path}...")
+        logger.info(f"Loading model from {model_path}...")
         
         if not os.path.exists(model_path):
-            print(f"Model file {model_path} not found. Using fallback mode.")
+            logger.warning(f"Model file {model_path} not found. Using fallback mode.")
             self.model = None
             return
         
@@ -283,132 +315,97 @@ class MirrorMindInference:
         pytorch_version = torch.__version__
         
         if pytorch_version.startswith(("2.8", "2.9")):
-            print(f"Detected PyTorch {pytorch_version} - using version-specific loading...")
+            logger.info(f"Detected PyTorch {pytorch_version} - using version-specific loading...")
             try:
-                print("Loading with weights_only=False...")
+                logger.info("Loading with weights_only=False...")
                 with warnings.catch_warnings():
                     warnings.simplefilter("ignore")
                     checkpoint = torch.load(model_path, map_location=self.device, weights_only=False)
-                print("✓ Successfully loaded complete model")
+                logger.info("✓ Successfully loaded complete model")
             except Exception as e1:
-                print(f"✗ Failed: {e1}")
+                logger.error(f"✗ Failed: {e1}")
                 try:
-                    print("Attempting state_dict loading with weights_only=True...")
+                    logger.info("Attempting state_dict loading with weights_only=True...")
                     checkpoint = torch.load(model_path, map_location=self.device, weights_only=True)
-                    print("✓ Loaded as state_dict")
+                    logger.info("✓ Loaded as state_dict")
                 except Exception as e2:
-                    print(f"✗ Failed: {e2}")
+                    logger.error(f"✗ Failed: {e2}")
                     checkpoint = None
         else:
             try:
-                print(f"Using standard loading for PyTorch {pytorch_version}...")
+                logger.info(f"Using standard loading for PyTorch {pytorch_version}...")
                 checkpoint = torch.load(model_path, map_location=self.device)
-                print("✓ Loaded with standard method")
+                logger.info("✓ Loaded with standard method")
             except Exception as e:
-                print(f"✗ Failed: {e}")
+                logger.error(f"✗ Failed: {e}")
                 checkpoint = None
         
         if checkpoint is None:
-            print("All loading methods failed. Using fallback mode.")
+            logger.warning("All loading methods failed. Using fallback mode.")
             self.model = None
             return
         
         if isinstance(checkpoint, dict):
-            print(f"Checkpoint keys: {list(checkpoint.keys())}")
+            logger.info(f"Checkpoint keys: {list(checkpoint.keys())}")
             
             if 'model' in checkpoint and 'state_dict' in checkpoint:
                 self.model = checkpoint['model']
                 self.model.load_state_dict(checkpoint['state_dict'])
-                print("✓ Loaded model architecture + state dict")
+                logger.info("✓ Loaded model architecture + state dict")
                 
             elif 'state_dict' in checkpoint:
-                print("Found 'state_dict' - attempting to reconstruct model...")
+                logger.info("Found 'state_dict' - attempting to reconstruct model...")
                 if 'model_config' in checkpoint:
                     self.model = MirrorMindModel(**checkpoint['model_config'])
                     self.model.load_state_dict(checkpoint['state_dict'])
-                    print("✓ Loaded using model_config + state_dict")
+                    logger.info("✓ Loaded using model_config + state_dict")
                 else:
-                    print("⚠️ No model_config. Using fallback.")
+                    logger.warning("⚠️ No model_config. Using fallback.")
                     self.model = None
                     return
                     
             elif 'model_state_dict' in checkpoint:
-                print("Found 'model_state_dict' - checking for model class info...")
+                logger.info("Found 'model_state_dict' - checking for model class info...")
                 state_dict = checkpoint['model_state_dict']
                 
                 if 'model_config' in checkpoint:
                     self.model = MirrorMindModel(**checkpoint['model_config'])
                     self.model.load_state_dict(state_dict)
-                    print("✓ Loaded using model_config + model_state_dict")
+                    logger.info("✓ Loaded using model_config + model_state_dict")
                 else:
-                    model_info = self.analyze_state_dict(state_dict)
-                    print(f"State dict analysis: {model_info}")
-                    print("⚠️ No model_config. Using fallback.")
+                    logger.warning("⚠️ No model_config. Using fallback.")
                     self.model = None
                     return
                     
             elif len(checkpoint.keys()) > 0 and all(isinstance(v, torch.Tensor) for v in checkpoint.values()):
-                print("Checkpoint appears to be a direct state dict")
-                model_info = self.analyze_state_dict(checkpoint)
-                print(f"Direct state dict analysis: {model_info}")
-                print("⚠️ Cannot reconstruct without model_config. Using fallback.")
+                logger.info("Checkpoint appears to be a direct state dict")
+                logger.warning("⚠️ Cannot reconstruct without model_config. Using fallback.")
                 self.model = None
                 return
                     
             else:
                 if hasattr(checkpoint, 'eval') and callable(checkpoint.eval):
                     self.model = checkpoint
-                    print("✓ Using checkpoint as complete model")
+                    logger.info("✓ Using checkpoint as complete model")
                 else:
-                    print("⚠️ Unrecognized format. Using fallback.")
+                    logger.warning("⚠️ Unrecognized format. Using fallback.")
                     self.model = None
                     return
         else:
             if hasattr(checkpoint, 'eval') and callable(checkpoint.eval):
                 self.model = checkpoint
-                print("✓ Loaded complete model object")
+                logger.info("✓ Loaded complete model object")
             else:
-                print("⚠️ Not a model object. Using fallback.")
+                logger.warning("⚠️ Not a model object. Using fallback.")
                 self.model = None
                 return
             
         if self.model is not None:
             self.model.to(self.device)
             self.model.eval()
-            print("Model loaded and ready for inference!")
+            logger.info("Model loaded and ready for inference!")
         else:
-            print("Model is None after loading. Using fallback.")
-    
-    def analyze_state_dict(self, state_dict):
-        info = {
-            'total_params': len(state_dict),
-            'layer_types': set(),
-            'input_features': None,
-            'output_features': None,
-            'has_conv': False,
-            'has_lstm': False,
-            'has_attention': False
-        }
-        
-        for key, tensor in state_dict.items():
-            if 'conv' in key.lower():
-                info['has_conv'] = True
-                info['layer_types'].add('conv')
-            elif 'lstm' in key.lower() or 'rnn' in key.lower():
-                info['has_lstm'] = True
-                info['layer_types'].add('lstm')
-            elif 'attention' in key.lower() or 'attn' in key.lower():
-                info['has_attention'] = True
-                info['layer_types'].add('attention')
-            elif 'linear' in key.lower() or 'fc' in key.lower():
-                info['layer_types'].add('linear')
-                if key.endswith('.weight'):
-                    if info['input_features'] is None:
-                        info['input_features'] = tensor.shape[-1]
-                    info['output_features'] = tensor.shape[0]
-        
-        info['layer_types'] = list(info['layer_types'])
-        return info
+            logger.warning("Model is None after loading. Using fallback.")
 
     def extract_video_frames(self, video_path: str, num_frames: int = 8) -> torch.Tensor:
         try:
@@ -442,7 +439,7 @@ class MirrorMindInference:
             return video_tensor
             
         except Exception as e:
-            print(f"Video extraction failed: {e}")
+            logger.error(f"Video extraction failed: {e}")
             dummy_frames = np.random.rand(num_frames, 3, 224, 224).astype(np.float32)
             return torch.from_numpy(dummy_frames).to(self.device)
 
@@ -465,7 +462,7 @@ class MirrorMindInference:
             
             return audio_tensor
         except Exception as e:
-            print(f"Audio extraction failed: {e}")
+            logger.error(f"Audio extraction failed: {e}")
             return torch.zeros(14).to(self.device)
     
     def predict(self, video_path: str) -> Dict[str, Any]:
@@ -485,7 +482,7 @@ class MirrorMindInference:
                     emotion_labels = ['Anger', 'Disgust', 'Fear', 'Happy', 'Neutral', 'Sad']
                     emotion_scores = dict(zip(emotion_labels, emotion_probs))
             else:
-                print("Using fallback predictions")
+                logger.info("Using fallback predictions")
                 neuroticism_score = np.random.uniform(0.2, 0.8)
                 emotion_scores = {
                     'Happy': np.random.uniform(0.1, 0.4),
@@ -507,7 +504,7 @@ class MirrorMindInference:
             }
             
         except Exception as e:
-            print(f"Prediction error: {e}")
+            logger.error(f"Prediction error: {e}")
             return {
                 'error': str(e),
                 'neuroticism': 0.0,
@@ -517,17 +514,90 @@ class MirrorMindInference:
                 'model_used': 'error'
             }
 
-def analyze_video(video_file) -> Tuple[float, str, str]:
-    if video_file is None:
-        return 0.0, "No video uploaded", "Please upload a video file"
+# Initialize model on startup
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global model_instance
+    logger.info("Starting MirrorMind API service...")
+    model_instance = MirrorMindInference()
+    logger.info(f"PyTorch version: {torch.__version__}")
+    logger.info(f"CUDA available: {torch.cuda.is_available()}")
+    yield
+    logger.info("Shutting down MirrorMind API service...")
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="MirrorMind API",
+    description="AI Personality & Emotion Analysis API",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Configure this for production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    return {
+        "message": "MirrorMind API is running",
+        "version": "1.0.0",
+        "pytorch_version": torch.__version__,
+        "cuda_available": torch.cuda.is_available(),
+        "model_loaded": model_instance.model is not None if model_instance else False
+    }
+
+@app.get("/health")
+async def health_check():
+    return {
+        "status": "healthy",
+        "model_status": "loaded" if model_instance and model_instance.model is not None else "fallback",
+        "device": str(model_instance.device) if model_instance else "unknown"
+    }
+
+@app.post("/analyze", response_model=AnalysisResult)
+async def analyze_video(file: UploadFile = File(...)):
+    """
+    Analyze a video file for personality traits and emotions.
+    
+    - **file**: Video file (MP4, AVI, MOV, WebM)
+    - Returns neuroticism score and emotion analysis
+    """
+    
+    if not model_instance:
+        raise HTTPException(status_code=503, detail="Model not initialized")
+    
+    # Validate file type
+    allowed_extensions = {'.mp4', '.avi', '.mov', '.webm', '.mkv'}
+    file_extension = os.path.splitext(file.filename.lower())[1]
+    
+    if file_extension not in allowed_extensions:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Unsupported file format. Allowed formats: {', '.join(allowed_extensions)}"
+        )
+    
+    # Create temporary file
+    temp_dir = tempfile.mkdtemp()
+    temp_file_path = os.path.join(temp_dir, f"uploaded_video{file_extension}")
     
     try:
-        video_path = video_file.name if hasattr(video_file, 'name') else str(video_file)
-        results = model.predict(video_path)
+        # Save uploaded file
+        with open(temp_file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+        
+        # Analyze video
+        results = model_instance.predict(temp_file_path)
         
         if 'error' in results:
-            return 0.0, f"Analysis Error: {results['error']}", str(results)
+            raise HTTPException(status_code=500, detail=f"Analysis failed: {results['error']}")
         
+        # Process results
         neuroticism_score = results['neuroticism']
         
         if neuroticism_score <= 0.3:
@@ -540,130 +610,106 @@ def analyze_video(video_file) -> Tuple[float, str, str]:
         emotions = results['emotions']
         dominant_emotion = max(emotions.keys(), key=lambda k: emotions[k])
         
-        emotion_text = f"**Dominant Emotion:** {dominant_emotion} ({emotions[dominant_emotion]:.1%})\n\n"
-        emotion_text += "**All Emotions:**\n"
-        for emotion, score in sorted(emotions.items(), key=lambda x: x[1], reverse=True):
-            emotion_text += f"- {emotion}: {score:.1%}\n"
+        confidence = "High" if results['model_used'] == 'real' else "Demo Mode"
         
-        model_status = " Real  Model" if results['model_used'] == 'real' else "⚠️ Fallback Mode"
-        detailed_results = f"""
-**Analysis Summary:**
-- Neuroticism Score: {neuroticism_score:.3f}
-- Neuroticism Level: {neuroticism_level}
-- Frames Processed: {results['frames_processed']}
-- Audio Features: {'✓' if results['audio_features_extracted'] else '✗'}
-
-**Technical Details:**
-- Model: {model_status}
-- Processing: Multimodal (Video + Audio)
-- Device: {'GPU' if torch.cuda.is_available() else 'CPU'}
-- Confidence: {'High' if results['model_used'] == 'real' else 'Demo Mode'}
-        """.strip()
-        
-        return neuroticism_score, emotion_text, detailed_results
+        return AnalysisResult(
+            neuroticism=neuroticism_score,
+            neuroticism_level=neuroticism_level,
+            emotions=EmotionScores(**emotions),
+            dominant_emotion=dominant_emotion,
+            frames_processed=results['frames_processed'],
+            audio_features_extracted=results['audio_features_extracted'],
+            model_used=results['model_used'],
+            confidence=confidence
+        )
         
+    except HTTPException:
+        raise
     except Exception as e:
-        error_msg = f"Processing error: {str(e)}"
-        return 0.0, error_msg, error_msg
+        logger.error(f"Analysis error: {e}")
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+    
+    finally:
+        # Clean up temporary files
+        try:
+            shutil.rmtree(temp_dir)
+        except Exception as e:
+            logger.warning(f"Failed to clean up temp directory: {e}")
 
-def create_interface():
-    css = """
-    .gradio-container {
-        font-family: 'Helvetica Neue', Arial, sans-serif;
-    }
-    .output-class {
-        font-size: 16px;
-    }
+@app.post("/analyze-from-url")
+async def analyze_video_from_url(video_url: str):
     """
+    Analyze a video from a URL (Firebase/Supabase storage).
     
-    with gr.Blocks(css=css, title="🧠 MirrorMind Analysis") as demo:
-        
-        model_status_text = " Real  Model Loaded" if model.model is not None else "⚠️ Demo Mode - Using fallback predictions"
-        
-        gr.Markdown(f"""
-        # 🧠 MirrorMind: AI Personality & Emotion Analysis
+    - **video_url**: Direct URL to video file
+    - Returns neuroticism score and emotion analysis
+    """
+    
+    if not model_instance:
+        raise HTTPException(status_code=503, detail="Model not initialized")
+    
+    import requests
+    
+    # Create temporary file
+    temp_dir = tempfile.mkdtemp()
+    temp_file_path = os.path.join(temp_dir, "downloaded_video.mp4")
+    
+    try:
+        # Download video from URL
+        response = requests.get(video_url, stream=True, timeout=30)
+        response.raise_for_status()
         
-        Upload a video to analyze personality traits and emotions using your trained MirrorMind model.
+        with open(temp_file_path, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
         
-        **Model Status:** {model_status_text}  
-        **PyTorch Version:** {torch.__version__}  
-        **CUDA Available:** {'Yes' if torch.cuda.is_available() else 'No'}
-        """)
+        # Analyze video
+        results = model_instance.predict(temp_file_path)
         
-        with gr.Row():
-            with gr.Column(scale=1):
-                video_input = gr.Video(
-                    label="Upload Video",
-                    sources=["upload"],
-                )
-                
-                analyze_btn = gr.Button(
-                    "🔍 Analyze Video", 
-                    variant="primary",
-                    scale=1
-                )
-                
-                gr.Markdown("""
-                **Supported formats:** MP4, AVI, MOV, WebM  
-                **Optimal duration:** 4-10 seconds  
-                **Requirements:** Clear face, good lighting, audio included
-                """)
-            
-            with gr.Column(scale=2):
-                neuroticism_output = gr.Number(
-                    label="🎭 Neuroticism Score (0.0 - 1.0)",
-                    precision=3
-                )
-                
-                emotion_output = gr.Markdown(
-                    label="😊 Emotion Analysis"
-                )
-                
-                details_output = gr.Markdown(
-                    label="📊 Detailed Results"
-                )
+        if 'error' in results:
+            raise HTTPException(status_code=500, detail=f"Analysis failed: {results['error']}")
         
-        analyze_btn.click(
-            fn=analyze_video,
-            inputs=[video_input],
-            outputs=[neuroticism_output, emotion_output, details_output]
-        )
+        # Process results (same as above)
+        neuroticism_score = results['neuroticism']
         
-        video_input.change(
-            fn=analyze_video,
-            inputs=[video_input],
-            outputs=[neuroticism_output, emotion_output, details_output]
-        )
+        if neuroticism_score <= 0.3:
+            neuroticism_level = "Low (Emotionally Stable)"
+        elif neuroticism_score <= 0.7:
+            neuroticism_level = "Medium (Moderate Reactivity)"
+        else:
+            neuroticism_level = "High (Emotionally Sensitive)"
         
-        gr.Markdown("""
-        ---
-        ### 📋 Understanding Your Results
+        emotions = results['emotions']
+        dominant_emotion = max(emotions.keys(), key=lambda k: emotions[k])
         
-        **Neuroticism Scale:**
-        - **0.0-0.3:** Low - Emotionally stable, calm under pressure
-        - **0.3-0.7:** Medium - Moderate emotional reactivity  
-        - **0.7-1.0:** High - More emotionally sensitive, reactive
+        confidence = "High" if results['model_used'] == 'real' else "Demo Mode"
         
-        **Emotions Detected:** Anger, Disgust, Fear, Happy, Neutral, Sad
+        return AnalysisResult(
+            neuroticism=neuroticism_score,
+            neuroticism_level=neuroticism_level,
+            emotions=EmotionScores(**emotions),
+            dominant_emotion=dominant_emotion,
+            frames_processed=results['frames_processed'],
+            audio_features_extracted=results['audio_features_extracted'],
+            model_used=results['model_used'],
+            confidence=confidence
+        )
         
-        **Model Information:** 
-        - Uses your trained `mirror_model.pth` for real AI predictions
-        - Processes both video frames and audio features
-        - Automatically falls back to demo mode if model loading fails
-        """)
+    except requests.RequestException as e:
+        raise HTTPException(status_code=400, detail=f"Failed to download video: {str(e)}")
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Analysis error: {e}")
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
     
-    return demo
-
-print("Initializing MirrorMind model...")
-model = MirrorMindInference()
+    finally:
+        # Clean up temporary files
+        try:
+            shutil.rmtree(temp_dir)
+        except Exception as e:
+            logger.warning(f"Failed to clean up temp directory: {e}")
 
 if __name__ == "__main__":
-    demo = create_interface()
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False,
-        debug=False,
-        show_error=True,
-        quiet=False
-    )
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)