🎥 Add Video AI: model/video_ai.py

model/video_ai.py

@@ -0,0 +1,575 @@
+"""
+🎥 Video AI System - Real-time Video Analysis & Processing
+🎬 Understand • 🎨 Edit • 📊 Extract • 🎯 Recognize • 📺 Real-time
+"""
+
+import cv2
+import numpy as np
+from typing import Dict, List, Tuple, Optional
+from datetime import datetime
+import json
+
+class VideoAnalyzer:
+    """🎥 Comprehensive video analysis and processing"""
+    
+    def __init__(self):
+        self.frame_cache = []
+        self.analysis_history = []
+        
+    def analyze_video_content(self, video_path: str) -> Dict:
+        """🎬 Understand what's in a video"""
+        print(f"🔍 Analyzing video: {video_path}")
+        
+        cap = cv2.VideoCapture(video_path)
+        
+        if not cap.isOpened():
+            return {"error": "❌ Cannot open video file"}
+        
+        # Get video properties
+        fps = int(cap.get(cv2.CAP_PROP_FPS))
+        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        duration = frame_count / fps if fps > 0 else 0
+        
+        print(f"📊 Video Info: {width}x{height}, {fps} FPS, {duration:.2f}s")
+        
+        # Analyze key frames
+        analysis = {
+            "video_info": {
+                "resolution": f"{width}x{height}",
+                "fps": fps,
+                "duration": f"{duration:.2f}s",
+                "frames": frame_count
+            },
+            "scenes": [],
+            "objects_detected": [],
+            "motion_analysis": {},
+            "quality_metrics": {}
+        }
+        
+        # Sample frames for analysis
+        sample_interval = max(1, frame_count // 10)  # Sample 10 frames
+        
+        for i in range(0, frame_count, sample_interval):
+            cap.set(cv2.CAP_PROP_POS_FRAMES, i)
+            ret, frame = cap.read()
+            
+            if ret:
+                # Analyze frame
+                frame_analysis = self._analyze_frame(frame, i)
+                analysis["scenes"].append(frame_analysis)
+        
+        cap.release()
+        
+        # Aggregate results
+        analysis["summary"] = self._generate_video_summary(analysis)
+        
+        return analysis
+    
+    def _analyze_frame(self, frame: np.ndarray, frame_num: int) -> Dict:
+        """🔍 Analyze individual frame"""
+        # Convert to different color spaces for analysis
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
+        
+        # Detect edges
+        edges = cv2.Canny(gray, 50, 150)
+        
+        # Detect objects using contours
+        contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        
+        # Calculate brightness
+        brightness = np.mean(gray)
+        
+        # Calculate color distribution
+        color_hist = cv2.calcHist([hsv], [0, 1], None, [50, 60], [0, 180, 0, 256])
+        
+        # Detect motion (simplified)
+        motion_score = np.std(gray)
+        
+        return {
+            "frame_number": frame_num,
+            "brightness": float(brightness),
+            "objects_count": len(contours),
+            "motion_score": float(motion_score),
+            "dominant_colors": self._extract_dominant_colors(color_hist),
+            "complexity": float(np.mean(edges))
+        }
+    
+    def _extract_dominant_colors(self, hist: np.ndarray) -> List[str]:
+        """🎨 Extract dominant colors from histogram"""
+        # Simplified color detection
+        colors = []
+        if np.max(hist) > 100:
+            colors.append("red")
+        if np.mean(hist) > 50:
+            colors.append("green")
+        if np.std(hist) > 30:
+            colors.append("blue")
+        return colors if colors else ["neutral"]
+    
+    def _generate_video_summary(self, analysis: Dict) -> Dict:
+        """📊 Generate video summary"""
+        scenes = analysis["scenes"]
+        
+        if not scenes:
+            return {"error": "No scenes analyzed"}
+        
+        avg_brightness = sum(s["brightness"] for s in scenes) / len(scenes)
+        avg_objects = sum(s["objects_count"] for s in scenes) / len(scenes)
+        avg_motion = sum(s["motion_score"] for s in scenes) / len(scenes)
+        
+        return {
+            "type": "educational" if avg_objects > 5 else "general",
+            "energy_level": "high" if avg_motion > 50 else "medium" if avg_motion > 30 else "low",
+            "brightness_level": "bright" if avg_brightness > 128 else "dark" if avg_brightness < 80 else "normal",
+            "content_complexity": "complex" if avg_objects > 10 else "moderate" if avg_objects > 5 else "simple",
+            "recommended_actions": self._recommend_actions(avg_brightness, avg_objects, avg_motion)
+        }
+    
+    def _recommend_actions(self, brightness: float, objects: float, motion: float) -> List[str]:
+        """💡 Recommend video improvements"""
+        actions = []
+        
+        if brightness < 80:
+            actions.append("🔆 Increase brightness for better visibility")
+        elif brightness > 200:
+            actions.append("🌙 Reduce brightness to avoid overexposure")
+        
+        if objects < 2:
+            actions.append("📦 Add more visual elements for engagement")
+        
+        if motion < 20:
+            actions.append("🎬 Add more dynamic movement")
+        
+        return actions
+    
+    def remove_objects(self, video_path: str, output_path: str, mask: np.ndarray = None) -> Dict:
+        """🎨 Remove objects from video using inpainting"""
+        print(f"🎨 Removing objects from video...")
+        
+        cap = cv2.VideoCapture(video_path)
+        
+        if not cap.isOpened():
+            return {"error": "❌ Cannot open video file"}
+        
+        # Get video properties
+        fps = int(cap.get(cv2.CAP_PROP_FPS))
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        
+        # Create video writer
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+        
+        frame_count = 0
+        processed_frames = 0
+        
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            
+            frame_count += 1
+            
+            # Apply object removal
+            if mask is not None:
+                # Use inpainting to remove masked objects
+                result = cv2.inpaint(frame, mask, 3, cv2.INPAINT_TELEA)
+            else:
+                result = frame
+            
+            out.write(result)
+            processed_frames += 1
+            
+            if processed_frames % 30 == 0:
+                print(f"✅ Processed {processed_frames} frames...")
+        
+        cap.release()
+        out.release()
+        
+        return {
+            "status": "✅ success",
+            "output_file": output_path,
+            "frames_processed": processed_frames,
+            "emoji": "🎨"
+        }
+    
+    def extract_information(self, video_path: str) -> Dict:
+        """📊 Extract information from video"""
+        print(f"📊 Extracting information from video...")
+        
+        analysis = self.analyze_video_content(video_path)
+        
+        # Extract text (simplified - would use OCR in production)
+        extracted_info = {
+            "video_metadata": analysis["video_info"],
+            "content_analysis": analysis["summary"],
+            "key_moments": self._identify_key_moments(analysis["scenes"]),
+            "detected_patterns": self._detect_patterns(analysis["scenes"]),
+            "emoji": "📊"
+        }
+        
+        return extracted_info
+    
+    def _identify_key_moments(self, scenes: List[Dict]) -> List[Dict]:
+        """🎯 Identify key moments in video"""
+        if not scenes:
+            return []
+        
+        # Find frames with high motion or many objects
+        key_moments = []
+        
+        for scene in scenes:
+            if scene["motion_score"] > 50 or scene["objects_count"] > 10:
+                key_moments.append({
+                    "frame": scene["frame_number"],
+                    "reason": "high_activity",
+                    "emoji": "🎬"
+                })
+        
+        return key_moments[:5]  # Top 5 key moments
+    
+    def _detect_patterns(self, scenes: List[Dict]) -> List[str]:
+        """🔍 Detect patterns in video"""
+        if not scenes:
+            return []
+        
+        patterns = []
+        
+        # Check for consistent brightness
+        brightness_values = [s["brightness"] for s in scenes]
+        if np.std(brightness_values) < 20:
+            patterns.append("🌟 Consistent lighting throughout")
+        
+        # Check for motion patterns
+        motion_values = [s["motion_score"] for s in scenes]
+        if np.mean(motion_values) > 50:
+            patterns.append("🎬 High-energy content")
+        elif np.mean(motion_values) < 20:
+            patterns.append("📚 Educational/tutorial content")
+        
+        return patterns
+    
+    def enhance_video_quality(self, video_path: str, output_path: str) -> Dict:
+        """🎯 Enhance video quality"""
+        print(f"🎯 Enhancing video quality...")
+        
+        cap = cv2.VideoCapture(video_path)
+        
+        if not cap.isOpened():
+            return {"error": "❌ Cannot open video file"}
+        
+        # Get video properties
+        fps = int(cap.get(cv2.CAP_PROP_FPS))
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        
+        # Create video writer
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+        
+        frame_count = 0
+        
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            
+            # Apply enhancements
+            enhanced = self._enhance_frame(frame)
+            
+            out.write(enhanced)
+            frame_count += 1
+        
+        cap.release()
+        out.release()
+        
+        return {
+            "status": "✅ success",
+            "output_file": output_path,
+            "frames_enhanced": frame_count,
+            "enhancements": [
+                "🔆 Brightness adjustment",
+                "🎨 Color correction",
+                "🔍 Sharpening",
+                "📊 Contrast enhancement"
+            ],
+            "emoji": "🎯"
+        }
+    
+    def _enhance_frame(self, frame: np.ndarray) -> np.ndarray:
+        """🎨 Enhance individual frame"""
+        # Convert to LAB color space
+        lab = cv2.cvtColor(frame, cv2.COLOR_BGR2LAB)
+        l, a, b = cv2.split(lab)
+        
+        # Apply CLAHE (Contrast Limited Adaptive Histogram Equalization)
+        clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
+        cl = clahe.apply(l)
+        
+        # Merge channels
+        limg = cv2.merge((cl, a, b))
+        enhanced = cv2.cvtColor(limg, cv2.COLOR_LAB2BGR)
+        
+        # Apply sharpening
+        kernel = np.array([[-1, -1, -1], [-1, 9, -1], [-1, -1, -1]])
+        sharpened = cv2.filter2D(enhanced, -1, kernel)
+        
+        # Blend original and sharpened
+        result = cv2.addWeighted(enhanced, 0.7, sharpened, 0.3, 0)
+        
+        return result
+    
+    def stabilize_video(self, video_path: str, output_path: str) -> Dict:
+        """📺 Stabilize video"""
+        print(f"📺 Stabilizing video...")
+        
+        cap = cv2.VideoCapture(video_path)
+        
+        if not cap.isOpened():
+            return {"error": "❌ Cannot open video file"}
+        
+        # Get video properties
+        fps = int(cap.get(cv2.CAP_PROP_FPS))
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        
+        # Read first frame
+        ret, prev_frame = cap.read()
+        if not ret:
+            return {"error": "❌ Cannot read first frame"}
+        
+        prev_gray = cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
+        
+        # Create video writer
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+        
+        # Stabilization transforms
+        transforms = []
+        frame_count = 0
+        
+        while True:
+            ret, curr_frame = cap.read()
+            if not ret:
+                break
+            
+            curr_gray = cv2.cvtColor(curr_frame, cv2.COLOR_BGR2GRAY)
+            
+            # Detect features
+            prev_pts = cv2.goodFeaturesToTrack(prev_gray, maxCorners=200, qualityLevel=0.01, minDistance=30, blockSize=3)
+            
+            if prev_pts is not None:
+                curr_pts, status, _ = cv2.calcOpticalFlowPyrLK(prev_gray, curr_gray, prev_pts, None)
+                
+                # Filter valid points
+                idx = np.where(status == 1)[0]
+                prev_pts = prev_pts[idx]
+                curr_pts = curr_pts[idx]
+                
+                if len(prev_pts) > 10:
+                    # Estimate transform
+                    m, _ = cv2.estimateAffinePartial2D(prev_pts, curr_pts)
+                    
+                    if m is not None:
+                        dx = m[0, 2]
+                        dy = m[1, 2]
+                        da = np.arctan2(m[1, 0], m[0, 0])
+                        
+                        transforms.append([dx, dy, da])
+            
+            prev_gray = curr_gray
+            frame_count += 1
+        
+        cap.release()
+        
+        # Apply stabilization
+        cap = cv2.VideoCapture(video_path)
+        
+        trajectory = np.cumsum(transforms, axis=0)
+        
+        # Smooth trajectory
+        smoothed = self._smooth_trajectory(trajectory)
+        
+        # Calculate stabilization transforms
+        diff = smoothed - trajectory
+        stabilization_transforms = []
+        
+        for i in range(len(diff)):
+            dx = diff[i, 0]
+            dy = diff[i, 1]
+            da = diff[i, 2]
+            
+            m = np.zeros((2, 3))
+            m[0, 0] = np.cos(da)
+            m[0, 1] = -np.sin(da)
+            m[1, 0] = np.sin(da)
+            m[1, 1] = np.cos(da)
+            m[0, 2] = dx
+            m[1, 2] = dy
+            
+            stabilization_transforms.append(m)
+        
+        # Apply transforms and write video
+        frame_idx = 0
+        
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            
+            if frame_idx < len(stabilization_transforms):
+                stabilized = cv2.warpAffine(frame, stabilization_transforms[frame_idx], (width, height))
+            else:
+                stabilized = frame
+            
+            out.write(stabilized)
+            frame_idx += 1
+        
+        cap.release()
+        out.release()
+        
+        return {
+            "status": "✅ success",
+            "output_file": output_path,
+            "frames_stabilized": frame_idx,
+            "stabilization_level": "high",
+            "emoji": "📺"
+        }
+    
+    def _smooth_trajectory(self, trajectory: np.ndarray, window_size: int = 30) -> np.ndarray:
+        """📊 Smooth trajectory using moving average"""
+        smoothed = np.zeros_like(trajectory)
+        
+        for i in range(3):  # x, y, angle
+            smoothed[:, i] = np.convolve(trajectory[:, i], 
+                                         np.ones(window_size)/window_size, 
+                                         mode='same')
+        
+        return smoothed
+    
+    def real_time_analysis(self, source=0) -> Dict:
+        """📺 Real-time video analysis from camera or screen"""
+        print(f"📺 Starting real-time analysis...")
+        print(f"📷 Source: {'Camera' if source == 0 else 'Screen'}")
+        
+        cap = cv2.VideoCapture(source)
+        
+        if not cap.isOpened():
+            return {"error": "❌ Cannot open video source"}
+        
+        print("✅ Real-time analysis started!")
+        print("👀 Press 'q' to quit")
+        
+        analysis_results = []
+        frame_count = 0
+        
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            
+            frame_count += 1
+            
+            # Analyze every 10th frame for performance
+            if frame_count % 10 == 0:
+                analysis = self._analyze_frame(frame, frame_count)
+                analysis_results.append(analysis)
+                
+                # Display analysis
+                print(f"\n📊 Frame {frame_count}:")
+                print(f"  🎯 Objects: {analysis['objects_count']}")
+                print(f"  💡 Brightness: {analysis['brightness']:.1f}")
+                print(f"  🎬 Motion: {analysis['motion_score']:.1f}")
+            
+            # Show frame (comment out for headless mode)
+            # cv2.imshow('Real-time Analysis', frame)
+            
+            # if cv2.waitKey(1) & 0xFF == ord('q'):
+            #     break
+        
+        cap.release()
+        # cv2.destroyAllWindows()
+        
+        return {
+            "status": "✅ success",
+            "frames_analyzed": frame_count,
+            "analysis_results": analysis_results,
+            "emoji": "📺"
+        }
+
+
+class CameraProcessor:
+    """📷 Camera processing and visual recognition"""
+    
+    def __init__(self):
+        self.recognition_history = []
+    
+    def process_camera_frame(self, frame: np.ndarray) -> Dict:
+        """📷 Process camera frame for visual recognition"""
+        # Convert to different formats
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
+        
+        # Detect faces (simplified)
+        face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
+        faces = face_cascade.detectMultiScale(gray, 1.1, 4)
+        
+        # Detect objects
+        edges = cv2.Canny(gray, 50, 150)
+        contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        
+        # Analyze scene
+        brightness = np.mean(gray)
+        contrast = np.std(gray)
+        
+        return {
+            "faces_detected": len(faces),
+            "objects_detected": len(contours),
+            "brightness": float(brightness),
+            "contrast": float(contrast),
+            "scene_type": self._classify_scene(brightness, len(contours)),
+            "emoji": "📷"
+        }
+    
+    def _classify_scene(self, brightness: float, object_count: int) -> str:
+        """🎯 Classify scene type"""
+        if brightness > 180 and object_count < 5:
+            return "🌟 Bright and simple"
+        elif brightness < 80:
+            return "🌙 Dark scene"
+        elif object_count > 20:
+            return "🎬 Complex scene"
+        else:
+            return "📚 Normal scene"
+    
+    def recognize_visual_elements(self, frame: np.ndarray) -> Dict:
+        """👁️ Recognize visual elements in frame"""
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        
+        # Detect edges and shapes
+        edges = cv2.Canny(gray, 50, 150)
+        
+        # Detect circles
+        circles = cv2.HoughCircles(gray, cv2.HOUGH_GRADIENT, 1, 20,
+                                   param1=50, param2=30, minRadius=0, maxRadius=0)
+        
+        # Detect lines
+        lines = cv2.HoughLinesP(edges, 1, np.pi/180, 50, minLineLength=50, maxLineGap=10)
+        
+        # Detect rectangles (simplified)
+        contours, _ = cv2.findContours(edges, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
+        rectangles = [c for c in contours if len(cv2.approxPolyDP(c, 0.02*cv2.contourArea(c), True)) == 4]
+        
+        return {
+            "circles": len(circles[0]) if circles is not None else 0,
+            "lines": len(lines) if lines is not None else 0,
+            "rectangles": len(rectangles),
+            "total_shapes": len(contours),
+            "emoji": "👁️"
+        }
+
+# Export classes
+__all__ = ['VideoAnalyzer', 'CameraProcessor']