Tuned for Angle-Based Flows

app.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🚦 Stage 3 — Wrong Direction Detection (Stable + Confidence + Hysteresis + Filter)
+# 🚦 Stage 3 — Wrong Direction Detection (Tuned + Confidence + Stability)
 # ============================================================
 
 import os, cv2, json, tempfile, numpy as np, gradio as gr
@@ -46,7 +46,7 @@ class Track:
         self.frames_seen += 1
         return [x, y]
 
-    def stable_status(self, new_status, new_conf, window=10, agree_ratio=0.6):
+    def stable_status(self, new_status, new_conf, window=15, agree_ratio=0.7):
         """Debounce flicker using recent window consensus."""
         self.status_history.append(new_status)
         if len(self.status_history) > window:
@@ -75,6 +75,7 @@ def smooth_direction(points, window=5):
         return None
     return avg_vec
 
+
 # ============================================================
 # 🧭 Wrong-Direction Detection Core
 # ============================================================
@@ -92,10 +93,10 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
 
     tracks, next_id = {}, 0
     DELAY_FRAMES = 8
-    MIN_FLOW_SPEED = 1.2
-    HYST_OK = 0.55
-    HYST_WRONG = 0.45
-    ALPHA = 0.6               # exponential smoothing weight
+    MIN_FLOW_SPEED = 1.8
+    HYST_OK = 0.65
+    HYST_WRONG = 0.35
+    ALPHA = 0.75
 
     while True:
         ret, frame = cap.read()
@@ -112,7 +113,7 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
                 dets.append([cx, cy])
         dets = np.array(dets)
 
-        # --- Tracker update ---
+        # --- Tracker association ---
         assigned = set()
         if len(dets) > 0 and len(tracks) > 0:
             existing = np.array([t.kf.x[:2].reshape(-1) for t in tracks.values()])
@@ -123,6 +124,7 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
                     tid = list(tracks.keys())[r]
                     tracks[tid].update(dets[c])
                     assigned.add(c)
+
         for i, d in enumerate(dets):
             if i not in assigned:
                 tracks[next_id] = Track(d, next_id)
@@ -134,32 +136,29 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
             pts = np.array(trk.history)
             if len(pts) > 1:
                 for i in range(1, len(pts)):
-                    cv2.line(frame, tuple(np.int32(pts[i-1])), tuple(np.int32(pts[i])), (0, 0, 255), 1)
+                    cv2.line(frame, tuple(np.int32(pts[i-1])), tuple(np.int32(pts[i])), (100, 100, 255), 1)
 
             motion = smooth_direction(pts)
-            if motion is None:
-                continue
-            if np.linalg.norm(motion) < MIN_FLOW_SPEED:
+            if motion is None or np.linalg.norm(motion) < MIN_FLOW_SPEED:
                 continue
 
             sims = [compute_cosine_similarity(motion, f) for f in lane_flows]
             best_sim = max(sims)
 
             if trk.frames_seen > DELAY_FRAMES:
-                # Exponential moving average
                 trk.ema_sim = ALPHA * best_sim + (1 - ALPHA) * getattr(trk, "ema_sim", best_sim)
 
-                # Hysteresis classification
+                # Hysteresis decision
                 if trk.ema_sim >= HYST_OK:
                     new_status = "OK"
                 elif trk.ema_sim <= HYST_WRONG:
                     new_status = "WRONG"
                 else:
-                    new_status = trk.status  # hold previous label
+                    new_status = trk.status
 
-                trk.stable_status(new_status, new_conf=trk.ema_sim, window=10, agree_ratio=0.6)
+                trk.stable_status(new_status, new_conf=trk.ema_sim, window=15, agree_ratio=0.7)
 
-                # --- Filter by UI controls ---
+                # --- Confidence-based label filtering ---
                 show_label = True
                 if trk.confidence < conf_threshold:
                     show_label = False
@@ -167,7 +166,13 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
                     show_label = False
 
                 if show_label:
-                    color = (0, 0, 255) if trk.status == "WRONG" else (0, 255, 0)
+                    if trk.status == "WRONG":
+                        color = (0, 0, 255)
+                    elif trk.confidence < 0.5:
+                        color = (0, 255, 255)
+                    else:
+                        color = (0, 255, 0)
+
                     label = f"ID:{tid} {trk.status} ({trk.confidence:.2f})"
                     cv2.putText(frame, label, tuple(np.int32(pos)),
                                 cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
@@ -178,16 +183,16 @@ def process_video(video_file, stage2_json, show_only_wrong=False, conf_threshold
     out.release()
     return out_path
 
+
 # ============================================================
 # 🎛️ Gradio Interface
 # ============================================================
 description = """
-### 🚦 Stage 3 — Wrong Direction Detection (Stable + Confidence + Filter)
-- ✅ Cosine similarity with exponential smoothing  
-- ✅ Hysteresis (OK≥0.55 / WRONG≤0.45) for stability  
-- ✅ 10-frame consensus voting (flicker-free)  
-- ✅ Confidence-based label filtering  
-- ✅ “Show Only Wrong” toggle
+### 🚦 Stage 3 — Wrong Direction Detection (Tuned for Angle-Based Flows)
+- ✅ Hysteresis: OK ≥ 0.65 / WRONG ≤ 0.35  
+- ✅ EMA smoothing α = 0.75  
+- ✅ 15-frame consensus for flicker-free labeling  
+- ✅ Confidence color coding (Green→OK, Yellow→Borderline, Red→Wrong)
 """
 
 demo = gr.Interface(
@@ -195,11 +200,11 @@ demo = gr.Interface(
     inputs=[
         gr.File(label="Input Video"),
         gr.File(label="Stage 2 Flow JSON"),
-        gr.Checkbox(label="Show ONLY Wrong Labels Overlay", value=False),
-        gr.Slider(0.0, 1.0, value=0.0, step=0.05, label="Confidence Level Filter (Show ≥ this value)")
+        gr.Checkbox(label="Show ONLY Wrong Labels", value=False),
+        gr.Slider(0.0, 1.0, value=0.0, step=0.05, label="Confidence Filter (Show ≥ this value)")
     ],
     outputs=gr.Video(label="Output Video"),
-    title="🚗 Stage 3 – Stable Wrong-Direction Detection (with Confidence Filter)",
+    title="🚗 Stage 3 – Tuned Wrong-Direction Detection (Confidence + Stability)",
     description=description
 )