Update app.py

app.py

@@ -17,6 +17,9 @@
 # -*- coding: utf-8 -*-
 # Fencing Scoreboard Clips - YOLO x AutoGluon (Gradio)
 
+# -*- coding: utf-8 -*-
+# Fencing Scoreboard Clips - YOLO x AutoGluon (Gradio)
+
 import os, sys, zipfile, shutil, subprocess, tempfile, pathlib
 from typing import List, Tuple
 import uuid
@@ -27,13 +30,15 @@ import cv2
 import gradio as gr
 
 # ----------------
-# Flags
+# Debug control
 # ----------------
-DEBUG_SAVE_FRAMES = False   # disable debug frames by default
+DEBUG_SAVE_FRAMES = False   # disable debug images by default
+
+DEBUG_DIR = pathlib.Path("debug_frames")
+if DEBUG_DIR.exists():
+    shutil.rmtree(DEBUG_DIR)   # wipe any leftover frames on startup
+DEBUG_DIR.mkdir(exist_ok=True)
 
-# ----------------
-# Utility
-# ----------------
 def _pip(pkgs: List[str]):
     import subprocess, sys
     subprocess.check_call([sys.executable, "-m", "pip", "install", "--quiet", *pkgs])
@@ -86,11 +91,6 @@ CLIP_PAD_S     = float(os.getenv("CLIP_PAD_S","2.0"))
 CACHE_DIR = pathlib.Path("hf_assets")
 CACHE_DIR.mkdir(parents=True, exist_ok=True)
 
-DEBUG_DIR = pathlib.Path("debug_frames")
-if DEBUG_DIR.exists():
-    shutil.rmtree(DEBUG_DIR)   # wipe old debug frames at startup
-DEBUG_DIR.mkdir(exist_ok=True)
-
 # ----------------
 # Model loaders
 # ----------------
@@ -182,9 +182,143 @@ def isolate_scoreboard_color(frame_bgr: np.ndarray,
 
     return gray
 
+def color_pixel_ratio(rgb: np.ndarray, ch: int) -> float:
+    R, G, B = rgb[:, :, 0], rgb[:, :, 1], rgb[:, :, 2]
+    if ch == 0:
+        mask = (R > 150) & (R > 1.2*G) & (R > 1.2*B)
+    else:
+        mask = (G > 100) & (G > 1.05*R) & (G > 1.05*B)
+    return np.sum(mask) / (rgb.shape[0]*rgb.shape[1] + 1e-9)
+
+def rolling_z(series: pd.Series, win: int = 45) -> pd.Series:
+    med = series.rolling(win, min_periods=5).median()
+    mad = series.rolling(win, min_periods=5).apply(
+        lambda x: np.median(np.abs(x - np.median(x))), raw=True
+    )
+    mad = mad.replace(0, mad[mad > 0].min() if (mad > 0).any() else 1.0)
+    return (series - med) / mad
+
+# ----------------------------
+# Video → features
+# ----------------------------
+def extract_feature_timeseries(video_path: str,
+                               frame_skip: int = FRAME_SKIP,
+                               debug: bool = False) -> Tuple[pd.DataFrame, float]:
+    print("[INFO] Starting frame extraction...")
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        return pd.DataFrame(), 0.0
+    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+    records, frame_idx = [], 0
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        if frame_idx % frame_skip == 0:
+            ts = frame_idx / fps
+            masked = isolate_scoreboard_color(frame, debug=debug, frame_id=frame_idx)
+            rgb = cv2.cvtColor(masked, cv2.COLOR_BGR2RGB)
+            red_ratio   = color_pixel_ratio(rgb, 0)
+            green_ratio = color_pixel_ratio(rgb, 1)
+            records.append({
+                "frame_id": frame_idx,
+                "timestamp": ts,
+                "red_ratio": red_ratio,
+                "green_ratio": green_ratio,
+            })
+        frame_idx += 1
+
+    cap.release()
+    df = pd.DataFrame(records)
+    print(f"[INFO] Processed {len(df)} frames out of {total_frames} (fps={fps:.2f})")
+
+    if df.empty:
+        return df, fps
+
+    df["red_diff"]   = df["red_ratio"].diff().fillna(0)
+    df["green_diff"] = df["green_ratio"].diff().fillna(0)
+    df["z_red"]      = rolling_z(df["red_ratio"])
+    df["z_green"]    = rolling_z(df["green_ratio"])
+
+    if DEBUG_SAVE_FRAMES and debug:
+        out_csv = DEBUG_DIR / f"features_{uuid.uuid4().hex}.csv"
+        df.to_csv(out_csv, index=False)
+
+    return df, fps
+
+# ----------------------------
+# Predictor + event picking
+# ----------------------------
+def predict_scores(df: pd.DataFrame) -> pd.Series:
+    feat_cols = ["red_ratio", "green_ratio", "red_diff", "green_diff", "z_red", "z_green"]
+    X = df[feat_cols].copy()
+    pred = ag_predictor().predict(X)
+    try:
+        proba = ag_predictor().predict_proba(X)
+        if isinstance(proba, pd.DataFrame) and (1 in proba.columns):
+            return proba[1]
+    except Exception:
+        pass
+    s = pd.Series(pred).astype(float)
+    rng = (s.quantile(0.95) - s.quantile(0.05)) or 1.0
+    return ((s - s.quantile(0.05)) / rng).clip(0, 1)
+
+def pick_events(df: pd.DataFrame, score: pd.Series, fps: float,
+                min_start_guard_s: float = 1.0,
+                guard_enable_min_duration_s: float = 6.0) -> List[float]:
+    max_score = score.max()
+    raw_cutoff = 0.7 * max_score if max_score > 0 else 0.4
+    z = rolling_z(score, win=45)
+    max_z = z.max()
+    z_cutoff = max(2.0, 0.6 * max_z)
+
+    print(f"[DEBUG] Predictor score stats: min={score.min():.3f}, max={max_score:.3f}, mean={score.mean():.3f}")
+    print(f"[DEBUG] Adaptive thresholds: raw>{raw_cutoff:.3f}, z>{z_cutoff:.2f}")
+
+    duration_est = float(df["timestamp"].max()) if not df.empty else 0.0
+    enforce_guard = duration_est >= guard_enable_min_duration_s
+    out_times = []
+    min_dist_frames = max(1, int(1.0 * max(1.0, fps)))
+    y = score.values
+    last_kept = -min_dist_frames
+
+    for i in range(1, len(y)-1):
+        ts = float(df.iloc[i]["timestamp"])
+        local_peak = y[i] > y[i-1] and y[i] > y[i+1]
+        if ((z.iloc[i] > z_cutoff) or (y[i] > raw_cutoff)) and local_peak and (i - last_kept) >= min_dist_frames:
+            if (not enforce_guard) or (ts >= min_start_guard_s):
+                out_times.append(ts)
+                last_kept = i
+
+    if not out_times and len(y) > 0:
+        best_idx = int(np.argmax(y))
+        ts_best = float(df.iloc[best_idx]["timestamp"])
+        if (not enforce_guard) or (ts_best >= min_start_guard_s):
+            out_times = [ts_best]
+            print(f"[DEBUG] Fallback → using global max at {ts_best:.2f}s")
+
+    out_times.sort()
+    grouped = []
+    for t in out_times:
+        if (not grouped) or (t - grouped[-1]) > GROUP_GAP_S:
+            grouped.append(t)
+    print(f"[DEBUG] Final detected events: {grouped}")
+    return grouped
+
 # ----------------------------
 # Clip helpers
 # ----------------------------
+def _probe_duration(video_path: str) -> float:
+    try:
+        if ffmpeg is None:
+            raise RuntimeError("ffmpeg-python not available")
+        meta = ffmpeg.probe(video_path)
+        return float(meta["format"]["duration"])
+    except:
+        return 0.0
+
 def cut_clip(video_path: str, start: float, end: float, out_path: str) -> str:
     try:
         cmd = ["ffmpeg", "-y", "-ss", str(max(0, start)), "-to", str(max(start, end)),
@@ -200,18 +334,18 @@ def cut_clip(video_path: str, start: float, end: float, out_path: str) -> str:
     return out_path
 
 # ----------------------------
-# Orchestrator (with cleanup)
+# Orchestrator
 # ----------------------------
-def extract_score_clips(video_path: str, debug: bool = False):
+def extract_score_clips(video_path: str, debug: bool = False) -> Tuple[List[Tuple[str, str]], str]:
     print("[INFO] Running full detection pipeline...")
-    from moviepy.editor import VideoFileClip
     df, fps = extract_feature_timeseries(video_path, frame_skip=FRAME_SKIP, debug=debug)
     if df.empty:
         return [], "No frames processed."
 
     score = predict_scores(df)
     if score.max() <= 1e-6:
-        return [], "⚠️ No scoreboard detected or illumination scores flat."
+        print("[WARN] Flat scores from predictor (possible YOLO miss or feature mismatch).")
+        return [], "⚠️ No scoreboard detected or illumination scores flat. Please check video or model."
 
     events = pick_events(df, score, fps)
     if not events:
@@ -221,10 +355,11 @@ def extract_score_clips(video_path: str, debug: bool = False):
     if duration <= 0:
         duration = float(df["timestamp"].max() + CLIP_PAD_S + 0.5)
 
-    clips, kept_paths = [], []
+    clips = []
     base = os.path.splitext(os.path.basename(video_path))[0]
     for i, t in enumerate(events):
-        s, e = t - CLIP_PAD_S, t + CLIP_PAD_S
+        s = t - CLIP_PAD_S
+        e = t + CLIP_PAD_S
         if s < 0:
             e = min(duration, e - s)
             s = 0
@@ -233,14 +368,118 @@ def extract_score_clips(video_path: str, debug: bool = False):
             e = duration
         clip_path = os.path.join(tempfile.gettempdir(), f"{base}_score_{i+1:02d}.mp4")
         cut_clip(video_path, s, e, clip_path)
-        clips.append((clip_path, f"Touch {i+1} @ {t:.2f}s"))
-        kept_paths.append(clip_path)
+        label = f"Touch {i+1} @ {t:.2f}s"
+        clips.append((clip_path, label))
 
-    # cleanup: delete unused temp clips
+    # cleanup: keep only returned clips
+    keep = {c[0] for c in clips}
     for f in pathlib.Path(tempfile.gettempdir()).glob(f"{base}_score_*.mp4"):
-        if str(f) not in kept_paths:
-            try: f.unlink()
-            except: pass
+        if str(f) not in keep:
+            try:
+                f.unlink()
+            except:
+                pass
 
     return clips, f"✅ Detected {len(clips)} event(s)."
 
+# ----------------------------
+# Gradio UI
+# ----------------------------
+CSS = """
+.gradio-container {max-width: 900px; margin: auto;}
+.header {text-align: center; margin-bottom: 20px;}
+.full-width {width: 100% !important;}
+.progress-bar {
+    width: 100%;
+    height: 30px;
+    background-color: #e0e0e0;
+    border-radius: 15px;
+    margin: 15px 0;
+    position: relative;
+    overflow: hidden;
+}
+.progress-fill {
+    height: 100%;
+    background-color: #4CAF50;
+    border-radius: 15px;
+    text-align: center;
+    line-height: 30px;
+    color: white;
+    font-weight: bold;
+    transition: width 0.3s;
+}
+.fencer {
+    position: absolute;
+    top: -5px;
+    font-size: 24px;
+    transition: left 0.3s;
+    transform: scaleX(-1);
+}
+"""
+
+def _make_progress_bar(percent: int, final_text: str = None):
+    text = f"{percent}%" if not final_text else final_text
+    return f"""
+    <div class="progress-bar">
+      <div id="progress-fill" class="progress-fill" style="width:{percent}%">{text}</div>
+      <div id="fencer" class="fencer" style="left:{percent}%">🤺</div>
+    </div>
+    """
+
+def run_with_progress(video_file):
+    if not video_file:
+        yield [], "Please upload a video file.", gr.update(visible=False)
+        return
+
+    yield [], "🔄 Extracting frames...", _make_progress_bar(20)
+    df, fps = extract_feature_timeseries(video_file, frame_skip=FRAME_SKIP, debug=False)
+    if df.empty:
+        yield [], "❌ No frames processed!", _make_progress_bar(100, "No Frames ❌")
+        return
+
+    yield [], "🔄 Scoring & detecting touches...", _make_progress_bar(80)
+    clips, status_msg = extract_score_clips(video_file, debug=False)
+
+    final_bar = _make_progress_bar(
+        100, f"Detected {len(clips)} Touches ⚡" if clips else "No Touches"
+    )
+    yield clips, status_msg, final_bar
+
+with gr.Blocks(css=CSS, title="Fencing Scoreboard Detector") as demo:
+    with gr.Row(elem_classes="header"):
+        gr.Markdown(
+            "## 🤺 Fencing Score Detector\n"
+            "Upload a fencing bout video. The system detects scoreboard lights "
+            "(YOLO + AutoGluon) and returns highlight clips around each scoring event."
+        )
+
+    in_video = gr.Video(label="Upload Bout Video", elem_classes="full-width", height=400)
+    run_btn  = gr.Button("⚡ Detect Touches", elem_classes="full-width")
+
+    progress_html = gr.HTML(value="", label="Processing Progress", visible=False)
+    status = gr.Markdown("Ready.")
+    gallery = gr.Gallery(
+        label="Detected Clips",
+        columns=1,
+        height=400,
+        preview=True,
+        allow_preview=True,
+        show_download_button=True,
+        visible=False
+    )
+
+    def wrapped_run(video_file):
+        yield gr.update(value=[], visible=False), "Processing started...", gr.update(value=_make_progress_bar(0), visible=True)
+        for clips, msg, bar in run_with_progress(video_file):
+            gallery_update = gr.update(value=clips, visible=bool(clips))
+            yield gallery_update, msg, gr.update(value=bar, visible=True)
+
+    run_btn.click(
+        fn=wrapped_run,
+        inputs=in_video,
+        outputs=[gallery, status, progress_html],
+    )
+
+if __name__ == "__main__":
+    demo.launch(debug=True)
+