# Copyright (c) 2026 SandAI. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import math from dataclasses import dataclass from enum import IntEnum from typing import Any, Literal, Optional, TYPE_CHECKING import torch from einops import rearrange from inference.common import DataProxyConfig, Modality, VarlenHandler from inference.model.dit.dit_module import FFAHandler from torch.nn import functional as F from unfoldNd import UnfoldNd if TYPE_CHECKING: from inference.pipeline.video_generate import EvalInput def calc_local_qk_range(num_video_tokens, num_audio_and_txt_tokens, num_frames, frame_receptive_field): token_per_frame = num_video_tokens // num_frames total_tokens = num_video_tokens + num_audio_and_txt_tokens q_range_list = [] k_range_list = [] for i in range(num_frames): local_q_range = torch.tensor([i * token_per_frame, (i + 1) * token_per_frame]) local_k_range = torch.tensor( [(i - frame_receptive_field) * token_per_frame, (i + frame_receptive_field + 1) * token_per_frame] ) q_range_list.append(local_q_range) k_range_list.append(local_k_range) local_q_range = torch.stack(q_range_list, dim=0) local_k_range = torch.stack(k_range_list, dim=0) local_k_range[local_k_range < 0] = 0 local_k_range[local_k_range > num_video_tokens] = num_video_tokens video_q_range = torch.tensor([[0, num_video_tokens]]) video_k_range = torch.tensor([[num_video_tokens, num_video_tokens + num_audio_and_txt_tokens]]) at_q_ranges = torch.tensor([[num_video_tokens, total_tokens]]) at_k_ranges = torch.tensor([[0, total_tokens]]) q_ranges = torch.cat([local_q_range, video_q_range, at_q_ranges], dim=0).to(torch.int32).to("cuda", non_blocking=True) k_ranges = torch.cat([local_k_range, video_k_range, at_k_ranges], dim=0).to(torch.int32).to("cuda", non_blocking=True) return (q_ranges, k_ranges) def calc_local_attn_ffa_handler(num_video_tokens, num_audio_and_txt_tokens, num_frames, frame_receptive_field): q_ranges, k_ranges = calc_local_qk_range(num_video_tokens, num_audio_and_txt_tokens, num_frames, frame_receptive_field) max_seqlen_q = num_video_tokens + num_audio_and_txt_tokens max_seqlen_k = num_video_tokens + num_audio_and_txt_tokens attn_type_map = torch.zeros([q_ranges.shape[0]], device="cuda", dtype=torch.int32) softmax_scale = None ffa_handler = FFAHandler( q_ranges=q_ranges, k_ranges=k_ranges, max_seqlen_q=max_seqlen_q, max_seqlen_k=max_seqlen_k, attn_type_map=attn_type_map, softmax_scale=softmax_scale, ) return ffa_handler def get_coords( shape: list[int], ref_feat_shape: list[int], offset_thw: list[int] = [0, 0, 0], device: torch.device = torch.device("cpu"), dtype: torch.dtype = torch.float32, ): """ Generate feature-grid coordinates and corresponding original/reference size metadata. Args: feat_shape: [T, H, W] original feature-map shape ref_feat_shape: [T_ref, H_ref, W_ref] reference feature-map shape device: device for coordinate tensors Returns: coords: tensor shape (T*H*W, 9), containing (t, h, w, T, H, W, ref_T, ref_H, ref_W) """ ori_t, ori_h, ori_w = shape ref_t, ref_h, ref_w = ref_feat_shape # Generate index ranges offset_t, offset_h, offset_w = offset_thw time_rng = torch.arange(ori_t, device=device, dtype=dtype) + offset_t height_rng = torch.arange(ori_h, device=device, dtype=dtype) + offset_h width_rng = torch.arange(ori_w, device=device, dtype=dtype) + offset_w # Use meshgrid to generate a 3D grid (T, H, W) time_grid, height_grid, width_grid = torch.meshgrid(time_rng, height_rng, width_rng, indexing="ij") # Stack and flatten coords_grid = torch.stack([time_grid, height_grid, width_grid], dim=-1) coords_flat = coords_grid.reshape(-1, 3) # Build and expand size metadata meta = torch.tensor([ori_t, ori_h, ori_w, ref_t, ref_h, ref_w], device=device, dtype=dtype) meta_expanded = meta.expand(coords_flat.size(0), -1) # Merge and return return torch.cat([coords_flat, meta_expanded], dim=-1) @dataclass class SingleData: video_x_t: torch.Tensor audio_x_t: torch.Tensor audio_feat_len: int txt_feat: torch.Tensor txt_feat_len: int t: int h: int w: int patch_size: int t_patch_size: int spatial_rope_interpolation: Literal["inter", "extra"] ref_audio_offset: int text_offset: int coords_style: Literal["v1", "v2"] = "v1" def __post_init__(self): self.video_token_num = self.video_x_t.shape[0] self.audio_x_t = self.audio_x_t[: self.audio_feat_len] self.txt_feat = self.txt_feat[: self.txt_feat_len] self.video_channel = self.video_x_t.shape[-1] self.audio_channel = self.audio_x_t.shape[-1] self.txt_channel = self.txt_feat.shape[-1] @property def device(self): return self.video_x_t.device @property def default_dtype(self): return self.video_x_t.dtype @property def total_token_num(self): return self.video_token_num + self.audio_feat_len + self.txt_feat_len @property def token_sequence(self): tensors_to_concat = [self.video_x_t, self.audio_x_t, self.txt_feat] max_channel = max(tensor.shape[-1] for tensor in tensors_to_concat) padded_tensors = [F.pad(t, (0, max_channel - t.shape[-1])) for t in tensors_to_concat] ret_val = torch.cat(padded_tensors, dim=0) return ret_val @property def modality_mapping(self): v_map = torch.full((self.video_token_num,), Modality.VIDEO, dtype=torch.int64, device=self.device) a_map = torch.full((self.audio_feat_len,), Modality.AUDIO, dtype=torch.int64, device=self.device) t_map = torch.full((self.txt_feat_len,), Modality.TEXT, dtype=torch.int64, device=self.device) modality_mapping = torch.cat([v_map, a_map, t_map], dim=0) return modality_mapping def default_coords(self, shape, ref_feat_shape, offset_thw=[0, 0, 0]): return get_coords( shape=shape, ref_feat_shape=ref_feat_shape, offset_thw=offset_thw, device=self.device, dtype=self.default_dtype ) @property def coords_mapping(self): if self.spatial_rope_interpolation == "inter": video_ref_feat_shape = (self.t // self.t_patch_size, 32, 32) else: video_ref_feat_shape = (self.t // self.t_patch_size, self.h // self.patch_size, self.w // self.patch_size) video_coords = self.default_coords( shape=(self.t // self.t_patch_size, self.h // self.patch_size, self.w // self.patch_size), ref_feat_shape=video_ref_feat_shape, ) if self.coords_style == "v1": audio_coords = self.default_coords( shape=(self.audio_feat_len, 1, 1), ref_feat_shape=(self.t // self.t_patch_size, 1, 1) ) text_coords = self.default_coords( shape=(self.txt_feat_len, 1, 1), ref_feat_shape=(2, 1, 1), offset_thw=[self.text_offset, 0, 0] ) elif self.coords_style == "v2": magic_audio_ref_t = (self.audio_feat_len - 1) // 4 + 1 audio_coords = self.default_coords( shape=(self.audio_feat_len, 1, 1), ref_feat_shape=(magic_audio_ref_t // self.t_patch_size, 1, 1) ) text_coords = self.default_coords( shape=(self.txt_feat_len, 1, 1), ref_feat_shape=(1, 1, 1), offset_thw=[-self.txt_feat_len, 0, 0] ) coords_mapping = torch.cat([video_coords, audio_coords, text_coords], dim=0) return coords_mapping def depack_token_sequence(self, token_sequence): video_x_t = token_sequence[: self.video_token_num, : self.video_channel] video_x_t = rearrange( video_x_t, "(T H W) (pT pH pW C) -> C (T pT) (H pH) (W pW)", H=self.h // self.patch_size, W=self.w // self.patch_size, pT=self.t_patch_size, pH=self.patch_size, pW=self.patch_size, ).contiguous() audio_x_t = token_sequence[self.video_token_num : self.video_token_num + self.audio_feat_len, : self.audio_channel] return video_x_t, audio_x_t @dataclass class SimplePackedData: items: list[SingleData] @property def token_sequence(self): return torch.cat([item.token_sequence for item in self.items], dim=0) @property def modality_mapping(self): return torch.cat([item.modality_mapping for item in self.items], dim=0) @property def coords_mapping(self): return torch.cat([item.coords_mapping for item in self.items], dim=0) @property def total_token_num(self): return sum([item.total_token_num for item in self.items]) def __getitem__(self, index): return self.items[index] @property def cu_seqlen(self): cu_seqlen = torch.cumsum(torch.tensor([item.total_token_num for item in self.items]), dim=0) cu_seqlen = torch.nn.functional.pad(cu_seqlen, (1, 0)) return cu_seqlen @property def max_seqlen(self): return torch.tensor(max([item.total_token_num for item in self.items])) def depack_token_sequence(self, token_sequence): video_x_t_list = [] audio_x_t_list = [] token_sequence_list = torch.split(token_sequence, [item.total_token_num for item in self.items], dim=0) for item, token_sequence in zip(self.items, token_sequence_list): video_x_t, audio_x_t = item.depack_token_sequence(token_sequence) video_x_t_list.append(video_x_t) audio_x_t_list.append(audio_x_t) return torch.stack(video_x_t_list, dim=0), torch.stack(audio_x_t_list, dim=0) class MagiDataProxy: def __init__(self, config: DataProxyConfig): self.patch_size = config.patch_size self.t_patch_size = config.t_patch_size self.frame_receptive_field = config.frame_receptive_field self.spatial_rope_interpolation = 'extra' self.ref_audio_offset = config.ref_audio_offset self.text_offset = config.text_offset self.unfold = UnfoldNd( kernel_size=(self.t_patch_size, self.patch_size, self.patch_size), stride=(self.t_patch_size, self.patch_size, self.patch_size), ) self.coords_style = config.coords_style self._saved_data: dict[str, Any] = {} def saved_for_output(self, **kwargs): """ Store intermediate data used by process_output. Supports keyword-argument style calls: saved_for_output(a=1, b=2) Can be called multiple times to accumulate data Args: **kwargs: key-value pairs to store """ # Directly update dict; supports accumulation across calls self._saved_data.update(kwargs) def get_saved_data(self, key: str): """ Get stored data """ return self._saved_data[key] def img2tokens(self, x_t: torch.Tensor): x_t_unfolded = self.unfold(x_t) # Transpose dimensions from (N, col_dim, num_tokens) -> (N, num_tokens, col_dim) x_t = rearrange(x_t_unfolded, "N col_dim num_tokens -> N num_tokens col_dim").contiguous() return x_t def process_input(self, transported_data: "EvalInput"): # init img2col module batch_size, _, t, h, w = transported_data.x_t.shape # 1. Process video features while keeping the batch dimension x_t = self.img2tokens(transported_data.x_t) # 2. Process audio features while keeping the batch dimension # Assume transported_data.audio_x_t shape is already (N, num_tokens, col_dim) audio_x_t = transported_data.audio_x_t.contiguous() # Here we assume text_in shape is (N, num_tokens, col_dim) text_in = transported_data.txt_feat.contiguous() simple_packed_data = SimplePackedData(items=[]) for i in range(batch_size): single_data = SingleData( video_x_t=x_t[i], audio_x_t=audio_x_t[i], audio_feat_len=transported_data.audio_feat_len[i], txt_feat=text_in[i], txt_feat_len=transported_data.txt_feat_len[i], t=t, h=h, w=w, patch_size=self.patch_size, t_patch_size=self.t_patch_size, spatial_rope_interpolation=self.spatial_rope_interpolation, ref_audio_offset=self.ref_audio_offset, text_offset=self.text_offset, coords_style=self.coords_style, ) simple_packed_data.items.append(single_data) if self.frame_receptive_field != -1: assert batch_size == 1, "local attention only supports batch size 1" local_attn_handler = calc_local_attn_ffa_handler( num_video_tokens=simple_packed_data[0].video_token_num, num_audio_and_txt_tokens=simple_packed_data[0].audio_feat_len + simple_packed_data[0].txt_feat_len, num_frames=t, frame_receptive_field=self.frame_receptive_field, ) if isinstance(local_attn_handler.max_seqlen_k, torch.Tensor): local_attn_handler.max_seqlen_k = local_attn_handler.max_seqlen_k.item() if isinstance(local_attn_handler.max_seqlen_q, torch.Tensor): local_attn_handler.max_seqlen_q = local_attn_handler.max_seqlen_q.item() else: local_attn_handler = None varlen_handler = VarlenHandler( cu_seqlens_q=simple_packed_data.cu_seqlen.to(torch.int32).cuda(), cu_seqlens_k=simple_packed_data.cu_seqlen.to(torch.int32).cuda(), max_seqlen_q=simple_packed_data.max_seqlen.to(torch.int32).cuda(), max_seqlen_k=simple_packed_data.max_seqlen.to(torch.int32).cuda(), ) self.saved_for_output(simple_packed_data=simple_packed_data) x = simple_packed_data.token_sequence coords_mapping = simple_packed_data.coords_mapping modality_mapping = simple_packed_data.modality_mapping return (x, coords_mapping, modality_mapping, varlen_handler, local_attn_handler) def process_output(self, x: torch.Tensor): # Inserting operations in between may corrupt parallel-runtime data and cause latent errors simple_packed_data: SimplePackedData = self.get_saved_data("simple_packed_data") x_video, x_audio = simple_packed_data.depack_token_sequence(x) return (x_video, x_audio)