import json
import os
import sys
import tempfile
import gradio as gr
import numpy as np
import spaces
from huggingface_hub import snapshot_download
MODEL_REPO = os.environ.get("CAMERAVLM_REPO", "NicolasCC/CameraVLM-9B")
MAX_FRAMES = 128
CHAT_HEIGHT = 380
GEOMETRY_FIELDS = ("extrinsics", "camera_to_world", "intrinsics", "pose_encoding", "tracks", "track_visibility")
# Weights and the `cameravlm` package live in a private model repo and are downloaded at runtime
# (never at build time), so they are not part of this Space's files or its Docker image.
model_dir = snapshot_download(MODEL_REPO, token=os.environ["HF_TOKEN"])
sys.path.insert(0, model_dir)
from cameravlm import CAMERA_MOTION_PROMPT, CameraVLM # noqa: E402
model = CameraVLM.from_pretrained(model_dir, device="cuda", attn_implementation="sdpa")
def gpu_duration(video, prompt, num_frames):
return int(30 + 0.25 * num_frames) # ~60 s at 128 frames; measured ~15 s end to end on ZeroGPU
@spaces.GPU(duration=gpu_duration)
def run_model(video, prompt, num_frames):
return model(video, prompt=prompt, num_frames=num_frames)
def infer(video, task, prompt, num_frames):
if not video:
raise gr.Error("Please upload a video first.")
if task == "camera_motion":
prompt = CAMERA_MOTION_PROMPT
elif not (prompt or "").strip():
raise gr.Error("Please enter a question.")
try:
return run_model(video, prompt, max(2, min(int(num_frames), MAX_FRAMES)))
except Exception as error:
# Log only the error type and message: tracebacks would expose source lines of the private code.
print(f"Inference failed: {type(error).__name__}: {error}", flush=True)
raise gr.Error("Inference failed on this video.") from None
def predict(video, task, prompt, num_frames):
"""API endpoint: the answer plus camera poses and point tracks as an .npz file."""
result = infer(video, task, prompt, num_frames)
path = tempfile.NamedTemporaryFile(prefix="cameravlm_", suffix=".npz", delete=False).name
groups = [list(group) + [group[-1]] * (2 - len(group)) for group in result.track_group_frames] # odd frame count
np.savez_compressed(
path,
**{name: np.asarray(getattr(result, name), dtype=np.float32) for name in GEOMETRY_FIELDS},
frame_indices=np.asarray(result.frame_indices), timestamps=np.asarray(result.timestamps),
image_size=np.asarray(result.image_size), track_group_frames=np.asarray(groups),
track_window_starts=np.asarray(result.track_window_starts),
)
return result.text, path
# ---------- chat UI ----------
def show_chat(history):
"""The upload box is shown until a video is attached; afterwards the conversation is shown."""
return gr.Chatbot(value=history, visible=bool(history)), gr.Video(value=None, visible=not history)
def upload_video(history, video):
if not video:
return (*show_chat(history), None)
history = history + [{"role": "user", "content": {"path": video}}]
return (*show_chat(history), video)
def attach_files(history, video, message):
"""Appends a video attached in the input box (if any); a new video replaces the current one."""
files = (message or {}).get("files") or []
if files:
video = files[-1]
history = history + [{"role": "user", "content": {"path": video}}]
return history, video, bool(files)
def add_message(history, video, message):
history, video, attached = attach_files(history, video, message)
text, request = ((message or {}).get("text") or "").strip(), None
if text:
if video:
history = history + [{"role": "user", "content": text}]
request = ("question", text)
else:
gr.Warning("Please upload a video first.")
return (*show_chat(history), video, request, gr.MultimodalTextbox(value=None if request or attached else message))
def add_camera_motion(history, video, message):
history, video, attached = attach_files(history, video, message)
if not video:
gr.Warning("Please upload a video first.")
return (*show_chat(history), video, None, gr.MultimodalTextbox(value=message))
history = history + [{"role": "user", "content": "Describe the camera motion."}]
# Keep any typed question in the box; only the attached file is consumed.
keep = {"text": (message or {}).get("text") or "", "files": []} if attached else message
return (*show_chat(history), video, ("camera_motion", ""), gr.MultimodalTextbox(value=keep))
def respond(history, video, request, num_frames):
if request is None:
return history
task, prompt = request
try:
text = infer(video, task, prompt, num_frames).text
except gr.Error as error:
text = f"**Error:** {error.message}"
else:
if task == "camera_motion":
try:
parsed = json.loads(text[text.index("{"):text.rindex("}") + 1])
text = "```json\n" + json.dumps(parsed, indent=2, ensure_ascii=False) + "\n```"
except ValueError:
pass
return history + [{"role": "assistant", "content": text}]
def drop_last_answer(history):
while history and history[-1]["role"] == "assistant":
history = history[:-1]
return history
DESCRIPTION = """
"""
NOTE = """
Each question is answered from the most recently attached video; attach another one with the paperclip to switch.
Log in to Hugging Face for more GPU quota. The API also returns per-frame camera poses and point tracks: see
Use via API at the bottom of the page.
"""
DEFAULT_PROMPT = "How does the camera move?"
API_DESCRIPTION = (
"Answers a question about a video. Returns (answer, geometry): answer is the text answer, a JSON string "
"when task=\"camera_motion\"; geometry is the path of an .npz with per-frame camera poses and point tracks "
"(fields: GitHub README)."
)
API_DOCS = {
"video": "Video to analyse: a local path or URL wrapped in handle_file() (.mp4, .mov, ...; up to 200 MB).",
"task": '"question" answers prompt; "camera_motion" returns a JSON camera-motion description and ignores prompt.',
"prompt": f'Question about the video (default: "{DEFAULT_PROMPT}"). Ignored when task="camera_motion".',
"num_frames": "Frames sampled uniformly from the video, 2-128. Fewer frames run faster and use less GPU quota.",
}
def document_api(component, text):
"""Replaces Gradio's generic per-parameter text in the 'Use via API' page with a real description."""
info = {**component.api_info_as_input(), "additional_description": text}
component.api_info_as_input = lambda: info
CSS = """
#chat video {max-height: 180px; width: auto !important; max-width: 320px;}
.header {text-align: center;}
.header h1 {font-size: 1.9rem; line-height: 1.25; margin: 0.4rem 0 0.5rem;}
.header .tagline {font-size: 1.05rem; opacity: 0.85; margin: 0 0 0.4rem;}
.note {font-size: 0.85rem; opacity: 0.8;}
"""
# System fonts only: no requests to external font servers.
THEME = gr.themes.Default(
font=["ui-sans-serif", "system-ui", "-apple-system", "Segoe UI", "Helvetica", "Arial", "sans-serif"],
font_mono=["ui-monospace", "SFMono-Regular", "Menlo", "Consolas", "monospace"],
)
with gr.Blocks(title="CameraVLM", delete_cache=(3600, 3600)) as demo:
gr.HTML(DESCRIPTION)
video_state = gr.State(None)
request_state = gr.State(None)
upload = gr.Video(label="Click or drop a video here", sources=["upload"], height=CHAT_HEIGHT)
chatbot = gr.Chatbot(show_label=False, height=CHAT_HEIGHT, visible=False, elem_id="chat")
message = gr.MultimodalTextbox(
show_label=False, file_types=["video"], file_count="single", sources=["upload"],
placeholder="Ask a question about the video, e.g. How does the camera move relative to the car?",
)
with gr.Row():
motion_btn = gr.Button("Describe camera motion", variant="primary")
regen_btn = gr.Button("Regenerate")
clear_btn = gr.Button("Clear")
with gr.Accordion("Advanced", open=False):
num_frames = gr.Slider(2, MAX_FRAMES, value=MAX_FRAMES, step=2, label="Frames sampled uniformly",
info="Fewer frames run faster and use less GPU quota.")
gr.HTML(NOTE)
private = {"api_visibility": "private"}
views = [chatbot, upload]
respond_inputs = [chatbot, video_state, request_state, num_frames]
upload.upload(upload_video, [chatbot, upload], [*views, video_state], **private)
message.submit(add_message, [chatbot, video_state, message], [*views, video_state, request_state, message],
**private).then(respond, respond_inputs, chatbot, **private)
motion_btn.click(add_camera_motion, [chatbot, video_state, message],
[*views, video_state, request_state, message], **private).then(
respond, respond_inputs, chatbot, **private)
regen_btn.click(drop_last_answer, chatbot, chatbot, **private).then(respond, respond_inputs, chatbot, **private)
for trigger in (clear_btn.click, chatbot.clear):
trigger(lambda: (*show_chat([]), None, None), None, [*views, video_state, request_state], **private)
# Programmatic API (hidden from the page): predict(video, task, prompt, num_frames) -> (answer, geometry)
with gr.Group(visible=False):
api_video = gr.Video(label="video")
api_task = gr.Radio(["question", "camera_motion"], value="question", label="task")
api_prompt = gr.Textbox(value=DEFAULT_PROMPT, label="prompt")
api_frames = gr.Number(value=MAX_FRAMES, precision=0, minimum=2, maximum=MAX_FRAMES, label="num_frames")
api_answer = gr.Textbox(label="answer")
api_geometry = gr.File(label="geometry")
api_btn = gr.Button()
for component in (api_video, api_task, api_prompt, api_frames):
document_api(component, API_DOCS[component.label])
api_btn.click(predict, [api_video, api_task, api_prompt, api_frames], [api_answer, api_geometry],
api_name="predict", api_description=API_DESCRIPTION)
demo.queue(max_size=20).launch(
theme=THEME,
css=CSS,
max_file_size="200mb",
blocked_paths=[model_dir],
show_error=False,
)