anilegin's picture
Upload inference-only latent diffusion model
a04730e verified
Raw
History Blame Contribute Delete
4.48 kB
from __future__ import annotations
import torch
from torch import nn
import torch.nn.functional as F
from src.models.autoencoder.blocks import (
ResBlock,
Upsample,
MidBlock,
normalization,
)
class Decoder(nn.Module):
"""
VAE decoder.
Converts latent tensor back into image space.
Shape path:
[B, 4, 32, 32]
-> conv_in
[B, 512, 32, 32]
-> mid block
[B, 512, 32, 32]
-> upsample
[B, 512, 64, 64]
-> upsample
[B, 256, 128, 128]
-> upsample
[B, 128, 256, 256]
-> conv_out
[B, 3, 256, 256]
Output is in [-1, 1] because training images are normalized to [-1, 1].
"""
def __init__(
self,
out_channels: int = 3,
latent_channels: int = 4,
base_channels: int = 128,
channel_multipliers: list[int] | tuple[int, ...] = (1, 2, 4, 4),
num_res_blocks: int = 2,
dropout: float = 0.0,
use_attention: bool = True,
attention_heads: int = 1
):
super().__init__()
if len(channel_multipliers) < 2:
raise ValueError("channel_multipliers must contain at least 2 levels.")
self.out_channels = out_channels
self.latent_channels = latent_channels
self.base_channels = base_channels
self.channel_multipliers = list(channel_multipliers)
self.num_res_blocks = num_res_blocks
# Number of spatial upsampling operations
# Example:
# [1, 2, 4, 4] has 4 levels, so decoder upsamples 3 times:
# 32 -> 64 -> 128 -> 256
self.num_upsamples = len(self.channel_multipliers) - 1
# Start from the deepest encoder channel count
current_channels = base_channels * self.channel_multipliers[-1]
self.conv_in = nn.Conv2d(
latent_channels,
current_channels,
kernel_size=3,
stride=1,
padding=1,
)
# Bottleneck block at the lowest spatial resolution.
self.mid = MidBlock(
channels=current_channels,
dropout=dropout,
use_attention=use_attention,
num_heads=attention_heads,
)
self.up_blocks = nn.ModuleList()
reversed_multipliers = list(reversed(self.channel_multipliers))
for level, multiplier in enumerate(reversed_multipliers):
out_stage_channels = base_channels * multiplier
resblocks = nn.ModuleList()
# one extra ResBlock per level
for _ in range(num_res_blocks + 1):
resblocks.append(
ResBlock(
in_channels=current_channels,
out_channels=out_stage_channels,
dropout=dropout,
)
)
current_channels = out_stage_channels
# Upsample after every stage except the full-resolution
if level < len(reversed_multipliers) - 1:
upsample = Upsample(
channels=current_channels
)
else:
upsample = nn.Identity()
self.up_blocks.append(
nn.ModuleDict(
{
"resblocks": resblocks,
"upsample": upsample,
}
)
)
self.norm_out = normalization(current_channels)
self.conv_out = nn.Conv2d(
current_channels,
out_channels,
kernel_size=3,
stride=1,
padding=1,
)
def forward(self, z: torch.Tensor) -> torch.Tensor:
"""
Args:
z:
Latent tensor with shape [B, latent_channels, H/8, W/8].
For 256x256 images and downsample factor 8:
[B, latent_channels, 32, 32]
Returns:
x_recon:
Reconstructed image tensor with shape [B, 3, H, W].
Values are in [-1, 1].
"""
h = self.conv_in(z)
h = self.mid(h)
for stage in self.up_blocks:
for block in stage["resblocks"]:
h = block(h)
h = stage["upsample"](h)
h = self.norm_out(h)
h = F.silu(h)
h = self.conv_out(h)
x_recon = torch.tanh(h)
return x_recon