Update app.py

app.py

@@ -17,9 +17,6 @@
 # -*- 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
@@ -29,6 +26,14 @@ import pandas as pd
 import cv2
 import gradio as gr
 
+# ----------------
+# Flags
+# ----------------
+DEBUG_SAVE_FRAMES = False   # disable debug frames by default
+
+# ----------------
+# Utility
+# ----------------
 def _pip(pkgs: List[str]):
     import subprocess, sys
     subprocess.check_call([sys.executable, "-m", "pip", "install", "--quiet", *pkgs])
@@ -82,6 +87,8 @@ 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)
 
 # ----------------
@@ -98,8 +105,11 @@ def load_autogluon_tabular_from_hub() -> TabularPredictor:
         shutil.rmtree(extract_dir)
     with zipfile.ZipFile(z, "r") as zip_ref:
         zip_ref.extractall(extract_dir)
-    return TabularPredictor.load(str(extract_dir), require_py_version_match=False)
-
+    return TabularPredictor.load(
+        str(extract_dir),
+        require_py_version_match=False,
+        require_version_match=False
+    )
 
 _YOLO = None
 _AG_PRED = None
@@ -162,155 +172,19 @@ def isolate_scoreboard_color(frame_bgr: np.ndarray,
                 if x2 > x1 and y2 > y1:
                     gray[y1:y2, x1:x2] = frame_bgr[y1:y2, x1:x2]
 
-    if debug and frame_id is not None:
+    if DEBUG_SAVE_FRAMES and debug and frame_id is not None:
         dbg = gray.copy()
         if chosen_box is not None:
             x1, y1, x2, y2 = [int(round(v)) for v in chosen_box]
             cv2.rectangle(dbg, (x1,y1), (x2,y2), (0,255,0), 2)
         out_path = DEBUG_DIR / f"frame_{frame_id:06d}.jpg"
         cv2.imwrite(str(out_path), dbg)
-        print(f"[DEBUG] Saved debug frame → {out_path}")
 
     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:
-        out_csv = DEBUG_DIR / f"features_{uuid.uuid4().hex}.csv"
-        df.to_csv(out_csv, index=False)
-        print(f"[DEBUG] Saved features CSV → {out_csv}")
-
-    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)),
@@ -326,18 +200,18 @@ def cut_clip(video_path: str, start: float, end: float, out_path: str) -> str:
     return out_path
 
 # ----------------------------
-# Orchestrator
+# Orchestrator (with cleanup)
 # ----------------------------
-def extract_score_clips(video_path: str, debug: bool = False) -> Tuple[List[Tuple[str, str]], str]:
+def extract_score_clips(video_path: str, debug: bool = False):
     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:
-        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."
+        return [], "⚠️ No scoreboard detected or illumination scores flat."
 
     events = pick_events(df, score, fps)
     if not events:
@@ -347,11 +221,10 @@ def extract_score_clips(video_path: str, debug: bool = False) -> Tuple[List[Tupl
     if duration <= 0:
         duration = float(df["timestamp"].max() + CLIP_PAD_S + 0.5)
 
-    clips = []
+    clips, kept_paths = [], []
     base = os.path.splitext(os.path.basename(video_path))[0]
     for i, t in enumerate(events):
-        s = t - CLIP_PAD_S
-        e = t + CLIP_PAD_S
+        s, e = t - CLIP_PAD_S, t + CLIP_PAD_S
         if s < 0:
             e = min(duration, e - s)
             s = 0
@@ -360,108 +233,14 @@ def extract_score_clips(video_path: str, debug: bool = False) -> Tuple[List[Tupl
             e = duration
         clip_path = os.path.join(tempfile.gettempdir(), f"{base}_score_{i+1:02d}.mp4")
         cut_clip(video_path, s, e, clip_path)
-        label = f"Touch {i+1} @ {t:.2f}s"
-        clips.append((clip_path, label))
-
-    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
+        clips.append((clip_path, f"Touch {i+1} @ {t:.2f}s"))
+        kept_paths.append(clip_path)
 
-    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)
+    # cleanup: delete unused temp 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
 
-    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],
-    )
+    return clips, f"✅ Detected {len(clips)} event(s)."
 
-if __name__ == "__main__":
-    demo.launch(debug=True)