|
from typing import *
|
|
import torch
|
|
import torch.nn as nn
|
|
import torch.nn.functional as F
|
|
from ..modules.norm import GroupNorm32, ChannelLayerNorm32
|
|
from ..modules.spatial import pixel_shuffle_3d
|
|
from ..modules.utils import zero_module, convert_module_to_f16, convert_module_to_f32
|
|
|
|
|
|
def norm_layer(norm_type: str, *args, **kwargs) -> nn.Module:
|
|
"""
|
|
Return a normalization layer.
|
|
"""
|
|
if norm_type == "group":
|
|
return GroupNorm32(32, *args, **kwargs)
|
|
elif norm_type == "layer":
|
|
return ChannelLayerNorm32(*args, **kwargs)
|
|
else:
|
|
raise ValueError(f"Invalid norm type {norm_type}")
|
|
|
|
|
|
class ResBlock3d(nn.Module):
|
|
def __init__(
|
|
self,
|
|
channels: int,
|
|
out_channels: Optional[int] = None,
|
|
norm_type: Literal["group", "layer"] = "layer",
|
|
):
|
|
super().__init__()
|
|
self.channels = channels
|
|
self.out_channels = out_channels or channels
|
|
|
|
self.norm1 = norm_layer(norm_type, channels)
|
|
self.norm2 = norm_layer(norm_type, self.out_channels)
|
|
self.conv1 = nn.Conv3d(channels, self.out_channels, 3, padding=1)
|
|
self.conv2 = zero_module(nn.Conv3d(self.out_channels, self.out_channels, 3, padding=1))
|
|
self.skip_connection = nn.Conv3d(channels, self.out_channels, 1) if channels != self.out_channels else nn.Identity()
|
|
|
|
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
|
h = self.norm1(x)
|
|
h = F.silu(h)
|
|
h = self.conv1(h)
|
|
h = self.norm2(h)
|
|
h = F.silu(h)
|
|
h = self.conv2(h)
|
|
h = h + self.skip_connection(x)
|
|
return h
|
|
|
|
|
|
class DownsampleBlock3d(nn.Module):
|
|
def __init__(
|
|
self,
|
|
in_channels: int,
|
|
out_channels: int,
|
|
mode: Literal["conv", "avgpool"] = "conv",
|
|
):
|
|
assert mode in ["conv", "avgpool"], f"Invalid mode {mode}"
|
|
|
|
super().__init__()
|
|
self.in_channels = in_channels
|
|
self.out_channels = out_channels
|
|
|
|
if mode == "conv":
|
|
self.conv = nn.Conv3d(in_channels, out_channels, 2, stride=2)
|
|
elif mode == "avgpool":
|
|
assert in_channels == out_channels, "Pooling mode requires in_channels to be equal to out_channels"
|
|
|
|
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
|
if hasattr(self, "conv"):
|
|
return self.conv(x)
|
|
else:
|
|
return F.avg_pool3d(x, 2)
|
|
|
|
|
|
class UpsampleBlock3d(nn.Module):
|
|
def __init__(
|
|
self,
|
|
in_channels: int,
|
|
out_channels: int,
|
|
mode: Literal["conv", "nearest"] = "conv",
|
|
):
|
|
assert mode in ["conv", "nearest"], f"Invalid mode {mode}"
|
|
|
|
super().__init__()
|
|
self.in_channels = in_channels
|
|
self.out_channels = out_channels
|
|
|
|
if mode == "conv":
|
|
self.conv = nn.Conv3d(in_channels, out_channels*8, 3, padding=1)
|
|
elif mode == "nearest":
|
|
assert in_channels == out_channels, "Nearest mode requires in_channels to be equal to out_channels"
|
|
|
|
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
|
if hasattr(self, "conv"):
|
|
x = self.conv(x)
|
|
return pixel_shuffle_3d(x, 2)
|
|
else:
|
|
return F.interpolate(x, scale_factor=2, mode="nearest")
|
|
|
|
|
|
class SparseStructureEncoder(nn.Module):
|
|
"""
|
|
Encoder for Sparse Structure (\mathcal{E}_S in the paper Sec. 3.3).
|
|
|
|
Args:
|
|
in_channels (int): Channels of the input.
|
|
latent_channels (int): Channels of the latent representation.
|
|
num_res_blocks (int): Number of residual blocks at each resolution.
|
|
channels (List[int]): Channels of the encoder blocks.
|
|
num_res_blocks_middle (int): Number of residual blocks in the middle.
|
|
norm_type (Literal["group", "layer"]): Type of normalization layer.
|
|
use_fp16 (bool): Whether to use FP16.
|
|
"""
|
|
def __init__(
|
|
self,
|
|
in_channels: int,
|
|
latent_channels: int,
|
|
num_res_blocks: int,
|
|
channels: List[int],
|
|
num_res_blocks_middle: int = 2,
|
|
norm_type: Literal["group", "layer"] = "layer",
|
|
use_fp16: bool = False,
|
|
):
|
|
super().__init__()
|
|
self.in_channels = in_channels
|
|
self.latent_channels = latent_channels
|
|
self.num_res_blocks = num_res_blocks
|
|
self.channels = channels
|
|
self.num_res_blocks_middle = num_res_blocks_middle
|
|
self.norm_type = norm_type
|
|
self.use_fp16 = use_fp16
|
|
self.dtype = torch.float16 if use_fp16 else torch.float32
|
|
|
|
self.input_layer = nn.Conv3d(in_channels, channels[0], 3, padding=1)
|
|
|
|
self.blocks = nn.ModuleList([])
|
|
for i, ch in enumerate(channels):
|
|
self.blocks.extend([
|
|
ResBlock3d(ch, ch)
|
|
for _ in range(num_res_blocks)
|
|
])
|
|
if i < len(channels) - 1:
|
|
self.blocks.append(
|
|
DownsampleBlock3d(ch, channels[i+1])
|
|
)
|
|
|
|
self.middle_block = nn.Sequential(*[
|
|
ResBlock3d(channels[-1], channels[-1])
|
|
for _ in range(num_res_blocks_middle)
|
|
])
|
|
|
|
self.out_layer = nn.Sequential(
|
|
norm_layer(norm_type, channels[-1]),
|
|
nn.SiLU(),
|
|
nn.Conv3d(channels[-1], latent_channels*2, 3, padding=1)
|
|
)
|
|
|
|
if use_fp16:
|
|
self.convert_to_fp16()
|
|
|
|
@property
|
|
def device(self) -> torch.device:
|
|
"""
|
|
Return the device of the model.
|
|
"""
|
|
return next(self.parameters()).device
|
|
|
|
def convert_to_fp16(self) -> None:
|
|
"""
|
|
Convert the torso of the model to float16.
|
|
"""
|
|
self.use_fp16 = True
|
|
self.dtype = torch.float16
|
|
self.blocks.apply(convert_module_to_f16)
|
|
self.middle_block.apply(convert_module_to_f16)
|
|
|
|
def convert_to_fp32(self) -> None:
|
|
"""
|
|
Convert the torso of the model to float32.
|
|
"""
|
|
self.use_fp16 = False
|
|
self.dtype = torch.float32
|
|
self.blocks.apply(convert_module_to_f32)
|
|
self.middle_block.apply(convert_module_to_f32)
|
|
|
|
def forward(self, x: torch.Tensor, sample_posterior: bool = False, return_raw: bool = False) -> torch.Tensor:
|
|
h = self.input_layer(x)
|
|
h = h.type(self.dtype)
|
|
|
|
for block in self.blocks:
|
|
h = block(h)
|
|
h = self.middle_block(h)
|
|
|
|
h = h.type(x.dtype)
|
|
h = self.out_layer(h)
|
|
|
|
mean, logvar = h.chunk(2, dim=1)
|
|
|
|
if sample_posterior:
|
|
std = torch.exp(0.5 * logvar)
|
|
z = mean + std * torch.randn_like(std)
|
|
else:
|
|
z = mean
|
|
|
|
if return_raw:
|
|
return z, mean, logvar
|
|
return z
|
|
|
|
|
|
class SparseStructureDecoder(nn.Module):
|
|
"""
|
|
Decoder for Sparse Structure (\mathcal{D}_S in the paper Sec. 3.3).
|
|
|
|
Args:
|
|
out_channels (int): Channels of the output.
|
|
latent_channels (int): Channels of the latent representation.
|
|
num_res_blocks (int): Number of residual blocks at each resolution.
|
|
channels (List[int]): Channels of the decoder blocks.
|
|
num_res_blocks_middle (int): Number of residual blocks in the middle.
|
|
norm_type (Literal["group", "layer"]): Type of normalization layer.
|
|
use_fp16 (bool): Whether to use FP16.
|
|
"""
|
|
def __init__(
|
|
self,
|
|
out_channels: int,
|
|
latent_channels: int,
|
|
num_res_blocks: int,
|
|
channels: List[int],
|
|
num_res_blocks_middle: int = 2,
|
|
norm_type: Literal["group", "layer"] = "layer",
|
|
use_fp16: bool = False,
|
|
):
|
|
super().__init__()
|
|
self.out_channels = out_channels
|
|
self.latent_channels = latent_channels
|
|
self.num_res_blocks = num_res_blocks
|
|
self.channels = channels
|
|
self.num_res_blocks_middle = num_res_blocks_middle
|
|
self.norm_type = norm_type
|
|
self.use_fp16 = use_fp16
|
|
self.dtype = torch.float16 if use_fp16 else torch.float32
|
|
|
|
self.input_layer = nn.Conv3d(latent_channels, channels[0], 3, padding=1)
|
|
|
|
self.middle_block = nn.Sequential(*[
|
|
ResBlock3d(channels[0], channels[0])
|
|
for _ in range(num_res_blocks_middle)
|
|
])
|
|
|
|
self.blocks = nn.ModuleList([])
|
|
for i, ch in enumerate(channels):
|
|
self.blocks.extend([
|
|
ResBlock3d(ch, ch)
|
|
for _ in range(num_res_blocks)
|
|
])
|
|
if i < len(channels) - 1:
|
|
self.blocks.append(
|
|
UpsampleBlock3d(ch, channels[i+1])
|
|
)
|
|
|
|
self.out_layer = nn.Sequential(
|
|
norm_layer(norm_type, channels[-1]),
|
|
nn.SiLU(),
|
|
nn.Conv3d(channels[-1], out_channels, 3, padding=1)
|
|
)
|
|
|
|
if use_fp16:
|
|
self.convert_to_fp16()
|
|
|
|
@property
|
|
def device(self) -> torch.device:
|
|
"""
|
|
Return the device of the model.
|
|
"""
|
|
return next(self.parameters()).device
|
|
|
|
def convert_to_fp16(self) -> None:
|
|
"""
|
|
Convert the torso of the model to float16.
|
|
"""
|
|
self.use_fp16 = True
|
|
self.dtype = torch.float16
|
|
self.blocks.apply(convert_module_to_f16)
|
|
self.middle_block.apply(convert_module_to_f16)
|
|
|
|
def convert_to_fp32(self) -> None:
|
|
"""
|
|
Convert the torso of the model to float32.
|
|
"""
|
|
self.use_fp16 = False
|
|
self.dtype = torch.float32
|
|
self.blocks.apply(convert_module_to_f32)
|
|
self.middle_block.apply(convert_module_to_f32)
|
|
|
|
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
|
h = self.input_layer(x)
|
|
|
|
h = h.type(self.dtype)
|
|
|
|
h = self.middle_block(h)
|
|
for block in self.blocks:
|
|
h = block(h)
|
|
|
|
h = h.type(x.dtype)
|
|
h = self.out_layer(h)
|
|
return h
|
|
|
