app.py

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 = """
<div class="header">
  <h1>When Left Is Right? Camera-Motion Understanding beyond Screen</h1>
  <p class="tagline">CameraVLM tells how the camera moves in a video, answers questions about it,
  and describes it as structured shot metadata.</p>
</div>
"""

NOTE = """
<p class="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
<i>Use via API</i> at the bottom of the page.
</p>
"""

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,
)