Skip to content

action_module

Classes

fastvideo.models.dits.matrixgame3.action_module.MatrixGame3ActionModule

MatrixGame3ActionModule(mouse_dim_in: int = 2, keyboard_dim_in: int = 6, hidden_size: int = 128, img_hidden_size: int = 1536, keyboard_hidden_dim: int = 1024, mouse_hidden_dim: int = 1024, vae_time_compression_ratio: int = 4, windows_size: int = 3, heads_num: int = 16, patch_size: list | None = None, qk_norm: bool = True, qkv_bias: bool = False, rope_dim_list: list | None = None, rope_theta=256, mouse_qk_dim_list: list | None = None, enable_mouse=True, enable_keyboard=True, blocks: list | None = None)

Bases: Module

action module from https://arxiv.org/pdf/2501.08325

Source code in fastvideo/models/dits/matrixgame3/action_module.py
def __init__(
    self,
    mouse_dim_in: int = 2,
    keyboard_dim_in: int = 6,
    hidden_size: int = 128,
    img_hidden_size: int = 1536,
    keyboard_hidden_dim: int = 1024,
    mouse_hidden_dim: int = 1024,
    vae_time_compression_ratio: int = 4,
    windows_size: int = 3,
    heads_num: int = 16,
    patch_size: list | None = None,
    qk_norm: bool = True,
    qkv_bias: bool = False,
    rope_dim_list: list | None = None,
    rope_theta=256,
    mouse_qk_dim_list: list | None = None,
    enable_mouse=True,
    enable_keyboard=True,
    blocks: list | None = None,
):
    super().__init__()
    patch_size = patch_size if patch_size is not None else [1, 2, 2]
    rope_dim_list = (
        rope_dim_list if rope_dim_list is not None else [8, 28, 28]
    )
    mouse_qk_dim_list = (
        mouse_qk_dim_list if mouse_qk_dim_list is not None else [8, 28, 28]
    )
    blocks = blocks if blocks is not None else []
    self.enable_mouse = enable_mouse
    self.enable_keyboard = enable_keyboard

    self.rope_dim_list = rope_dim_list
    self.rope_theta = rope_theta
    if self.enable_keyboard:
        self.keyboard_embed = nn.Sequential(
            nn.Linear(keyboard_dim_in, hidden_size, bias=True),
            nn.SiLU(),
            nn.Linear(hidden_size, hidden_size, bias=True),
        )

    self.mouse_qk_dim_list = mouse_qk_dim_list
    self.heads_num = heads_num
    if self.enable_mouse:
        c = mouse_hidden_dim
        self.mouse_mlp = nn.Sequential(
            nn.Linear(
                mouse_dim_in * vae_time_compression_ratio * windows_size
                + img_hidden_size,
                c,
                bias=True,
            ),
            nn.GELU(approximate="tanh"),
            nn.Linear(c, c),
            FP32LayerNorm(c, elementwise_affine=True),
        )

        head_dim = c // heads_num
        self.t_qkv = ReplicatedLinear(c, c * 3, bias=qkv_bias)
        self.img_attn_q_norm = (
            WanRMSNorm(head_dim, eps=1e-6) if qk_norm else nn.Identity()
        )
        self.img_attn_k_norm = (
            WanRMSNorm(head_dim, eps=1e-6) if qk_norm else nn.Identity()
        )
        self.proj_mouse = ReplicatedLinear(
            c, img_hidden_size, bias=qkv_bias
        )

    if self.enable_keyboard:
        head_dim_key = keyboard_hidden_dim // heads_num
        self.key_attn_q_norm = (
            WanRMSNorm(head_dim_key, eps=1e-6) if qk_norm else nn.Identity()
        )
        self.key_attn_k_norm = (
            WanRMSNorm(head_dim_key, eps=1e-6) if qk_norm else nn.Identity()
        )

        self.mouse_attn_q = ReplicatedLinear(
            img_hidden_size, keyboard_hidden_dim, bias=qkv_bias
        )
        self.keyboard_attn_kv = ReplicatedLinear(
            hidden_size * windows_size * vae_time_compression_ratio,
            keyboard_hidden_dim * 2,
            bias=qkv_bias,
        )
        self.proj_keyboard = ReplicatedLinear(
            keyboard_hidden_dim, img_hidden_size, bias=qkv_bias
        )

    self.mouse_attn_layer = (
        LocalAttention(
            num_heads=heads_num,
            head_size=mouse_hidden_dim // heads_num,
            causal=False,
            supported_attention_backends=(
                AttentionBackendEnum.FLASH_ATTN,
                AttentionBackendEnum.TORCH_SDPA,
            ),
        )
        if self.enable_mouse
        else None
    )

    self.keyboard_attn_layer = (
        LocalAttention(
            num_heads=heads_num,
            head_size=keyboard_hidden_dim // heads_num,
            causal=False,
            supported_attention_backends=(
                AttentionBackendEnum.FLASH_ATTN,
                AttentionBackendEnum.TORCH_SDPA,
            ),
        )
        if self.enable_keyboard
        else None
    )

    self.vae_time_compression_ratio = vae_time_compression_ratio
    self.windows_size = windows_size
    self.patch_size = patch_size
    # Lazy initialization: freqs will be created on first forward pass
    self._freqs_cos = None
    self._freqs_sin = None

Methods:

fastvideo.models.dits.matrixgame3.action_module.MatrixGame3ActionModule.forward
forward(x: Tensor, tt: int, th: int, tw: int, mouse_condition: Tensor | None = None, keyboard_condition: Tensor | None = None, mouse_cond_memory: Tensor | None = None, keyboard_cond_memory: Tensor | None = None)

hidden_states: B, ttthtw, C mouse_condition: B, N_frames, C1 keyboard_condition: B, N_frames, C2

Source code in fastvideo/models/dits/matrixgame3/action_module.py
def forward(
    self,
    x: torch.Tensor,
    tt: int,
    th: int,
    tw: int,
    mouse_condition: torch.Tensor | None = None,
    keyboard_condition: torch.Tensor | None = None,
    mouse_cond_memory: torch.Tensor | None = None,
    keyboard_cond_memory: torch.Tensor | None = None,
):
    """
    hidden_states: B, tt*th*tw, C
    mouse_condition: B, N_frames, C1
    keyboard_condition: B, N_frames, C2
    """
    target_device = x.device
    target_dtype = x.dtype
    if mouse_condition is not None:
        mouse_condition = mouse_condition.to(
            device=target_device, dtype=target_dtype
        )
    if keyboard_condition is not None:
        keyboard_condition = keyboard_condition.to(
            device=target_device, dtype=target_dtype
        )
    else:
        return x

    B, N_frames, C = keyboard_condition.shape
    assert tt * th * tw == x.shape[1]
    assert (
        ((N_frames - 1) + self.vae_time_compression_ratio) % self.vae_time_compression_ratio == 0
        or (N_frames % self.vae_time_compression_ratio == 0)
    )
    if ((N_frames - 1) + self.vae_time_compression_ratio) % self.vae_time_compression_ratio == 0:
        N_feats = int((N_frames - 1) / self.vae_time_compression_ratio) + 1
    else:
        N_feats = N_frames // self.vae_time_compression_ratio

    # Lazy initialization of freqs
    if (self._freqs_cos is None or self._freqs_sin is None
            or self._freqs_cos.device != x.device):
        fc, fs = self.get_rotary_pos_embed(
            7500,
            self.patch_size[1],
            self.patch_size[2],
            64,
            self.mouse_qk_dim_list,
        )
        self._freqs_cos = fc.to(x.device)
        self._freqs_sin = fs.to(x.device)

    freqs_cis = (self._freqs_cos, self._freqs_sin)

    if self.enable_mouse and mouse_condition is not None:
        hidden_states = rearrange(
            x, "B (T S) C -> (B S) T C", T=tt, S=th * tw
        )
        B, N_frames, C = mouse_condition.shape
    else:
        hidden_states = x

    pad_t = self.vae_time_compression_ratio * self.windows_size
    if self.enable_mouse and mouse_condition is not None:
        attn = self._forward_mouse(
            hidden_states,
            mouse_condition,
            mouse_cond_memory,
            pad_t=pad_t,
            freqs_cis=freqs_cis,
            N_feats=N_feats,
            B=B,
            C=C,
            th=th,
            tw=tw,
        )
        hidden_states = rearrange(x, "(B S) T C -> B (T S) C", B=B)
        hidden_states = hidden_states + attn

    if self.enable_keyboard and keyboard_condition is not None:
        attn = self._forward_keyboard(
            hidden_states,
            keyboard_condition,
            keyboard_cond_memory,
            pad_t=pad_t,
            freqs_cis=freqs_cis,
            N_feats=N_feats,
            B=B,
            th=th,
            tw=tw,
        )
        hidden_states = hidden_states + attn
    return hidden_states
fastvideo.models.dits.matrixgame3.action_module.MatrixGame3ActionModule.patchify
patchify(x, patch_size)

x : (N C T H W)

Source code in fastvideo/models/dits/matrixgame3/action_module.py
def patchify(self, x, patch_size):
    """
    x : (N C T H W)
    """
    pt, ph, pw = self.patch_size
    t, h, w = x.shape[2] // pt, x.shape[3] // ph, x.shape[4] // pw
    c = x.shape[1]
    x = x.reshape(shape=(x.shape[0], c, t, pt, h, ph, w, pw))
    x = torch.einsum("nctohpwq->nthwcopq", x)
    x = x.reshape(shape=(x.shape[0], t * h * w, c * pt * ph * pw))
    return x
fastvideo.models.dits.matrixgame3.action_module.MatrixGame3ActionModule.unpatchify
unpatchify(x, t, h, w, patch_size)

x: (N, T, patch_size**2 * C) imgs: (N, H, W, C)

Source code in fastvideo/models/dits/matrixgame3/action_module.py
def unpatchify(self, x, t, h, w, patch_size):
    """
    x: (N, T, patch_size**2 * C)
    imgs: (N, H, W, C)
    """
    c = x.shape[2] // patch_size
    pt, ph, pw = self.patch_size
    assert t * h * w == x.shape[1]

    x = x.reshape(shape=(x.shape[0], t, h, w, c, pt, ph, pw))
    x = torch.einsum("nthwcopq->nctohpwq", x)
    imgs = x.reshape(shape=(x.shape[0], c, t * pt, h * ph, w * pw))

    return imgs