Update app.py

app.py

@@ -1,102 +1,65 @@
+import gradio as gr
+import cv2
 import os
 import random
-import cv2
-import numpy as np
-import gradio as gr
-from services.overlay_service import overlay_boxes
+import time
 from services.video_service import get_random_video_frame
+from services.metrics_service import update_metrics
+from services.overlay_service import overlay_boxes
 
-# Global control variables
-paused = False
-frame_rate = 1.0  # seconds between frames
-
-# Temporary image path for current frame
+# Constants
 TEMP_IMAGE_PATH = "temp_frame.jpg"
+VIDEO_FOLDER = "data/sample_videos"
 
-# Metrics
-total_detections = 0
-anomaly_counter = {}
-
-# Define list of possible anomaly labels for simulation
-possible_labels = ["Panel Crack", "Dust Accumulation", "Human Intrusion", "Overheating"]
-
+# State Variables
+paused = False
+frame_rate = 1  # Default 1 FPS
 
+# Helper Functions
 def monitor_feed():
-    global total_detections, anomaly_counter
-    if paused:
-        frame = np.zeros((480, 640, 3), dtype=np.uint8)
-        cv2.putText(frame, "PAUSED", (200, 250), cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 0, 255), 5)
-    else:
-        frame = get_random_video_frame()
-        if frame is None:
-            frame = np.zeros((480, 640, 3), dtype=np.uint8)
-        else:
-            # Simulate random detections
-            detections = []
-            for _ in range(random.randint(1, 5)):
-                label = random.choice(possible_labels)
-                x1, y1 = random.randint(0, frame.shape[1]//2), random.randint(0, frame.shape[0]//2)
-                x2, y2 = x1 + random.randint(50, 150), y1 + random.randint(50, 150)
-                detections.append({"box": [x1, y1, x2, y2], "label": label})
-
-            frame = overlay_boxes(frame, detections)
-
-            # Update metrics
-            total_detections += len(detections)
-            for det in detections:
-                anomaly_counter[det["label"]] = anomaly_counter.get(det["label"], 0) + 1
-
-    cv2.imwrite(TEMP_IMAGE_PATH, frame)
-    return TEMP_IMAGE_PATH, generate_metrics_text()
-
-
-def generate_metrics_text():
-    metrics_text = f"Total Detections: {total_detections}\n"
-    for label, count in anomaly_counter.items():
-        metrics_text += f"{label}: {count}\n"
-    return metrics_text
-
-
-def pause_monitor():
     global paused
-    paused = True
-
-def resume_monitor():
+    frame = get_random_video_frame()
+    if frame is None:
+        return None, update_metrics()
+    if not paused:
+        frame = overlay_boxes(frame)
+        cv2.imwrite(TEMP_IMAGE_PATH, frame)
+    return TEMP_IMAGE_PATH, update_metrics()
+
+def toggle_pause():
     global paused
-    paused = False
+    paused = not paused
 
-def update_frame_rate(new_rate):
+def set_frame_rate(value):
     global frame_rate
-    frame_rate = new_rate
+    frame_rate = int(value)
 
+def loop_monitor():
+    while True:
+        time.sleep(1 / frame_rate if frame_rate > 0 else 1)
+        yield monitor_feed()
 
+# Gradio App
 def build_interface():
     with gr.Blocks() as app:
-        gr.Markdown("# Solar Surveillance - Live Monitor \ud83d\ude80")
-
-        with gr.Row():
-            video_output = gr.Image(label="Live Feed", show_label=True)
-            metrics_output = gr.Textbox(label="Live Metrics", interactive=False)
-
         with gr.Row():
             with gr.Column():
-                pause_btn = gr.Button("Pause")
-                resume_btn = gr.Button("Resume")
-                frame_rate_slider = gr.Slider(0.1, 5.0, value=1.0, label="Frame Rate (seconds)")
+                video_output = gr.Image(label="Live Drone Feed")
+            with gr.Column():
+                metrics_output = gr.Label(label="Live Metrics")
 
-        pause_btn.click(fn=pause_monitor)
-        resume_btn.click(fn=resume_monitor)
-        frame_rate_slider.change(fn=update_frame_rate, inputs=frame_rate_slider)
+        with gr.Row():
+            pause_button = gr.Button("Pause / Resume")
+            frame_rate_slider = gr.Slider(1, 10, value=1, step=1, label="Frame Rate (Frames per second)")
 
-        def loop_monitor():
-            while True:
-                yield monitor_feed()
+        pause_button.click(toggle_pause)
+        frame_rate_slider.change(set_frame_rate, inputs=[frame_rate_slider])
 
-        app.live(loop_monitor, outputs=[video_output, metrics_output], interval=lambda: frame_rate)
+        app.load(fn=loop_monitor, outputs=[video_output, metrics_output])
 
     return app
 
-
-demo = build_interface()
-
-demo.launch(server_name="0.0.0.0", server_port=7860)
+# Launch
+if __name__ == "__main__":
+    demo = build_interface()
+    demo.queue().launch()