Update app.py

app.py

@@ -131,82 +131,40 @@ def process_live_frame(frame):
         return processed, status_txt, None
 
 
-
-# Constants for the video experiment
-VIDEO_FPS = 30.0
-CHUNK_SIZE_SECONDS = 2
-CHUNK_FRAME_COUNT = int(VIDEO_FPS * CHUNK_SIZE_SECONDS)
-TEMP_VIDEO_FILE = "temp_video_chunk.mp4"
-
-def process_video_chunk(frame, state_dict):
+# ───────────────────────────── NEW: Frame Processing for Tab 2 (Analysis-Only)
+def process_for_stats_only(frame):
+    """
+    Processes a frame but does not return any video/image output.
+    This is the fastest method, focused only on status and alerts.
+    """
     if frame is None:
-        return None, "Status: Inactive", None, state_dict
-
-    # Unpack the state
-    frame_buffer = state_dict['buffer']
-    start_time = state_dict['start_time']
-    last_buffer_fill_time = state_dict['last_buffer_fill_time']
-    last_encoding_time = state_dict['last_encoding_time']
-
-    # If the buffer is empty, this is the first frame of a new chunk. Record start time.
-    if not frame_buffer:
-        start_time = time.perf_counter()
+        return "Status: Inactive", None
 
-    # --- Real-time detection and alerting (This is not delayed) ---
+    t0 = time.perf_counter()
     try:
-        processed_frame, indic = detector.process_frame(frame)
+        # We still call the same detector, but we will ignore the processed frame it returns.
+        _, indic = detector.process_frame(frame)
     except Exception as e:
         logging.error(f"Error processing frame: {e}")
-        processed_frame = np.zeros_like(frame)
         indic = {"drowsiness_level": "Error", "lighting": "Unknown", "details": {"Score": 0.0}}
 
     level = indic.get("drowsiness_level", "Awake")
     lighting = indic.get("lighting", "Good")
     score = indic.get("details", {}).get("Score", 0.0)
-    audio_payload = alert_manager.trigger_alert(level, lighting)
-    audio_out = gr.Audio(value=audio_payload, autoplay=True) if audio_payload else None
 
-    # --- Video Buffering Logic ---
-    frame_buffer.append(processed_frame)
-
-    video_out = None
-    if len(frame_buffer) >= CHUNK_FRAME_COUNT:
-        buffer_fill_time = time.perf_counter() - start_time
-        logging.info(f"BUFFER: Filled {len(frame_buffer)} frames in {buffer_fill_time:.2f} seconds.")
-        
-        encoding_start_time = time.perf_counter()
-        h, w, _ = frame_buffer[0].shape
-        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-        writer = cv2.VideoWriter(TEMP_VIDEO_FILE, fourcc, VIDEO_FPS, (w, h))
-        for f in frame_buffer:
-            writer.write(f)
-        writer.release()
-        encoding_time = time.perf_counter() - encoding_start_time
-        logging.info(f"ENCODING: Video chunk encoded in {encoding_time:.2f} seconds.")
-        
-        video_out = TEMP_VIDEO_FILE
-        last_buffer_fill_time = buffer_fill_time
-        last_encoding_time = encoding_time
-        frame_buffer = []
-        start_time = None
+    dt_ms = (time.perf_counter() - t0) * 1000.0
+    logging.info(f"ANALYSIS ONLY │ {dt_ms:6.1f} ms │ {lighting:<4} │ {level:<14} │ score={score:.2f}")
 
-    # --- NEW: Update status text to include buffer/encoding times ---
     status_txt = (
         f"Status: {level} (Score: {score:.2f})\n"
-        f"Buffer Fill Time: {last_buffer_fill_time:.2f}s\n"
-        f"Video Encode Time: {last_encoding_time:.2f}s"
+        f"Lighting: {lighting}\n"
+        f"Processing Time: {dt_ms:.1f} ms"
     )
-    logging.info(f"Status: {status_txt}")
-
-    # Pack state back up to return
-    new_state = {
-        'buffer': frame_buffer,
-        'start_time': start_time,
-        'last_buffer_fill_time': last_buffer_fill_time,
-        'last_encoding_time': last_encoding_time
-    }
-    
-    return video_out, status_txt, audio_out, new_state
+
+    audio_payload = alert_manager.trigger_alert(level, lighting)
+    audio_out = gr.Audio(value=audio_payload, autoplay=True) if audio_payload else None
+
+    return status_txt, audio_out
 
     
 # ───────────────────────────── UI Definition
@@ -287,46 +245,33 @@ def create_detection_tab():
             inputs=[cam],
             outputs=[out_img, out_text, out_audio] # The output now targets the placeholder
         )
-
-def create_video_experiment_tab():
-    """Creates the content for the Video Chunk experiment tab - UPDATED."""
-    with gr.Blocks() as video_tab:
-        gr.Markdown("## 🧪 Video Output Experiment")
-        gr.Markdown(f"This feed buffers processed frames and outputs them as **{CHUNK_SIZE_SECONDS}-second video chunks**. Notice the trade-off between smoothness and latency. Alerts remain real-time.")
-        with gr.Row():
-            with gr.Column(scale=2):
-                cam_video = gr.Image(sources=["webcam"], streaming=True, label="Live Camera Feed")
-            with gr.Column(scale=1):
-                # --- NEW: Use the blank video as the initial value ---
-                out_video = gr.Video(label="Processed Video Chunk", value=BLANK_VIDEO_FILE)
-                # --- NEW: Textbox is now larger to show more info ---
-                out_text_video = gr.Textbox(label="Live Status", lines=4, interactive=False)
-                out_audio_video = gr.Audio(label="Alert", autoplay=True, visible=False)
-
-        # --- NEW: State is a dictionary to hold the buffer and timing info ---
-        initial_state = {
-            'buffer': [], 
-            'start_time': None, 
-            'last_buffer_fill_time': 0.0,
-            'last_encoding_time': 0.0
-        }
-        state = gr.State(initial_state)
         
-        cam_video.stream(
-            fn=process_video_chunk,
-            inputs=[cam_video, state],
-            outputs=[out_video, out_text_video, out_audio_video, state]
-        )
-    return video_tab
-
+def create_analysis_only_tab():
+    """Creates the content for the Analysis-Only Mode tab."""
+    gr.Markdown("## ⚡ Analysis-Only Mode")
+    gr.Markdown("This mode provides the fastest possible analysis by not sending any video back to the browser. The camera is still active for detection, but you will only see the live status and hear alerts.")
+    with gr.Row():
+        with gr.Column(scale=1):
+            # The input camera is visible so the user knows it's working,
+            # but there is no corresponding video output component.
+            cam_analysis = gr.Image(sources=["webcam"], streaming=True, label="Live Camera Feed (for detection)")
+        with gr.Column(scale=1):
+            out_text_analysis = gr.Textbox(label="Live Status & Performance", lines=4, interactive=False)
+            out_audio_analysis = gr.Audio(label="Alert", autoplay=True, visible=False)
+    
+    cam_analysis.stream(
+        fn=process_for_stats_only,
+        inputs=[cam_analysis],
+        outputs=[out_text_analysis, out_audio_analysis]
+    )
 
 with gr.Blocks(title="🚗 Drive Paddy – Drowsiness Detection", theme=gr.themes.Soft()) as app:
     gr.Markdown("# 🚗 **Drive Paddy**")
     with gr.Tabs():
         with gr.TabItem("Live Detection"):
             create_detection_tab()
-        with gr.TabItem("Video Output Experiment"):
-            create_video_experiment_tab()
+        with gr.TabItem("Analysis-Only Mode"):
+            create_analysis_only_tab()
         with gr.TabItem("About this App"):
             create_readme_tab()