diff --git a/apps/cam2v/README.md b/apps/cam2v/README.md new file mode 100644 index 000000000..6ffafa118 --- /dev/null +++ b/apps/cam2v/README.md @@ -0,0 +1,32 @@ +# FlashDreams Cam2V application + +`flashdreams-cam2v` owns the reusable v2 application, session, model-generation +loop, camera controls, and timing for interactive camera-to-video models. +Concrete integrations supply an existing runner config plus an input resolver +that turns their asset format into `Cam2VConditioning`. + +The application owns the loaded pipeline. Each session owns its autoregressive +cache, keyboard state, camera pose, and SlangPy UI overlay. The +io-thread runs the UI loop over the current video frame; the +model-generation-thread runs the model loop and is the only thread that mutates +rollout state. Model status crosses to the UI loop through `invoke_async` +messages. + +The overlay is enabled by default. Pass `-- --no-ui` after the application +arguments to use the default model-output blitter for headless or benchmark +runs. + +For UI testing without loading a real model, run the packaged dummy pipeline: + +```bash +uv run flashdreams-run-v2 cam2v-dummy --mode webrtc \ + --host 0.0.0.0 --port 8089 -- \ + --step-wait-seconds 0.9 --frames-per-chunk 12 +``` + +The model-generation-thread waits on a `threading.Event` for each synthetic +step while the io-thread continues collecting browser input and presenting +generated frames. + +See `integrations_v2/cam2v_lingbot/cam2v_lingbot/app.py` for the minimal +specialization pattern. diff --git a/apps/cam2v/__init__.py b/apps/cam2v/__init__.py new file mode 100644 index 000000000..ca187cc5f --- /dev/null +++ b/apps/cam2v/__init__.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Reusable interactive camera-to-video application primitives.""" + +from .application import Cam2VApplication +from .controls import CameraPoseIntegrator, KeyboardResampler, PoseSegment +from .defaults import ( + Cam2VApplicationDefaults, + Cam2VConditioning, + Cam2VInputResolver, +) +from .session import ( + Cam2VModelLoop, + Cam2VModelState, + Cam2VSession, + Cam2VSessionConfig, + CameraControlInput, +) +from .ui import Cam2VSlangPyUILoop, Cam2VUIState, Cam2VUIStatus + +__all__ = [ + "Cam2VApplication", + "Cam2VApplicationDefaults", + "Cam2VConditioning", + "Cam2VInputResolver", + "Cam2VModelLoop", + "Cam2VModelState", + "Cam2VSession", + "Cam2VSessionConfig", + "Cam2VSlangPyUILoop", + "Cam2VUIState", + "Cam2VUIStatus", + "CameraControlInput", + "CameraPoseIntegrator", + "KeyboardResampler", + "PoseSegment", +] diff --git a/apps/cam2v/application.py b/apps/cam2v/application.py new file mode 100644 index 000000000..ff73bbaa2 --- /dev/null +++ b/apps/cam2v/application.py @@ -0,0 +1,265 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Reusable camera-to-video application on the FlashDreams v2 API.""" + +from __future__ import annotations + +import argparse +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import torch + +from flashdreams.api_v2.application import IApplication +from flashdreams.api_v2.session import ISession +from flashdreams.infra.config import derive_config +from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.video_tensor import VideoTensorLayout + +from .defaults import Cam2VApplicationDefaults, Cam2VConditioning +from .session import Cam2VSession, Cam2VSessionConfig + + +class Cam2VApplication(IApplication): + """Reusable interactive camera-to-video application. + + The shared class owns command-line parsing, pipeline lifetime, session + validation, and model-generation-loop construction. A concrete model + integration contributes a runner config and an input resolver through + :class:`Cam2VApplicationDefaults`. + """ + + def __init__(self, *, defaults: Cam2VApplicationDefaults) -> None: + self.defaults = defaults + self._pipeline_config = defaults.pipeline_config + self._device = defaults.device + self._total_blocks = defaults.total_blocks + self._log_every_blocks = defaults.log_every_blocks + self._warmup_blocks = defaults.warmup_blocks + self._use_ui = True + self._input_values: dict[str, Any] | None = None + self._pipeline: Any | None = None + + @property + def pipeline_config(self) -> Any: + """Return the model configuration after command-line overrides.""" + return self._pipeline_config + + def init(self, commandline_args: Sequence[str]) -> None: + """Parse shared camera-to-video inputs without loading the model.""" + parser = argparse.ArgumentParser( + prog="flashdreams-run-v2 CAM2V_SLUG --", + description="Generate video from a first frame and keyboard camera input.", + ) + input_defaults = self.defaults.input_defaults + parser.add_argument("--prompt", default=input_defaults.get("prompt", "")) + parser.add_argument( + "--prompt-path", + type=Path, + default=input_defaults.get("prompt_path"), + ) + parser.add_argument( + "--image-path", + type=Path, + default=input_defaults.get("image_path"), + ) + parser.add_argument( + "--pose-path", + type=Path, + default=input_defaults.get("pose_path"), + ) + parser.add_argument( + "--intrinsic-path", + type=Path, + default=input_defaults.get("intrinsic_path"), + ) + parser.add_argument( + "--world-scale", + type=float, + default=input_defaults.get("world_scale"), + ) + parser.add_argument( + "--example-data", + action=argparse.BooleanOptionalAction, + default=bool(input_defaults.get("example_data", False)), + ) + parser.add_argument( + "--example-idx", + type=int, + default=int(input_defaults.get("example_idx", 0)), + ) + parser.add_argument( + "--device", + default=self.defaults.device, + help="Device used by the shared model. Default: %(default)s.", + ) + parser.add_argument( + "--total-blocks", + type=int, + default=self.defaults.total_blocks, + help="Autoregressive chunks generated per rollout. Default: %(default)s.", + ) + parser.add_argument( + "--log-every-blocks", + type=int, + default=self.defaults.log_every_blocks, + help="Emit live timing every N steady-state chunks.", + ) + parser.add_argument( + "--warmup-blocks", + type=int, + default=self.defaults.warmup_blocks, + help="Leading chunks excluded from steady-state FPS.", + ) + parser.add_argument( + "--ui", + action=argparse.BooleanOptionalAction, + default=True, + help="Render the shared camera controls and timing overlay.", + ) + parser.add_argument( + "--compile", + action=argparse.BooleanOptionalAction, + default=None, + ) + parser.add_argument("--seed", type=int, default=None) + self._configure_argument_parser(parser) + args = parser.parse_args(list(commandline_args)) + + self._pipeline_config = self.defaults.pipeline_config + self._validate_arguments(args) + self._apply_parsed_arguments(args) + if args.compile is not None: + self._pipeline_config = derive_config( + self._pipeline_config, + diffusion_model={"transformer": {"compile_network": args.compile}}, + ) + if args.seed is not None: + self._pipeline_config = derive_config( + self._pipeline_config, + diffusion_model={"seed": args.seed}, + ) + self._device = args.device + self._total_blocks = args.total_blocks + self._log_every_blocks = args.log_every_blocks + self._warmup_blocks = args.warmup_blocks + self._use_ui = args.ui + self._input_values = { + "prompt": args.prompt, + "prompt_path": args.prompt_path, + "image_path": args.image_path, + "pose_path": args.pose_path, + "intrinsic_path": args.intrinsic_path, + "world_scale": args.world_scale, + "example_data": args.example_data, + "example_idx": args.example_idx, + "total_blocks": args.total_blocks, + } + + def session_desc(self) -> SessionDesc: + """Return the model's default output shape and interactive rates.""" + return SessionDesc( + output_layout=self.defaults.output_layout, + backpressure_mode=self.defaults.backpressure_mode, + presentation_mode=self.defaults.presentation_mode, + frames_per_second_for_ui=self.defaults.ui_fps, + frames_per_second_for_step=self.defaults.fps, + video_width=self.defaults.pixel_width, + video_height=self.defaults.pixel_height, + ) + + def create_session(self, session_desc: SessionDesc) -> ISession: + """Create an isolated rollout after lazily loading the shared pipeline.""" + input_values = self._input_values + if input_values is None: + raise RuntimeError( + f"{type(self).__name__}.init() must run before create_session()." + ) + self._validate_layout(session_desc) + resolved_values = { + **input_values, + "pixel_height": session_desc.video_height, + "pixel_width": session_desc.video_width, + "fps": session_desc.frames_per_second_for_step, + } + conditioning = self.defaults.input_resolver(resolved_values) + if not isinstance(conditioning, Cam2VConditioning): + raise TypeError( + "Cam2VApplicationDefaults.input_resolver must return Cam2VConditioning." + ) + + pipeline = self._pipeline + if pipeline is None: + pipeline = self._pipeline_config.setup().to(self._device).eval() + self._pipeline = pipeline + self._validate_frame_size(session_desc, pipeline) + return Cam2VSession( + pipeline=pipeline, + session_desc=session_desc, + config=Cam2VSessionConfig( + conditioning=conditioning, + total_blocks=self._total_blocks, + device=torch.device(self._device), + log_every_blocks=self._log_every_blocks, + warmup_blocks=self._warmup_blocks, + install_hint=self.defaults.install_hint, + ), + use_ui=self._use_ui, + ) + + def close(self) -> None: + """Release the application-owned pipeline after all sessions stop.""" + pipeline = self._pipeline + self._pipeline = None + self._input_values = None + close = getattr(pipeline, "close", None) + if callable(close): + close() + + def _configure_argument_parser(self, parser: argparse.ArgumentParser) -> None: + """Add integration-specific application arguments to ``parser``.""" + + def _apply_parsed_arguments(self, args: argparse.Namespace) -> None: + """Retain integration-specific arguments after shared validation.""" + + def _validate_arguments(self, args: argparse.Namespace) -> None: + """Reject invalid rollout and timing settings.""" + if args.total_blocks <= 0: + raise ValueError("--total-blocks must be > 0.") + if args.log_every_blocks <= 0: + raise ValueError("--log-every-blocks must be > 0.") + if args.warmup_blocks < 0: + raise ValueError("--warmup-blocks must be >= 0.") + if args.world_scale is not None and args.world_scale < 0: + raise ValueError("--world-scale must be >= 0 when set.") + + def _validate_layout(self, session_desc: SessionDesc) -> None: + """Reject output layouts that differ from the model's declared layout.""" + if session_desc.output_layout is not self.defaults.output_layout: + raise ValueError( + "This camera-to-video model only produces " + f"{self.defaults.output_layout.value} output, got " + f"{session_desc.output_layout.value}." + ) + if self._use_ui and session_desc.output_layout is not VideoTensorLayout.tchw: + raise ValueError("The Cam2V SlangPy UI overlay requires tchw output.") + + def _validate_frame_size(self, session_desc: SessionDesc, pipeline: Any) -> None: + """Reject frame dimensions that cannot map to integral latents.""" + decoder = getattr(pipeline, "decoder", None) + ratio = getattr(decoder, "spatial_compression_ratio", None) + if not isinstance(ratio, int) or ratio <= 0: + raise TypeError( + "Cam2V requires a decoder with a positive integer " + "spatial_compression_ratio." + ) + if session_desc.video_width % ratio or session_desc.video_height % ratio: + raise ValueError( + f"Frame dimensions must be multiples of {ratio}, got " + f"{session_desc.video_width}x{session_desc.video_height}." + ) + + +__all__ = ["Cam2VApplication"] diff --git a/apps/cam2v/assets/dummy_frame.ppm b/apps/cam2v/assets/dummy_frame.ppm new file mode 100644 index 000000000..1feee36e5 --- /dev/null +++ b/apps/cam2v/assets/dummy_frame.ppm @@ -0,0 +1,5 @@ +P3 +2 2 +255 +34 42 58 52 64 82 +52 64 82 34 42 58 diff --git a/apps/cam2v/controls.py b/apps/cam2v/controls.py new file mode 100644 index 000000000..647927207 --- /dev/null +++ b/apps/cam2v/controls.py @@ -0,0 +1,322 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Keyboard sampling and camera-pose integration shared by Cam2V apps.""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass, field +from typing import Literal + +import numpy as np + +from flashdreams.runtime.keyboard import DEFAULT_SUPPORTED_KEYS, KeyboardState + +PoseSegment = tuple[float, float, frozenset[str]] +"""One time interval and the camera-control keys held throughout it.""" + + +@dataclass(frozen=True, slots=True) +class _KeyboardEdge: + """One timestamped keyboard state transition.""" + + arrival_t: float + """Seconds since the WebRTC session began.""" + + event: str + """KeyboardState transition or the internal ``release_all`` action.""" + + key: str | None = None + """Browser key identifier, or ``None`` for ``release_all``.""" + + +class KeyboardResampler: + """Resample sparse key-down/key-up edges into a camera-control timeline.""" + + def __init__( + self, + *, + fps: float, + start_v: float = 0.0, + supported_keys: frozenset[str] = DEFAULT_SUPPORTED_KEYS, + ) -> None: + if fps <= 0: + raise ValueError("fps must be > 0") + self._fps = float(fps) + self._dt = 1.0 / self._fps + self._supported_keys = supported_keys + self.next_chunk_start_v = start_v + self._event_log: deque[_KeyboardEdge] = deque() + self._carried_state = KeyboardState(supported_keys=supported_keys) + + @property + def fps(self) -> float: + """Return the target camera sampling rate.""" + return self._fps + + @property + def dt(self) -> float: + """Return the interval between adjacent camera samples.""" + return self._dt + + def on_edge(self, *, arrival_t: float, event: str, key: str) -> None: + """Record one keyboard edge in timestamp order.""" + self._record_edge(_KeyboardEdge(arrival_t, event, key)) + + def release_all(self, *, arrival_t: float) -> None: + """Release every held key at ``arrival_t``, such as on focus loss.""" + self._record_edge(_KeyboardEdge(arrival_t, "release_all")) + + def _record_edge(self, edge: _KeyboardEdge) -> None: + """Insert ``edge`` while preserving timestamp order.""" + if not self._event_log or edge.arrival_t >= self._event_log[-1].arrival_t: + self._event_log.append(edge) + return + for index, queued_edge in enumerate(self._event_log): + if edge.arrival_t < queued_edge.arrival_t: + self._event_log.insert(index, edge) + return + self._event_log.append(edge) + + def _apply_edge(self, edge: _KeyboardEdge) -> None: + """Apply one queued edge to the carried keyboard state.""" + if edge.event == "release_all": + self._carried_state = KeyboardState( + supported_keys=self._supported_keys, + ) + return + assert edge.key is not None + self._carried_state.apply_event(event=edge.event, key=edge.key) + + def sample_chunk(self, num_frames: int) -> tuple[list[PoseSegment], list[float]]: + """Return key-state segments and sample times for one model chunk.""" + if num_frames < 1: + raise ValueError("num_frames must be >= 1") + + chunk_start_v = self.next_chunk_start_v + chunk_end_v = chunk_start_v + num_frames * self._dt + while self._event_log and self._event_log[0].arrival_t < chunk_start_v: + self._apply_edge(self._event_log.popleft()) + + segments: list[PoseSegment] = [] + previous_t = chunk_start_v + previous_state = self._carried_state.resolved_effective_keys() + while self._event_log and self._event_log[0].arrival_t <= chunk_end_v: + edge = self._event_log.popleft() + event_t = edge.arrival_t + if event_t > previous_t: + segments.append((previous_t, event_t, previous_state)) + self._apply_edge(edge) + previous_state = self._carried_state.resolved_effective_keys() + previous_t = event_t + if previous_t < chunk_end_v: + segments.append((previous_t, chunk_end_v, previous_state)) + elif not segments: + segments.append((chunk_start_v, chunk_end_v, previous_state)) + + frame_times = [ + chunk_start_v + (index + 1) * self._dt for index in range(num_frames) + ] + self.next_chunk_start_v = chunk_end_v + return segments, frame_times + + def reset(self, *, start_v: float) -> None: + """Discard queued edges and restart the virtual camera clock.""" + self._event_log.clear() + self._carried_state = KeyboardState(supported_keys=self._supported_keys) + self.next_chunk_start_v = start_v + + def event_log_size(self) -> int: + """Return the number of keyboard edges awaiting consumption.""" + return len(self._event_log) + + +def _rotation_matrix(axis: str, angle_rad: float) -> np.ndarray: + """Return a float32 three-dimensional rotation matrix.""" + cos_t = np.float32(np.cos(angle_rad)) + sin_t = np.float32(np.sin(angle_rad)) + if axis == "x": + return np.array( + [[1.0, 0.0, 0.0], [0.0, cos_t, -sin_t], [0.0, sin_t, cos_t]], + dtype=np.float32, + ) + if axis == "y": + return np.array( + [[cos_t, 0.0, sin_t], [0.0, 1.0, 0.0], [-sin_t, 0.0, cos_t]], + dtype=np.float32, + ) + if axis == "z": + return np.array( + [[cos_t, -sin_t, 0.0], [sin_t, cos_t, 0.0], [0.0, 0.0, 1.0]], + dtype=np.float32, + ) + return np.eye(3, dtype=np.float32) + + +@dataclass(slots=True) +class CameraPoseIntegrator: + """Integrate piecewise-constant keyboard intent into camera poses.""" + + move_speed_per_s: float = 0.8 + """Camera translation speed in world units per second.""" + + rotate_speed_rad_per_s: float = float(np.deg2rad(32.0)) + """Camera yaw and pitch speed in radians per second.""" + + pitch_limit_rad: float = float(np.deg2rad(85.0)) + """Maximum absolute camera pitch.""" + + coordinate_system: Literal["RDF", "FLU"] = "RDF" + """Camera basis: right-down-forward or forward-left-up.""" + + _current_pose: np.ndarray = field( + default_factory=lambda: np.eye(4, dtype=np.float32), + ) + _current_pitch: float = 0.0 + + def __post_init__(self) -> None: + if self.coordinate_system not in {"RDF", "FLU"}: + raise ValueError( + "coordinate_system must be 'RDF' (right-down-forward) " + "or 'FLU' (forward-left-up)" + ) + + def reset(self, pose: np.ndarray | None = None) -> None: + """Reset integration to identity or to ``pose``.""" + if pose is None: + self._current_pose = np.eye(4, dtype=np.float32) + self._current_pitch = 0.0 + return + if pose.shape != (4, 4): + raise ValueError(f"Expected pose shape (4, 4), got {pose.shape}") + self._current_pose = pose.astype(np.float32, copy=True) + if self.coordinate_system == "FLU": + self._current_pitch = float(np.arcsin(np.clip(pose[2, 0], -1.0, 1.0))) + else: + self._current_pitch = float(np.arctan2(pose[2, 1], pose[1, 1])) + + def current_pose(self) -> np.ndarray: + """Return a copy of the most recently integrated camera pose.""" + return self._current_pose.copy() + + def _advance(self, *, state: frozenset[str], duration: float) -> None: + """Advance the current pose through one constant-key interval.""" + if duration <= 0: + return + + yaw_rate = 0.0 + if self.coordinate_system == "FLU": + if "a" in state or "j" in state: + yaw_rate += self.rotate_speed_rad_per_s + if "d" in state or "l" in state: + yaw_rate -= self.rotate_speed_rad_per_s + else: + if "a" in state or "j" in state: + yaw_rate -= self.rotate_speed_rad_per_s + if "d" in state or "l" in state: + yaw_rate += self.rotate_speed_rad_per_s + pitch_rate = 0.0 + if "i" in state: + pitch_rate += self.rotate_speed_rad_per_s + if "k" in state: + pitch_rate -= self.rotate_speed_rad_per_s + + yaw_delta = yaw_rate * duration + pitch_delta = pitch_rate * duration + new_pitch = self._current_pitch + pitch_delta + if -self.pitch_limit_rad <= new_pitch <= self.pitch_limit_rad: + self._current_pitch = new_pitch + else: + pitch_delta = 0.0 + + rotation = self._current_pose[:3, :3] + translation = self._current_pose[:3, 3] + if self.coordinate_system == "FLU": + pitch_rotation = _rotation_matrix("y", -pitch_delta) + yaw_rotation = _rotation_matrix("z", yaw_delta) + else: + pitch_rotation = _rotation_matrix("x", pitch_delta) + yaw_rotation = _rotation_matrix("y", yaw_delta) + new_rotation = yaw_rotation @ rotation @ pitch_rotation + + forward_rate = 0.0 + if "w" in state: + forward_rate += self.move_speed_per_s + if "s" in state: + forward_rate -= self.move_speed_per_s + right_rate = 0.0 + if "e" in state: + right_rate += self.move_speed_per_s + if "q" in state: + right_rate -= self.move_speed_per_s + + if self.coordinate_system == "FLU": + forward = new_rotation[:, 0] + right = -new_rotation[:, 1] + flat_forward = np.array([forward[0], forward[1], 0.0], dtype=np.float32) + flat_right = np.array([right[0], right[1], 0.0], dtype=np.float32) + else: + right = new_rotation[:, 0] + forward = new_rotation[:, 2] + flat_forward = np.array([forward[0], 0.0, forward[2]], dtype=np.float32) + flat_right = np.array([right[0], 0.0, right[2]], dtype=np.float32) + forward_norm = np.linalg.norm(flat_forward) + right_norm = np.linalg.norm(flat_right) + if forward_norm > 0: + flat_forward /= forward_norm + if right_norm > 0: + flat_right /= right_norm + + movement = flat_forward * (forward_rate * duration) + flat_right * ( + right_rate * duration + ) + self._current_pose = np.eye(4, dtype=np.float32) + self._current_pose[:3, :3] = new_rotation + self._current_pose[:3, 3] = translation + movement + + def integrate_chunk( + self, + *, + segments: list[PoseSegment], + frame_times: list[float], + ) -> np.ndarray: + """Return one camera-to-world matrix at each requested frame time.""" + if not segments: + raise ValueError("segments must be non-empty") + if not frame_times: + raise ValueError("frame_times must be non-empty") + chunk_start = segments[0][0] + chunk_end = segments[-1][1] + if any( + frame_times[index] >= frame_times[index + 1] + for index in range(len(frame_times) - 1) + ): + raise ValueError("frame_times must be strictly increasing") + if frame_times[0] < chunk_start - 1e-9 or frame_times[-1] > chunk_end + 1e-9: + raise ValueError( + "frame_times must lie within the chunk window " + f"[{chunk_start}, {chunk_end}]" + ) + + poses: list[np.ndarray] = [] + current_t = chunk_start + frame_index = 0 + for _, segment_end, segment_state in segments: + while ( + frame_index < len(frame_times) + and frame_times[frame_index] <= segment_end + ): + target_t = frame_times[frame_index] + self._advance(state=segment_state, duration=target_t - current_t) + current_t = target_t + poses.append(self._current_pose.copy()) + frame_index += 1 + if segment_end > current_t: + self._advance(state=segment_state, duration=segment_end - current_t) + current_t = segment_end + + return np.stack(poses, axis=0).astype(np.float32) + + +__all__ = ["CameraPoseIntegrator", "KeyboardResampler", "PoseSegment"] diff --git a/apps/cam2v/defaults.py b/apps/cam2v/defaults.py new file mode 100644 index 000000000..6e30cc05e --- /dev/null +++ b/apps/cam2v/defaults.py @@ -0,0 +1,184 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration-owned defaults and resolved inputs for camera-to-video apps.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from types import MappingProxyType +from typing import Any + +import torch + +from flashdreams.runtime_v2.session_desc import BackpressureMode, PresentationMode +from flashdreams.runtime_v2.video_tensor import VideoTensorLayout + +Cam2VInputResolver = Callable[[Mapping[str, Any]], "Cam2VConditioning"] +"""Resolve application arguments into one session's camera conditioning.""" + + +@dataclass(frozen=True, kw_only=True, slots=True) +class Cam2VConditioning: + """Static camera-to-video conditioning resolved before a session starts.""" + + prompt: str + """Text condition used to initialize the model cache.""" + + first_frame_path: Path + """Image used to initialize the camera-to-video rollout.""" + + base_intrinsics: torch.Tensor + """Pixel-space camera intrinsics ``[fx, fy, cx, cy]``.""" + + world_scale: float + """Scale applied to camera translations by the model's camera encoder.""" + + def __post_init__(self) -> None: + first_frame_path = Path(self.first_frame_path) + intrinsics = torch.as_tensor(self.base_intrinsics, dtype=torch.float32) + if intrinsics.numel() != 4: + raise ValueError( + "Cam2VConditioning.base_intrinsics must contain four values." + ) + if self.world_scale < 0: + raise ValueError("Cam2VConditioning.world_scale must be >= 0.") + object.__setattr__(self, "prompt", self.prompt.strip()) + object.__setattr__(self, "first_frame_path", first_frame_path) + object.__setattr__(self, "base_intrinsics", intrinsics.reshape(1, 4).clone()) + + +@dataclass(frozen=True, kw_only=True, slots=True) +class Cam2VApplicationDefaults: + """Defaults one model integration contributes to the shared Cam2V app.""" + + pipeline_config: Any + """Model pipeline configuration owned by the integration.""" + + input_resolver: Cam2VInputResolver + """Integration hook that resolves paths and camera calibration.""" + + total_blocks: int + """Default number of autoregressive blocks in one rollout.""" + + pixel_width: int + """Default generated frame width.""" + + pixel_height: int + """Default generated frame height.""" + + device: str = "cuda" + """Device on which the application constructs the shared pipeline.""" + + fps: int = 16 + """Initial video frame rate and model-generation-loop pacing limit.""" + + output_layout: VideoTensorLayout = VideoTensorLayout.tchw + """Tensor layout emitted by the model pipeline.""" + + backpressure_mode: BackpressureMode = BackpressureMode.BLOCK + """Preserve every generated model frame in presentation order.""" + + presentation_mode: PresentationMode = PresentationMode.ONLY_PRESENT_NEW + """Render and transmit the overlay once for each selected model frame.""" + + ui_fps: int = 60 + """Rate at which the io-thread reads inputs and runs the UI loop.""" + + log_every_blocks: int = 1 + """Default interval between steady-state timing log records.""" + + warmup_blocks: int = 5 + """Leading blocks excluded from steady-state FPS.""" + + install_hint: str = "" + """Optional dependency hint included in first-frame loading failures.""" + + input_defaults: Mapping[str, Any] = field(default_factory=dict) + """Integration-owned default prompt, asset paths, and example selection.""" + + def __post_init__(self) -> None: + if self.total_blocks <= 0: + raise ValueError("Cam2VApplicationDefaults.total_blocks must be > 0.") + if self.pixel_width <= 0 or self.pixel_height <= 0: + raise ValueError("Cam2VApplicationDefaults dimensions must be > 0.") + if self.fps <= 0 or self.ui_fps <= 0: + raise ValueError("Cam2VApplicationDefaults frame rates must be > 0.") + if self.log_every_blocks <= 0: + raise ValueError("Cam2VApplicationDefaults.log_every_blocks must be > 0.") + if self.warmup_blocks < 0: + raise ValueError("Cam2VApplicationDefaults.warmup_blocks must be >= 0.") + object.__setattr__( + self, + "input_defaults", + MappingProxyType(dict(self.input_defaults)), + ) + + @classmethod + def from_runner_config( + cls, + runner_config: Any, + *, + input_resolver: Cam2VInputResolver, + total_blocks: int | None = None, + install_hint: str = "", + ) -> "Cam2VApplicationDefaults": + """Read shared application defaults from an integration runner config.""" + required = ["pipeline", "pixel_height", "pixel_width"] + if total_blocks is None: + required.append("total_blocks") + missing = [name for name in required if not hasattr(runner_config, name)] + if missing: + raise TypeError( + f"Runner config {type(runner_config).__name__} is missing " + f"camera-to-video application defaults: {missing}." + ) + input_names = ( + "prompt", + "prompt_path", + "image_path", + "pose_path", + "intrinsic_path", + "world_scale", + "example_data", + "example_idx", + ) + return cls( + pipeline_config=runner_config.pipeline, + input_resolver=input_resolver, + total_blocks=( + int(runner_config.total_blocks) + if total_blocks is None + else int(total_blocks) + ), + pixel_width=int(runner_config.pixel_width), + pixel_height=int(runner_config.pixel_height), + device=str(getattr(runner_config, "device", "cuda")), + fps=int(getattr(runner_config, "fps", 16)), + output_layout=_output_layout(runner_config), + install_hint=install_hint, + input_defaults={ + name: getattr(runner_config, name) + for name in input_names + if hasattr(runner_config, name) + }, + ) + + +def _output_layout(runner_config: Any) -> VideoTensorLayout: + """Return a runner's output layout using the v2 enum spelling.""" + value = getattr(runner_config, "postprocess_output_layout", None) + if value is None: + return VideoTensorLayout.tchw + if isinstance(value, VideoTensorLayout): + return value + return VideoTensorLayout(str(value)) + + +__all__ = [ + "Cam2VApplicationDefaults", + "Cam2VConditioning", + "Cam2VInputResolver", +] diff --git a/apps/cam2v/dummy.py b/apps/cam2v/dummy.py new file mode 100644 index 000000000..ccb93c9dd --- /dev/null +++ b/apps/cam2v/dummy.py @@ -0,0 +1,253 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Slow dummy model and application for interactive Cam2V UI testing.""" + +from __future__ import annotations + +import argparse +import threading +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import torch +from torch import Tensor + +from flashdreams.api_v2.application import IApplication + +from .application import Cam2VApplication +from .defaults import Cam2VApplicationDefaults, Cam2VConditioning +from .session import CameraControlInput + +_DUMMY_FRAME_PATH = Path(__file__).with_name("assets") / "dummy_frame.ppm" +"""Packaged first frame used by the dummy rollout.""" + + +@dataclass(frozen=True, slots=True) +class DummyCam2VCache: + """Per-rollout first frame retained by the dummy pipeline.""" + + first_frame: Tensor + """Normalized ``[C, H, W]`` background on the requested device.""" + + +class DummyCam2VDecoder: + """Expose the spatial contract required by :class:`Cam2VApplication`.""" + + spatial_compression_ratio = 1 + """Keep dummy pixels at their requested output resolution.""" + + +class DummyCam2VPipeline: + """Generate lightweight camera-tinted frames after a configurable wait.""" + + def __init__(self, *, step_wait_seconds: float, frames_per_chunk: int) -> None: + """Configure simulated model latency and chunk size. + + Args: + step_wait_seconds: Time each generation call waits. + frames_per_chunk: Frames returned by each generation call. + + Raises: + ValueError: The wait is negative or the chunk is empty. + """ + if step_wait_seconds < 0: + raise ValueError("step_wait_seconds must be >= 0.") + if frames_per_chunk <= 0: + raise ValueError("frames_per_chunk must be > 0.") + self.step_wait_seconds = float(step_wait_seconds) + self.frames_per_chunk = int(frames_per_chunk) + self.decoder = DummyCam2VDecoder() + self._device = torch.device("cpu") + self._sleep = threading.Event() + + def to(self, device: torch.device | str) -> "DummyCam2VPipeline": + """Select the device used for generated dummy frames.""" + self._device = torch.device(device) + return self + + def eval(self) -> "DummyCam2VPipeline": + """Return this stateless dummy pipeline in inference form.""" + return self + + def initialize_cache(self, *, text: list[str], image: Tensor) -> DummyCam2VCache: + """Retain one normalized first frame for a dummy rollout. + + Args: + text: Prompt accepted for Cam2V pipeline compatibility. + image: First frame shaped ``[1, C, H, W]``. + + Returns: + Per-rollout background state. + + Raises: + ValueError: ``image`` does not contain one RGB frame. + """ + del text + if image.ndim != 4 or image.shape[:2] != (1, 3): + raise ValueError("Dummy Cam2V requires one RGB first frame.") + return DummyCam2VCache( + first_frame=image[0].to(device=self._device, dtype=torch.float32) + ) + + def get_num_output_frames(self, autoregressive_index: int) -> int: + """Return the fixed dummy chunk size.""" + del autoregressive_index + return self.frames_per_chunk + + def generate( + self, + *, + autoregressive_index: int, + cache: DummyCam2VCache, + input: CameraControlInput, + ) -> Tensor: + """Wait like the real model and tint frames from camera motion. + + Args: + autoregressive_index: Zero-based dummy chunk index. + cache: Per-rollout first-frame state. + input: Integrated camera intrinsics and poses. + + Returns: + Normalized video shaped ``[T, C, H, W]``. + """ + self._sleep.wait(self.step_wait_seconds) + poses = input.poses.to(device=self._device, dtype=torch.float32) + translations = poses[:, :3, 3] + yaw = poses[:, 0, 2:3] + tint = torch.cat( + ( + translations[:, 0:1] + yaw, + translations[:, 1:2], + translations[:, 2:3] - yaw, + ), + dim=1, + ).tanh() + phase = torch.arange( + self.frames_per_chunk, + device=self._device, + dtype=torch.float32, + ) + phase = (phase + autoregressive_index * self.frames_per_chunk) % 32 + tint[:, 2] += (phase / 31.0 - 0.5) * 0.12 + background = cache.first_frame.unsqueeze(0).expand( + self.frames_per_chunk, -1, -1, -1 + ) + return (background + tint[:, :, None, None] * 0.35).clamp(-1.0, 1.0) + + def finalize( + self, + *, + autoregressive_index: int, + cache: DummyCam2VCache, + ) -> dict[str, float]: + """Return the configured synthetic generation latency.""" + del autoregressive_index, cache + return {"dummy_wait_s": self.step_wait_seconds} + + def close(self) -> None: + """Release any future dummy waits.""" + self._sleep.set() + + +@dataclass(frozen=True, slots=True) +class DummyCam2VPipelineConfig: + """Configuration that constructs a slow dummy Cam2V pipeline.""" + + step_wait_seconds: float = 0.9 + """Synthetic wall time for one model-generation step.""" + + frames_per_chunk: int = 12 + """Video frames produced by one model-generation step.""" + + def setup(self) -> DummyCam2VPipeline: + """Construct the configured dummy pipeline.""" + return DummyCam2VPipeline( + step_wait_seconds=self.step_wait_seconds, + frames_per_chunk=self.frames_per_chunk, + ) + + +def _resolve_dummy_conditioning(values: Mapping[str, Any]) -> Cam2VConditioning: + pixel_width = int(values["pixel_width"]) + pixel_height = int(values["pixel_height"]) + focal_length = float(max(pixel_width, pixel_height)) + return Cam2VConditioning( + prompt="dummy camera UI test", + first_frame_path=_DUMMY_FRAME_PATH, + base_intrinsics=torch.tensor( + [ + focal_length, + focal_length, + pixel_width / 2.0, + pixel_height / 2.0, + ] + ), + world_scale=1.0, + ) + + +class DummyCam2VApplication(Cam2VApplication): + """Run the shared Cam2V UI against a sleeping synthetic model.""" + + def __init__(self) -> None: + super().__init__( + defaults=Cam2VApplicationDefaults( + pipeline_config=DummyCam2VPipelineConfig(), + input_resolver=_resolve_dummy_conditioning, + total_blocks=10_000, + pixel_width=640, + pixel_height=360, + device="cuda", + fps=16, + ui_fps=60, + warmup_blocks=1, + install_hint="Install the Cam2V application with runner extras.", + ) + ) + + def _configure_argument_parser(self, parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--step-wait-seconds", + type=float, + default=0.9, + help="Wall time simulated by each dummy model step.", + ) + parser.add_argument( + "--frames-per-chunk", + type=int, + default=12, + help="Frames emitted by each dummy model step.", + ) + + def _apply_parsed_arguments(self, args: argparse.Namespace) -> None: + self._pipeline_config = DummyCam2VPipelineConfig( + step_wait_seconds=args.step_wait_seconds, + frames_per_chunk=args.frames_per_chunk, + ) + + def _validate_arguments(self, args: argparse.Namespace) -> None: + super()._validate_arguments(args) + if args.step_wait_seconds < 0: + raise ValueError("--step-wait-seconds must be >= 0.") + if args.frames_per_chunk <= 0: + raise ValueError("--frames-per-chunk must be > 0.") + if args.compile is not None or args.seed is not None: + raise ValueError("The dummy Cam2V model does not use --compile or --seed.") + + +def create_app() -> IApplication: + """Return the slow dummy Cam2V application.""" + return DummyCam2VApplication() + + +__all__ = [ + "DummyCam2VApplication", + "DummyCam2VCache", + "DummyCam2VPipeline", + "DummyCam2VPipelineConfig", + "create_app", +] diff --git a/apps/cam2v/pyproject.toml b/apps/cam2v/pyproject.toml new file mode 100644 index 000000000..a6c8ca27b --- /dev/null +++ b/apps/cam2v/pyproject.toml @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "flashdreams-cam2v" +version = "0.1.0" +description = "Reusable interactive FlashDreams camera-to-video application primitives" +readme = "README.md" +requires-python = ">=3.10" +dependencies = ["flashdreams[local-window,runners,serving]"] + +[project.entry-points."flashdreams.applications_v2"] +"cam2v-dummy" = "cam2v.dummy:create_app" + +[tool.uv.sources] +flashdreams = { workspace = true } + +[tool.setuptools] +packages = ["cam2v"] +package-dir = { cam2v = "." } + +[tool.setuptools.package-data] +cam2v = ["assets/*.ppm"] diff --git a/apps/cam2v/session.py b/apps/cam2v/session.py new file mode 100644 index 000000000..ab2809704 --- /dev/null +++ b/apps/cam2v/session.py @@ -0,0 +1,424 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Model-generation loop and session shared by camera-to-video apps.""" + +from __future__ import annotations + +import time +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +import torch +from loguru import logger + +from flashdreams.api_v2.loop import IModelLoop, invoke_async +from flashdreams.api_v2.session import ISession +from flashdreams.infra.runner_io import load_first_frame_tensor +from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.step_result import StepResult +from flashdreams.runtime_v2.user_input_event import ( + FocusUserInputEventData, + KeyboardInputState, + KeyboardUserInputEventData, +) +from flashdreams.runtime_v2.user_input_events import UserInputEvents + +from .controls import CameraPoseIntegrator, KeyboardResampler +from .defaults import Cam2VConditioning +from .ui import Cam2VSlangPyUILoop, Cam2VUIState, Cam2VUIStatus + + +@dataclass(kw_only=True, slots=True) +class CameraControlInput: + """Model-neutral per-step camera payload.""" + + intrinsics: torch.Tensor + """Per-frame intrinsics shaped ``[T, 4]``.""" + + poses: torch.Tensor + """Per-frame camera-to-world matrices shaped ``[T, 4, 4]``.""" + + world_scale: float + """Scale applied to camera translations by the model camera encoder.""" + + +@dataclass(frozen=True, kw_only=True, slots=True) +class Cam2VSessionConfig: + """Resolved immutable settings for one camera-to-video rollout.""" + + conditioning: Cam2VConditioning + """Prompt, first frame, and calibrated camera values.""" + + total_blocks: int + """Number of model steps generated before the rollout completes.""" + + device: torch.device + """Device holding model inputs and cache state.""" + + log_every_blocks: int + """Interval between live timing records after warmup.""" + + warmup_blocks: int + """Leading blocks excluded from steady-state FPS.""" + + install_hint: str = "" + """Optional first-frame loader hint for missing integration dependencies.""" + + def __post_init__(self) -> None: + if self.total_blocks <= 0: + raise ValueError("Cam2VSessionConfig.total_blocks must be > 0.") + if self.log_every_blocks <= 0: + raise ValueError("Cam2VSessionConfig.log_every_blocks must be > 0.") + if self.warmup_blocks < 0: + raise ValueError("Cam2VSessionConfig.warmup_blocks must be >= 0.") + + +@dataclass(slots=True) +class Cam2VModelState: + """Mutable rollout state owned exclusively by the model-generation-thread.""" + + pipeline: Any + """Application-owned, loaded model pipeline.""" + + session_desc: SessionDesc + """Output shape, layout, and rates accepted for this session.""" + + config: Cam2VSessionConfig + """Resolved inputs and rollout controls.""" + + keyboard_resampler: KeyboardResampler + """Timestamped camera-control state sampled on the model frame clock.""" + + cache: Any | None = None + """Session-local autoregressive model cache.""" + + blocks_generated: int = 0 + """Number of completed autoregressive model steps.""" + + frames_generated: int = 0 + """Number of generated video frames on the virtual camera clock.""" + + pose_integrator: CameraPoseIntegrator = field(default_factory=CameraPoseIntegrator) + """Session-local continuous camera state.""" + + steady_started_at: float | None = None + """Wall-clock origin immediately after excluded warmup blocks.""" + + steady_frames_generated: int = 0 + """Frames generated since :attr:`steady_started_at`.""" + + ui_loop: Cam2VSlangPyUILoop | None = None + """Registered UI-loop handle used only through ``invoke_async``.""" + + +class Cam2VModelLoop(IModelLoop[Cam2VModelState]): + """Generate one camera-controlled video chunk per model-loop iteration.""" + + def step(self, step_index: int, events: UserInputEvents) -> list[StepResult]: + """Apply new keyboard edges and generate one autoregressive block.""" + state = self.state + step_started_at = time.perf_counter() + if state.blocks_generated == state.config.warmup_blocks: + state.steady_started_at = step_started_at + + conditioning = state.config.conditioning + if state.cache is None: + first_frame = load_first_frame_tensor( + conditioning.first_frame_path, + pixel_height=state.session_desc.video_height, + pixel_width=state.session_desc.video_width, + device=state.config.device, + dtype=torch.bfloat16, + interpolation="cubic", + install_hint=state.config.install_hint, + ) + state.cache = state.pipeline.initialize_cache( + text=[conditioning.prompt], + image=first_frame, + ) + assert state.cache is not None + + frame_count = int(state.pipeline.get_num_output_frames(step_index)) + if frame_count <= 0: + raise ValueError( + "Cam2V pipelines must generate at least one frame per step." + ) + event_times = _buffer_keyboard_events(state.keyboard_resampler, events) + _catch_up_keyboard_timeline( + state.keyboard_resampler, + frame_count=frame_count, + event_times=event_times, + ) + segments, frame_times = state.keyboard_resampler.sample_chunk(frame_count) + poses = state.pose_integrator.integrate_chunk( + segments=segments, + frame_times=frame_times, + ) + camera_input = CameraControlInput( + intrinsics=conditioning.base_intrinsics.repeat(frame_count, 1).to( + device=state.config.device, + dtype=torch.float32, + ), + poses=torch.from_numpy(poses).to( + device=state.config.device, + dtype=torch.float32, + ), + world_scale=conditioning.world_scale, + ) + input_preparation_s = time.perf_counter() - step_started_at + + generate_started_at = time.perf_counter() + frames = state.pipeline.generate( + autoregressive_index=step_index, + cache=state.cache, + input=camera_input, + ) + generate_call_s = time.perf_counter() - generate_started_at + + finalize_started_at = time.perf_counter() + metrics = _numeric_metrics( + state.pipeline.finalize( + autoregressive_index=step_index, + cache=state.cache, + ) + ) + if state.config.device.type == "cuda" and torch.cuda.is_available(): + torch.cuda.current_stream(state.config.device).synchronize() + finalize_and_sync_s = time.perf_counter() - finalize_started_at + model_step_wall_s = time.perf_counter() - step_started_at + + state.blocks_generated += 1 + state.frames_generated += frame_count + metrics.update( + { + "input_prepare_s": input_preparation_s, + "generate_call_s": generate_call_s, + "finalize_and_sync_s": finalize_and_sync_s, + "model_step_wall_s": model_step_wall_s, + "chunk_fps": frame_count / model_step_wall_s, + } + ) + if state.steady_started_at is not None: + state.steady_frames_generated += frame_count + steady_elapsed_s = time.perf_counter() - state.steady_started_at + metrics["steady_state_fps"] = ( + state.steady_frames_generated / steady_elapsed_s + ) + if ( + state.steady_started_at is not None + and state.blocks_generated % state.config.log_every_blocks == 0 + ): + _log_step_timing( + step_index=step_index, + frame_count=frame_count, + metrics=metrics, + ) + _publish_ui_status(state, metrics) + return [ + StepResult( + step_index=step_index, + output=frames.detach(), + frame_count=frame_count, + output_layout=state.session_desc.output_layout, + metrics=metrics, + ) + ] + + def is_finished(self) -> bool: + """Return whether this rollout generated its requested blocks.""" + return self.state.blocks_generated >= self.state.config.total_blocks + + def reset(self) -> None: + """Discard model and camera state for a new session generation.""" + state = self.state + state.cache = None + state.blocks_generated = 0 + state.frames_generated = 0 + state.keyboard_resampler.reset(start_v=0.0) + state.pose_integrator.reset() + state.steady_started_at = None + state.steady_frames_generated = 0 + + def close(self) -> None: + """Release session-owned tensors while retaining the application model.""" + self.state.cache = None + + +class Cam2VSession(ISession): + """One camera-controlled rollout sharing its application's loaded model.""" + + def __init__( + self, + *, + pipeline: Any, + config: Cam2VSessionConfig, + session_desc: SessionDesc, + use_ui: bool = True, + ) -> None: + """Configure one rollout without initializing model or UI resources. + + Args: + pipeline: Application-owned model pipeline. + config: Resolved inputs and rollout controls. + session_desc: Output dimensions, layout, and loop rates. + use_ui: Whether to register the shared Cam2V overlay. + """ + self._pipeline = pipeline + self._config = config + self._session_desc = session_desc + self._use_ui = use_ui + + @property + def session_desc(self) -> SessionDesc: + """Return the resolved output dimensions and loop rates.""" + return self._session_desc + + def init(self) -> None: + """Register the UI and model-generation loops with isolated state.""" + ui_loop = None + if self._use_ui: + registered_ui = self.register_ui_loop( + Cam2VSlangPyUILoop, + state=Cam2VUIState( + total_blocks=self._config.total_blocks, + target_fps=self._session_desc.frames_per_second_for_step, + warmup_blocks=self._config.warmup_blocks, + ), + width=self._session_desc.video_width, + height=self._session_desc.video_height, + ) + assert isinstance(registered_ui, Cam2VSlangPyUILoop) + ui_loop = registered_ui + self.register_model_loop( + Cam2VModelLoop, + state=Cam2VModelState( + pipeline=self._pipeline, + session_desc=self._session_desc, + config=self._config, + keyboard_resampler=KeyboardResampler( + fps=self._session_desc.frames_per_second_for_step, + ), + ui_loop=ui_loop, + ), + ) + + +def _buffer_keyboard_events( + keyboard_resampler: KeyboardResampler, + events: UserInputEvents, +) -> list[float]: + """Queue timestamped WebRTC keyboard and focus edges for resampling.""" + event_times: list[float] = [] + for event in events.get_events(): + data = event.get_event_data() + event_t = float(event.get_timestamp()) / 1_000_000.0 + if isinstance(data, FocusUserInputEventData): + if not data.focused: + keyboard_resampler.release_all(arrival_t=event_t) + event_times.append(event_t) + continue + if not isinstance(data, KeyboardUserInputEventData): + continue + keyboard_resampler.on_edge( + arrival_t=event_t, + event=("keydown" if data.state is KeyboardInputState.PRESSED else "keyup"), + key=data.key, + ) + event_times.append(event_t) + return event_times + + +def _catch_up_keyboard_timeline( + keyboard_resampler: KeyboardResampler, + *, + frame_count: int, + event_times: list[float], +) -> None: + """Keep stale wall-clock input from waiting behind the model clock. + + Model warm-up and slow generation can leave the virtual camera timeline + behind WebRTC's session clock. If the newest unread edge lies beyond the + next chunk, skip only stale virtual time and retain up to one chunk of the + batch's original edge timing. + """ + if not event_times: + return + chunk_duration = frame_count * keyboard_resampler.dt + chunk_end = keyboard_resampler.next_chunk_start_v + chunk_duration + latest_event_t = max(event_times) + if latest_event_t <= chunk_end: + return + earliest_event_t = min(event_times) + keyboard_resampler.next_chunk_start_v = max( + keyboard_resampler.next_chunk_start_v, + earliest_event_t, + latest_event_t - chunk_duration, + ) + + +def _publish_ui_status( + state: Cam2VModelState, + metrics: Mapping[str, float | int], +) -> None: + """Send immutable model status to the UI loop without sharing state.""" + ui_loop = state.ui_loop + if ui_loop is None: + return + steady_state_fps = metrics.get("steady_state_fps") + status = Cam2VUIStatus( + completed_blocks=state.blocks_generated, + frames_generated=state.frames_generated, + chunk_fps=float(metrics["chunk_fps"]), + steady_state_fps=( + None if steady_state_fps is None else float(steady_state_fps) + ), + model_step_wall_s=float(metrics["model_step_wall_s"]), + ) + invoke_async( + ui_loop, + lambda ui_state, status=status: ui_state.update_status(status), + ) + + +def _numeric_metrics(stats: object) -> dict[str, float | int]: + """Keep numeric pipeline metrics accepted by the v2 result contract.""" + if not isinstance(stats, Mapping): + return {} + return { + str(name): value + for name, value in stats.items() + if isinstance(value, int | float) and not isinstance(value, bool) + } + + +def _log_step_timing( + *, + step_index: int, + frame_count: int, + metrics: Mapping[str, float | int], +) -> None: + """Log one chunk's wall-time breakdown and steady-state throughput.""" + logger.info( + "Cam2V block={} frames={} steady_state_fps={:.2f} chunk_fps={:.2f} " + "wall={:.3f}s input={:.3f}s generate_call={:.3f}s " + "finalize_and_sync={:.3f}s", + step_index, + frame_count, + metrics["steady_state_fps"], + metrics["chunk_fps"], + metrics["model_step_wall_s"], + metrics["input_prepare_s"], + metrics["generate_call_s"], + metrics["finalize_and_sync_s"], + ) + + +__all__ = [ + "Cam2VModelState", + "Cam2VModelLoop", + "Cam2VSession", + "Cam2VSessionConfig", + "CameraControlInput", +] diff --git a/apps/cam2v/tests/test_application.py b/apps/cam2v/tests/test_application.py new file mode 100644 index 000000000..c12273556 --- /dev/null +++ b/apps/cam2v/tests/test_application.py @@ -0,0 +1,535 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CPU tests for the shared camera-to-video v2 application.""" + +from __future__ import annotations + +import queue +import threading +from collections.abc import Mapping +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import Mock + +import cam2v.ui as cam2v_ui +import pytest +import tomli as tomllib +import torch +from cam2v import ( + Cam2VApplication, + Cam2VApplicationDefaults, + Cam2VConditioning, + Cam2VModelLoop, + Cam2VModelState, + Cam2VSession, + Cam2VSessionConfig, + Cam2VSlangPyUILoop, + Cam2VUIState, + Cam2VUIStatus, + CameraControlInput, + KeyboardResampler, +) +from cam2v.dummy import DummyCam2VPipelineConfig +from cam2v.dummy import create_app as create_dummy_app +from numpy import uint64 + +from flashdreams.runtime_v2.blit_model_output_to_screen_loop import ( + BlitModelOutputToScreenLoop, +) +from flashdreams.runtime_v2.presentation_manager import PresentationManager +from flashdreams.runtime_v2.session_desc import ( + BackpressureMode, + PresentationMode, + SessionDesc, +) +from flashdreams.runtime_v2.step_result import StepResult +from flashdreams.runtime_v2.user_input_event import ( + FocusUserInputEventData, + KeyboardInputState, + KeyboardUserInputEventData, + UserInputEvent, +) +from flashdreams.runtime_v2.user_input_events import UserInputEvents +from flashdreams.runtime_v2.video_tensor import VideoTensorLayout + +pytestmark = pytest.mark.ci_cpu + + +class _Decoder: + spatial_compression_ratio = 1 + + +class _Pipeline: + """Small CPU pipeline recording shared Cam2V inputs and lifecycle calls.""" + + def __init__(self) -> None: + self.decoder = _Decoder() + self.camera_input: CameraControlInput | None = None + self.device: str | None = None + self.closed = False + + def to(self, device: str) -> "_Pipeline": + """Record application-owned device placement.""" + self.device = device + return self + + def eval(self) -> "_Pipeline": + """Match the real pipeline construction chain.""" + return self + + def get_num_output_frames(self, step_index: int) -> int: + """Return a two-frame chunk for each model-generation step.""" + del step_index + return 2 + + def generate( + self, + *, + autoregressive_index: int, + cache: object, + input: CameraControlInput, + ) -> torch.Tensor: + """Record camera conditioning and return a deterministic chunk.""" + del autoregressive_index, cache + self.camera_input = input + return torch.zeros((2, 3, 1, 1), dtype=torch.float32) + + def finalize( + self, + *, + autoregressive_index: int, + cache: object, + ) -> dict[str, float]: + """Return one model-provided timing metric.""" + del autoregressive_index, cache + return {"model_step_s": 1.0} + + def close(self) -> None: + """Record application cleanup.""" + self.closed = True + + +class _PipelineConfig: + """Return one retained stand-in pipeline from ``setup``.""" + + def __init__(self) -> None: + self.pipeline = _Pipeline() + + def setup(self) -> _Pipeline: + """Return the application-owned pipeline.""" + return self.pipeline + + +def _conditioning() -> Cam2VConditioning: + return Cam2VConditioning( + prompt="camera demo", + first_frame_path=Path("first.jpg"), + base_intrinsics=torch.tensor([1.0, 1.0, 0.5, 0.5]), + world_scale=1.0, + ) + + +def test_model_loop_maps_wasd_to_shared_camera_input_and_metrics() -> None: + """Keep keyboard-to-pose conversion outside concrete integrations.""" + pipeline = _Pipeline() + ui_state = Cam2VUIState(total_blocks=1, target_fps=16, warmup_blocks=0) + shutdown_event = threading.Event() + failure_queue: queue.Queue[BaseException] = queue.Queue() + presentation_manager = PresentationManager() + ui_loop = Cam2VSlangPyUILoop(renderer=Mock()) + ui_loop.register_session_loop_objects( + state=ui_state, + frequency=60, + shutdown_event=shutdown_event, + failure_queue=failure_queue, + ) + ui_loop.register_session_ui_loop_objects( + output_layout=VideoTensorLayout.tchw, + presentation_manager=presentation_manager, + ) + state = Cam2VModelState( + pipeline=pipeline, + session_desc=SessionDesc( + output_layout=VideoTensorLayout.tchw, + frames_per_second_for_step=16, + video_width=1, + video_height=1, + ), + config=Cam2VSessionConfig( + conditioning=_conditioning(), + total_blocks=1, + device=torch.device("cpu"), + log_every_blocks=1, + warmup_blocks=0, + ), + keyboard_resampler=KeyboardResampler(fps=16), + cache=object(), + ui_loop=ui_loop, + ) + model_loop = Cam2VModelLoop() + model_loop.register_session_loop_objects( + state=state, + frequency=16, + shutdown_event=shutdown_event, + failure_queue=failure_queue, + ) + events = UserInputEvents( + [ + UserInputEvent( + timestamp=uint64(0), + event_data=KeyboardUserInputEventData( + key="w", + state=KeyboardInputState.PRESSED, + ), + ) + ] + ) + + result = model_loop.step(0, events)[0] + + assert result.frame_count == 2 + assert result.metrics["model_step_s"] == 1.0 + assert result.metrics["generate_call_s"] >= 0 + assert result.metrics["finalize_and_sync_s"] >= 0 + assert result.metrics["steady_state_fps"] > 0 + assert result.metrics["model_step_wall_s"] > 0 + ui_loop._run_message_batch() + assert ui_state.status is not None + assert ui_state.status.completed_blocks == 1 + assert ui_state.status.frames_generated == 2 + assert pipeline.camera_input is not None + assert pipeline.camera_input.poses.shape == (2, 4, 4) + assert pipeline.camera_input.poses[-1, 2, 3] > 0 + assert model_loop.is_finished() + + +def _input_test_model_loop() -> tuple[Cam2VModelLoop, Cam2VModelState, _Pipeline]: + """Return a registered CPU model loop for camera-input tests.""" + pipeline = _Pipeline() + state = Cam2VModelState( + pipeline=pipeline, + session_desc=SessionDesc( + output_layout=VideoTensorLayout.tchw, + frames_per_second_for_step=16, + video_width=1, + video_height=1, + ), + config=Cam2VSessionConfig( + conditioning=_conditioning(), + total_blocks=4, + device=torch.device("cpu"), + log_every_blocks=1, + warmup_blocks=4, + ), + keyboard_resampler=KeyboardResampler(fps=16), + cache=object(), + ) + model_loop = Cam2VModelLoop() + model_loop.register_session_loop_objects( + state=state, + frequency=16, + shutdown_event=threading.Event(), + failure_queue=queue.Queue(), + ) + return model_loop, state, pipeline + + +def test_model_loop_preserves_a_quick_tap_after_wall_clock_stall() -> None: + """Apply a short press in the next chunk even when the model clock lags.""" + model_loop, state, pipeline = _input_test_model_loop() + events = UserInputEvents( + [ + UserInputEvent( + timestamp=uint64(10_000_000), + event_data=KeyboardUserInputEventData( + key="w", + state=KeyboardInputState.PRESSED, + ), + ), + UserInputEvent( + timestamp=uint64(10_080_000), + event_data=KeyboardUserInputEventData( + key="w", + state=KeyboardInputState.RELEASED, + ), + ), + ] + ) + + model_loop.step(0, events) + + assert pipeline.camera_input is not None + assert pipeline.camera_input.poses[-1, 2, 3].item() == pytest.approx(0.064) + assert state.keyboard_resampler.next_chunk_start_v == pytest.approx(10.125) + + +def test_model_loop_releases_camera_controls_when_browser_loses_focus() -> None: + """Stop retained movement at the timestamped browser focus-loss edge.""" + model_loop, _, pipeline = _input_test_model_loop() + events = UserInputEvents( + [ + UserInputEvent( + timestamp=uint64(0), + event_data=KeyboardUserInputEventData( + key="w", + state=KeyboardInputState.PRESSED, + ), + ), + UserInputEvent( + timestamp=uint64(50_000), + event_data=FocusUserInputEventData(focused=False), + ), + ] + ) + + model_loop.step(0, events) + + assert pipeline.camera_input is not None + poses = pipeline.camera_input.poses + assert poses[0, 2, 3].item() == pytest.approx(0.04) + assert poses[1, 2, 3].item() == pytest.approx(0.04) + + +def test_model_loop_normalizes_browser_arrow_keys() -> None: + """Use the shared keyboard normalization without an input registry.""" + model_loop, _, pipeline = _input_test_model_loop() + events = UserInputEvents( + [ + UserInputEvent( + timestamp=uint64(0), + event_data=KeyboardUserInputEventData( + key="ArrowUp", + state=KeyboardInputState.PRESSED, + ), + ) + ] + ) + + model_loop.step(0, events) + + assert pipeline.camera_input is not None + assert pipeline.camera_input.poses[-1, 2, 3].item() == pytest.approx(0.1) + + +def test_slangpy_overlay_tracks_controls_and_model_status(monkeypatch: Any) -> None: + """Keep immediate input display in UI-loop-owned state.""" + logger = Mock() + monkeypatch.setattr(cam2v_ui, "logger", logger) + state = Cam2VUIState(total_blocks=4, target_fps=16, warmup_blocks=1) + presentation_manager = PresentationManager() + presentation_manager.publish( + 0, + [ + StepResult( + step_index=0, + output=torch.zeros((3, 3, 2, 2), dtype=torch.bfloat16), + frame_count=3, + output_layout=VideoTensorLayout.tchw, + ) + ], + ) + assert presentation_manager.advance(0)[0] + ui_loop = Cam2VSlangPyUILoop(renderer=Mock()) + ui_loop.register_session_loop_objects( + state=state, + frequency=60, + shutdown_event=threading.Event(), + failure_queue=queue.Queue(), + ) + ui_loop.register_session_ui_loop_objects( + output_layout=VideoTensorLayout.tchw, + presentation_manager=presentation_manager, + ) + ui = SimpleNamespace( + screen=object(), + Window=Mock(return_value=object()), + Text=Mock(side_effect=lambda parent, text: SimpleNamespace(text=text)), + ) + events = UserInputEvents( + [ + UserInputEvent( + timestamp=uint64(0), + event_data=KeyboardUserInputEventData( + key="ArrowUp", + state=KeyboardInputState.PRESSED, + ), + ) + ] + ) + state.update_status( + Cam2VUIStatus( + completed_blocks=2, + frames_generated=24, + chunk_fps=13.5, + steady_state_fps=13.25, + model_step_wall_s=0.89, + ) + ) + + back_buffer = ui_loop.step_ui(ui, 0, events) + + assert back_buffer is not None + assert back_buffer.dtype is torch.float32 + displayed = [widget.text for widget in state.status_widgets] + assert "Rollout: 2/4 blocks" in displayed + assert "Presented: 1 frames (24 generated)" in displayed + assert "Latest model rate: 13.50 FPS" in displayed + assert state.active_keys_widget is not None + assert state.active_keys_widget.text == "Active keys: W" + logger.info.assert_called_once_with( + "Cam2V SlangPy UI loop processed keyboard event " + "key={} state={} timestamp_us={} held_keys={}", + "w", + "Pressed", + 0, + "w", + ) + + assert presentation_manager.advance(0)[0] + ui_loop.step_ui(ui, 1, UserInputEvents([])) + displayed = [widget.text for widget in state.status_widgets] + assert "Presented: 2 frames (24 generated)" in displayed + + +def test_cam2v_session_registers_the_shared_slangpy_ui_loop() -> None: + """Construct the overlay at the shared session boundary.""" + session = Cam2VSession( + pipeline=_Pipeline(), + session_desc=SessionDesc( + output_layout=VideoTensorLayout.tchw, + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, + frames_per_second_for_step=16, + video_width=8, + video_height=4, + ), + config=Cam2VSessionConfig( + conditioning=_conditioning(), + total_blocks=2, + device=torch.device("cpu"), + log_every_blocks=1, + warmup_blocks=0, + ), + ) + + session.init() + + assert isinstance(session.ui_loop, Cam2VSlangPyUILoop) + assert session.ui_loop.state.total_blocks == 2 + assert session.model_loop.state.keyboard_resampler.fps == 16 + assert session.session_desc.backpressure_mode is BackpressureMode.BLOCK + assert session.session_desc.presentation_mode is PresentationMode.ONLY_PRESENT_NEW + + +def test_application_owns_pipeline_and_resolves_inputs_per_session_desc() -> None: + """Keep the loaded model application-scoped and rollout inputs session-scoped.""" + pipeline_config = _PipelineConfig() + seen: list[Mapping[str, Any]] = [] + + def resolve(values: Mapping[str, Any]) -> Cam2VConditioning: + seen.append(values) + return _conditioning() + + app = Cam2VApplication( + defaults=Cam2VApplicationDefaults( + pipeline_config=pipeline_config, + input_resolver=resolve, + total_blocks=3, + pixel_width=8, + pixel_height=4, + device="cpu", + fps=16, + ) + ) + app.init(["--total-blocks", "2", "--warmup-blocks", "0", "--no-ui"]) + + session = app.create_session(app.session_desc()) + + assert isinstance(session, Cam2VSession) + session.init() + ui_loop, _ = session._take_loops() + assert session.session_desc.video_width == 8 + assert isinstance(ui_loop, BlitModelOutputToScreenLoop) + assert seen[0]["pixel_width"] == 8 + assert seen[0]["pixel_height"] == 4 + assert seen[0]["fps"] == 16 + assert pipeline_config.pipeline.device == "cpu" + app.close() + assert pipeline_config.pipeline.closed + + +def test_defaults_reject_invalid_timing_configuration() -> None: + """Fail before model construction when timing defaults are invalid.""" + with pytest.raises(ValueError, match="warmup_blocks"): + Cam2VApplicationDefaults( + pipeline_config=object(), + input_resolver=lambda values: _conditioning(), + total_blocks=1, + pixel_width=1, + pixel_height=1, + warmup_blocks=-1, + ) + + +def test_dummy_cam2v_pipeline_simulates_generation_without_a_model() -> None: + """Exercise camera-dependent dummy output without image or GPU dependencies.""" + pipeline = DummyCam2VPipelineConfig( + step_wait_seconds=0.0, + frames_per_chunk=2, + ).setup() + cache = pipeline.initialize_cache( + text=["dummy"], + image=torch.zeros((1, 3, 4, 8), dtype=torch.float32), + ) + poses = torch.eye(4).repeat(2, 1, 1) + poses[:, 0, 3] = 1.0 + + frames = pipeline.generate( + autoregressive_index=0, + cache=cache, + input=CameraControlInput( + intrinsics=torch.zeros((2, 4)), + poses=poses, + world_scale=1.0, + ), + ) + + assert frames.shape == (2, 3, 4, 8) + assert frames[:, 0].mean() > frames[:, 1].mean() + + +def test_dummy_cam2v_application_exposes_slow_step_controls() -> None: + """Keep synthetic latency configurable through application arguments.""" + app = create_dummy_app() + + app.init( + [ + "--step-wait-seconds", + "0.25", + "--frames-per-chunk", + "3", + "--total-blocks", + "1", + ] + ) + + assert isinstance(app, Cam2VApplication) + assert app.session_desc().backpressure_mode is BackpressureMode.BLOCK + assert app.session_desc().presentation_mode is PresentationMode.ONLY_PRESENT_NEW + assert app.pipeline_config == DummyCam2VPipelineConfig( + step_wait_seconds=0.25, + frames_per_chunk=3, + ) + + +def test_shared_cam2v_package_registers_the_dummy_application() -> None: + """Expose the slow dummy through the v2 application registry.""" + manifest = tomllib.loads((Path(__file__).parents[1] / "pyproject.toml").read_text()) + + assert ( + manifest["project"]["entry-points"]["flashdreams.applications_v2"][ + "cam2v-dummy" + ] + == "cam2v.dummy:create_app" + ) diff --git a/apps/cam2v/ui.py b/apps/cam2v/ui.py new file mode 100644 index 000000000..fd8820da6 --- /dev/null +++ b/apps/cam2v/ui.py @@ -0,0 +1,219 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""SlangPy status and camera-control overlay for Cam2V applications.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +import torch +from loguru import logger +from torch import Tensor + +from flashdreams.runtime.keyboard import normalize_key +from flashdreams.runtime_v2.slangpy_ui_loop import SlangPyUILoop +from flashdreams.runtime_v2.user_input_event import ( + FocusUserInputEventData, + KeyboardInputState, + KeyboardUserInputEventData, +) +from flashdreams.runtime_v2.user_input_events import UserInputEvents + +_CAMERA_KEY_ORDER = ("w", "s", "q", "e", "a", "d", "j", "l", "i", "k") +"""Stable order used when active camera controls are displayed.""" + +_CAMERA_KEYS = frozenset(_CAMERA_KEY_ORDER) +"""Keyboard controls recognized by the shared camera pose integrator.""" + + +@dataclass(frozen=True, slots=True) +class Cam2VUIStatus: + """Latest model-generation status copied to the UI loop.""" + + completed_blocks: int + """Number of autoregressive blocks completed in this rollout.""" + + frames_generated: int + """Number of video frames generated in this rollout.""" + + chunk_fps: float + """Frame throughput measured across the latest model step.""" + + steady_state_fps: float | None + """Cumulative post-warmup throughput, or ``None`` during warmup.""" + + model_step_wall_s: float + """Wall time spent producing the latest model chunk.""" + + +@dataclass(slots=True) +class Cam2VUIState: + """Mutable Cam2V overlay state owned exclusively by the UI loop.""" + + total_blocks: int + """Number of autoregressive blocks requested for the rollout.""" + + target_fps: int + """Configured generated-video frame rate.""" + + warmup_blocks: int + """Leading blocks excluded from steady-state throughput.""" + + held_keys: set[str] = field(default_factory=set) + """Camera-control keys currently held by the client.""" + + status: Cam2VUIStatus | None = None + """Latest model status received from the model-generation loop.""" + + frames_presented: int = 0 + """Number of model frames selected by the io-thread in this rollout.""" + + window: Any | None = field(default=None, init=False, repr=False) + """Retained SlangPy controls window.""" + + status_widgets: list[Any] = field(default_factory=list, init=False, repr=False) + """Retained SlangPy text widgets for model status.""" + + active_keys_widget: Any | None = field(default=None, init=False, repr=False) + """Retained SlangPy text widget for active camera controls.""" + + def update_status(self, status: Cam2VUIStatus) -> None: + """Replace the displayed model-generation status.""" + self.status = status + + def reset(self) -> None: + """Clear transient controls and model status for a new generation.""" + self.held_keys.clear() + self.status = None + self.frames_presented = 0 + + +class Cam2VSlangPyUILoop(SlangPyUILoop[Cam2VUIState]): + """Draw Cam2V controls and model throughput over generated video.""" + + def step_ui( + self, + ui: Any, + step_index: int, + events: UserInputEvents, + ) -> Tensor | None: + """Update retained widgets and return the current model frame.""" + del step_index + _apply_ui_input(self.state, events) + frame = self.presented_model_frame() + self.state.frames_presented = self._presentation_manager.presented_frame_count + _ensure_widgets(ui, self.state) + _refresh_widgets(self.state) + + if frame is None: + return None + if frame.is_floating_point(): + return frame.to(torch.float32) + return frame.to(torch.float32).mul_(2.0 / 255.0).sub_(1.0) + + def reset(self) -> None: + """Clear UI-loop state for a new generation.""" + self.state.reset() + super().reset() + + +def _ensure_widgets(ui: Any, state: Cam2VUIState) -> None: + if state.window is not None: + return + state.window = ui.Window( + ui.screen, + "Camera controls", + position=(16, 16), + size=(360, 280), + ) + state.status_widgets = [ + ui.Text(state.window, line) for line in _status_lines(state) + ] + ui.Text(state.window, "Move: W/S Strafe: Q/E") + ui.Text(state.window, "Yaw: A/D or J/L Pitch: I/K") + state.active_keys_widget = ui.Text(state.window, _active_keys_text(state)) + ui.Text(state.window, "Click the video before using keyboard controls.") + + +def _refresh_widgets(state: Cam2VUIState) -> None: + for widget, line in zip( + state.status_widgets, + _status_lines(state), + strict=True, + ): + widget.text = line + if state.active_keys_widget is not None: + state.active_keys_widget.text = _active_keys_text(state) + + +def _status_lines(state: Cam2VUIState) -> tuple[str, ...]: + status = state.status + if status is None: + return ( + "Waiting for the first generated chunk...", + f"Presented: {state.frames_presented} frames", + "Latest model rate: waiting", + f"Steady state: warming up (0/{state.warmup_blocks})", + f"Target video rate: {state.target_fps} FPS", + "Latest model step: waiting", + ) + + if status.steady_state_fps is None: + warmup_done = min(status.completed_blocks, state.warmup_blocks) + steady_state = f"Steady state: warming up ({warmup_done}/{state.warmup_blocks})" + else: + steady_state = f"Steady-state model rate: {status.steady_state_fps:.2f} FPS" + return ( + f"Rollout: {status.completed_blocks}/{state.total_blocks} blocks", + f"Presented: {state.frames_presented} frames " + f"({status.frames_generated} generated)", + f"Latest model rate: {status.chunk_fps:.2f} FPS", + steady_state, + f"Target video rate: {state.target_fps} FPS", + f"Latest model step: {status.model_step_wall_s * 1_000.0:.0f} ms", + ) + + +def _active_keys_text(state: Cam2VUIState) -> str: + active = [key.upper() for key in _CAMERA_KEY_ORDER if key in state.held_keys] + return f"Active keys: {', '.join(active) if active else 'none'}" + + +def _apply_ui_input(state: Cam2VUIState, events: UserInputEvents) -> None: + for event in events.get_events(): + data = event.get_event_data() + if isinstance(data, FocusUserInputEventData) and not data.focused: + state.held_keys.clear() + continue + if not isinstance(data, KeyboardUserInputEventData): + continue + key = normalize_key(data.key) + if key not in _CAMERA_KEYS: + logger.info( + "Cam2V SlangPy UI loop ignored keyboard event " + "key={} state={} timestamp_us={} reason=unsupported", + data.key, + data.state.value, + int(event.get_timestamp()), + ) + continue + if data.state is KeyboardInputState.PRESSED: + state.held_keys.add(key) + else: + state.held_keys.discard(key) + held_keys = ",".join( + item for item in _CAMERA_KEY_ORDER if item in state.held_keys + ) + logger.info( + "Cam2V SlangPy UI loop processed keyboard event " + "key={} state={} timestamp_us={} held_keys={}", + key, + data.state.value, + int(event.get_timestamp()), + held_keys or "none", + ) + + +__all__ = ["Cam2VSlangPyUILoop", "Cam2VUIState", "Cam2VUIStatus"] diff --git a/flashdreams/flashdreams/api_v2/README.md b/flashdreams/flashdreams/api_v2/README.md index b5d9269b8..07caac287 100644 --- a/flashdreams/flashdreams/api_v2/README.md +++ b/flashdreams/flashdreams/api_v2/README.md @@ -50,6 +50,11 @@ which draws every model channel into one frame as if they were image layers. Each loop is registered with the state it owns, and the call returns the loop: +| Runs on | Calls | Owns | Frame rate | +| --- | --- | --- | --- | +| The io-thread | `IUILoop.step` | UI-loop state, `run_session` state | `frames_per_second_for_ui` | +| The model-generation-thread | `IModelLoop.step` | Model-loop state and model logic | `frames_per_second_for_step` | + ```python self.register_model_loop(ModelLoop, state=ModelState(self._desc)) ``` @@ -58,6 +63,14 @@ self.register_model_loop(ModelLoop, state=ModelState(self._desc)) comes from the session description: the model loop steps at `frames_per_second_for_step`, and the UI ticks at `frames_per_second_for_ui`. +The io-thread initially selects frames from model chunks at +`frames_per_second_for_step`, then uses the model-generation-thread's rolling +two-second output rate. This paces chunked output evenly without tying input +and UI redraws to model throughput. +`PresentationMode.ONLY_PRESENT_NEWEST` lets an `IUILoop` redraw continuously; +`PresentationMode.ONLY_PRESENT_NEW` runs it only when the selected model frame +changes. + ## Loops The two loops run on different threads, so neither should reach into the other's @@ -151,6 +164,26 @@ This loop runs forever. Override `is_finished` to make it stop. ## Writing a UI loop +`SessionDesc.backpressure_mode` handles the model-generation-loop producing +frames faster than the io-thread can consume them: + +- `BackpressureMode.BLOCK` waits when the presentation queue is full. This keeps + every generated frame and can slow the model-generation-thread to the + io-thread's pace. +- `BackpressureMode.DROP_OLDEST` discards old buffered work so the UI can catch + up to newer output. This favors low latency over preserving every frame. + +`SessionDesc.presentation_mode` handles the UI loop ticking faster than the +model-generation-loop produces frames: + +- `PresentationMode.ONLY_PRESENT_NEWEST` runs the UI every tick and may reuse + the newest generated model frame. +- `PresentationMode.ONLY_PRESENT_NEW` runs the UI only after the presentation + manager advances to a new model frame. + +Use `PresentationMode.ONLY_PRESENT_NEW` with `BackpressureMode.BLOCK` when every +generated frame must be presented exactly once and in order. + For widgets drawn over the model output, subclass `SlangPyUILoop` from `flashdreams.runtime_v2.slangpy_ui_loop` and implement `step_ui(ui, step_index, events)` rather than `step`. The diff --git a/flashdreams/flashdreams/runtime_v2/presentation_manager.py b/flashdreams/flashdreams/runtime_v2/presentation_manager.py index 64b6190b1..08a4bff19 100644 --- a/flashdreams/flashdreams/runtime_v2/presentation_manager.py +++ b/flashdreams/flashdreams/runtime_v2/presentation_manager.py @@ -15,13 +15,13 @@ class PresentationManager: - """Buffer model output for a session's UI thread. + """Buffer model output for a session's io-thread. - The model thread publishes a chunk of channels per step into a bounded - queue; the UI thread calls :meth:`advance` once per tick to move to the next - frame. A chunk holding several frames is walked frame by frame before - another is taken, so a step that generated twelve frames is presented over - twelve ticks rather than eleven being skipped. + The model-generation-thread publishes a chunk of channels per step into a + bounded queue; the io-thread calls :meth:`advance` once per tick to move to + the next frame. A chunk holding several frames is walked frame by frame + before another is taken, so a step that generated twelve frames is + presented over twelve ticks rather than eleven being skipped. :class:`BackpressureMode` decides what publishing does when the queue is full. Chunks that could not be kept are counted in @@ -37,6 +37,7 @@ def __init__(self) -> None: self._generation = 0 self._presented_chunk: list[StepResult] | None = None self._frame_index = -1 + self._presented_frame_count = 0 self.dropped_for_space = 0 """Chunks dropped because the UI could not keep up with the model.""" @@ -80,9 +81,9 @@ def publish( ) -> None: """Add one completed model step to the presentation queue. - Called on the model thread. ``BLOCK`` waits here when the queue is full, - until there is room or the session stops; ``DROP_OLDEST`` evicts instead - and returns. + Called on the model-generation-thread. ``BLOCK`` waits here when the + queue is full, until there is room or the session stops; + ``DROP_OLDEST`` evicts instead and returns. Args: generation: Reset generation the chunk was generated in. A chunk @@ -115,7 +116,7 @@ def publish( def advance(self, generation: int) -> tuple[bool, list[StepResult] | None]: """Move to the next model frame, if one is available. - Called on the UI thread, once per tick. A ``generation`` other than the + Called on the io-thread, once per tick. A ``generation`` other than the last one seen drops what is being presented, so nothing generated before a reset survives it. @@ -131,33 +132,32 @@ def advance(self, generation: int) -> tuple[bool, list[StepResult] | None]: self._generation = generation self._presented_chunk = None self._frame_index = -1 - - if self._backpressure_mode is BackpressureMode.DROP_OLDEST: - chunk = self._take_buffered_chunk(generation, latest=True) - if chunk is not None: - if ( - self._presented_chunk is not None - and self._frame_index + 1 < self._presented_chunk[0].frame_count - ): - self.dropped_for_space += 1 - self._presented_chunk = chunk - self._frame_index = 0 - return True, chunk + self._presented_frame_count = 0 if ( self._presented_chunk is not None and self._frame_index + 1 < self._presented_chunk[0].frame_count ): self._frame_index += 1 + self._presented_frame_count += 1 return True, None - chunk = self._take_buffered_chunk(generation, latest=False) + chunk = self._take_buffered_chunk( + generation, + latest=self._backpressure_mode is BackpressureMode.DROP_OLDEST, + ) if chunk is None: return False, None self._presented_chunk = chunk self._frame_index = 0 + self._presented_frame_count += 1 return True, chunk + @property + def presented_frame_count(self) -> int: + """Return frames selected one-by-one in the current generation.""" + return self._presented_frame_count + def presented_frame(self, channel_index: int) -> Tensor | None: """Return the current ``[C, H, W]`` frame from one model channel.""" if self._presented_chunk is None: @@ -210,6 +210,7 @@ def clear(self) -> None: """Discard buffered and currently presented model results.""" self._presented_chunk = None self._frame_index = -1 + self._presented_frame_count = 0 while True: try: self._buffer.get_nowait() diff --git a/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py b/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py index aa8c08781..ac384fc60 100644 --- a/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py +++ b/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py @@ -12,9 +12,10 @@ import threading import time from collections.abc import Callable +from dataclasses import dataclass from fractions import Fraction from importlib.resources import files -from typing import Any +from typing import Any, TypeAlias import numpy as np import torch @@ -22,8 +23,9 @@ from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription from aiortc.mediastreams import MediaStreamError from av import VideoFrame +from loguru import logger -from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.runtime_v2.step_result import StepResult from flashdreams.runtime_v2.user_input_event import ( CloseUserInputEventData, @@ -40,52 +42,168 @@ _BROWSER_PAGE = _WEB_RESOURCES.joinpath("index.html").read_text(encoding="utf-8") _BROWSER_SCRIPT = _WEB_RESOURCES.joinpath("app.js").read_text(encoding="utf-8") +_CUDA_EVENT_POLL_SECONDS = 0.001 +"""Polling interval that keeps CUDA waits off the WebRTC event-loop thread.""" + +_INTERACTIVE_FRAME_QUEUE_SIZE = 1 +"""Pending sender frames retained for latest-frame presentation.""" + +_RGBArray: TypeAlias = np.ndarray[Any, np.dtype[np.uint8]] + + +@dataclass(frozen=True, slots=True) +class _PendingRGBFrame: + """Pinned host frame whose asynchronous CUDA transfer is in flight.""" + + host_frames: torch.Tensor + """Pinned ``[T, H, W, C]`` uint8 storage shared by one result.""" + + frame_index: int + """Frame selected from ``host_frames`` after the transfer completes.""" + + ready_event: torch.cuda.Event + """Event recorded after the device-to-host transfer.""" + + async def resolve(self) -> _RGBArray: + """Wait without blocking the event loop and return the host array.""" + while not self.ready_event.query(): + await asyncio.sleep(_CUDA_EVENT_POLL_SECONDS) + return np.asarray(self.host_frames[self.frame_index].numpy()) + + +_QueuedRGBFrame: TypeAlias = _RGBArray | _PendingRGBFrame + + +@dataclass(frozen=True, slots=True) +class _PresentedRGBFrame: + """One prepared frame with its io-thread presentation time.""" + + frame: _QueuedRGBFrame + """RGB pixels, possibly awaiting an asynchronous CUDA transfer.""" + + presented_at: float + """Event-loop timestamp at which the io-thread submitted this frame.""" + + +class _FramePacer: + """Pace source frames against drift-free absolute deadlines.""" + + def __init__(self, frames_per_second: int) -> None: + self._minimum_interval = 1.0 / frames_per_second + self._last_source_at: float | None = None + self._next_frame_at: float | None = None + + def delay_seconds(self, *, now: float, source_at: float) -> float: + """Return the delay before presenting one source frame. + + Small scheduling overruns are recovered by the next absolute deadline + instead of accumulating. A stall of at least one frame interval + reanchors the schedule so queued frames are not emitted in a burst. + """ + last_source_at = self._last_source_at + next_frame_at = self._next_frame_at + self._last_source_at = source_at + if last_source_at is None or next_frame_at is None: + self._next_frame_at = now + return 0.0 + + source_interval = max( + self._minimum_interval, + source_at - last_source_at, + ) + next_frame_at += source_interval + if now - next_frame_at >= self._minimum_interval: + next_frame_at = now + self._next_frame_at = next_frame_at + return max(0.0, next_frame_at - now) + class _VideoTrack(MediaStreamTrack): """Video track whose frames are supplied by the server.""" kind = "video" - def __init__(self, frames_per_second: int) -> None: + def __init__(self, frames_per_second: int, *, drop_oldest: bool = False) -> None: + """Configure frame pacing and optional latest-frame delivery. + + Args: + frames_per_second: RTP clock and maximum delivery rate. + drop_oldest: Whether a newly presented frame replaces a queued one. + """ super().__init__() self._frames_per_second = frames_per_second + self._frame_interval = 1.0 / frames_per_second self._time_base = Fraction(1, frames_per_second) - self._frames: asyncio.Queue[np.ndarray[Any, np.dtype[np.uint8]] | None] = ( - asyncio.Queue() + self._drop_oldest = drop_oldest + self._frames: asyncio.Queue[_PresentedRGBFrame | None] = asyncio.Queue( + maxsize=_INTERACTIVE_FRAME_QUEUE_SIZE if drop_oldest else 0 ) - self._next_frame_time: float | None = None - self._pts = 0 + self._retired_pending_frames: list[_PendingRGBFrame] = [] + self._dropped_for_lag = 0 + self._pacer = _FramePacer(frames_per_second) + self._presentation_started_at: float | None = None + self._next_pts = 0 self._closed = False - async def enqueue( - self, frames: tuple[np.ndarray[Any, np.dtype[np.uint8]], ...] - ) -> None: + @property + def dropped_for_lag(self) -> int: + """Return the number of stale sender frames replaced before encoding.""" + return self._dropped_for_lag + + def qsize(self) -> int: + """Return the number of frames waiting for the WebRTC sender.""" + return self._frames.qsize() + + async def enqueue(self, frames: tuple[_QueuedRGBFrame, ...]) -> None: """Append generated RGB frames for the WebRTC sender.""" if self._closed: return - for frame in frames: - await self._frames.put(frame) + self._reap_retired_pending_frames() + presented_at = asyncio.get_running_loop().time() + for frame_index, frame in enumerate(frames): + presented_frame = _PresentedRGBFrame( + frame=frame, + presented_at=presented_at + frame_index * self._frame_interval, + ) + if self._drop_oldest: + self._drop_queued_frame() + self._frames.put_nowait(presented_frame) + else: + await self._frames.put(presented_frame) async def recv(self) -> VideoFrame: """Return the next generated frame when aiortc requests one.""" if self._closed: raise MediaStreamError - frame = await self._frames.get() - if frame is None: + presented_frame = await self._frames.get() + if presented_frame is None: raise MediaStreamError + self._reap_retired_pending_frames() + queued_frame = presented_frame.frame + frame = ( + await queued_frame.resolve() + if isinstance(queued_frame, _PendingRGBFrame) + else queued_frame + ) loop = asyncio.get_running_loop() now = loop.time() - if self._next_frame_time is None: - self._next_frame_time = now - else: - self._next_frame_time += 1.0 / self._frames_per_second - await asyncio.sleep(max(0.0, self._next_frame_time - now)) + wait_seconds = self._pacer.delay_seconds( + now=now, + source_at=presented_frame.presented_at, + ) + if wait_seconds > 0: + await asyncio.sleep(wait_seconds) + + if self._presentation_started_at is None: + self._presentation_started_at = presented_frame.presented_at + elapsed = presented_frame.presented_at - self._presentation_started_at + pts = max(self._next_pts, round(elapsed * self._frames_per_second)) video_frame = VideoFrame.from_ndarray(frame, format="rgb24") - video_frame.pts = self._pts + video_frame.pts = pts video_frame.time_base = self._time_base - self._pts += 1 + self._next_pts = pts + 1 return video_frame async def close(self) -> None: @@ -93,9 +211,43 @@ async def close(self) -> None: if self._closed: return self._closed = True + while True: + try: + queued = self._frames.get_nowait() + except asyncio.QueueEmpty: + break + if queued is not None: + self._retire_pending_frame(queued.frame) + if self._retired_pending_frames: + await asyncio.gather( + *(frame.resolve() for frame in self._retired_pending_frames) + ) + self._retired_pending_frames.clear() self._frames.put_nowait(None) self.stop() + def _drop_queued_frame(self) -> None: + if not self._frames.full(): + return + try: + stale = self._frames.get_nowait() + except asyncio.QueueEmpty: + return + if stale is not None: + self._retire_pending_frame(stale.frame) + self._dropped_for_lag += 1 + + def _retire_pending_frame(self, frame: _QueuedRGBFrame) -> None: + if isinstance(frame, _PendingRGBFrame) and not frame.ready_event.query(): + self._retired_pending_frames.append(frame) + + def _reap_retired_pending_frames(self) -> None: + self._retired_pending_frames = [ + frame + for frame in self._retired_pending_frames + if not frame.ready_event.query() + ] + class WebRTCServer: """Own the HTTP, signaling, input buffering, and media transport.""" @@ -215,11 +367,16 @@ def write(self, result: StepResult) -> None: session_desc = self._session_desc if session_desc is None: raise RuntimeError("Open the WebRTC server before writing.") - frames = _result_to_rgb_frames(result, session_desc) + frames = _validated_result_frames(result, session_desc) + if self._video_track is None: + return + queued_frames = _prepare_rgb_frames(frames) loop = self._loop if loop is None: raise RuntimeError("WebRTC server is not running.") - future = asyncio.run_coroutine_threadsafe(self._enqueue_frames(frames), loop) + future = asyncio.run_coroutine_threadsafe( + self._enqueue_frames(queued_frames), loop + ) future.result() def close(self) -> None: @@ -321,7 +478,12 @@ async def _offer(self, request: web.Request) -> web.Response: ) peer_connection = RTCPeerConnection() - video_track = _VideoTrack(session_desc.frames_per_second_for_ui) + video_track = _VideoTrack( + session_desc.frames_per_second_for_ui, + drop_oldest=( + session_desc.presentation_mode is PresentationMode.ONLY_PRESENT_NEWEST + ), + ) peer_connection.addTrack(video_track) self._peer_connection = peer_connection self._video_track = video_track @@ -450,6 +612,13 @@ def _append_event( return timestamp_us = np.uint64((time.monotonic_ns() - session_start_ns) // 1_000) event = UserInputEvent(timestamp=timestamp_us, event_data=event_data) + if isinstance(event_data, KeyboardUserInputEventData): + logger.info( + "WebRTC received keyboard event key={} state={} timestamp_us={}", + event_data.key, + event_data.state.value, + int(timestamp_us), + ) callback = self._input_callback if callback is None: raise RuntimeError("WebRTC input callback is not registered.") @@ -465,9 +634,7 @@ def _record_client_disconnect(self) -> None: if not self._closed: self._append_event(CloseUserInputEventData()) - async def _enqueue_frames( - self, frames: tuple[np.ndarray[Any, np.dtype[np.uint8]], ...] - ) -> None: + async def _enqueue_frames(self, frames: tuple[_QueuedRGBFrame, ...]) -> None: """Append frames to the active media track, if connected.""" track = self._video_track if track is not None: @@ -507,10 +674,10 @@ def _normalized_coordinate(value: object, *, label: str) -> float: return result -def _result_to_rgb_frames( +def _validated_result_frames( result: StepResult, session_desc: SessionDesc -) -> tuple[np.ndarray[Any, np.dtype[np.uint8]], ...]: - """Convert one result to time-major RGB uint8 frames.""" +) -> torch.Tensor: + """Return validated time-major frames without materializing them on the host.""" output = result.output.detach() if result.output_layout == VideoTensorLayout.tchw: frames = output @@ -542,10 +709,40 @@ def _result_to_rgb_frames( if result.output_layout != session_desc.output_layout: raise ValueError("StepResult.output_layout does not match SessionDesc.") + return frames + + +def _rgb_uint8_thwc(frames: torch.Tensor) -> torch.Tensor: + """Convert validated frames to contiguous ``[T, H, W, C]`` uint8 storage.""" + if frames.shape[1] == 1: frames = frames.repeat(1, 3, 1, 1) if frames.is_floating_point(): frames = ((frames.to(torch.float32).clamp(-1.0, 1.0) + 1.0) * 127.5).round() frames = frames.clamp(0, 255).to(torch.uint8) - frames = frames.permute(0, 2, 3, 1).contiguous().cpu() - return tuple(np.asarray(frame.numpy()) for frame in frames) + return frames.permute(0, 2, 3, 1).contiguous() + + +def _prepare_rgb_frames(frames: torch.Tensor) -> tuple[_QueuedRGBFrame, ...]: + """Prepare RGB frames without synchronizing the calling thread on CUDA work.""" + frames = _rgb_uint8_thwc(frames) + if not frames.is_cuda: + return tuple(np.asarray(frame.numpy()) for frame in frames.cpu()) + + host_frames = torch.empty( + frames.shape, + dtype=torch.uint8, + device="cpu", + pin_memory=True, + ) + host_frames.copy_(frames, non_blocking=True) + ready_event = torch.cuda.Event() + ready_event.record(torch.cuda.current_stream(frames.device)) + return tuple( + _PendingRGBFrame( + host_frames=host_frames, + frame_index=frame_index, + ready_event=ready_event, + ) + for frame_index in range(frames.shape[0]) + ) diff --git a/flashdreams/flashdreams/runtime_v2/session_desc.py b/flashdreams/flashdreams/runtime_v2/session_desc.py index 525c8122a..b741b48c5 100644 --- a/flashdreams/flashdreams/runtime_v2/session_desc.py +++ b/flashdreams/flashdreams/runtime_v2/session_desc.py @@ -12,7 +12,7 @@ class BackpressureMode(Enum): - """What the model thread does when the presentation queue is full.""" + """What the model-generation-thread does when its output queue is full.""" BLOCK = "block" """Wait when the presentation queue is full.""" @@ -22,7 +22,7 @@ class BackpressureMode(Enum): class PresentationMode(Enum): - """What the UI thread does when no new model frame is ready.""" + """What the UI loop does when no new model frame is ready.""" ONLY_PRESENT_NEW = "only_present_new" """Present only after advancing to a new model frame.""" @@ -44,16 +44,16 @@ class SessionDesc: """Declared tensor layout for generated video results.""" backpressure_mode: BackpressureMode = BackpressureMode.BLOCK - """What the model thread does when the presentation queue is full.""" + """What the model-generation-thread does when its output queue is full.""" presentation_mode: PresentationMode = PresentationMode.ONLY_PRESENT_NEWEST - """What the UI thread does when no new model frame is ready.""" + """What the UI loop does when no new model frame is ready.""" frames_per_second_for_ui: int = 60 - """Rate to read input and present finished results at, in frames per second.""" + """Rate to read input and run continuous UI redraws, in frames per second.""" frames_per_second_for_step: int = 30 - """Maximum model-loop iterations per second.""" + """Initial video rate and maximum model-loop iterations per second.""" video_width: int = 1280 """Output video width in pixels.""" diff --git a/flashdreams/flashdreams/runtime_v2/session_runner.py b/flashdreams/flashdreams/runtime_v2/session_runner.py index 38391cb9b..acea6c509 100644 --- a/flashdreams/flashdreams/runtime_v2/session_runner.py +++ b/flashdreams/flashdreams/runtime_v2/session_runner.py @@ -5,6 +5,8 @@ import logging import threading +import time +from collections import deque from flashdreams.api_v2.client_window import IClientWindow from flashdreams.api_v2.loop import IModelLoop, IUILoop @@ -21,6 +23,108 @@ _MODEL_THREAD_NAME = "flashdreams-model-generation-thread" _UI_READER_ID = 0 _MODEL_READER_ID = 1 +_MODEL_FPS_WINDOW_SECONDS = 2.0 +"""Wall-time window used to estimate generated-frame throughput.""" + + +class _PresentationClock: + """Schedule model-frame advances at recent model throughput.""" + + def __init__(self, frames_per_second: int) -> None: + self._fallback_frame_interval = 1.0 / frames_per_second + self._frame_interval = self._fallback_frame_interval + self._next_frame_at: float | None = None + self._generation: int | None = None + self._observations: deque[tuple[float, int]] = deque() + self._lock = threading.Lock() + + @property + def frames_per_second(self) -> float: + """Return the current model-frame presentation rate.""" + with self._lock: + return 1.0 / self._frame_interval + + def observe_model_output( + self, + *, + now: float, + generation: int, + frame_count: int, + ) -> None: + """Add one completed model chunk to the rolling FPS estimate. + + Args: + now: Monotonic completion time for the chunk. + generation: Session generation that produced the chunk. + frame_count: Number of generated frames in the chunk. + + Raises: + ValueError: ``frame_count`` is not positive or ``now`` precedes the + latest observation. + """ + if frame_count <= 0: + raise ValueError(f"frame_count must be > 0, got {frame_count}.") + with self._lock: + if self._generation is None or generation > self._generation: + self._reset_generation(generation) + elif generation < self._generation: + return + + if self._observations and now < self._observations[-1][0]: + raise ValueError("now must not precede the latest observation.") + observed_frames = ( + frame_count + if not self._observations + else self._observations[-1][1] + frame_count + ) + if self._observations and now == self._observations[-1][0]: + self._observations[-1] = (now, observed_frames) + else: + self._observations.append((now, observed_frames)) + self._update_frame_interval(now) + + def is_due(self, now: float, generation: int) -> bool: + """Return whether the next model frame may be selected.""" + with self._lock: + if generation != self._generation: + self._reset_generation(generation) + return self._next_frame_at is None or now >= self._next_frame_at + + def mark_advanced(self, now: float) -> None: + """Record one selected frame without catching up after a long stall.""" + with self._lock: + next_frame_at = self._next_frame_at + if next_frame_at is None or now - next_frame_at >= self._frame_interval: + self._next_frame_at = now + self._frame_interval + else: + self._next_frame_at = next_frame_at + self._frame_interval + + def _reset_generation(self, generation: int) -> None: + self._generation = generation + self._frame_interval = self._fallback_frame_interval + self._next_frame_at = None + self._observations.clear() + + def _update_frame_interval(self, now: float) -> None: + cutoff = now - _MODEL_FPS_WINDOW_SECONDS + while len(self._observations) >= 3 and self._observations[1][0] <= cutoff: + self._observations.popleft() + if len(self._observations) < 2: + return + + first_at, first_frames = self._observations[0] + last_at, last_frames = self._observations[-1] + window_started_at = max(first_at, cutoff) + frames_at_window_start = float(first_frames) + if window_started_at > first_at: + second_at, second_frames = self._observations[1] + fraction = (window_started_at - first_at) / (second_at - first_at) + frames_at_window_start += fraction * (second_frames - first_frames) + + elapsed = last_at - window_started_at + generated_frames = last_frames - frames_at_window_start + if elapsed > 0.0 and generated_frames > 0.0: + self._frame_interval = elapsed / generated_frames def _contains(events: UserInputEvents, event_type: type[UserInputEventData]) -> bool: @@ -45,10 +149,10 @@ def run_session( ) -> None: """Run a session's UI and model loops. - The calling thread handles the window and UI. The model runs on a separate - Python thread. Returns when the client closes the window, when the model - loop has finished and no generated frames are still waiting, or when either - loop fails. + The calling io-thread handles the window and UI. A model-generation-thread + runs the model loop. Returns when the client closes the window, when the + model loop has finished and no generated frames are still waiting, or when + either loop fails. Both loops are shut down, every sink opened is closed, and the session is closed, before this returns or raises. @@ -74,6 +178,7 @@ def run_session( session_desc = session.session_desc tick_seconds = 1.0 / session_desc.frames_per_second_for_ui + presentation_clock = _PresentationClock(session_desc.frames_per_second_for_step) event_buffer = EventBuffer() stop = session._shutdown_event presentation_manager = session._presentation_manager @@ -114,14 +219,26 @@ def publish_model_results( generation: int, results: list[StepResult], ) -> None: + if results: + presentation_clock.observe_model_output( + now=time.monotonic(), + generation=generation, + frame_count=results[0].frame_count, + ) presentation_manager.publish(generation, results) if metrics_output_sink is not None: for result in results: metrics_output_sink.write(result) def tick_ui() -> None: - model_advanced, _ = presentation_manager.advance(event_buffer.generation) - # Safe presentation does not redraw a frame the UI already consumed. + assert ui_loop is not None + generation = event_buffer.generation + now = time.monotonic() + model_advanced = False + if presentation_clock.is_due(now, generation): + model_advanced, _ = presentation_manager.advance(generation) + if model_advanced: + presentation_clock.mark_advanced(now) if ( session_desc.presentation_mode is PresentationMode.ONLY_PRESENT_NEW and not model_advanced @@ -159,6 +276,7 @@ def tick_ui() -> None: name=_MODEL_THREAD_NAME, ) model_thread_handle.start() + next_tick_at = time.monotonic() + tick_seconds # Keep servicing input and presenting queued frames until shutdown, # or until the model finishes and no generated frames remain. @@ -169,13 +287,18 @@ def tick_ui() -> None: and not presentation_manager.has_pending_frames() ): break - if stop.wait(tick_seconds): + wait_seconds = max(0.0, next_tick_at - time.monotonic()) + if stop.wait(wait_seconds): break collect_input() if stop.is_set(): break tick_ui() event_buffer.collect_garbage() + next_tick_at += tick_seconds + completed_at = time.monotonic() + if next_tick_at <= completed_at: + next_tick_at = completed_at + tick_seconds except BaseException as error: high_level_failures = error finally: diff --git a/flashdreams/flashdreams/t2v_v2/application.py b/flashdreams/flashdreams/t2v_v2/application.py index f21ad8364..15467e276 100644 --- a/flashdreams/flashdreams/t2v_v2/application.py +++ b/flashdreams/flashdreams/t2v_v2/application.py @@ -11,7 +11,7 @@ from flashdreams.api_v2.application import IApplication from flashdreams.api_v2.session import ISession from flashdreams.infra.config import derive_config -from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.t2v_v2.defaults import T2VApplicationDefaults from flashdreams.t2v_v2.session import T2VSession @@ -144,6 +144,7 @@ def session_desc(self) -> SessionDesc: """ return SessionDesc( output_layout=self.defaults.output_layout, + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, frames_per_second_for_ui=_FRAMES_PER_SECOND_FOR_UI, frames_per_second_for_step=self.defaults.fps, video_width=self.defaults.pixel_width, diff --git a/flashdreams/flashdreams/t2v_v2/testing.py b/flashdreams/flashdreams/t2v_v2/testing.py index 4751ee907..9cfbc6f85 100644 --- a/flashdreams/flashdreams/t2v_v2/testing.py +++ b/flashdreams/flashdreams/t2v_v2/testing.py @@ -11,7 +11,7 @@ import os import shutil from collections.abc import Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any @@ -22,7 +22,11 @@ from flashdreams.api_v2.client_window import IClientWindow from flashdreams.api_v2.output_sink import OutputSink from flashdreams.runtime_v2.mp4_output_sink import Mp4OutputSink -from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.session_desc import ( + BackpressureMode, + PresentationMode, + SessionDesc, +) from flashdreams.runtime_v2.session_runner import run_session from flashdreams.runtime_v2.step_result import StepResult from flashdreams.runtime_v2.user_input_events import UserInputEvents @@ -106,7 +110,8 @@ def check_t2v_model_impl( The coverage an integration gets from one call: the application loads, resolves a session, generates, and what it generated is a video rather than a run that merely finished. It is initialized and closed here, and frames - are read the way a sink reads them. + are read the way a sink reads them. Blocking backpressure and new-frame-only + presentation keep the inspected frames identical to model output. Args: application: Uninitialized application to run. @@ -133,6 +138,11 @@ def check_t2v_model_impl( try: if session_desc is None: session_desc = application.session_desc() + session_desc = replace( + session_desc, + backpressure_mode=BackpressureMode.BLOCK, + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, + ) run_session( application.create_session(session_desc), _InspectingClientWindow(inspector), diff --git a/flashdreams/test_v2/test_session_runner.py b/flashdreams/test_v2/test_session_runner.py index 171de6ee8..1c18652bb 100644 --- a/flashdreams/test_v2/test_session_runner.py +++ b/flashdreams/test_v2/test_session_runner.py @@ -23,7 +23,7 @@ PresentationMode, SessionDesc, ) -from flashdreams.runtime_v2.session_runner import run_session +from flashdreams.runtime_v2.session_runner import _PresentationClock, run_session from flashdreams.runtime_v2.step_result import StepResult from flashdreams.runtime_v2.user_input_event import ( CloseUserInputEventData, @@ -57,6 +57,62 @@ def test_session_modes_are_independent() -> None: assert SessionDesc().presentation_mode is PresentationMode.ONLY_PRESENT_NEWEST +def test_presentation_clock_paces_frames_and_reanchors_after_a_stall() -> None: + clock = _PresentationClock(frames_per_second=4) + + assert clock.is_due(now=1.0, generation=0) + clock.mark_advanced(now=1.0) + assert not clock.is_due(now=1.24, generation=0) + assert clock.is_due(now=1.25, generation=0) + + clock.mark_advanced(now=2.0) + assert not clock.is_due(now=2.24, generation=0) + assert clock.is_due(now=2.25, generation=0) + assert clock.is_due(now=2.0, generation=1) + + +def test_presentation_clock_uses_recent_model_fps() -> None: + clock = _PresentationClock(frames_per_second=16) + + clock.observe_model_output(now=1.0, generation=0, frame_count=12) + assert clock.frames_per_second == 16 + + clock.observe_model_output(now=1.9, generation=0, frame_count=12) + assert clock.frames_per_second == pytest.approx(12 / 0.9) + + assert clock.is_due(now=2.0, generation=0) + clock.mark_advanced(now=2.0) + assert not clock.is_due(now=2.074, generation=0) + assert clock.is_due(now=2.075, generation=0) + + +def test_presentation_clock_limits_estimate_to_recent_two_seconds() -> None: + clock = _PresentationClock(frames_per_second=30) + + clock.observe_model_output(now=0.0, generation=0, frame_count=10) + clock.observe_model_output(now=1.0, generation=0, frame_count=10) + assert clock.frames_per_second == pytest.approx(10.0) + + clock.observe_model_output(now=2.0, generation=0, frame_count=20) + assert clock.frames_per_second == pytest.approx(15.0) + + clock.observe_model_output(now=3.0, generation=0, frame_count=20) + assert clock.frames_per_second == pytest.approx(20.0) + + +def test_presentation_clock_resets_estimate_for_a_new_generation() -> None: + clock = _PresentationClock(frames_per_second=16) + clock.observe_model_output(now=1.0, generation=0, frame_count=12) + clock.observe_model_output(now=2.0, generation=0, frame_count=12) + assert clock.frames_per_second == pytest.approx(12.0) + + assert clock.is_due(now=2.1, generation=1) + assert clock.frames_per_second == 16 + + clock.observe_model_output(now=2.2, generation=0, frame_count=120) + assert clock.frames_per_second == 16 + + class CallLog: """Record calls made from either thread, with the thread that made them.""" @@ -404,6 +460,40 @@ def test_run_session_calls_ui_run_on_the_io_thread() -> None: assert log.threads_for("session.step(0)") == {_STEP_THREAD_NAME} +def test_continuous_ui_processes_input_while_model_generation_waits() -> None: + """Keep the io-thread responsive during a slow model-generation step.""" + log = CallLog() + input_processed = threading.Event() + + class SlowModelSession(FakeSession): + def step(self, step_index: int, events: UserInputEvents) -> StepResult: + assert input_processed.wait(timeout=1.0) + return super().step(step_index, events) + + def run_ui(self, step_index: int, events: UserInputEvents) -> StepResult | None: + if events.get_events(): + input_processed.set() + return super().run_ui(step_index, events) + + session = SlowModelSession( + _session_desc( + presentation_mode=PresentationMode.ONLY_PRESENT_NEWEST, + ui_fps=100, + model_fps=1, + ), + log, + ) + window = RecordingClientWindow( + log, + [UserInputEvents([]), _key_event()], + ) + + run_session(session, window, steps=1) + + assert input_processed.is_set() + assert "ui_loop.step" in log.calls + + def test_each_message_queue_runs_on_its_owning_thread() -> None: log = CallLog() @@ -469,6 +559,7 @@ def test_default_ui_composites_channels_and_holds_the_latest_frame() -> None: assert held is not None assert torch.equal(held.output, first.output) assert not manager.advance(1)[0] + assert manager.presented_frame_count == 0 assert ui.step(2, UserInputEvents([])) is None @@ -481,8 +572,8 @@ def step(self, step_index: int, events: UserInputEvents) -> StepResult: self._log.record(f"session.step({step_index})") return StepResult( step_index=step_index, - output=torch.arange(6, dtype=torch.float32).reshape(1, 3, 2, 1, 1), - frame_count=2, + output=torch.arange(36, dtype=torch.float32).reshape(1, 3, 12, 1, 1), + frame_count=12, output_layout=self.session_desc.output_layout, metrics={"total_ms": 1.5}, ) @@ -509,17 +600,39 @@ def close(self) -> None: steps=1, ) - assert [result.frame_count for result in window.results] == [1, 1] - assert [result.output[0, 0, 0, 0, 0].item() for result in window.results] == [0, 1] - assert [result.metrics for result in window.results] == [ - {"ui_ms": 0.25}, - {"ui_ms": 0.25}, - ] + assert [result.frame_count for result in window.results] == [1] * 12 + assert [result.output[0, 0, 0, 0, 0].item() for result in window.results] == list( + range(12) + ) + assert [result.metrics for result in window.results] == [{"ui_ms": 0.25}] * 12 assert len(metrics.results) == 1 assert metrics.results[0].metrics == {"total_ms": 1.5} -def test_drop_oldest_preempts_the_rest_of_a_stale_chunk() -> None: +def test_default_ui_does_not_redraw_an_unchanged_model_frame() -> None: + log = CallLog() + + class DefaultUISession(FakeSession): + def init(self) -> None: + self._log.record("session.init") + self.register_model_loop(FakeModelLoop, state=self) + + session = DefaultUISession( + _session_desc( + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, + ui_fps=100, + model_fps=30, + ), + log, + ) + window = RecordingClientWindow(log) + + run_session(session, window, steps=3) + + assert [result.step_index for result in window.results] == [0, 1, 2] + + +def test_drop_oldest_finishes_active_chunk_before_newest_waiting_chunk() -> None: manager = PresentationManager() manager.configure( max_pending=1, @@ -531,18 +644,38 @@ def test_drop_oldest_preempts_the_rest_of_a_stale_chunk() -> None: def result(step_index: int, frames: int) -> StepResult: return StepResult( step_index=step_index, - output=torch.full((frames, 3, 1, 1), float(step_index)), + output=( + torch.arange(frames, dtype=torch.float32).reshape(frames, 1, 1, 1) + + step_index * 10 + ).expand(-1, 3, -1, -1), frame_count=frames, output_layout=VideoTensorLayout.tchw, ) - manager.publish(0, [result(0, 2)]) + manager.publish(0, [result(0, 3)]) assert manager.advance(0)[0] + assert manager.presented_frame_count == 1 manager.publish(0, [result(1, 1)]) + manager.publish(0, [result(2, 1)]) + + assert manager.advance(0)[0] + second = manager.presented_frame(0) + assert second is not None + assert second[0, 0, 0] == 1 + assert manager.presented_frame_count == 2 + + assert manager.advance(0)[0] + third = manager.presented_frame(0) + assert third is not None + assert third[0, 0, 0] == 2 + assert manager.presented_frame_count == 3 + assert manager.advance(0)[0] newest = manager.presented_frame(0) assert newest is not None - assert newest[0, 0, 0] == 1 + assert newest[0, 0, 0] == 20 + assert manager.presented_frame_count == 4 + assert manager.dropped_for_space == 1 def test_run_session_opens_window_with_the_resolved_session_desc() -> None: @@ -587,7 +720,7 @@ def test_run_session_resets_the_session_and_the_step_index() -> None: run_session(session, window, steps=2) - # Ignore UI calls when checking the model thread's order. + # Ignore UI calls when checking the model-generation-loop order. calls = [call for call in log.calls if call.startswith("session.reset")] + [ call for call in log.calls if call.startswith("session.step(") ] diff --git a/flashdreams/test_v2/test_t2v_application.py b/flashdreams/test_v2/test_t2v_application.py index 0efebf7da..a9d8d0694 100644 --- a/flashdreams/test_v2/test_t2v_application.py +++ b/flashdreams/test_v2/test_t2v_application.py @@ -13,7 +13,7 @@ import pytest -from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.runtime_v2.video_tensor import VideoTensorLayout from flashdreams.t2v_v2.application import T2VApplication from flashdreams.t2v_v2.defaults import T2VApplicationDefaults @@ -149,6 +149,7 @@ def _session_desc( ) -> SessionDesc: return SessionDesc( output_layout=layout, + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, frames_per_second_for_ui=60, frames_per_second_for_step=_FPS, video_width=width, diff --git a/flashdreams/test_v2/test_webrtc_client_window.py b/flashdreams/test_v2/test_webrtc_client_window.py index 18fba8860..a0f23b83f 100644 --- a/flashdreams/test_v2/test_webrtc_client_window.py +++ b/flashdreams/test_v2/test_webrtc_client_window.py @@ -5,6 +5,8 @@ import asyncio import json +from typing import Any, cast +from unittest.mock import ANY, Mock, call import pytest import torch @@ -23,7 +25,13 @@ ) from av import VideoFrame -from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.serving import webrtc_server +from flashdreams.runtime_v2.serving.webrtc_server import ( + _FramePacer, + _PendingRGBFrame, + _VideoTrack, +) +from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.runtime_v2.step_result import StepResult from flashdreams.runtime_v2.user_input_event import ( FocusUserInputEventData, @@ -38,6 +46,7 @@ def _session_desc() -> SessionDesc: return SessionDesc( output_layout=VideoTensorLayout.tchw, + presentation_mode=PresentationMode.ONLY_PRESENT_NEW, frames_per_second_for_ui=30, frames_per_second_for_step=30, video_width=16, @@ -84,7 +93,9 @@ def on_track(track: MediaStreamTrack) -> None: @pytest.mark.asyncio -async def test_window_buffers_browser_events_until_drained() -> None: +async def test_window_buffers_browser_events_until_drained(monkeypatch: Any) -> None: + logger = Mock() + monkeypatch.setattr(webrtc_server, "logger", logger) window = WebRTCClientWindow() peer: RTCPeerConnection | None = None try: @@ -141,6 +152,24 @@ async def test_window_buffers_browser_events_until_drained() -> None: ("w", KeyboardInputState.PRESSED), ("w", KeyboardInputState.RELEASED), ] + assert ( + call( + "WebRTC received keyboard event key={} state={} timestamp_us={}", + "w", + "Pressed", + ANY, + ) + in logger.info.call_args_list + ) + assert ( + call( + "WebRTC received keyboard event key={} state={} timestamp_us={}", + "w", + "Released", + ANY, + ) + in logger.info.call_args_list + ) assert events[0].get_timestamp() <= events[1].get_timestamp() mouse = next( data @@ -189,3 +218,109 @@ async def test_write_delivers_a_video_frame_to_the_browser() -> None: if peer is not None: await peer.close() window.close() + + +@pytest.mark.asyncio +async def test_video_track_resolves_pending_cuda_transfer_without_blocking() -> None: + class FakeCUDAEvent: + def __init__(self) -> None: + self.queries = 0 + + def query(self) -> bool: + self.queries += 1 + return self.queries >= 3 + + ready_event = FakeCUDAEvent() + pending = _PendingRGBFrame( + host_frames=torch.full((1, 16, 16, 3), 23, dtype=torch.uint8), + frame_index=0, + ready_event=cast(Any, ready_event), + ) + track = _VideoTrack(frames_per_second=30) + try: + await track.enqueue((pending,)) + frame = await asyncio.wait_for(track.recv(), timeout=1) + + assert ready_event.queries == 3 + pixels = frame.to_ndarray(format="rgb24") + assert pixels.shape == (16, 16, 3) + assert abs(float(pixels.mean()) - 23.0) <= 2.0 + finally: + await track.close() + + +@pytest.mark.asyncio +async def test_video_track_does_not_burst_to_catch_up_after_a_stall() -> None: + track = _VideoTrack(frames_per_second=30) + frame = torch.zeros((16, 16, 3), dtype=torch.uint8).numpy() + try: + await track.enqueue((frame, frame, frame)) + await track.recv() + await asyncio.sleep(0.1) + + await track.recv() + resumed_at = asyncio.get_running_loop().time() + await track.recv() + next_frame_at = asyncio.get_running_loop().time() + + assert next_frame_at - resumed_at >= 0.02 + finally: + await track.close() + + +def test_video_track_pacer_does_not_accumulate_wakeup_delay() -> None: + """Keep repeated scheduler overshoot out of subsequent frame deadlines.""" + pacer = _FramePacer(frames_per_second=60) + now = 0.0 + sent_at: list[float] = [] + overshoot = 0.001 + + for frame_index in range(240): + delay = pacer.delay_seconds( + now=now, + source_at=frame_index / 120.0, + ) + now += delay + if delay > 0.0: + now += overshoot + sent_at.append(now) + + ideal_last_frame_at = (len(sent_at) - 1) / 60.0 + assert sent_at[-1] - ideal_last_frame_at == pytest.approx(overshoot) + + +@pytest.mark.asyncio +async def test_video_track_timestamps_sparse_frames_at_their_source_cadence() -> None: + track = _VideoTrack(frames_per_second=30) + frame = torch.zeros((16, 16, 3), dtype=torch.uint8).numpy() + try: + await track.enqueue((frame,)) + first = await track.recv() + await asyncio.sleep(0.1) + await track.enqueue((frame,)) + second = await track.recv() + + assert first.pts == 0 + assert second.pts is not None + assert second.pts >= 2 + finally: + await track.close() + + +@pytest.mark.asyncio +async def test_drop_oldest_video_track_keeps_the_latest_ui_frame() -> None: + track = _VideoTrack(frames_per_second=60, drop_oldest=True) + frames = tuple( + torch.full((16, 16, 3), value, dtype=torch.uint8).numpy() + for value in (10, 20, 30) + ) + try: + for frame in frames: + await track.enqueue((frame,)) + + assert track.qsize() == 1 + assert track.dropped_for_lag == 2 + latest = await track.recv() + assert abs(float(latest.to_ndarray(format="rgb24").mean()) - 30.0) <= 2.0 + finally: + await track.close() diff --git a/integrations/lingbot/README.md b/integrations/lingbot/README.md index e993a539d..0122ac762 100644 --- a/integrations/lingbot/README.md +++ b/integrations/lingbot/README.md @@ -90,6 +90,13 @@ uv run flashdreams-run lingbot-world-fast \ --prompt "your text prompt here" --total-blocks 21 ``` +## FlashDreams v2 Cam2V application + +The v2 `cam2v-lingbot` application lives in the standalone +[`integrations_v2/cam2v_lingbot`](../../integrations_v2/cam2v_lingbot/README.md) +package. This distribution continues to own the Lingbot model, runners, and +legacy serving paths consumed by that application. + Multi-GPU via context-parallelism (Wan 2.1 CP assumes `cp_size == world_size`): ```bash diff --git a/integrations/lingbot/lingbot/controls.py b/integrations/lingbot/lingbot/controls.py index da7cbf16c..85e945c8f 100644 --- a/integrations/lingbot/lingbot/controls.py +++ b/integrations/lingbot/lingbot/controls.py @@ -1,283 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Lingbot-owned keyboard segmenting and camera pose integration.""" - -from __future__ import annotations - -from collections import deque -from dataclasses import dataclass, field -from typing import Literal - -import numpy as np - -from flashdreams.runtime.keyboard import DEFAULT_SUPPORTED_KEYS, KeyboardState - -PoseSegment = tuple[float, float, frozenset[str]] - - -class KeyboardResampler: - """Resample sparse keydown/keyup edges into a Lingbot camera timeline.""" - - def __init__( - self, - *, - fps: float, - start_v: float = 0.0, - supported_keys: frozenset[str] = DEFAULT_SUPPORTED_KEYS, - ) -> None: - if fps <= 0: - raise ValueError("fps must be > 0") - self._fps = float(fps) - self._dt = 1.0 / self._fps - self._supported_keys = supported_keys - self.next_chunk_start_v = start_v - self._event_log: deque[tuple[float, dict[str, str]]] = deque() - self._carried_state = KeyboardState(supported_keys=supported_keys) - - @property - def fps(self) -> float: - return self._fps - - @property - def dt(self) -> float: - return self._dt - - def on_edge(self, *, arrival_t: float, event: str, key: str) -> None: - entry = (arrival_t, {"event": event, "key": key}) - if not self._event_log or arrival_t >= self._event_log[-1][0]: - self._event_log.append(entry) - return - for index, (event_t, _) in enumerate(self._event_log): - if arrival_t < event_t: - self._event_log.insert(index, entry) - return - self._event_log.append(entry) - - def sample_chunk(self, num_frames: int) -> tuple[list[PoseSegment], list[float]]: - if num_frames < 1: - raise ValueError("num_frames must be >= 1") - - chunk_start_v = self.next_chunk_start_v - chunk_end_v = chunk_start_v + num_frames * self._dt - - while self._event_log and self._event_log[0][0] < chunk_start_v: - _, payload = self._event_log.popleft() - self._carried_state.apply_event(**payload) - - segments: list[PoseSegment] = [] - prev_t = chunk_start_v - prev_state = self._carried_state.resolved_effective_keys() - while self._event_log and self._event_log[0][0] <= chunk_end_v: - event_t, payload = self._event_log.popleft() - if event_t > prev_t: - segments.append((prev_t, event_t, prev_state)) - self._carried_state.apply_event(**payload) - prev_state = self._carried_state.resolved_effective_keys() - prev_t = event_t - if prev_t < chunk_end_v: - segments.append((prev_t, chunk_end_v, prev_state)) - elif not segments: - segments.append((chunk_start_v, chunk_end_v, prev_state)) - - frame_times = [chunk_start_v + (i + 1) * self._dt for i in range(num_frames)] - self.next_chunk_start_v = chunk_end_v - return segments, frame_times - - def reset(self, *, start_v: float) -> None: - self._event_log.clear() - self._carried_state = KeyboardState(supported_keys=self._supported_keys) - self.next_chunk_start_v = start_v - - def event_log_size(self) -> int: - return len(self._event_log) - - -def _rotation_matrix(axis: str, angle_rad: float) -> np.ndarray: - cos_t = np.float32(np.cos(angle_rad)) - sin_t = np.float32(np.sin(angle_rad)) - if axis == "x": - return np.array( - [ - [1.0, 0.0, 0.0], - [0.0, cos_t, -sin_t], - [0.0, sin_t, cos_t], - ], - dtype=np.float32, - ) - if axis == "y": - return np.array( - [ - [cos_t, 0.0, sin_t], - [0.0, 1.0, 0.0], - [-sin_t, 0.0, cos_t], - ], - dtype=np.float32, - ) - if axis == "z": - return np.array( - [ - [cos_t, -sin_t, 0.0], - [sin_t, cos_t, 0.0], - [0.0, 0.0, 1.0], - ], - dtype=np.float32, - ) - return np.eye(3, dtype=np.float32) - - -@dataclass(slots=True) -class CameraPoseIntegrator: - """Integrate a piecewise-constant keyboard timeline into a camera path.""" - - move_speed_per_s: float = 0.8 - rotate_speed_rad_per_s: float = float(np.deg2rad(32.0)) - pitch_limit_rad: float = float(np.deg2rad(85.0)) - coordinate_system: Literal["RDF", "FLU"] = "RDF" - _current_pose: np.ndarray = field( - default_factory=lambda: np.eye(4, dtype=np.float32), - ) - _current_pitch: float = 0.0 - - def __post_init__(self) -> None: - if self.coordinate_system not in {"RDF", "FLU"}: - raise ValueError( - "coordinate_system must be 'RDF' (right-down-forward) " - "or 'FLU' (forward-left-up)" - ) - - def reset(self, pose: np.ndarray | None = None) -> None: - if pose is None: - self._current_pose = np.eye(4, dtype=np.float32) - self._current_pitch = 0.0 - return - if pose.shape != (4, 4): - raise ValueError(f"Expected pose shape (4, 4), got {pose.shape}") - self._current_pose = pose.astype(np.float32, copy=True) - if self.coordinate_system == "FLU": - self._current_pitch = float(np.arcsin(np.clip(pose[2, 0], -1.0, 1.0))) - else: - self._current_pitch = float(np.arctan2(pose[2, 1], pose[1, 1])) - - def current_pose(self) -> np.ndarray: - return self._current_pose.copy() - - def _advance(self, *, state: frozenset[str], duration: float) -> None: - if duration <= 0: - return - - yaw_rate = 0.0 - if self.coordinate_system == "FLU": - if "a" in state or "j" in state: - yaw_rate += self.rotate_speed_rad_per_s - if "d" in state or "l" in state: - yaw_rate -= self.rotate_speed_rad_per_s - else: - if "a" in state or "j" in state: - yaw_rate -= self.rotate_speed_rad_per_s - if "d" in state or "l" in state: - yaw_rate += self.rotate_speed_rad_per_s - pitch_rate = 0.0 - if "i" in state: - pitch_rate += self.rotate_speed_rad_per_s - if "k" in state: - pitch_rate -= self.rotate_speed_rad_per_s - - yaw_delta = yaw_rate * duration - pitch_delta = pitch_rate * duration - - new_pitch = self._current_pitch + pitch_delta - if -self.pitch_limit_rad <= new_pitch <= self.pitch_limit_rad: - self._current_pitch = new_pitch - else: - pitch_delta = 0.0 - - rot = self._current_pose[:3, :3] - trans = self._current_pose[:3, 3] - if self.coordinate_system == "FLU": - rot_pitch = _rotation_matrix("y", -pitch_delta) - rot_yaw = _rotation_matrix("z", yaw_delta) - else: - rot_pitch = _rotation_matrix("x", pitch_delta) - rot_yaw = _rotation_matrix("y", yaw_delta) - rot_new = rot_yaw @ rot @ rot_pitch - - forward_rate = 0.0 - if "w" in state: - forward_rate += self.move_speed_per_s - if "s" in state: - forward_rate -= self.move_speed_per_s - right_rate = 0.0 - if "e" in state: - right_rate += self.move_speed_per_s - if "q" in state: - right_rate -= self.move_speed_per_s - - if self.coordinate_system == "FLU": - vec_forward = rot_new[:, 0] - vec_right = -rot_new[:, 1] - forward_flat = np.array( - [vec_forward[0], vec_forward[1], 0.0], dtype=np.float32 - ) - right_flat = np.array([vec_right[0], vec_right[1], 0.0], dtype=np.float32) - else: - vec_right = rot_new[:, 0] - vec_forward = rot_new[:, 2] - forward_flat = np.array( - [vec_forward[0], 0.0, vec_forward[2]], dtype=np.float32 - ) - right_flat = np.array([vec_right[0], 0.0, vec_right[2]], dtype=np.float32) - forward_norm = np.linalg.norm(forward_flat) - right_norm = np.linalg.norm(right_flat) - if forward_norm > 0: - forward_flat /= forward_norm - if right_norm > 0: - right_flat /= right_norm - - move_vec = forward_flat * (forward_rate * duration) + right_flat * ( - right_rate * duration - ) - self._current_pose = np.eye(4, dtype=np.float32) - self._current_pose[:3, :3] = rot_new - self._current_pose[:3, 3] = trans + move_vec - - def integrate_chunk( - self, - *, - segments: list[PoseSegment], - frame_times: list[float], - ) -> np.ndarray: - if not segments: - raise ValueError("segments must be non-empty") - if not frame_times: - raise ValueError("frame_times must be non-empty") - chunk_start = segments[0][0] - chunk_end = segments[-1][1] - if any( - frame_times[i] >= frame_times[i + 1] for i in range(len(frame_times) - 1) - ): - raise ValueError("frame_times must be strictly increasing") - if frame_times[0] < chunk_start - 1e-9 or frame_times[-1] > chunk_end + 1e-9: - raise ValueError( - "frame_times must lie within the chunk window " - f"[{chunk_start}, {chunk_end}]" - ) - - poses: list[np.ndarray] = [] - cur_t = chunk_start - ft_idx = 0 - for _, seg_end, seg_state in segments: - while ft_idx < len(frame_times) and frame_times[ft_idx] <= seg_end: - target_t = frame_times[ft_idx] - self._advance(state=seg_state, duration=target_t - cur_t) - cur_t = target_t - poses.append(self._current_pose.copy()) - ft_idx += 1 - if seg_end > cur_t: - self._advance(state=seg_state, duration=seg_end - cur_t) - cur_t = seg_end - - return np.stack(poses, axis=0).astype(np.float32) +"""Compatibility exports for camera controls now owned by ``cam2v``.""" +from cam2v.controls import CameraPoseIntegrator, KeyboardResampler, PoseSegment __all__ = ["CameraPoseIntegrator", "KeyboardResampler", "PoseSegment"] diff --git a/integrations/lingbot/lingbot/encoder/camctrl.py b/integrations/lingbot/lingbot/encoder/camctrl.py index 6a4bbc3bd..1a604f3af 100644 --- a/integrations/lingbot/lingbot/encoder/camctrl.py +++ b/integrations/lingbot/lingbot/encoder/camctrl.py @@ -44,16 +44,16 @@ @dataclass(kw_only=True) class CamCtrlInput: - """Per-AR-step camera payload.""" + """Per-AR-step camera payload consumed by the Lingbot encoder.""" intrinsics: Tensor - """Per-frame camera intrinsics of shape ``[..., T, 4]`` (fx, fy, cx, cy).""" + """Per-frame camera intrinsics shaped ``[..., T, 4]``.""" poses: Tensor - """Per-frame camera-to-world poses of shape ``[..., T, 4, 4]``.""" + """Per-frame camera-to-world poses shaped ``[..., T, 4, 4]``.""" world_scale: float - """Scalar applied to translations when normalizing world coordinates.""" + """Scale applied to translations when normalizing world coordinates.""" @dataclass(kw_only=True) diff --git a/integrations/lingbot/lingbot/input_mapping.py b/integrations/lingbot/lingbot/input_mapping.py index ecd508d97..c23f4b073 100644 --- a/integrations/lingbot/lingbot/input_mapping.py +++ b/integrations/lingbot/lingbot/input_mapping.py @@ -44,10 +44,7 @@ ) from flashdreams.runtime.mapping import InputMappingSchema from flashdreams.runtime.types import StepRequest -from lingbot.controls import ( - CameraPoseIntegrator, - PoseSegment, -) +from lingbot.controls import CameraPoseIntegrator, PoseSegment FIELD_CAMERA_TRAJECTORY = "camera_trajectory" FIELD_CAMERA_INTRINSICS = "camera_intrinsics" diff --git a/integrations/lingbot/pyproject.toml b/integrations/lingbot/pyproject.toml index f0175fb35..fbbc42b8c 100644 --- a/integrations/lingbot/pyproject.toml +++ b/integrations/lingbot/pyproject.toml @@ -25,6 +25,7 @@ readme = "README.md" requires-python = ">=3.10" dependencies = [ "flashdreams", + "flashdreams-cam2v", "aiohttp>=3.9", "aiortc>=1.9", "mediapy>=1.1", @@ -34,6 +35,7 @@ dependencies = [ [tool.uv.sources] flashdreams = { workspace = true } +flashdreams-cam2v = { workspace = true } [project.optional-dependencies] dev = [ diff --git a/integrations_v2/README.md b/integrations_v2/README.md index fd957a1c0..c42a24674 100644 --- a/integrations_v2/README.md +++ b/integrations_v2/README.md @@ -18,6 +18,8 @@ follows is already done for you. - `red_screen` — the smallest interactive one, streaming to a browser. - `slangpy_ui_demo` — three applications that draw widgets over model output, and the reference for writing a UI loop. +- `cam2v_lingbot` — the Lingbot World specialization of the shared interactive + camera-to-video application. - `t2v_self_forcing`, `t2v_causal_forcing`, `t2v_fastvideo_causal_wan22`, `t2v_wan21`, `t2v_cosmos_predict2` — real models, each a thin wrapper over `flashdreams.t2v_v2`. diff --git a/integrations_v2/cam2v_lingbot/README.md b/integrations_v2/cam2v_lingbot/README.md new file mode 100644 index 000000000..d2fcebff2 --- /dev/null +++ b/integrations_v2/cam2v_lingbot/README.md @@ -0,0 +1,45 @@ + + +# Lingbot camera-to-video + +The FlashDreams v2 camera-to-video application for Lingbot World. This package +contains only the application boundary: it combines the shared +`flashdreams-cam2v` lifecycle and controls with the existing +`flashdreams-lingbot` model config. Its CLI input, example-data, intrinsics, +and world-scale resolution live in this package. + +The application loads the pipeline once. Each session owns its autoregressive +cache, keyboard state, camera pose, and UI state. + +## Run + +```bash +uv sync --package flashdreams-cam2v-lingbot --inexact +uv run --no-sync flashdreams-run-v2 cam2v-lingbot \ + --mode webrtc --host 0.0.0.0 --port 8089 -- --example-data +``` + +The command prints the browser URL. Use `W`/`S` to move, `A`/`D` to yaw, +`Q`/`E` to strafe, and `I`/`K` to pitch the generated camera. + +The application logs warmup-excluded `steady_state_fps` and a per-block timing +breakdown. `model_step_wall_s` includes input preparation, generation, +finalization, and CUDA completion. The pipeline profiling log provides its GPU +stage breakdown. + +For custom inputs, pass `--image-path` and `--intrinsic-path`. Also pass either +`--world-scale` directly or `--pose-path` so the application can infer the +translation normalizer. Input resolution and example-data downloads are owned +by this package and do not use the legacy Lingbot runtime/schema path. + +Use `--log-every-blocks N` to reduce log frequency and `--warmup-blocks N` to +change the five-block default warmup exclusion. + +## Tests + +```bash +uv run pytest integrations_v2/cam2v_lingbot -m ci_cpu -v +``` diff --git a/integrations_v2/cam2v_lingbot/cam2v_lingbot/__init__.py b/integrations_v2/cam2v_lingbot/cam2v_lingbot/__init__.py new file mode 100644 index 000000000..5f3405764 --- /dev/null +++ b/integrations_v2/cam2v_lingbot/cam2v_lingbot/__init__.py @@ -0,0 +1,8 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Lingbot camera-to-video application for the FlashDreams v2 API.""" + +from .app import LingbotCam2VApplication, create_app + +__all__ = ["LingbotCam2VApplication", "create_app"] diff --git a/integrations_v2/cam2v_lingbot/cam2v_lingbot/app.py b/integrations_v2/cam2v_lingbot/cam2v_lingbot/app.py new file mode 100644 index 000000000..a8a4877c1 --- /dev/null +++ b/integrations_v2/cam2v_lingbot/cam2v_lingbot/app.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Lingbot specialization of the shared camera-to-video application.""" + +from __future__ import annotations + +from cam2v import Cam2VApplication, Cam2VApplicationDefaults + +from flashdreams.api_v2.application import IApplication +from lingbot.config import RUNNER_LINGBOT_WORLD_FAST_TAEHV_WINDOW15_SINK3 + +from .conditioning import resolve_lingbot_conditioning + +_INSTALL_HINT = ( + "Install the Lingbot Cam2V application: pip install flashdreams-cam2v-lingbot." +) + + +class LingbotCam2VApplication(Cam2VApplication): + """Lingbot World configured through its existing interactive runner config.""" + + def __init__(self) -> None: + super().__init__( + defaults=Cam2VApplicationDefaults.from_runner_config( + RUNNER_LINGBOT_WORLD_FAST_TAEHV_WINDOW15_SINK3, + input_resolver=resolve_lingbot_conditioning, + install_hint=_INSTALL_HINT, + ) + ) + + +def create_app() -> IApplication: + """Return a Lingbot camera-to-video application.""" + return LingbotCam2VApplication() + + +__all__ = ["LingbotCam2VApplication", "create_app"] diff --git a/integrations_v2/cam2v_lingbot/cam2v_lingbot/conditioning.py b/integrations_v2/cam2v_lingbot/cam2v_lingbot/conditioning.py new file mode 100644 index 000000000..809f2390f --- /dev/null +++ b/integrations_v2/cam2v_lingbot/cam2v_lingbot/conditioning.py @@ -0,0 +1,206 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Resolve Lingbot assets directly into the shared Cam2V contract.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import numpy as np +import torch + +from cam2v import Cam2VConditioning + +from flashdreams.core.io.disk import default_flashdreams_cache_dir +from flashdreams.core.io.download import download_to_cache + +_EXAMPLE_DATA_BASE_URL = ( + "https://raw.githubusercontent.com/Robbyant/lingbot-world-v2/main/examples" +) +_EXAMPLE_DATA_DIR = default_flashdreams_cache_dir() / "example_data/lingbot_world" +_EXAMPLE_DATA_INDICES = frozenset(range(6)) +_EXAMPLE_PROMPT_INDICES = frozenset({0, 1, 2, 5}) +_INTRINSICS_REFERENCE_HEIGHT = 480 +_INTRINSICS_REFERENCE_WIDTH = 832 +_DEFAULT_PIXEL_HEIGHT = 464 +_DEFAULT_PIXEL_WIDTH = 832 +_TEMPORAL_COMPRESSION_RATIO = 4 +_TRANSFORMER_CHUNK_FRAMES = 3 + + +def resolve_lingbot_conditioning(values: Mapping[str, Any]) -> Cam2VConditioning: + """Resolve application arguments without the legacy input-mapping runtime.""" + example_idx = int(values.get("example_idx", 0)) + if example_idx not in _EXAMPLE_DATA_INDICES: + raise ValueError( + f"Lingbot example_idx must be one of {sorted(_EXAMPLE_DATA_INDICES)}." + ) + + image_path = _optional_path(values.get("image_path")) + pose_path = _optional_path(values.get("pose_path")) + intrinsic_path = _optional_path(values.get("intrinsic_path")) + prompt_path = _optional_path(values.get("prompt_path")) + prompt = _nonempty_text(values.get("prompt")) + world_scale = _optional_float(values.get("world_scale")) + + if _as_bool(values.get("example_data", False)): + example_dir = _ensure_example_data(example_idx) + image_path = image_path or example_dir / "image.jpg" + pose_path = pose_path or example_dir / "poses.npy" + intrinsic_path = intrinsic_path or example_dir / "intrinsics.npy" + if not prompt and prompt_path is None and example_idx in _EXAMPLE_PROMPT_INDICES: + prompt_path = example_dir / "prompt.txt" + + first_frame_path = _require_existing_path(image_path, label="image_path") + intrinsics_path = _require_existing_path( + intrinsic_path, + label="intrinsic_path", + ) + if not prompt and prompt_path is not None: + prompt = _read_first_line( + _require_existing_path(prompt_path, label="prompt_path") + ) + if world_scale is None: + poses_path = _require_existing_path(pose_path, label="pose_path") + world_scale = _infer_world_scale(poses_path) + + return Cam2VConditioning( + prompt=prompt, + first_frame_path=first_frame_path, + base_intrinsics=_load_base_intrinsics( + intrinsics_path, + pixel_height=int(values.get("pixel_height", _DEFAULT_PIXEL_HEIGHT)), + pixel_width=int(values.get("pixel_width", _DEFAULT_PIXEL_WIDTH)), + ), + world_scale=world_scale, + ) + + +def _ensure_example_data(example_idx: int) -> Path: + dirname = f"{example_idx:02d}" + cache_dir = _EXAMPLE_DATA_DIR / dirname + filenames = ["image.jpg", "poses.npy", "intrinsics.npy"] + if example_idx in _EXAMPLE_PROMPT_INDICES: + filenames.append("prompt.txt") + + distributed = torch.distributed.is_initialized() + if not distributed or torch.distributed.get_rank() == 0: + for filename in filenames: + download_to_cache( + f"{_EXAMPLE_DATA_BASE_URL}/{dirname}/{filename}", + cache_dir=cache_dir, + filename=filename, + ) + if distributed: + torch.distributed.barrier() + return cache_dir + + +def _load_base_intrinsics( + path: Path, + *, + pixel_height: int, + pixel_width: int, +) -> torch.Tensor: + intrinsics = np.asarray(np.load(path), dtype=np.float32) + if intrinsics.ndim == 1: + intrinsics = intrinsics[None, :] + if intrinsics.ndim != 2 or intrinsics.shape[0] == 0 or intrinsics.shape[1] != 4: + raise ValueError( + "Lingbot intrinsics must have shape [T, 4], got " + f"{tuple(intrinsics.shape)}." + ) + scale = np.array( + [ + pixel_width / _INTRINSICS_REFERENCE_WIDTH, + pixel_height / _INTRINSICS_REFERENCE_HEIGHT, + pixel_width / _INTRINSICS_REFERENCE_WIDTH, + pixel_height / _INTRINSICS_REFERENCE_HEIGHT, + ], + dtype=np.float32, + ) + return torch.from_numpy(np.ascontiguousarray(intrinsics[0] * scale)) + + +def _infer_world_scale(path: Path) -> float: + """Return the legacy pose normalizer without constructing a camera trace.""" + poses = np.asarray(np.load(path), dtype=np.float64) + if poses.ndim != 3 or poses.shape[1:] != (4, 4): + raise ValueError( + f"Lingbot poses must have shape [T, 4, 4], got {tuple(poses.shape)}." + ) + + raw_frame_count = poses.shape[0] + compatible_frame_count = ( + (raw_frame_count - 1) // _TEMPORAL_COMPRESSION_RATIO + ) * _TEMPORAL_COMPRESSION_RATIO + 1 + encoded_frame_count = ( + (compatible_frame_count - 1) // _TEMPORAL_COMPRESSION_RATIO + ) + 1 + if encoded_frame_count < _TRANSFORMER_CHUNK_FRAMES: + minimum = ( + (_TRANSFORMER_CHUNK_FRAMES - 1) * _TEMPORAL_COMPRESSION_RATIO + 1 + ) + raise ValueError( + f"Expected at least {minimum} poses to infer world scale, " + f"got {raw_frame_count}." + ) + encoded_frame_count -= encoded_frame_count % _TRANSFORMER_CHUNK_FRAMES + + source_indices = np.arange(compatible_frame_count, dtype=np.float64) + target_indices = np.linspace( + 0, + compatible_frame_count - 1, + encoded_frame_count, + ) + translations = poses[:compatible_frame_count, :3, 3] + encoded_translations = np.stack( + [ + np.interp(target_indices, source_indices, translations[:, axis]) + for axis in range(3) + ], + axis=1, + ) + step_distances = np.linalg.norm(np.diff(encoded_translations, axis=0), axis=1) + return float(step_distances.max(initial=0.0)) + + +def _optional_path(value: str | Path | None) -> Path | None: + return None if value is None or value == "" else Path(value) + + +def _require_existing_path(path: Path | None, *, label: str) -> Path: + if path is None: + raise ValueError(f"Lingbot Cam2V requires {label}.") + if not path.exists(): + raise FileNotFoundError(f"Lingbot Cam2V missing {label}: {path}") + return path + + +def _read_first_line(path: Path) -> str: + lines = path.read_text(encoding="utf-8").splitlines() + return _nonempty_text(lines[0]) if lines else "" + + +def _nonempty_text(value: object) -> str: + return "" if value is None else " ".join(str(value).split()) + + +def _optional_float(value: str | int | float | None) -> float | None: + return None if value is None or value == "" else float(value) + + +def _as_bool(value: object) -> bool: + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + return bool(value) + + +__all__ = ["resolve_lingbot_conditioning"] diff --git a/integrations_v2/cam2v_lingbot/cam2v_lingbot/tests/test_app.py b/integrations_v2/cam2v_lingbot/cam2v_lingbot/tests/test_app.py new file mode 100644 index 000000000..90255fb88 --- /dev/null +++ b/integrations_v2/cam2v_lingbot/cam2v_lingbot/tests/test_app.py @@ -0,0 +1,144 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CPU tests for Lingbot's thin shared Cam2V specialization.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest +import tomli as tomllib +import torch +from cam2v import Cam2VApplication, Cam2VConditioning +from cam2v_lingbot import LingbotCam2VApplication, create_app +from cam2v_lingbot import conditioning +from lingbot.config import RUNNER_LINGBOT_WORLD_FAST_TAEHV_WINDOW15_SINK3 + +pytestmark = pytest.mark.ci_cpu + +_PACKAGE_ROOT = Path(__file__).resolve().parents[2] + + +def test_package_registers_the_shared_cam2v_application() -> None: + """Keep v2 entry-point ownership in the v2 integration package.""" + manifest = tomllib.loads((_PACKAGE_ROOT / "pyproject.toml").read_text()) + + dependencies = manifest["project"]["dependencies"] + assert "flashdreams-cam2v" in dependencies + assert "flashdreams-lingbot" in dependencies + assert ( + manifest["project"]["entry-points"]["flashdreams.applications_v2"][ + "cam2v-lingbot" + ] + == "cam2v_lingbot.app:create_app" + ) + + +def test_application_reuses_lingbot_runner_config() -> None: + """Avoid restating the model's pipeline, geometry, rate, or rollout length.""" + application = LingbotCam2VApplication() + runner = RUNNER_LINGBOT_WORLD_FAST_TAEHV_WINDOW15_SINK3 + + assert isinstance(application, Cam2VApplication) + assert application.pipeline_config is runner.pipeline + assert application.defaults.total_blocks == runner.total_blocks + assert application.session_desc().video_width == runner.pixel_width + assert application.session_desc().video_height == runner.pixel_height + assert application.session_desc().frames_per_second_for_step == runner.fps + assert isinstance(create_app(), LingbotCam2VApplication) + + +def test_resolver_builds_conditioning_without_legacy_runtime(tmp_path: Path) -> None: + """Resolve prompt, calibration, and scale entirely in the v2 package.""" + image_path = tmp_path / "image.jpg" + image_path.touch() + prompt_path = tmp_path / "prompt.txt" + prompt_path.write_text(" move through the room \nignored\n") + intrinsic_path = tmp_path / "intrinsics.npy" + np.save(intrinsic_path, np.array([[832.0, 480.0, 416.0, 240.0]])) + pose_path = tmp_path / "poses.npy" + poses = np.repeat(np.eye(4)[None], 13, axis=0) + poses[:, 0, 3] = np.arange(13) + np.save(pose_path, poses) + + result = conditioning.resolve_lingbot_conditioning( + { + "prompt": "", + "prompt_path": prompt_path, + "image_path": image_path, + "pose_path": pose_path, + "intrinsic_path": intrinsic_path, + "world_scale": None, + "example_data": False, + "example_idx": 0, + "pixel_height": 240, + "pixel_width": 416, + } + ) + + assert isinstance(result, Cam2VConditioning) + assert result.prompt == "move through the room" + assert result.first_frame_path == image_path + assert torch.equal( + result.base_intrinsics, + torch.tensor([[416.0, 240.0, 208.0, 120.0]]), + ) + assert result.world_scale == pytest.approx(6.0) + + +def test_explicit_world_scale_does_not_require_replay_poses(tmp_path: Path) -> None: + """Live Cam2V control only needs poses when deriving the normalizer.""" + image_path = tmp_path / "image.jpg" + image_path.touch() + intrinsic_path = tmp_path / "intrinsics.npy" + np.save(intrinsic_path, np.array([832.0, 480.0, 416.0, 240.0])) + + result = conditioning.resolve_lingbot_conditioning( + { + "prompt": "forward", + "image_path": image_path, + "pose_path": None, + "intrinsic_path": intrinsic_path, + "world_scale": 2.5, + "example_data": False, + "example_idx": 0, + "pixel_height": 480, + "pixel_width": 832, + } + ) + + assert result.world_scale == 2.5 + + +def test_example_data_fills_missing_assets( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep the documented example-data command independent of legacy helpers.""" + (tmp_path / "image.jpg").touch() + (tmp_path / "prompt.txt").write_text("example prompt\n") + np.save(tmp_path / "intrinsics.npy", np.array([[832.0, 480.0, 416.0, 240.0]])) + poses = np.repeat(np.eye(4)[None], 9, axis=0) + np.save(tmp_path / "poses.npy", poses) + monkeypatch.setattr(conditioning, "_ensure_example_data", lambda index: tmp_path) + + result = conditioning.resolve_lingbot_conditioning( + { + "prompt": "", + "prompt_path": None, + "image_path": None, + "pose_path": None, + "intrinsic_path": None, + "world_scale": None, + "example_data": True, + "example_idx": 0, + "pixel_height": 480, + "pixel_width": 832, + } + ) + + assert result.prompt == "example prompt" + assert result.first_frame_path == tmp_path / "image.jpg" + assert result.world_scale == 0.0 diff --git a/integrations_v2/cam2v_lingbot/pyproject.toml b/integrations_v2/cam2v_lingbot/pyproject.toml new file mode 100644 index 000000000..2f209bc87 --- /dev/null +++ b/integrations_v2/cam2v_lingbot/pyproject.toml @@ -0,0 +1,32 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "flashdreams-cam2v-lingbot" +version = "0.1.0" +description = "Lingbot camera-to-video application for the FlashDreams v2 API." +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "flashdreams", + "flashdreams-cam2v", + "flashdreams-lingbot", +] + +[tool.uv.sources] +flashdreams = { workspace = true } +flashdreams-cam2v = { workspace = true } +flashdreams-lingbot = { workspace = true } + +[project.entry-points."flashdreams.applications_v2"] +"cam2v-lingbot" = "cam2v_lingbot.app:create_app" + +[tool.setuptools.packages.find] +include = ["cam2v_lingbot*"] + +[tool.uv] +managed = true diff --git a/pyproject.toml b/pyproject.toml index a82f8ed88..fcb1e8129 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,6 +60,7 @@ extraPaths = [ "integrations/wan21", "integrations/wan22", "integrations_v2/color_fade", + "integrations_v2/cam2v_lingbot", "integrations_v2/null_model", "integrations_v2/red_screen", "integrations_v2/t2v_causal_forcing", @@ -94,6 +95,7 @@ extra-paths = [ "integrations/wan21", "integrations/wan22", "integrations_v2/color_fade", + "integrations_v2/cam2v_lingbot", "integrations_v2/null_model", "integrations_v2/red_screen", "integrations_v2/t2v_causal_forcing", diff --git a/uv.lock b/uv.lock index 7a410ea30..6efb02692 100644 --- a/uv.lock +++ b/uv.lock @@ -20,6 +20,8 @@ conflicts = [[ [manifest] members = [ "flashdreams", + "flashdreams-cam2v", + "flashdreams-cam2v-lingbot", "flashdreams-causal-forcing", "flashdreams-color-fade", "flashdreams-cosmos-predict2", @@ -1120,6 +1122,34 @@ cuda13 = [ { name = "torchvision", marker = "sys_platform == 'win32'", specifier = ">=0.24", index = "https://download.pytorch.org/whl/cu130" }, ] +[[package]] +name = "flashdreams-cam2v" +version = "0.1.0" +source = { editable = "apps/cam2v" } +dependencies = [ + { name = "flashdreams", extra = ["local-window", "runners", "serving"] }, +] + +[package.metadata] +requires-dist = [{ name = "flashdreams", extras = ["local-window", "runners", "serving"], editable = "flashdreams" }] + +[[package]] +name = "flashdreams-cam2v-lingbot" +version = "0.1.0" +source = { editable = "integrations_v2/cam2v_lingbot" } +dependencies = [ + { name = "flashdreams" }, + { name = "flashdreams-cam2v" }, + { name = "flashdreams-lingbot" }, +] + +[package.metadata] +requires-dist = [ + { name = "flashdreams", editable = "flashdreams" }, + { name = "flashdreams-cam2v", editable = "apps/cam2v" }, + { name = "flashdreams-lingbot", editable = "integrations/lingbot" }, +] + [[package]] name = "flashdreams-causal-forcing" version = "0.1.0" @@ -1283,6 +1313,7 @@ dependencies = [ { name = "aiohttp" }, { name = "aiortc" }, { name = "flashdreams" }, + { name = "flashdreams-cam2v" }, { name = "mediapy" }, { name = "opencv-python-headless" }, { name = "scipy", version = "1.15.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11' or (extra == 'extra-11-flashdreams-dev' and extra == 'group-11-flashdreams-cuda12') or (extra == 'group-11-flashdreams-cuda12' and extra == 'group-11-flashdreams-cuda13')" }, @@ -1301,6 +1332,7 @@ requires-dist = [ { name = "aiohttp", specifier = ">=3.9" }, { name = "aiortc", specifier = ">=1.9" }, { name = "flashdreams", editable = "flashdreams" }, + { name = "flashdreams-cam2v", editable = "apps/cam2v" }, { name = "mediapy", specifier = ">=1.1" }, { name = "opencv-python-headless", specifier = ">=4.5" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" },