app.py

import gradio as gr
import torch
import cv2
import numpy as np
from PIL import Image
import spaces
import base64
import io
from transformers import BlipProcessor, BlipForConditionalGeneration, AutoTokenizer, AutoModelForCausalLM
import warnings
warnings.filterwarnings("ignore")

# Global variables
vision_model = None
vision_processor = None
text_model = None
text_tokenizer = None
device = "cuda" if torch.cuda.is_available() else "cpu"
model_loaded = False

@spaces.GPU
def load_models():
    """Load BLIP for vision and a language model for analysis"""
    global vision_model, vision_processor, text_model, text_tokenizer, model_loaded
    
    try:
        print("🔄 Loading AI models for video analysis...")
        
        # Load BLIP for image understanding
        print("Loading BLIP vision model...")
        vision_processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-large")
        vision_model = BlipForConditionalGeneration.from_pretrained(
            "Salesforce/blip-image-captioning-large",
            torch_dtype=torch.float16,
            device_map="auto"
        )
        
        # Load a conversational model for analysis
        print("Loading language model...")
        text_tokenizer = AutoTokenizer.from_pretrained("microsoft/DialoGPT-medium")
        text_model = AutoModelForCausalLM.from_pretrained(
            "microsoft/DialoGPT-medium",
            torch_dtype=torch.float16,
            device_map="auto"
        )
        
        # Add padding token if needed
        if text_tokenizer.pad_token is None:
            text_tokenizer.pad_token = text_tokenizer.eos_token
        
        model_loaded = True
        success_msg = "✅ AI models loaded successfully! You can now analyze videos."
        print(success_msg)
        return success_msg
        
    except Exception as e:
        model_loaded = False
        error_msg = f"❌ Failed to load models: {str(e)}"
        print(error_msg)
        return error_msg

def extract_key_frames(video_path, max_frames=8):
    """Extract key frames from video"""
    try:
        cap = cv2.VideoCapture(video_path)
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        fps = cap.get(cv2.CAP_PROP_FPS)
        duration = total_frames / fps if fps > 0 else 0
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        
        if total_frames == 0:
            return [], None
        
        # Get evenly spaced frames
        frame_indices = np.linspace(0, total_frames-1, min(max_frames, total_frames), dtype=int)
        frames = []
        timestamps = []
        
        for frame_idx in frame_indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
            ret, frame = cap.read()
            if ret:
                # Convert BGR to RGB
                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                
                # Resize if too large
                if max(width, height) > 512:
                    scale = 512 / max(width, height)
                    new_width = int(width * scale)
                    new_height = int(height * scale)
                    frame_rgb = cv2.resize(frame_rgb, (new_width, new_height))
                
                frames.append(Image.fromarray(frame_rgb))
                timestamp = frame_idx / fps if fps > 0 else frame_idx
                timestamps.append(timestamp)
        
        cap.release()
        
        video_info = {
            "duration": duration,
            "fps": fps,
            "total_frames": total_frames,
            "resolution": f"{width}x{height}",
            "extracted_frames": len(frames)
        }
        
        return frames, video_info, timestamps
        
    except Exception as e:
        print(f"Error extracting frames: {e}")
        return [], None, []

@spaces.GPU
def analyze_frame_with_blip(frame, custom_question=None):
    """Analyze a single frame with BLIP"""
    try:
        if custom_question:
            # Use BLIP for visual question answering
            inputs = vision_processor(frame, custom_question, return_tensors="pt").to(device)
        else:
            # Use BLIP for image captioning
            inputs = vision_processor(frame, return_tensors="pt").to(device)
        
        with torch.no_grad():
            if custom_question:
                output_ids = vision_model.generate(**inputs, max_new_tokens=100)
            else:
                output_ids = vision_model.generate(**inputs, max_new_tokens=50)
        
        caption = vision_processor.decode(output_ids[0], skip_special_tokens=True)
        return caption
        
    except Exception as e:
        return f"Error analyzing frame: {str(e)}"

def synthesize_video_analysis(frame_descriptions, question, video_info):
    """Create comprehensive video analysis from frame descriptions"""
    
    # Combine all frame descriptions
    all_descriptions = " ".join(frame_descriptions)
    
    # Create analysis based on question type
    question_lower = question.lower()
    
    analysis = f"""🎥 **AI Video Analysis**

❓ **Your Question:** {question}

🤖 **Detailed Analysis:**

"""
    
    if any(word in question_lower for word in ['what', 'happening', 'describe', 'see']):
        analysis += f"Based on my analysis of {len(frame_descriptions)} key frames from the video:\n\n"
        
        for i, desc in enumerate(frame_descriptions):
            timestamp = i * (video_info['duration'] / len(frame_descriptions))
            analysis += f"• **At {timestamp:.1f}s:** {desc}\n"
        
        analysis += f"\n**Overall Summary:** This {video_info['duration']:.1f}-second video shows {all_descriptions.lower()}. "
        
        # Add contextual insights
        if len(set(frame_descriptions)) < len(frame_descriptions) * 0.3:
            analysis += "The scene appears relatively static with consistent elements throughout."
        else:
            analysis += "The video shows dynamic content with changing scenes and activities."
    
    elif any(word in question_lower for word in ['people', 'person', 'human', 'who']):
        people_mentions = [desc for desc in frame_descriptions if any(word in desc.lower() for word in ['person', 'people', 'man', 'woman', 'child', 'human'])]
        if people_mentions:
            analysis += f"**People in the video:** {' '.join(people_mentions)}\n\n"
        else:
            analysis += "**People analysis:** No clear human figures were detected in the analyzed frames.\n\n"
    
    elif any(word in question_lower for word in ['object', 'item', 'thing']):
        analysis += "**Objects and items visible:**\n"
        for desc in frame_descriptions:
            analysis += f"• {desc}\n"
    
    elif any(word in question_lower for word in ['setting', 'location', 'place', 'where']):
        analysis += "**Setting and location analysis:**\n"
        analysis += f"Based on the visual elements: {all_descriptions}\n\n"
    
    elif any(word in question_lower for word in ['mood', 'emotion', 'feeling', 'atmosphere']):
        analysis += "**Mood and atmosphere:**\n"
        analysis += f"The visual elements suggest: {all_descriptions}\n\n"
    
    else:
        # General analysis
        analysis += f"**Frame-by-frame analysis:**\n"
        for i, desc in enumerate(frame_descriptions):
            analysis += f"{i+1}. {desc}\n"
    
    return analysis

@spaces.GPU
def analyze_video_with_ai(video_file, question, progress=gr.Progress()):
    """Main video analysis function"""
    
    if video_file is None:
        return "❌ Please upload a video file first."
    
    if not question.strip():
        return "❌ Please enter a question about the video."
    
    if not model_loaded:
        return "❌ AI models are not loaded. Please click 'Load AI Models' first and wait for completion."
    
    try:
        progress(0.1, desc="Extracting video frames...")
        
        # Extract frames
        frames, video_info, timestamps = extract_key_frames(video_file, max_frames=8)
        
        if not frames or video_info is None:
            return "❌ Could not process video. Please check the video format."
        
        progress(0.3, desc="Analyzing frames with AI...")
        
        # Analyze each frame
        frame_descriptions = []
        for i, frame in enumerate(frames):
            progress(0.3 + (i / len(frames)) * 0.5, desc=f"Analyzing frame {i+1}/{len(frames)}...")
            
            # Create frame-specific question if relevant
            if any(word in question.lower() for word in ['what', 'describe', 'see', 'happening']):
                frame_question = f"What do you see in this image? {question}"
                description = analyze_frame_with_blip(frame, frame_question)
            else:
                description = analyze_frame_with_blip(frame)
            
            frame_descriptions.append(description)
        
        progress(0.8, desc="Synthesizing analysis...")
        
        # Create comprehensive analysis
        analysis = synthesize_video_analysis(frame_descriptions, question, video_info)
        
        # Add technical information
        analysis += f"""

📊 **Technical Information:**
• Duration: {video_info['duration']:.1f} seconds
• Frame Rate: {video_info['fps']:.1f} FPS
• Total Frames: {video_info['total_frames']:,}
• Analyzed Frames: {video_info['extracted_frames']}
• Resolution: {video_info['resolution']}

⚡ **Powered by:** BLIP Vision AI + Advanced Analysis
"""
        
        progress(1.0, desc="Analysis complete!")
        
        return analysis
        
    except Exception as e:
        error_msg = f"❌ Error during analysis: {str(e)}"
        print(error_msg)
        return error_msg

def create_interface():
    """Create the Gradio interface"""
    
    with gr.Blocks(title="AI Video Analyzer", theme=gr.themes.Soft()) as demo:
        gr.Markdown("# 🎥 AI Video Analysis Tool")
        gr.Markdown("Upload videos and get detailed AI-powered analysis using advanced computer vision!")
        
        # Model loading section
        with gr.Row():
            with gr.Column(scale=3):
                model_status = gr.Textbox(
                    label="🤖 Model Status", 
                    value="Models not loaded - Click the button to load AI models →",
                    interactive=False,
                    lines=2
                )
            with gr.Column(scale=1):
                load_btn = gr.Button("🚀 Load AI Models", variant="primary", size="lg")
        
        load_btn.click(load_models, outputs=model_status)
        
        gr.Markdown("---")
        
        # Main interface
        with gr.Row():
            with gr.Column(scale=1):
                video_input = gr.Video(
                    label="📹 Upload Video (MP4, AVI, MOV, WebM)",
                    height=350
                )
                question_input = gr.Textbox(
                    label="❓ Ask about the video",
                    placeholder="What is happening in this video? Describe it in detail.",
                    lines=3,
                    max_lines=5
                )
                analyze_btn = gr.Button("🔍 Analyze Video with AI", variant="primary", size="lg")
                
            with gr.Column(scale=1):
                output = gr.Textbox(
                    label="🎯 AI Analysis Results",
                    lines=25,
                    max_lines=30,
                    show_copy_button=True
                )
        
        # Example questions
        gr.Markdown("### 💡 Example Questions (click to use):")
        
        example_questions = [
            "What is happening in this video? Describe the scene in detail.",
            "Who are the people in this video and what are they doing?",
            "Describe the setting, location, and environment shown.",
            "What objects, animals, or items can you see in the video?",
            "What is the mood, atmosphere, or emotion conveyed?",
            "Summarize the key events that occur chronologically."
        ]
        
        with gr.Row():
            for i in range(0, len(example_questions), 2):
                with gr.Column():
                    if i < len(example_questions):
                        btn1 = gr.Button(example_questions[i], size="sm")
                        btn1.click(lambda x=example_questions[i]: x, outputs=question_input)
                    if i+1 < len(example_questions):
                        btn2 = gr.Button(example_questions[i+1], size="sm") 
                        btn2.click(lambda x=example_questions[i+1]: x, outputs=question_input)
        
        # Connect the analyze button
        analyze_btn.click(
            analyze_video_with_ai,
            inputs=[video_input, question_input],
            outputs=output,
            show_progress=True
        )
        
        gr.Markdown("---")
        gr.Markdown("""
        ### 📋 Instructions:
        1. **First:** Click "Load AI Models" and wait for it to complete (~3-5 minutes)
        2. **Then:** Upload your video file (works with most formats)  
        3. **Ask:** Type your question about the video content
        4. **Analyze:** Click "Analyze Video with AI" to get detailed insights
        
        💡 **How it works:** 
        - Extracts key frames from your video
        - Analyzes each frame with BLIP vision AI
        - Synthesizes comprehensive analysis based on your question
        - Works reliably with standard video formats
        """)
    
    return demo

if __name__ == "__main__":
    demo = create_interface()
    demo.launch()