app.py

# ---------------------------------------------------------------------------
# Silence harmless asyncio "Invalid file descriptor: -1" warnings that occur
# when Python's GC finalises temporary event loops created by dependencies
# (gradio, spaces SDK, httpx/httpcore) during import.  The __del__ on
# BaseEventLoop tries to close an already-closed selector and raises
# ValueError, which Python logs as "Exception ignored".  We wrap __del__ to
# swallow that specific error since it has zero runtime impact.
# ---------------------------------------------------------------------------
import asyncio as _asyncio  # noqa: E402

_orig_del = getattr(_asyncio.BaseEventLoop, "__del__", None)

if _orig_del is not None:
    def _quiet_del(self):  # type: ignore[no-untyped-def]
        try:
            _orig_del(self)
        except (ValueError, OSError):
            pass

    _asyncio.BaseEventLoop.__del__ = _quiet_del  # type: ignore[attr-defined]

import base64
import gc
import json
import math
import os
import tempfile
import traceback
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

import cv2
import gradio as gr
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from PIL import Image, ImageDraw, ImageFont

try:
    import spaces
except Exception:
    class _SpacesFallback:
        @staticmethod
        def GPU(*args, **kwargs):
            if args and callable(args[0]) and len(args) == 1 and not kwargs:
                return args[0]

            def decorator(fn):
                return fn

            return decorator

    spaces = _SpacesFallback()

try:
    from skimage.metrics import structural_similarity as structural_similarity
except Exception:
    structural_similarity = None

os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

APP_TITLE = "Temporal Difference Lab"
SPACE_REPO_ID = "charbel-malo/temporal-difference-lab"
VLM_MODEL_ID = os.getenv("VLM_MODEL_ID", "OpenGVLab/InternVL3_5-30B-A3B")
VJEPA_MODEL_ID = os.getenv("VJEPA_MODEL_ID", "facebook/vjepa2-vitl-fpc64-256")
SKIP_MODEL_LOAD = os.getenv("SKIP_MODEL_LOAD", "0").lower() in {"1", "true", "yes"}
ENABLE_VLM_RUNTIME = os.getenv("ENABLE_VLM", "1").lower() not in {"0", "false", "no"}
ENABLE_VLM_STARTUP = os.getenv("ENABLE_VLM_STARTUP", "0").lower() in {"1", "true", "yes"}
MAX_HARD_DURATION_SECONDS = 120
DEFAULT_MAX_DURATION_SECONDS = 60
DEFAULT_SAMPLE_FPS = 2.0
MAX_PREPROCESS_FRAMES = 240
MAX_VLM_EVENTS = 3
MAX_VLM_FRAMES_PER_EVENT = 8
RESIZE_LONG_EDGE = 768

PRESETS = [
    "Hail vs Pebbles",
    "Shader Artifact",
    "Particle Burst",
    "Animation Keyframe",
    "Camera Cut",
    "General Change",
]

ACCENT = "#3758f9"
WARNING = "#c75b3f"
TEXT = "#171717"
MUTED = "#6b7280"
GRID = "rgba(17, 24, 39, 0.12)"
SURFACE = "#ffffff"
SOFT = "#f7f7f5"

_VLM_PIPE = None
_VLM_MODEL = None
_VLM_PROCESSOR = None
_VLM_LOAD_ERROR = None


@dataclass
class VideoSamples:
    video_path: str
    fps: float
    duration: float
    source_width: int
    source_height: int
    frames: List[np.ndarray]
    timestamps: List[float]
    source_indices: List[int]


def _noop_gpu_duration(*args, **kwargs):
    return 1


@spaces.GPU(duration=1)
def zerogpu_healthcheck() -> str:
    return "ready"


def ensure_vlm_loaded() -> None:
    global _VLM_PIPE, _VLM_MODEL, _VLM_PROCESSOR, _VLM_LOAD_ERROR
    if SKIP_MODEL_LOAD:
        _VLM_LOAD_ERROR = "Model loading skipped by environment."
        return
    if not ENABLE_VLM_RUNTIME:
        _VLM_LOAD_ERROR = "InternVL runtime disabled by ENABLE_VLM=0."
        return
    if _VLM_MODEL is not None or _VLM_PIPE is not None:
        _VLM_LOAD_ERROR = None
        return

    try:
        import torch
        from transformers import pipeline

        dtype = torch.bfloat16 if torch.cuda.is_available() else torch.float32
        _VLM_PIPE = pipeline(
            "image-text-to-text",
            model=VLM_MODEL_ID,
            torch_dtype=dtype,
            device_map="auto",
            trust_remote_code=True,
        )
        _VLM_LOAD_ERROR = None
        return
    except Exception as pipe_error:
        try:
            import torch
            from transformers import AutoModelForImageTextToText, AutoProcessor

            dtype = torch.bfloat16 if torch.cuda.is_available() else torch.float32
            _VLM_PROCESSOR = AutoProcessor.from_pretrained(
                VLM_MODEL_ID,
                trust_remote_code=True,
            )
            _VLM_MODEL = AutoModelForImageTextToText.from_pretrained(
                VLM_MODEL_ID,
                torch_dtype=dtype,
                device_map="auto",
                trust_remote_code=True,
                low_cpu_mem_usage=True,
            ).eval()
            _VLM_LOAD_ERROR = None
            return
        except Exception as direct_error:
            _VLM_LOAD_ERROR = (
                "InternVL load failed. Pipeline: "
                f"{type(pipe_error).__name__}: {pipe_error}. Direct loader: "
                f"{type(direct_error).__name__}: {direct_error}"
            )


def _load_vlm_at_startup() -> None:
    if ENABLE_VLM_STARTUP:
        ensure_vlm_loaded()
    elif SKIP_MODEL_LOAD:
        global _VLM_LOAD_ERROR
        _VLM_LOAD_ERROR = "Model loading skipped by environment."


_load_vlm_at_startup()


def clamp(value: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, value))


def safe_float(value: Any, default: float = 0.0) -> float:
    try:
        if value is None:
            return default
        if isinstance(value, float) and math.isnan(value):
            return default
        return float(value)
    except Exception:
        return default


def resize_long_edge(frame: np.ndarray, long_edge: int = RESIZE_LONG_EDGE) -> np.ndarray:
    height, width = frame.shape[:2]
    longest = max(height, width)
    if longest <= long_edge:
        return frame
    scale = long_edge / float(longest)
    new_size = (max(1, int(width * scale)), max(1, int(height * scale)))
    return cv2.resize(frame, new_size, interpolation=cv2.INTER_AREA)


def read_video_samples(
    video_path: str,
    max_duration_seconds: float = DEFAULT_MAX_DURATION_SECONDS,
    sample_fps: float = DEFAULT_SAMPLE_FPS,
    long_edge: int = RESIZE_LONG_EDGE,
) -> VideoSamples:
    if not video_path:
        raise ValueError("Upload a video file first.")

    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise ValueError("OpenCV could not open the uploaded video.")

    fps = safe_float(cap.get(cv2.CAP_PROP_FPS), 30.0)
    if fps <= 1e-6:
        fps = 30.0
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
    source_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
    source_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
    source_duration = frame_count / fps if frame_count else 0.0
    duration_limit = clamp(safe_float(max_duration_seconds, DEFAULT_MAX_DURATION_SECONDS), 1.0, MAX_HARD_DURATION_SECONDS)
    duration = min(source_duration if source_duration > 0 else duration_limit, duration_limit)
    target_fps = clamp(safe_float(sample_fps, DEFAULT_SAMPLE_FPS), 0.5, 4.0)
    step = max(1, int(round(fps / target_fps)))
    max_source_index = int(duration * fps) if duration > 0 else frame_count

    frames: List[np.ndarray] = []
    timestamps: List[float] = []
    source_indices: List[int] = []
    index = 0
    while True:
        ok, frame_bgr = cap.read()
        if not ok:
            break
        if index > max_source_index:
            break
        if index % step == 0:
            rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
            rgb = resize_long_edge(rgb, long_edge=long_edge)
            frames.append(rgb)
            timestamps.append(index / fps)
            source_indices.append(index)
            if len(frames) >= MAX_PREPROCESS_FRAMES:
                break
        index += 1
    cap.release()

    if len(frames) < 2:
        raise ValueError("Need at least two sampled frames to analyze temporal change.")

    return VideoSamples(
        video_path=video_path,
        fps=fps,
        duration=duration,
        source_width=source_width,
        source_height=source_height,
        frames=frames,
        timestamps=timestamps,
        source_indices=source_indices,
    )


def gray_frame(frame: np.ndarray) -> np.ndarray:
    return cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)


def downscale_for_flow(gray: np.ndarray, max_edge: int = 384) -> np.ndarray:
    height, width = gray.shape[:2]
    longest = max(height, width)
    if longest <= max_edge:
        return gray
    scale = max_edge / float(longest)
    return cv2.resize(gray, (int(width * scale), int(height * scale)), interpolation=cv2.INTER_AREA)


def fallback_ssim_loss(prev_gray: np.ndarray, next_gray: np.ndarray) -> float:
    prev = prev_gray.astype(np.float32) / 255.0
    nxt = next_gray.astype(np.float32) / 255.0
    mse = float(np.mean((prev - nxt) ** 2))
    return clamp(mse * 4.0, 0.0, 1.0)


def pair_metrics(prev_rgb: np.ndarray, next_rgb: np.ndarray) -> Dict[str, float]:
    prev_gray = gray_frame(prev_rgb)
    next_gray = gray_frame(next_rgb)
    absdiff = cv2.absdiff(prev_rgb, next_rgb)
    graydiff = cv2.absdiff(prev_gray, next_gray)

    if structural_similarity is not None:
        try:
            similarity = structural_similarity(prev_gray, next_gray, data_range=255)
            ssim_loss = 1.0 - float(similarity)
        except Exception:
            ssim_loss = fallback_ssim_loss(prev_gray, next_gray)
    else:
        ssim_loss = fallback_ssim_loss(prev_gray, next_gray)

    prev_small = downscale_for_flow(prev_gray)
    next_small = downscale_for_flow(next_gray)
    flow = cv2.calcOpticalFlowFarneback(
        prev_small,
        next_small,
        None,
        0.5,
        3,
        15,
        3,
        5,
        1.2,
        0,
    )
    mag, ang = cv2.cartToPolar(flow[..., 0], flow[..., 1])
    hsv_prev = cv2.cvtColor(prev_rgb, cv2.COLOR_RGB2HSV)
    hsv_next = cv2.cvtColor(next_rgb, cv2.COLOR_RGB2HSV)
    edges_prev = cv2.Canny(prev_gray, 80, 160)
    edges_next = cv2.Canny(next_gray, 80, 160)
    edge_delta = float(np.mean(cv2.absdiff(edges_prev, edges_next))) / 255.0

    return {
        "frame_diff": float(np.mean(absdiff)) / 255.0,
        "ssim_loss": clamp(ssim_loss, 0.0, 1.0),
        "flow_mean": float(np.mean(mag)),
        "flow_p95": float(np.percentile(mag, 95)),
        "flow_angle_std": float(np.std(ang)),
        "edge_delta": edge_delta,
        "color_delta": float(np.mean(cv2.absdiff(hsv_prev, hsv_next))) / 255.0,
        "intensity_delta": float(abs(np.mean(prev_gray) - np.mean(next_gray))) / 255.0,
        "psnr_proxy_loss": float(np.mean(graydiff.astype(np.float32) ** 2)) / (255.0 ** 2),
    }


def robust_normalize(values: Sequence[float]) -> List[float]:
    if not values:
        return []
    arr = np.asarray(values, dtype=np.float32)
    hi = float(np.percentile(arr, 95))
    if hi <= 1e-8:
        hi = float(np.max(arr))
    if hi <= 1e-8:
        return [0.0 for _ in values]
    return [float(clamp(v / hi, 0.0, 1.5)) for v in arr]


def detect_scene_cuts(video_path: str) -> List[float]:
    try:
        from scenedetect import ContentDetector, SceneManager, open_video

        video = open_video(video_path)
        scene_manager = SceneManager()
        scene_manager.add_detector(ContentDetector(threshold=27.0))
        scene_manager.detect_scenes(video)
        scenes = scene_manager.get_scene_list()
        return [scene[0].get_seconds() for scene in scenes[1:]]
    except Exception:
        return []


def compute_temporal_series(samples: VideoSamples) -> List[Dict[str, Any]]:
    raw_metrics = []
    for index in range(1, len(samples.frames)):
        metrics = pair_metrics(samples.frames[index - 1], samples.frames[index])
        metrics["timestamp"] = samples.timestamps[index]
        metrics["sample_index"] = index
        metrics["source_frame"] = samples.source_indices[index]
        raw_metrics.append(metrics)

    keys = [
        "frame_diff",
        "ssim_loss",
        "flow_p95",
        "edge_delta",
        "color_delta",
        "intensity_delta",
        "psnr_proxy_loss",
    ]
    normalized = {key: robust_normalize([row[key] for row in raw_metrics]) for key in keys}
    scene_cuts = detect_scene_cuts(samples.video_path)
    tolerance = max(0.25, 1.0 / max(DEFAULT_SAMPLE_FPS, 1.0))

    series = []
    for idx, row in enumerate(raw_metrics):
        scene_cut = any(abs(row["timestamp"] - cut) <= tolerance for cut in scene_cuts)
        score = (
            0.23 * normalized["frame_diff"][idx]
            + 0.17 * normalized["ssim_loss"][idx]
            + 0.25 * normalized["flow_p95"][idx]
            + 0.13 * normalized["edge_delta"][idx]
            + 0.08 * normalized["color_delta"][idx]
            + 0.07 * normalized["intensity_delta"][idx]
            + 0.07 * normalized["psnr_proxy_loss"][idx]
        )
        if scene_cut:
            score = max(score, 1.15)
        event_row = dict(row)
        event_row.update(
            {
                "score": float(score),
                "scene_cut": bool(scene_cut),
                "norm_frame_diff": normalized["frame_diff"][idx],
                "norm_ssim_loss": normalized["ssim_loss"][idx],
                "norm_flow_p95": normalized["flow_p95"][idx],
                "norm_edge_delta": normalized["edge_delta"][idx],
                "norm_color_delta": normalized["color_delta"][idx],
                "norm_intensity_delta": normalized["intensity_delta"][idx],
            }
        )
        series.append(event_row)
    return series


def evidence_type(row: Dict[str, Any]) -> str:
    if row.get("scene_cut"):
        return "Camera cut / scene boundary"
    flow = row.get("norm_flow_p95", 0.0)
    diff = row.get("norm_frame_diff", 0.0)
    edge = row.get("norm_edge_delta", 0.0)
    color = row.get("norm_color_delta", 0.0)
    intensity = row.get("norm_intensity_delta", 0.0)
    if intensity > 0.85 and flow < 0.55:
        return "Lighting / exposure change"
    if flow > 0.75 and diff > 0.55:
        return "Motion field change"
    if edge > 0.75 and color > 0.45:
        return "Shader / surface artifact"
    if diff > 0.75 and flow < 0.45:
        return "Appearance change"
    if flow > 0.65:
        return "Subtle motion change"
    return "Low-amplitude temporal change"


def select_events(series: List[Dict[str, Any]], sensitivity: float, max_events: int) -> List[Dict[str, Any]]:
    if not series:
        return []
    sensitivity = clamp(safe_float(sensitivity, 0.55), 0.1, 1.0)
    max_events = int(clamp(safe_float(max_events, 5), 1, 5))
    scores = np.asarray([row["score"] for row in series], dtype=np.float32)
    absolute_floor = 0.10 + (1.0 - sensitivity) * 0.22
    quantile = 88.0 - sensitivity * 28.0
    adaptive_floor = float(np.percentile(scores, quantile))
    threshold = max(absolute_floor, adaptive_floor)
    candidates = [row for row in series if row["score"] >= threshold]
    candidates.sort(key=lambda row: row["score"], reverse=True)

    selected: List[Dict[str, Any]] = []
    min_separation = 0.65
    for row in candidates:
        if all(abs(row["timestamp"] - other["timestamp"]) >= min_separation for other in selected):
            event = dict(row)
            event["event_id"] = len(selected) + 1
            event["evidence_type"] = evidence_type(row)
            selected.append(event)
        if len(selected) >= max_events:
            break

    selected.sort(key=lambda row: row["timestamp"])
    for index, row in enumerate(selected, start=1):
        row["event_id"] = index
    return selected


def nearest_sample_index(samples: VideoSamples, timestamp: float) -> int:
    distances = [abs(ts - timestamp) for ts in samples.timestamps]
    return int(np.argmin(distances))


def event_frame_indices(samples: VideoSamples, event: Dict[str, Any], max_frames: int = MAX_VLM_FRAMES_PER_EVENT) -> List[int]:
    center = nearest_sample_index(samples, event["timestamp"])
    offsets = [-5, -3, -2, -1, 0, 1, 2, 3, 5]
    indices = []
    for offset in offsets:
        idx = center + offset
        if 0 <= idx < len(samples.frames) and idx not in indices:
            indices.append(idx)
    if len(indices) > max_frames:
        keep = np.linspace(0, len(indices) - 1, max_frames).round().astype(int).tolist()
        indices = [indices[i] for i in keep]
    return indices


def annotate_image(frame: np.ndarray, text: str) -> Image.Image:
    image = Image.fromarray(frame)
    draw = ImageDraw.Draw(image)
    try:
        font = ImageFont.truetype("Arial.ttf", 18)
    except Exception:
        font = ImageFont.load_default()
    pad = 8
    bbox = draw.textbbox((pad, pad), text, font=font)
    draw.rectangle(
        [bbox[0] - 5, bbox[1] - 3, bbox[2] + 5, bbox[3] + 4],
        fill=(255, 255, 255, 225),
    )
    draw.text((pad, pad), text, fill=(17, 17, 17), font=font)
    return image


def make_grid(images: List[Image.Image], columns: int = 4, gap: int = 8, fill: Tuple[int, int, int] = (247, 247, 245)) -> Image.Image:
    if not images:
        return Image.new("RGB", (320, 180), fill)
    width = max(img.width for img in images)
    height = max(img.height for img in images)
    rows = int(math.ceil(len(images) / float(columns)))
    canvas = Image.new("RGB", (columns * width + (columns - 1) * gap, rows * height + (rows - 1) * gap), fill)
    for i, img in enumerate(images):
        row = i // columns
        col = i % columns
        canvas.paste(img.resize((width, height)), (col * (width + gap), row * (height + gap)))
    return canvas


def diff_heatmap(prev_rgb: np.ndarray, next_rgb: np.ndarray) -> Image.Image:
    diff = cv2.absdiff(prev_rgb, next_rgb)
    diff_gray = cv2.cvtColor(diff, cv2.COLOR_RGB2GRAY)
    colored = cv2.applyColorMap(diff_gray, cv2.COLORMAP_INFERNO)
    return Image.fromarray(cv2.cvtColor(colored, cv2.COLOR_BGR2RGB))


def flow_visualization(prev_rgb: np.ndarray, next_rgb: np.ndarray) -> Image.Image:
    prev_gray = downscale_for_flow(gray_frame(prev_rgb), max_edge=512)
    next_gray = downscale_for_flow(gray_frame(next_rgb), max_edge=512)
    flow = cv2.calcOpticalFlowFarneback(prev_gray, next_gray, None, 0.5, 3, 15, 3, 5, 1.2, 0)
    mag, ang = cv2.cartToPolar(flow[..., 0], flow[..., 1])
    hsv = np.zeros((prev_gray.shape[0], prev_gray.shape[1], 3), dtype=np.uint8)
    hsv[..., 0] = (ang * 180 / np.pi / 2).astype(np.uint8)
    hsv[..., 1] = 210
    norm_mag = cv2.normalize(mag, None, 0, 255, cv2.NORM_MINMAX)
    hsv[..., 2] = norm_mag.astype(np.uint8)
    bgr = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)
    return Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB))


def save_event_artifacts(samples: VideoSamples, events: List[Dict[str, Any]], output_dir: Path) -> Tuple[List[Tuple[str, str]], List[Dict[str, Any]]]:
    gallery: List[Tuple[str, str]] = []
    payloads: List[Dict[str, Any]] = []
    output_dir.mkdir(parents=True, exist_ok=True)
    for event in events:
        indices = event_frame_indices(samples, event)
        annotated = [
            annotate_image(samples.frames[idx], f"#{event['event_id']} t={samples.timestamps[idx]:.2f}s")
            for idx in indices
        ]
        frame_grid = make_grid(annotated, columns=min(4, max(1, len(annotated))))
        grid_path = output_dir / f"event_{event['event_id']}_frames.jpg"
        frame_grid.save(grid_path, quality=92)
        gallery.append((str(grid_path), f"Event {event['event_id']} keyframes at {event['timestamp']:.2f}s"))

        center_idx = nearest_sample_index(samples, event["timestamp"])
        prev_idx = max(0, center_idx - 1)
        next_idx = min(len(samples.frames) - 1, center_idx)
        diff_path = output_dir / f"event_{event['event_id']}_diff.jpg"
        flow_path = output_dir / f"event_{event['event_id']}_flow.jpg"
        diff_heatmap(samples.frames[prev_idx], samples.frames[next_idx]).save(diff_path, quality=92)
        flow_visualization(samples.frames[prev_idx], samples.frames[next_idx]).save(flow_path, quality=92)
        gallery.append((str(diff_path), f"Event {event['event_id']} frame-difference map"))
        gallery.append((str(flow_path), f"Event {event['event_id']} optical-flow field"))

        raw_paths = []
        for frame_idx in indices[:MAX_VLM_FRAMES_PER_EVENT]:
            raw_path = output_dir / f"event_{event['event_id']}_frame_{frame_idx}.jpg"
            Image.fromarray(samples.frames[frame_idx]).save(raw_path, quality=90)
            raw_paths.append(str(raw_path))

        payload = {
            "event_id": event["event_id"],
            "timestamp": event["timestamp"],
            "evidence_type": event["evidence_type"],
            "score": event["score"],
            "metrics": {key: event.get(key) for key in event if key.startswith("norm_") or key in {"frame_diff", "ssim_loss", "flow_p95", "edge_delta", "color_delta", "intensity_delta", "scene_cut"}},
            "frame_paths": raw_paths,
            "diff_path": str(diff_path),
            "flow_path": str(flow_path),
        }
        payloads.append(payload)
    return gallery, payloads


def cpu_temporal_feature_evidence(series: List[Dict[str, Any]], events: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not series:
        return {"mode": "deterministic", "summary": "No temporal series available.", "events": []}
    scores = np.asarray([row["score"] for row in series], dtype=np.float32)
    median = float(np.median(scores))
    p90 = float(np.percentile(scores, 90))
    event_evidence = []
    for event in events:
        novelty = (event["score"] - median) / max(p90 - median, 1e-6)
        if event.get("norm_flow_p95", 0.0) > 0.7:
            scale = "motion-dominant temporal feature"
        elif event.get("norm_intensity_delta", 0.0) > 0.75:
            scale = "global intensity feature"
        elif event.get("norm_edge_delta", 0.0) > 0.7:
            scale = "edge/surface feature"
        else:
            scale = "low-amplitude temporal feature"
        event_evidence.append(
            {
                "event_id": event["event_id"],
                "embedding_source": "deterministic-proxy",
                "novelty": round(float(clamp(novelty, 0.0, 3.0)), 3),
                "temporal_label": scale,
            }
        )
    return {
        "mode": "deterministic-proxy",
        "summary": "V-JEPA is disabled or unavailable; deterministic metrics are serialized as JEPA-style temporal evidence.",
        "baseline_score_median": round(median, 4),
        "baseline_score_p90": round(p90, 4),
        "events": event_evidence,
    }


def estimate_jepa_duration(frame_paths: List[str], *args, **kwargs) -> int:
    return int(clamp(45 + len(frame_paths) * 2, 60, 240))


@spaces.GPU(size="large", duration=estimate_jepa_duration)
def extract_vjepa_evidence_gpu(frame_paths: List[str], cpu_evidence: Dict[str, Any]) -> Dict[str, Any]:
    if not frame_paths:
        return cpu_evidence
    try:
        import torch
        from transformers import AutoModel, AutoVideoProcessor

        processor = AutoVideoProcessor.from_pretrained(VJEPA_MODEL_ID)
        model = AutoModel.from_pretrained(VJEPA_MODEL_ID).eval()
        if torch.cuda.is_available():
            model = model.to("cuda")

        frames = [Image.open(path).convert("RGB") for path in frame_paths[:64]]
        if len(frames) < 64:
            frames = frames + [frames[-1]] * (64 - len(frames))
        inputs = processor(videos=[frames], return_tensors="pt")
        if torch.cuda.is_available():
            inputs = {key: value.to("cuda") for key, value in inputs.items()}
        with torch.no_grad():
            outputs = model(**inputs)
        hidden = getattr(outputs, "last_hidden_state", None)
        if hidden is None:
            hidden = outputs[0]
        pooled = hidden.float().mean(dim=1).detach().cpu().numpy()[0]
        magnitude = float(np.linalg.norm(pooled) / max(len(pooled), 1))
        evidence = dict(cpu_evidence)
        evidence["mode"] = "vjepa2"
        evidence["model"] = VJEPA_MODEL_ID
        evidence["summary"] = "V-JEPA 2 embedding extracted successfully; novelty is combined with deterministic event evidence."
        evidence["embedding_norm_per_dim"] = round(magnitude, 6)
        del model
        gc.collect()
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
        return evidence
    except Exception as exc:
        evidence = dict(cpu_evidence)
        evidence["mode"] = "deterministic-proxy"
        evidence["vjepa_error"] = f"{type(exc).__name__}: {exc}"
        return evidence


def format_metric(value: Any) -> str:
    return f"{safe_float(value):.3f}"


def build_prompt(event_payload: Dict[str, Any], preset: str, question: str, jepa_evidence: Dict[str, Any]) -> str:
    metrics = event_payload.get("metrics", {})
    event_id = event_payload["event_id"]
    matching_jepa = next(
        (row for row in jepa_evidence.get("events", []) if row.get("event_id") == event_id),
        {},
    )
    return f"""
You are analyzing a short video segment for temporal change. Use the images in order:
1. sampled keyframes around the event,
2. a frame-difference heatmap,
3. an optical-flow visualization.

Task preset: {preset}
User question: {question or "What changed over time, and what visual evidence supports it?"}

Event timestamp: {event_payload['timestamp']:.2f}s
Deterministic evidence type: {event_payload['evidence_type']}
Normalized metrics:
- overall score: {event_payload['score']:.3f}
- frame diff: {format_metric(metrics.get('norm_frame_diff'))}
- SSIM loss: {format_metric(metrics.get('norm_ssim_loss'))}
- optical flow p95: {format_metric(metrics.get('norm_flow_p95'))}
- edge delta: {format_metric(metrics.get('norm_edge_delta'))}
- color delta: {format_metric(metrics.get('norm_color_delta'))}
- intensity delta: {format_metric(metrics.get('norm_intensity_delta'))}
- scene cut: {metrics.get('scene_cut')}

Temporal embedding evidence:
{json.dumps(matching_jepa, indent=2)}

Answer as compact JSON with keys:
verdict, timestamp_seconds, visual_change, evidence, uncertainty, caveat.
Do not claim certainty if the evidence is only frame differencing or flow.
""".strip()


def estimate_vlm_duration(event_payloads: List[Dict[str, Any]], *args, **kwargs) -> int:
    return int(clamp(120 + len(event_payloads) * 70, 180, 600))


def _run_pipeline_vlm(images: List[Image.Image], prompt: str) -> str:
    content = [{"type": "image", "image": image} for image in images]
    content.append({"type": "text", "text": prompt})
    output = _VLM_PIPE(text=[{"role": "user", "content": content}], max_new_tokens=384)
    if isinstance(output, list) and output:
        item = output[0]
        if isinstance(item, dict):
            return str(item.get("generated_text") or item.get("text") or item)
    return str(output)


def _run_processor_vlm(images: List[Image.Image], prompt: str) -> str:
    import torch

    content = [{"type": "image", "image": image} for image in images]
    content.append({"type": "text", "text": prompt})
    messages = [{"role": "user", "content": content}]
    inputs = _VLM_PROCESSOR.apply_chat_template(
        messages,
        add_generation_prompt=True,
        tokenize=True,
        return_dict=True,
        return_tensors="pt",
    )
    device = next(_VLM_MODEL.parameters()).device
    inputs = {key: value.to(device) for key, value in inputs.items()}
    with torch.no_grad():
        output_ids = _VLM_MODEL.generate(**inputs, max_new_tokens=384, do_sample=False)
    input_len = inputs["input_ids"].shape[-1]
    return _VLM_PROCESSOR.decode(output_ids[0][input_len:], skip_special_tokens=True)


@spaces.GPU(size="xlarge", duration=estimate_vlm_duration)
def run_internvl_reasoning_gpu(
    event_payloads: List[Dict[str, Any]],
    preset: str,
    question: str,
    jepa_evidence: Dict[str, Any],
) -> List[Dict[str, Any]]:
    ensure_vlm_loaded()
    if _VLM_LOAD_ERROR:
        raise RuntimeError(_VLM_LOAD_ERROR)
    if _VLM_MODEL is None and _VLM_PIPE is None:
        raise RuntimeError("InternVL model is not loaded.")

    results = []
    for payload in event_payloads[:MAX_VLM_EVENTS]:
        image_paths = payload.get("frame_paths", [])[:MAX_VLM_FRAMES_PER_EVENT]
        image_paths = image_paths + [payload.get("diff_path"), payload.get("flow_path")]
        images = [Image.open(path).convert("RGB") for path in image_paths if path and Path(path).exists()]
        prompt = build_prompt(payload, preset, question, jepa_evidence)
        if _VLM_MODEL is not None and _VLM_PROCESSOR is not None:
            text = _run_processor_vlm(images, prompt)
        else:
            text = _run_pipeline_vlm(images, prompt)
        results.append({"event_id": payload["event_id"], "raw_response": text})
    return results


def parse_jsonish_response(text: str) -> Dict[str, Any]:
    try:
        start = text.find("{")
        end = text.rfind("}")
        if start >= 0 and end > start:
            return json.loads(text[start : end + 1])
    except Exception:
        pass
    return {
        "verdict": "Model response",
        "visual_change": text.strip()[:1200],
        "evidence": "InternVL returned free-form text.",
        "uncertainty": "unknown",
        "caveat": "Response was not valid JSON.",
    }


def fallback_event_reasoning(event: Dict[str, Any], preset: str) -> Dict[str, Any]:
    evidence = event.get("evidence_type", "Temporal change")
    flow = safe_float(event.get("norm_flow_p95"))
    diff = safe_float(event.get("norm_frame_diff"))
    intensity = safe_float(event.get("norm_intensity_delta"))
    if preset == "Hail vs Pebbles" and flow > 0.65 and diff > 0.45:
        verdict = "New falling or moving material is likely present."
        visual_change = "The event has a high optical-flow field and strong frame difference, which fits falling hail or moving particles more than static pebbles."
    elif "Shader" in preset or evidence == "Shader / surface artifact":
        verdict = "Surface or shader-state change is plausible."
        visual_change = "Edge and color deltas dominate the event, which can indicate specular, normal-map, texture, or lighting-state changes."
    elif intensity > 0.75:
        verdict = "Global lighting or exposure change."
        visual_change = "Intensity changes dominate while motion is weaker, so the change is likely global illumination, exposure, or flicker."
    elif event.get("scene_cut"):
        verdict = "Camera cut or scene boundary."
        visual_change = "The timestamp aligns with scene-boundary evidence."
    else:
        verdict = evidence
        visual_change = "The deterministic layer found measurable temporal change, but no VLM interpretation was available."
    return {
        "verdict": verdict,
        "timestamp_seconds": round(safe_float(event.get("timestamp")), 3),
        "visual_change": visual_change,
        "evidence": f"{evidence}; score={safe_float(event.get('score')):.3f}",
        "uncertainty": "medium" if safe_float(event.get("score")) < 0.8 else "low",
        "caveat": "CPU fallback reasoning; enable InternVL on ZeroGPU for semantic interpretation.",
    }


def empty_timeline(message: str = "Upload a video and run analysis.") -> go.Figure:
    fig = go.Figure()
    fig.add_annotation(text=message, showarrow=False, font={"size": 15, "color": MUTED})
    fig.update_layout(
        height=360,
        margin={"l": 42, "r": 22, "t": 42, "b": 46},
        paper_bgcolor=SURFACE,
        plot_bgcolor=SOFT,
        xaxis={"visible": False},
        yaxis={"visible": False},
    )
    return fig


def timeline_figure(series: List[Dict[str, Any]], events: List[Dict[str, Any]]) -> go.Figure:
    fig = go.Figure()
    x = [row["timestamp"] for row in series]
    y = [row["score"] for row in series]
    fig.add_trace(
        go.Scatter(
            x=x,
            y=y,
            mode="lines",
            name="Change score",
            line={"color": ACCENT, "width": 2},
            hovertemplate="t=%{x:.2f}s<br>score=%{y:.3f}<extra></extra>",
        )
    )
    if events:
        fig.add_trace(
            go.Scatter(
                x=[row["timestamp"] for row in events],
                y=[row["score"] for row in events],
                mode="markers+text",
                name="Selected events",
                marker={"color": WARNING, "size": 11, "line": {"width": 1, "color": "#fff"}},
                text=[f"#{row['event_id']}" for row in events],
                textposition="top center",
                hovertemplate="Event %{text}<br>t=%{x:.2f}s<br>score=%{y:.3f}<extra></extra>",
            )
        )
    fig.update_layout(
        title="Temporal Change Score",
        height=390,
        margin={"l": 52, "r": 22, "t": 58, "b": 54},
        paper_bgcolor=SURFACE,
        plot_bgcolor=SOFT,
        font={"color": TEXT, "family": "Inter, Arial, sans-serif"},
        legend={"orientation": "h", "yanchor": "bottom", "y": 1.02, "x": 0},
        xaxis={"title": "Time (seconds)", "gridcolor": GRID, "zeroline": False},
        yaxis={"title": "Normalized evidence score", "gridcolor": GRID, "zeroline": False},
    )
    return fig


def event_rows(events: List[Dict[str, Any]], interpretations: Dict[int, Dict[str, Any]]) -> pd.DataFrame:
    rows = []
    for event in events:
        interp = interpretations.get(event["event_id"], fallback_event_reasoning(event, "General Change"))
        rows.append(
            {
                "event": event["event_id"],
                "timestamp_s": round(safe_float(event["timestamp"]), 3),
                "score": round(safe_float(event["score"]), 3),
                "evidence_type": event["evidence_type"],
                "verdict": interp.get("verdict", ""),
                "uncertainty": interp.get("uncertainty", ""),
                "caveat": interp.get("caveat", ""),
            }
        )
    return pd.DataFrame(rows)


def summary_markdown(
    samples: Optional[VideoSamples],
    events: List[Dict[str, Any]],
    interpretations: Dict[int, Dict[str, Any]],
    jepa_evidence: Dict[str, Any],
    status: str,
) -> str:
    if samples is None:
        return f"### Ready\n{status}"
    if not events:
        return (
            "### No major temporal event detected\n"
            f"Analyzed {len(samples.frames)} sampled frames over {samples.duration:.2f}s. "
            "The deterministic score did not cross the selected sensitivity threshold.\n\n"
            f"**Status:** {status}"
        )
    first = interpretations.get(events[0]["event_id"], fallback_event_reasoning(events[0], "General Change"))
    bullets = []
    for event in events:
        interp = interpretations.get(event["event_id"], fallback_event_reasoning(event, "General Change"))
        bullets.append(
            f"- **Event {event['event_id']} at {event['timestamp']:.2f}s:** "
            f"{interp.get('verdict', event['evidence_type'])}. "
            f"{interp.get('visual_change', '')}"
        )
    return (
        f"### {first.get('verdict', 'Temporal change detected')}\n"
        f"Analyzed {len(samples.frames)} sampled frames from a {samples.source_width}x{samples.source_height} video. "
        f"Selected {len(events)} event(s) for evidence review.\n\n"
        + "\n".join(bullets)
        + "\n\n"
        f"**Temporal feature layer:** {jepa_evidence.get('summary', 'No JEPA evidence.')}\n\n"
        f"**Status:** {status}"
    )


def write_json_report(report: Dict[str, Any], output_dir: Path) -> str:
    path = output_dir / "temporal_difference_report.json"
    path.write_text(json.dumps(report, indent=2), encoding="utf-8")
    return str(path)


def run_analysis(
    video_path: Optional[str],
    task_preset: str,
    question: str,
    sensitivity: float,
    max_events: int,
    sample_fps: float,
    max_duration_seconds: float,
    use_jepa: bool,
    run_vlm: bool,
):
    if not video_path:
        return (
            "### Ready\nUpload a video to begin.",
            empty_timeline(),
            [],
            pd.DataFrame(),
            None,
            "{}",
        )

    output_dir = Path(tempfile.mkdtemp(prefix="tdl_"))
    samples = None
    try:
        samples = read_video_samples(video_path, max_duration_seconds, sample_fps)
        series = compute_temporal_series(samples)
        events = select_events(series, sensitivity, max_events)
        gallery, event_payloads = save_event_artifacts(samples, events, output_dir)
        cpu_evidence = cpu_temporal_feature_evidence(series, events)
        jepa_evidence = cpu_evidence

        if use_jepa and event_payloads:
            frame_paths = []
            for payload in event_payloads:
                frame_paths.extend(payload.get("frame_paths", []))
            jepa_evidence = extract_vjepa_evidence_gpu(frame_paths[:64], cpu_evidence)

        interpretations: Dict[int, Dict[str, Any]] = {}
        status_bits = ["Deterministic preprocessing complete."]
        if run_vlm and event_payloads:
            try:
                vlm_results = run_internvl_reasoning_gpu(
                    event_payloads[:MAX_VLM_EVENTS],
                    task_preset,
                    question,
                    jepa_evidence,
                )
                for item in vlm_results:
                    interpretations[int(item["event_id"])] = parse_jsonish_response(item.get("raw_response", ""))
                status_bits.append(f"InternVL reasoning complete using {VLM_MODEL_ID}.")
            except Exception as exc:
                status_bits.append(f"InternVL unavailable: {type(exc).__name__}: {exc}")

        for event in events:
            if event["event_id"] not in interpretations:
                interpretations[event["event_id"]] = fallback_event_reasoning(event, task_preset)

        report = {
            "app": APP_TITLE,
            "space_repo_id": SPACE_REPO_ID,
            "video": {
                "duration_analyzed_seconds": samples.duration,
                "source_width": samples.source_width,
                "source_height": samples.source_height,
                "native_fps": samples.fps,
                "sample_count": len(samples.frames),
            },
            "inputs": {
                "task_preset": task_preset,
                "question": question,
                "sensitivity": sensitivity,
                "max_events": max_events,
                "sample_fps": sample_fps,
                "max_duration_seconds": max_duration_seconds,
                "use_jepa": use_jepa,
                "run_vlm": run_vlm,
            },
            "temporal_feature_evidence": jepa_evidence,
            "events": events,
            "interpretations": interpretations,
        }
        json_path = write_json_report(report, output_dir)
        status = " ".join(status_bits)
        return (
            summary_markdown(samples, events, interpretations, jepa_evidence, status),
            timeline_figure(series, events),
            gallery,
            event_rows(events, interpretations),
            json_path,
            json.dumps(report, indent=2),
        )
    except Exception as exc:
        report = {
            "app": APP_TITLE,
            "error": f"{type(exc).__name__}: {exc}",
            "traceback": traceback.format_exc(limit=8),
        }
        json_path = write_json_report(report, output_dir)
        return (
            f"### Analysis failed\n{type(exc).__name__}: {exc}",
            empty_timeline("Analysis failed. See JSON report for details."),
            [],
            pd.DataFrame(),
            json_path,
            json.dumps(report, indent=2),
        )


FORCE_LIGHT_JS = """
function refresh() {
    const url = new URL(window.location);
    if (url.searchParams.get('__theme') !== 'light') {
        url.searchParams.set('__theme', 'light');
        window.location.href = url.href;
    }
}
"""

CSS = """
:root,
.gradio-container {
  --radius: 8px;
  --body-background-fill: #f7f7f5;
  --body-text-color: #171717;
  --block-background-fill: #ffffff;
  --block-border-color: #e5e7eb;
  --button-primary-background-fill: #3758f9;
  --button-primary-background-fill-hover: #2844cb;
  --button-primary-text-color: #ffffff;
  color-scheme: light;
}

/* Override Gradio dark mode – force light colours even if .dark is applied */
.dark,
.dark .gradio-container {
  --body-background-fill: #f7f7f5 !important;
  --body-text-color: #171717 !important;
  --block-background-fill: #ffffff !important;
  --block-border-color: #e5e7eb !important;
  --button-primary-background-fill: #3758f9 !important;
  --button-primary-background-fill-hover: #2844cb !important;
  --button-primary-text-color: #ffffff !important;
  --neutral-50: #fafafa !important;
  --neutral-100: #f5f5f5 !important;
  --neutral-200: #e5e5e5 !important;
  --neutral-300: #d4d4d4 !important;
  --neutral-400: #a3a3a3 !important;
  --neutral-500: #737373 !important;
  --neutral-600: #525252 !important;
  --neutral-700: #404040 !important;
  --neutral-800: #262626 !important;
  --neutral-900: #171717 !important;
  --neutral-950: #0a0a0a !important;
  --background-fill-primary: #ffffff !important;
  --background-fill-secondary: #f7f7f5 !important;
  --border-color-primary: #e5e7eb !important;
  --input-background-fill: #ffffff !important;
  --block-label-background-fill: #f7f7f5 !important;
  --block-label-text-color: #171717 !important;
  --block-title-text-color: #171717 !important;
  --checkbox-background-color: #ffffff !important;
  --table-even-background-fill: #f9fafb !important;
  --table-odd-background-fill: #ffffff !important;
  color-scheme: light !important;
}

.gradio-container {
  max-width: 1480px !important;
  margin: 0 auto !important;
  padding: 24px !important;
  font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif !important;
}

.tdl-shell {
  display: grid;
  gap: 18px;
}

.tdl-hero {
  display: grid;
  grid-template-columns: minmax(0, 1fr) auto;
  gap: 16px;
  align-items: start;
  border-bottom: 1px solid #e5e7eb;
  padding-bottom: 16px;
}

.tdl-hero h1 {
  margin: 0;
  color: #111827;
  font-size: clamp(30px, 4vw, 52px);
  line-height: 1;
  letter-spacing: 0 !important;
}

.tdl-hero p {
  max-width: 880px;
  margin: 8px 0 0;
  color: #6b7280;
  font-size: 15px;
  line-height: 1.45;
}

.tdl-lockup {
  display: inline-flex;
  align-items: center;
  gap: 8px;
  white-space: nowrap;
  font-weight: 600;
}

.tdl-mark {
  width: 18px;
  height: 18px;
  border-radius: 5px;
  background: linear-gradient(135deg, #3758f9, #c75b3f);
}

.tdl-pill-row {
  display: flex;
  flex-wrap: wrap;
  gap: 8px;
  margin-top: 14px;
}

.tdl-pill {
  display: inline-flex;
  align-items: center;
  height: 30px;
  border: 1px solid #e5e7eb;
  border-radius: 999px;
  padding: 0 10px;
  background: #fff;
  color: #4b5563;
  font-size: 13px;
}

.tdl-panel {
  border: 1px solid #e5e7eb !important;
  border-radius: 8px !important;
  background: #fff !important;
  padding: 14px !important;
}

.tdl-panel h2 {
  margin: 0 0 8px;
  color: #111827;
  font-size: 18px;
}

.tdl-note {
  border-left: 3px solid #3758f9;
  padding: 8px 10px;
  background: #f7f7f5;
  color: #4b5563;
  font-size: 13px;
  line-height: 1.4;
}

button.primary {
  min-height: 44px !important;
  border-radius: 8px !important;
}

@media (max-width: 820px) {
  html,
  body {
    max-width: 100vw !important;
    overflow-x: hidden !important;
  }

  .gradio-container {
    padding: 16px 12px 28px !important;
    width: 100vw !important;
    max-width: 100vw !important;
    min-width: 0 !important;
    box-sizing: border-box !important;
  }

  .gradio-container > .main,
  .gradio-container > .main > .wrap,
  .gradio-container > .main > .wrap > main {
    width: 100% !important;
    max-width: 100% !important;
    min-width: 0 !important;
    margin-left: 0 !important;
    margin-right: 0 !important;
  }

  .tdl-hero {
    grid-template-columns: 1fr;
  }

  .tdl-main-row,
  .tdl-report-row {
    flex-direction: column !important;
    align-items: stretch !important;
  }

  .tdl-main-row > *,
  .tdl-report-row > * {
    width: 100% !important;
    min-width: 0 !important;
    max-width: 100% !important;
  }

  .tdl-panel {
    padding: 12px !important;
  }

  .tdl-pill {
    height: auto;
    min-height: 30px;
    white-space: normal;
  }

  .tdl-lockup {
    justify-self: start;
  }
}

@media (max-width: 480px) {
  .gradio-container {
    padding: 12px 10px 24px !important;
  }

  .tdl-hero h1 {
    font-size: 30px;
    line-height: 1.05;
  }

  .tdl-hero p {
    font-size: 14px;
  }
}
"""

HEAD = """
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="color-scheme" content="light only">
"""

THEME = gr.themes.Base(primary_hue="blue", secondary_hue="gray", neutral_hue="gray")

with gr.Blocks(title=APP_TITLE) as demo:
    gr.HTML(
        """
        <div class="tdl-shell">
          <header class="tdl-hero">
            <div>
              <h1>Temporal Difference Lab</h1>
              <p>Analyze subtle video changes with deterministic temporal preprocessing, optional V-JEPA evidence, and optional InternVL3.5-30B-A3B reasoning on ZeroGPU.</p>
              <div class="tdl-pill-row">
                <span class="tdl-pill">PySceneDetect + OpenCV flow</span>
                <span class="tdl-pill">Diff / SSIM / edge maps</span>
                <span class="tdl-pill">V-JEPA evidence path</span>
                <span class="tdl-pill">InternVL ZeroGPU xlarge</span>
              </div>
            </div>
            <div class="tdl-lockup"><span class="tdl-mark"></span><span>TDL</span></div>
          </header>
        </div>
        """
    )

    with gr.Row(elem_classes=["tdl-main-row"]):
        with gr.Column(scale=4, elem_classes=["tdl-panel"]):
            gr.HTML("<h2>Input</h2>")
            video_input = gr.Video(label="Video input", sources=["upload"], format=None)
            task_preset = gr.Dropdown(
                label="Task preset",
                choices=PRESETS,
                value="Hail vs Pebbles",
                interactive=True,
            )
            question = gr.Textbox(
                label="Analysis question",
                value="What changed over time, and is the change more consistent with falling hail than static pebbles?",
                lines=3,
            )
        with gr.Column(scale=3, elem_classes=["tdl-panel"]):
            gr.HTML("<h2>Controls</h2>")
            sensitivity = gr.Slider(0.1, 1.0, value=0.55, step=0.05, label="Detection sensitivity")
            max_events = gr.Slider(1, 5, value=5, step=1, label="Candidate events")
            sample_fps = gr.Slider(0.5, 4.0, value=DEFAULT_SAMPLE_FPS, step=0.5, label="Sample FPS")
            max_duration = gr.Slider(5, MAX_HARD_DURATION_SECONDS, value=DEFAULT_MAX_DURATION_SECONDS, step=5, label="Max seconds to analyze")
            use_jepa = gr.Checkbox(value=False, label="Use V-JEPA evidence when available")
            run_vlm = gr.Checkbox(value=not SKIP_MODEL_LOAD, label="Run InternVL reasoning")
            analyze_btn = gr.Button("Analyze video", variant="primary")
            gr.HTML(
                """
                <div class="tdl-note">CPU preprocessing always runs. V-JEPA and InternVL paths fall back cleanly if models are disabled, unavailable, or out of quota.</div>
                """
            )

    summary_output = gr.Markdown(value="### Ready\nUpload a video to begin.")
    timeline_output = gr.Plot(value=empty_timeline(), label="Temporal timeline", show_label=False)
    gallery_output = gr.Gallery(label="Frame, difference, and flow evidence", columns=3, height=520)
    table_output = gr.Dataframe(label="Event table", interactive=False, wrap=True)
    with gr.Row(elem_classes=["tdl-report-row"]):
        json_file_output = gr.File(label="Download JSON report")
        json_output = gr.Code(label="Structured report", language="json", lines=18)

    analyze_btn.click(
        fn=run_analysis,
        inputs=[
            video_input,
            task_preset,
            question,
            sensitivity,
            max_events,
            sample_fps,
            max_duration,
            use_jepa,
            run_vlm,
        ],
        outputs=[
            summary_output,
            timeline_output,
            gallery_output,
            table_output,
            json_file_output,
            json_output,
        ],
        api_name="analyze_video",
    )

    gr.Examples(
        examples=[
            ["Hail vs Pebbles", "What changed over time, and is it more consistent with falling hail than static pebbles?", 0.55, 5, 2.0, 60, False, False],
            ["Shader Artifact", "Did the surface shading, normal map, or specular response change unexpectedly?", 0.65, 5, 2.0, 60, False, False],
            ["Particle Burst", "When did a new particle system appear, and what motion evidence supports it?", 0.55, 5, 2.0, 60, False, False],
        ],
        inputs=[task_preset, question, sensitivity, max_events, sample_fps, max_duration, use_jepa, run_vlm],
    )


if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch(
        server_name="0.0.0.0",
        server_port=7860,
        theme=THEME,
        css=CSS,
        head=HEAD,
        js=FORCE_LIGHT_JS,
        ssr_mode=False,
        show_error=True,
    )