# ---------------------------------------------------------------------------
# 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 score=%{y:.3f}",
)
)
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} t=%{x:.2f}s score=%{y:.3f}",
)
)
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 = """
"""
THEME = gr.themes.Base(primary_hue="blue", secondary_hue="gray", neutral_hue="gray")
with gr.Blocks(title=APP_TITLE) as demo:
gr.HTML(
"""
Temporal Difference Lab
Analyze subtle video changes with deterministic temporal preprocessing, optional V-JEPA evidence, and optional InternVL3.5-30B-A3B reasoning on ZeroGPU.
"""
)
with gr.Row(elem_classes=["tdl-main-row"]):
with gr.Column(scale=4, elem_classes=["tdl-panel"]):
gr.HTML("
Input
")
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("