Update app.py

app.py

@@ -20,6 +20,8 @@ import numpy as np
 import pathlib
 import zipfile
 import shutil
+import pandas as pd
+import subprocess
 from ultralytics import YOLO
 from autogluon.tabular import TabularPredictor
 from huggingface_hub import hf_hub_download
@@ -61,9 +63,11 @@ def ag_predictor():
             shutil.rmtree(extract_dir)
         with zipfile.ZipFile(z, "r") as zip_ref:
             zip_ref.extractall(extract_dir)
-        _ag_predictor = TabularPredictor.load(str(extract_dir),
-                                              require_version_match=False,
-                                              require_py_version_match=False)
+        _ag_predictor = TabularPredictor.load(
+            str(extract_dir),
+            require_version_match=False,
+            require_py_version_match=False
+        )
     return _ag_predictor
 
 # -------------------
@@ -117,6 +121,82 @@ def isolate_scoreboard_color(frame_bgr: np.ndarray,
 
     return gray
 
+# -------------------
+# Event picking
+# -------------------
+def pick_events(df: pd.DataFrame, score: pd.Series, fps: float) -> list:
+    # simple hybrid threshold (as tuned earlier)
+    max_score = score.max()
+    raw_cutoff = 0.7 * max_score if max_score > 0 else 0.4
+    z = (score - score.rolling(45, min_periods=1).mean()) / (score.rolling(45, min_periods=1).std()+1e-9)
+    z_cutoff = max(2.0, 0.6 * z.max())
+
+    out_times = []
+    for i in range(1, len(score)-1):
+        ts = float(df.iloc[i]["timestamp"])
+        if ((score.iloc[i] > raw_cutoff) or (z.iloc[i] > z_cutoff)):
+            if score.iloc[i] > score.iloc[i-1] and score.iloc[i] > score.iloc[i+1]:
+                if ts >= 1.0:  # guard against first second
+                    out_times.append(ts)
+
+    grouped = []
+    for t in out_times:
+        if (not grouped) or (t - grouped[-1]) > GROUP_GAP_S:
+            grouped.append(t)
+    return grouped
+
+# -------------------
+# Video clipping
+# -------------------
+def cut_clip(video_path, start, end, out_path):
+    cmd = [
+        "ffmpeg", "-y", "-i", str(video_path),
+        "-ss", str(start), "-to", str(end),
+        "-c", "copy", str(out_path)
+    ]
+    subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+def extract_score_clips(video_path: str, debug: bool = False):
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        return [], "❌ Could not open video."
+
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    frames = []
+    timestamps = []
+    idx = 0
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        ts = idx / fps
+        masked = isolate_scoreboard_color(frame, debug=debug, frame_id=idx if debug else None)
+        red_ratio = float((masked[:,:,2] > 150).mean())  # crude feature
+        frames.append([ts, red_ratio])
+        timestamps.append(ts)
+        idx += 1
+    cap.release()
+
+    if not frames:
+        return [], "⚠️ No frames processed."
+
+    df = pd.DataFrame(frames, columns=["timestamp","red_ratio"])
+    pred = ag_predictor().predict_proba(df[["red_ratio"]])
+    score = pd.Series(pred[1].values, index=df.index)
+
+    events = pick_events(df, score, fps)
+    if not events:
+        return [], "⚠️ No touches confidently detected in this video."
+
+    clips = []
+    for i, t in enumerate(events, 1):
+        s = max(0.0, t - CLIP_PAD_S)
+        e = min(df["timestamp"].max(), t + CLIP_PAD_S)
+        out_path = f"clip_{i}.mp4"
+        cut_clip(video_path, s, e, out_path)
+        clips.append((out_path, f"Touch at {t:.2f}s"))
+    return clips, f"✅ Detected {len(events)} touches."
+
 # -------------------
 # Progress bar builder
 # -------------------
@@ -130,51 +210,37 @@ def _make_progress_bar(percent: int, final_text: str = None):
     """
 
 # -------------------
-# Main pipeline
+# Wrapped run (step-based)
 # -------------------
-def extract_score_clips(video_path: str, debug: bool = False):
-    print("[INFO] Starting frame extraction...")
-    # Placeholder for real processing code
-    # This would extract frames, isolate scoreboard, run predictor, detect events, cut clips
-    # For now we mock it as a "success" to show UI
-    return [], "⚠️ No touches confidently detected in this video."
-
 def wrapped_run(video_file):
     if not video_file:
         yield gr.update(value=[], visible=False), "Please upload a video file.", gr.update(value="", visible=False)
         return
 
-    # Fake step progress (jumping updates)
     yield gr.update(value=[], visible=False), "Processing started...", gr.update(value=_make_progress_bar(10), visible=True)
     yield gr.update(value=[], visible=False), "Extracting frames...", gr.update(value=_make_progress_bar(40), visible=True)
     yield gr.update(value=[], visible=False), "Running predictor...", gr.update(value=_make_progress_bar(70), visible=True)
 
     clips, status_msg = extract_score_clips(video_file, debug=False)
-
     final_bar = _make_progress_bar(100, "✅ Done")
     yield gr.update(value=clips, visible=bool(clips)), status_msg, gr.update(value=final_bar, visible=True)
 
 # -------------------
-# Build Gradio UI
+# Gradio UI
 # -------------------
 with gr.Blocks() as demo:
     gr.Markdown("## 🤺 Fencing Score Detector\nUpload a bout video and detect touches.")
 
     in_video = gr.Video(label="Upload Bout Video", type="filepath")
-
     run_btn = gr.Button("Detect Touches", elem_id="detect-btn")
 
     status = gr.Markdown("Status messages will appear here.")
     progress_html = gr.HTML("")
     gallery = gr.Gallery(label="Detected Clips", visible=False)
 
-    run_btn.click(
-        fn=wrapped_run,
-        inputs=in_video,
-        outputs=[gallery, status, progress_html],
-    )
+    run_btn.click(fn=wrapped_run, inputs=in_video, outputs=[gallery, status, progress_html])
 
-# Launch with queue
 if __name__ == "__main__":
     demo.queue(max_size=20)
     demo.launch(debug=True)
+