Create app.py

app.py

@@ -0,0 +1,453 @@
+# app.py
+# A lightweight, user-friendly FFmpeg UI for extracting video frames on Hugging Face Spaces.
+# - Works on CPU (no GPU required)
+# - Shows the exact ffmpeg command used
+# - Lets you extract every N seconds, every Nth frame, or at an exact FPS
+# - Optional start/end time trims, resize, JPG quality / PNG compression, scene-change detection
+# - Returns a ZIP of frames and a gallery preview
+
+import os
+import re
+import io
+import sys
+import json
+import math
+import time
+import shutil
+import zipfile
+import tempfile
+import subprocess
+from pathlib import Path
+from typing import List, Tuple, Optional
+
+import gradio as gr
+
+# ─────────────────────────────────────────────────────────────
+# Utility: check for ffmpeg/ffprobe availability
+# ─────────────────────────────────────────────────────────────
+
+def _which(name: str) -> Optional[str]:
+    from shutil import which
+    return which(name)
+
+FFMPEG = _which("ffmpeg")
+FFPROBE = _which("ffprobe")
+
+if not FFMPEG or not FFPROBE:
+    # Friendly message shown in the UI footer if ffmpeg is missing
+    MISSING_MSG = (
+        "⚠️ FFmpeg not found. On Hugging Face Spaces, add a file named 'packages.txt' "
+        "with a single line 'ffmpeg' (and optionally 'libsm6' 'libxext6'). Then restart the Space."
+    )
+else:
+    MISSING_MSG = ""
+
+# ─────────────────────────────────────────────────────────────
+# Video probing via ffprobe
+# ─────────────────────────────────────────────────────────────
+
+def ffprobe_json(input_path: str) -> dict:
+    if not FFPROBE:
+        return {}
+    cmd = [
+        FFPROBE,
+        "-v", "error",
+        "-print_format", "json",
+        "-show_streams",
+        "-show_format",
+        input_path,
+    ]
+    res = subprocess.run(cmd, capture_output=True, text=True)
+    if res.returncode != 0:
+        return {}
+    try:
+        return json.loads(res.stdout)
+    except Exception:
+        return {}
+
+
+def parse_video_info(meta: dict) -> dict:
+    info = {"duration": None, "fps": None, "width": None, "height": None, "codec": None}
+    if not meta:
+        return info
+
+    # Duration from format
+    try:
+        info["duration"] = float(meta.get("format", {}).get("duration", None))
+    except Exception:
+        pass
+
+    # Find the first video stream
+    vstreams = [s for s in meta.get("streams", []) if s.get("codec_type") == "video"]
+    if vstreams:
+        v = vstreams[0]
+        info["codec"] = v.get("codec_name")
+        info["width"] = v.get("width")
+        info["height"] = v.get("height")
+        # FPS from r_frame_rate
+        rfr = v.get("r_frame_rate") or v.get("avg_frame_rate")
+        if rfr and "/" in rfr:
+            num, den = rfr.split("/")
+            try:
+                num = float(num)
+                den = float(den)
+                if den != 0:
+                    info["fps"] = num / den
+            except Exception:
+                pass
+    return info
+
+
+# ─────────────────────────────────────────────────────────────
+# FFmpeg command builder
+# ─────────────────────────────────────────────────────────────
+
+def build_ffmpeg_command(
+    input_path: str,
+    mode: str,
+    every_seconds: float,
+    nth_frame: int,
+    exact_fps: float,
+    start_time: str,
+    end_time: str,
+    long_side: int,
+    out_format: str,
+    jpg_quality: int,
+    png_level: int,
+    scene_detect: bool,
+    scene_thresh: float,
+    out_pattern: str,
+) -> List[str]:
+    """Return a full ffmpeg command list for subprocess.run."""
+    if not FFMPEG:
+        raise RuntimeError("FFmpeg is not available on this system.")
+
+    cmd = [FFMPEG, "-y"]
+
+    # Optional in/out trims
+    if start_time:
+        cmd += ["-ss", start_time]
+
+    cmd += ["-i", input_path]
+
+    if end_time:
+        # Use -to for end timestamp relative to input start
+        cmd += ["-to", end_time]
+
+    # Build filter chain
+    vf_parts = []
+
+    # 1) Frame selection / rate
+    if mode == "Every N seconds":
+        # fps=1/seconds
+        rate = 1.0 / max(every_seconds, 0.000001)
+        vf_parts.append(f"fps={rate}")
+    elif mode == "Every Nth frame":
+        # select='not(mod(n\,N))' -> then set fps to input fps to avoid duplicating
+        vf_parts.append(f"select='not(mod(n,{max(nth_frame,1)}))'")
+        vf_parts.append("setpts=N/FRAME_RATE/TB")
+    elif mode == "Exact FPS":
+        vf_parts.append(f"fps={max(exact_fps, 0.000001)}")
+    elif mode == "All frames":
+        # No explicit fps filter — pass all frames
+        pass
+    else:
+        vf_parts.append("fps=1")
+
+    # 2) Scene change detection (grabs frames when scene changes by threshold)
+    if scene_detect:
+        # Use select filter: 'gt(scene,THRESH)' outputs only scene-change frames
+        vf_parts.append(f"select='gt(scene,{scene_thresh})',showinfo")
+        vf_parts.append("setpts=N/FRAME_RATE/TB")
+
+    # 3) Resize by long side
+    if long_side and long_side > 0:
+        # Maintain aspect: scale=LONG:-1 sets height auto; but we need to pick which side is longer
+        # Use force_original_aspect_ratio=decrease and -1 for one dim with eval
+        # To keep the *long* side at long_side, we can use scale logic with if(gt(iw,ih),...)
+        vf_parts.append(
+            f"scale='if(gt(iw,ih),{long_side},-1)':'if(gt(iw,ih),-1,{long_side})':force_original_aspect_ratio=decrease"
+        )
+
+    # Join vf
+    if vf_parts:
+        cmd += ["-vf", ",".join(vf_parts)]
+
+    # 4) Output options per format
+    ext = out_format.lower()
+    if ext == "jpg":
+        # Lower -q:v is higher quality. Map slider (2..31) directly
+        cmd += ["-q:v", str(jpg_quality)]
+    elif ext == "png":
+        # Compression level 0..9
+        cmd += ["-compression_level", str(png_level)]
+
+    # Avoid timestamps gaps in patterns
+    cmd += ["-frame_pts", "1"]
+
+    # Output pattern
+    cmd += [out_pattern]
+    return cmd
+
+
+# ─────────────────────────────────────────────────────────────
+# Extraction runtime
+# ─────────────────────────────────────────────────────────────
+
+def extract_frames(
+    video: gr.File | None,
+    mode: str,
+    every_seconds: float,
+    nth_frame: int,
+    exact_fps: float,
+    start_time: str,
+    end_time: str,
+    long_side: int,
+    out_format: str,
+    jpg_quality: int,
+    png_level: int,
+    scene_detect: bool,
+    scene_thresh: float,
+    prefix: str,
+    progress=gr.Progress(track_tqdm=True),
+):
+    if not video or not video.name:
+        return None, None, "Please upload a video.", ""
+
+    if not FFMPEG or not FFPROBE:
+        return None, None, "FFmpeg is not available. See the note below.", MISSING_MSG
+
+    # Probe video info
+    meta = ffprobe_json(video.name)
+    info = parse_video_info(meta)
+
+    # Prepare temp dir
+    work = Path(tempfile.mkdtemp(prefix="frames_"))
+    out_dir = work / "frames"
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    # Output pattern
+    ext = out_format.lower()
+    pattern = str(out_dir / f"{prefix}_%05d.{ext}")
+
+    # Build command
+    cmd = build_ffmpeg_command(
+        input_path=video.name,
+        mode=mode,
+        every_seconds=every_seconds,
+        nth_frame=nth_frame,
+        exact_fps=exact_fps,
+        start_time=start_time.strip(),
+        end_time=end_time.strip(),
+        long_side=long_side,
+        out_format=ext,
+        jpg_quality=jpg_quality,
+        png_level=png_level,
+        scene_detect=scene_detect,
+        scene_thresh=scene_thresh,
+        out_pattern=pattern,
+    )
+
+    # Friendly command preview
+    command_preview = " ".join([sh if " " not in sh else f'"{sh}"' for sh in cmd])
+
+    # Estimate total frames for progress (best-effort)
+    total = None
+    try:
+        duration = info.get("duration")
+        in_fps = info.get("fps") or 30
+        if start_time:
+            # Roughly convert HH:MM:SS.mmm to seconds
+            parts = [float(x) for x in re.split(r"[:]", start_time)]
+            if len(parts) == 3:
+                duration = max(0.0, (duration or 0) - (parts[0]*3600 + parts[1]*60 + parts[2]))
+        if end_time:
+            parts = [float(x) for x in re.split(r"[:]", end_time)]
+            if len(parts) == 3:
+                duration = min(duration or 0, (parts[0]*3600 + parts[1]*60 + parts[2]))
+        if duration:
+            if mode == "Every N seconds" and every_seconds > 0:
+                total = int(math.ceil(duration / every_seconds))
+            elif mode == "Every Nth frame" and in_fps and nth_frame > 0:
+                total = int(math.ceil((duration * in_fps) / nth_frame))
+            elif mode == "Exact FPS" and exact_fps > 0:
+                total = int(math.ceil(duration * exact_fps))
+            elif mode == "All frames" and in_fps:
+                total = int(math.ceil(duration * in_fps))
+    except Exception:
+        total = None
+
+    # Run ffmpeg and stream stderr for incremental progress by counting files created
+    proc = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, text=True, bufsize=1)
+
+    created = 0
+    last_update = time.time()
+    while True:
+        line = proc.stderr.readline()
+        if not line and proc.poll() is not None:
+            break
+        # Periodically refresh progress based on files present
+        if time.time() - last_update > 0.2:
+            created = len(list(out_dir.glob(f"{prefix}_*.{ext}")))
+            if total:
+                progress(created / max(total, 1))
+            last_update = time.time()
+
+    ret = proc.wait()
+
+    # Final count
+    frame_files = sorted(out_dir.glob(f"{prefix}_*.{ext}"))
+    created = len(frame_files)
+
+    if ret != 0 or created == 0:
+        # Read remaining stderr to show message
+        try:
+            err_rest = proc.stderr.read() if proc.stderr else ""
+        except Exception:
+            err_rest = ""
+        return None, None, f"FFmpeg failed or produced no frames.\n\nStderr:\n{err_rest}", command_preview
+
+    # Build a small gallery (cap to avoid huge RAM)
+    gallery_cap = 60
+    gallery_paths = [str(p) for p in frame_files[:gallery_cap]]
+
+    # Zip everything
+    zip_path = work / f"{prefix}_frames.zip"
+    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for p in frame_files:
+            zf.write(p, p.name)
+
+    # Info text
+    info_lines = []
+    if info.get("fps"):
+        info_lines.append(f"Input FPS: {info['fps']:.3f}")
+    if info.get("duration"):
+        info_lines.append(f"Duration: {info['duration']:.2f}s")
+    if info.get("width") and info.get("height"):
+        info_lines.append(f"Resolution: {info['width']}×{info['height']}")
+    info_lines.append(f"Frames extracted: {created}")
+
+    details = "\n".join(info_lines)
+
+    return gallery_paths, str(zip_path), details, command_preview
+
+
+# ─────────────────────────────────────────────────────────────
+# UI
+# ─────────────────────────────────────────────────────────────
+
+def build_ui():
+    with gr.Blocks(theme=gr.themes.Soft(), css="""
+    .cf-title { font-size: 1.6rem; font-weight: 800; }
+    .cf-sub { opacity: .8; }
+    .cmdbox textarea { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 12px; }
+    """) as demo:
+        gr.Markdown("""
+        <div class="cf-title">FFmpeg Frames UI</div>
+        <div class="cf-sub">Extract JPG/PNG frames from videos — friendly controls, exact command shown.</div>
+        """)
+
+        with gr.Row():
+            video = gr.File(label="Upload video", file_types=[".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"], type="filepath")
+
+        with gr.Row():
+            mode = gr.Dropdown(
+                ["Every N seconds", "Every Nth frame", "Exact FPS", "All frames"],
+                value="Every N seconds", label="Extraction mode"
+            )
+            every_seconds = gr.Number(value=1.0, label="Every N seconds (e.g., 0.5 = every 500ms)")
+            nth_frame = gr.Number(value=30, label="Every Nth frame (e.g., 30 = 1 frame per 30)")
+            exact_fps = gr.Number(value=1.0, label="Exact FPS (e.g., 2.0)")
+
+        with gr.Row():
+            start_time = gr.Textbox(value="", label="Start time (HH:MM:SS.mmm, optional)")
+            end_time = gr.Textbox(value="", label="End time (HH:MM:SS.mmm, optional)")
+            long_side = gr.Number(value=0, label="Resize long side px (0 = no resize)")
+
+        with gr.Row():
+            out_format = gr.Dropdown(["jpg", "png"], value="jpg", label="Output format")
+            jpg_quality = gr.Slider(2, 31, value=3, step=1, label="JPG quality (2=best, 31=worst)")
+            png_level = gr.Slider(0, 9, value=2, step=1, label="PNG compression level (0..9)")
+
+        with gr.Row():
+            scene_detect = gr.Checkbox(value=False, label="Enable scene-change extraction (advanced)")
+            scene_thresh = gr.Slider(0.0, 1.0, value=0.3, step=0.01, label="Scene threshold (0..1)")
+            prefix = gr.Textbox(value="frame", label="Filename prefix")
+
+        with gr.Row():
+            run_btn = gr.Button("Extract Frames", variant="primary")
+            info_btn = gr.Button("Probe Video (FPS / Duration)")
+
+        with gr.Row():
+            gallery = gr.Gallery(label="Preview (first 60 frames)", columns=6, height=300)
+
+        with gr.Row():
+            zip_out = gr.File(label="Download all frames as ZIP")
+
+        with gr.Row():
+            details = gr.Markdown("Frames extracted: —")
+
+        with gr.Accordion("Show exact ffmpeg command", open=False):
+            cmd_preview = gr.Textbox(label="ffmpeg command", lines=4, elem_classes=["cmdbox"])
+
+        # Footer
+        if MISSING_MSG:
+            gr.Markdown(f"<span style='color:#b45309'>{MISSING_MSG}</span>")
+
+        # Wire behavior: enable/disable param groups depending on mode / format
+        def _toggle_params(mode_val, fmt):
+            return (
+                gr.update(visible=(mode_val == "Every N seconds")),
+                gr.update(visible=(mode_val == "Every Nth frame")),
+                gr.update(visible=(mode_val == "Exact FPS")),
+                gr.update(visible=(fmt == "jpg")),
+                gr.update(visible=(fmt == "png")),
+            )
+
+        mode.change(
+            _toggle_params,
+            inputs=[mode, out_format],
+            outputs=[every_seconds, nth_frame, exact_fps, jpg_quality, png_level],
+        )
+        out_format.change(
+            _toggle_params,
+            inputs=[mode, out_format],
+            outputs=[every_seconds, nth_frame, exact_fps, jpg_quality, png_level],
+        )
+        # Initialize visibility
+        demo.load(_toggle_params, inputs=[mode, out_format], outputs=[every_seconds, nth_frame, exact_fps, jpg_quality, png_level])
+
+        # Actions
+        run_btn.click(
+            extract_frames,
+            inputs=[
+                video, mode, every_seconds, nth_frame, exact_fps,
+                start_time, end_time, long_side, out_format, jpg_quality, png_level,
+                scene_detect, scene_thresh, prefix
+            ],
+            outputs=[gallery, zip_out, details, cmd_preview],
+            api_name="extract_frames",
+        )
+
+        def probe(video: gr.File | None):
+            if not video or not video.name:
+                return "Upload a video to probe."
+            meta = ffprobe_json(video.name)
+            info = parse_video_info(meta)
+            lines = ["**Video info**"]
+            if info.get("fps"):
+                lines.append(f"• FPS: **{info['fps']:.3f}**")
+            if info.get("duration"):
+                lines.append(f"• Duration: **{info['duration']:.2f}s**")
+            if info.get("width") and info.get("height"):
+                lines.append(f"• Resolution: **{info['width']}×{info['height']}**")
+            return "\n".join(lines)
+
+        info_btn.click(probe, inputs=[video], outputs=[details])
+
+    return demo
+
+
+if __name__ == "__main__":
+    demo = build_ui()
+    demo.queue().launch()