Spaces:
Running on Zero
Running on Zero
Postroll Q&A: re-anchor mailed prompts on the last frame (a frameless prompt splice parks the model on its turn-opening <|silence|>); never let session.close() skip the mailbox cleanup.
fe86be8 | """Session engine for the MOSS-VL-Realtime Space. | |
| One realtime session == one GPU call. The Gradio side (app.py) starts a session | |
| bound to the staged media; prompts typed while the session is live reach the | |
| GPU worker through an on-disk mailbox (ZeroGPU runs @spaces.GPU functions in a | |
| forked worker on the same container, so /tmp is shared between the main | |
| process and the worker). | |
| MOCK mode (MOSS_DEMO_MOCK=1): no torch / spaces / torchcodec imports; a | |
| scripted session drives the exact same event protocol so the full UI can be | |
| exercised on a CPU-only box. | |
| """ | |
| import json | |
| import os | |
| import re | |
| import shutil | |
| import time | |
| import traceback | |
| from collections import deque | |
| MOCK = os.getenv("MOSS_DEMO_MOCK") == "1" | |
| MODEL_ID = os.getenv("MOSS_MODEL_ID", "OpenMOSS-Team/MOSS-VL-Realtime") # hub id or local path | |
| MAILBOX_ROOT = "/tmp/moss_sessions" | |
| CONTROL_ROUND_START = "<|round_start|>" | |
| CONTROL_ROUND_END = "<|round_end|>" | |
| CONTROL_RESPONSE = "<|response|>" # real model's round-start marker | |
| CONTROL_SILENCE = "<|silence|>" | |
| # Session budgets (seconds). The paced stream is capped so a session always | |
| # closes gracefully before the ZeroGPU duration kill. | |
| SESSION_VIDEO_CAP_S = 120.0 | |
| LIVE_CAP_S = float(os.getenv("MOSS_LIVE_CAP_S", "180")) # live-camera session length | |
| POSTROLL_IDLE_S = 45.0 | |
| HARD_MARGIN_S = 15.0 | |
| CLOSE_GRACE_S = 8.0 | |
| IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tif", ".tiff"} | |
| VIDEO_EXTS = {".mp4", ".mov", ".webm", ".avi", ".mkv", ".ogg", ".m4v"} | |
| FRAME_MAX_SIDE = 1280 # downscale before pickling frames into the GPU worker | |
| if not MOCK: | |
| import ctypes | |
| import site | |
| # nvidia-npp-cu12 installs libnppicc.so.12 inside site-packages/nvidia/npp/lib/, | |
| # which is not on LD_LIBRARY_PATH. Load it globally before torchcodec is imported | |
| # so the dynamic linker can resolve it when torchcodec dlopen's its shared libs. | |
| def _preload_npp(): | |
| for _sp in site.getsitepackages(): | |
| _p = os.path.join(_sp, "nvidia", "npp", "lib", "libnppicc.so.12") | |
| if os.path.exists(_p): | |
| ctypes.CDLL(_p, mode=ctypes.RTLD_GLOBAL) | |
| return | |
| _preload_npp() | |
| try: | |
| import spaces # MUST come before torch / any CUDA-touching import (ZeroGPU) | |
| except ImportError: | |
| spaces = None # bare GPU box: decorator no-ops, model runs on real CUDA | |
| import torch | |
| from transformers import AutoModelForCausalLM, AutoProcessor | |
| print("Loading processor...") | |
| processor = AutoProcessor.from_pretrained( | |
| MODEL_ID, trust_remote_code=True, frame_extract_num_threads=1 | |
| ) | |
| print("Loading model...") | |
| model = AutoModelForCausalLM.from_pretrained( | |
| MODEL_ID, | |
| trust_remote_code=True, | |
| torch_dtype=torch.bfloat16, | |
| attn_implementation="sdpa", | |
| ).to("cuda") | |
| model.eval() | |
| print("Model ready.") | |
| if spaces is not None: | |
| GPU = spaces.GPU | |
| else: | |
| def GPU(*d_args, **d_kwargs): | |
| if d_args and callable(d_args[0]): | |
| return d_args[0] | |
| def _wrap(fn): | |
| return fn | |
| return _wrap | |
| else: | |
| def GPU(*d_args, **d_kwargs): | |
| """Effect-free stand-in for spaces.GPU in MOCK mode.""" | |
| if d_args and callable(d_args[0]): | |
| return d_args[0] | |
| def _wrap(fn): | |
| return fn | |
| return _wrap | |
| processor = None | |
| model = None # replaced by _MockModel via get_model() | |
| # --- Media normalization --- | |
| def classify_media(path): | |
| ext = os.path.splitext(path)[1].lower() | |
| if ext in IMAGE_EXTS: | |
| return "image" | |
| if ext in VIDEO_EXTS: | |
| return "video" | |
| return "video" # gr.Video/gr.Image constrain uploads; default to video | |
| def snapshot_stage(stage_video, stage_image): | |
| """Resolve the staged media at session start. | |
| Returns (kind, path, warning) where warning is a user-facing note or None. | |
| Video wins when both stages are populated (surfaced, not silent). | |
| """ | |
| if stage_video and stage_image: | |
| return ( | |
| "video", | |
| stage_video, | |
| "Both video and image are staged — the session runs on the video. " | |
| "两者都已上传，本次会话使用视频。", | |
| ) | |
| if stage_video: | |
| return "video", stage_video, None | |
| if stage_image: | |
| return "image", stage_image, None | |
| return None, None, None | |
| def _downscale(img, max_side=FRAME_MAX_SIDE): | |
| w, h = img.size | |
| scale = max(w, h) / float(max_side) | |
| if scale > 1.0: | |
| img = img.resize((int(w / scale), int(h / scale))) | |
| return img | |
| def extract_frames(video_path, video_fps, max_frames): | |
| """Decode a video into [(PIL.Image, timestamp_seconds)] sampled at video_fps. | |
| Runs on CPU in the main process (outside the GPU lease). | |
| """ | |
| if MOCK: | |
| return _mock_frames(video_fps, max_frames) | |
| from torchcodec.decoders import VideoDecoder | |
| from torchvision.transforms.functional import to_pil_image | |
| decoder = VideoDecoder(video_path) | |
| duration = float(decoder.metadata.duration_seconds or 0.0) | |
| if duration <= 0: | |
| frame = decoder[0] | |
| return [(_downscale(to_pil_image(frame)), 0.0)] | |
| step = 1.0 / float(video_fps) if float(video_fps) > 0 else 1.0 | |
| timestamps = [] | |
| t = 0.0 | |
| # keep a small epsilon away from the very end (no frame plays exactly at duration) | |
| end = max(duration - 1e-3, 0.0) | |
| while t <= end and len(timestamps) < int(max_frames): | |
| timestamps.append(round(t, 3)) | |
| t += step | |
| if not timestamps: | |
| timestamps = [0.0] | |
| batch = decoder.get_frames_played_at(seconds=timestamps) | |
| frames = [] | |
| for i in range(batch.data.shape[0]): | |
| img = _downscale(to_pil_image(batch.data[i])) | |
| ts = float(batch.pts_seconds[i]) | |
| frames.append((img, ts)) | |
| return frames | |
| def load_image_frame(image_path): | |
| from PIL import Image | |
| img = Image.open(image_path).convert("RGB") | |
| return [(_downscale(img), 0.0)] | |
| def _mock_frames(video_fps, max_frames): | |
| from PIL import Image | |
| step = 1.0 / float(video_fps) if float(video_fps) > 0 else 1.0 | |
| n = min(int(max_frames), 12) | |
| frames = [] | |
| for i in range(n): | |
| shade = 40 + (i * 160) // max(n - 1, 1) | |
| frames.append((Image.new("RGB", (64, 36), (shade, shade, 96)), round(i * step, 3))) | |
| return frames | |
| # --- Mailbox: main process -> GPU worker channel --- | |
| class Mailbox: | |
| """Per-session directory under /tmp shared with the forked GPU worker. | |
| prompts.jsonl : appended by the UI process, tailed by the worker | |
| stop : flag file — graceful session shutdown | |
| """ | |
| def _dir(sid): | |
| return os.path.join(MAILBOX_ROOT, sid) | |
| def create(sid): | |
| os.makedirs(Mailbox._dir(sid), exist_ok=True) | |
| def is_live(sid): | |
| return bool(sid) and os.path.isdir(Mailbox._dir(sid)) | |
| def write_prompt(sid, text): | |
| path = os.path.join(Mailbox._dir(sid), "prompts.jsonl") | |
| with open(path, "a", encoding="utf-8") as f: | |
| f.write(json.dumps({"text": text, "wall_ts": time.time()}) + "\n") | |
| def read_new_prompts(sid, offset): | |
| """Return (prompts, new_offset) for lines appended past byte offset.""" | |
| path = os.path.join(Mailbox._dir(sid), "prompts.jsonl") | |
| if not os.path.exists(path): | |
| return [], offset | |
| prompts = [] | |
| with open(path, "r", encoding="utf-8") as f: | |
| f.seek(offset) | |
| for line in f: | |
| if not line.endswith("\n"): | |
| break # partial write; re-read next tick | |
| offset += len(line.encode("utf-8")) | |
| try: | |
| prompts.append(json.loads(line)["text"]) | |
| except (ValueError, KeyError): | |
| continue | |
| return prompts, offset | |
| def mark_live_camera(sid): | |
| open(os.path.join(Mailbox._dir(sid), "live_camera"), "w").close() | |
| def is_live_camera(sid): | |
| return bool(sid) and os.path.exists(os.path.join(Mailbox._dir(sid), "live_camera")) | |
| def write_frame(sid, pil_image): | |
| """Store a live-camera frame for the GPU worker (name = capture time in ns).""" | |
| d = os.path.join(Mailbox._dir(sid), "frames") | |
| os.makedirs(d, exist_ok=True) | |
| name = f"{time.time_ns():020d}.jpg" | |
| tmp = os.path.join(d, "." + name) | |
| pil_image.save(tmp, "JPEG", quality=85) | |
| os.replace(tmp, os.path.join(d, name)) | |
| def read_new_frames(sid, after_name): | |
| """Return ([(path, name)], last_name) for frames newer than after_name.""" | |
| d = os.path.join(Mailbox._dir(sid), "frames") | |
| if not os.path.isdir(d): | |
| return [], after_name | |
| names = sorted(n for n in os.listdir(d) if not n.startswith(".") and n > (after_name or "")) | |
| return [(os.path.join(d, n), n) for n in names], (names[-1] if names else after_name) | |
| def signal_stop(sid): | |
| if Mailbox.is_live(sid): | |
| open(os.path.join(Mailbox._dir(sid), "stop"), "w").close() | |
| def should_stop(sid): | |
| return os.path.exists(os.path.join(Mailbox._dir(sid), "stop")) | |
| def cleanup(sid): | |
| shutil.rmtree(Mailbox._dir(sid), ignore_errors=True) | |
| def cleanup_stale(max_age_s=3600): | |
| if not os.path.isdir(MAILBOX_ROOT): | |
| return | |
| now = time.time() | |
| for name in os.listdir(MAILBOX_ROOT): | |
| path = os.path.join(MAILBOX_ROOT, name) | |
| try: | |
| if now - os.path.getmtime(path) > max_age_s: | |
| shutil.rmtree(path, ignore_errors=True) | |
| except OSError: | |
| continue | |
| # --- Round parsing (CPU side) --- | |
| _CTRL_RE = re.compile(r"(<\|[a-zA-Z_]+\|>)") | |
| class RoundParser: | |
| """Turn raw session chunks into UI ops. | |
| Ops: ("round_open", ts) | ("text", delta) | ("round_break", ts) | |
| | ("round_close", ts) | ("silence", ts) | ("control", token) for any | |
| other unknown <|...|> control token, which must stay out of the chat | |
| text but is worth logging in the raw view. | |
| Rounds open on <|round_start|> or the real model's <|response|>; they close | |
| on <|round_end|> or when silence resumes (the real model has no end marker). | |
| The real model RE-EMITS <|response|> every frame while narrating one | |
| continuous utterance — that yields ("round_break", ts): a raw-view round | |
| boundary that must NOT break the flowing chat text. | |
| Control tokens normally arrive as standalone chunks; the regex split is a | |
| defensive path for tokens embedded inside a larger chunk. | |
| """ | |
| def __init__(self): | |
| self.in_round = False | |
| self._round_has_text = False | |
| def _close(self, ops, ts): | |
| if self.in_round: | |
| self.in_round = False | |
| self._round_has_text = False | |
| ops.append(("round_close", ts)) | |
| def feed(self, chunk, ts): | |
| ops = [] | |
| pieces = [chunk] if _CTRL_RE.fullmatch(chunk) else [p for p in _CTRL_RE.split(chunk) if p] | |
| for piece in pieces: | |
| if piece in (CONTROL_ROUND_START, CONTROL_RESPONSE): | |
| if self.in_round and self._round_has_text: | |
| # per-frame re-emitted marker mid-narration: raw-view | |
| # boundary only — the utterance keeps flowing in chat | |
| self._round_has_text = False | |
| ops.append(("round_break", ts)) | |
| elif not self.in_round: | |
| self.in_round = True | |
| self._round_has_text = False | |
| ops.append(("round_open", ts)) | |
| # else: duplicate marker in a still-empty round — ignore | |
| elif piece == CONTROL_ROUND_END: | |
| self._close(ops, ts) | |
| elif piece == CONTROL_SILENCE: | |
| # the real model has no explicit round end — silence resuming | |
| # after a response marks the round as finished | |
| self._close(ops, ts) | |
| ops.append(("silence", ts)) | |
| elif _CTRL_RE.fullmatch(piece): | |
| ops.append(("control", piece)) | |
| else: | |
| if not self.in_round: | |
| # text without an explicit round marker — open one implicitly | |
| self.in_round = True | |
| ops.append(("round_open", ts)) | |
| self._round_has_text = True | |
| ops.append(("text", piece)) | |
| return ops | |
| # --- The session generator (runs in the GPU worker) --- | |
| def _speed_factor(playback_speed): | |
| return {"1×": 1.0, "2×": 2.0, "Fast-forward": 0.0}.get(playback_speed, 1.0) | |
| def estimate_duration(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s=POSTROLL_IDLE_S, live=False): | |
| """Dynamic @spaces.GPU duration: paced stream span + post-roll + margin.""" | |
| if live: | |
| return int(LIVE_CAP_S + HARD_MARGIN_S) | |
| speed = _speed_factor(playback_speed) | |
| span = frames[-1][1] if frames else 0.0 | |
| if speed > 0: | |
| paced = min(span / speed, SESSION_VIDEO_CAP_S) | |
| else: | |
| paced = min(len(frames) * 0.35 + 10.0, 90.0) | |
| return int(paced + postroll_idle_s + HARD_MARGIN_S) | |
| def _drain(session): | |
| chunks = [] | |
| while True: | |
| chunk = session.poll_output(timeout=0.0) | |
| if chunk is None: | |
| break | |
| chunks.append(chunk) | |
| return chunks | |
| def _poll_mailbox(session, sid, offset, video_ts, events): | |
| """Push any newly mailed prompts into the session; emit ack events.""" | |
| if sid is None: | |
| return offset | |
| prompts, offset = Mailbox.read_new_prompts(sid, offset) | |
| for text in prompts: | |
| session.push_prompt(text) | |
| events.append({"type": "prompt", "text": text, "video_ts": video_ts}) | |
| return offset | |
| def gpu_session(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s=POSTROLL_IDLE_S, live=False): | |
| """Run one realtime session; yields typed event dicts. | |
| Uploaded media: frames are paced against wall clock (speed factor from | |
| playback_speed; fast-forward pushes as fast as the model consumes), then a | |
| post-roll keeps the session open for Q&A. Live camera (live=True): frames | |
| arrive through the mailbox from the browser's webcam stream and are pushed | |
| with capture-time timestamps until stop flag / budget. | |
| """ | |
| speed = _speed_factor(playback_speed) | |
| budget = estimate_duration(sid, frames, initial_prompt, gen_kwargs, playback_speed, postroll_idle_s, live) | |
| total = len(frames) if frames else 0 | |
| mail_offset = 0 | |
| dropped_frames = 0 | |
| session = get_model().create_realtime_session( | |
| get_processor(), initial_prompt="", **gen_kwargs | |
| ) | |
| try: | |
| # The model runs a single realtime loop at a time; a just-closed session | |
| # can take a few seconds to release it. Wait it out instead of failing. | |
| for _ in range(20): | |
| try: | |
| session.start() | |
| break | |
| except RuntimeError as exc: | |
| if "active realtime generation" not in str(exc): | |
| raise | |
| time.sleep(0.5) | |
| else: | |
| yield { | |
| "type": "error", | |
| "message": "The model is busy with another session — try again in a moment. 模型正忙，请稍后重试。", | |
| } | |
| return | |
| t0 = time.monotonic() | |
| deadline = t0 + budget - CLOSE_GRACE_S | |
| yield {"type": "session_start", "frames_total": total, "budget_s": budget, "live": live} | |
| if initial_prompt: | |
| session.push_prompt(initial_prompt) | |
| yield {"type": "prompt", "text": initial_prompt, "video_ts": 0.0} | |
| if live: | |
| yield from _live_loop(session, sid, deadline) | |
| return | |
| end_reason = "stream ended" | |
| last_ts = 0.0 | |
| for i, (img, ts) in enumerate(frames): | |
| if Mailbox.should_stop(sid) if sid else False: | |
| end_reason = "stopped" | |
| break | |
| if time.monotonic() > deadline: | |
| end_reason = "session budget reached" | |
| break | |
| # pace: wait until this frame's wall-clock slot, staying responsive | |
| target = t0 + (ts / speed) if speed > 0 else 0.0 | |
| while time.monotonic() < target: | |
| events = [] | |
| mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events) | |
| chunks = _drain(session) | |
| if chunks: | |
| events.append( | |
| {"type": "chunk_batch", "video_ts": last_ts, "chunks": chunks} | |
| ) | |
| for ev in events: | |
| yield ev | |
| if (sid and Mailbox.should_stop(sid)) or time.monotonic() > deadline: | |
| break | |
| time.sleep(0.05) | |
| if sid and Mailbox.should_stop(sid): | |
| end_reason = "stopped" | |
| break | |
| dropped_frames += 1 if session.push_frame(img, timestamp=ts) else 0 | |
| last_ts = ts | |
| events = [] | |
| mail_offset = _poll_mailbox(session, sid, mail_offset, ts, events) | |
| chunks = _drain(session) | |
| for ev in events: | |
| yield ev | |
| yield { | |
| "type": "frame", | |
| "frame": i + 1, | |
| "total": total, | |
| "video_ts": ts, | |
| "chunks": chunks, | |
| "dropped_frames": dropped_frames, | |
| } | |
| else: | |
| end_reason = "stream ended" | |
| # Post-roll Q&A: the session stays open on the observed stream. | |
| if end_reason == "stream ended": | |
| yield {"type": "postroll", "video_ts": last_ts} | |
| idle_deadline = time.monotonic() + postroll_idle_s | |
| last_emit = time.monotonic() | |
| while time.monotonic() < min(idle_deadline, deadline): | |
| if sid and Mailbox.should_stop(sid): | |
| end_reason = "stopped" | |
| break | |
| events = [] | |
| prev_offset = mail_offset | |
| mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events) | |
| if mail_offset != prev_offset: | |
| idle_deadline = time.monotonic() + postroll_idle_s | |
| # A prompt spliced without a frame parks the model: the | |
| # assistant turn opens with <|silence|> (training format) | |
| # and the loop waits for new input, so the question is | |
| # never answered. Re-anchor postroll questions on the last | |
| # frame — prompt+frame in one drain cycle is the path that | |
| # actually generates a response (same as `analyze`). | |
| if frames: | |
| session.push_frame(frames[-1][0], timestamp=last_ts) | |
| chunk = session.poll_output(timeout=0.2) | |
| if chunk is not None: | |
| events.append( | |
| {"type": "chunk_batch", "video_ts": last_ts, "chunks": [chunk]} | |
| ) | |
| idle_deadline = time.monotonic() + postroll_idle_s | |
| if not events and time.monotonic() - last_emit > 0.3: | |
| events.append({"type": "tick", "video_ts": last_ts}) | |
| if events: | |
| last_emit = time.monotonic() | |
| for ev in events: | |
| yield ev | |
| else: | |
| if time.monotonic() >= deadline: | |
| end_reason = "session budget reached" | |
| elif end_reason == "stream ended": | |
| end_reason = "idle timeout" | |
| yield {"type": "session_end", "reason": end_reason, "video_ts": last_ts} | |
| except Exception as exc: | |
| traceback.print_exc() | |
| yield {"type": "error", "message": f"{type(exc).__name__}: {exc}"} | |
| finally: | |
| # close() can raise (join timeout / late worker error) — never let that | |
| # skip the mailbox cleanup, or the sid stays "live" and blocks every | |
| # new session until the stale sweep an hour later. | |
| try: | |
| session.close() | |
| except Exception: | |
| traceback.print_exc() | |
| if sid: | |
| Mailbox.cleanup(sid) | |
| def _live_loop(session, sid, deadline): | |
| """Consume live-camera frames from the mailbox until stop / budget.""" | |
| from PIL import Image | |
| mail_offset = 0 | |
| last_frame_name = None | |
| first_frame_ns = None | |
| pushed = 0 | |
| last_ts = 0.0 | |
| end_reason = "stopped" | |
| t0 = time.monotonic() | |
| last_emit = t0 # heartbeat so the UI can flush pending updates while idle | |
| while True: | |
| if sid and Mailbox.should_stop(sid): | |
| end_reason = "stopped" | |
| break | |
| if time.monotonic() > deadline: | |
| end_reason = "session budget reached" | |
| break | |
| frame_files, last_frame_name = Mailbox.read_new_frames(sid, last_frame_name) | |
| for path, name in frame_files: | |
| ns = int(name.split(".")[0]) | |
| if first_frame_ns is None: | |
| first_frame_ns = ns | |
| ts = max((ns - first_frame_ns) / 1e9, last_ts) | |
| try: | |
| img = Image.open(path).convert("RGB") | |
| except OSError: | |
| continue | |
| finally: | |
| try: | |
| os.remove(path) | |
| except OSError: | |
| pass | |
| session.push_frame(_downscale(img), timestamp=ts) | |
| last_ts = ts | |
| pushed += 1 | |
| events = [] | |
| mail_offset = _poll_mailbox(session, sid, mail_offset, last_ts, events) | |
| chunks = _drain(session) | |
| for ev in events: | |
| yield ev | |
| if frame_files or chunks: | |
| yield { | |
| "type": "frame", | |
| "frame": pushed, | |
| "total": 0, # unbounded live stream | |
| "video_ts": last_ts, | |
| "chunks": chunks, | |
| "dropped_frames": 0, | |
| } | |
| last_emit = time.monotonic() | |
| elif time.monotonic() - last_emit > 0.3: | |
| yield {"type": "tick", "video_ts": time.monotonic() - t0, "frames": pushed} | |
| last_emit = time.monotonic() | |
| time.sleep(0.1) | |
| # flush any final output briefly before closing | |
| flush_deadline = time.monotonic() + 2.0 | |
| while time.monotonic() < flush_deadline: | |
| chunk = session.poll_output(timeout=0.2) | |
| if chunk is None: | |
| continue | |
| yield {"type": "chunk_batch", "video_ts": last_ts, "chunks": [chunk]} | |
| yield {"type": "session_end", "reason": end_reason, "video_ts": last_ts} | |
| # --- Model access (real or mock) --- | |
| def get_model(): | |
| if MOCK: | |
| global model | |
| if model is None: | |
| model = _MockModel() | |
| return model | |
| return model | |
| def get_processor(): | |
| return processor | |
| class _MockSession: | |
| """Scripted realtime session mirroring the wire protocol. | |
| Emits silences while 'observing', two scripted rounds during the stream, | |
| and an echo round for every pushed prompt (proves the mailbox path). | |
| """ | |
| _ROUND_A = ["The stream opens on ", "a synthetic test pattern ", "fading in."] | |
| _ROUND_B = ["Brightness keeps increasing — ", "the pattern is nearly white now."] | |
| def __init__(self): | |
| self._out = deque() | |
| self._frames = 0 | |
| def start(self): | |
| return self | |
| def _queue_round(self, chunks): | |
| self._out.append(CONTROL_ROUND_START) | |
| self._out.append("<|response|>") # real model emits this inside rounds | |
| self._out.extend(chunks) | |
| self._out.append(CONTROL_ROUND_END) | |
| def push_frame(self, img, timestamp=None, drop_oldest=True): | |
| self._frames += 1 | |
| if self._frames == 4: | |
| self._queue_round(self._ROUND_A) | |
| elif self._frames == 9: | |
| self._queue_round(self._ROUND_B) | |
| elif self._frames % 3 == 0: | |
| self._out.append(CONTROL_SILENCE) | |
| return False | |
| def push_prompt(self, prompt): | |
| self._queue_round( | |
| ["(mock) You asked: ", f"“{prompt}” — ", f"I have seen {self._frames} frames so far."] | |
| ) | |
| def poll_output(self, timeout=0.0): | |
| if self._out: | |
| return self._out.popleft() | |
| if timeout > 0: | |
| time.sleep(min(timeout, 0.05)) | |
| if self._out: | |
| return self._out.popleft() | |
| return None | |
| def close(self, timeout=None): | |
| pass | |
| class _MockModel: | |
| def create_realtime_session(self, processor_, initial_prompt="", **kwargs): | |
| return _MockSession() | |
| # --- Stateless one-shot for MCP --- | |
| def analyze( | |
| media: str, | |
| prompt: str, | |
| max_new_tokens: int = 512, | |
| temperature: float = 0.0, | |
| video_fps: float = 1.0, | |
| max_frames: int = 64, | |
| ) -> str: | |
| """Analyze a video or image with MOSS-VL-Realtime and return the answer text. | |
| The media is streamed through a realtime session frame by frame (images are | |
| a single frame) and all model responses are collected and returned. | |
| Args: | |
| media: Path or http(s) URL of a video (.mp4/.mov/.webm) or image (.png/.jpg/...). | |
| prompt: The question or instruction about the media. | |
| max_new_tokens: Maximum number of tokens to generate per response round. | |
| temperature: Sampling temperature (0 = deterministic). | |
| video_fps: Frames per second sampled from a video. | |
| max_frames: Maximum number of frames sampled from a video. | |
| """ | |
| if media.startswith(("http://", "https://")): | |
| import tempfile | |
| import urllib.request | |
| suffix = os.path.splitext(media.split("?")[0])[1] or ".mp4" | |
| with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: | |
| with urllib.request.urlopen(media, timeout=60) as resp: | |
| shutil.copyfileobj(resp, tmp) | |
| media = tmp.name | |
| kind = classify_media(media) | |
| if kind == "image": | |
| frames = load_image_frame(media) | |
| else: | |
| frames = extract_frames(media, video_fps, max_frames) | |
| gen_kwargs = { | |
| "max_new_tokens": int(max_new_tokens), | |
| "temperature": float(temperature), | |
| "do_sample": float(temperature) > 0.0, | |
| } | |
| parser = RoundParser() | |
| rounds, current = [], [] | |
| # short post-roll: a one-shot call should not idle out the GPU lease | |
| for event in gpu_session(None, frames, prompt, gen_kwargs, "Fast-forward", postroll_idle_s=8.0): | |
| if event["type"] == "error": | |
| raise RuntimeError(event["message"]) | |
| for chunk in event.get("chunks", []): | |
| for op, payload in parser.feed(chunk, event.get("video_ts", 0.0)): | |
| if op == "text": | |
| current.append(payload) | |
| elif op == "round_close" and current: | |
| rounds.append("".join(current).strip()) | |
| current = [] | |
| if current: | |
| rounds.append("".join(current).strip()) | |
| return "\n\n".join(r for r in rounds if r) or "(the model stayed silent)" | |