"""
Video Blur AI - Main Gradio Web Application
Text-prompted video object blurring using Grounding DINO + SAM 2
Runs locally and on Hugging Face Spaces (including ZeroGPU).
Local usage:
python app.py
Then open http://localhost:7860 in your browser.
"""
import os
import sys
import time
import tempfile
import subprocess
from pathlib import Path
# ─── ZeroGPU / Hugging Face Spaces support ─────────────────────
# `spaces` MUST be imported before torch (pulled in transitively by
# config/pipeline) so ZeroGPU can patch CUDA. Locally the package is
# absent, so we fall back to a no-op decorator that supports the same
# @spaces.GPU and @spaces.GPU(duration=...) forms.
try:
import spaces # type: ignore
_HAS_SPACES = True
except Exception:
_HAS_SPACES = False
class _SpacesShim:
@staticmethod
def GPU(*args, **kwargs):
# Used as a bare decorator: @spaces.GPU
if len(args) == 1 and callable(args[0]) and not kwargs:
return args[0]
# Used with arguments: @spaces.GPU(duration=...)
def _decorator(fn):
return fn
return _decorator
spaces = _SpacesShim() # type: ignore
import gradio as gr
import numpy as np
# ─── Work around a gradio_client JSON-schema bug ───────────────
# In some gradio 4.x builds, gradio_client.utils.get_type() raises
# "argument of type 'bool' is not iterable" when a component's schema has a
# boolean `additionalProperties`. This happens during API-info generation,
# which Gradio calls in its startup health check — the failed check then
# makes launch() abort demanding share=True. We make the schema walkers
# tolerate non-dict / boolean schemas so startup succeeds on Spaces.
try:
import gradio_client.utils as _gc_utils
_orig_get_type = _gc_utils.get_type
def _get_type_safe(schema):
if not isinstance(schema, dict):
return "Any"
return _orig_get_type(schema)
_gc_utils.get_type = _get_type_safe
_orig_j2pt = _gc_utils._json_schema_to_python_type
def _j2pt_safe(schema, defs=None):
if isinstance(schema, bool):
return "Any"
return _orig_j2pt(schema, defs)
_gc_utils._json_schema_to_python_type = _j2pt_safe
except Exception as _patch_err:
print(f"⚠️ Could not apply gradio_client schema patch: {_patch_err}")
# ─── Compat shim for Starlette TemplateResponse ────────────────
# Gradio 4.x calls templates.TemplateResponse(name, context) (old signature).
# Newer Starlette put `request` first and dropped old-style support, so the
# template name gets mistaken for the request and Gradio's page render crashes
# with "unhashable type: 'dict'" (which then makes launch() demand share=True).
# We restore old-style support. We patch the EXACT instance Gradio's route uses
# (gradio.routes.templates) — the most reliable target — and the class as a
# fallback, so it works regardless of Starlette version.
def _old_style_args(args, kwargs):
"""Return (name, request, context, rest, kwargs) for an old-style call."""
name = args[0]
context = args[1] if len(args) > 1 else kwargs.pop("context", None)
context = dict(context) if isinstance(context, dict) else {}
request = context.get("request")
return name, request, context, args[2:], kwargs
def _install_templateresponse_compat():
patched = []
# 1) Patch Gradio's actual templates instance (primary — bulletproof).
try:
import gradio.routes as _gr_routes
_inst = getattr(_gr_routes, "templates", None)
if _inst is not None and not getattr(_inst.TemplateResponse, "_vb_compat", False):
_orig_bound = _inst.TemplateResponse # bound method (self baked in)
def _inst_compat(*args, **kwargs): # instance attr => no self
if args and isinstance(args[0], str):
name, request, context, rest, kwargs = _old_style_args(args, kwargs)
return _orig_bound(request, name, context, *rest, **kwargs)
return _orig_bound(*args, **kwargs)
_inst_compat._vb_compat = True
_inst.TemplateResponse = _inst_compat
patched.append("gradio.routes.templates")
except Exception as e:
print(f" (instance-level TemplateResponse patch skipped: {e})")
# 2) Patch the class too (fallback for any other Jinja2Templates instances).
try:
import starlette.templating as _st
_JT = _st.Jinja2Templates
if not getattr(_JT.TemplateResponse, "_vb_compat", False):
_class_orig = _JT.TemplateResponse
def _class_compat(self, *args, **kwargs):
if args and isinstance(args[0], str):
name, request, context, rest, kwargs = _old_style_args(args, kwargs)
return _class_orig(self, request, name, context, *rest, **kwargs)
return _class_orig(self, *args, **kwargs)
_class_compat._vb_compat = True
_JT.TemplateResponse = _class_compat
patched.append("Jinja2Templates(class)")
except Exception as e:
print(f" (class-level TemplateResponse patch skipped: {e})")
return patched
try:
_tr_patched = _install_templateresponse_compat()
if _tr_patched:
print(f"✅ TemplateResponse compat shim installed on: {', '.join(_tr_patched)}")
else:
print("ℹ️ TemplateResponse compat shim not needed (nothing to patch).")
except Exception as _tr_err:
print(f"⚠️ TemplateResponse compat shim failed: {_tr_err}")
from config import get_config, AppConfig, IS_ZERO_GPU
from pipeline import VideoBlurPipeline
# ─── Global State ──────────────────────────────────────────────
config = get_config()
pipeline = VideoBlurPipeline(config)
# GPU time budget (seconds) requested from ZeroGPU per call.
PROCESS_DURATION = getattr(config, "gpu_duration", 120)
PREVIEW_DURATION = 60
# NOTE: models are loaded LAZILY inside the @spaces.GPU functions below
# (pipeline.ensure_models_loaded() runs at the start of process/preview).
# We deliberately do NOT preload at import time: on ZeroGPU the real GPU is
# only attached *inside* a @spaces.GPU call, so building SAM 2 on CUDA at
# import mixes devices ("cuda:0 (ZeroGPU) and cpu"). Loading on first request
# — inside the GPU context — avoids that. The first request is a little
# slower because it loads the weights; subsequent ones reuse the cache.
# ─── Custom CSS ────────────────────────────────────────────────
CUSTOM_CSS = """
/* Main container */
.gradio-container {
max-width: 1200px !important;
margin: auto !important;
font-family: 'Segoe UI', system-ui, -apple-system, sans-serif !important;
}
/* Header */
.app-header {
text-align: center;
padding: 20px 0;
margin-bottom: 10px;
background: linear-gradient(135deg, #0f0c29, #302b63, #24243e);
border-radius: 16px;
color: white;
}
.app-header h1 {
font-size: 2.2em !important;
font-weight: 700 !important;
margin: 0 !important;
background: linear-gradient(90deg, #00d2ff, #7b2ff7, #ff6b6b);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
background-clip: text;
}
.app-header p {
color: #a0a0c0;
font-size: 1.05em;
margin: 8px 0 0 0;
}
/* Prompt input */
.prompt-input textarea {
font-size: 1.15em !important;
border: 2px solid #7b2ff7 !important;
border-radius: 12px !important;
padding: 14px !important;
transition: all 0.3s ease !important;
}
.prompt-input textarea:focus {
border-color: #00d2ff !important;
box-shadow: 0 0 0 3px rgba(0, 210, 255, 0.15) !important;
}
/* Process button */
.process-btn {
background: linear-gradient(135deg, #7b2ff7, #00d2ff) !important;
border: none !important;
font-size: 1.15em !important;
font-weight: 600 !important;
padding: 14px 32px !important;
border-radius: 12px !important;
transition: all 0.3s ease !important;
}
.process-btn:hover {
transform: translateY(-2px) !important;
box-shadow: 0 6px 20px rgba(123, 47, 247, 0.35) !important;
}
/* Preview button */
.preview-btn {
background: linear-gradient(135deg, #ff6b6b, #ffa07a) !important;
border: none !important;
border-radius: 12px !important;
font-weight: 600 !important;
}
/* Status */
.status-text {
font-size: 1.05em;
padding: 10px;
border-radius: 8px;
text-align: center;
}
/* Settings panel */
.settings-panel {
background: rgba(123, 47, 247, 0.04);
border: 1px solid rgba(123, 47, 247, 0.12);
border-radius: 12px;
padding: 16px;
}
/* Example prompts */
.example-box {
border: 1px solid rgba(123, 47, 247, 0.2);
border-radius: 10px;
padding: 12px;
margin: 4px 0;
}
"""
# ─── Processing Functions ──────────────────────────────────────
def trim_clip(video_path, start, length):
"""Cut [start, start+length] seconds from a video with ffmpeg.
Returns the path to a new .mp4. Re-encodes (fast preset) so the cut is
frame-accurate. This is CPU-only work (no GPU needed).
"""
start = max(0.0, float(start))
length = max(0.1, float(length))
out_dir = tempfile.mkdtemp(prefix="trim_")
out_path = os.path.join(out_dir, f"trimmed_{Path(video_path).stem}.mp4")
cmd = [
"ffmpeg", "-y",
"-ss", str(start),
"-i", video_path,
"-t", str(length),
"-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
"-c:a", "aac",
"-movflags", "+faststart",
out_path,
]
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0 or not os.path.exists(out_path):
raise RuntimeError((proc.stderr or "ffmpeg failed")[-400:])
return out_path
def preview_trim(video_file, trim_start, trim_length):
"""Produce a preview of the trimmed segment so the user can confirm it."""
if video_file is None:
gr.Warning("⚠️ Please upload a video first!")
return None
try:
video_path = video_file if isinstance(video_file, str) else video_file.name
cap = getattr(config, "max_video_duration", 600)
length = min(float(trim_length) if trim_length else cap, cap)
return trim_clip(video_path, trim_start, length)
except Exception as e:
gr.Warning(f"Trim error: {e}")
return None
@spaces.GPU(duration=PROCESS_DURATION)
def process_video(
video_file,
text_prompt,
blur_type,
blur_strength,
edge_feather,
processing_mode,
keyframe_interval,
detection_threshold,
model_quality,
trim_enable,
trim_start,
trim_length,
progress=gr.Progress()
):
"""Main video processing function"""
# Validate inputs
if video_file is None:
gr.Warning("⚠️ Please upload a video first!")
return None, "❌ No video uploaded"
if not text_prompt or not text_prompt.strip():
gr.Warning("⚠️ Please enter a text prompt describing what to blur!")
return None, "❌ No prompt entered"
text_prompt = text_prompt.strip()
# Update model based on quality selection
if model_quality == "base":
config.model.gdino_model_id = "IDEA-Research/grounding-dino-base"
else:
config.model.gdino_model_id = "IDEA-Research/grounding-dino-tiny"
# Progress wrapper for Gradio
def gradio_progress(value, text):
progress(value, desc=text)
try:
# Get video path
if isinstance(video_file, str):
video_path = video_file
else:
video_path = video_file.name if hasattr(video_file, 'name') else str(video_file)
# ── Trim (for long videos) ──
# Trim when the user asked to, OR automatically when the clip is longer
# than the cap — so long videos are handled instead of rejected.
max_secs = getattr(config, "max_video_duration", 600)
try:
_info = pipeline.video_processor.get_video_info(video_path)
duration = _info.duration or 0.0
except Exception:
duration = 0.0
should_trim = bool(trim_enable) or (duration and duration > max_secs)
if should_trim:
length = min(float(trim_length) if trim_length else max_secs, max_secs)
start = max(0.0, float(trim_start) if trim_start else 0.0)
if duration and start >= duration:
start = 0.0
gradio_progress(0.02, f"✂️ Trimming to {start:.0f}–{start + length:.0f}s...")
try:
video_path = trim_clip(video_path, start, length)
except Exception as e:
gr.Warning(f"Trim failed: {e}")
return None, f"❌ Trim failed: {e}"
# Generate output path (from the possibly-trimmed video)
output_dir = tempfile.mkdtemp(prefix="blur_output_")
output_name = f"blurred_{Path(video_path).stem}.mp4"
output_path = os.path.join(output_dir, output_name)
# Process
result_path = pipeline.process_video(
video_path=video_path,
text_prompt=text_prompt,
output_path=output_path,
blur_type=blur_type,
blur_strength=int(blur_strength),
edge_feather=int(edge_feather),
processing_mode=processing_mode,
keyframe_interval=int(keyframe_interval),
detection_threshold=detection_threshold,
progress_callback=gradio_progress,
)
status = "✅ Done! Video saved successfully."
return result_path, status
except Exception as e:
error_msg = f"❌ Error: {str(e)}"
gr.Warning(error_msg)
return None, error_msg
@spaces.GPU(duration=PREVIEW_DURATION)
def preview_detection(video_file, text_prompt, frame_slider):
"""Preview detection on a single frame"""
if video_file is None:
gr.Warning("⚠️ Please upload a video first!")
return None
if not text_prompt or not text_prompt.strip():
gr.Warning("⚠️ Please enter a prompt!")
return None
try:
video_path = video_file if isinstance(video_file, str) else video_file.name
vis = pipeline.preview_detection(
video_path,
text_prompt.strip(),
frame_number=int(frame_slider)
)
# Convert BGR -> RGB for Gradio
vis_rgb = vis[..., ::-1]
return vis_rgb
except Exception as e:
gr.Warning(f"Preview error: {str(e)}")
return None
def get_video_info_text(video_file):
"""Get video info when uploaded, and set up the trim controls."""
cap = getattr(config, "max_video_duration", 600)
if video_file is None:
return (
"No video uploaded",
gr.Slider(maximum=0, value=0), # frame_slider
gr.Checkbox(value=False), # trim_enable
gr.Slider(maximum=1, value=0), # trim_start
gr.Slider(maximum=cap, value=cap), # trim_length
"", # trim_info
)
try:
video_path = video_file if isinstance(video_file, str) else video_file.name
info = pipeline.video_processor.get_video_info(video_path)
text = (
f"📹 **{Path(video_path).name}**\n"
f"Resolution: {info.width}×{info.height} | "
f"FPS: {info.fps:.1f} | "
f"Frames: {info.total_frames} | "
f"Duration: {info.duration:.1f}s | "
f"Audio: {'Yes ✅' if info.has_audio else 'No ❌'}"
)
max_frame = max(0, info.total_frames - 1)
dur = info.duration or 0.0
too_long = bool(dur and dur > cap)
length_default = min(cap, dur) if dur else cap
trim_info = ""
if too_long:
trim_info = (
f"⏱️ This clip is **{dur:.0f}s**, longer than the **{cap}s** limit. "
f"It will be trimmed automatically — set the start below to choose "
f"which part to keep (defaults to the first {cap}s), and optionally "
f"**Preview trimmed clip** to check it."
)
return (
text,
gr.Slider(maximum=max_frame, value=0), # frame_slider
gr.Checkbox(value=too_long), # trim_enable (auto-on if long)
gr.Slider(maximum=max(0.1, dur), value=0), # trim_start
gr.Slider(maximum=cap, value=length_default), # trim_length
trim_info, # trim_info
)
except Exception as e:
return (
f"❌ Error reading video: {e}",
gr.Slider(maximum=0, value=0),
gr.Checkbox(value=False),
gr.Slider(maximum=1, value=0),
gr.Slider(maximum=cap, value=cap),
"",
)
# ─── Build UI ──────────────────────────────────────────────────
def create_app():
"""Build the Gradio interface"""
# Match the model-quality selector to the active config so the first
# run doesn't trigger an unnecessary model reload.
default_quality = "tiny" if "tiny" in config.model.gdino_model_id else "base"
# Note shown to users about the environment limits (ZeroGPU only).
limits_note = ""
if IS_ZERO_GPU:
limits_note = (
f"Running on ZeroGPU · "
f"processed at {config.video.max_resolution}p · clips up to "
f"{config.max_video_duration}s (longer videos can be trimmed)
"
)
with gr.Blocks(
title="Video Blur AI",
css=CUSTOM_CSS,
theme=gr.themes.Soft(primary_hue="violet", secondary_hue="cyan"),
) as app:
# ── Header ──
gr.HTML(f"""
""")
with gr.Row():
# ── Left Column: Input ──
with gr.Column(scale=1):
# Video Upload
video_input = gr.Video(
label="📤 Upload Video",
)
# Video Info
video_info_text = gr.Markdown("No video uploaded")
# ── Trim controls (for long videos) ──
with gr.Accordion("✂️ Trim (for long videos)", open=False):
trim_info = gr.Markdown("")
trim_enable = gr.Checkbox(
label="Trim this video before processing",
value=False,
info=(
f"Clips here are limited to {config.max_video_duration}s. "
"Turn this on to process only a chosen segment. Long videos "
"are trimmed automatically."
),
)
with gr.Row():
trim_start = gr.Slider(
minimum=0, maximum=60, value=0, step=0.5,
label="Start (seconds)",
)
trim_length = gr.Slider(
minimum=1, maximum=config.max_video_duration,
value=config.max_video_duration, step=1,
label=f"Clip length (max {config.max_video_duration}s)",
)
trim_preview_btn = gr.Button(
"✂️ Preview trimmed clip", variant="secondary"
)
trim_preview_video = gr.Video(
label="Trimmed preview", interactive=False
)
# Prompt Input — this is where the user WRITES the regions to blur
text_prompt = gr.Textbox(
label="✏️ What to blur? (Text Prompt)",
placeholder="e.g., face, hand, license plate, person, text, logo...",
info="Describe the regions you want to blur. Use periods to separate multiple objects: 'face. hand. text.'",
lines=2,
elem_classes=["prompt-input"],
)
# Example Prompts
gr.Examples(
examples=[
["face."],
["face. hand."],
["license plate."],
["text. writing."],
["person."],
["logo. brand."],
["phone. screen."],
["face. license plate. text."],
],
inputs=[text_prompt],
label="💡 Example Prompts (click to use)",
)
# Settings
with gr.Accordion("⚙️ Advanced Settings", open=False):
model_quality = gr.Radio(
choices=[
("⚡ Fast (Tiny model ~350MB, good accuracy)", "tiny"),
("🎯 Accurate (Base model ~999MB, best accuracy)", "base"),
],
value=default_quality,
label="Model Quality",
info="Tiny downloads faster and runs faster. Base is more accurate."
)
blur_type = gr.Radio(
choices=["gaussian", "pixelate", "black"],
value="gaussian",
label="Blur Type",
info="Gaussian = smooth blur, Pixelate = mosaic, Black = solid cover"
)
blur_strength = gr.Slider(
minimum=11, maximum=151, value=51, step=2,
label="Blur Strength",
info="Higher = more blur (must be odd number)"
)
edge_feather = gr.Slider(
minimum=3, maximum=51, value=11, step=2,
label="Edge Feathering",
info="Smooth transition at blur edges"
)
processing_mode = gr.Radio(
choices=[
("🎯 SAM 2 Video Tracking (recommended)", "video_tracking"),
("⚡ Frame-by-Frame (faster, less consistent)", "frame_by_frame"),
],
value="video_tracking",
label="Processing Mode"
)
keyframe_interval = gr.Slider(
minimum=1, maximum=30, value=5, step=1,
label="Keyframe Interval (frame-by-frame mode)",
info="Re-detect every N frames"
)
detection_threshold = gr.Slider(
minimum=0.1, maximum=0.9, value=0.3, step=0.05,
label="Detection Confidence Threshold",
info="Lower = more detections, higher = more precise"
)
# ── Right Column: Output ──
with gr.Column(scale=1):
# Preview
with gr.Tab("🔍 Preview Detection"):
preview_image = gr.Image(
label="Detection Preview",
type="numpy"
)
frame_slider = gr.Slider(
minimum=0, maximum=100, value=0, step=1,
label="Frame Number"
)
preview_btn = gr.Button(
"🔍 Preview Detection",
variant="secondary",
elem_classes=["preview-btn"]
)
# Output Video
with gr.Tab("🎬 Result"):
video_output = gr.Video(
label="Blurred Video",
)
status_text = gr.Markdown(
"Ready to process",
elem_classes=["status-text"]
)
# Process Button
process_btn = gr.Button(
"🚀 Process Video",
variant="primary",
size="lg",
elem_classes=["process-btn"]
)
# ── Usage Guide ──
with gr.Accordion("📖 How to Use", open=False):
gr.Markdown("""
### Quick Start
1. **Upload** a video file (MP4, AVI, MOV, MKV supported)
2. **Type** the regions you want to blur in the text prompt
3. **Click** "Preview Detection" to check if the AI found the right objects
4. **Click** "Process Video" to apply the blur
5. **Download** your blurred video
### Tips for Better Results
- Use **specific** descriptions: "face" works better than "person's head"
- Separate multiple objects with **periods**: `face. license plate. text.`
- Lower the **detection threshold** if objects aren't being found
- Use **SAM 2 Video Tracking** mode for smooth, consistent blur across frames
- Use **Frame-by-Frame** mode if objects appear/disappear frequently
### Processing Modes
- **SAM 2 Video Tracking**: Detects on first frame, then tracks objects through the entire video using SAM 2's memory mechanism. Best for consistent results.
- **Frame-by-Frame**: Runs detection periodically and reuses masks between keyframes. Faster but less temporally consistent.
### Running on ZeroGPU
- The first run loads the models onto the GPU, so it's a bit slower; later runs are faster.
- Each run gets the GPU for a limited time, so clips are processed at a reduced resolution and capped in length by default for speed.
- **Long video?** Open **✂️ Trim** to pick a segment. Videos over the limit are trimmed automatically (to the first part, or your chosen start).
- Want longer / higher-res? Upgrade the Space GPU and raise `max_video_duration`, `max_resolution` and the `@spaces.GPU(duration=...)` budget.
""")
# ── Event Handlers ──
# Video upload → show info + set up trim controls
video_input.change(
fn=get_video_info_text,
inputs=[video_input],
outputs=[
video_info_text, frame_slider,
trim_enable, trim_start, trim_length, trim_info,
]
)
# Trim preview button
trim_preview_btn.click(
fn=preview_trim,
inputs=[video_input, trim_start, trim_length],
outputs=[trim_preview_video]
)
# Preview button
preview_btn.click(
fn=preview_detection,
inputs=[video_input, text_prompt, frame_slider],
outputs=[preview_image]
)
# Process button
process_btn.click(
fn=process_video,
inputs=[
video_input, text_prompt, blur_type, blur_strength,
edge_feather, processing_mode, keyframe_interval,
detection_threshold, model_quality,
trim_enable, trim_start, trim_length
],
outputs=[video_output, status_text]
)
return app
# ─── Entry Point ───────────────────────────────────────────────
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Video Blur AI")
parser.add_argument("--port", type=int, default=7860, help="Server port")
parser.add_argument("--share", action="store_true", help="Create public link")
parser.add_argument("--host", type=str, default="0.0.0.0", help="Server host")
args = parser.parse_args()
print("\n" + "=" * 60)
print(" 🎬 Video Blur AI - Starting...")
print("=" * 60)
app = create_app()
app.queue() # required for long jobs + progress + ZeroGPU scheduling
app.launch(
server_name=args.host,
server_port=args.port,
share=args.share,
show_api=False, # no public API page; also avoids schema edge cases
)