Upload cleaning_heatmap.py

cleaning_heatmap.py

@@ -0,0 +1,1040 @@
+import cv2
+import numpy as np
+import json
+import os
+from datetime import datetime
+import gradio as gr
+from pathlib import Path
+import threading
+import time
+from collections import defaultdict
+import torch
+from ultralytics import YOLO
+import logging
+from typing import Dict, List, Tuple, Optional
+import base64
+from io import BytesIO
+from PIL import Image
+import matplotlib.pyplot as plt
+import matplotlib.cm as cm
+import cv2.legacy as cv2_legacy
+import shutil
+import tempfile
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class HygieneMonitor:
+    """Professional hygiene monitoring system for catering kitchen surveillance."""
+    
+    def __init__(self, model_path: str, confidence_threshold: float = 0.5):
+        """
+        Initialize the hygiene monitoring system.
+        
+        Args:
+            model_path: Path to the custom YOLO model
+            confidence_threshold: Minimum confidence for detections
+        """
+        self.model_path = model_path
+        self.confidence_threshold = confidence_threshold
+        self.model = None
+        # CHANGED: Replace heatmap_data with red_mask_data and erased_mask_data
+        self.red_mask_data = defaultdict(lambda: np.zeros((480, 640), dtype=np.uint8))
+        self.erased_mask_data = defaultdict(lambda: np.zeros((480, 640), dtype=np.uint8))
+        self.red_mask_created = defaultdict(bool)  # FIXED: Add this missing attribute
+        self.processing_active = False
+        self.current_video_path = None
+        self.table_mask = None
+        self.detection_history = []
+        
+        # CHANGED: Mask parameters instead of heatmap parameters
+        self.mask_intensity = 255               # Full intensity for red mask
+        self.gaussian_sigma = 80                # Blur radius for mask smoothing
+        self.intensity_threshold = 30           # Threshold for detecting table changes
+        
+        # NEW: Frame difference tracking for detecting table changes
+        self.previous_frame = None
+        self.table_changed = False
+        
+        # Tracking parameters
+        self.tracker = None
+        self.tracker_active = False
+        self.last_detection_bbox = None
+        
+        # Cleaning status tracking
+        self.detection_frames_count = 0
+        self.no_detection_frames_count = 0
+        self.cleaning_active = False
+        self.cleaning_start_threshold = 4  # frames
+        self.cleaning_stop_threshold = 10  # frames
+        
+        self._load_model()
+    
+    def _load_model(self) -> None:
+        """Load the custom YOLO model."""
+        try:
+            if not os.path.exists(self.model_path):
+                logger.error(f"Model file not found: {self.model_path}")
+                return
+            
+            self.model = YOLO(self.model_path)
+            logger.info(f"Model loaded successfully from {self.model_path}")
+        except Exception as e:
+            logger.error(f"Failed to load model: {str(e)}")
+            self.model = None
+    
+    def load_table_mask(self, mask_path: str) -> bool:
+        """
+        Load the binary mask for table areas.
+        
+        Args:
+            mask_path: Path to the binary mask image (white = table area)
+            
+        Returns:
+            bool: True if mask loaded successfully
+        """
+        try:
+            if not os.path.exists(mask_path):
+                logger.error(f"Mask file not found: {mask_path}")
+                return False
+            
+            mask = cv2.imread(mask_path, cv2.IMREAD_GRAYSCALE)
+            if mask is None:
+                logger.error(f"Failed to load mask from {mask_path}")
+                return False
+            
+            # Normalize mask to 0-1 range
+            self.table_mask = (mask > 128).astype(np.uint8)
+            logger.info(f"Table mask loaded successfully: {mask.shape}")
+            return True
+        except Exception as e:
+            logger.error(f"Error loading table mask: {str(e)}")
+            return False
+    
+    def create_default_mask(self, height: int, width: int) -> None:
+        """Create a default mask covering the entire frame."""
+        self.table_mask = np.ones((height, width), dtype=np.uint8)
+        logger.info("Using default mask (entire frame)")
+    
+    def detect_hand_with_cloth(self, frame: np.ndarray) -> List[Dict]:
+        """
+        Detect hands with cloth in the frame.
+        
+        Args:
+            frame: Input frame as numpy array
+            
+        Returns:
+            List of detection dictionaries with bbox and confidence
+        """
+        if self.model is None:
+            logger.warning("Model not loaded")
+            return []
+        
+        try:
+            results = self.model(frame, conf=self.confidence_threshold)
+            detections = []
+            
+            for result in results:
+                boxes = result.boxes
+                if boxes is not None:
+                    for box in boxes:
+                        confidence = float(box.conf[0])
+                        bbox = box.xyxy[0].cpu().numpy()  # x1, y1, x2, y2
+                        
+                        detection = {
+                            'bbox': bbox.tolist(),
+                            'confidence': confidence,
+                            'center': [(bbox[0] + bbox[2]) / 2, (bbox[1] + bbox[3]) / 2],
+                            'timestamp': datetime.now().isoformat()
+                        }
+                        detections.append(detection)
+            
+            return detections
+        except Exception as e:
+            logger.error(f"Detection error: {str(e)}")
+            return []
+
+    def init_tracker(self, frame: np.ndarray, bbox: List[float]) -> bool:
+        """
+        Initialize CSRT tracker with detection bbox.
+        
+        Args:
+            frame: Current frame
+            bbox: Bounding box [x1, y1, x2, y2]
+        
+        Returns:
+            bool: True if tracker initialized successfully
+        """
+        try:
+            self.tracker = cv2_legacy.TrackerCSRT_create()
+            # Convert bbox format from [x1, y1, x2, y2] to [x, y, w, h]
+            x1, y1, x2, y2 = bbox
+            tracker_bbox = (int(x1), int(y1), int(x2-x1), int(y2-y1))
+            
+            success = self.tracker.init(frame, tracker_bbox)
+            self.tracker_active = success
+            self.last_detection_bbox = bbox
+            
+            logger.info(f"Tracker initialized: {success}")
+            return success
+        except Exception as e:
+            logger.error(f"Failed to initialize tracker: {str(e)}")
+            return False
+
+    def update_tracker(self, frame: np.ndarray) -> Optional[Dict]:
+        """
+        Update tracker and return tracking result.
+        
+        Args:
+            frame: Current frame
+        
+        Returns:
+            Dict with tracking result or None if tracking failed
+        """
+        if not self.tracker_active or self.tracker is None:
+            return None
+        
+        try:
+            success, tracker_bbox = self.tracker.update(frame)
+            
+            if success:
+                # Convert back to [x1, y1, x2, y2] format
+                x, y, w, h = tracker_bbox
+                bbox = [x, y, x + w, y + h]
+                
+                tracking_result = {
+                    'bbox': bbox,
+                    'confidence': 0.7,  # Assign reasonable confidence for tracker
+                    'center': [(bbox[0] + bbox[2]) / 2, (bbox[1] + bbox[3]) / 2],
+                    'timestamp': datetime.now().isoformat(),
+                    'source': 'tracker'
+                }
+                
+                return tracking_result
+            else:
+                # Tracking failed, deactivate tracker
+                self.tracker_active = False
+                logger.info("Tracking lost")
+                return None
+                
+        except Exception as e:
+            logger.error(f"Tracker update error: {str(e)}")
+            self.tracker_active = False
+            return None
+
+    def update_cleaning_status(self, detections: List[Dict], frame_shape: Tuple[int, int]) -> str:
+        """
+        Update cleaning status based on detection patterns.
+        
+        Args:
+            detections: List of detection dictionaries
+            frame_shape: (height, width) of the frame
+        
+        Returns:
+            str: Current cleaning status
+        """
+        height, width = frame_shape[:2]
+        
+        # Check if any detection is in table area
+        table_detection = False
+        for detection in detections:
+            center_x, center_y = detection['center']
+            center_x, center_y = int(center_x), int(center_y)
+            
+            if (0 <= center_y < height and 0 <= center_x < width and 
+                self.table_mask is not None and self.table_mask[center_y, center_x] > 0):
+                table_detection = True
+                break
+        
+        # Update counters based on detection
+        if table_detection:
+            self.detection_frames_count += 1
+            self.no_detection_frames_count = 0  # Reset no detection counter
+        else:
+            self.no_detection_frames_count += 1
+            self.detection_frames_count = 0  # Reset detection counter
+        
+        # Update cleaning status
+        if not self.cleaning_active and self.detection_frames_count >= self.cleaning_start_threshold:
+            self.cleaning_active = True
+            logger.info("Cleaning started")
+            return "CLEANING STARTED"
+        elif self.cleaning_active and self.no_detection_frames_count >= self.cleaning_stop_threshold:
+            self.cleaning_active = False
+            logger.info("Cleaning stopped")
+            return "CLEANING STOPPED"
+        
+        # Return current status
+        return "CLEANING ACTIVE" if self.cleaning_active else "NO CLEANING"
+    
+    def _draw_professional_status_panel(self, frame: np.ndarray, cleaning_status: str, detection_count: int, tracking_active: bool) -> None:
+        """
+        Draw professional status panel with gradient background and modern styling.
+        
+        Args:
+            frame: Frame to draw on
+            cleaning_status: Current cleaning status
+            detection_count: Number of detections
+            tracking_active: Whether tracking is active
+        """
+        height, width = frame.shape[:2]
+        
+        # Panel dimensions and position
+        panel_width = 380
+        panel_height = 120
+        panel_x = width - panel_width - 20
+        panel_y = 20
+        
+        # Create gradient background overlay
+        overlay = frame.copy()
+        
+        # Draw rounded rectangle background with gradient effect
+        # Main panel background (dark semi-transparent)
+        cv2.rectangle(overlay, (panel_x, panel_y), (panel_x + panel_width, panel_y + panel_height), (20, 20, 20), -1)
+        
+        # Add subtle border
+        cv2.rectangle(overlay, (panel_x-2, panel_y-2), (panel_x + panel_width + 2, panel_y + panel_height + 2), (60, 60, 60), 2)
+        
+        # Blend overlay with original frame
+        cv2.addWeighted(frame, 0.3, overlay, 0.7, 0, frame)
+        
+        # Status color coding
+        if "ACTIVE" in cleaning_status or "STARTED" in cleaning_status:
+            status_color = (0, 200, 0)  # Green
+            status_bg_color = (0, 60, 0)  # Dark green
+        elif "STOPPED" in cleaning_status:
+            status_color = (0, 100, 255)  # Orange/Red
+            status_bg_color = (0, 30, 80)  # Dark red
+        else:
+            status_color = (128, 128, 128)  # Gray
+            status_bg_color = (40, 40, 40)  # Dark gray
+        
+        # Draw status indicator bar
+        status_bar_height = 8
+        cv2.rectangle(frame, (panel_x, panel_y), (panel_x + panel_width, panel_y + status_bar_height), status_color, -1)
+        
+        # Title section
+        title_y = panel_y + 30
+        cv2.putText(frame, "HYGIENE MONITOR", (panel_x + 15, title_y), 
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
+        
+        # Main status
+        main_status_y = title_y + 30
+        status_text = cleaning_status.replace("_", " ")
+        cv2.putText(frame, status_text, (panel_x + 15, main_status_y), 
+                    cv2.FONT_HERSHEY_DUPLEX, 0.8, status_color, 2)
+        
+        # Add timestamp
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        cv2.putText(frame, timestamp, (panel_x + panel_width - 80, panel_y + panel_height - 10), 
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.4, (150, 150, 150), 1)
+        
+        # Add detection/tracking indicators (small dots)
+        if detection_count > 0:
+            cv2.circle(frame, (panel_x + 350, title_y - 5), 4, (0, 255, 0), -1)  # Green dot for detection
+        
+        if tracking_active:
+            cv2.circle(frame, (panel_x + 365, title_y - 5), 4, (255, 255, 0), -1)
+
+    def draw_overlays(self, frame: np.ndarray, detections: List[Dict], tracker_result: Optional[Dict], cleaning_status: str) -> np.ndarray:
+        """
+        Draw professional status overlay on frame (no bounding boxes).
+        
+        Args:
+            frame: Input frame
+            detections: List of detections (for counting only)
+            tracker_result: Tracker result if available (for counting only)
+            cleaning_status: Current cleaning status
+        
+        Returns:
+            Frame with professional status overlay
+        """
+        result_frame = frame.copy()
+        height, width = result_frame.shape[:2]
+        
+        # Create professional status panel
+        self._draw_professional_status_panel(result_frame, cleaning_status, len(detections), tracker_result is not None)
+        
+        return result_frame
+
+    # CHANGED: New method to detect table changes using frame difference
+    def detect_table_changes(self, current_frame: np.ndarray) -> bool:
+        """
+        Detect if there are intensity changes on table area.
+        
+        Args:
+            current_frame: Current frame
+            
+        Returns:
+            bool: True if table changes detected
+        """
+        if self.previous_frame is None:
+            self.previous_frame = current_frame.copy()
+            return False
+        
+        # Convert to grayscale for comparison
+        current_gray = cv2.cvtColor(current_frame, cv2.COLOR_BGR2GRAY)
+        previous_gray = cv2.cvtColor(self.previous_frame, cv2.COLOR_BGR2GRAY)
+        
+        # Calculate frame difference
+        diff = cv2.absdiff(current_gray, previous_gray)
+        
+        # Apply table mask to focus only on table area
+        if self.table_mask is not None:
+            height, width = diff.shape
+            if self.table_mask.shape != (height, width):
+                table_mask_resized = cv2.resize(self.table_mask, (width, height))
+            else:
+                table_mask_resized = self.table_mask
+            
+            masked_diff = diff * table_mask_resized
+        else:
+            masked_diff = diff
+        
+        # Check if changes exceed threshold
+        mean_change = np.mean(masked_diff)
+        table_changed = mean_change > self.intensity_threshold
+        
+        # Update previous frame
+        self.previous_frame = current_frame.copy()
+        
+        return table_changed
+
+    # CHANGED: Replace update_heatmap with update_red_mask_and_erase
+    def update_red_mask_and_erase(self, detections: List[Dict], frame_shape: Tuple[int, int], table_changed: bool) -> None:
+        """
+        Update red mask based on table changes and erase it based on detections.
+        
+        Args:
+            detections: List of detection dictionaries
+            frame_shape: (height, width) of the frame
+            table_changed: Whether table changes were detected
+        """
+        height, width = frame_shape[:2]
+        video_key = self.current_video_path or "live"
+        
+        # Ensure masks match frame dimensions
+        if video_key not in self.red_mask_data:
+            self.red_mask_data[video_key] = np.zeros((height, width), dtype=np.uint8)
+            self.erased_mask_data[video_key] = np.zeros((height, width), dtype=np.uint8)
+        
+        # Ensure table mask matches frame dimensions
+        if self.table_mask is None:
+            self.create_default_mask(height, width)
+        elif self.table_mask.shape != (height, width):
+            self.table_mask = cv2.resize(self.table_mask, (width, height))
+        
+        # STEP 1: If table changes detected OR first detection on empty table, create red mask on table
+        if table_changed or (len(detections) > 0 and not self.red_mask_created[video_key]):
+            # Apply red mask to entire table area
+            self.red_mask_data[video_key] = np.where(
+                self.table_mask > 0, 
+                self.mask_intensity, 
+                self.red_mask_data[video_key]
+            )
+            self.red_mask_created[video_key] = True
+            logger.info("Red mask applied to table")
+        
+        # STEP 2: Erase red mask where detections occur
+        for detection in detections:
+            bbox = detection['bbox']  # x1, y1, x2, y2
+            
+            # Calculate 30% inner circular area of bounding box
+            bbox_width = bbox[2] - bbox[0]
+            bbox_height = bbox[3] - bbox[1]
+            
+            # Use smaller dimension to ensure circular area stays within bbox
+            min_dimension = min(bbox_width, bbox_height)
+            effective_radius = (min_dimension * 0.5) / 2
+            
+            center_x, center_y = detection['center']
+            center_x, center_y = int(center_x), int(center_y)
+            
+            if (0 <= center_y < height and 0 <= center_x < width and 
+                self.table_mask[center_y, center_x] > 0):
+                
+                # Create Gaussian blob around detection center with limited radius
+                y_indices, x_indices = np.ogrid[:height, :width]
+                distance_sq = (x_indices - center_x) ** 2 + (y_indices - center_y) ** 2
+                
+                # Use effective radius with much smoother falloff (increased multiplier and minimum)
+                gaussian_sigma = max(effective_radius * 2.5, 20)
+
+                # Create much smoother Gaussian distribution with softer edges
+                gaussian_mask = np.exp(-distance_sq / (2 * gaussian_sigma ** 2))
+                
+                # Apply table mask to the gaussian
+                masked_gaussian = gaussian_mask * self.table_mask
+                
+                # ERASE red mask where detection occurs with smooth blending
+                erase_intensity = masked_gaussian * self.mask_intensity * detection['confidence'] * 3
+
+                # Update erased mask with smooth blending instead of hard maximum
+                current_erased = self.erased_mask_data[video_key].astype(np.float32)
+                new_erase = erase_intensity.astype(np.float32)
+
+                # Smooth blending: use weighted average for overlapping areas
+                blended_erase = np.where(current_erased > 0, 
+                                        np.maximum(current_erased, current_erased * 0.7 + new_erase * 0.3),
+                                        new_erase)
+
+                self.erased_mask_data[video_key] = np.clip(blended_erase, 0, 255).astype(np.uint8)
+                
+                # Store detection in history
+                self.detection_history.append({
+                    'timestamp': detection['timestamp'],
+                    'center': [center_x, center_y],
+                    'confidence': detection['confidence'],
+                    'video': video_key
+                })
+
+    # CHANGED: Replace generate_heatmap_overlay with generate_red_mask_overlay
+    def generate_red_mask_overlay(self, frame: np.ndarray, alpha: float = 0.6) -> np.ndarray:
+        """
+        Generate red mask overlay on the frame, with erased areas removed.
+        
+        Args:
+            frame: Original frame
+            alpha: Transparency of red mask overlay
+            
+        Returns:
+            Frame with red mask overlay
+        """
+        video_key = self.current_video_path or "live"
+        red_mask = self.red_mask_data[video_key]
+        erased_mask = self.erased_mask_data[video_key]
+        
+        if red_mask.max() == 0:
+            return frame
+        
+        # Apply table mask to red mask
+        if self.table_mask is not None:
+            # Ensure mask dimensions match
+            if self.table_mask.shape != red_mask.shape:
+                mask_resized = cv2.resize(self.table_mask.astype(np.uint8), 
+                                        (red_mask.shape[1], red_mask.shape[0]))
+            else:
+                mask_resized = self.table_mask
+            
+            # Apply table mask
+            masked_red_mask = red_mask * mask_resized
+        else:
+            masked_red_mask = red_mask
+        
+        # Subtract erased areas from red mask
+        final_red_mask = np.maximum(0, masked_red_mask.astype(np.int16) - erased_mask.astype(np.int16))
+        final_red_mask = final_red_mask.astype(np.uint8)
+        
+        if final_red_mask.max() == 0:
+            return frame
+        
+        # Create red colored mask
+        red_colored_mask = np.zeros_like(frame)
+        red_colored_mask[:, :, 2] = final_red_mask  # Red channel
+        
+        # Create mask for non-zero areas
+        mask = final_red_mask > 5
+        
+        # Blend with original frame
+        result = frame.copy()
+        result[mask] = cv2.addWeighted(frame[mask], 1-alpha, red_colored_mask[mask], alpha, 0)
+        
+        return result
+    
+    def process_video(self, video_path: str, output_dir: str = "output") -> Dict:
+        """
+        Process entire video and generate red mask data with tracking and status overlay.
+        
+        Args:
+            video_path: Path to input video
+            output_dir: Directory for output files
+            
+        Returns:
+            Dictionary with processing results
+        """
+        if not os.path.exists(video_path):
+            return {"error": "Video file not found"}
+        
+        self.current_video_path = video_path
+        self.processing_active = True
+        
+        # Reset tracking and status variables
+        self.tracker = None
+        self.tracker_active = False
+        self.detection_frames_count = 0
+        self.no_detection_frames_count = 0
+        self.cleaning_active = False
+        self.previous_frame = None  # Reset frame difference tracking
+        
+        # Create output directory
+        os.makedirs(output_dir, exist_ok=True)
+        
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            return {"error": "Failed to open video"}
+        
+        # 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))
+        
+        # Initialize masks for this video
+        self.red_mask_data[video_path] = np.zeros((height, width), dtype=np.uint8)
+        self.erased_mask_data[video_path] = np.zeros((height, width), dtype=np.uint8)
+        
+        # FIXED: Better output video path and codec handling
+        output_video_path = os.path.join(output_dir, f"{Path(video_path).stem}_hygiene_monitor.mp4")
+        
+        # FIXED: Try different codecs for better compatibility
+        fourcc_options = [
+            cv2.VideoWriter_fourcc(*'mp4v'),
+            cv2.VideoWriter_fourcc(*'XVID'),
+            cv2.VideoWriter_fourcc(*'MJPG'),
+            cv2.VideoWriter_fourcc(*'H264')
+        ]
+        
+        out = None
+        for fourcc in fourcc_options:
+            out = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height))
+            if out.isOpened():
+                logger.info(f"Video writer initialized with codec: {fourcc}")
+                break
+            out.release()
+        
+        if out is None or not out.isOpened():
+            cap.release()
+            return {"error": "Failed to initialize video writer"}
+
+        # Process frames
+        frame_idx = 0
+        all_detections = []
+        
+        try:
+            logger.info(f"Starting video processing: {frame_count} frames")
+            
+            while cap.isOpened() and self.processing_active:
+                ret, frame = cap.read()
+                if not ret:
+                    logger.info(f"End of video reached at frame {frame_idx}")
+                    break
+                
+                # Detect table changes
+                table_changed = self.detect_table_changes(frame)
+                
+                # Detect hands with cloth
+                detections = self.detect_hand_with_cloth(frame)
+                
+                # Handle tracking
+                tracker_result = None
+                all_objects = []  # Combined detections and tracking
+                
+                if detections:
+                    # We have detections, add them to objects list
+                    all_objects.extend(detections)
+                    
+                    # Initialize tracker if not active
+                    if not self.tracker_active and len(detections) > 0:
+                        self.init_tracker(frame, detections[0]['bbox'])  # Use first detection
+                    
+                    # Reset no-detection counter since we have detections
+                    self.tracker_active = False  # Prioritize detection over tracking
+                else:
+                    # No detections, try tracking
+                    if self.tracker_active:
+                        tracker_result = self.update_tracker(frame)
+                        if tracker_result:
+                            all_objects.append(tracker_result)
+                    elif self.last_detection_bbox is not None:
+                        # Try to reinitialize tracker from last known position
+                        self.init_tracker(frame, self.last_detection_bbox)
+                
+                # Update cleaning status
+                cleaning_status = self.update_cleaning_status(all_objects, frame.shape)
+                
+                # Update red mask and erase instead of heatmap
+                self.update_red_mask_and_erase(all_objects, frame.shape, table_changed)
+                
+                # Generate red mask overlay instead of heatmap
+                frame_with_mask = self.generate_red_mask_overlay(frame, alpha=0.4)
+                
+                # Draw status overlay
+                final_frame = self.draw_overlays(frame_with_mask, detections, tracker_result, cleaning_status)
+                
+                # FIXED: Ensure frame is properly formatted before writing
+                if final_frame is not None and final_frame.shape[0] > 0 and final_frame.shape[1] > 0:
+                    # Ensure frame dimensions match video writer expectations
+                    if final_frame.shape[:2] != (height, width):
+                        final_frame = cv2.resize(final_frame, (width, height))
+                    
+                    # Write frame to output video
+                    success = out.write(final_frame)
+                    if not success and frame_idx < 10:  # Only warn for first few frames
+                        logger.warning(f"Failed to write frame {frame_idx}")
+                
+                # Store frame detections
+                frame_detections = {
+                    'frame_id': frame_idx,
+                    'timestamp': frame_idx / fps,
+                    'detections': detections,
+                    'tracker_result': tracker_result,
+                    'cleaning_status': cleaning_status,
+                    'table_changed': bool(table_changed)  # Convert to Python bool
+                }
+                all_detections.append(frame_detections)
+                
+                frame_idx += 1
+                
+                # Progress update
+                if frame_idx % 30 == 0:
+                    progress = (frame_idx / frame_count) * 100
+                    logger.info(f"Processing progress: {progress:.1f}% ({frame_idx}/{frame_count} frames)")
+        
+        except Exception as e:
+            logger.error(f"Error during video processing: {str(e)}")
+            return {"error": f"Processing error: {str(e)}"}
+        
+        finally:
+            cap.release()
+            if out is not None:
+                out.release()
+            logger.info(f"Video processing completed. Processed {frame_idx} frames")
+
+        # FIXED: Verify output video was created
+        if not os.path.exists(output_video_path):
+            return {"error": "Output video file was not created"}
+        
+        # Check if output video has reasonable size
+        if os.path.getsize(output_video_path) < 1024:  # Less than 1KB
+            return {"error": "Output video file is too small - may be corrupted"}
+
+        # Generate output files
+        results = self._save_results(video_path, all_detections, output_dir)
+        
+        # FIXED: Add output video path to results
+        results['output_video_path'] = output_video_path
+        results['frames_processed'] = frame_idx
+        
+        self.processing_active = False
+        
+        logger.info(f"Final output video saved to: {output_video_path}")
+        return results
+
+        
+    
+    def _save_results(self, video_path: str, detections: List[Dict], output_dir: str) -> Dict:
+        """Save processing results to files."""
+        try:
+            video_name = Path(video_path).stem
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            
+            # FIXED: Get output video path
+            output_video_path = os.path.join(output_dir, f"{video_name}_hygiene_monitor.mp4")
+            
+            # Save JSON data
+            json_path = os.path.join(output_dir, f"{video_name}_{timestamp}.json")
+            results_data = {
+                'video_path': video_path,
+                'output_video_path': output_video_path,  # ADDED
+                'processing_timestamp': datetime.now().isoformat(),
+                'total_detections': len([d for frame in detections for d in frame['detections']]),
+                'total_frames': len(detections),
+                'red_mask_stats': {
+                    'max_red_intensity': int(self.red_mask_data[video_path].max()),
+                    'total_red_area': int(np.sum(self.red_mask_data[video_path] > 0)),
+                    'erased_area': int(np.sum(self.erased_mask_data[video_path] > 0)),
+                    'remaining_red_area': int(np.sum((self.red_mask_data[video_path] > 0) & (self.erased_mask_data[video_path] == 0)))
+                },
+                'frame_detections': detections
+            }
+            
+            with open(json_path, 'w') as f:
+                json.dump(results_data, f, indent=2)
+            
+            # Save red mask visualization
+            mask_path = os.path.join(output_dir, f"{video_name}_{timestamp}_redmask.png")
+            self._save_red_mask_image(video_path, mask_path)
+            
+            return {
+                'success': True,
+                'json_path': json_path,
+                'heatmap_path': mask_path,
+                'output_video_path': output_video_path,  # ADDED
+                'stats': results_data['red_mask_stats']
+            }
+        
+        except Exception as e:
+            logger.error(f"Error saving results: {str(e)}")
+            return {'error': str(e)}
+    
+    # CHANGED: New method to save red mask image
+    def _save_red_mask_image(self, video_key: str, output_path: str) -> None:
+        """Save red mask as image file."""
+        red_mask = self.red_mask_data[video_key]
+        erased_mask = self.erased_mask_data[video_key]
+        
+        if red_mask.max() == 0:
+            return
+        
+        # Calculate final mask (red - erased)
+        final_mask = np.maximum(0, red_mask.astype(np.int16) - erased_mask.astype(np.int16))
+        
+        plt.figure(figsize=(12, 8))
+        plt.imshow(final_mask, cmap='Reds', interpolation='bilinear')
+        plt.colorbar(label='Red Mask Intensity (Remaining)')
+        plt.title('Table Red Mask (After Cleaning Erasure)')
+        plt.axis('off')
+        plt.tight_layout()
+        plt.savefig(output_path, dpi=300, bbox_inches='tight')
+        plt.close()
+    
+    def reset_heatmap(self, video_key: str = None) -> None:  # Keep method name for interface compatibility
+        """Reset mask data."""
+        if video_key:
+            self.red_mask_data[video_key] = np.zeros_like(self.red_mask_data[video_key])
+            self.erased_mask_data[video_key] = np.zeros_like(self.erased_mask_data[video_key])
+        else:
+            self.red_mask_data.clear()
+            self.erased_mask_data.clear()
+        self.detection_history.clear()
+        self.red_mask_created.clear()
+        self.previous_frame = None
+        logger.info("Red mask data reset")
+
+# Gradio Interface
+class HygieneMonitorInterface:
+    """Professional Gradio interface for the hygiene monitoring system."""
+    
+    def __init__(self):
+        self.monitor = None
+        self.live_processing = False
+        self.live_thread = None
+    
+    def initialize_monitor(self, model_file, confidence: float, mask_file=None) -> str:
+        """Initialize the monitoring system with proper file handling."""
+        try:
+            # FIXED: Handle Gradio file objects properly
+            if model_file is None:
+                return "❌ Please upload a model file"
+            
+            # Get the file path from Gradio file object
+            if hasattr(model_file, 'name'):
+                model_path = model_file.name
+            else:
+                model_path = str(model_file)
+                
+            # Validate model file exists and has correct extension
+            if not os.path.exists(model_path):
+                return f"❌ Model file not found: {model_path}"
+            
+            if not model_path.lower().endswith(('.pt', '.pth')):
+                return "❌ Please upload a valid YOLO model file (.pt or .pth)"
+            
+            # FIXED: Copy model to a safe location to avoid permission issues
+            temp_dir = tempfile.gettempdir()
+            safe_model_path = os.path.join(temp_dir, f"model_{int(time.time())}.pt")
+            
+            try:
+                shutil.copy2(model_path, safe_model_path)
+                model_path = safe_model_path
+            except Exception as copy_error:
+                logger.warning(f"Could not copy model file: {copy_error}. Using original path.")
+            
+            # Initialize monitor with safe path
+            self.monitor = HygieneMonitor(model_path, confidence)
+            
+            # Handle mask file if provided
+            if mask_file is not None:
+                if hasattr(mask_file, 'name'):
+                    mask_path = mask_file.name
+                else:
+                    mask_path = str(mask_file)
+                    
+                if os.path.exists(mask_path):
+                    success = self.monitor.load_table_mask(mask_path)
+                    if not success:
+                        return "⚠️ Model loaded but failed to load table mask. Using default mask."
+                else:
+                    return "⚠️ Model loaded but mask file not found. Using default mask."
+            
+            return "✅ System initialized successfully!"
+            
+        except Exception as e:
+            logger.error(f"Initialization error: {str(e)}")
+            return f"❌ Initialization failed: {str(e)}"
+    
+    def process_video_interface(self, video_file, progress=gr.Progress()) -> Tuple[str, str, str]:
+        """Process video through Gradio interface with proper error handling."""
+        if self.monitor is None:
+            return "❌ Please initialize the system first", "", ""
+        
+        if video_file is None:
+            return "❌ Please upload a video file", "", ""
+        
+        try:
+            progress(0, desc="Starting video processing...")
+            
+            # FIXED: Handle Gradio video file object properly
+            if hasattr(video_file, 'name'):
+                video_path = video_file.name
+            else:
+                video_path = str(video_file)
+            
+            # Validate video file
+            if not os.path.exists(video_path):
+                return f"❌ Video file not found: {video_path}", "", ""
+            
+            # Check if it's a file (not directory)
+            if not os.path.isfile(video_path):
+                return f"❌ Path is not a file: {video_path}", "", ""
+            
+            # FIXED: Create a safe output directory in temp
+            # output_dir = os.path.join(tempfile.gettempdir(), f"hygiene_output_{int(time.time())}")
+            output_dir = os.path.join("output/", f"hygiene_output_{int(time.time())}")
+            os.makedirs(output_dir, exist_ok=True)
+            
+            # Process video
+            results = self.monitor.process_video(video_path, output_dir)
+            
+            if 'error' in results:
+                return f"❌ Processing failed: {results['error']}", "", ""
+            
+            progress(1, desc="Processing complete!")
+            
+            # Prepare results
+            stats_text = f"""
+            📊 **Processing Results:**
+            - JSON Output: {results['json_path']}
+            - Red Mask Image: {results['heatmap_path']}
+            - Max Red Intensity: {results['stats']['max_red_intensity']}
+            - Total Red Area: {results['stats']['total_red_area']} pixels
+            - Erased Area: {results['stats']['erased_area']} pixels
+            - Remaining Red Area: {results['stats']['remaining_red_area']} pixels
+                        """
+            
+            return "✅ Video processed successfully!", stats_text, results['heatmap_path']
+        
+        except Exception as e:
+            logger.error(f"Processing error: {str(e)}")
+            return f"❌ Processing error: {str(e)}", "", ""
+    
+    def start_live_monitoring(self, camera_index: int = 0) -> str:
+        """Start live camera monitoring."""
+        if self.monitor is None:
+            return "❌ Please initialize the system first"
+        
+        if self.live_processing:
+            return "⚠️ Live monitoring already active"
+        
+        self.live_processing = True
+        self.live_thread = threading.Thread(target=self._live_monitoring_loop, args=(camera_index,))
+        self.live_thread.daemon = True
+        self.live_thread.start()
+        
+        return "✅ Live monitoring started"
+    
+    def stop_live_monitoring(self) -> str:
+        """Stop live monitoring."""
+        self.live_processing = False
+        if self.live_thread:
+            self.live_thread.join(timeout=2)
+        return "🛑 Live monitoring stopped"
+    
+    def _live_monitoring_loop(self, camera_index: int) -> None:
+        """Live monitoring loop (runs in separate thread)."""
+        cap = cv2.VideoCapture(camera_index)
+        
+        try:
+            while self.live_processing and cap.isOpened():
+                ret, frame = cap.read()
+                if not ret:
+                    continue
+                
+                # Detect table changes
+                table_changed = self.monitor.detect_table_changes(frame)
+                
+                # Process frame
+                detections = self.monitor.detect_hand_with_cloth(frame)
+                self.monitor.update_red_mask_and_erase(detections, frame.shape, table_changed)
+                
+                time.sleep(0.1)  # Limit processing rate
+        finally:
+            cap.release()
+    
+    def create_interface(self) -> gr.Interface:
+        """Create the Gradio interface."""
+        with gr.Blocks(title="Kitchen Hygiene Monitor", theme=gr.themes.Soft()) as interface:
+            gr.Markdown("""
+            # 🍽️ Kitchen Hygiene Monitoring System
+            Professional AI-powered solution for monitoring table cleaning activities in catering kitchens.
+            """)
+            
+            with gr.Tab("🔧 System Setup"):
+                gr.Markdown("### Initialize the monitoring system")
+                
+                with gr.Row():
+                    model_file = gr.File(
+                        label="Upload YOLO Model (.pt)", 
+                        file_types=[".pt", ".pth"],
+                        file_count="single"
+                    )
+                    mask_file = gr.File(
+                        label="Upload Table Mask (optional)", 
+                        file_types=[".png", ".jpg", ".jpeg"],
+                        file_count="single"
+                    )
+                
+                confidence_slider = gr.Slider(0.1, 1.0, value=0.3, label="Detection Confidence Threshold")
+                init_btn = gr.Button("Initialize System", variant="primary")
+                init_status = gr.Textbox(label="Status", interactive=False)
+                
+                init_btn.click(
+                    self.initialize_monitor,
+                    inputs=[model_file, confidence_slider, mask_file],
+                    outputs=init_status
+                )
+            
+            with gr.Tab("📹 Video Processing"):
+                gr.Markdown("### Process video files for hygiene analysis")
+                
+                video_input = gr.File(
+                    label="Upload Video", 
+                    file_types=[".mp4", ".avi", ".mov", ".mkv"],
+                    file_count="single"
+                )
+                process_btn = gr.Button("Process Video", variant="primary")
+                
+                with gr.Row():
+                    with gr.Column():
+                        process_status = gr.Textbox(label="Processing Status", interactive=False)
+                        results_text = gr.Markdown(label="Results")
+                    
+                    with gr.Column():
+                        heatmap_output = gr.Image(label="Generated Red Mask")
+                
+                process_btn.click(
+                    self.process_video_interface,
+                    inputs=[video_input],
+                    outputs=[process_status, results_text, heatmap_output]
+                )
+
+        return interface
+
+def main():
+    """Main function to run the application."""
+    # Create interface
+    interface_manager = HygieneMonitorInterface()
+    app = interface_manager.create_interface()
+    
+    # Launch with appropriate settings for RunPod/production
+    app.launch(
+        server_name="0.0.0.0",  # Allow external connections
+        server_port=7860,
+        share=True,  # Create shareable link
+        show_error=True,
+        quiet=False
+    )
+
+if __name__ == "__main__":
+    main()