Update app.py

app.py

@@ -1,130 +1,84 @@
 import gradio as gr
 import cv2
-import os
+import numpy as np
 import random
-import time
-
 from services.video_service import get_random_video_frame
 from services.overlay_service import overlay_boxes
 from services.detection_service import detect_objects
 from services.thermal_service import detect_thermal_anomalies
 
+# Constants
 TEMP_IMAGE_PATH = "temp.jpg"
 
-def monitor_feed():
-    frame = get_random_video_frame()
+# Global Variables
+paused = False
+frame_rate = 1  # default frame rate in seconds
 
+# Helper functions
+def monitor_feed():
+    frame, anomaly_type = get_random_video_frame()
     if frame is None:
-        # Safely return if no frame
-        return None, "No frame available", 0, 0
+        return None, "Error loading video frame"
 
-    # Save current frame
     cv2.imwrite(TEMP_IMAGE_PATH, frame)
 
-    # Object detection
     detections = detect_objects(TEMP_IMAGE_PATH)
-
-    # Thermal detection
     thermal_detections = detect_thermal_anomalies(TEMP_IMAGE_PATH)
 
     # Merge all detections
-    all_detections = detections + thermal_detections
-
-    # Overlay bounding boxes
-    result_image = overlay_boxes(TEMP_IMAGE_PATH, all_detections)
-
-    if result_image is not None:
-        cv2.imwrite(TEMP_IMAGE_PATH, result_image)
-    
-    anomaly_count = len([d for d in all_detections if d[4] != "normal"])
-    total_count = len(all_detections)
-
-    return TEMP_IMAGE_PATH, f"Detected {anomaly_count} anomalies", anomaly_count, total_count
-
-
+    combined_detections = detections + thermal_detections
+
+    # Overlay the detections
+    annotated_frame = overlay_boxes(frame.copy(), combined_detections)
+
+    # Update status text
+    status_text = f"Active Anomaly: {anomaly_type} | Objects Detected: {len(detections)} | Thermal Alerts: {len(thermal_detections)}"
+
+    return annotated_frame, status_text
+
+def loop_monitor():
+    global paused
+    while True:
+        if not paused:
+            yield monitor_feed()
+        else:
+            yield None, "Paused"
+        import time
+        time.sleep(frame_rate)
+
+# Pause/Resume control
+def toggle_pause():
+    global paused
+    paused = not paused
+    return "Resume" if paused else "Pause"
+
+# Frame Rate control
+def update_frame_rate(new_rate):
+    global frame_rate
+    frame_rate = new_rate
+    return f"Frame Rate set to {new_rate} sec"
+
+# Build Gradio Interface
 def build_interface():
-    with gr.Blocks() as demo:
+    with gr.Blocks() as app:
         with gr.Row():
-            image_display = gr.Image(label="Solar Drone Monitoring", interactive=False)
-            with gr.Column():
-                anomaly_info = gr.Textbox(label="Anomaly Summary", interactive=False)
-                anomaly_count = gr.Number(label="Anomaly Count", interactive=False)
-                total_objects = gr.Number(label="Total Detections", interactive=False)
-
-        def loop_monitor():
-            while True:
-                time.sleep(1)  # every 1 second
-                yield monitor_feed()
-
-        demo.load(loop_monitor, outputs=[image_display, anomaly_info, anomaly_count, total_objects])
-
-    return demo
-
-if __name__ == "__main__":
-    demo = build_interface()
-    demo.queue()
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=True)
-import gradio as gr
-import cv2
-import os
-import random
-import time
-
-from services.video_service import get_random_video_frame
-from services.overlay_service import overlay_boxes
-from services.detection_service import detect_objects
-from services.thermal_service import detect_thermal_anomalies
-
-TEMP_IMAGE_PATH = "temp.jpg"
-
-def monitor_feed():
-    frame = get_random_video_frame()
-    if frame is None:
-        return None, "No frame available", 0, 0
-    
-    # Save current frame
-    cv2.imwrite(TEMP_IMAGE_PATH, frame)
-
-    # Object detection
-    detections = detect_objects(TEMP_IMAGE_PATH)
-
-    # Thermal detection
-    thermal_detections = detect_thermal_anomalies(TEMP_IMAGE_PATH)
-
-    # Merge all detections
-    all_detections = detections + thermal_detections
-
-    # Overlay bounding boxes
-    result_image = overlay_boxes(TEMP_IMAGE_PATH, all_detections)
-
-    if result_image is not None:
-        cv2.imwrite(TEMP_IMAGE_PATH, result_image)
-    
-    # For metric panel
-    anomaly_count = len([d for d in all_detections if d[4] != "normal"])
-    total_count = len(all_detections)
-
-    return TEMP_IMAGE_PATH, f"Detected {anomaly_count} anomalies", anomaly_count, total_count
-
-def build_interface():
-    with gr.Blocks() as demo:
+            video_output = gr.Image(label="Surveillance Feed", interactive=False)
+            metrics_output = gr.Textbox(label="Live Metrics", interactive=False)
         with gr.Row():
-            image_display = gr.Image(label="Solar Drone Monitoring", interactive=False)
-            with gr.Column():
-                anomaly_info = gr.Textbox(label="Anomaly Summary", interactive=False)
-                anomaly_count = gr.Number(label="Anomaly Count", interactive=False)
-                total_objects = gr.Number(label="Total Detections", interactive=False)
+            pause_button = gr.Button("Pause")
+            frame_rate_slider = gr.Slider(minimum=0.5, maximum=5, value=1, step=0.5, label="Frame Update Interval (seconds)")
+
+        # Live update feed
+        app.load(loop_monitor, outputs=[video_output, metrics_output], every=frame_rate)
 
-        def loop_monitor():
-            while True:
-                time.sleep(1)  # every 1 second
-                yield monitor_feed()
+        # Pause/Resume button logic
+        pause_button.click(toggle_pause, outputs=pause_button)
 
-        demo.load(loop_monitor, outputs=[image_display, anomaly_info, anomaly_count, total_objects])
+        # Frame Rate Slider logic
+        frame_rate_slider.change(update_frame_rate, inputs=frame_rate_slider, outputs=None)
 
-    return demo
+    return app
 
-if __name__ == "__main__":
-    demo = build_interface()
-    demo.queue()
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=True)
+# Launch the app
+demo = build_interface()
+demo.launch()