Delete trellis/modules

trellis/modules/attention/__init__.py

@@ -1,36 +0,0 @@
-from typing import *
-
-BACKEND = 'flash_attn' 
-DEBUG = False
-
-def __from_env():
-    import os
-    
-    global BACKEND
-    global DEBUG
-    
-    env_attn_backend = os.environ.get('ATTN_BACKEND')
-    env_sttn_debug = os.environ.get('ATTN_DEBUG')
-    
-    if env_attn_backend is not None and env_attn_backend in ['xformers', 'flash_attn', 'sdpa', 'naive']:
-        BACKEND = env_attn_backend
-    if env_sttn_debug is not None:
-        DEBUG = env_sttn_debug == '1'
-
-    print(f"[ATTENTION] Using backend: {BACKEND}")
-        
-
-__from_env()
-    
-
-def set_backend(backend: Literal['xformers', 'flash_attn']):
-    global BACKEND
-    BACKEND = backend
-
-def set_debug(debug: bool):
-    global DEBUG
-    DEBUG = debug
-
-
-from .full_attn import *
-from .modules import *

trellis/modules/attention/full_attn.py

@@ -1,140 +0,0 @@
-from typing import *
-import torch
-import math
-from . import DEBUG, BACKEND
-
-if BACKEND == 'xformers':
-    import xformers.ops as xops
-elif BACKEND == 'flash_attn':
-    import flash_attn
-elif BACKEND == 'sdpa':
-    from torch.nn.functional import scaled_dot_product_attention as sdpa
-elif BACKEND == 'naive':
-    pass
-else:
-    raise ValueError(f"Unknown attention backend: {BACKEND}")
-
-
-__all__ = [
-    'scaled_dot_product_attention',
-]
-
-
-def _naive_sdpa(q, k, v):
-    """
-    Naive implementation of scaled dot product attention.
-    """
-    q = q.permute(0, 2, 1, 3)   # [N, H, L, C]
-    k = k.permute(0, 2, 1, 3)   # [N, H, L, C]
-    v = v.permute(0, 2, 1, 3)   # [N, H, L, C]
-    scale_factor = 1 / math.sqrt(q.size(-1))
-    attn_weight = q @ k.transpose(-2, -1) * scale_factor
-    attn_weight = torch.softmax(attn_weight, dim=-1)
-    out = attn_weight @ v
-    out = out.permute(0, 2, 1, 3)   # [N, L, H, C]
-    return out
-
-
-@overload
-def scaled_dot_product_attention(qkv: torch.Tensor) -> torch.Tensor:
-    """
-    Apply scaled dot product attention.
-
-    Args:
-        qkv (torch.Tensor): A [N, L, 3, H, C] tensor containing Qs, Ks, and Vs.
-    """
-    ...
-
-@overload
-def scaled_dot_product_attention(q: torch.Tensor, kv: torch.Tensor) -> torch.Tensor:
-    """
-    Apply scaled dot product attention.
-
-    Args:
-        q (torch.Tensor): A [N, L, H, C] tensor containing Qs.
-        kv (torch.Tensor): A [N, L, 2, H, C] tensor containing Ks and Vs.
-    """
-    ...
-
-@overload
-def scaled_dot_product_attention(q: torch.Tensor, k: torch.Tensor, v: torch.Tensor) -> torch.Tensor:
-    """
-    Apply scaled dot product attention.
-
-    Args:
-        q (torch.Tensor): A [N, L, H, Ci] tensor containing Qs.
-        k (torch.Tensor): A [N, L, H, Ci] tensor containing Ks.
-        v (torch.Tensor): A [N, L, H, Co] tensor containing Vs.
-
-    Note:
-        k and v are assumed to have the same coordinate map.
-    """
-    ...
-
-def scaled_dot_product_attention(*args, **kwargs):
-    arg_names_dict = {
-        1: ['qkv'],
-        2: ['q', 'kv'],
-        3: ['q', 'k', 'v']
-    }
-    num_all_args = len(args) + len(kwargs)
-    assert num_all_args in arg_names_dict, f"Invalid number of arguments, got {num_all_args}, expected 1, 2, or 3"
-    for key in arg_names_dict[num_all_args][len(args):]:
-        assert key in kwargs, f"Missing argument {key}"
-
-    if num_all_args == 1:
-        qkv = args[0] if len(args) > 0 else kwargs['qkv']
-        assert len(qkv.shape) == 5 and qkv.shape[2] == 3, f"Invalid shape for qkv, got {qkv.shape}, expected [N, L, 3, H, C]"
-        device = qkv.device
-
-    elif num_all_args == 2:
-        q = args[0] if len(args) > 0 else kwargs['q']
-        kv = args[1] if len(args) > 1 else kwargs['kv']
-        assert q.shape[0] == kv.shape[0], f"Batch size mismatch, got {q.shape[0]} and {kv.shape[0]}"
-        assert len(q.shape) == 4, f"Invalid shape for q, got {q.shape}, expected [N, L, H, C]"
-        assert len(kv.shape) == 5, f"Invalid shape for kv, got {kv.shape}, expected [N, L, 2, H, C]"
-        device = q.device
-
-    elif num_all_args == 3:
-        q = args[0] if len(args) > 0 else kwargs['q']
-        k = args[1] if len(args) > 1 else kwargs['k']
-        v = args[2] if len(args) > 2 else kwargs['v']
-        assert q.shape[0] == k.shape[0] == v.shape[0], f"Batch size mismatch, got {q.shape[0]}, {k.shape[0]}, and {v.shape[0]}"
-        assert len(q.shape) == 4, f"Invalid shape for q, got {q.shape}, expected [N, L, H, Ci]"
-        assert len(k.shape) == 4, f"Invalid shape for k, got {k.shape}, expected [N, L, H, Ci]"
-        assert len(v.shape) == 4, f"Invalid shape for v, got {v.shape}, expected [N, L, H, Co]"
-        device = q.device    
-
-    if BACKEND == 'xformers':
-        if num_all_args == 1:
-            q, k, v = qkv.unbind(dim=2)
-        elif num_all_args == 2:
-            k, v = kv.unbind(dim=2)
-        out = xops.memory_efficient_attention(q, k, v)
-    elif BACKEND == 'flash_attn':
-        if num_all_args == 1:
-            out = flash_attn.flash_attn_qkvpacked_func(qkv)
-        elif num_all_args == 2:
-            out = flash_attn.flash_attn_kvpacked_func(q, kv)
-        elif num_all_args == 3:
-            out = flash_attn.flash_attn_func(q, k, v)
-    elif BACKEND == 'sdpa':
-        if num_all_args == 1:
-            q, k, v = qkv.unbind(dim=2)
-        elif num_all_args == 2:
-            k, v = kv.unbind(dim=2)
-        q = q.permute(0, 2, 1, 3)   # [N, H, L, C]
-        k = k.permute(0, 2, 1, 3)   # [N, H, L, C]
-        v = v.permute(0, 2, 1, 3)   # [N, H, L, C]
-        out = sdpa(q, k, v)         # [N, H, L, C]
-        out = out.permute(0, 2, 1, 3)   # [N, L, H, C]
-    elif BACKEND == 'naive':
-        if num_all_args == 1:
-            q, k, v = qkv.unbind(dim=2)
-        elif num_all_args == 2:
-            k, v = kv.unbind(dim=2)
-        out = _naive_sdpa(q, k, v)
-    else:
-        raise ValueError(f"Unknown attention module: {BACKEND}")
-    
-    return out

trellis/modules/attention/modules.py

@@ -1,146 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from .full_attn import scaled_dot_product_attention
-
-
-class MultiHeadRMSNorm(nn.Module):
-    def __init__(self, dim: int, heads: int):
-        super().__init__()
-        self.scale = dim ** 0.5
-        self.gamma = nn.Parameter(torch.ones(heads, dim))
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return (F.normalize(x.float(), dim = -1) * self.gamma * self.scale).to(x.dtype)
-
-
-class RotaryPositionEmbedder(nn.Module):
-    def __init__(self, hidden_size: int, in_channels: int = 3):
-        super().__init__()
-        assert hidden_size % 2 == 0, "Hidden size must be divisible by 2"
-        self.hidden_size = hidden_size
-        self.in_channels = in_channels
-        self.freq_dim = hidden_size // in_channels // 2
-        self.freqs = torch.arange(self.freq_dim, dtype=torch.float32) / self.freq_dim
-        self.freqs = 1.0 / (10000 ** self.freqs)
-        
-    def _get_phases(self, indices: torch.Tensor) -> torch.Tensor:
-        self.freqs = self.freqs.to(indices.device)
-        phases = torch.outer(indices, self.freqs)
-        phases = torch.polar(torch.ones_like(phases), phases)
-        return phases
-        
-    def _rotary_embedding(self, x: torch.Tensor, phases: torch.Tensor) -> torch.Tensor:
-        x_complex = torch.view_as_complex(x.float().reshape(*x.shape[:-1], -1, 2))
-        x_rotated = x_complex * phases
-        x_embed = torch.view_as_real(x_rotated).reshape(*x_rotated.shape[:-1], -1).to(x.dtype)
-        return x_embed
-        
-    def forward(self, q: torch.Tensor, k: torch.Tensor, indices: Optional[torch.Tensor] = None) -> Tuple[torch.Tensor, torch.Tensor]:
-        """
-        Args:
-            q (sp.SparseTensor): [..., N, D] tensor of queries
-            k (sp.SparseTensor): [..., N, D] tensor of keys
-            indices (torch.Tensor): [..., N, C] tensor of spatial positions
-        """
-        if indices is None:
-            indices = torch.arange(q.shape[-2], device=q.device)
-            if len(q.shape) > 2:
-                indices = indices.unsqueeze(0).expand(q.shape[:-2] + (-1,))
-        
-        phases = self._get_phases(indices.reshape(-1)).reshape(*indices.shape[:-1], -1)
-        if phases.shape[1] < self.hidden_size // 2:
-            phases = torch.cat([phases, torch.polar(
-                torch.ones(*phases.shape[:-1], self.hidden_size // 2 - phases.shape[1], device=phases.device),
-                torch.zeros(*phases.shape[:-1], self.hidden_size // 2 - phases.shape[1], device=phases.device)
-            )], dim=-1)
-        q_embed = self._rotary_embedding(q, phases)
-        k_embed = self._rotary_embedding(k, phases)
-        return q_embed, k_embed
-    
-
-class MultiHeadAttention(nn.Module):
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        ctx_channels: Optional[int]=None,
-        type: Literal["self", "cross"] = "self",
-        attn_mode: Literal["full", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        qkv_bias: bool = True,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-    ):
-        super().__init__()
-        assert channels % num_heads == 0
-        assert type in ["self", "cross"], f"Invalid attention type: {type}"
-        assert attn_mode in ["full", "windowed"], f"Invalid attention mode: {attn_mode}"
-        assert type == "self" or attn_mode == "full", "Cross-attention only supports full attention"
-        
-        if attn_mode == "windowed":
-            raise NotImplementedError("Windowed attention is not yet implemented")
-        
-        self.channels = channels
-        self.head_dim = channels // num_heads
-        self.ctx_channels = ctx_channels if ctx_channels is not None else channels
-        self.num_heads = num_heads
-        self._type = type
-        self.attn_mode = attn_mode
-        self.window_size = window_size
-        self.shift_window = shift_window
-        self.use_rope = use_rope
-        self.qk_rms_norm = qk_rms_norm
-
-        if self._type == "self":
-            self.to_qkv = nn.Linear(channels, channels * 3, bias=qkv_bias)
-        else:
-            self.to_q = nn.Linear(channels, channels, bias=qkv_bias)
-            self.to_kv = nn.Linear(self.ctx_channels, channels * 2, bias=qkv_bias)
-            
-        if self.qk_rms_norm:
-            self.q_rms_norm = MultiHeadRMSNorm(self.head_dim, num_heads)
-            self.k_rms_norm = MultiHeadRMSNorm(self.head_dim, num_heads)
-            
-        self.to_out = nn.Linear(channels, channels)
-
-        if use_rope:
-            self.rope = RotaryPositionEmbedder(channels)
-    
-    def forward(self, x: torch.Tensor, context: Optional[torch.Tensor] = None, indices: Optional[torch.Tensor] = None) -> torch.Tensor:
-        B, L, C = x.shape
-        if self._type == "self":
-            qkv = self.to_qkv(x)
-            qkv = qkv.reshape(B, L, 3, self.num_heads, -1)
-            if self.use_rope:
-                q, k, v = qkv.unbind(dim=2)
-                q, k = self.rope(q, k, indices)
-                qkv = torch.stack([q, k, v], dim=2)
-            if self.attn_mode == "full":
-                if self.qk_rms_norm:
-                    q, k, v = qkv.unbind(dim=2)
-                    q = self.q_rms_norm(q)
-                    k = self.k_rms_norm(k)
-                    h = scaled_dot_product_attention(q, k, v)
-                else:
-                    h = scaled_dot_product_attention(qkv)
-            elif self.attn_mode == "windowed":
-                raise NotImplementedError("Windowed attention is not yet implemented")
-        else:
-            Lkv = context.shape[1]
-            q = self.to_q(x)
-            kv = self.to_kv(context)
-            q = q.reshape(B, L, self.num_heads, -1)
-            kv = kv.reshape(B, Lkv, 2, self.num_heads, -1)
-            if self.qk_rms_norm:
-                q = self.q_rms_norm(q)
-                k, v = kv.unbind(dim=2)
-                k = self.k_rms_norm(k)
-                h = scaled_dot_product_attention(q, k, v)
-            else:
-                h = scaled_dot_product_attention(q, kv)
-        h = h.reshape(B, L, -1)
-        h = self.to_out(h)
-        return h

trellis/modules/norm.py

@@ -1,25 +0,0 @@
-import torch
-import torch.nn as nn
-
-
-class LayerNorm32(nn.LayerNorm):
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return super().forward(x.float()).type(x.dtype)
-    
-
-class GroupNorm32(nn.GroupNorm):
-    """
-    A GroupNorm layer that converts to float32 before the forward pass.
-    """
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return super().forward(x.float()).type(x.dtype)
-    
-    
-class ChannelLayerNorm32(LayerNorm32):
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        DIM = x.dim()
-        x = x.permute(0, *range(2, DIM), 1).contiguous()
-        x = super().forward(x)
-        x = x.permute(0, DIM-1, *range(1, DIM-1)).contiguous()
-        return x
-    

trellis/modules/sparse/__init__.py

@@ -1,102 +0,0 @@
-from typing import *
-
-BACKEND = 'spconv' 
-DEBUG = False
-ATTN = 'flash_attn'
-
-def __from_env():
-    import os
-    
-    global BACKEND
-    global DEBUG
-    global ATTN
-    
-    env_sparse_backend = os.environ.get('SPARSE_BACKEND')
-    env_sparse_debug = os.environ.get('SPARSE_DEBUG')
-    env_sparse_attn = os.environ.get('SPARSE_ATTN_BACKEND')
-    if env_sparse_attn is None:
-        env_sparse_attn = os.environ.get('ATTN_BACKEND')
-
-    if env_sparse_backend is not None and env_sparse_backend in ['spconv', 'torchsparse']:
-        BACKEND = env_sparse_backend
-    if env_sparse_debug is not None:
-        DEBUG = env_sparse_debug == '1'
-    if env_sparse_attn is not None and env_sparse_attn in ['xformers', 'flash_attn']:
-        ATTN = env_sparse_attn
-        
-    print(f"[SPARSE] Backend: {BACKEND}, Attention: {ATTN}")
-        
-
-__from_env()
-    
-
-def set_backend(backend: Literal['spconv', 'torchsparse']):
-    global BACKEND
-    BACKEND = backend
-
-def set_debug(debug: bool):
-    global DEBUG
-    DEBUG = debug
-
-def set_attn(attn: Literal['xformers', 'flash_attn']):
-    global ATTN
-    ATTN = attn
-    
-    
-import importlib
-
-__attributes = {
-    'SparseTensor': 'basic',
-    'sparse_batch_broadcast': 'basic',
-    'sparse_batch_op': 'basic',
-    'sparse_cat': 'basic',
-    'sparse_unbind': 'basic',
-    'SparseGroupNorm': 'norm',
-    'SparseLayerNorm': 'norm',
-    'SparseGroupNorm32': 'norm',
-    'SparseLayerNorm32': 'norm',
-    'SparseReLU': 'nonlinearity',
-    'SparseSiLU': 'nonlinearity',
-    'SparseGELU': 'nonlinearity',
-    'SparseActivation': 'nonlinearity',
-    'SparseLinear': 'linear',
-    'sparse_scaled_dot_product_attention': 'attention',
-    'SerializeMode': 'attention',
-    'sparse_serialized_scaled_dot_product_self_attention': 'attention',
-    'sparse_windowed_scaled_dot_product_self_attention': 'attention',
-    'SparseMultiHeadAttention': 'attention',
-    'SparseConv3d': 'conv',
-    'SparseInverseConv3d': 'conv',
-    'SparseDownsample': 'spatial',
-    'SparseUpsample': 'spatial',
-    'SparseSubdivide' : 'spatial'
-}
-
-__submodules = ['transformer']
-
-__all__ = list(__attributes.keys()) + __submodules
-
-def __getattr__(name):
-    if name not in globals():
-        if name in __attributes:
-            module_name = __attributes[name]
-            module = importlib.import_module(f".{module_name}", __name__)
-            globals()[name] = getattr(module, name)
-        elif name in __submodules:
-            module = importlib.import_module(f".{name}", __name__)
-            globals()[name] = module
-        else:
-            raise AttributeError(f"module {__name__} has no attribute {name}")
-    return globals()[name]
-
-
-# For Pylance
-if __name__ == '__main__':
-    from .basic import *
-    from .norm import *
-    from .nonlinearity import *
-    from .linear import *
-    from .attention import *
-    from .conv import *
-    from .spatial import *
-    import transformer

trellis/modules/sparse/attention/__init__.py

@@ -1,4 +0,0 @@
-from .full_attn import *
-from .serialized_attn import *
-from .windowed_attn import *
-from .modules import *

trellis/modules/sparse/attention/full_attn.py

@@ -1,215 +0,0 @@
-from typing import *
-import torch
-from .. import SparseTensor
-from .. import DEBUG, ATTN
-
-if ATTN == 'xformers':
-    import xformers.ops as xops
-elif ATTN == 'flash_attn':
-    import flash_attn
-else:
-    raise ValueError(f"Unknown attention module: {ATTN}")
-
-
-__all__ = [
-    'sparse_scaled_dot_product_attention',
-]
-
-
-@overload
-def sparse_scaled_dot_product_attention(qkv: SparseTensor) -> SparseTensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        qkv (SparseTensor): A [N, *, 3, H, C] sparse tensor containing Qs, Ks, and Vs.
-    """
-    ...
-
-@overload
-def sparse_scaled_dot_product_attention(q: SparseTensor, kv: Union[SparseTensor, torch.Tensor]) -> SparseTensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        q (SparseTensor): A [N, *, H, C] sparse tensor containing Qs.
-        kv (SparseTensor or torch.Tensor): A [N, *, 2, H, C] sparse tensor or a [N, L, 2, H, C] dense tensor containing Ks and Vs.
-    """
-    ...
-
-@overload
-def sparse_scaled_dot_product_attention(q: torch.Tensor, kv: SparseTensor) -> torch.Tensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        q (SparseTensor): A [N, L, H, C] dense tensor containing Qs.
-        kv (SparseTensor or torch.Tensor): A [N, *, 2, H, C] sparse tensor containing Ks and Vs.
-    """
-    ...
-
-@overload
-def sparse_scaled_dot_product_attention(q: SparseTensor, k: SparseTensor, v: SparseTensor) -> SparseTensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        q (SparseTensor): A [N, *, H, Ci] sparse tensor containing Qs.
-        k (SparseTensor): A [N, *, H, Ci] sparse tensor containing Ks.
-        v (SparseTensor): A [N, *, H, Co] sparse tensor containing Vs.
-
-    Note:
-        k and v are assumed to have the same coordinate map.
-    """
-    ...
-
-@overload
-def sparse_scaled_dot_product_attention(q: SparseTensor, k: torch.Tensor, v: torch.Tensor) -> SparseTensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        q (SparseTensor): A [N, *, H, Ci] sparse tensor containing Qs.
-        k (torch.Tensor): A [N, L, H, Ci] dense tensor containing Ks.
-        v (torch.Tensor): A [N, L, H, Co] dense tensor containing Vs.
-    """
-    ...
-
-@overload
-def sparse_scaled_dot_product_attention(q: torch.Tensor, k: SparseTensor, v: SparseTensor) -> torch.Tensor:
-    """
-    Apply scaled dot product attention to a sparse tensor.
-
-    Args:
-        q (torch.Tensor): A [N, L, H, Ci] dense tensor containing Qs.
-        k (SparseTensor): A [N, *, H, Ci] sparse tensor containing Ks.
-        v (SparseTensor): A [N, *, H, Co] sparse tensor containing Vs.
-    """
-    ...
-
-def sparse_scaled_dot_product_attention(*args, **kwargs):
-    arg_names_dict = {
-        1: ['qkv'],
-        2: ['q', 'kv'],
-        3: ['q', 'k', 'v']
-    }
-    num_all_args = len(args) + len(kwargs)
-    assert num_all_args in arg_names_dict, f"Invalid number of arguments, got {num_all_args}, expected 1, 2, or 3"
-    for key in arg_names_dict[num_all_args][len(args):]:
-        assert key in kwargs, f"Missing argument {key}"
-
-    if num_all_args == 1:
-        qkv = args[0] if len(args) > 0 else kwargs['qkv']
-        assert isinstance(qkv, SparseTensor), f"qkv must be a SparseTensor, got {type(qkv)}"
-        assert len(qkv.shape) == 4 and qkv.shape[1] == 3, f"Invalid shape for qkv, got {qkv.shape}, expected [N, *, 3, H, C]"
-        device = qkv.device
-
-        s = qkv
-        q_seqlen = [qkv.layout[i].stop - qkv.layout[i].start for i in range(qkv.shape[0])]
-        kv_seqlen = q_seqlen
-        qkv = qkv.feats     # [T, 3, H, C]
-
-    elif num_all_args == 2:
-        q = args[0] if len(args) > 0 else kwargs['q']
-        kv = args[1] if len(args) > 1 else kwargs['kv']
-        assert isinstance(q, SparseTensor) and isinstance(kv, (SparseTensor, torch.Tensor)) or \
-               isinstance(q, torch.Tensor) and isinstance(kv, SparseTensor), \
-               f"Invalid types, got {type(q)} and {type(kv)}"
-        assert q.shape[0] == kv.shape[0], f"Batch size mismatch, got {q.shape[0]} and {kv.shape[0]}"
-        device = q.device
-
-        if isinstance(q, SparseTensor):
-            assert len(q.shape) == 3, f"Invalid shape for q, got {q.shape}, expected [N, *, H, C]"
-            s = q
-            q_seqlen = [q.layout[i].stop - q.layout[i].start for i in range(q.shape[0])]
-            q = q.feats     # [T_Q, H, C]
-        else:
-            assert len(q.shape) == 4, f"Invalid shape for q, got {q.shape}, expected [N, L, H, C]"
-            s = None
-            N, L, H, C = q.shape
-            q_seqlen = [L] * N
-            q = q.reshape(N * L, H, C)   # [T_Q, H, C]
-
-        if isinstance(kv, SparseTensor):
-            assert len(kv.shape) == 4 and kv.shape[1] == 2, f"Invalid shape for kv, got {kv.shape}, expected [N, *, 2, H, C]"
-            kv_seqlen = [kv.layout[i].stop - kv.layout[i].start for i in range(kv.shape[0])]
-            kv = kv.feats     # [T_KV, 2, H, C]
-        else:
-            assert len(kv.shape) == 5, f"Invalid shape for kv, got {kv.shape}, expected [N, L, 2, H, C]"
-            N, L, _, H, C = kv.shape
-            kv_seqlen = [L] * N
-            kv = kv.reshape(N * L, 2, H, C)   # [T_KV, 2, H, C]
-
-    elif num_all_args == 3:
-        q = args[0] if len(args) > 0 else kwargs['q']
-        k = args[1] if len(args) > 1 else kwargs['k']
-        v = args[2] if len(args) > 2 else kwargs['v']
-        assert isinstance(q, SparseTensor) and isinstance(k, (SparseTensor, torch.Tensor)) and type(k) == type(v) or \
-               isinstance(q, torch.Tensor) and isinstance(k, SparseTensor) and isinstance(v, SparseTensor), \
-               f"Invalid types, got {type(q)}, {type(k)}, and {type(v)}"
-        assert q.shape[0] == k.shape[0] == v.shape[0], f"Batch size mismatch, got {q.shape[0]}, {k.shape[0]}, and {v.shape[0]}"
-        device = q.device
-
-        if isinstance(q, SparseTensor):
-            assert len(q.shape) == 3, f"Invalid shape for q, got {q.shape}, expected [N, *, H, Ci]"
-            s = q
-            q_seqlen = [q.layout[i].stop - q.layout[i].start for i in range(q.shape[0])]
-            q = q.feats     # [T_Q, H, Ci]
-        else:
-            assert len(q.shape) == 4, f"Invalid shape for q, got {q.shape}, expected [N, L, H, Ci]"
-            s = None
-            N, L, H, CI = q.shape
-            q_seqlen = [L] * N
-            q = q.reshape(N * L, H, CI)  # [T_Q, H, Ci]
-
-        if isinstance(k, SparseTensor):
-            assert len(k.shape) == 3, f"Invalid shape for k, got {k.shape}, expected [N, *, H, Ci]"
-            assert len(v.shape) == 3, f"Invalid shape for v, got {v.shape}, expected [N, *, H, Co]"
-            kv_seqlen = [k.layout[i].stop - k.layout[i].start for i in range(k.shape[0])]
-            k = k.feats     # [T_KV, H, Ci]
-            v = v.feats     # [T_KV, H, Co]
-        else:
-            assert len(k.shape) == 4, f"Invalid shape for k, got {k.shape}, expected [N, L, H, Ci]"
-            assert len(v.shape) == 4, f"Invalid shape for v, got {v.shape}, expected [N, L, H, Co]"
-            N, L, H, CI, CO = *k.shape, v.shape[-1]
-            kv_seqlen = [L] * N
-            k = k.reshape(N * L, H, CI)     # [T_KV, H, Ci]
-            v = v.reshape(N * L, H, CO)     # [T_KV, H, Co]
-
-    if DEBUG:
-        if s is not None:
-            for i in range(s.shape[0]):
-                assert (s.coords[s.layout[i]] == i).all(), f"SparseScaledDotProductSelfAttention: batch index mismatch"
-        if num_all_args in [2, 3]:
-            assert q.shape[:2] == [1, sum(q_seqlen)], f"SparseScaledDotProductSelfAttention: q shape mismatch"
-        if num_all_args == 3:
-            assert k.shape[:2] == [1, sum(kv_seqlen)], f"SparseScaledDotProductSelfAttention: k shape mismatch"
-            assert v.shape[:2] == [1, sum(kv_seqlen)], f"SparseScaledDotProductSelfAttention: v shape mismatch"
-
-    if ATTN == 'xformers':
-        if num_all_args == 1:
-            q, k, v = qkv.unbind(dim=1)
-        elif num_all_args == 2:
-            k, v = kv.unbind(dim=1)
-        q = q.unsqueeze(0)
-        k = k.unsqueeze(0)
-        v = v.unsqueeze(0)
-        mask = xops.fmha.BlockDiagonalMask.from_seqlens(q_seqlen, kv_seqlen)
-        out = xops.memory_efficient_attention(q, k, v, mask)[0]
-    elif ATTN == 'flash_attn':
-        cu_seqlens_q = torch.cat([torch.tensor([0]), torch.cumsum(torch.tensor(q_seqlen), dim=0)]).int().to(device)
-        if num_all_args in [2, 3]:
-            cu_seqlens_kv = torch.cat([torch.tensor([0]), torch.cumsum(torch.tensor(kv_seqlen), dim=0)]).int().to(device)
-        if num_all_args == 1:
-            out = flash_attn.flash_attn_varlen_qkvpacked_func(qkv, cu_seqlens_q, max(q_seqlen))
-        elif num_all_args == 2:
-            out = flash_attn.flash_attn_varlen_kvpacked_func(q, kv, cu_seqlens_q, cu_seqlens_kv, max(q_seqlen), max(kv_seqlen))
-        elif num_all_args == 3:
-            out = flash_attn.flash_attn_varlen_func(q, k, v, cu_seqlens_q, cu_seqlens_kv, max(q_seqlen), max(kv_seqlen))
-    else:
-        raise ValueError(f"Unknown attention module: {ATTN}")
-    
-    if s is not None:
-        return s.replace(out)
-    else:
-        return out.reshape(N, L, H, -1)

trellis/modules/sparse/attention/modules.py

@@ -1,139 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from .. import SparseTensor
-from .full_attn import sparse_scaled_dot_product_attention
-from .serialized_attn import SerializeMode, sparse_serialized_scaled_dot_product_self_attention
-from .windowed_attn import sparse_windowed_scaled_dot_product_self_attention
-from ...attention import RotaryPositionEmbedder
-
-
-class SparseMultiHeadRMSNorm(nn.Module):
-    def __init__(self, dim: int, heads: int):
-        super().__init__()
-        self.scale = dim ** 0.5
-        self.gamma = nn.Parameter(torch.ones(heads, dim))
-
-    def forward(self, x: Union[SparseTensor, torch.Tensor]) -> Union[SparseTensor, torch.Tensor]:
-        x_type = x.dtype
-        x = x.float()
-        if isinstance(x, SparseTensor):
-            x = x.replace(F.normalize(x.feats, dim=-1))
-        else:
-            x = F.normalize(x, dim=-1)            
-        return (x * self.gamma * self.scale).to(x_type)
-
-
-class SparseMultiHeadAttention(nn.Module):
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        ctx_channels: Optional[int] = None,
-        type: Literal["self", "cross"] = "self",
-        attn_mode: Literal["full", "serialized", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_sequence: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        serialize_mode: Optional[SerializeMode] = None,
-        qkv_bias: bool = True,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-    ):
-        super().__init__()
-        assert channels % num_heads == 0
-        assert type in ["self", "cross"], f"Invalid attention type: {type}"
-        assert attn_mode in ["full", "serialized", "windowed"], f"Invalid attention mode: {attn_mode}"
-        assert type == "self" or attn_mode == "full", "Cross-attention only supports full attention"
-        assert type == "self" or use_rope is False, "Rotary position embeddings only supported for self-attention"
-        self.channels = channels
-        self.ctx_channels = ctx_channels if ctx_channels is not None else channels
-        self.num_heads = num_heads
-        self._type = type
-        self.attn_mode = attn_mode
-        self.window_size = window_size
-        self.shift_sequence = shift_sequence
-        self.shift_window = shift_window
-        self.serialize_mode = serialize_mode
-        self.use_rope = use_rope
-        self.qk_rms_norm = qk_rms_norm
-
-        if self._type == "self":
-            self.to_qkv = nn.Linear(channels, channels * 3, bias=qkv_bias)
-        else:
-            self.to_q = nn.Linear(channels, channels, bias=qkv_bias)
-            self.to_kv = nn.Linear(self.ctx_channels, channels * 2, bias=qkv_bias)
-        
-        if self.qk_rms_norm:
-            self.q_rms_norm = SparseMultiHeadRMSNorm(channels // num_heads, num_heads)
-            self.k_rms_norm = SparseMultiHeadRMSNorm(channels // num_heads, num_heads)
-            
-        self.to_out = nn.Linear(channels, channels)
-
-        if use_rope:
-            self.rope = RotaryPositionEmbedder(channels)
-
-    @staticmethod
-    def _linear(module: nn.Linear, x: Union[SparseTensor, torch.Tensor]) -> Union[SparseTensor, torch.Tensor]:
-        if isinstance(x, SparseTensor):
-            return x.replace(module(x.feats))
-        else:
-            return module(x)
-
-    @staticmethod
-    def _reshape_chs(x: Union[SparseTensor, torch.Tensor], shape: Tuple[int, ...]) -> Union[SparseTensor, torch.Tensor]:
-        if isinstance(x, SparseTensor):
-            return x.reshape(*shape)
-        else:
-            return x.reshape(*x.shape[:2], *shape)
-
-    def _fused_pre(self, x: Union[SparseTensor, torch.Tensor], num_fused: int) -> Union[SparseTensor, torch.Tensor]:
-        if isinstance(x, SparseTensor):
-            x_feats = x.feats.unsqueeze(0)
-        else:
-            x_feats = x
-        x_feats = x_feats.reshape(*x_feats.shape[:2], num_fused, self.num_heads, -1)
-        return x.replace(x_feats.squeeze(0)) if isinstance(x, SparseTensor) else x_feats
-
-    def _rope(self, qkv: SparseTensor) -> SparseTensor:
-        q, k, v = qkv.feats.unbind(dim=1)   # [T, H, C]
-        q, k = self.rope(q, k, qkv.coords[:, 1:])
-        qkv = qkv.replace(torch.stack([q, k, v], dim=1)) 
-        return qkv
-    
-    def forward(self, x: Union[SparseTensor, torch.Tensor], context: Optional[Union[SparseTensor, torch.Tensor]] = None) -> Union[SparseTensor, torch.Tensor]:
-        if self._type == "self":
-            qkv = self._linear(self.to_qkv, x)
-            qkv = self._fused_pre(qkv, num_fused=3)
-            if self.use_rope:
-                qkv = self._rope(qkv)
-            if self.qk_rms_norm:
-                q, k, v = qkv.unbind(dim=1)
-                q = self.q_rms_norm(q)
-                k = self.k_rms_norm(k)
-                qkv = qkv.replace(torch.stack([q.feats, k.feats, v.feats], dim=1))
-            if self.attn_mode == "full":
-                h = sparse_scaled_dot_product_attention(qkv)
-            elif self.attn_mode == "serialized":
-                h = sparse_serialized_scaled_dot_product_self_attention(
-                    qkv, self.window_size, serialize_mode=self.serialize_mode, shift_sequence=self.shift_sequence, shift_window=self.shift_window
-                )
-            elif self.attn_mode == "windowed":
-                h = sparse_windowed_scaled_dot_product_self_attention(
-                    qkv, self.window_size, shift_window=self.shift_window
-                )
-        else:
-            q = self._linear(self.to_q, x)
-            q = self._reshape_chs(q, (self.num_heads, -1))
-            kv = self._linear(self.to_kv, context)
-            kv = self._fused_pre(kv, num_fused=2)
-            if self.qk_rms_norm:
-                q = self.q_rms_norm(q)
-                k, v = kv.unbind(dim=1)
-                k = self.k_rms_norm(k)
-                kv = kv.replace(torch.stack([k.feats, v.feats], dim=1))
-            h = sparse_scaled_dot_product_attention(q, kv)
-        h = self._reshape_chs(h, (-1,))
-        h = self._linear(self.to_out, h)
-        return h

trellis/modules/sparse/attention/serialized_attn.py

@@ -1,193 +0,0 @@
-from typing import *
-from enum import Enum
-import torch
-import math
-from .. import SparseTensor
-from .. import DEBUG, ATTN
-
-if ATTN == 'xformers':
-    import xformers.ops as xops
-elif ATTN == 'flash_attn':
-    import flash_attn
-else:
-    raise ValueError(f"Unknown attention module: {ATTN}")
-
-
-__all__ = [
-    'sparse_serialized_scaled_dot_product_self_attention',
-]
-
-
-class SerializeMode(Enum):
-    Z_ORDER = 0
-    Z_ORDER_TRANSPOSED = 1
-    HILBERT = 2
-    HILBERT_TRANSPOSED = 3
-
-
-SerializeModes = [
-    SerializeMode.Z_ORDER,
-    SerializeMode.Z_ORDER_TRANSPOSED,
-    SerializeMode.HILBERT,
-    SerializeMode.HILBERT_TRANSPOSED
-]
-
-
-def calc_serialization(
-    tensor: SparseTensor,
-    window_size: int,
-    serialize_mode: SerializeMode = SerializeMode.Z_ORDER,
-    shift_sequence: int = 0,
-    shift_window: Tuple[int, int, int] = (0, 0, 0)
-) -> Tuple[torch.Tensor, torch.Tensor, List[int]]:
-    """
-    Calculate serialization and partitioning for a set of coordinates.
-
-    Args:
-        tensor (SparseTensor): The input tensor.
-        window_size (int): The window size to use.
-        serialize_mode (SerializeMode): The serialization mode to use.
-        shift_sequence (int): The shift of serialized sequence.
-        shift_window (Tuple[int, int, int]): The shift of serialized coordinates.
-
-    Returns:
-        (torch.Tensor, torch.Tensor): Forwards and backwards indices.
-    """
-    fwd_indices = []
-    bwd_indices = []
-    seq_lens = []
-    seq_batch_indices = []
-    offsets = [0]
-    
-    if 'vox2seq' not in globals():
-        import vox2seq
-
-    # Serialize the input
-    serialize_coords = tensor.coords[:, 1:].clone()
-    serialize_coords += torch.tensor(shift_window, dtype=torch.int32, device=tensor.device).reshape(1, 3)
-    if serialize_mode == SerializeMode.Z_ORDER:
-        code = vox2seq.encode(serialize_coords, mode='z_order', permute=[0, 1, 2])
-    elif serialize_mode == SerializeMode.Z_ORDER_TRANSPOSED:
-        code = vox2seq.encode(serialize_coords, mode='z_order', permute=[1, 0, 2])
-    elif serialize_mode == SerializeMode.HILBERT:
-        code = vox2seq.encode(serialize_coords, mode='hilbert', permute=[0, 1, 2])
-    elif serialize_mode == SerializeMode.HILBERT_TRANSPOSED:
-        code = vox2seq.encode(serialize_coords, mode='hilbert', permute=[1, 0, 2])
-    else:
-        raise ValueError(f"Unknown serialize mode: {serialize_mode}")
-    
-    for bi, s in enumerate(tensor.layout):
-        num_points = s.stop - s.start
-        num_windows = (num_points + window_size - 1) // window_size
-        valid_window_size = num_points / num_windows
-        to_ordered = torch.argsort(code[s.start:s.stop])
-        if num_windows == 1:
-            fwd_indices.append(to_ordered)
-            bwd_indices.append(torch.zeros_like(to_ordered).scatter_(0, to_ordered, torch.arange(num_points, device=tensor.device)))
-            fwd_indices[-1] += s.start
-            bwd_indices[-1] += offsets[-1]
-            seq_lens.append(num_points)
-            seq_batch_indices.append(bi)
-            offsets.append(offsets[-1] + seq_lens[-1])
-        else:
-            # Partition the input
-            offset = 0
-            mids = [(i + 0.5) * valid_window_size + shift_sequence for i in range(num_windows)]
-            split = [math.floor(i * valid_window_size + shift_sequence) for i in range(num_windows + 1)]
-            bwd_index = torch.zeros((num_points,), dtype=torch.int64, device=tensor.device)
-            for i in range(num_windows):
-                mid = mids[i]
-                valid_start = split[i]
-                valid_end = split[i + 1]
-                padded_start = math.floor(mid - 0.5 * window_size)
-                padded_end = padded_start + window_size
-                fwd_indices.append(to_ordered[torch.arange(padded_start, padded_end, device=tensor.device) % num_points])
-                offset += valid_start - padded_start
-                bwd_index.scatter_(0, fwd_indices[-1][valid_start-padded_start:valid_end-padded_start], torch.arange(offset, offset + valid_end - valid_start, device=tensor.device))
-                offset += padded_end - valid_start
-                fwd_indices[-1] += s.start
-            seq_lens.extend([window_size] * num_windows)
-            seq_batch_indices.extend([bi] * num_windows)
-            bwd_indices.append(bwd_index + offsets[-1])
-            offsets.append(offsets[-1] + num_windows * window_size)
-
-    fwd_indices = torch.cat(fwd_indices)
-    bwd_indices = torch.cat(bwd_indices)
-
-    return fwd_indices, bwd_indices, seq_lens, seq_batch_indices
-    
-
-def sparse_serialized_scaled_dot_product_self_attention(
-    qkv: SparseTensor,
-    window_size: int,
-    serialize_mode: SerializeMode = SerializeMode.Z_ORDER,
-    shift_sequence: int = 0,
-    shift_window: Tuple[int, int, int] = (0, 0, 0)
-) -> SparseTensor:
-    """
-    Apply serialized scaled dot product self attention to a sparse tensor.
-
-    Args:
-        qkv (SparseTensor): [N, *, 3, H, C] sparse tensor containing Qs, Ks, and Vs.
-        window_size (int): The window size to use.
-        serialize_mode (SerializeMode): The serialization mode to use.
-        shift_sequence (int): The shift of serialized sequence.
-        shift_window (Tuple[int, int, int]): The shift of serialized coordinates.
-        shift (int): The shift to use.
-    """
-    assert len(qkv.shape) == 4 and qkv.shape[1] == 3, f"Invalid shape for qkv, got {qkv.shape}, expected [N, *, 3, H, C]"
-
-    serialization_spatial_cache_name = f'serialization_{serialize_mode}_{window_size}_{shift_sequence}_{shift_window}'
-    serialization_spatial_cache = qkv.get_spatial_cache(serialization_spatial_cache_name)
-    if serialization_spatial_cache is None:
-        fwd_indices, bwd_indices, seq_lens, seq_batch_indices = calc_serialization(qkv, window_size, serialize_mode, shift_sequence, shift_window)
-        qkv.register_spatial_cache(serialization_spatial_cache_name, (fwd_indices, bwd_indices, seq_lens, seq_batch_indices))
-    else:
-        fwd_indices, bwd_indices, seq_lens, seq_batch_indices = serialization_spatial_cache
-
-    M = fwd_indices.shape[0]
-    T = qkv.feats.shape[0]
-    H = qkv.feats.shape[2]
-    C = qkv.feats.shape[3]
-    
-    qkv_feats = qkv.feats[fwd_indices]      # [M, 3, H, C]
-
-    if DEBUG:
-        start = 0
-        qkv_coords = qkv.coords[fwd_indices]
-        for i in range(len(seq_lens)):
-            assert (qkv_coords[start:start+seq_lens[i], 0] == seq_batch_indices[i]).all(), f"SparseWindowedScaledDotProductSelfAttention: batch index mismatch"
-            start += seq_lens[i]
-
-    if all([seq_len == window_size for seq_len in seq_lens]):
-        B = len(seq_lens)
-        N = window_size
-        qkv_feats = qkv_feats.reshape(B, N, 3, H, C)
-        if ATTN == 'xformers':
-            q, k, v = qkv_feats.unbind(dim=2)                       # [B, N, H, C]
-            out = xops.memory_efficient_attention(q, k, v)          # [B, N, H, C]
-        elif ATTN == 'flash_attn':
-            out = flash_attn.flash_attn_qkvpacked_func(qkv_feats)   # [B, N, H, C]
-        else:
-            raise ValueError(f"Unknown attention module: {ATTN}")
-        out = out.reshape(B * N, H, C)                              # [M, H, C]
-    else:
-        if ATTN == 'xformers':
-            q, k, v = qkv_feats.unbind(dim=1)                       # [M, H, C]
-            q = q.unsqueeze(0)                                      # [1, M, H, C]
-            k = k.unsqueeze(0)                                      # [1, M, H, C]
-            v = v.unsqueeze(0)                                      # [1, M, H, C]
-            mask = xops.fmha.BlockDiagonalMask.from_seqlens(seq_lens)
-            out = xops.memory_efficient_attention(q, k, v, mask)[0] # [M, H, C]
-        elif ATTN == 'flash_attn':
-            cu_seqlens = torch.cat([torch.tensor([0]), torch.cumsum(torch.tensor(seq_lens), dim=0)], dim=0) \
-                        .to(qkv.device).int()
-            out = flash_attn.flash_attn_varlen_qkvpacked_func(qkv_feats, cu_seqlens, max(seq_lens)) # [M, H, C]
-
-    out = out[bwd_indices]      # [T, H, C]
-
-    if DEBUG:
-        qkv_coords = qkv_coords[bwd_indices]
-        assert torch.equal(qkv_coords, qkv.coords), "SparseWindowedScaledDotProductSelfAttention: coordinate mismatch"
-
-    return qkv.replace(out)

trellis/modules/sparse/attention/windowed_attn.py

@@ -1,135 +0,0 @@
-from typing import *
-import torch
-import math
-from .. import SparseTensor
-from .. import DEBUG, ATTN
-
-if ATTN == 'xformers':
-    import xformers.ops as xops
-elif ATTN == 'flash_attn':
-    import flash_attn
-else:
-    raise ValueError(f"Unknown attention module: {ATTN}")
-
-
-__all__ = [
-    'sparse_windowed_scaled_dot_product_self_attention',
-]
-
-
-def calc_window_partition(
-    tensor: SparseTensor,
-    window_size: Union[int, Tuple[int, ...]],
-    shift_window: Union[int, Tuple[int, ...]] = 0
-) -> Tuple[torch.Tensor, torch.Tensor, List[int], List[int]]:
-    """
-    Calculate serialization and partitioning for a set of coordinates.
-
-    Args:
-        tensor (SparseTensor): The input tensor.
-        window_size (int): The window size to use.
-        shift_window (Tuple[int, ...]): The shift of serialized coordinates.
-
-    Returns:
-        (torch.Tensor): Forwards indices.
-        (torch.Tensor): Backwards indices.
-        (List[int]): Sequence lengths.
-        (List[int]): Sequence batch indices.
-    """
-    DIM = tensor.coords.shape[1] - 1
-    shift_window = (shift_window,) * DIM if isinstance(shift_window, int) else shift_window
-    window_size = (window_size,) * DIM if isinstance(window_size, int) else window_size
-    shifted_coords = tensor.coords.clone().detach()
-    shifted_coords[:, 1:] += torch.tensor(shift_window, device=tensor.device, dtype=torch.int32).unsqueeze(0)
-
-    MAX_COORDS = shifted_coords[:, 1:].max(dim=0).values.tolist()
-    NUM_WINDOWS = [math.ceil((mc + 1) / ws) for mc, ws in zip(MAX_COORDS, window_size)]
-    OFFSET = torch.cumprod(torch.tensor([1] + NUM_WINDOWS[::-1]), dim=0).tolist()[::-1]
-
-    shifted_coords[:, 1:] //= torch.tensor(window_size, device=tensor.device, dtype=torch.int32).unsqueeze(0)
-    shifted_indices = (shifted_coords * torch.tensor(OFFSET, device=tensor.device, dtype=torch.int32).unsqueeze(0)).sum(dim=1)
-    fwd_indices = torch.argsort(shifted_indices)
-    bwd_indices = torch.empty_like(fwd_indices)
-    bwd_indices[fwd_indices] = torch.arange(fwd_indices.shape[0], device=tensor.device)
-    seq_lens = torch.bincount(shifted_indices)
-    seq_batch_indices = torch.arange(seq_lens.shape[0], device=tensor.device, dtype=torch.int32) // OFFSET[0]
-    mask = seq_lens != 0
-    seq_lens = seq_lens[mask].tolist()
-    seq_batch_indices = seq_batch_indices[mask].tolist()
-
-    return fwd_indices, bwd_indices, seq_lens, seq_batch_indices
-    
-
-def sparse_windowed_scaled_dot_product_self_attention(
-    qkv: SparseTensor,
-    window_size: int,
-    shift_window: Tuple[int, int, int] = (0, 0, 0)
-) -> SparseTensor:
-    """
-    Apply windowed scaled dot product self attention to a sparse tensor.
-
-    Args:
-        qkv (SparseTensor): [N, *, 3, H, C] sparse tensor containing Qs, Ks, and Vs.
-        window_size (int): The window size to use.
-        shift_window (Tuple[int, int, int]): The shift of serialized coordinates.
-        shift (int): The shift to use.
-    """
-    assert len(qkv.shape) == 4 and qkv.shape[1] == 3, f"Invalid shape for qkv, got {qkv.shape}, expected [N, *, 3, H, C]"
-
-    serialization_spatial_cache_name = f'window_partition_{window_size}_{shift_window}'
-    serialization_spatial_cache = qkv.get_spatial_cache(serialization_spatial_cache_name)
-    if serialization_spatial_cache is None:
-        fwd_indices, bwd_indices, seq_lens, seq_batch_indices = calc_window_partition(qkv, window_size, shift_window)
-        qkv.register_spatial_cache(serialization_spatial_cache_name, (fwd_indices, bwd_indices, seq_lens, seq_batch_indices))
-    else:
-        fwd_indices, bwd_indices, seq_lens, seq_batch_indices = serialization_spatial_cache
-
-    M = fwd_indices.shape[0]
-    T = qkv.feats.shape[0]
-    H = qkv.feats.shape[2]
-    C = qkv.feats.shape[3]
-    
-    qkv_feats = qkv.feats[fwd_indices]      # [M, 3, H, C]
-
-    if DEBUG:
-        start = 0
-        qkv_coords = qkv.coords[fwd_indices]
-        for i in range(len(seq_lens)):
-            seq_coords = qkv_coords[start:start+seq_lens[i]]
-            assert (seq_coords[:, 0] == seq_batch_indices[i]).all(), f"SparseWindowedScaledDotProductSelfAttention: batch index mismatch"
-            assert (seq_coords[:, 1:].max(dim=0).values - seq_coords[:, 1:].min(dim=0).values < window_size).all(), \
-                    f"SparseWindowedScaledDotProductSelfAttention: window size exceeded"
-            start += seq_lens[i]
-
-    if all([seq_len == window_size for seq_len in seq_lens]):
-        B = len(seq_lens)
-        N = window_size
-        qkv_feats = qkv_feats.reshape(B, N, 3, H, C)
-        if ATTN == 'xformers':
-            q, k, v = qkv_feats.unbind(dim=2)                       # [B, N, H, C]
-            out = xops.memory_efficient_attention(q, k, v)          # [B, N, H, C]
-        elif ATTN == 'flash_attn':
-            out = flash_attn.flash_attn_qkvpacked_func(qkv_feats)   # [B, N, H, C]
-        else:
-            raise ValueError(f"Unknown attention module: {ATTN}")
-        out = out.reshape(B * N, H, C)                              # [M, H, C]
-    else:
-        if ATTN == 'xformers':
-            q, k, v = qkv_feats.unbind(dim=1)                       # [M, H, C]
-            q = q.unsqueeze(0)                                      # [1, M, H, C]
-            k = k.unsqueeze(0)                                      # [1, M, H, C]
-            v = v.unsqueeze(0)                                      # [1, M, H, C]
-            mask = xops.fmha.BlockDiagonalMask.from_seqlens(seq_lens)
-            out = xops.memory_efficient_attention(q, k, v, mask)[0] # [M, H, C]
-        elif ATTN == 'flash_attn':
-            cu_seqlens = torch.cat([torch.tensor([0]), torch.cumsum(torch.tensor(seq_lens), dim=0)], dim=0) \
-                        .to(qkv.device).int()
-            out = flash_attn.flash_attn_varlen_qkvpacked_func(qkv_feats, cu_seqlens, max(seq_lens)) # [M, H, C]
-
-    out = out[bwd_indices]      # [T, H, C]
-
-    if DEBUG:
-        qkv_coords = qkv_coords[bwd_indices]
-        assert torch.equal(qkv_coords, qkv.coords), "SparseWindowedScaledDotProductSelfAttention: coordinate mismatch"
-
-    return qkv.replace(out)

trellis/modules/sparse/basic.py

@@ -1,459 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from . import BACKEND, DEBUG
-SparseTensorData = None # Lazy import
-
-
-__all__ = [
-    'SparseTensor',
-    'sparse_batch_broadcast',
-    'sparse_batch_op',
-    'sparse_cat',
-    'sparse_unbind',
-]
-
-
-class SparseTensor:
-    """
-    Sparse tensor with support for both torchsparse and spconv backends.
-    
-    Parameters:
-    - feats (torch.Tensor): Features of the sparse tensor.
-    - coords (torch.Tensor): Coordinates of the sparse tensor.
-    - shape (torch.Size): Shape of the sparse tensor.
-    - layout (List[slice]): Layout of the sparse tensor for each batch
-    - data (SparseTensorData): Sparse tensor data used for convolusion
-
-    NOTE:
-    - Data corresponding to a same batch should be contiguous.
-    - Coords should be in [0, 1023]
-    """
-    @overload
-    def __init__(self, feats: torch.Tensor, coords: torch.Tensor, shape: Optional[torch.Size] = None, layout: Optional[List[slice]] = None, **kwargs): ...
-
-    @overload
-    def __init__(self, data, shape: Optional[torch.Size] = None, layout: Optional[List[slice]] = None, **kwargs): ...
-
-    def __init__(self, *args, **kwargs):
-        # Lazy import of sparse tensor backend
-        global SparseTensorData
-        if SparseTensorData is None:
-            import importlib
-            if BACKEND == 'torchsparse':
-                SparseTensorData = importlib.import_module('torchsparse').SparseTensor
-            elif BACKEND == 'spconv':
-                SparseTensorData = importlib.import_module('spconv.pytorch').SparseConvTensor
-                
-        method_id = 0
-        if len(args) != 0:
-            method_id = 0 if isinstance(args[0], torch.Tensor) else 1
-        else:
-            method_id = 1 if 'data' in kwargs else 0
-
-        if method_id == 0:
-            feats, coords, shape, layout = args + (None,) * (4 - len(args))
-            if 'feats' in kwargs:
-                feats = kwargs['feats']
-                del kwargs['feats']
-            if 'coords' in kwargs:
-                coords = kwargs['coords']
-                del kwargs['coords']
-            if 'shape' in kwargs:
-                shape = kwargs['shape']
-                del kwargs['shape']
-            if 'layout' in kwargs:
-                layout = kwargs['layout']
-                del kwargs['layout']
-
-            if shape is None:
-                shape = self.__cal_shape(feats, coords)
-            if layout is None:
-                layout = self.__cal_layout(coords, shape[0])
-            if BACKEND == 'torchsparse':
-                self.data = SparseTensorData(feats, coords, **kwargs)
-            elif BACKEND == 'spconv':
-                spatial_shape = list(coords.max(0)[0] + 1)[1:]
-                self.data = SparseTensorData(feats.reshape(feats.shape[0], -1), coords, spatial_shape, shape[0], **kwargs)
-                self.data._features = feats
-        elif method_id == 1:
-            data, shape, layout = args + (None,) * (3 - len(args))
-            if 'data' in kwargs:
-                data = kwargs['data']
-                del kwargs['data']
-            if 'shape' in kwargs:
-                shape = kwargs['shape']
-                del kwargs['shape']
-            if 'layout' in kwargs:
-                layout = kwargs['layout']
-                del kwargs['layout']
-
-            self.data = data
-            if shape is None:
-                shape = self.__cal_shape(self.feats, self.coords)
-            if layout is None:
-                layout = self.__cal_layout(self.coords, shape[0])
-
-        self._shape = shape
-        self._layout = layout
-        self._scale = kwargs.get('scale', (1, 1, 1))
-        self._spatial_cache = kwargs.get('spatial_cache', {})
-
-        if DEBUG:
-            try:
-                assert self.feats.shape[0] == self.coords.shape[0], f"Invalid feats shape: {self.feats.shape}, coords shape: {self.coords.shape}"
-                assert self.shape == self.__cal_shape(self.feats, self.coords), f"Invalid shape: {self.shape}"
-                assert self.layout == self.__cal_layout(self.coords, self.shape[0]), f"Invalid layout: {self.layout}"
-                for i in range(self.shape[0]):
-                    assert torch.all(self.coords[self.layout[i], 0] == i), f"The data of batch {i} is not contiguous"
-            except Exception as e:
-                print('Debugging information:')
-                print(f"- Shape: {self.shape}")
-                print(f"- Layout: {self.layout}")
-                print(f"- Scale: {self._scale}")
-                print(f"- Coords: {self.coords}")
-                raise e
-        
-    def __cal_shape(self, feats, coords):
-        shape = []
-        shape.append(coords[:, 0].max().item() + 1)
-        shape.extend([*feats.shape[1:]])
-        return torch.Size(shape)
-    
-    def __cal_layout(self, coords, batch_size):
-        seq_len = torch.bincount(coords[:, 0], minlength=batch_size)
-        offset = torch.cumsum(seq_len, dim=0) 
-        layout = [slice((offset[i] - seq_len[i]).item(), offset[i].item()) for i in range(batch_size)]
-        return layout
-    
-    @property
-    def shape(self) -> torch.Size:
-        return self._shape
-    
-    def dim(self) -> int:
-        return len(self.shape)
-    
-    @property
-    def layout(self) -> List[slice]:
-        return self._layout
-
-    @property
-    def feats(self) -> torch.Tensor:
-        if BACKEND == 'torchsparse':
-            return self.data.F
-        elif BACKEND == 'spconv':
-            return self.data.features
-    
-    @feats.setter
-    def feats(self, value: torch.Tensor):
-        if BACKEND == 'torchsparse':
-            self.data.F = value
-        elif BACKEND == 'spconv':
-            self.data.features = value
-
-    @property
-    def coords(self) -> torch.Tensor:
-        if BACKEND == 'torchsparse':
-            return self.data.C
-        elif BACKEND == 'spconv':
-            return self.data.indices
-        
-    @coords.setter
-    def coords(self, value: torch.Tensor):
-        if BACKEND == 'torchsparse':
-            self.data.C = value
-        elif BACKEND == 'spconv':
-            self.data.indices = value
-
-    @property
-    def dtype(self):
-        return self.feats.dtype
-
-    @property
-    def device(self):
-        return self.feats.device
-
-    @overload
-    def to(self, dtype: torch.dtype) -> 'SparseTensor': ...
-
-    @overload
-    def to(self, device: Optional[Union[str, torch.device]] = None, dtype: Optional[torch.dtype] = None) -> 'SparseTensor': ...
-
-    def to(self, *args, **kwargs) -> 'SparseTensor':
-        device = None
-        dtype = None
-        if len(args) == 2:
-            device, dtype = args
-        elif len(args) == 1:
-            if isinstance(args[0], torch.dtype):
-                dtype = args[0]
-            else:
-                device = args[0]
-        if 'dtype' in kwargs:
-            assert dtype is None, "to() received multiple values for argument 'dtype'"
-            dtype = kwargs['dtype']
-        if 'device' in kwargs:
-            assert device is None, "to() received multiple values for argument 'device'"
-            device = kwargs['device']
-        
-        new_feats = self.feats.to(device=device, dtype=dtype)
-        new_coords = self.coords.to(device=device)
-        return self.replace(new_feats, new_coords)
-
-    def type(self, dtype):
-        new_feats = self.feats.type(dtype)
-        return self.replace(new_feats)
-
-    def cpu(self) -> 'SparseTensor':
-        new_feats = self.feats.cpu()
-        new_coords = self.coords.cpu()
-        return self.replace(new_feats, new_coords)
-    
-    def cuda(self) -> 'SparseTensor':
-        new_feats = self.feats.cuda()
-        new_coords = self.coords.cuda()
-        return self.replace(new_feats, new_coords)
-
-    def half(self) -> 'SparseTensor':
-        new_feats = self.feats.half()
-        return self.replace(new_feats)
-    
-    def float(self) -> 'SparseTensor':
-        new_feats = self.feats.float()
-        return self.replace(new_feats)
-    
-    def detach(self) -> 'SparseTensor':
-        new_coords = self.coords.detach()
-        new_feats = self.feats.detach()
-        return self.replace(new_feats, new_coords)
-
-    def dense(self) -> torch.Tensor:
-        if BACKEND == 'torchsparse':
-            return self.data.dense()
-        elif BACKEND == 'spconv':
-            return self.data.dense()
-
-    def reshape(self, *shape) -> 'SparseTensor':
-        new_feats = self.feats.reshape(self.feats.shape[0], *shape)
-        return self.replace(new_feats)
-    
-    def unbind(self, dim: int) -> List['SparseTensor']:
-        return sparse_unbind(self, dim)
-
-    def replace(self, feats: torch.Tensor, coords: Optional[torch.Tensor] = None) -> 'SparseTensor':
-        new_shape = [self.shape[0]]
-        new_shape.extend(feats.shape[1:])
-        if BACKEND == 'torchsparse':
-            new_data = SparseTensorData(
-                feats=feats,
-                coords=self.data.coords if coords is None else coords,
-                stride=self.data.stride,
-                spatial_range=self.data.spatial_range,
-            )
-            new_data._caches = self.data._caches
-        elif BACKEND == 'spconv':
-            new_data = SparseTensorData(
-                self.data.features.reshape(self.data.features.shape[0], -1),
-                self.data.indices,
-                self.data.spatial_shape,
-                self.data.batch_size,
-                self.data.grid,
-                self.data.voxel_num,
-                self.data.indice_dict
-            )
-            new_data._features = feats
-            new_data.benchmark = self.data.benchmark
-            new_data.benchmark_record = self.data.benchmark_record
-            new_data.thrust_allocator = self.data.thrust_allocator
-            new_data._timer = self.data._timer
-            new_data.force_algo = self.data.force_algo
-            new_data.int8_scale = self.data.int8_scale
-            if coords is not None:
-                new_data.indices = coords
-        new_tensor = SparseTensor(new_data, shape=torch.Size(new_shape), layout=self.layout, scale=self._scale, spatial_cache=self._spatial_cache)
-        return new_tensor
-
-    @staticmethod
-    def full(aabb, dim, value, dtype=torch.float32, device=None) -> 'SparseTensor':
-        N, C = dim
-        x = torch.arange(aabb[0], aabb[3] + 1)
-        y = torch.arange(aabb[1], aabb[4] + 1)
-        z = torch.arange(aabb[2], aabb[5] + 1)
-        coords = torch.stack(torch.meshgrid(x, y, z, indexing='ij'), dim=-1).reshape(-1, 3)
-        coords = torch.cat([
-            torch.arange(N).view(-1, 1).repeat(1, coords.shape[0]).view(-1, 1),
-            coords.repeat(N, 1),
-        ], dim=1).to(dtype=torch.int32, device=device)
-        feats = torch.full((coords.shape[0], C), value, dtype=dtype, device=device)
-        return SparseTensor(feats=feats, coords=coords)
-
-    def __merge_sparse_cache(self, other: 'SparseTensor') -> dict:
-        new_cache = {}
-        for k in set(list(self._spatial_cache.keys()) + list(other._spatial_cache.keys())):
-            if k in self._spatial_cache:
-                new_cache[k] = self._spatial_cache[k]
-            if k in other._spatial_cache:
-                if k not in new_cache:
-                    new_cache[k] = other._spatial_cache[k]
-                else:
-                    new_cache[k].update(other._spatial_cache[k])
-        return new_cache
-
-    def __neg__(self) -> 'SparseTensor':
-        return self.replace(-self.feats)
-    
-    def __elemwise__(self, other: Union[torch.Tensor, 'SparseTensor'], op: callable) -> 'SparseTensor':
-        if isinstance(other, torch.Tensor):
-            try:
-                other = torch.broadcast_to(other, self.shape)
-                other = sparse_batch_broadcast(self, other)
-            except:
-                pass
-        if isinstance(other, SparseTensor):
-            other = other.feats
-        new_feats = op(self.feats, other)
-        new_tensor = self.replace(new_feats)
-        if isinstance(other, SparseTensor):
-            new_tensor._spatial_cache = self.__merge_sparse_cache(other)
-        return new_tensor
-
-    def __add__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.add)
-
-    def __radd__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.add)
-    
-    def __sub__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.sub)
-    
-    def __rsub__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, lambda x, y: torch.sub(y, x))
-
-    def __mul__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.mul)
-
-    def __rmul__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.mul)
-
-    def __truediv__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, torch.div)
-
-    def __rtruediv__(self, other: Union[torch.Tensor, 'SparseTensor', float]) -> 'SparseTensor':
-        return self.__elemwise__(other, lambda x, y: torch.div(y, x))
-
-    def __getitem__(self, idx):
-        if isinstance(idx, int):
-            idx = [idx]
-        elif isinstance(idx, slice):
-            idx = range(*idx.indices(self.shape[0]))
-        elif isinstance(idx, torch.Tensor):
-            if idx.dtype == torch.bool:
-                assert idx.shape == (self.shape[0],), f"Invalid index shape: {idx.shape}"
-                idx = idx.nonzero().squeeze(1)
-            elif idx.dtype in [torch.int32, torch.int64]:
-                assert len(idx.shape) == 1, f"Invalid index shape: {idx.shape}"
-            else:
-                raise ValueError(f"Unknown index type: {idx.dtype}")
-        else:
-            raise ValueError(f"Unknown index type: {type(idx)}")
-        
-        coords = []
-        feats = []
-        for new_idx, old_idx in enumerate(idx):
-            coords.append(self.coords[self.layout[old_idx]].clone())
-            coords[-1][:, 0] = new_idx
-            feats.append(self.feats[self.layout[old_idx]])
-        coords = torch.cat(coords, dim=0).contiguous()
-        feats = torch.cat(feats, dim=0).contiguous()
-        return SparseTensor(feats=feats, coords=coords)
-
-    def register_spatial_cache(self, key, value) -> None:
-        """
-        Register a spatial cache.
-        The spatial cache can be any thing you want to cache.
-        The registery and retrieval of the cache is based on current scale.
-        """
-        scale_key = str(self._scale)
-        if scale_key not in self._spatial_cache:
-            self._spatial_cache[scale_key] = {}
-        self._spatial_cache[scale_key][key] = value
-
-    def get_spatial_cache(self, key=None):
-        """
-        Get a spatial cache.
-        """
-        scale_key = str(self._scale)
-        cur_scale_cache = self._spatial_cache.get(scale_key, {})
-        if key is None:
-            return cur_scale_cache
-        return cur_scale_cache.get(key, None)
-
-
-def sparse_batch_broadcast(input: SparseTensor, other: torch.Tensor) -> torch.Tensor:
-    """
-    Broadcast a 1D tensor to a sparse tensor along the batch dimension then perform an operation.
-    
-    Args:
-        input (torch.Tensor): 1D tensor to broadcast.
-        target (SparseTensor): Sparse tensor to broadcast to.
-        op (callable): Operation to perform after broadcasting. Defaults to torch.add.
-    """
-    coords, feats = input.coords, input.feats
-    broadcasted = torch.zeros_like(feats)
-    for k in range(input.shape[0]):
-        broadcasted[input.layout[k]] = other[k]
-    return broadcasted
-
-
-def sparse_batch_op(input: SparseTensor, other: torch.Tensor, op: callable = torch.add) -> SparseTensor:
-    """
-    Broadcast a 1D tensor to a sparse tensor along the batch dimension then perform an operation.
-    
-    Args:
-        input (torch.Tensor): 1D tensor to broadcast.
-        target (SparseTensor): Sparse tensor to broadcast to.
-        op (callable): Operation to perform after broadcasting. Defaults to torch.add.
-    """
-    return input.replace(op(input.feats, sparse_batch_broadcast(input, other)))
-
-
-def sparse_cat(inputs: List[SparseTensor], dim: int = 0) -> SparseTensor:
-    """
-    Concatenate a list of sparse tensors.
-    
-    Args:
-        inputs (List[SparseTensor]): List of sparse tensors to concatenate.
-    """
-    if dim == 0:
-        start = 0
-        coords = []
-        for input in inputs:
-            coords.append(input.coords.clone())
-            coords[-1][:, 0] += start
-            start += input.shape[0]
-        coords = torch.cat(coords, dim=0)
-        feats = torch.cat([input.feats for input in inputs], dim=0)
-        output = SparseTensor(
-            coords=coords,
-            feats=feats,
-        )
-    else:
-        feats = torch.cat([input.feats for input in inputs], dim=dim)
-        output = inputs[0].replace(feats)
-
-    return output
-
-
-def sparse_unbind(input: SparseTensor, dim: int) -> List[SparseTensor]:
-    """
-    Unbind a sparse tensor along a dimension.
-    
-    Args:
-        input (SparseTensor): Sparse tensor to unbind.
-        dim (int): Dimension to unbind.
-    """
-    if dim == 0:
-        return [input[i] for i in range(input.shape[0])]
-    else:
-        feats = input.feats.unbind(dim)
-        return [input.replace(f) for f in feats]

trellis/modules/sparse/conv/__init__.py

@@ -1,21 +0,0 @@
-from .. import BACKEND
-
-
-SPCONV_ALGO = 'auto'    # 'auto', 'implicit_gemm', 'native'
-
-def __from_env():
-    import os
-        
-    global SPCONV_ALGO
-    env_spconv_algo = os.environ.get('SPCONV_ALGO')
-    if env_spconv_algo is not None and env_spconv_algo in ['auto', 'implicit_gemm', 'native']:
-        SPCONV_ALGO = env_spconv_algo
-    print(f"[SPARSE][CONV] spconv algo: {SPCONV_ALGO}")
-        
-
-__from_env()
-
-if BACKEND == 'torchsparse':
-    from .conv_torchsparse import *
-elif BACKEND == 'spconv':
-    from .conv_spconv import *

trellis/modules/sparse/conv/conv_spconv.py

@@ -1,80 +0,0 @@
-import torch
-import torch.nn as nn
-from .. import SparseTensor
-from .. import DEBUG
-from . import SPCONV_ALGO
-
-class SparseConv3d(nn.Module):
-    def __init__(self, in_channels, out_channels, kernel_size, stride=1, dilation=1, padding=None, bias=True, indice_key=None):
-        super(SparseConv3d, self).__init__()
-        if 'spconv' not in globals():
-            import spconv.pytorch as spconv
-        algo = None
-        if SPCONV_ALGO == 'native':
-            algo = spconv.ConvAlgo.Native
-        elif SPCONV_ALGO == 'implicit_gemm':
-            algo = spconv.ConvAlgo.MaskImplicitGemm
-        if stride == 1 and (padding is None):
-            self.conv = spconv.SubMConv3d(in_channels, out_channels, kernel_size, dilation=dilation, bias=bias, indice_key=indice_key, algo=algo)
-        else:
-            self.conv = spconv.SparseConv3d(in_channels, out_channels, kernel_size, stride=stride, dilation=dilation, padding=padding, bias=bias, indice_key=indice_key, algo=algo)
-        self.stride = tuple(stride) if isinstance(stride, (list, tuple)) else (stride, stride, stride)
-        self.padding = padding
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        spatial_changed = any(s != 1 for s in self.stride) or (self.padding is not None)
-        new_data = self.conv(x.data)
-        new_shape = [x.shape[0], self.conv.out_channels]
-        new_layout = None if spatial_changed else x.layout
-
-        if spatial_changed and (x.shape[0] != 1):
-            # spconv was non-1 stride will break the contiguous of the output tensor, sort by the coords
-            fwd = new_data.indices[:, 0].argsort()
-            bwd = torch.zeros_like(fwd).scatter_(0, fwd, torch.arange(fwd.shape[0], device=fwd.device))
-            sorted_feats = new_data.features[fwd]
-            sorted_coords = new_data.indices[fwd]
-            unsorted_data = new_data
-            new_data = spconv.SparseConvTensor(sorted_feats, sorted_coords, unsorted_data.spatial_shape, unsorted_data.batch_size)  # type: ignore
-
-        out = SparseTensor(
-            new_data, shape=torch.Size(new_shape), layout=new_layout,
-            scale=tuple([s * stride for s, stride in zip(x._scale, self.stride)]),
-            spatial_cache=x._spatial_cache,
-        )
-
-        if spatial_changed and (x.shape[0] != 1):
-            out.register_spatial_cache(f'conv_{self.stride}_unsorted_data', unsorted_data)
-            out.register_spatial_cache(f'conv_{self.stride}_sort_bwd', bwd)
- 
-        return out
-
-
-class SparseInverseConv3d(nn.Module):
-    def __init__(self, in_channels, out_channels, kernel_size, stride=1, dilation=1, bias=True, indice_key=None):
-        super(SparseInverseConv3d, self).__init__()
-        if 'spconv' not in globals():
-            import spconv.pytorch as spconv
-        self.conv = spconv.SparseInverseConv3d(in_channels, out_channels, kernel_size, bias=bias, indice_key=indice_key)
-        self.stride = tuple(stride) if isinstance(stride, (list, tuple)) else (stride, stride, stride)
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        spatial_changed = any(s != 1 for s in self.stride)
-        if spatial_changed:
-            # recover the original spconv order
-            data = x.get_spatial_cache(f'conv_{self.stride}_unsorted_data')
-            bwd = x.get_spatial_cache(f'conv_{self.stride}_sort_bwd')
-            data = data.replace_feature(x.feats[bwd])
-            if DEBUG:
-                assert torch.equal(data.indices, x.coords[bwd]), 'Recover the original order failed'
-        else:
-            data = x.data
-
-        new_data = self.conv(data)
-        new_shape = [x.shape[0], self.conv.out_channels]
-        new_layout = None if spatial_changed else x.layout
-        out = SparseTensor(
-            new_data, shape=torch.Size(new_shape), layout=new_layout,
-            scale=tuple([s // stride for s, stride in zip(x._scale, self.stride)]),
-            spatial_cache=x._spatial_cache,
-        )
-        return out

trellis/modules/sparse/conv/conv_torchsparse.py

@@ -1,38 +0,0 @@
-import torch
-import torch.nn as nn
-from .. import SparseTensor
-
-
-class SparseConv3d(nn.Module):
-    def __init__(self, in_channels, out_channels, kernel_size, stride=1, dilation=1, bias=True, indice_key=None):
-        super(SparseConv3d, self).__init__()
-        if 'torchsparse' not in globals():
-            import torchsparse
-        self.conv = torchsparse.nn.Conv3d(in_channels, out_channels, kernel_size, stride, 0, dilation, bias)
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        out = self.conv(x.data)
-        new_shape = [x.shape[0], self.conv.out_channels]
-        out = SparseTensor(out, shape=torch.Size(new_shape), layout=x.layout if all(s == 1 for s in self.conv.stride) else None)
-        out._spatial_cache = x._spatial_cache
-        out._scale = tuple([s * stride for s, stride in zip(x._scale, self.conv.stride)])
-        return out
-
-
-class SparseInverseConv3d(nn.Module):
-    def __init__(self, in_channels, out_channels, kernel_size, stride=1, dilation=1, bias=True, indice_key=None):
-        super(SparseInverseConv3d, self).__init__()
-        if 'torchsparse' not in globals():
-            import torchsparse
-        self.conv = torchsparse.nn.Conv3d(in_channels, out_channels, kernel_size, stride, 0, dilation, bias, transposed=True)
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        out = self.conv(x.data)        
-        new_shape = [x.shape[0], self.conv.out_channels]
-        out = SparseTensor(out, shape=torch.Size(new_shape), layout=x.layout if all(s == 1 for s in self.conv.stride) else None)
-        out._spatial_cache = x._spatial_cache
-        out._scale = tuple([s // stride for s, stride in zip(x._scale, self.conv.stride)])
-        return out
-
-
-

trellis/modules/sparse/linear.py

@@ -1,15 +0,0 @@
-import torch
-import torch.nn as nn
-from . import SparseTensor
-
-__all__ = [
-    'SparseLinear'
-]
-
-
-class SparseLinear(nn.Linear):
-    def __init__(self, in_features, out_features, bias=True):
-        super(SparseLinear, self).__init__(in_features, out_features, bias)
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        return input.replace(super().forward(input.feats))

trellis/modules/sparse/nonlinearity.py

@@ -1,35 +0,0 @@
-import torch
-import torch.nn as nn
-from . import SparseTensor
-
-__all__ = [
-    'SparseReLU',
-    'SparseSiLU',
-    'SparseGELU',
-    'SparseActivation'
-]
-
-
-class SparseReLU(nn.ReLU):
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        return input.replace(super().forward(input.feats))
-    
-
-class SparseSiLU(nn.SiLU):
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        return input.replace(super().forward(input.feats))
-
-
-class SparseGELU(nn.GELU):
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        return input.replace(super().forward(input.feats))
-
-
-class SparseActivation(nn.Module):
-    def __init__(self, activation: nn.Module):
-        super().__init__()
-        self.activation = activation
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        return input.replace(self.activation(input.feats))
-    

trellis/modules/sparse/norm.py

@@ -1,58 +0,0 @@
-import torch
-import torch.nn as nn
-from . import SparseTensor
-from . import DEBUG
-
-__all__ = [
-    'SparseGroupNorm',
-    'SparseLayerNorm',
-    'SparseGroupNorm32',
-    'SparseLayerNorm32',
-]
-
-
-class SparseGroupNorm(nn.GroupNorm):
-    def __init__(self, num_groups, num_channels, eps=1e-5, affine=True):
-        super(SparseGroupNorm, self).__init__(num_groups, num_channels, eps, affine)
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        nfeats = torch.zeros_like(input.feats)
-        for k in range(input.shape[0]):
-            if DEBUG:
-                assert (input.coords[input.layout[k], 0] == k).all(), f"SparseGroupNorm: batch index mismatch"
-            bfeats = input.feats[input.layout[k]]
-            bfeats = bfeats.permute(1, 0).reshape(1, input.shape[1], -1)
-            bfeats = super().forward(bfeats)
-            bfeats = bfeats.reshape(input.shape[1], -1).permute(1, 0)
-            nfeats[input.layout[k]] = bfeats
-        return input.replace(nfeats)
-
-
-class SparseLayerNorm(nn.LayerNorm):
-    def __init__(self, normalized_shape, eps=1e-5, elementwise_affine=True):
-        super(SparseLayerNorm, self).__init__(normalized_shape, eps, elementwise_affine)
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        nfeats = torch.zeros_like(input.feats)
-        for k in range(input.shape[0]):
-            bfeats = input.feats[input.layout[k]]
-            bfeats = bfeats.permute(1, 0).reshape(1, input.shape[1], -1)
-            bfeats = super().forward(bfeats)
-            bfeats = bfeats.reshape(input.shape[1], -1).permute(1, 0)
-            nfeats[input.layout[k]] = bfeats
-        return input.replace(nfeats)
-
-
-class SparseGroupNorm32(SparseGroupNorm):
-    """
-    A GroupNorm layer that converts to float32 before the forward pass.
-    """
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        return super().forward(x.float()).type(x.dtype)
-
-class SparseLayerNorm32(SparseLayerNorm):
-    """
-    A LayerNorm layer that converts to float32 before the forward pass.
-    """
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        return super().forward(x.float()).type(x.dtype)

trellis/modules/sparse/spatial.py

@@ -1,110 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from . import SparseTensor
-
-__all__ = [
-    'SparseDownsample',
-    'SparseUpsample',
-    'SparseSubdivide'
-]
-
-
-class SparseDownsample(nn.Module):
-    """
-    Downsample a sparse tensor by a factor of `factor`.
-    Implemented as average pooling.
-    """
-    def __init__(self, factor: Union[int, Tuple[int, ...], List[int]]):
-        super(SparseDownsample, self).__init__()
-        self.factor = tuple(factor) if isinstance(factor, (list, tuple)) else factor
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        DIM = input.coords.shape[-1] - 1
-        factor = self.factor if isinstance(self.factor, tuple) else (self.factor,) * DIM
-        assert DIM == len(factor), 'Input coordinates must have the same dimension as the downsample factor.'
-
-        coord = list(input.coords.unbind(dim=-1))
-        for i, f in enumerate(factor):
-            coord[i+1] = coord[i+1] // f
-
-        MAX = [coord[i+1].max().item() + 1 for i in range(DIM)]
-        OFFSET = torch.cumprod(torch.tensor(MAX[::-1]), 0).tolist()[::-1] + [1]
-        code = sum([c * o for c, o in zip(coord, OFFSET)])
-        code, idx = code.unique(return_inverse=True)
-
-        new_feats = torch.scatter_reduce(
-            torch.zeros(code.shape[0], input.feats.shape[1], device=input.feats.device, dtype=input.feats.dtype),
-            dim=0,
-            index=idx.unsqueeze(1).expand(-1, input.feats.shape[1]),
-            src=input.feats,
-            reduce='mean'
-        )
-        new_coords = torch.stack(
-            [code // OFFSET[0]] +
-            [(code // OFFSET[i+1]) % MAX[i] for i in range(DIM)],
-            dim=-1
-        )
-        out = SparseTensor(new_feats, new_coords, input.shape,)
-        out._scale = tuple([s // f for s, f in zip(input._scale, factor)])
-        out._spatial_cache = input._spatial_cache
-
-        out.register_spatial_cache(f'upsample_{factor}_coords', input.coords)
-        out.register_spatial_cache(f'upsample_{factor}_layout', input.layout)
-        out.register_spatial_cache(f'upsample_{factor}_idx', idx)
-
-        return out
-
-
-class SparseUpsample(nn.Module):
-    """
-    Upsample a sparse tensor by a factor of `factor`.
-    Implemented as nearest neighbor interpolation.
-    """
-    def __init__(self, factor: Union[int, Tuple[int, int, int], List[int]]):
-        super(SparseUpsample, self).__init__()
-        self.factor = tuple(factor) if isinstance(factor, (list, tuple)) else factor
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        DIM = input.coords.shape[-1] - 1
-        factor = self.factor if isinstance(self.factor, tuple) else (self.factor,) * DIM
-        assert DIM == len(factor), 'Input coordinates must have the same dimension as the upsample factor.'
-
-        new_coords = input.get_spatial_cache(f'upsample_{factor}_coords')
-        new_layout = input.get_spatial_cache(f'upsample_{factor}_layout')
-        idx = input.get_spatial_cache(f'upsample_{factor}_idx')
-        if any([x is None for x in [new_coords, new_layout, idx]]):
-            raise ValueError('Upsample cache not found. SparseUpsample must be paired with SparseDownsample.')
-        new_feats = input.feats[idx]
-        out = SparseTensor(new_feats, new_coords, input.shape, new_layout)
-        out._scale = tuple([s * f for s, f in zip(input._scale, factor)])
-        out._spatial_cache = input._spatial_cache
-        return out
-    
-class SparseSubdivide(nn.Module):
-    """
-    Upsample a sparse tensor by a factor of `factor`.
-    Implemented as nearest neighbor interpolation.
-    """
-    def __init__(self):
-        super(SparseSubdivide, self).__init__()
-
-    def forward(self, input: SparseTensor) -> SparseTensor:
-        DIM = input.coords.shape[-1] - 1
-        # upsample scale=2^DIM
-        n_cube = torch.ones([2] * DIM, device=input.device, dtype=torch.int)
-        n_coords = torch.nonzero(n_cube)
-        n_coords = torch.cat([torch.zeros_like(n_coords[:, :1]), n_coords], dim=-1)
-        factor = n_coords.shape[0]
-        assert factor == 2 ** DIM
-        # print(n_coords.shape)
-        new_coords = input.coords.clone()
-        new_coords[:, 1:] *= 2
-        new_coords = new_coords.unsqueeze(1) + n_coords.unsqueeze(0).to(new_coords.dtype)
-        
-        new_feats = input.feats.unsqueeze(1).expand(input.feats.shape[0], factor, *input.feats.shape[1:])
-        out = SparseTensor(new_feats.flatten(0, 1), new_coords.flatten(0, 1), input.shape)
-        out._scale = input._scale * 2
-        out._spatial_cache = input._spatial_cache
-        return out
-

trellis/modules/sparse/transformer/__init__.py

@@ -1,2 +0,0 @@
-from .blocks import *
-from .modulated import *

trellis/modules/sparse/transformer/blocks.py

@@ -1,151 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from ..basic import SparseTensor
-from ..linear import SparseLinear
-from ..nonlinearity import SparseGELU
-from ..attention import SparseMultiHeadAttention, SerializeMode
-from ...norm import LayerNorm32
-
-
-class SparseFeedForwardNet(nn.Module):
-    def __init__(self, channels: int, mlp_ratio: float = 4.0):
-        super().__init__()
-        self.mlp = nn.Sequential(
-            SparseLinear(channels, int(channels * mlp_ratio)),
-            SparseGELU(approximate="tanh"),
-            SparseLinear(int(channels * mlp_ratio), channels),
-        )
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        return self.mlp(x)
-
-
-class SparseTransformerBlock(nn.Module):
-    """
-    Sparse Transformer block (MSA + FFN).
-    """
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "shift_window", "shift_sequence", "shift_order", "swin"] = "full",
-        window_size: Optional[int] = None,
-        shift_sequence: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        serialize_mode: Optional[SerializeMode] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qkv_bias: bool = True,
-        ln_affine: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.norm1 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.attn = SparseMultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_sequence=shift_sequence,
-            shift_window=shift_window,
-            serialize_mode=serialize_mode,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.mlp = SparseFeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-
-    def _forward(self, x: SparseTensor) -> SparseTensor:
-        h = x.replace(self.norm1(x.feats))
-        h = self.attn(h)
-        x = x + h
-        h = x.replace(self.norm2(x.feats))
-        h = self.mlp(h)
-        x = x + h
-        return x
-
-    def forward(self, x: SparseTensor) -> SparseTensor:
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, use_reentrant=False)
-        else:
-            return self._forward(x)
-
-
-class SparseTransformerCrossBlock(nn.Module):
-    """
-    Sparse Transformer cross-attention block (MSA + MCA + FFN).
-    """
-    def __init__(
-        self,
-        channels: int,
-        ctx_channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "shift_window", "shift_sequence", "shift_order", "swin"] = "full",
-        window_size: Optional[int] = None,
-        shift_sequence: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        serialize_mode: Optional[SerializeMode] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qk_rms_norm_cross: bool = False,
-        qkv_bias: bool = True,
-        ln_affine: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.norm1 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm3 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.self_attn = SparseMultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            type="self",
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_sequence=shift_sequence,
-            shift_window=shift_window,
-            serialize_mode=serialize_mode,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.cross_attn = SparseMultiHeadAttention(
-            channels,
-            ctx_channels=ctx_channels,
-            num_heads=num_heads,
-            type="cross",
-            attn_mode="full",
-            qkv_bias=qkv_bias,
-            qk_rms_norm=qk_rms_norm_cross,
-        )
-        self.mlp = SparseFeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-
-    def _forward(self, x: SparseTensor, mod: torch.Tensor, context: torch.Tensor):
-        h = x.replace(self.norm1(x.feats))
-        h = self.self_attn(h)
-        x = x + h
-        h = x.replace(self.norm2(x.feats))
-        h = self.cross_attn(h, context)
-        x = x + h
-        h = x.replace(self.norm3(x.feats))
-        h = self.mlp(h)
-        x = x + h
-        return x
-
-    def forward(self, x: SparseTensor, context: torch.Tensor):
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, context, use_reentrant=False)
-        else:
-            return self._forward(x, context)

trellis/modules/sparse/transformer/modulated.py

@@ -1,166 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from ..basic import SparseTensor
-from ..attention import SparseMultiHeadAttention, SerializeMode
-from ...norm import LayerNorm32
-from .blocks import SparseFeedForwardNet
-
-
-class ModulatedSparseTransformerBlock(nn.Module):
-    """
-    Sparse Transformer block (MSA + FFN) with adaptive layer norm conditioning.
-    """
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "shift_window", "shift_sequence", "shift_order", "swin"] = "full",
-        window_size: Optional[int] = None,
-        shift_sequence: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        serialize_mode: Optional[SerializeMode] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qkv_bias: bool = True,
-        share_mod: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.share_mod = share_mod
-        self.norm1 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.attn = SparseMultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_sequence=shift_sequence,
-            shift_window=shift_window,
-            serialize_mode=serialize_mode,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.mlp = SparseFeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-        if not share_mod:
-            self.adaLN_modulation = nn.Sequential(
-                nn.SiLU(),
-                nn.Linear(channels, 6 * channels, bias=True)
-            )
-
-    def _forward(self, x: SparseTensor, mod: torch.Tensor) -> SparseTensor:
-        if self.share_mod:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = mod.chunk(6, dim=1)
-        else:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = self.adaLN_modulation(mod).chunk(6, dim=1)
-        h = x.replace(self.norm1(x.feats))
-        h = h * (1 + scale_msa) + shift_msa
-        h = self.attn(h)
-        h = h * gate_msa
-        x = x + h
-        h = x.replace(self.norm2(x.feats))
-        h = h * (1 + scale_mlp) + shift_mlp
-        h = self.mlp(h)
-        h = h * gate_mlp
-        x = x + h
-        return x
-
-    def forward(self, x: SparseTensor, mod: torch.Tensor) -> SparseTensor:
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, mod, use_reentrant=False)
-        else:
-            return self._forward(x, mod)
-
-
-class ModulatedSparseTransformerCrossBlock(nn.Module):
-    """
-    Sparse Transformer cross-attention block (MSA + MCA + FFN) with adaptive layer norm conditioning.
-    """
-    def __init__(
-        self,
-        channels: int,
-        ctx_channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "shift_window", "shift_sequence", "shift_order", "swin"] = "full",
-        window_size: Optional[int] = None,
-        shift_sequence: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        serialize_mode: Optional[SerializeMode] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qk_rms_norm_cross: bool = False,
-        qkv_bias: bool = True,
-        share_mod: bool = False,
-
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.share_mod = share_mod
-        self.norm1 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=True, eps=1e-6)
-        self.norm3 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.self_attn = SparseMultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            type="self",
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_sequence=shift_sequence,
-            shift_window=shift_window,
-            serialize_mode=serialize_mode,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.cross_attn = SparseMultiHeadAttention(
-            channels,
-            ctx_channels=ctx_channels,
-            num_heads=num_heads,
-            type="cross",
-            attn_mode="full",
-            qkv_bias=qkv_bias,
-            qk_rms_norm=qk_rms_norm_cross,
-        )
-        self.mlp = SparseFeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-        if not share_mod:
-            self.adaLN_modulation = nn.Sequential(
-                nn.SiLU(),
-                nn.Linear(channels, 6 * channels, bias=True)
-            )
-
-    def _forward(self, x: SparseTensor, mod: torch.Tensor, context: torch.Tensor) -> SparseTensor:
-        if self.share_mod:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = mod.chunk(6, dim=1)
-        else:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = self.adaLN_modulation(mod).chunk(6, dim=1)
-        h = x.replace(self.norm1(x.feats))
-        h = h * (1 + scale_msa) + shift_msa
-        h = self.self_attn(h)
-        h = h * gate_msa
-        x = x + h
-        h = x.replace(self.norm2(x.feats))
-        h = self.cross_attn(h, context)
-        x = x + h
-        h = x.replace(self.norm3(x.feats))
-        h = h * (1 + scale_mlp) + shift_mlp
-        h = self.mlp(h)
-        h = h * gate_mlp
-        x = x + h
-        return x
-
-    def forward(self, x: SparseTensor, mod: torch.Tensor, context: torch.Tensor) -> SparseTensor:
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, mod, context, use_reentrant=False)
-        else:
-            return self._forward(x, mod, context)

trellis/modules/spatial.py

@@ -1,48 +0,0 @@
-import torch
-
-
-def pixel_shuffle_3d(x: torch.Tensor, scale_factor: int) -> torch.Tensor:
-    """
-    3D pixel shuffle.
-    """
-    B, C, H, W, D = x.shape
-    C_ = C // scale_factor**3
-    x = x.reshape(B, C_, scale_factor, scale_factor, scale_factor, H, W, D)
-    x = x.permute(0, 1, 5, 2, 6, 3, 7, 4)
-    x = x.reshape(B, C_, H*scale_factor, W*scale_factor, D*scale_factor)
-    return x
-
-
-def patchify(x: torch.Tensor, patch_size: int):
-    """
-    Patchify a tensor.
-
-    Args:
-        x (torch.Tensor): (N, C, *spatial) tensor
-        patch_size (int): Patch size
-    """
-    DIM = x.dim() - 2
-    for d in range(2, DIM + 2):
-        assert x.shape[d] % patch_size == 0, f"Dimension {d} of input tensor must be divisible by patch size, got {x.shape[d]} and {patch_size}"
-
-    x = x.reshape(*x.shape[:2], *sum([[x.shape[d] // patch_size, patch_size] for d in range(2, DIM + 2)], []))
-    x = x.permute(0, 1, *([2 * i + 3 for i in range(DIM)] + [2 * i + 2 for i in range(DIM)]))
-    x = x.reshape(x.shape[0], x.shape[1] * (patch_size ** DIM), *(x.shape[-DIM:]))
-    return x
-
-
-def unpatchify(x: torch.Tensor, patch_size: int):
-    """
-    Unpatchify a tensor.
-
-    Args:
-        x (torch.Tensor): (N, C, *spatial) tensor
-        patch_size (int): Patch size
-    """
-    DIM = x.dim() - 2
-    assert x.shape[1] % (patch_size ** DIM) == 0, f"Second dimension of input tensor must be divisible by patch size to unpatchify, got {x.shape[1]} and {patch_size ** DIM}"
-
-    x = x.reshape(x.shape[0], x.shape[1] // (patch_size ** DIM), *([patch_size] * DIM), *(x.shape[-DIM:]))
-    x = x.permute(0, 1, *(sum([[2 + DIM + i, 2 + i] for i in range(DIM)], [])))
-    x = x.reshape(x.shape[0], x.shape[1], *[x.shape[2 + 2 * i] * patch_size for i in range(DIM)])
-    return x

trellis/modules/transformer/__init__.py

@@ -1,2 +0,0 @@
-from .blocks import *
-from .modulated import *

trellis/modules/transformer/blocks.py

@@ -1,182 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from ..attention import MultiHeadAttention
-from ..norm import LayerNorm32
-
-
-class AbsolutePositionEmbedder(nn.Module):
-    """
-    Embeds spatial positions into vector representations.
-    """
-    def __init__(self, channels: int, in_channels: int = 3):
-        super().__init__()
-        self.channels = channels
-        self.in_channels = in_channels
-        self.freq_dim = channels // in_channels // 2
-        self.freqs = torch.arange(self.freq_dim, dtype=torch.float32) / self.freq_dim
-        self.freqs = 1.0 / (10000 ** self.freqs)
-        
-    def _sin_cos_embedding(self, x: torch.Tensor) -> torch.Tensor:
-        """
-        Create sinusoidal position embeddings.
-
-        Args:
-            x: a 1-D Tensor of N indices
-
-        Returns:
-            an (N, D) Tensor of positional embeddings.
-        """
-        self.freqs = self.freqs.to(x.device)
-        out = torch.outer(x, self.freqs)
-        out = torch.cat([torch.sin(out), torch.cos(out)], dim=-1)
-        return out
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        """
-        Args:
-            x (torch.Tensor): (N, D) tensor of spatial positions
-        """
-        N, D = x.shape
-        assert D == self.in_channels, "Input dimension must match number of input channels"
-        embed = self._sin_cos_embedding(x.reshape(-1))
-        embed = embed.reshape(N, -1)
-        if embed.shape[1] < self.channels:
-            embed = torch.cat([embed, torch.zeros(N, self.channels - embed.shape[1], device=embed.device)], dim=-1)
-        return embed
-
-
-class FeedForwardNet(nn.Module):
-    def __init__(self, channels: int, mlp_ratio: float = 4.0):
-        super().__init__()
-        self.mlp = nn.Sequential(
-            nn.Linear(channels, int(channels * mlp_ratio)),
-            nn.GELU(approximate="tanh"),
-            nn.Linear(int(channels * mlp_ratio), channels),
-        )
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        return self.mlp(x)
-
-
-class TransformerBlock(nn.Module):
-    """
-    Transformer block (MSA + FFN).
-    """
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_window: Optional[int] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qkv_bias: bool = True,
-        ln_affine: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.norm1 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.attn = MultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_window=shift_window,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.mlp = FeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-
-    def _forward(self, x: torch.Tensor) -> torch.Tensor:
-        h = self.norm1(x)
-        h = self.attn(h)
-        x = x + h
-        h = self.norm2(x)
-        h = self.mlp(h)
-        x = x + h
-        return x
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, use_reentrant=False)
-        else:
-            return self._forward(x)
-
-
-class TransformerCrossBlock(nn.Module):
-    """
-    Transformer cross-attention block (MSA + MCA + FFN).
-    """
-    def __init__(
-        self,
-        channels: int,
-        ctx_channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qk_rms_norm_cross: bool = False,
-        qkv_bias: bool = True,
-        ln_affine: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.norm1 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.norm3 = LayerNorm32(channels, elementwise_affine=ln_affine, eps=1e-6)
-        self.self_attn = MultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            type="self",
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_window=shift_window,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.cross_attn = MultiHeadAttention(
-            channels,
-            ctx_channels=ctx_channels,
-            num_heads=num_heads,
-            type="cross",
-            attn_mode="full",
-            qkv_bias=qkv_bias,
-            qk_rms_norm=qk_rms_norm_cross,
-        )
-        self.mlp = FeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-
-    def _forward(self, x: torch.Tensor, context: torch.Tensor):
-        h = self.norm1(x)
-        h = self.self_attn(h)
-        x = x + h
-        h = self.norm2(x)
-        h = self.cross_attn(h, context)
-        x = x + h
-        h = self.norm3(x)
-        h = self.mlp(h)
-        x = x + h
-        return x
-
-    def forward(self, x: torch.Tensor, context: torch.Tensor):
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, context, use_reentrant=False)
-        else:
-            return self._forward(x, context)
-        

trellis/modules/transformer/modulated.py

@@ -1,157 +0,0 @@
-from typing import *
-import torch
-import torch.nn as nn
-from ..attention import MultiHeadAttention
-from ..norm import LayerNorm32
-from .blocks import FeedForwardNet
-
-
-class ModulatedTransformerBlock(nn.Module):
-    """
-    Transformer block (MSA + FFN) with adaptive layer norm conditioning.
-    """
-    def __init__(
-        self,
-        channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qkv_bias: bool = True,
-        share_mod: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.share_mod = share_mod
-        self.norm1 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.attn = MultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_window=shift_window,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.mlp = FeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-        if not share_mod:
-            self.adaLN_modulation = nn.Sequential(
-                nn.SiLU(),
-                nn.Linear(channels, 6 * channels, bias=True)
-            )
-
-    def _forward(self, x: torch.Tensor, mod: torch.Tensor) -> torch.Tensor:
-        if self.share_mod:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = mod.chunk(6, dim=1)
-        else:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = self.adaLN_modulation(mod).chunk(6, dim=1)
-        h = self.norm1(x)
-        h = h * (1 + scale_msa.unsqueeze(1)) + shift_msa.unsqueeze(1)
-        h = self.attn(h)
-        h = h * gate_msa.unsqueeze(1)
-        x = x + h
-        h = self.norm2(x)
-        h = h * (1 + scale_mlp.unsqueeze(1)) + shift_mlp.unsqueeze(1)
-        h = self.mlp(h)
-        h = h * gate_mlp.unsqueeze(1)
-        x = x + h
-        return x
-
-    def forward(self, x: torch.Tensor, mod: torch.Tensor) -> torch.Tensor:
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, mod, use_reentrant=False)
-        else:
-            return self._forward(x, mod)
-
-
-class ModulatedTransformerCrossBlock(nn.Module):
-    """
-    Transformer cross-attention block (MSA + MCA + FFN) with adaptive layer norm conditioning.
-    """
-    def __init__(
-        self,
-        channels: int,
-        ctx_channels: int,
-        num_heads: int,
-        mlp_ratio: float = 4.0,
-        attn_mode: Literal["full", "windowed"] = "full",
-        window_size: Optional[int] = None,
-        shift_window: Optional[Tuple[int, int, int]] = None,
-        use_checkpoint: bool = False,
-        use_rope: bool = False,
-        qk_rms_norm: bool = False,
-        qk_rms_norm_cross: bool = False,
-        qkv_bias: bool = True,
-        share_mod: bool = False,
-    ):
-        super().__init__()
-        self.use_checkpoint = use_checkpoint
-        self.share_mod = share_mod
-        self.norm1 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.norm2 = LayerNorm32(channels, elementwise_affine=True, eps=1e-6)
-        self.norm3 = LayerNorm32(channels, elementwise_affine=False, eps=1e-6)
-        self.self_attn = MultiHeadAttention(
-            channels,
-            num_heads=num_heads,
-            type="self",
-            attn_mode=attn_mode,
-            window_size=window_size,
-            shift_window=shift_window,
-            qkv_bias=qkv_bias,
-            use_rope=use_rope,
-            qk_rms_norm=qk_rms_norm,
-        )
-        self.cross_attn = MultiHeadAttention(
-            channels,
-            ctx_channels=ctx_channels,
-            num_heads=num_heads,
-            type="cross",
-            attn_mode="full",
-            qkv_bias=qkv_bias,
-            qk_rms_norm=qk_rms_norm_cross,
-        )
-        self.mlp = FeedForwardNet(
-            channels,
-            mlp_ratio=mlp_ratio,
-        )
-        if not share_mod:
-            self.adaLN_modulation = nn.Sequential(
-                nn.SiLU(),
-                nn.Linear(channels, 6 * channels, bias=True)
-            )
-
-    def _forward(self, x: torch.Tensor, mod: torch.Tensor, context: torch.Tensor):
-        if self.share_mod:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = mod.chunk(6, dim=1)
-        else:
-            shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = self.adaLN_modulation(mod).chunk(6, dim=1)
-        h = self.norm1(x)
-        h = h * (1 + scale_msa.unsqueeze(1)) + shift_msa.unsqueeze(1)
-        h = self.self_attn(h)
-        h = h * gate_msa.unsqueeze(1)
-        x = x + h
-        h = self.norm2(x)
-        h = self.cross_attn(h, context)
-        x = x + h
-        h = self.norm3(x)
-        h = h * (1 + scale_mlp.unsqueeze(1)) + shift_mlp.unsqueeze(1)
-        h = self.mlp(h)
-        h = h * gate_mlp.unsqueeze(1)
-        x = x + h
-        return x
-
-    def forward(self, x: torch.Tensor, mod: torch.Tensor, context: torch.Tensor):
-        if self.use_checkpoint:
-            return torch.utils.checkpoint.checkpoint(self._forward, x, mod, context, use_reentrant=False)
-        else:
-            return self._forward(x, mod, context)
-        

trellis/modules/utils.py

@@ -1,54 +0,0 @@
-import torch.nn as nn
-from ..modules import sparse as sp
-
-FP16_MODULES = (
-    nn.Conv1d,
-    nn.Conv2d,
-    nn.Conv3d,
-    nn.ConvTranspose1d,
-    nn.ConvTranspose2d,
-    nn.ConvTranspose3d,
-    nn.Linear,
-    sp.SparseConv3d,
-    sp.SparseInverseConv3d,
-    sp.SparseLinear,
-)
-
-def convert_module_to_f16(l):
-    """
-    Convert primitive modules to float16.
-    """
-    if isinstance(l, FP16_MODULES):
-        for p in l.parameters():
-            p.data = p.data.half()
-
-
-def convert_module_to_f32(l):
-    """
-    Convert primitive modules to float32, undoing convert_module_to_f16().
-    """
-    if isinstance(l, FP16_MODULES):
-        for p in l.parameters():
-            p.data = p.data.float()
-
-
-def zero_module(module):
-    """
-    Zero out the parameters of a module and return it.
-    """
-    for p in module.parameters():
-        p.detach().zero_()
-    return module
-
-
-def scale_module(module, scale):
-    """
-    Scale the parameters of a module and return it.
-    """
-    for p in module.parameters():
-        p.detach().mul_(scale)
-    return module
-
-
-def modulate(x, shift, scale):
-    return x * (1 + scale.unsqueeze(1)) + shift.unsqueeze(1)