Update app.py

app.py

@@ -1,78 +1,102 @@
 import os
-import cv2
 import random
-import time
-import gradio as gr
+import cv2
 import numpy as np
-from services.video_service import get_random_video_frame
+import gradio as gr
 from services.overlay_service import overlay_boxes
+from services.video_service import get_random_video_frame
 
-# Constants
-TEMP_IMAGE_PATH = "temp.jpg"
-
-# Global control flags
-is_paused = False
-frame_rate = 1  # Default frame rate (frames per second)
-
-# Live Metrics
-metrics = {
-    "total_frames": 0,
-    "detected_objects": 0,
-}
-
-def monitor_feed():
-    global is_paused, frame_rate, metrics
-
-    while True:
-        if not is_paused:
-            frame, anomaly_label = get_random_video_frame()
-
-            if frame is None:
-                time.sleep(1)
-                continue
-
-            # Overlay anomaly text
-            frame = overlay_boxes(frame, anomaly_label)
-
-            # Save temporary image
-            cv2.imwrite(TEMP_IMAGE_PATH, frame)
+# Global control variables
+paused = False
+frame_rate = 1.0  # seconds between frames
 
-            metrics["total_frames"] += 1
-            metrics["detected_objects"] += random.randint(1, 3)  # Simulated detection count
+# Temporary image path for current frame
+TEMP_IMAGE_PATH = "temp_frame.jpg"
 
-            yield TEMP_IMAGE_PATH, f"Frames: {metrics['total_frames']}, Detections: {metrics['detected_objects']}"
+# Metrics
+total_detections = 0
+anomaly_counter = {}
 
-        time.sleep(1 / max(frame_rate, 1))
+# Define list of possible anomaly labels for simulation
+possible_labels = ["Panel Crack", "Dust Accumulation", "Human Intrusion", "Overheating"]
 
-def toggle_pause():
-    global is_paused
-    is_paused = not is_paused
-    return "Paused" if is_paused else "Running"
 
-def set_frame_rate(new_rate):
+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():
+    global paused
+    paused = False
+
+def update_frame_rate(new_rate):
     global frame_rate
     frame_rate = new_rate
-    return f"Frame Rate: {frame_rate} fps"
+
 
 def build_interface():
-    with gr.Blocks() as demo:
-        gr.Markdown("# Solar Surveillance POC")
-        
+    with gr.Blocks() as app:
+        gr.Markdown("# Solar Surveillance - Live Monitor \ud83d\ude80")
+
         with gr.Row():
-            video_output = gr.Image(label="Live Feed")
-            metrics_output = gr.Label(label="Live Metrics")
+            video_output = gr.Image(label="Live Feed", show_label=True)
+            metrics_output = gr.Textbox(label="Live Metrics", interactive=False)
 
         with gr.Row():
-            pause_button = gr.Button("Pause / Resume")
-            frame_rate_slider = gr.Slider(1, 10, value=1, label="Frame Rate (frames/sec)")
+            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)")
+
+        pause_btn.click(fn=pause_monitor)
+        resume_btn.click(fn=resume_monitor)
+        frame_rate_slider.change(fn=update_frame_rate, inputs=frame_rate_slider)
+
+        def loop_monitor():
+            while True:
+                yield monitor_feed()
+
+        app.live(loop_monitor, outputs=[video_output, metrics_output], interval=lambda: frame_rate)
 
-        pause_button.click(toggle_pause, outputs=[])
-        frame_rate_slider.change(set_frame_rate, inputs=frame_rate_slider, outputs=[])
+    return app
 
-        demo.load(monitor_feed, outputs=[video_output, metrics_output], every=None)
 
-    return demo
+demo = build_interface()
 
-if __name__ == "__main__":
-    demo = build_interface()
-    demo.launch(share=True)  # Important for Hugging Face Spaces!
+demo.launch(server_name="0.0.0.0", server_port=7860)