Delete app.py

app.py

@@ -1,299 +0,0 @@
-"""
-Optical Flow Server for Hugging Face Spaces
-============================================
-Complete server for cloud-based optical flow processing.
-Deploy this directly to Hugging Face Spaces.
-"""
-
-import gradio as gr
-import cv2
-import numpy as np
-import base64
-from io import BytesIO
-from PIL import Image
-from sklearn.cluster import DBSCAN
-import colorsys
-from collections import deque
-import uuid
-import time
-
-
-class ServerOpticalFlowEngine:
-    """Optical Flow Engine optimized for server-side processing."""
-    
-    def __init__(self):
-        self.prev_gray = None
-        self.trail_points = {}
-        self.max_trail_length = 30
-        
-        self.bg_subtractor = cv2.createBackgroundSubtractorMOG2(
-            history=500, varThreshold=16, detectShadows=True
-        )
-        self.color_pool = self._generate_colors(20)
-        
-    def _generate_colors(self, n):
-        colors = []
-        for i in range(n):
-            hue = i / n
-            rgb = colorsys.hsv_to_rgb(hue, 0.9, 0.95)
-            bgr = (int(rgb[2] * 255), int(rgb[1] * 255), int(rgb[0] * 255))
-            colors.append(bgr)
-        return colors
-    
-    def compute_flow(self, gray):
-        if self.prev_gray is None:
-            self.prev_gray = gray.copy()
-            return None
-        
-        flow = cv2.calcOpticalFlowFarneback(
-            self.prev_gray, gray, None,
-            pyr_scale=0.5, levels=5, winsize=13,
-            iterations=10, poly_n=5, poly_sigma=1.1,
-            flags=cv2.OPTFLOW_FARNEBACK_GAUSSIAN
-        )
-        self.prev_gray = gray.copy()
-        return flow
-    
-    def segment_motion(self, frame):
-        fg_mask = self.bg_subtractor.apply(frame)
-        fg_mask[fg_mask == 127] = 0
-        
-        kernel_small = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3))
-        fg_mask = cv2.erode(fg_mask, kernel_small, iterations=2)
-        
-        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (7, 7))
-        fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, kernel)
-        
-        kernel_large = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (15, 15))
-        fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_CLOSE, kernel_large)
-        
-        fg_mask = cv2.dilate(fg_mask, kernel_small, iterations=1)
-        return fg_mask
-    
-    def detect_objects(self, mask, flow=None, min_area=2000):
-        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
-        
-        if not contours:
-            return None, {}
-        
-        label_image = np.full(mask.shape, -1, dtype=np.int32)
-        objects = {}
-        label_id = 0
-        
-        for contour in contours:
-            area = cv2.contourArea(contour)
-            if area < min_area:
-                continue
-            
-            x, y, w, h = cv2.boundingRect(contour)
-            aspect_ratio = float(h) / w if w > 0 else 0
-            if aspect_ratio < 0.2 or aspect_ratio > 8.0:
-                continue
-            
-            hull = cv2.convexHull(contour)
-            hull_area = cv2.contourArea(hull)
-            if hull_area > 0 and (area / hull_area) < 0.3:
-                continue
-            
-            cv2.drawContours(label_image, [contour], -1, label_id, -1)
-            
-            M = cv2.moments(contour)
-            cx = int(M['m10'] / M['m00']) if M['m00'] > 0 else x + w // 2
-            cy = int(M['m01'] / M['m00']) if M['m00'] > 0 else y + h // 2
-            
-            avg_velocity = (0, 0)
-            if flow is not None:
-                contour_mask = np.zeros(mask.shape, dtype=np.uint8)
-                cv2.drawContours(contour_mask, [contour], -1, 255, -1)
-                flow_x = flow[:, :, 0][contour_mask > 0]
-                flow_y = flow[:, :, 1][contour_mask > 0]
-                if len(flow_x) > 0:
-                    avg_velocity = (float(np.mean(flow_x)), float(np.mean(flow_y)))
-            
-            objects[label_id] = {
-                'centroid': (cx, cy),
-                'bbox': (x, y, x + w, y + h),
-                'velocity': avg_velocity,
-                'area': area,
-                'contour': contour
-            }
-            label_id += 1
-        
-        return label_image, objects if objects else (None, {})
-    
-    def update_trails(self, objects):
-        current_ids = set(objects.keys())
-        for old_id in set(self.trail_points.keys()) - current_ids:
-            del self.trail_points[old_id]
-        
-        for obj_id, obj in objects.items():
-            if obj_id not in self.trail_points:
-                self.trail_points[obj_id] = deque(maxlen=self.max_trail_length)
-            self.trail_points[obj_id].append(obj['centroid'])
-    
-    def draw_results(self, frame, objects, label_image):
-        output = frame.copy()
-        
-        for label, obj in objects.items():
-            color = self.color_pool[label % len(self.color_pool)]
-            x1, y1, x2, y2 = obj['bbox']
-            cv2.rectangle(output, (x1, y1), (x2, y2), color, 2)
-            
-            cx, cy = obj['centroid']
-            cv2.circle(output, (cx, cy), 5, color, -1)
-            
-            vx, vy = obj['velocity']
-            if np.sqrt(vx**2 + vy**2) > 1:
-                cv2.arrowedLine(output, (cx, cy), 
-                              (int(cx + vx * 5), int(cy + vy * 5)),
-                              (0, 255, 255), 3, tipLength=0.3)
-            
-            cv2.putText(output, f"Obj {label+1}", (x1, y1 - 25),
-                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
-            cv2.putText(output, f"Speed: {np.sqrt(vx**2 + vy**2):.1f}", 
-                       (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 1)
-            
-            if label_image is not None:
-                mask = (label_image == label).astype(np.uint8) * 255
-                contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
-                cv2.drawContours(output, contours, -1, color, 2)
-        
-        return output
-    
-    def draw_trails(self, frame):
-        output = frame.copy()
-        for obj_id, trail in self.trail_points.items():
-            if len(trail) < 2:
-                continue
-            color = self.color_pool[obj_id % len(self.color_pool)]
-            points = np.array(list(trail), dtype=np.int32)
-            for i in range(1, len(points)):
-                thickness = int(1 + (i / len(points)) * 3)
-                cv2.line(output, tuple(points[i-1]), tuple(points[i]), color, thickness)
-            cv2.circle(output, tuple(points[-1]), 6, color, -1)
-        return output
-    
-    def compute_heatmap(self, flow):
-        if flow is None:
-            return None
-        magnitude = np.sqrt(flow[..., 0]**2 + flow[..., 1]**2)
-        normalized = cv2.normalize(magnitude, None, 0, 255, cv2.NORM_MINMAX)
-        return cv2.applyColorMap(normalized.astype(np.uint8), cv2.COLORMAP_HOT)
-    
-    def process_frame(self, frame):
-        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-        flow = self.compute_flow(gray)
-        motion_mask = self.segment_motion(frame)
-        
-        results = {
-            'num_objects': 0,
-            'tracked': frame.copy(),
-            'trails': frame.copy(),
-            'heatmap': np.zeros_like(frame)
-        }
-        
-        if flow is not None:
-            results['heatmap'] = self.compute_heatmap(flow)
-            label_image, objects = self.detect_objects(motion_mask, flow)
-            
-            if objects:
-                results['num_objects'] = len(objects)
-                self.update_trails(objects)
-                results['tracked'] = self.draw_results(frame, objects, label_image)
-                results['trails'] = self.draw_trails(results['tracked'])
-        
-        return results
-    
-    def reset(self):
-        self.prev_gray = None
-        self.trail_points = {}
-
-
-# Session management
-sessions = {}
-SESSION_TIMEOUT = 300
-
-
-def get_or_create_session(session_id):
-    current_time = time.time()
-    expired = [sid for sid, data in sessions.items()
-               if current_time - data['last_access'] > SESSION_TIMEOUT]
-    for sid in expired:
-        del sessions[sid]
-    
-    if session_id not in sessions:
-        sessions[session_id] = {
-            'engine': ServerOpticalFlowEngine(),
-            'last_access': current_time
-        }
-    else:
-        sessions[session_id]['last_access'] = current_time
-    
-    return sessions[session_id]['engine']
-
-
-def decode_frame(base64_data):
-    img_data = base64.b64decode(base64_data)
-    img = Image.open(BytesIO(img_data))
-    return cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
-
-
-def encode_frame(frame):
-    _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
-    return base64.b64encode(buffer).decode('utf-8')
-
-
-def process_frame_api(frame_base64, session_id):
-    """Process a single frame."""
-    try:
-        engine = get_or_create_session(session_id)
-        frame = decode_frame(frame_base64)
-        results = engine.process_frame(frame)
-        
-        return {
-            'success': True,
-            'num_objects': results['num_objects'],
-            'tracked': encode_frame(results['tracked']),
-            'trails': encode_frame(results['trails']),
-            'heatmap': encode_frame(results['heatmap'])
-        }
-    except Exception as e:
-        return {'success': False, 'error': str(e)}
-
-
-def reset_session_api(session_id):
-    """Reset session state."""
-    if session_id in sessions:
-        sessions[session_id]['engine'].reset()
-    return {'success': True}
-
-
-def create_new_session():
-    """Create new session."""
-    session_id = str(uuid.uuid4())
-    get_or_create_session(session_id)
-    return session_id
-
-
-# Gradio interface
-with gr.Blocks(title="Optical Flow Server") as demo:
-    gr.Markdown("# 🎥 Optical Flow Processing Server")
-    gr.Markdown("Use with: `python optical_flow_advanced.py --server <this-url>`")
-    
-    with gr.Tab("Test"):
-        with gr.Row():
-            input_image = gr.Image(label="Input Frame", type="numpy")
-            output_image = gr.Image(label="Processed")
-        
-        def test_process(image):
-            if image is None:
-                return None
-            engine = ServerOpticalFlowEngine()
-            engine.process_frame(image)
-            return engine.process_frame(image)['tracked']
-        
-        input_image.change(test_process, inputs=input_image, outputs=output_image)
-
-
-if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)