from __future__ import annotations
import json
import os
import re
import subprocess
import sys
import traceback
import uuid
from dataclasses import dataclass
from pathlib import Path
from threading import Lock
from typing import Any, Dict, Iterator, List, Optional, Tuple
SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_DIR = SCRIPT_DIR.parent
if str(PROJECT_DIR) not in sys.path:
sys.path.insert(0, str(PROJECT_DIR))
COLLECTOR_PROMPT = """You are the Collector for a public video reasoning demo.
Watch the short video and write a concise factual summary that helps answer the user question.
Focus on visible actions, objects, scene changes, and any obvious temporal order.
Do not answer the question directly."""
PLANNER_PROMPT = """You are the Planner for a public video reasoning demo.
Decide whether the question needs a focused time span from the video.
Return valid JSON only:
{"use_grounder": true or false, "grounding_query": "short retrieval query", "reason": "short reason"}"""
GROUNDER_PROMPT = """You are a lightweight textual Grounder.
Identify the most relevant time span in the video for the question.
Return valid JSON only:
{"start_sec": number, "end_sec": number, "reason": "short reason"}
Rules:
- Use seconds from the start of the video.
- The span must be short and useful.
- If unsure, choose a slightly broader span.
- Do not return markdown."""
ANSWER_PROMPT = """You are the Answerer for a public video reasoning demo.
Use the video and the context summary to answer the user question.
Return a short explanation and put the final answer inside ...."""
REVIEW_PROMPT = """You are a lightweight Reviewer.
Judge whether the answer seems well supported by the visible video evidence.
Return valid JSON only:
{"confidence": "low" | "medium" | "high", "review": "one short paragraph"}"""
def extract_json_object(text: str) -> Dict[str, Any]:
match = re.search(r"(\{.*\})", text, re.DOTALL)
if not match:
return {}
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
return {}
def extract_answer(text: str) -> str:
match = re.search(r"\s*(.*?)\s*", text, re.DOTALL)
return match.group(1).strip() if match else text.strip()
def format_messages_markdown(title: str, messages: List[Dict[str, Any]]) -> str:
payload = json.dumps(messages, ensure_ascii=False, indent=2)
return f"### {title}\n```json\n{payload}\n```"
def probe_video(video_path: str) -> Dict[str, Any]:
import av
info = {
"duration_sec": 0.0,
"width": 0,
"height": 0,
"size_mb": round(os.path.getsize(video_path) / (1024 * 1024), 2),
}
with av.open(video_path) as container:
info["duration_sec"] = round((container.duration or 0) / 1_000_000.0, 2)
video_stream = next((stream for stream in container.streams if stream.type == "video"), None)
if video_stream is not None:
info["width"] = int(video_stream.width or 0)
info["height"] = int(video_stream.height or 0)
return info
def trim_video_ffmpeg(video_path: str, start: float, end: float, output_path: str) -> str:
cmd = [
"ffmpeg",
"-y",
"-ss",
f"{max(start, 0.0):.2f}",
"-to",
f"{max(end, start + 0.1):.2f}",
"-i",
video_path,
"-c:v",
"libx264",
"-preset",
"veryfast",
"-crf",
"28",
"-c:a",
"aac",
output_path,
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or "ffmpeg trim failed")
return output_path
def normalize_span(span: Dict[str, Any], duration_sec: float, max_window_sec: float) -> Optional[Tuple[float, float]]:
try:
start = float(span.get("start_sec"))
end = float(span.get("end_sec"))
except (TypeError, ValueError):
return None
if duration_sec <= 0:
return None
start = max(0.0, min(start, duration_sec))
end = max(0.0, min(end, duration_sec))
if end <= start:
return None
if end - start > max_window_sec:
end = min(duration_sec, start + max_window_sec)
if end <= start:
return None
return round(start, 2), round(end, 2)
def build_demo_query(question: str, sample: Optional[Dict[str, Any]] = None) -> str:
full_question = question.strip()
if sample and sample.get("problem", "").strip() == full_question:
problem_type = sample.get("problem_type", "free-form")
if problem_type in {"multiple choice", "emer_ov_mc"}:
options = sample.get("options") or []
if options:
full_question = f"{full_question}\nOptions:\n" + "\n".join(options)
return full_question
def _import_torch():
import torch
return torch
def _import_vision_helpers():
try:
from qwen_vl_utils import process_vision_info
return process_vision_info
except ImportError:
from videomind.dataset.utils import process_vision_info
return process_vision_info
@dataclass
class RuntimeConfig:
model_id: str = "Qwen/Qwen2-VL-2B-Instruct"
backend_mode: str = "cpu_fallback"
artifact_root: str = "/tmp/intentbench_space_artifacts"
local_device: str = "cpu"
remote_inference_url: str = ""
remote_api_key: str = ""
cpu_max_duration_sec: float = 8.0
cpu_max_frames: int = 8
cpu_fps: float = 1.0
cpu_max_pixels: int = 128 * 28 * 28
cpu_max_size_mb: float = 80.0
cpu_max_edge: int = 1400
local_max_duration_sec: float = 30.0
local_max_frames: int = 24
local_fps: float = 1.0
local_max_pixels: int = 256 * 28 * 28
max_new_tokens_collector: int = 96
max_new_tokens_planner: int = 64
max_new_tokens_grounder: int = 48
max_new_tokens_answer: int = 96
max_new_tokens_review: int = 64
max_grounded_window_sec: float = 12.0
@classmethod
def from_env(cls) -> "RuntimeConfig":
return cls(
model_id=os.environ.get("MODEL_ID", "Qwen/Qwen2-VL-2B-Instruct"),
backend_mode=os.environ.get("INFERENCE_BACKEND", "cpu_fallback"),
artifact_root=os.environ.get("INTENTBENCH_DEMO_ARTIFACT_ROOT", "/tmp/intentbench_space_artifacts"),
local_device=os.environ.get("LOCAL_DEVICE", "cpu"),
remote_inference_url=os.environ.get("REMOTE_INFERENCE_URL", ""),
remote_api_key=os.environ.get("REMOTE_API_KEY", ""),
cpu_max_duration_sec=float(os.environ.get("CPU_MAX_DURATION_SEC", "8")),
cpu_max_frames=int(os.environ.get("CPU_MAX_FRAMES", "8")),
cpu_fps=float(os.environ.get("CPU_FPS", "1.0")),
cpu_max_pixels=int(os.environ.get("CPU_MAX_PIXELS", str(128 * 28 * 28))),
cpu_max_size_mb=float(os.environ.get("CPU_MAX_SIZE_MB", "80")),
cpu_max_edge=int(os.environ.get("CPU_MAX_EDGE", "1400")),
local_max_duration_sec=float(os.environ.get("LOCAL_MAX_DURATION_SEC", "30")),
local_max_frames=int(os.environ.get("LOCAL_MAX_FRAMES", "24")),
local_fps=float(os.environ.get("LOCAL_FPS", "1.0")),
local_max_pixels=int(os.environ.get("LOCAL_MAX_PIXELS", str(256 * 28 * 28))),
max_new_tokens_collector=int(os.environ.get("MAX_NEW_TOKENS_COLLECTOR", "96")),
max_new_tokens_planner=int(os.environ.get("MAX_NEW_TOKENS_PLANNER", "64")),
max_new_tokens_grounder=int(os.environ.get("MAX_NEW_TOKENS_GROUNDER", "48")),
max_new_tokens_answer=int(os.environ.get("MAX_NEW_TOKENS_ANSWER", "96")),
max_new_tokens_review=int(os.environ.get("MAX_NEW_TOKENS_REVIEW", "64")),
max_grounded_window_sec=float(os.environ.get("MAX_GROUNDED_WINDOW_SEC", "12")),
)
def backend_label(self) -> str:
mapping = {
"cpu_fallback": "CPU fallback",
"local_gpu": "Local GPU",
"remote_api": "Remote API",
"disabled": "Disabled",
}
return mapping.get(self.backend_mode, self.backend_mode)
def cpu_limits_text(self) -> str:
return (
f"max {self.cpu_max_duration_sec:.0f}s video, "
f"{self.cpu_fps:.1f} fps, "
f"{self.cpu_max_frames} frames, "
f"{self.cpu_max_size_mb:.0f} MB, "
f"edge <= {self.cpu_max_edge}px"
)
class InferenceBackend:
mode = "disabled"
def __init__(self, config: RuntimeConfig):
self.config = config
def describe(self) -> Dict[str, str]:
return {
"mode": self.mode,
"title": self.config.backend_label(),
"message": "Inference is not configured.",
}
def run_pipeline(
self,
video_path: str,
question: str,
grounder_mode: str,
sample: Optional[Dict[str, Any]] = None,
) -> Iterator[Dict[str, Any]]:
yield {
"stage": "error",
"message": "Inference backend is unavailable.",
"backend_mode": self.mode,
"traceback": "",
}
class DisabledBackend(InferenceBackend):
mode = "disabled"
def describe(self) -> Dict[str, str]:
return {
"mode": self.mode,
"title": "UI-only mode",
"message": "The public page is online, but real inference is disabled until GPU grant or a remote backend is connected.",
}
def run_pipeline(self, video_path: str, question: str, grounder_mode: str, sample: Optional[Dict[str, Any]] = None):
yield {
"stage": "status",
"message": self.describe()["message"],
"backend_mode": self.mode,
"warnings": ["Switch to remote_api or local_gpu to enable real inference."],
}
yield {
"stage": "done",
"message": "UI-only mode complete.",
"backend_mode": self.mode,
"final_answer": "Inference is currently unavailable on this deployment.",
"review_summary": "No model execution happened.",
}
class RemoteAPIBackend(InferenceBackend):
mode = "remote_api"
def describe(self) -> Dict[str, str]:
if not self.config.remote_inference_url:
return {
"mode": self.mode,
"title": "Remote API",
"message": "Remote mode is selected, but REMOTE_INFERENCE_URL is not configured.",
}
return {
"mode": self.mode,
"title": "Remote API",
"message": "The Space UI is live and forwards inference to an external GPU service.",
}
def run_pipeline(self, video_path: str, question: str, grounder_mode: str, sample: Optional[Dict[str, Any]] = None):
import requests
if not self.config.remote_inference_url:
yield {
"stage": "error",
"message": "REMOTE_INFERENCE_URL is missing.",
"backend_mode": self.mode,
"traceback": "",
}
return
yield {
"stage": "status",
"message": "Sending request to remote inference service...",
"backend_mode": self.mode,
}
headers = {}
if self.config.remote_api_key:
headers["Authorization"] = f"Bearer {self.config.remote_api_key}"
with open(video_path, "rb") as handle:
response = requests.post(
self.config.remote_inference_url,
headers=headers,
data={
"question": question,
"grounder_mode": grounder_mode,
"sample": json.dumps(sample or {}, ensure_ascii=False),
},
files={"video": (Path(video_path).name, handle, "video/mp4")},
timeout=600,
)
response.raise_for_status()
payload = response.json()
if payload.get("collector_summary"):
yield {
"stage": "collector",
"message": "Collector finished via remote API.",
"backend_mode": self.mode,
"collector_summary": payload.get("collector_summary", ""),
"collector_raw": payload.get("collector_raw", payload.get("collector_summary", "")),
"raw_prompt": payload.get("collector_prompt", ""),
}
if payload.get("planner_decision"):
yield {
"stage": "planner",
"message": "Planner finished via remote API.",
"backend_mode": self.mode,
"planner_decision": payload.get("planner_decision", ""),
"planner_raw": payload.get("planner_raw", payload.get("planner_decision", "")),
"grounder_span_text": payload.get("grounder_span", ""),
"raw_prompt": payload.get("planner_prompt", ""),
}
if payload.get("grounder_span"):
yield {
"stage": "grounder",
"message": "Grounder finished via remote API.",
"backend_mode": self.mode,
"grounder_span_text": payload.get("grounder_span", ""),
"grounder_raw": payload.get("grounder_raw", payload.get("grounder_span", "")),
"grounded_video": payload.get("highlight_clip_path"),
"raw_prompt": payload.get("grounder_prompt", ""),
}
yield {
"stage": "answer",
"message": "Answer received from remote API.",
"backend_mode": self.mode,
"final_answer": payload.get("final_answer", ""),
"answer_raw": payload.get("answer_raw", payload.get("final_answer", "")),
"raw_prompt": payload.get("answer_prompt", ""),
}
yield {
"stage": "review",
"message": "Review received from remote API.",
"backend_mode": self.mode,
"review_summary": payload.get("review_summary", ""),
"review_raw": payload.get("review_raw", payload.get("review_summary", "")),
"raw_prompt": payload.get("review_prompt", ""),
}
yield {
"stage": "done",
"message": "Remote inference completed.",
"backend_mode": self.mode,
"final_answer": payload.get("final_answer", ""),
"review_summary": payload.get("review_summary", ""),
"grounded_video": payload.get("highlight_clip_path"),
}
class LocalQwenVLBackend(InferenceBackend):
mode = "local"
def __init__(self, config: RuntimeConfig):
super().__init__(config)
self.model = None
self.processor = None
self.process_vision_info = None
self.torch = None
self.device = config.local_device
self.dtype = None
self._load_lock = Lock()
self._run_lock = Lock()
self.artifact_root = Path(config.artifact_root)
self.artifact_root.mkdir(parents=True, exist_ok=True)
def describe(self) -> Dict[str, str]:
return {
"mode": self.mode,
"title": self.config.backend_label(),
"message": f"Single-model Qwen2-VL pipeline on {self.device}.",
}
def ensure_loaded(self) -> None:
if self.model is not None and self.processor is not None:
return
with self._load_lock:
if self.model is not None and self.processor is not None:
return
self.torch = _import_torch()
from transformers import AutoProcessor, Qwen2VLForConditionalGeneration
self.process_vision_info = _import_vision_helpers()
if self.device.startswith("cuda") and self.torch.cuda.is_available():
self.dtype = self.torch.bfloat16
else:
self.device = "cpu"
self.dtype = self.torch.float32
kwargs: Dict[str, Any] = {
"torch_dtype": self.dtype,
"low_cpu_mem_usage": True,
}
self.model = Qwen2VLForConditionalGeneration.from_pretrained(self.config.model_id, **kwargs)
self.model = self.model.to(self.device)
self.model.eval()
self.processor = AutoProcessor.from_pretrained(self.config.model_id)
def _make_run_dir(self) -> Path:
run_dir = self.artifact_root / f"run_{uuid.uuid4().hex[:8]}"
run_dir.mkdir(parents=True, exist_ok=True)
return run_dir
def _processor_inputs(
self,
messages: List[Dict[str, Any]],
) -> Dict[str, Any]:
chat_text = self.processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
image_inputs, video_inputs = self.process_vision_info(messages)
inputs = self.processor(
text=[chat_text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
for key, value in inputs.items():
if hasattr(value, "to"):
inputs[key] = value.to(self.device)
if getattr(inputs[key], "is_floating_point", lambda: False)():
inputs[key] = inputs[key].to(self.dtype)
return inputs
def _generate(
self,
messages: List[Dict[str, Any]],
max_new_tokens: int,
) -> str:
inputs = self._processor_inputs(messages)
with self.torch.inference_mode():
output_ids = self.model.generate(
**inputs,
max_new_tokens=max_new_tokens,
do_sample=False,
)
generated = output_ids[0][inputs["input_ids"].size(1):]
return self.processor.decode(generated, skip_special_tokens=True)
def _video_messages(
self,
video_path: str,
prompt: str,
fps: float,
max_frames: int,
max_pixels: int,
) -> List[Dict[str, Any]]:
return [
{
"role": "user",
"content": [
{
"type": "video",
"video": video_path,
"fps": fps,
"max_frames": max_frames,
"max_pixels": max_pixels,
},
{"type": "text", "text": prompt},
],
}
]
def _run_collector(self, video_path: str, question: str, fps: float, max_frames: int, max_pixels: int):
prompt = f"{COLLECTOR_PROMPT}\n\nUser question: {question}"
messages = self._video_messages(video_path, prompt, fps, max_frames, max_pixels)
output = self._generate(messages, self.config.max_new_tokens_collector)
return output, messages
def _run_planner(
self,
video_path: str,
question: str,
collector_summary: str,
fps: float,
max_frames: int,
max_pixels: int,
):
prompt = (
f"{PLANNER_PROMPT}\n\n"
f"User question: {question}\n\n"
f"Collector summary:\n{collector_summary}"
)
messages = self._video_messages(video_path, prompt, fps, max_frames, max_pixels)
output = self._generate(messages, self.config.max_new_tokens_planner)
return extract_json_object(output), output, messages
def _run_grounder(
self,
video_path: str,
question: str,
grounding_query: str,
fps: float,
max_frames: int,
max_pixels: int,
):
prompt = (
f"{GROUNDER_PROMPT}\n\n"
f"User question: {question}\n"
f"Grounding query: {grounding_query or question}"
)
messages = self._video_messages(video_path, prompt, fps, max_frames, max_pixels)
output = self._generate(messages, self.config.max_new_tokens_grounder)
return extract_json_object(output), output, messages
def _run_answer(
self,
video_path: str,
question: str,
collector_summary: str,
fps: float,
max_frames: int,
max_pixels: int,
):
prompt = (
f"{ANSWER_PROMPT}\n\n"
f"User question: {build_demo_query(question)}\n\n"
f"Collector summary:\n{collector_summary}"
)
messages = self._video_messages(video_path, prompt, fps, max_frames, max_pixels)
output = self._generate(messages, self.config.max_new_tokens_answer)
return output, messages
def _run_review(
self,
video_path: str,
question: str,
answer_text: str,
fps: float,
max_frames: int,
max_pixels: int,
):
prompt = (
f"{REVIEW_PROMPT}\n\n"
f"User question: {question}\n\n"
f"Candidate answer:\n{answer_text}"
)
messages = self._video_messages(video_path, prompt, fps, max_frames, max_pixels)
output = self._generate(messages, self.config.max_new_tokens_review)
return extract_json_object(output), output, messages
class CPUFallbackBackend(LocalQwenVLBackend):
mode = "cpu_fallback"
def describe(self) -> Dict[str, str]:
return {
"mode": self.mode,
"title": "CPU fallback",
"message": f"Very short clips only. Limits: {self.config.cpu_limits_text()}",
}
def _check_limits(self, video_info: Dict[str, Any]) -> List[str]:
issues = []
if video_info["duration_sec"] > self.config.cpu_max_duration_sec:
issues.append(
f"Video is {video_info['duration_sec']:.2f}s; CPU fallback only supports up to {self.config.cpu_max_duration_sec:.0f}s."
)
if video_info["size_mb"] > self.config.cpu_max_size_mb:
issues.append(
f"Video is {video_info['size_mb']:.1f} MB; CPU fallback only supports up to {self.config.cpu_max_size_mb:.0f} MB."
)
if max(video_info["width"], video_info["height"]) > self.config.cpu_max_edge:
issues.append(
f"Video edge is {max(video_info['width'], video_info['height'])} px; CPU fallback only supports up to {self.config.cpu_max_edge}px."
)
return issues
def run_pipeline(
self,
video_path: str,
question: str,
grounder_mode: str,
sample: Optional[Dict[str, Any]] = None,
) -> Iterator[Dict[str, Any]]:
if not video_path:
yield {"stage": "error", "message": "A video is required.", "backend_mode": self.mode, "traceback": ""}
return
if not question.strip():
yield {"stage": "error", "message": "A question is required.", "backend_mode": self.mode, "traceback": ""}
return
with self._run_lock:
run_dir = self._make_run_dir()
try:
info = probe_video(video_path)
issues = self._check_limits(info)
yield {
"stage": "status",
"message": f"CPU fallback mode active. Video: {info['duration_sec']:.2f}s, {info['width']}x{info['height']}, {info['size_mb']} MB.",
"backend_mode": self.mode,
"warnings": issues,
}
if issues:
yield {
"stage": "error",
"message": "CPU fallback limits exceeded.",
"backend_mode": self.mode,
"traceback": "\n".join(issues),
}
return
self.ensure_loaded()
collector_summary, collector_messages = self._run_collector(
video_path,
question,
self.config.cpu_fps,
self.config.cpu_max_frames,
self.config.cpu_max_pixels,
)
yield {
"stage": "collector",
"message": "Collector finished.",
"backend_mode": self.mode,
"collector_summary": collector_summary,
"collector_raw": collector_summary,
"raw_prompt": format_messages_markdown("Collector Input", collector_messages),
}
planner_json, planner_raw, planner_messages = self._run_planner(
video_path,
question,
collector_summary,
self.config.cpu_fps,
self.config.cpu_max_frames,
self.config.cpu_max_pixels,
)
use_grounder = bool(planner_json.get("use_grounder"))
if grounder_mode == "Always On":
use_grounder = True
elif grounder_mode == "Off":
use_grounder = False
grounding_query = str(planner_json.get("grounding_query", "")).strip()
planner_text = (
f"Use Grounder: {use_grounder}\n"
f"Grounding Query: {grounding_query or '-'}\n"
f"Reason: {planner_json.get('reason', '-')}"
)
yield {
"stage": "planner",
"message": "Planner finished.",
"backend_mode": self.mode,
"planner_decision": planner_text,
"planner_raw": planner_raw,
"raw_prompt": format_messages_markdown("Planner Input", planner_messages),
}
generation_video = video_path
grounded_span_text = "Grounder skipped."
grounded_video = None
if use_grounder:
span_json, grounder_raw, grounder_messages = self._run_grounder(
video_path,
question,
grounding_query or question,
self.config.cpu_fps,
self.config.cpu_max_frames,
self.config.cpu_max_pixels,
)
normalized = normalize_span(span_json, info["duration_sec"], self.config.max_grounded_window_sec)
if normalized:
grounded_video = str(run_dir / "highlight.mp4")
trim_video_ffmpeg(video_path, normalized[0], normalized[1], grounded_video)
generation_video = grounded_video
grounded_span_text = (
f"[{normalized[0]:.2f}s, {normalized[1]:.2f}s] "
f"- {span_json.get('reason', 'no reason provided')}"
)
else:
grounded_span_text = f"Grounder returned an invalid span.\n\nRaw:\n{grounder_raw}"
yield {
"stage": "grounder",
"message": "Grounder finished.",
"backend_mode": self.mode,
"grounder_span_text": grounded_span_text,
"grounder_raw": grounder_raw,
"grounded_video": grounded_video,
"raw_prompt": format_messages_markdown("Grounder Input", grounder_messages),
}
else:
yield {
"stage": "grounder",
"message": "Grounder skipped.",
"backend_mode": self.mode,
"grounder_span_text": grounded_span_text,
"grounder_raw": grounded_span_text,
"grounded_video": None,
}
answer_raw, answer_messages = self._run_answer(
generation_video,
question,
collector_summary,
self.config.cpu_fps,
self.config.cpu_max_frames,
self.config.cpu_max_pixels,
)
final_answer = extract_answer(answer_raw)
yield {
"stage": "answer",
"message": "Answer finished.",
"backend_mode": self.mode,
"final_answer": final_answer,
"answer_raw": answer_raw,
"grounded_video": grounded_video,
"raw_prompt": format_messages_markdown("Answer Input", answer_messages),
}
review_json, review_raw, review_messages = self._run_review(
generation_video,
question,
answer_raw,
self.config.cpu_fps,
self.config.cpu_max_frames,
self.config.cpu_max_pixels,
)
review_summary = (
f"Confidence: {review_json.get('confidence', 'unknown')}\n\n"
f"{review_json.get('review', review_raw)}"
)
yield {
"stage": "review",
"message": "Review finished.",
"backend_mode": self.mode,
"review_summary": review_summary,
"review_raw": review_raw,
"raw_prompt": format_messages_markdown("Review Input", review_messages),
}
yield {
"stage": "done",
"message": "CPU fallback pipeline completed.",
"backend_mode": self.mode,
"final_answer": final_answer,
"review_summary": review_summary,
"grounded_video": grounded_video,
}
except Exception as exc:
yield {
"stage": "error",
"message": f"{type(exc).__name__}: {exc}",
"backend_mode": self.mode,
"traceback": traceback.format_exc(),
}
class LocalGPUBackend(CPUFallbackBackend):
mode = "local_gpu"
def __init__(self, config: RuntimeConfig):
super().__init__(config)
self.device = config.local_device
def describe(self) -> Dict[str, str]:
return {
"mode": self.mode,
"title": "Local GPU",
"message": "Local GPU mode is enabled for the single-model Qwen2-VL pipeline.",
}
def _check_limits(self, video_info: Dict[str, Any]) -> List[str]:
issues = []
if video_info["duration_sec"] > self.config.local_max_duration_sec:
issues.append(
f"Video is {video_info['duration_sec']:.2f}s; local mode supports up to {self.config.local_max_duration_sec:.0f}s by default."
)
return issues
def run_pipeline(self, video_path: str, question: str, grounder_mode: str, sample: Optional[Dict[str, Any]] = None):
old_fps = self.config.cpu_fps
old_frames = self.config.cpu_max_frames
old_pixels = self.config.cpu_max_pixels
self.config.cpu_fps = self.config.local_fps
self.config.cpu_max_frames = self.config.local_max_frames
self.config.cpu_max_pixels = self.config.local_max_pixels
try:
for event in super().run_pipeline(video_path, question, grounder_mode, sample):
event["backend_mode"] = self.mode
yield event
finally:
self.config.cpu_fps = old_fps
self.config.cpu_max_frames = old_frames
self.config.cpu_max_pixels = old_pixels
class IntentBenchDemoRuntime:
def __init__(self, config: RuntimeConfig):
self.config = config
self.backends = {
"disabled": DisabledBackend(config),
"remote_api": RemoteAPIBackend(config),
"cpu_fallback": CPUFallbackBackend(config),
"local_gpu": LocalGPUBackend(config),
}
def shutdown(self) -> None:
return None
def describe_backend(self) -> Dict[str, str]:
backend = self.backends.get(self.config.backend_mode, self.backends["disabled"])
return backend.describe()
def run_pipeline(
self,
video_path: str,
question: str,
grounder_mode: str = "Auto",
sample: Optional[Dict[str, Any]] = None,
) -> Iterator[Dict[str, Any]]:
backend = self.backends.get(self.config.backend_mode, self.backends["disabled"])
return backend.run_pipeline(video_path, question, grounder_mode, sample)