Update app.py

app.py

@@ -1,123 +1,231 @@
 import gradio as gr
-from azure.storage.blob import BlobServiceClient
+from azure.storage.blob import BlobServiceClient, BlobClient
 import os
 import cv2
 import tempfile
 from ultralytics import YOLO
-import numpy as np
+import logging
 from datetime import datetime
 
-# Azure Storage config
-SAS_TOKEN = "sv=2024-11-04&ss=bfqt&srt=sco&sp=rwdlacupiytfx&se=2025-04-30T04:25:22Z&st=2025-04-16T20:25:22Z&spr=https&sig=HYrJBoOYc4PRe%2BoqBMl%2FmoL5Kz4ZYugbTLuEh63sbeo%3D"
-ACCOUNT_NAME = "assentian"
-CONTAINER_URL = f"https://{ACCOUNT_NAME}.blob.core.windows.net/logs"
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Azure Storage Configuration
+AZURE_ACCOUNT_NAME = "assentian"
+AZURE_SAS_TOKEN = "sv=2024-11-04&ss=bfqt&srt=sco&sp=rwdlacupiytfx&se=2025-04-30T04:25:22Z&st=2025-04-16T20:25:22Z&spr=https&sig=HYrJBoOYc4PRe%2BoqBMl%2FmoL5Kz4ZYugbTLuEh63sbeo%3D"
 CONTAINER_NAME = "logs"
-VIDEOS_FOLDER = ""
+VIDEO_PREFIX = ""
+
+# Initialize YOLO Model
+try:
+    YOLO_MODEL = YOLO("./best_yolov11.pt")
+    logger.info("YOLO model loaded successfully")
+except Exception as e:
+    logger.error(f"Failed to load YOLO model: {e}")
+    raise
 
-# YOLO model
-YOLO_MODEL = YOLO("./best_yolov11.pt")  # Ensure model is uploaded
+# Azure Blob Service Client
+def get_blob_service_client():
+    return BlobServiceClient(
+        account_url=f"https://{AZURE_ACCOUNT_NAME}.blob.core.windows.net",
+        credential=AZURE_SAS_TOKEN
+    )
 
-def list_videos():
+def list_azure_videos():
     try:
-        blob_service_client = BlobServiceClient(account_url=f"https://{ACCOUNT_NAME}.blob.core.windows.net", credential=SAS_TOKEN)
+        blob_service_client = get_blob_service_client()
         container_client = blob_service_client.get_container_client(CONTAINER_NAME)
-        blobs = container_client.list_blobs(name_starts_with=VIDEOS_FOLDER)
-        videos = [blob.name for blob in blobs if blob.name.endswith(".mp4")]
-        return videos if videos else ["No videos found"]
+        blobs = container_client.list_blobs(name_starts_with=VIDEO_PREFIX)
+        return [blob.name for blob in blobs if blob.name.lower().endswith(".mp4")]
     except Exception as e:
-        return [f"Error: {str(e)}"]
+        logger.error(f"Error listing videos: {e}")
+        return []
 
-def get_latest_video():
+def get_latest_azure_video():
     try:
-        blob_service_client = BlobServiceClient(account_url=f"https://{ACCOUNT_NAME}.blob.core.windows.net", credential=SAS_TOKEN)
+        blob_service_client = get_blob_service_client()
         container_client = blob_service_client.get_container_client(CONTAINER_NAME)
-        blobs = container_client.list_blobs(name_starts_with=VIDEOS_FOLDER)
+        blobs = container_client.list_blobs(name_starts_with=VIDEO_PREFIX)
+        
         latest_blob = None
         latest_time = None
+        
         for blob in blobs:
-            if blob.name.endswith(".mp4"):
+            if blob.name.lower().endswith(".mp4"):
                 blob_client = container_client.get_blob_client(blob.name)
-                props = blob_client.get_blob_properties()
-                if not latest_time or props.last_modified > latest_time:
-                    latest_time = props.last_modified
+                properties = blob_client.get_blob_properties()
+                if not latest_time or properties.last_modified > latest_time:
+                    latest_time = properties.last_modified
                     latest_blob = blob.name
-        return latest_blob if latest_blob else "No videos found"
+        
+        return latest_blob if latest_blob else None
     except Exception as e:
-        return f"Error: {str(e)}"
+        logger.error(f"Error finding latest video: {e}")
+        return None
 
-def annotate_video_with_bboxes(video_path):
+def download_azure_video(blob_name):
     try:
-        if not video_path or video_path == "No videos found":
-            return "Error: No video selected"
+        blob_service_client = get_blob_service_client()
+        blob_client = blob_service_client.get_blob_client(
+            container=CONTAINER_NAME,
+            blob=blob_name
+        )
         
-        # Download video from Azure
-        blob_service_client = BlobServiceClient(account_url=f"https://{ACCOUNT_NAME}.blob.core.windows.net", credential=SAS_TOKEN)
-        blob_client = blob_service_client.get_blob_client(CONTAINER_NAME, video_path)
-        video_data = blob_client.download_blob().readall()
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_file:
+            download_stream = blob_client.download_blob()
+            temp_file.write(download_stream.readall())
+            return temp_file.name
         
-        # Save to temp file
-        temp_input = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
-        temp_input.write(video_data)
-        temp_input.close()
-        
-        # Process video with YOLO
-        cap = cv2.VideoCapture(temp_input.name)
+    except Exception as e:
+        logger.error(f"Download failed: {e}")
+        return None
+
+def annotate_video(input_video_path):
+    try:
+        if not input_video_path or not os.path.exists(input_video_path):
+            logger.error("Invalid input video path")
+            return None
+
+        cap = cv2.VideoCapture(input_video_path)
         if not cap.isOpened():
-            os.remove(temp_input.name)
-            return "Error: Could not open video"
-        
+            logger.error("Failed to open video file")
+            return None
+
+        # Video writer setup
+        frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
         fps = cap.get(cv2.CAP_PROP_FPS)
-        w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-        h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        out_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
-        annotated_video_path = out_file.name
-        out_file.close()
-        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-        writer = cv2.VideoWriter(annotated_video_path, fourcc, fps, (w, h))
         
-        while True:
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_output:
+            output_path = temp_output.name
+        
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        writer = cv2.VideoWriter(output_path, fourcc, fps, (frame_width, frame_height))
+
+        # Processing loop
+        while cap.isOpened():
             ret, frame = cap.read()
             if not ret:
                 break
+
+            # YOLO inference
             results = YOLO_MODEL(frame)
-            frame_counts = {}
-            for r in results:
-                boxes = r.boxes
-                for box in boxes:
+            class_counts = {}
+
+            for result in results:
+                for box in result.boxes:
                     cls_id = int(box.cls[0])
                     conf = float(box.conf[0])
                     if conf < 0.5:
                         continue
-                    x1, y1, x2, y2 = box.xyxy[0]
+
+                    # Bounding box
+                    x1, y1, x2, y2 = map(int, box.xyxy[0])
                     class_name = YOLO_MODEL.names[cls_id]
-                    x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
-                    color = (0, 255, 0)
+                    color = (0, 255, 0)  # BGR format
+                    
+                    # Draw rectangle
                     cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
-                    label_text = f"{class_name} {conf:.2f}"
-                    cv2.putText(frame, label_text, (x1, y1 - 6),
-                                cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255,255,255), 1)
-                    frame_counts[class_name] = frame_counts.get(class_name, 0) + 1
-            summary_str = ", ".join(f"{cls_name}: {count}" for cls_name, count in frame_counts.items())
-            cv2.putText(frame, summary_str, (15, 30),
-                        cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 0), 2)
+                    
+                    # Text label
+                    label = f"{class_name} {conf:.2f}"
+                    cv2.putText(frame, label, (x1, y1 - 10),
+                                cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2)
+                    
+                    # Update counts
+                    class_counts[class_name] = class_counts.get(class_name, 0) + 1
+
+            # Add summary overlay
+            summary_text = " | ".join([f"{k}: {v}" for k, v in class_counts.items()])
+            cv2.putText(frame, summary_text, (10, 30),
+                        cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)
+            
             writer.write(frame)
-        
+
+        # Cleanup
         cap.release()
         writer.release()
-        os.remove(temp_input.name)
-        return annotated_video_path
+        os.remove(input_video_path)
+        
+        return output_path
+
+    except Exception as e:
+        logger.error(f"Annotation failed: {e}")
+        if 'cap' in locals(): cap.release()
+        if 'writer' in locals(): writer.release()
+        return None
+
+def process_video(blob_name):
+    try:
+        local_path = download_azure_video(blob_name)
+        if not local_path:
+            return None
+        return annotate_video(local_path)
     except Exception as e:
-        return f"Error: {str(e)}"
-
-# Gradio UI
-with gr.Blocks() as demo:
-    gr.Markdown("## PRISM Site Diary V3 - Video Annotator")
-    video_dropdown = gr.Dropdown(label="Select Video from Azure Blob", choices=list_videos(), interactive=True, allow_custom_value=True)
-    auto_btn = gr.Button("Annotate Latest Video")
-    manual_btn = gr.Button("Annotate Selected Video")
-    output_video = gr.Video(label="Annotated Video")
-    auto_btn.click(fn=annotate_video_with_bboxes, inputs=gr.State(get_latest_video()), outputs=output_video)
-    manual_btn.click(fn=annotate_video_with_bboxes, inputs=video_dropdown, outputs=output_video)
-    video_dropdown.change(fn=list_videos, outputs=video_dropdown)
-
-demo.launch()
+        logger.error(f"Processing failed: {e}")
+        return None
+
+# Gradio Interface
+with gr.Blocks(title="PRISM Video Annotator", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🎥 PRISM Site Diary - Video Analyzer")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("## Azure Storage Controls")
+            refresh_btn = gr.Button("🔄 Refresh Video List", variant="secondary")
+            video_dropdown = gr.Dropdown(
+                label="Available Videos",
+                choices=list_azure_videos(),
+                interactive=True
+            )
+            latest_btn = gr.Button("⏩ Process Latest Video", variant="primary")
+            selected_btn = gr.Button("✅ Process Selected Video", variant="primary")
+        
+        with gr.Column(scale=2):
+            gr.Markdown("## Annotated Output")
+            output_video = gr.Video(
+                label="Processed Video",
+                format="mp4",
+                interactive=False
+            )
+            status = gr.Textbox(label="Processing Status")
+
+    def update_ui():
+        new_choices = list_azure_videos()
+        return gr.Dropdown.update(choices=new_choices)
+
+    def handle_latest():
+        latest = get_latest_azure_video()
+        if latest:
+            output = process_video(latest)
+            return output if output else None
+        return None
+
+    # Event handlers
+    refresh_btn.click(
+        fn=update_ui,
+        outputs=video_dropdown,
+        queue=False
+    )
+    
+    latest_btn.click(
+        fn=handle_latest,
+        outputs=output_video,
+        api_name="process_latest"
+    )
+    
+    selected_btn.click(
+        fn=lambda x: process_video(x),
+        inputs=video_dropdown,
+        outputs=output_video,
+        api_name="process_selected"
+    )
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        show_error=True,
+        share=False
+    )