Spaces:
Running on L40S
Running on L40S
Cosmos3-Action-Viewer / cosmos-framework /packages /diffusers-cosmos3 /diffusers_cosmos3 /sequence_packing.py
| # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. | |
| # SPDX-License-Identifier: OpenMDW-1.1 | |
| # Copied from cosmos3._src.vfm.datasets.sequence_packing | |
| import math | |
| from dataclasses import dataclass, field | |
| from typing import Any, Dict, List, Optional, Tuple | |
| import torch | |
| class GenerationDataClean: | |
| batch_size: int | |
| is_image_batch: bool | |
| raw_state_vision: Optional[List[torch.Tensor]] = None | |
| x0_tokens_vision: Optional[List[torch.Tensor]] = None | |
| fps_vision: Optional[torch.Tensor] = None | |
| num_vision_items_per_sample: Optional[List[int]] = None | |
| x0_tokens_action: Optional[List[torch.Tensor]] = None | |
| fps_action: Optional[torch.Tensor] = None | |
| action_domain_id: Optional[List[torch.Tensor]] = None | |
| raw_action_dim: Optional[List[Optional[torch.Tensor]]] = None | |
| x0_tokens_sound: Optional[List[torch.Tensor]] = None | |
| fps_sound: Optional[torch.Tensor] = None | |
| # "Fake" types for readability; everything is plain dict at runtime. | |
| FactoredSequencePack = dict[str, Any] | |
| JointSequencePack = dict[str, Any] | |
| SequencePack = FactoredSequencePack | JointSequencePack | |
| # ------------------------------------ | |
| # Internal helpers | |
| # ------------------------------------ | |
| def _pad_to_N(N, x: torch.Tensor) -> torch.Tensor: | |
| assert x.shape[0] <= N | |
| padded = x.new_zeros((N, *x.shape[1:])) | |
| padded[: x.shape[0]] = x | |
| return padded | |
| def _pad( | |
| causal_seq: torch.Tensor, full_only_seq: torch.Tensor, max_causal_len: int, max_full_len: int | |
| ) -> tuple[torch.Tensor, torch.Tensor]: | |
| causal_seq = _pad_to_N(max_causal_len, causal_seq) | |
| full_only_seq = _pad_to_N(max_full_len, full_only_seq) | |
| return causal_seq, full_only_seq | |
| def _compute_mode_indices_and_offsets( | |
| split_lens: list[int], | |
| attn_modes: list[str], | |
| mode: str, | |
| device: torch.device, | |
| ) -> tuple[torch.Tensor, torch.Tensor]: | |
| indices = [] | |
| offsets = [0] | |
| next_offset = 0 | |
| start = 0 | |
| for split_len, attn_mode in zip(split_lens, attn_modes): | |
| if attn_mode == mode: | |
| indices.extend(range(start, start + split_len)) | |
| next_offset += split_len | |
| offsets.append(next_offset) | |
| start += split_len | |
| return ( | |
| torch.tensor(indices, dtype=torch.int32, device=device), | |
| torch.tensor(offsets, dtype=torch.int32, device=device), | |
| ) | |
| def _init_sequence_pack( | |
| sample_lens: list[int], | |
| split_lens: list[int], | |
| attn_modes: list[str], | |
| device: torch.device, | |
| ) -> dict: | |
| _max_sample_len = max(sample_lens) | |
| _max_causal_len = max((split_lens[i] for i in range(len(split_lens)) if attn_modes[i] == "causal"), default=0) | |
| _max_full_len = max((split_lens[i] for i in range(len(split_lens)) if attn_modes[i] == "full"), default=0) | |
| sample_lens_cu = torch.tensor([0] + sample_lens, device=device, dtype=torch.int32) | |
| _sample_offsets = torch.cumsum(sample_lens_cu, dim=0, dtype=torch.int32) | |
| _causal_indices, _causal_seq_offsets = _compute_mode_indices_and_offsets(split_lens, attn_modes, "causal", device) | |
| _full_indices, _full_only_seq_offsets = _compute_mode_indices_and_offsets(split_lens, attn_modes, "full", device) | |
| return dict( | |
| sample_offsets=_sample_offsets, | |
| max_sample_len=_max_sample_len, | |
| max_causal_len=_max_causal_len, | |
| max_full_len=_max_full_len, | |
| _causal_indices=_causal_indices, | |
| _full_indices=_full_indices, | |
| _causal_seq_offsets=_causal_seq_offsets, | |
| _full_only_seq_offsets=_full_only_seq_offsets, | |
| _num_causal_tokens=len(_causal_indices), | |
| _num_full_tokens=len(_full_indices), | |
| split_lens=split_lens, | |
| attn_modes=attn_modes, | |
| ) | |
| def _find_non_causal_text_token_idx( | |
| attn_modes: list[str], | |
| split_lens: list[int], | |
| und_token_indexes: list[int], | |
| ) -> list[int]: | |
| out = [] | |
| full_offset = 0 | |
| packed_idx = 0 | |
| und_token_set = set(und_token_indexes) | |
| for attn_mode, split_len in zip(attn_modes, split_lens): | |
| if attn_mode == "full": | |
| for local_idx, split_idx in enumerate(range(packed_idx, packed_idx + split_len)): | |
| if split_idx in und_token_set: | |
| out.append(full_offset + local_idx) | |
| full_offset += split_len | |
| packed_idx += split_len | |
| return out | |
| def factored_from_joint_sequence( | |
| packed_sequence: torch.Tensor, | |
| attn_modes: list[str], | |
| split_lens: list[int], | |
| sample_lens: list[int], | |
| packed_und_token_indexes: torch.Tensor, | |
| packed_gen_token_indexes: torch.Tensor, | |
| is_image_batch: bool = False, | |
| cp_world_size: int = 1, | |
| pad_for_cuda_graphs: bool = False, | |
| ) -> FactoredSequencePack: | |
| non_causal_text_idxs = _find_non_causal_text_token_idx(attn_modes, split_lens, packed_und_token_indexes.tolist()) | |
| assert len(non_causal_text_idxs) == 0, "non_causal_text_idxs should be empty" | |
| assert sum(sample_lens) == packed_sequence.shape[0] | |
| meta = _init_sequence_pack(sample_lens, split_lens, attn_modes, packed_sequence.device) | |
| causal_seq = packed_sequence[meta["_causal_indices"]] | |
| full_only_seq = packed_sequence[meta["_full_indices"]] | |
| return { | |
| **meta, | |
| "max_num_tokens": sum(sample_lens), | |
| "causal_seq": causal_seq, | |
| "full_only_seq": full_only_seq, | |
| "is_sharded": False, | |
| } | |
| class SplitInfo: | |
| def __init__( | |
| self, | |
| split_lens: list[int], | |
| attn_modes: list[str], | |
| sample_lens: list[int], | |
| actual_len: int, | |
| ): | |
| assert sum(sample_lens) == sum(split_lens) | |
| max_causal_len = 0 | |
| max_full_len = 0 | |
| for split_len, attn_mode in zip(split_lens, attn_modes): | |
| if attn_mode == "causal": | |
| max_causal_len = max(max_causal_len, split_len) | |
| elif attn_mode == "full": | |
| max_full_len = max(max_full_len, split_len) | |
| self.max_causal_len = max_causal_len | |
| self.max_full_len = max_full_len | |
| self.max_sample_len = max(sample_lens) | |
| self.split_lens = split_lens | |
| self.attn_modes = attn_modes | |
| self.sample_lens = sample_lens | |
| def build_packed_sequence( | |
| joint_attn_implementation: str, | |
| *, | |
| packed_sequence: torch.Tensor, | |
| attn_modes: list[str], | |
| split_lens: list[int], | |
| sample_lens: list[int], | |
| packed_und_token_indexes: torch.LongTensor, | |
| packed_gen_token_indexes: torch.LongTensor, | |
| num_heads: int, | |
| head_dim: int, | |
| num_layers: int, | |
| token_shapes=None, | |
| natten_parameter_list=None, | |
| block_size: int = 128, | |
| is_image_batch: bool = False, | |
| cp_world_size: int = 1, | |
| video_temporal_causal: bool = False, | |
| vision_token_shapes=None, | |
| action_token_shapes=None, | |
| temporal_compression_factor_vision: int = 4, | |
| null_action_supertokens: bool = False, | |
| pad_for_cuda_graphs: bool = False, | |
| ) -> tuple[FactoredSequencePack, SplitInfo, None]: | |
| assert joint_attn_implementation == "two_way", ( | |
| f"Only two_way attention is supported, got {joint_attn_implementation!r}" | |
| ) | |
| device = packed_sequence.device | |
| attention_meta = SplitInfo( | |
| split_lens=split_lens, | |
| attn_modes=attn_modes, | |
| sample_lens=sample_lens, | |
| actual_len=int(packed_sequence.shape[0]), | |
| ) | |
| input_pack = factored_from_joint_sequence( | |
| packed_sequence=packed_sequence, | |
| attn_modes=attn_modes, | |
| split_lens=split_lens, | |
| sample_lens=sample_lens, | |
| packed_und_token_indexes=packed_und_token_indexes.to(device), | |
| packed_gen_token_indexes=packed_gen_token_indexes.to(device), | |
| is_image_batch=is_image_batch, | |
| cp_world_size=cp_world_size, | |
| pad_for_cuda_graphs=pad_for_cuda_graphs, | |
| ) | |
| input_pack.pop("split_lens", None) | |
| input_pack.pop("attn_modes", None) | |
| return input_pack, attention_meta, None | |
| def _ensure_core_metadata(pack: SequencePack) -> None: | |
| required = [ | |
| "sample_offsets", | |
| "max_sample_len", | |
| "max_causal_len", | |
| "max_full_len", | |
| "_causal_indices", | |
| "_full_indices", | |
| "_causal_seq_offsets", | |
| "_full_only_seq_offsets", | |
| "is_sharded", | |
| ] | |
| for key in required: | |
| if key not in pack: | |
| raise KeyError(f"Missing required pack field: {key}") | |
| def from_mode_splits( | |
| causal_seq: torch.Tensor, | |
| full_only_seq: torch.Tensor, | |
| orig: FactoredSequencePack | JointSequencePack, | |
| is_sharded: bool | None = None, | |
| ): | |
| """ | |
| Create a new sequence pack from two mode splits. | |
| Args: | |
| causal_seq (torch.Tensor): The causal sequence. | |
| full_only_seq (torch.Tensor): The full-only sequence. | |
| orig (FactoredSequencePack | JointSequencePack): The metadata source to copy from. | |
| is_sharded (bool | None): If True, create a local pack for context parallel. | |
| If None, inherits from orig. | |
| """ | |
| _ensure_core_metadata(orig) | |
| if is_sharded is None: | |
| is_sharded = orig.get("is_sharded", False) | |
| if "packed_sequence" in orig: | |
| all_len = int(orig["_causal_indices"].shape[0] + orig["_full_indices"].shape[0]) | |
| packed_sequence = causal_seq.new_zeros((all_len, *causal_seq.shape[1:])) # [seq_len,D] | |
| packed_sequence[orig["_causal_indices"]] = causal_seq | |
| packed_sequence[orig["_full_indices"]] = full_only_seq | |
| return from_joint(packed_sequence, orig) | |
| else: | |
| out = dict(orig) | |
| out["causal_seq"] = causal_seq | |
| out["full_only_seq"] = full_only_seq | |
| out["is_sharded"] = is_sharded | |
| return out | |
| # ------------------------------------ | |
| # Public API | |
| # ------------------------------------ | |
| def zeros_like(orig: FactoredSequencePack | JointSequencePack, shape: Tuple[int, ...] | torch.Size | None = None): | |
| """ | |
| Create a new sequence pack with the same metadata as the original, but with all tokens set to zero. | |
| Args: | |
| orig (FactoredSequencePack | JointSequencePack): The original sequence pack to copy metadata from. | |
| shape (Tuple[int, ...] | torch.Size | None): The shape of the new sequence pack. If None, the shape will be the same as the original. | |
| """ | |
| _ensure_core_metadata(orig) | |
| if "packed_sequence" in orig: | |
| if shape is None: | |
| shape_ = orig["packed_sequence"].shape | |
| else: | |
| assert len(shape) >= 1 and shape[0] == -1 | |
| shape_ = (orig["packed_sequence"].shape[0],) + tuple(shape)[1:] | |
| packed_sequence = torch.zeros( | |
| shape_, device=orig["packed_sequence"].device, dtype=orig["packed_sequence"].dtype | |
| ) # [seq_len,D] | |
| return from_joint(packed_sequence, orig) | |
| else: | |
| if shape is None: | |
| shape_causal = orig["causal_seq"].shape | |
| shape_full = orig["full_only_seq"].shape | |
| else: | |
| assert len(shape) >= 1 and shape[0] == -1 | |
| shape_causal = (orig["causal_seq"].shape[0],) + tuple(shape)[1:] | |
| shape_full = (orig["full_only_seq"].shape[0],) + tuple(shape)[1:] | |
| causal_seq = torch.zeros( | |
| shape_causal, device=orig["causal_seq"].device, dtype=orig["causal_seq"].dtype | |
| ) # [N_causal_tokens,D] | |
| full_only_seq = torch.zeros( | |
| shape_full, device=orig["full_only_seq"].device, dtype=orig["full_only_seq"].dtype | |
| ) # [N_full_tokens,D] | |
| return from_mode_splits(causal_seq, full_only_seq, orig) | |
| def from_joint(packed_sequence: torch.Tensor, metadata_source: FactoredSequencePack | JointSequencePack): | |
| """ | |
| Create a new sequence pack from a packed sequence and another sequence pack with the same metadata. | |
| Args: | |
| packed_sequence (torch.Tensor): Tensor containing all tokens in the batch of sequences. | |
| metadata_source (FactoredSequencePack | JointSequencePack): The metadata source to copy from. | |
| """ | |
| _ensure_core_metadata(metadata_source) | |
| if "packed_sequence" in metadata_source: | |
| out = dict(metadata_source) | |
| out["packed_sequence"] = packed_sequence | |
| return out | |
| else: | |
| if metadata_source["is_sharded"]: | |
| # Use sharded sequences as is when is_sharded is True (used in Context Parallel) | |
| causal_seq = packed_sequence[: len(metadata_source["causal_seq"])] # [N_causal_tokens,D] | |
| full_only_seq = packed_sequence[len(metadata_source["causal_seq"]) :] # [N_full_tokens,D] | |
| else: | |
| causal_seq = packed_sequence[metadata_source["_causal_indices"]] # [N_causal_tokens,D] | |
| full_only_seq = packed_sequence[metadata_source["_full_indices"]] # [N_full_tokens,D] | |
| causal_seq, full_only_seq = _pad( | |
| causal_seq, | |
| full_only_seq, | |
| max_causal_len=metadata_source["causal_seq"].shape[0], | |
| max_full_len=metadata_source["full_only_seq"].shape[0], | |
| ) | |
| return from_mode_splits(causal_seq, full_only_seq, metadata_source) | |
| def from_und_gen_splits(und_seq: torch.Tensor, gen_seq: torch.Tensor, orig: FactoredSequencePack | JointSequencePack): | |
| """ | |
| Create a new sequence pack from two und/gen splits. | |
| Args: | |
| und_seq (torch.Tensor): The understanding sequence. | |
| gen_seq (torch.Tensor): The generating sequence. | |
| orig (FactoredSequencePack | JointSequencePack): The metadata source to copy from. | |
| """ | |
| # If we have a joint pack (single packed_sequence), place by und/gen indexes. | |
| if "packed_sequence" in orig and "packed_und_token_indexes" in orig and "packed_gen_token_indexes" in orig: | |
| all_len = int(und_seq.shape[0] + gen_seq.shape[0]) | |
| packed_sequence = und_seq.new_zeros((all_len, *und_seq.shape[1:])) # [seq_len,D] | |
| packed_sequence[orig["packed_und_token_indexes"]] = und_seq | |
| packed_sequence[orig["packed_gen_token_indexes"]] = gen_seq | |
| return from_joint(packed_sequence, orig) | |
| # Otherwise, treat und/gen as mode splits (und == causal; gen == full). | |
| return from_mode_splits(und_seq, gen_seq, orig) | |
| def get_und_seq(pack: SequencePack) -> torch.Tensor: | |
| """ | |
| Get all understanding tokens in a sequence pack in a single tensor. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the understanding sequence from. | |
| Returns: | |
| torch.Tensor: All understanding tokens concatenated over all sequences in the batch. | |
| """ | |
| if "causal_seq" in pack: | |
| return pack["causal_seq"] | |
| if "packed_sequence" in pack and "packed_und_token_indexes" in pack: | |
| return pack["packed_sequence"][pack["packed_und_token_indexes"]] | |
| raise KeyError("Cannot derive und_seq from provided pack") | |
| def set_und_seq(pack: SequencePack, value: torch.Tensor) -> None: | |
| """ | |
| Override the understanding tokens in a sequence pack. | |
| The order of tokens passed in must correspond to the order of tokens returned by get_und_seq. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to set the understanding sequence in. | |
| value (torch.Tensor): The understanding sequence to set. | |
| """ | |
| if "packed_sequence" in pack and "packed_und_token_indexes" in pack: | |
| pack["packed_sequence"][pack["packed_und_token_indexes"]] = value | |
| elif "causal_seq" in pack: | |
| pack["causal_seq"] = value | |
| else: | |
| raise KeyError("Cannot set und_seq from provided pack") | |
| def get_gen_seq(pack: SequencePack) -> torch.Tensor: | |
| """ | |
| Get all generating tokens in a sequence pack in a single tensor. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the generating sequence from. | |
| Returns: | |
| torch.Tensor: All generating tokens concatenated over all sequences in the batch. | |
| """ | |
| if "full_only_seq" in pack: | |
| return pack["full_only_seq"] | |
| if "packed_sequence" in pack and "packed_gen_token_indexes" in pack: | |
| return pack["packed_sequence"][pack["packed_gen_token_indexes"]] | |
| raise KeyError("Cannot derive gen_seq from provided pack") | |
| def set_gen_seq(pack: SequencePack, value: torch.Tensor) -> None: | |
| """ | |
| Override the generating tokens in a sequence pack. | |
| The order of tokens passed in must correspond to the order of tokens returned by get_gen_seq. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to set the generating sequence in. | |
| value (torch.Tensor): The generating sequence to set. | |
| """ | |
| if "packed_sequence" in pack and "packed_gen_token_indexes" in pack: | |
| pack["packed_sequence"][pack["packed_gen_token_indexes"]] = value | |
| elif "full_only_seq" in pack: | |
| pack["full_only_seq"] = value | |
| else: | |
| raise KeyError("Cannot set gen_seq from provided pack") | |
| def get_device_and_dtype(pack: SequencePack) -> Tuple[torch.device, torch.dtype]: | |
| """ | |
| Get the device and dtype of a sequence pack. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the device and dtype from. | |
| Returns: | |
| Tuple[torch.device, torch.dtype]: The device and dtype of the sequence pack. | |
| """ | |
| if "packed_sequence" in pack: | |
| return pack["packed_sequence"].device, pack["packed_sequence"].dtype | |
| if "causal_seq" in pack and "full_only_seq" in pack: | |
| return pack["causal_seq"].device, pack["causal_seq"].dtype | |
| raise KeyError("Cannot derive device and dtype from provided pack") | |
| def get_all_seq(pack: SequencePack) -> torch.Tensor: | |
| """ | |
| Get all tokens in a sequence pack in a single tensor. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the all sequence from. | |
| Returns: | |
| torch.Tensor: All tokens concatenated over all sequences in the batch. | |
| """ | |
| if "all_seq" in pack: | |
| return pack["all_seq"] | |
| if "packed_sequence" in pack: | |
| return pack["packed_sequence"] | |
| if "causal_seq" in pack and "full_only_seq" in pack: | |
| _ensure_core_metadata(pack) | |
| if pack["is_sharded"]: | |
| assert False, "get_all_seq is not supported in context parallel sharded mode" | |
| else: | |
| out = pack["causal_seq"].new_zeros( | |
| int(pack["_causal_indices"].shape[0] + pack["_full_indices"].shape[0]), *pack["causal_seq"].shape[1:] | |
| ) # [seq_len,D] | |
| if pack["causal_seq"].shape[0] > 0: | |
| out[pack["_causal_indices"]] = pack["causal_seq"][: pack["_causal_indices"].shape[0]] | |
| if pack["full_only_seq"].shape[0] > 0: | |
| out[pack["_full_indices"]] = pack["full_only_seq"][: pack["_full_indices"].shape[0]] | |
| return out | |
| raise KeyError("Cannot derive all_seq from provided pack") | |
| def get_causal_seq(pack: SequencePack) -> Tuple[torch.Tensor, torch.Tensor]: | |
| """ | |
| Get the causal sequence and its offsets in a sequence pack. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the causal sequence from. | |
| Returns: | |
| Tuple[torch.Tensor, torch.Tensor]: The concatenated causal sub-sequences and the starting offset for each sub-sequence. | |
| """ | |
| _ensure_core_metadata(pack) | |
| if "causal_seq" in pack: | |
| return pack["causal_seq"], pack["_causal_seq_offsets"] | |
| assert "packed_sequence" in pack | |
| return pack["packed_sequence"][pack["_causal_indices"]], pack["_causal_seq_offsets"] | |
| def get_full_only_seq(pack: SequencePack) -> Tuple[torch.Tensor, torch.Tensor]: | |
| """ | |
| Get the full-only sequence and its offsets in a sequence pack. | |
| Args: | |
| pack (FactoredSequencePack | JointSequencePack): The sequence pack to get the full-only sequence from. | |
| Returns: | |
| Tuple[torch.Tensor, torch.Tensor]: The concatenated full-only sub-sequences and the starting offset for each sub-sequence. | |
| """ | |
| _ensure_core_metadata(pack) | |
| if "full_only_seq" in pack: | |
| return pack["full_only_seq"], pack["_full_only_seq_offsets"] | |
| assert "packed_sequence" in pack | |
| return pack["packed_sequence"][pack["_full_indices"]], pack["_full_only_seq_offsets"] | |
| # ============================================================================ | |
| # 3D mRoPE position ID utilities | |
| # Copied from cosmos3._src.vfm.models.mot.unified_3dmrope_utils | |
| # ============================================================================ | |
| def get_3d_mrope_ids_text_tokens( | |
| num_tokens: int, | |
| temporal_offset: int | float, | |
| use_float_positions: bool = False, | |
| ) -> tuple[torch.Tensor, int | float]: | |
| """Generate 3D mRoPE position IDs for text tokens. | |
| For text tokens, all three axes (temporal, height, width) share the same | |
| monotonically increasing position IDs, starting from ``temporal_offset``. | |
| Args: | |
| num_tokens: Number of text tokens. | |
| temporal_offset: Current temporal offset to start from. | |
| use_float_positions: If True, generate float position IDs. | |
| Returns: | |
| Tuple of position IDs tensor of shape (3, num_tokens) and updated temporal offset. | |
| """ | |
| if use_float_positions: | |
| ids = torch.arange(num_tokens, dtype=torch.float32) + temporal_offset | |
| else: | |
| ids = torch.arange(num_tokens, dtype=torch.long) + int(temporal_offset) | |
| mrope_ids = ids.unsqueeze(0).expand(3, -1).contiguous() # [3,num_tokens] | |
| next_temporal_offset = temporal_offset + num_tokens | |
| return mrope_ids, next_temporal_offset | |
| def get_3d_mrope_ids_vae_tokens( | |
| grid_t: int, | |
| grid_h: int, | |
| grid_w: int, | |
| temporal_offset: int | float, | |
| reset_spatial_indices: bool = True, | |
| fps: float | None = None, | |
| base_fps: float = 24.0, | |
| temporal_compression_factor: int = 4, | |
| base_temporal_compression_factor: int | None = None, | |
| start_frame_offset: int = 0, | |
| ) -> tuple[torch.Tensor, int | float]: | |
| """Generate 3D mRoPE position IDs for VAE vision tokens (image/video latents). | |
| Args: | |
| grid_t: Number of temporal frames in the latent grid. | |
| grid_h: Height of the latent grid (after patchification). | |
| grid_w: Width of the latent grid (after patchification). | |
| temporal_offset: Current temporal offset. | |
| reset_spatial_indices: If True, spatial indices start from 0 for each vision segment. | |
| fps: Frames per second. If None, FPS modulation is disabled. | |
| base_fps: Base FPS for normalization. | |
| temporal_compression_factor: VAE temporal compression factor. | |
| base_temporal_compression_factor: Base temporal compression factor. | |
| start_frame_offset: Offset added to frame indices before FPS scaling. | |
| Returns: | |
| Tuple of position IDs tensor of shape (3, grid_t * grid_h * grid_w) and updated offset. | |
| """ | |
| fps_modulation_enabled = fps is not None and grid_t > 1 | |
| effective_base_tcf = ( | |
| base_temporal_compression_factor | |
| if base_temporal_compression_factor is not None | |
| else temporal_compression_factor | |
| ) | |
| if fps_modulation_enabled: | |
| tps = fps / temporal_compression_factor | |
| base_tps = base_fps / effective_base_tcf | |
| frame_indices = torch.arange(grid_t, dtype=torch.float32) | |
| scaled_t = (frame_indices + start_frame_offset) / tps * base_tps + temporal_offset | |
| t_index = scaled_t.view(-1, 1).expand(-1, grid_h * grid_w).flatten() | |
| t_dtype = torch.float32 | |
| else: | |
| t_index = ( | |
| torch.arange(grid_t, dtype=torch.long).view(-1, 1).expand(-1, grid_h * grid_w).flatten() | |
| + int(temporal_offset) | |
| + start_frame_offset | |
| ) | |
| t_dtype = torch.long | |
| h_index = torch.arange(grid_h, dtype=torch.long).view(1, -1, 1).expand(grid_t, -1, grid_w).flatten() | |
| w_index = torch.arange(grid_w, dtype=torch.long).view(1, 1, -1).expand(grid_t, grid_h, -1).flatten() | |
| if not reset_spatial_indices: | |
| spatial_offset = int(temporal_offset) | |
| h_index = h_index + spatial_offset | |
| w_index = w_index + spatial_offset | |
| if fps_modulation_enabled: | |
| mrope_ids = torch.stack([t_index, h_index.to(torch.float32), w_index.to(torch.float32)], dim=0) | |
| else: | |
| mrope_ids = torch.stack([t_index, h_index, w_index], dim=0) | |
| max_position = mrope_ids.max().item() | |
| next_temporal_offset = math.ceil(max_position) + 1 | |
| return mrope_ids, next_temporal_offset | |
| # ============================================================================ | |
| # Data structures for sequence packing | |
| # Copied from cosmos3._src.vfm.datasets.sequence_packing | |
| # ============================================================================ | |
| class ModalityData: | |
| """Unified container for a single generation modality's data. | |
| This dataclass serves dual purposes: | |
| 1. During packing: Acts as a builder, accumulating data in lists | |
| 2. After finalize(): Holds finalized tensors ready for model consumption | |
| """ | |
| sequence_indexes: list[int] | torch.Tensor = field(default_factory=list) | |
| timesteps: list[float] | torch.Tensor = field(default_factory=list) | |
| mse_loss_indexes: list[int] | torch.Tensor = field(default_factory=list) | |
| token_shapes: list = field(default_factory=list) | |
| tokens: list[torch.Tensor] = field(default_factory=list) | |
| condition_mask: list[torch.Tensor] = field(default_factory=list) | |
| noisy_frame_indexes: list[torch.Tensor] = field(default_factory=list) | |
| domain_id: list[torch.Tensor] = field(default_factory=list) | |
| raw_action_dim: list[torch.Tensor | None] | None = field(default_factory=list) | |
| def to_cuda(self) -> None: | |
| if isinstance(self.sequence_indexes, torch.Tensor): | |
| self.sequence_indexes = self.sequence_indexes.cuda() | |
| if isinstance(self.timesteps, torch.Tensor): | |
| self.timesteps = self.timesteps.cuda() | |
| if isinstance(self.mse_loss_indexes, torch.Tensor): | |
| self.mse_loss_indexes = self.mse_loss_indexes.cuda() | |
| self.tokens = [token.cuda() for token in self.tokens] | |
| self.condition_mask = [cm.cuda() for cm in self.condition_mask] | |
| self.noisy_frame_indexes = [ni.cuda() for ni in self.noisy_frame_indexes] | |
| self.domain_id = [d.cuda() for d in self.domain_id] | |
| if self.raw_action_dim is not None: | |
| self.raw_action_dim = [d.cuda() if d is not None else None for d in self.raw_action_dim] | |
| class PackedSequence: | |
| """Unified sequence container - works as builder during packing and final output.""" | |
| # Sequence structure | |
| sample_lens: list[int] = field(default_factory=list) | |
| split_lens: list[int] = field(default_factory=list) | |
| attn_modes: list[str] = field(default_factory=list) | |
| is_image_batch: bool = False | |
| sequence_length: int = 0 | |
| # Build-time tracking | |
| curr: int = 0 | |
| # Text modality (list during build, tensor after finalize) | |
| text_ids: list[int] | torch.Tensor = field(default_factory=list) | |
| text_indexes: list[int] | torch.Tensor = field(default_factory=list) | |
| position_ids: list[int] | torch.Tensor = field(default_factory=list) | |
| # Loss computation - Cross Entropy (text) | |
| label_ids: list[int] | torch.Tensor | None = field(default_factory=list) | |
| ce_loss_indexes: list[int] | torch.Tensor | None = field(default_factory=list) | |
| ce_loss_weights: list[float] | torch.Tensor | None = field(default_factory=list) | |
| # Build-time mRoPE tracking | |
| _use_mrope: bool = False | |
| _mrope_temporal_offset: int | float = 0 | |
| _mrope_reset_spatial: bool = True | |
| # Temporal causal | |
| null_action_supertokens: bool = False | |
| # Generation modalities | |
| vision: ModalityData | None = None | |
| action: ModalityData | None = None | |
| sound: ModalityData | None = None | |
| def finalize( | |
| self, | |
| gen_data_clean: "GenerationDataClean", | |
| ) -> "PackedSequence": | |
| """Convert all lists to tensors and compute derived values.""" | |
| sequence_length = sum(self.sample_lens) | |
| sample_lens = self.sample_lens.copy() | |
| split_lens = self.split_lens.copy() | |
| attn_modes = self.attn_modes.copy() | |
| label_ids: torch.Tensor | None = None | |
| ce_loss_indexes: torch.Tensor | None = None | |
| ce_loss_weights: torch.Tensor | None = None | |
| if self.label_ids and len(self.label_ids) > 0: | |
| label_ids = torch.tensor(self.label_ids) | |
| ce_loss_indexes = torch.tensor(self.ce_loss_indexes) | |
| ce_loss_weights = torch.tensor(self.ce_loss_weights) | |
| vision: ModalityData | None = None | |
| if self.vision is not None and len(self.vision.sequence_indexes) > 0: | |
| vision = ModalityData( | |
| sequence_indexes=torch.tensor(self.vision.sequence_indexes, dtype=torch.long), | |
| timesteps=torch.tensor(self.vision.timesteps), | |
| mse_loss_indexes=torch.tensor(self.vision.mse_loss_indexes, dtype=torch.long), | |
| token_shapes=list(self.vision.token_shapes), | |
| tokens=self.vision.tokens, | |
| condition_mask=list(self.vision.condition_mask), | |
| noisy_frame_indexes=list(self.vision.noisy_frame_indexes), | |
| ) | |
| action: ModalityData | None = None | |
| if self.action is not None and len(self.action.sequence_indexes) > 0: | |
| action = ModalityData( | |
| sequence_indexes=torch.tensor(self.action.sequence_indexes, dtype=torch.long), | |
| timesteps=torch.tensor(self.action.timesteps), | |
| mse_loss_indexes=torch.tensor(self.action.mse_loss_indexes, dtype=torch.long), | |
| token_shapes=list(self.action.token_shapes), | |
| tokens=self.action.tokens, | |
| condition_mask=list(self.action.condition_mask), | |
| noisy_frame_indexes=list(self.action.noisy_frame_indexes), | |
| domain_id=( | |
| gen_data_clean.action_domain_id | |
| if gen_data_clean.action_domain_id is not None | |
| else [torch.zeros(1, dtype=torch.long)] * len(self.action.token_shapes) | |
| ), | |
| raw_action_dim=gen_data_clean.raw_action_dim, | |
| ) | |
| sound: ModalityData | None = None | |
| if self.sound is not None and len(self.sound.sequence_indexes) > 0: | |
| sound = ModalityData( | |
| sequence_indexes=torch.tensor(self.sound.sequence_indexes, dtype=torch.long), | |
| timesteps=torch.tensor(self.sound.timesteps), | |
| mse_loss_indexes=torch.tensor(self.sound.mse_loss_indexes, dtype=torch.long), | |
| token_shapes=list(self.sound.token_shapes), | |
| tokens=self.sound.tokens, | |
| condition_mask=list(self.sound.condition_mask), | |
| noisy_frame_indexes=list(self.sound.noisy_frame_indexes), | |
| ) | |
| if self._use_mrope and len(self.position_ids) > 0 and isinstance(self.position_ids[0], torch.Tensor): | |
| mrope_tensors: list[torch.Tensor] = self.position_ids # type: ignore[assignment] | |
| position_ids = torch.cat(mrope_tensors, dim=1) # [3,actual_seq_len] | |
| else: | |
| position_ids = torch.tensor(self.position_ids) # [seq_len] | |
| return PackedSequence( | |
| sequence_length=sequence_length, | |
| sample_lens=sample_lens, | |
| split_lens=split_lens, | |
| attn_modes=attn_modes, | |
| is_image_batch=gen_data_clean.is_image_batch, | |
| text_ids=torch.tensor(self.text_ids, dtype=torch.long), | |
| text_indexes=torch.tensor(self.text_indexes, dtype=torch.long), | |
| position_ids=position_ids, | |
| label_ids=label_ids, | |
| ce_loss_indexes=ce_loss_indexes, | |
| ce_loss_weights=ce_loss_weights, | |
| vision=vision, | |
| action=action, | |
| sound=sound, | |
| null_action_supertokens=self.null_action_supertokens, | |
| ) | |
| def to_cuda(self) -> None: | |
| if isinstance(self.text_ids, torch.Tensor): | |
| self.text_ids = self.text_ids.cuda() | |
| if isinstance(self.text_indexes, torch.Tensor): | |
| self.text_indexes = self.text_indexes.cuda() | |
| if isinstance(self.position_ids, torch.Tensor): | |
| self.position_ids = self.position_ids.cuda() | |
| if isinstance(self.label_ids, torch.Tensor): | |
| self.label_ids = self.label_ids.cuda() | |
| if isinstance(self.ce_loss_indexes, torch.Tensor): | |
| self.ce_loss_indexes = self.ce_loss_indexes.cuda() | |
| if isinstance(self.ce_loss_weights, torch.Tensor): | |
| self.ce_loss_weights = self.ce_loss_weights.cuda() | |
| if self.vision is not None: | |
| self.vision.to_cuda() | |
| if self.action is not None: | |
| self.action.to_cuda() | |
| if self.sound is not None: | |
| self.sound.to_cuda() | |
| class SequencePlan: | |
| """Plan describing which modalities are present in a sample.""" | |
| has_text: bool | |
| has_vision: bool = False | |
| condition_frame_indexes_vision: list[int] = field(default_factory=list) | |
| has_action: bool = False | |
| condition_frame_indexes_action: list[int] = field(default_factory=list) | |
| has_sound: bool = False | |
| condition_frame_indexes_sound: list[int] = field(default_factory=list) | |
| def as_dict(self) -> dict: | |
| return { | |
| "has_text": self.has_text, | |
| "has_vision": self.has_vision, | |
| "has_action": self.has_action, | |
| "has_sound": self.has_sound, | |
| "condition_frame_indexes_vision": self.condition_frame_indexes_vision, | |
| "condition_frame_indexes_action": self.condition_frame_indexes_action, | |
| "condition_frame_indexes_sound": self.condition_frame_indexes_sound, | |
| } | |
| # ============================================================================ | |
| # Helper functions for packing sequences | |
| # ============================================================================ | |
| def _pack_text_tokens( | |
| packed_seq: PackedSequence, | |
| text_ids: List[int], | |
| special_tokens: Dict[str, int], | |
| curr_rope_id: int, | |
| has_generation: bool, | |
| use_float_positions: bool = False, | |
| ) -> Tuple[int, int, int]: | |
| """Pack text tokens into the sequence.""" | |
| assert isinstance(packed_seq.text_ids, list), "PackedSequence must be in build mode" | |
| assert isinstance(packed_seq.text_indexes, list) | |
| assert isinstance(packed_seq.position_ids, list) | |
| assert isinstance(packed_seq.label_ids, list) | |
| assert isinstance(packed_seq.ce_loss_indexes, list) | |
| assert isinstance(packed_seq.ce_loss_weights, list) | |
| curr = packed_seq.curr | |
| if "bos_token_id" in special_tokens: | |
| shifted_text_ids = [special_tokens["bos_token_id"]] + text_ids | |
| else: | |
| shifted_text_ids = text_ids | |
| split_len = 0 | |
| packed_seq.text_ids.extend(shifted_text_ids) | |
| packed_seq.text_indexes.extend(range(curr, curr + len(shifted_text_ids))) | |
| packed_seq.ce_loss_indexes.extend(range(curr, curr + len(shifted_text_ids))) | |
| packed_seq.ce_loss_weights.extend([1.0] * len(shifted_text_ids)) | |
| packed_seq.label_ids.extend(text_ids[1:] + [special_tokens["eos_token_id"]]) | |
| curr += len(shifted_text_ids) | |
| split_len += len(shifted_text_ids) | |
| packed_seq.text_ids.append(special_tokens["eos_token_id"]) | |
| packed_seq.text_indexes.append(curr) | |
| curr += 1 | |
| split_len += 1 | |
| if has_generation: | |
| packed_seq.text_ids.append(special_tokens["start_of_generation"]) | |
| packed_seq.text_indexes.append(curr) | |
| curr += 1 | |
| split_len += 1 | |
| if packed_seq._use_mrope: | |
| text_mrope_ids, packed_seq._mrope_temporal_offset = get_3d_mrope_ids_text_tokens( | |
| num_tokens=split_len, | |
| temporal_offset=packed_seq._mrope_temporal_offset, | |
| use_float_positions=use_float_positions, | |
| ) | |
| packed_seq.position_ids.append(text_mrope_ids) | |
| else: | |
| packed_seq.position_ids.extend(range(curr_rope_id, curr_rope_id + split_len)) | |
| packed_seq.attn_modes.append("causal") | |
| packed_seq.split_lens.append(split_len) | |
| packed_seq.curr = curr | |
| return curr_rope_id + split_len, split_len, split_len | |
| def _pack_vision_tokens( | |
| packed_seq: PackedSequence, | |
| input_vision_tokens: torch.Tensor, | |
| condition_frame_indexes_vision: list[int], | |
| input_timestep: float | torch.Tensor, | |
| curr_rope_id: int, | |
| latent_patch_size: int = 1, | |
| vision_fps: float | None = None, | |
| enable_fps_modulation: bool = False, | |
| base_fps: float = 24.0, | |
| temporal_compression_factor: int = 4, | |
| ) -> int: | |
| """Pack vision tokens into the sequence.""" | |
| assert isinstance(packed_seq.position_ids, list), "PackedSequence must be in build mode" | |
| curr = packed_seq.curr | |
| vision_split_len = 0 | |
| if packed_seq.vision is None: | |
| packed_seq.vision = ModalityData() | |
| assert isinstance(packed_seq.vision.sequence_indexes, list) | |
| assert isinstance(packed_seq.vision.mse_loss_indexes, list) | |
| assert isinstance(packed_seq.vision.timesteps, list) | |
| assert isinstance(packed_seq.vision.tokens, list) | |
| _, _, latent_t, latent_h, latent_w = input_vision_tokens.shape | |
| if latent_patch_size < 1: | |
| raise ValueError(f"latent_patch_size must be >= 1, got {latent_patch_size}") | |
| patch_h = math.ceil(latent_h / latent_patch_size) | |
| patch_w = math.ceil(latent_w / latent_patch_size) | |
| packed_seq.vision.token_shapes.append((latent_t, patch_h, patch_w)) | |
| packed_seq.vision.tokens.append(input_vision_tokens) | |
| num_vision_tokens = latent_t * patch_h * patch_w | |
| packed_seq.vision.sequence_indexes.extend(range(curr, curr + num_vision_tokens)) | |
| condition_set = {idx for idx in condition_frame_indexes_vision if 0 <= idx < latent_t} | |
| assert isinstance(packed_seq.vision.condition_mask, list) | |
| vision_condition_mask = torch.zeros( | |
| (latent_t, 1, 1), device=input_vision_tokens.device, dtype=input_vision_tokens.dtype | |
| ) | |
| for frame_idx in condition_set: | |
| vision_condition_mask[frame_idx, 0, 0] = 1.0 | |
| packed_seq.vision.condition_mask.append(vision_condition_mask) | |
| vision_noisy_frame_indexes = torch.tensor( | |
| [idx for idx in range(latent_t) if idx not in condition_set], | |
| device=input_vision_tokens.device, | |
| dtype=torch.long, | |
| ) | |
| assert isinstance(packed_seq.vision.noisy_frame_indexes, list) | |
| packed_seq.vision.noisy_frame_indexes.append(vision_noisy_frame_indexes) | |
| frame_token_stride = patch_h * patch_w | |
| for frame_idx in range(latent_t): | |
| if frame_idx in condition_set: | |
| continue | |
| frame_start = curr + frame_idx * frame_token_stride | |
| frame_end = frame_start + frame_token_stride | |
| packed_seq.vision.mse_loss_indexes.extend(range(frame_start, frame_end)) | |
| if isinstance(input_timestep, torch.Tensor): | |
| frame_ts = input_timestep[frame_idx].item() | |
| else: | |
| frame_ts = input_timestep | |
| packed_seq.vision.timesteps.extend([frame_ts] * frame_token_stride) | |
| curr += num_vision_tokens | |
| vision_split_len += num_vision_tokens | |
| if packed_seq._use_mrope: | |
| effective_fps = vision_fps if enable_fps_modulation else None | |
| vision_mrope_ids, packed_seq._mrope_temporal_offset = get_3d_mrope_ids_vae_tokens( | |
| grid_t=latent_t, | |
| grid_h=patch_h, | |
| grid_w=patch_w, | |
| temporal_offset=packed_seq._mrope_temporal_offset, | |
| reset_spatial_indices=packed_seq._mrope_reset_spatial, | |
| fps=effective_fps, | |
| base_fps=base_fps, | |
| temporal_compression_factor=temporal_compression_factor, | |
| ) | |
| packed_seq.position_ids.append(vision_mrope_ids) | |
| else: | |
| packed_seq.position_ids.extend([curr_rope_id] * vision_split_len) | |
| packed_seq.curr = curr | |
| return vision_split_len | |
| def _pack_action_tokens( | |
| packed_seq: PackedSequence, | |
| input_action_tokens: torch.Tensor, | |
| condition_frame_indexes_action: list[int], | |
| input_timestep: float, | |
| curr_rope_id: int, | |
| action_temporal_offset: int | float = 0, | |
| enable_fps_modulation: bool = False, | |
| base_fps: float = 24.0, | |
| action_fps: float | None = None, | |
| base_temporal_compression_factor: int | None = None, | |
| ) -> int: | |
| """Pack action tokens into the sequence.""" | |
| assert isinstance(packed_seq.position_ids, list), "PackedSequence must be in build mode" | |
| curr = packed_seq.curr | |
| action_split_len = input_action_tokens.shape[0] | |
| if packed_seq.action is None: | |
| packed_seq.action = ModalityData() | |
| assert isinstance(packed_seq.action.sequence_indexes, list) | |
| assert isinstance(packed_seq.action.mse_loss_indexes, list) | |
| assert isinstance(packed_seq.action.timesteps, list) | |
| assert isinstance(packed_seq.action.tokens, list) | |
| action_indexes = list(range(curr, curr + action_split_len)) | |
| packed_seq.action.sequence_indexes.extend(action_indexes) | |
| packed_seq.action.token_shapes.append((action_split_len,)) | |
| packed_seq.action.tokens.append(input_action_tokens) | |
| condition_set = {idx for idx in condition_frame_indexes_action if 0 <= idx < action_split_len} | |
| assert isinstance(packed_seq.action.condition_mask, list) | |
| action_condition_mask = torch.zeros( | |
| (action_split_len, 1), device=input_action_tokens.device, dtype=input_action_tokens.dtype | |
| ) | |
| for frame_idx in condition_set: | |
| action_condition_mask[frame_idx, 0] = 1.0 | |
| packed_seq.action.condition_mask.append(action_condition_mask) | |
| action_noisy_frame_indexes = torch.tensor( | |
| [idx for idx in range(action_split_len) if idx not in condition_set], | |
| device=input_action_tokens.device, | |
| dtype=torch.long, | |
| ) | |
| assert isinstance(packed_seq.action.noisy_frame_indexes, list) | |
| packed_seq.action.noisy_frame_indexes.append(action_noisy_frame_indexes) | |
| frame_token_stride = 1 | |
| for frame_idx in range(action_split_len): | |
| if frame_idx in condition_set: | |
| continue | |
| frame_start = curr + frame_idx * frame_token_stride | |
| frame_end = frame_start + frame_token_stride | |
| packed_seq.action.mse_loss_indexes.extend(range(frame_start, frame_end)) | |
| packed_seq.action.timesteps.extend([input_timestep] * frame_token_stride) | |
| if packed_seq._use_mrope: | |
| effective_fps = action_fps if enable_fps_modulation else None | |
| action_mrope_ids, _ = get_3d_mrope_ids_vae_tokens( | |
| grid_t=action_split_len, | |
| grid_h=1, | |
| grid_w=1, | |
| temporal_offset=action_temporal_offset, | |
| reset_spatial_indices=packed_seq._mrope_reset_spatial, | |
| fps=effective_fps, | |
| base_fps=base_fps, | |
| temporal_compression_factor=1, | |
| base_temporal_compression_factor=base_temporal_compression_factor, | |
| start_frame_offset=1, | |
| ) | |
| packed_seq.position_ids.append(action_mrope_ids) | |
| else: | |
| packed_seq.position_ids.extend([curr_rope_id] * action_split_len) | |
| packed_seq.curr = curr + action_split_len | |
| return action_split_len | |
| def _pack_sound_tokens( | |
| packed_seq: PackedSequence, | |
| input_sound_tokens: torch.Tensor, | |
| condition_frame_indexes_sound: list[int], | |
| input_timestep: float, | |
| curr_rope_id: int, | |
| sound_temporal_offset: int | float = 0, | |
| enable_fps_modulation: bool = False, | |
| base_fps: float = 24.0, | |
| sound_fps: float | None = None, | |
| ) -> int: | |
| """Pack sound/audio tokens into the sequence.""" | |
| assert isinstance(packed_seq.position_ids, list), "PackedSequence must be in build mode" | |
| curr = packed_seq.curr | |
| _, sound_split_len = input_sound_tokens.shape | |
| if packed_seq.sound is None: | |
| packed_seq.sound = ModalityData() | |
| assert isinstance(packed_seq.sound.sequence_indexes, list) | |
| assert isinstance(packed_seq.sound.mse_loss_indexes, list) | |
| assert isinstance(packed_seq.sound.timesteps, list) | |
| assert isinstance(packed_seq.sound.tokens, list) | |
| packed_seq.sound.token_shapes.append((sound_split_len, 1, 1)) | |
| packed_seq.sound.sequence_indexes.extend(range(curr, curr + sound_split_len)) | |
| packed_seq.sound.tokens.append(input_sound_tokens) | |
| condition_set = {idx for idx in condition_frame_indexes_sound if 0 <= idx < sound_split_len} | |
| assert isinstance(packed_seq.sound.condition_mask, list) | |
| sound_condition_mask = torch.zeros( | |
| (sound_split_len, 1), device=input_sound_tokens.device, dtype=input_sound_tokens.dtype | |
| ) | |
| for frame_idx in condition_set: | |
| sound_condition_mask[frame_idx, 0] = 1.0 | |
| packed_seq.sound.condition_mask.append(sound_condition_mask) | |
| sound_noisy_frame_indexes = torch.tensor( | |
| [idx for idx in range(sound_split_len) if idx not in condition_set], | |
| device=input_sound_tokens.device, | |
| dtype=torch.long, | |
| ) | |
| assert isinstance(packed_seq.sound.noisy_frame_indexes, list) | |
| packed_seq.sound.noisy_frame_indexes.append(sound_noisy_frame_indexes) | |
| for frame_idx in range(sound_split_len): | |
| if frame_idx in condition_set: | |
| continue | |
| frame_start = curr + frame_idx | |
| frame_end = frame_start + 1 | |
| packed_seq.sound.mse_loss_indexes.extend(range(frame_start, frame_end)) | |
| packed_seq.sound.timesteps.extend([input_timestep]) | |
| if packed_seq._use_mrope: | |
| effective_fps = sound_fps if enable_fps_modulation else None | |
| sound_mrope_ids, _ = get_3d_mrope_ids_vae_tokens( | |
| grid_t=sound_split_len, | |
| grid_h=1, | |
| grid_w=1, | |
| temporal_offset=sound_temporal_offset, | |
| reset_spatial_indices=packed_seq._mrope_reset_spatial, | |
| fps=effective_fps, | |
| base_fps=base_fps, | |
| temporal_compression_factor=1, | |
| start_frame_offset=0, | |
| ) | |
| packed_seq.position_ids.append(sound_mrope_ids) | |
| else: | |
| packed_seq.position_ids.extend([curr_rope_id] * sound_split_len) | |
| packed_seq.curr = curr + sound_split_len | |
| return sound_split_len | |
| def _pack_supertokens_temporal_causal( | |
| packed_seq: "PackedSequence", | |
| input_vision_tokens: torch.Tensor, | |
| input_action_tokens: torch.Tensor | None, | |
| condition_frame_indexes_vision: list[int], | |
| input_timestep: float | torch.Tensor, | |
| curr_rope_id: int, | |
| latent_patch_size: int, | |
| temporal_compression_factor: int, | |
| action_dim: int, | |
| vision_fps: float | None = None, | |
| action_fps: float | None = None, | |
| enable_fps_modulation: bool = False, | |
| base_fps: float = 24.0, | |
| ) -> tuple[int, bool]: | |
| """Pack vision and action tokens in interleaved supertoken order for temporal causal attention. | |
| Buffer layout: [action_t0, vision_t0, action_t1, vision_t1, ..., action_{T-1}, vision_{T-1}] | |
| """ | |
| assert isinstance(packed_seq.position_ids, list), "PackedSequence must be in build mode" | |
| _, _, latent_t, latent_h, latent_w = input_vision_tokens.shape | |
| patch_h = math.ceil(latent_h / latent_patch_size) | |
| patch_w = math.ceil(latent_w / latent_patch_size) | |
| tcf = temporal_compression_factor | |
| patches_per_frame = patch_h * patch_w | |
| supertoken_len = tcf + patches_per_frame | |
| if packed_seq.vision is None: | |
| packed_seq.vision = ModalityData() | |
| if packed_seq.action is None: | |
| packed_seq.action = ModalityData() | |
| assert isinstance(packed_seq.vision.sequence_indexes, list) | |
| assert isinstance(packed_seq.vision.mse_loss_indexes, list) | |
| assert isinstance(packed_seq.vision.timesteps, list) | |
| assert isinstance(packed_seq.vision.tokens, list) | |
| assert isinstance(packed_seq.vision.condition_mask, list) | |
| assert isinstance(packed_seq.action.sequence_indexes, list) | |
| assert isinstance(packed_seq.action.mse_loss_indexes, list) | |
| assert isinstance(packed_seq.action.timesteps, list) | |
| assert isinstance(packed_seq.action.tokens, list) | |
| assert isinstance(packed_seq.action.condition_mask, list) | |
| device = input_vision_tokens.device | |
| dtype = input_vision_tokens.dtype | |
| null_tokens = torch.zeros(tcf, action_dim, device=device, dtype=dtype) | |
| if input_action_tokens is not None: | |
| if input_action_tokens.dim() == 3: | |
| real_actions = input_action_tokens.squeeze(0) | |
| else: | |
| real_actions = input_action_tokens | |
| if latent_t == 1: | |
| all_action_tokens = real_actions | |
| else: | |
| all_action_tokens = torch.cat([null_tokens, real_actions], dim=0) | |
| else: | |
| all_action_tokens = torch.zeros(latent_t * tcf, action_dim, device=device, dtype=dtype) | |
| null_action_flag = not (latent_t == 1 and input_action_tokens is not None) | |
| packed_seq.vision.token_shapes.append((latent_t, patch_h, patch_w)) | |
| packed_seq.vision.tokens.append(input_vision_tokens) | |
| condition_set_vision = {idx for idx in condition_frame_indexes_vision if 0 <= idx < latent_t} | |
| vision_condition_mask = torch.zeros((latent_t, 1, 1), device=device, dtype=dtype) | |
| for fidx in condition_set_vision: | |
| vision_condition_mask[fidx, 0, 0] = 1.0 | |
| packed_seq.vision.condition_mask.append(vision_condition_mask) | |
| vision_noisy_frame_indexes = torch.tensor( | |
| [idx for idx in range(latent_t) if idx not in condition_set_vision], | |
| device=device, | |
| dtype=torch.long, | |
| ) | |
| packed_seq.vision.noisy_frame_indexes.append(vision_noisy_frame_indexes) | |
| packed_seq.action.token_shapes.append((latent_t * tcf,)) | |
| packed_seq.action.tokens.append(all_action_tokens) | |
| action_condition_mask = torch.ones((latent_t * tcf, 1), device=device, dtype=dtype) | |
| packed_seq.action.condition_mask.append(action_condition_mask) | |
| curr = packed_seq.curr | |
| total_split_len = 0 | |
| if packed_seq._use_mrope: | |
| temporal_offset = packed_seq._mrope_temporal_offset | |
| effective_action_fps = action_fps if enable_fps_modulation else None | |
| effective_vision_fps = vision_fps if enable_fps_modulation else None | |
| fps_active = effective_action_fps is not None | |
| t_dtype = torch.float32 if fps_active else torch.long | |
| t_offset = float(temporal_offset) if fps_active else int(temporal_offset) | |
| null_t = torch.full((tcf,), t_offset, dtype=t_dtype) | |
| null_hw = torch.zeros(tcf, dtype=t_dtype) | |
| null_ids = torch.stack([null_t, null_hw, null_hw]) # [3,tcf] | |
| def _real_action_ids(n_frames: int, start_frame_offset: int) -> torch.Tensor: | |
| flat, _ = get_3d_mrope_ids_vae_tokens( | |
| grid_t=n_frames * tcf, | |
| grid_h=1, | |
| grid_w=1, | |
| temporal_offset=temporal_offset, | |
| reset_spatial_indices=packed_seq._mrope_reset_spatial, | |
| fps=effective_action_fps, | |
| base_fps=base_fps, | |
| temporal_compression_factor=1, | |
| base_temporal_compression_factor=tcf, | |
| start_frame_offset=start_frame_offset, | |
| ) | |
| return flat.reshape(3, n_frames, tcf) | |
| if latent_t > 1: | |
| null_ids_3d = null_ids.reshape(3, 1, tcf) | |
| real_ids_3d = _real_action_ids(latent_t - 1, start_frame_offset=1) | |
| action_ids_3d = torch.cat([null_ids_3d, real_ids_3d], dim=1) | |
| elif input_action_tokens is None: | |
| action_ids_3d = null_ids.reshape(3, 1, tcf) | |
| else: | |
| action_ids_3d = _real_action_ids(1, start_frame_offset=0) | |
| vision_ids_flat, new_offset = get_3d_mrope_ids_vae_tokens( | |
| grid_t=latent_t, | |
| grid_h=patch_h, | |
| grid_w=patch_w, | |
| temporal_offset=temporal_offset, | |
| reset_spatial_indices=packed_seq._mrope_reset_spatial, | |
| fps=effective_vision_fps, | |
| base_fps=base_fps, | |
| temporal_compression_factor=tcf, | |
| ) | |
| vision_ids_3d = vision_ids_flat.reshape(3, latent_t, patches_per_frame) | |
| interleaved_ids = torch.cat([action_ids_3d, vision_ids_3d], dim=2).reshape(3, latent_t * supertoken_len) | |
| packed_seq.position_ids.append(interleaved_ids) | |
| packed_seq._mrope_temporal_offset = new_offset | |
| for frame_t in range(latent_t): | |
| action_indexes = list(range(curr, curr + tcf)) | |
| packed_seq.action.sequence_indexes.extend(action_indexes) | |
| curr += tcf | |
| total_split_len += tcf | |
| if not packed_seq._use_mrope: | |
| packed_seq.position_ids.extend([curr_rope_id] * tcf) | |
| frame_indexes = list(range(curr, curr + patches_per_frame)) | |
| packed_seq.vision.sequence_indexes.extend(frame_indexes) | |
| curr += patches_per_frame | |
| total_split_len += patches_per_frame | |
| if not packed_seq._use_mrope: | |
| packed_seq.position_ids.extend([curr_rope_id] * patches_per_frame) | |
| if frame_t not in condition_set_vision: | |
| packed_seq.vision.mse_loss_indexes.extend(frame_indexes) | |
| frame_ts = input_timestep[frame_t].item() if isinstance(input_timestep, torch.Tensor) else input_timestep | |
| packed_seq.vision.timesteps.extend([frame_ts] * patches_per_frame) | |
| packed_seq.curr = curr | |
| return total_split_len, null_action_flag | |
| # ============================================================================ | |
| # Main packing functions | |
| # ============================================================================ | |
| def pack_input_sequence( | |
| sequence_plans: list[SequencePlan], | |
| input_text_indexes: list[list[int]], | |
| gen_data_clean: GenerationDataClean, | |
| input_timesteps: torch.Tensor, | |
| special_tokens: dict[str, int], | |
| max_num_tokens: int | None = None, | |
| latent_patch_size: int = 1, | |
| skip_text_tokens: bool = False, | |
| include_end_of_generation_token: bool = False, | |
| position_embedding_type: str = "3d_rope", | |
| unified_3d_mrope_reset_spatial_ids: bool = True, | |
| unified_3d_mrope_temporal_modality_margin: int = 0, | |
| enable_fps_modulation: bool = False, | |
| base_fps: float = 24.0, | |
| temporal_compression_factor: int = 4, | |
| video_temporal_causal: bool = False, | |
| action_dim: int = 32, | |
| initial_mrope_temporal_offset: int | float = 0, | |
| ) -> PackedSequence: | |
| """Pack a sequence of input strings and VAE latents into a packed tensor format. | |
| Args: | |
| sequence_plans: List of SequencePlan items describing which modalities are present. | |
| input_text_indexes: List of text token ID sequences. | |
| gen_data_clean: GenerationDataClean containing vision, action, and sound tensors. | |
| input_timesteps: Diffusion timesteps for each sample. Shape (B,) or (B, 1). | |
| special_tokens: Dictionary containing special token IDs. | |
| max_num_tokens: Maximum number of tokens (unused, kept for API compatibility). | |
| latent_patch_size: Patch size used by the network to pack latents. | |
| skip_text_tokens: If True, skip packing text tokens. | |
| include_end_of_generation_token: If True, append end-of-generation token. | |
| position_embedding_type: Position embedding type for vision tokens. | |
| unified_3d_mrope_reset_spatial_ids: If True, spatial indices start from 0 per segment. | |
| unified_3d_mrope_temporal_modality_margin: Temporal margin between text and vision. | |
| enable_fps_modulation: If True, scale temporal position IDs based on video FPS. | |
| base_fps: Base FPS for normalization. | |
| temporal_compression_factor: VAE temporal compression factor. | |
| video_temporal_causal: If True, pack vision and action as interleaved supertokens. | |
| action_dim: Action token dimension for temporal causal packing. | |
| initial_mrope_temporal_offset: Initial temporal offset for AR inference. | |
| Returns: | |
| PackedSequence containing all packed tensors and metadata. | |
| """ | |
| del max_num_tokens | |
| assert special_tokens is not None, "Special tokens must be provided" | |
| assert isinstance(input_timesteps, torch.Tensor), "input_timesteps must be a tensor" | |
| if input_timesteps.is_cuda: | |
| raise ValueError("input_timesteps must be on CPU, not CUDA") | |
| if isinstance(input_text_indexes, torch.Tensor): | |
| raise ValueError("input_text_tokens must be a list, not a tensor") | |
| packed_seq = PackedSequence() | |
| packed_seq._use_mrope = position_embedding_type == "unified_3d_mrope" | |
| packed_seq._mrope_reset_spatial = unified_3d_mrope_reset_spatial_ids | |
| idx_text = 0 | |
| idx_vision = 0 | |
| idx_action = 0 | |
| idx_sound = 0 | |
| null_action_flags: list[bool] = [] | |
| if not skip_text_tokens: | |
| for plan in sequence_plans: | |
| assert plan.has_text, "All sequence plans must have has_text=True when skip_text_tokens=False" | |
| for sample_idx, sequence_plan in enumerate(sequence_plans): | |
| curr_rope_id = 0 | |
| sample_len = 0 | |
| packed_seq._mrope_temporal_offset = initial_mrope_temporal_offset | |
| _ts = input_timesteps[sample_idx] | |
| input_timestep = _ts.item() if _ts.numel() == 1 else _ts | |
| if sequence_plan.has_text and not skip_text_tokens: | |
| text_ids = input_text_indexes[idx_text] | |
| idx_text += 1 | |
| has_generation_for_sample = sequence_plan.has_vision or sequence_plan.has_action or sequence_plan.has_sound | |
| curr_rope_id, _, text_sample_len = _pack_text_tokens( | |
| packed_seq, | |
| text_ids, | |
| special_tokens, | |
| curr_rope_id, | |
| has_generation=has_generation_for_sample, | |
| use_float_positions=enable_fps_modulation, | |
| ) | |
| sample_len += text_sample_len | |
| packed_seq._mrope_temporal_offset += unified_3d_mrope_temporal_modality_margin | |
| vision_start_temporal_offset = packed_seq._mrope_temporal_offset | |
| if video_temporal_causal and sequence_plan.has_vision: | |
| assert position_embedding_type == "unified_3d_mrope", ( | |
| "video_temporal_causal=True requires position_embedding_type='unified_3d_mrope'" | |
| ) | |
| input_vision_tokens = gen_data_clean.x0_tokens_vision[idx_vision] | |
| idx_vision += 1 | |
| vision_fps = None | |
| if ( | |
| enable_fps_modulation | |
| and gen_data_clean.fps_vision is not None | |
| and idx_vision - 1 < len(gen_data_clean.fps_vision) | |
| ): | |
| vision_fps = float(gen_data_clean.fps_vision[idx_vision - 1].item()) | |
| input_action_tokens_tc: torch.Tensor | None = None | |
| action_fps_tc: float | None = None | |
| if sequence_plan.has_action: | |
| input_action_tokens_tc = gen_data_clean.x0_tokens_action[idx_action] | |
| if ( | |
| enable_fps_modulation | |
| and gen_data_clean.fps_action is not None | |
| and idx_action < len(gen_data_clean.fps_action) | |
| ): | |
| action_fps_tc = float(gen_data_clean.fps_action[idx_action].item()) | |
| idx_action += 1 | |
| supertoken_split_len, null_flag = _pack_supertokens_temporal_causal( | |
| packed_seq=packed_seq, | |
| input_vision_tokens=input_vision_tokens, | |
| input_action_tokens=input_action_tokens_tc, | |
| condition_frame_indexes_vision=sequence_plan.condition_frame_indexes_vision, | |
| input_timestep=input_timestep, | |
| curr_rope_id=curr_rope_id, | |
| latent_patch_size=latent_patch_size, | |
| temporal_compression_factor=temporal_compression_factor, | |
| action_dim=action_dim, | |
| vision_fps=vision_fps, | |
| action_fps=action_fps_tc, | |
| enable_fps_modulation=enable_fps_modulation, | |
| base_fps=base_fps, | |
| ) | |
| null_action_flags.append(null_flag) | |
| sample_len += supertoken_split_len | |
| vision_split_len = supertoken_split_len | |
| action_split_len = 0 | |
| else: | |
| if sequence_plan.has_vision: | |
| num_vis = ( | |
| gen_data_clean.num_vision_items_per_sample[sample_idx] | |
| if gen_data_clean.num_vision_items_per_sample is not None | |
| else 1 | |
| ) | |
| vision_split_len = 0 | |
| for item_idx in range(num_vis): | |
| input_vision_tokens = gen_data_clean.x0_tokens_vision[idx_vision] | |
| vision_fps: float | None = None | |
| if ( | |
| enable_fps_modulation | |
| and gen_data_clean.fps_vision is not None | |
| and idx_vision < len(gen_data_clean.fps_vision) | |
| ): | |
| vision_fps = float(gen_data_clean.fps_vision[idx_vision].item()) | |
| idx_vision += 1 | |
| if num_vis > 1 and item_idx < num_vis - 1: | |
| latent_t = input_vision_tokens.shape[2] | |
| item_condition_frames = list(range(latent_t)) | |
| else: | |
| item_condition_frames = sequence_plan.condition_frame_indexes_vision | |
| item_split_len = _pack_vision_tokens( | |
| packed_seq=packed_seq, | |
| input_vision_tokens=input_vision_tokens, | |
| condition_frame_indexes_vision=item_condition_frames, | |
| input_timestep=input_timestep, | |
| curr_rope_id=curr_rope_id, | |
| latent_patch_size=latent_patch_size, | |
| vision_fps=vision_fps, | |
| enable_fps_modulation=enable_fps_modulation, | |
| base_fps=base_fps, | |
| temporal_compression_factor=temporal_compression_factor, | |
| ) | |
| vision_split_len += item_split_len | |
| sample_len += vision_split_len | |
| else: | |
| vision_split_len = 0 | |
| if sequence_plan.has_action: | |
| input_action_tokens = gen_data_clean.x0_tokens_action[idx_action] | |
| action_fps: float | None = None | |
| if ( | |
| enable_fps_modulation | |
| and gen_data_clean.fps_action is not None | |
| and idx_action < len(gen_data_clean.fps_action) | |
| ): | |
| action_fps = float(gen_data_clean.fps_action[idx_action].item()) | |
| idx_action += 1 | |
| action_split_len = _pack_action_tokens( | |
| packed_seq=packed_seq, | |
| input_action_tokens=input_action_tokens, | |
| condition_frame_indexes_action=sequence_plan.condition_frame_indexes_action, | |
| input_timestep=input_timestep, | |
| curr_rope_id=curr_rope_id, | |
| action_temporal_offset=vision_start_temporal_offset, | |
| enable_fps_modulation=enable_fps_modulation, | |
| base_fps=base_fps, | |
| action_fps=action_fps, | |
| base_temporal_compression_factor=temporal_compression_factor, | |
| ) | |
| sample_len += action_split_len | |
| else: | |
| action_split_len = 0 | |
| if sequence_plan.has_sound: | |
| input_sound_tokens = gen_data_clean.x0_tokens_sound[idx_sound] | |
| sound_fps: float | None = None | |
| if ( | |
| enable_fps_modulation | |
| and gen_data_clean.fps_sound is not None | |
| and idx_sound < len(gen_data_clean.fps_sound) | |
| ): | |
| sound_fps = float(gen_data_clean.fps_sound[idx_sound].item()) | |
| idx_sound += 1 | |
| sound_split_len = _pack_sound_tokens( | |
| packed_seq=packed_seq, | |
| input_sound_tokens=input_sound_tokens, | |
| condition_frame_indexes_sound=sequence_plan.condition_frame_indexes_sound, | |
| input_timestep=input_timestep, | |
| curr_rope_id=curr_rope_id, | |
| sound_temporal_offset=vision_start_temporal_offset, | |
| enable_fps_modulation=enable_fps_modulation, | |
| base_fps=base_fps, | |
| sound_fps=sound_fps, | |
| ) | |
| sample_len += sound_split_len | |
| else: | |
| sound_split_len = 0 | |
| eov_len = 0 | |
| has_any_generation = sequence_plan.has_vision or sequence_plan.has_action or sequence_plan.has_sound | |
| if include_end_of_generation_token and has_any_generation: | |
| assert isinstance(packed_seq.text_ids, list) | |
| assert isinstance(packed_seq.text_indexes, list) | |
| assert isinstance(packed_seq.position_ids, list) | |
| packed_seq.text_ids.append(special_tokens["end_of_generation"]) | |
| packed_seq.text_indexes.append(packed_seq.curr) | |
| if packed_seq._use_mrope: | |
| eov_dtype = torch.float32 if enable_fps_modulation else torch.long | |
| eov_mrope_ids = torch.full((3, 1), packed_seq._mrope_temporal_offset, dtype=eov_dtype) | |
| packed_seq.position_ids.append(eov_mrope_ids) # type: ignore[arg-type] | |
| packed_seq._mrope_temporal_offset += 1 | |
| else: | |
| packed_seq.position_ids.append(curr_rope_id) # type: ignore[arg-type] | |
| packed_seq.curr += 1 | |
| eov_len = 1 | |
| sample_len += 1 | |
| combined_split_len = vision_split_len + action_split_len + sound_split_len + eov_len | |
| packed_seq.attn_modes.append("full") | |
| packed_seq.split_lens.append(combined_split_len) | |
| packed_seq.sample_lens.append(sample_len) | |
| if null_action_flags: | |
| assert len(set(null_action_flags)) == 1, ( | |
| f"Inconsistent null_action_supertokens across samples: {null_action_flags}." | |
| ) | |
| packed_seq.null_action_supertokens = null_action_flags[0] | |
| return packed_seq.finalize(gen_data_clean=gen_data_clean) | |
| def build_sequence_plans_from_data_batch( | |
| data_batch: dict, | |
| input_video_key, | |
| input_image_key: str, | |
| ) -> list[SequencePlan]: | |
| """Build or retrieve sequence plans from a data batch dictionary. | |
| This function extracts sequence plans from the data batch if they exist, | |
| otherwise creates default SequencePlan objects for each sample in the batch. | |
| Args: | |
| data_batch: Dictionary containing the data batch from the dataloader. | |
| input_video_key: Key for video tensors in the batch. | |
| input_image_key: Key for image tensors in the batch. | |
| Returns: | |
| List of SequencePlan objects, one per sample in the batch. | |
| """ | |
| # NOTE: this function is ONLY intended for backward compatibility. | |
| # For new modalities, please generate the sequence_plan in the dataset class. | |
| if "sequence_plan" in data_batch: | |
| return data_batch["sequence_plan"] | |
| assert "action" not in data_batch or data_batch["action"] is None, "Action data SHOULD have sequence_plans!" | |
| assert "sound" not in data_batch or data_batch["sound"] is None, "Sound data SHOULD have sequence_plans!" | |
| batch_size = 0 | |
| for key in [input_video_key, input_image_key]: | |
| if key in data_batch: | |
| val = data_batch[key] | |
| if isinstance(val, torch.Tensor): | |
| batch_size = val.shape[0] | |
| break | |
| elif isinstance(val, list): | |
| batch_size = len(val) | |
| break | |
| if batch_size == 0: | |
| raise ValueError( | |
| f"Cannot determine batch size from data_batch. Expected {input_video_key}, {input_image_key}, or similar key." | |
| ) | |
| return [ | |
| SequencePlan( | |
| has_text=True, | |
| has_vision=True, | |
| condition_frame_indexes_vision=[], | |
| ) | |
| for _ in range(batch_size) | |
| ] | |