models/control/controlnet.py

import torch
import torch.nn as nn
from typing import List, Optional, Union
from models.svd.sgm.util import default
from models.svd.sgm.modules.video_attention import SpatialVideoTransformer
from models.svd.sgm.modules.diffusionmodules.openaimodel import *
from models.diffusion.video_model import VideoResBlock, VideoUNet
from einops import repeat, rearrange
from models.svd.sgm.modules.diffusionmodules.wrappers import OpenAIWrapper


class Merger(nn.Module):
    """
    Merges the controlnet latents with the conditioning embedding (encoding of control frames).

    """

    def __init__(self, merge_mode: str = "addition", input_channels=0, frame_expansion="last_frame") -> None:
        super().__init__()
        self.merge_mode = merge_mode
        self.frame_expansion = frame_expansion

    def forward(self, x, condition_signal, num_video_frames, num_video_frames_conditional):
        x = rearrange(x, "(B F) C H W -> B F C H W", F=num_video_frames)

        condition_signal = rearrange(
            condition_signal, "(B F) C H W -> B F C H W", B=x.shape[0])

        if x.shape[1] - condition_signal.shape[1] > 0:
            if self.frame_expansion == "last_frame":
                fillup_latent = repeat(
                    condition_signal[:, -1], "B C H W -> B F C H W", F=x.shape[1] - condition_signal.shape[1])
            elif self.frame_expansion == "zero":
                fillup_latent = torch.zeros(
                    (x.shape[0], num_video_frames-num_video_frames_conditional, *x.shape[2:]), device=x.device, dtype=x.dtype)

            if self.frame_expansion != "none":
                condition_signal = torch.cat(
                    [condition_signal, fillup_latent], dim=1)

        if self.merge_mode == "addition":
            out = x + condition_signal
        else:
            raise NotImplementedError(
                f"Merging mode {self.merge_mode} not implemented.")

        out = rearrange(out, "B F C H W -> (B F) C H W")
        return out


class ControlNetConditioningEmbedding(nn.Module):
    """
    Quoting from https://arxiv.org/abs/2302.05543: "Stable Diffusion uses a pre-processing method similar to VQ-GAN
    [11] to convert the entire dataset of 512 × 512 images into smaller 64 × 64 “latent images” for stabilized
    training. This requires ControlNets to convert image-based conditions to 64 × 64 feature space to match the
    convolution size. We use a tiny network E(·) of four convolution layers with 4 × 4 kernels and 2 × 2 strides
    (activated by ReLU, channels are 16, 32, 64, 128, initialized with Gaussian weights, trained jointly with the full
    model) to encode image-space conditions ... into feature maps ..."
    """

    def __init__(
        self,
        conditioning_embedding_channels: int,
        conditioning_channels: int = 3,
        block_out_channels: Tuple[int] = (16, 32, 96, 256),
        downsample: bool = True,
        final_3d_conv: bool = False,
        zero_init: bool = True,
        use_controlnet_mask: bool = False,
        use_normalization: bool = False,
    ):
        super().__init__()

        self.final_3d_conv = final_3d_conv
        self.conv_in = nn.Conv2d(
            conditioning_channels, block_out_channels[0], kernel_size=3, padding=1)
        if final_3d_conv:
            print("USING 3D CONV in ControlNET")

        self.blocks = nn.ModuleList([])
        if use_normalization:
            self.norms = nn.ModuleList([])
        self.use_normalization = use_normalization

        stride = 2 if downsample else 1

        for i in range(len(block_out_channels) - 1):
            channel_in = block_out_channels[i]
            channel_out = block_out_channels[i + 1]
            self.blocks.append(
                nn.Conv2d(channel_in, channel_in, kernel_size=3, padding=1))
            if use_normalization:
                self.norms.append(nn.LayerNorm((channel_in)))
            self.blocks.append(
                nn.Conv2d(channel_in, channel_out, kernel_size=3, padding=1, stride=stride))
            if use_normalization:
                self.norms.append(nn.LayerNorm((channel_out)))

        self.conv_out = zero_module(
            nn.Conv2d(
                block_out_channels[-1]+int(use_controlnet_mask), conditioning_embedding_channels, kernel_size=3, padding=1), reset=zero_init
        )

    def forward(self, conditioning):
        embedding = self.conv_in(conditioning)
        embedding = F.silu(embedding)

        if self.use_normalization:
            for block, norm in zip(self.blocks, self.norms):
                embedding = block(embedding)
                embedding = rearrange(embedding, " ... C W H -> ... W H C")
                embedding = norm(embedding)
                embedding = rearrange(embedding, "... W H C -> ... C W H")
                embedding = F.silu(embedding)
        else:
            for block in self.blocks:
                embedding = block(embedding)
                embedding = F.silu(embedding)

        embedding = self.conv_out(embedding)
        return embedding


class ControlNet(nn.Module):

    def __init__(
        self,
        in_channels: int,
        model_channels: int,
        out_channels: int,
        num_res_blocks: int,
        attention_resolutions: Union[List[int], int],
        dropout: float = 0.0,
        channel_mult: List[int] = (1, 2, 4, 8),
        conv_resample: bool = True,
        dims: int = 2,
        num_classes: Optional[Union[int, str]] = None,
        use_checkpoint: bool = False,
        num_heads: int = -1,
        num_head_channels: int = -1,
        num_heads_upsample: int = -1,
        use_scale_shift_norm: bool = False,
        resblock_updown: bool = False,
        transformer_depth: Union[List[int], int] = 1,
        transformer_depth_middle: Optional[int] = None,
        context_dim: Optional[int] = None,
        time_downup: bool = False,
        time_context_dim: Optional[int] = None,
        extra_ff_mix_layer: bool = False,
        use_spatial_context: bool = False,
        merge_strategy: str = "fixed",
        merge_factor: float = 0.5,
        spatial_transformer_attn_type: str = "softmax",
        video_kernel_size: Union[int, List[int]] = 3,
        use_linear_in_transformer: bool = False,
        adm_in_channels: Optional[int] = None,
        disable_temporal_crossattention: bool = False,
        max_ddpm_temb_period: int = 10000,
        conditioning_embedding_out_channels: Optional[Tuple[int]] = (
            16, 32, 96, 256),
        condition_encoder: str = "",
        use_controlnet_mask: bool = False,
        downsample_controlnet_cond: bool = True,
        use_image_encoder_normalization: bool = False,
        zero_conv_mode: str = "Identity",
        frame_expansion: str = "none",
        merging_mode: str = "addition",
    ):
        super().__init__()
        assert zero_conv_mode == "Identity", "Zero convolution not implemented"

        assert context_dim is not None

        if num_heads_upsample == -1:
            num_heads_upsample = num_heads

        if num_heads == -1:
            assert num_head_channels != -1

        if num_head_channels == -1:
            assert num_heads != -1

        self.in_channels = in_channels
        self.model_channels = model_channels
        self.out_channels = out_channels
        if isinstance(transformer_depth, int):
            transformer_depth = len(channel_mult) * [transformer_depth]
        transformer_depth_middle = default(
            transformer_depth_middle, transformer_depth[-1]
        )

        self.num_res_blocks = num_res_blocks
        self.attention_resolutions = attention_resolutions
        self.dropout = dropout
        self.channel_mult = channel_mult
        self.conv_resample = conv_resample
        self.num_classes = num_classes
        self.use_checkpoint = use_checkpoint
        self.num_heads = num_heads
        self.num_head_channels = num_head_channels
        self.num_heads_upsample = num_heads_upsample
        self.dims = dims
        self.use_scale_shift_norm = use_scale_shift_norm
        self.resblock_updown = resblock_updown
        self.transformer_depth = transformer_depth
        self.transformer_depth_middle = transformer_depth_middle
        self.context_dim = context_dim
        self.time_downup = time_downup
        self.time_context_dim = time_context_dim
        self.extra_ff_mix_layer = extra_ff_mix_layer
        self.use_spatial_context = use_spatial_context
        self.merge_strategy = merge_strategy
        self.merge_factor = merge_factor
        self.spatial_transformer_attn_type = spatial_transformer_attn_type
        self.video_kernel_size = video_kernel_size
        self.use_linear_in_transformer = use_linear_in_transformer
        self.adm_in_channels = adm_in_channels
        self.disable_temporal_crossattention = disable_temporal_crossattention
        self.max_ddpm_temb_period = max_ddpm_temb_period

        time_embed_dim = model_channels * 4
        self.time_embed = nn.Sequential(
            linear(model_channels, time_embed_dim),
            nn.SiLU(),
            linear(time_embed_dim, time_embed_dim),
        )

        if self.num_classes is not None:
            if isinstance(self.num_classes, int):
                self.label_emb = nn.Embedding(num_classes, time_embed_dim)
            elif self.num_classes == "continuous":
                print("setting up linear c_adm embedding layer")
                self.label_emb = nn.Linear(1, time_embed_dim)
            elif self.num_classes == "timestep":
                self.label_emb = nn.Sequential(
                    Timestep(model_channels),
                    nn.Sequential(
                        linear(model_channels, time_embed_dim),
                        nn.SiLU(),
                        linear(time_embed_dim, time_embed_dim),
                    ),
                )

            elif self.num_classes == "sequential":
                assert adm_in_channels is not None
                self.label_emb = nn.Sequential(
                    nn.Sequential(
                        linear(adm_in_channels, time_embed_dim),
                        nn.SiLU(),
                        linear(time_embed_dim, time_embed_dim),
                    )
                )
            else:
                raise ValueError()

        self.input_blocks = nn.ModuleList(
            [
                TimestepEmbedSequential(
                    conv_nd(dims, in_channels, model_channels, 3, padding=1)
                )
            ]
        )
        self._feature_size = model_channels
        input_block_chans = [model_channels]
        ch = model_channels
        ds = 1

        def get_attention_layer(
            ch,
            num_heads,
            dim_head,
            depth=1,
            context_dim=None,
            use_checkpoint=False,
            disabled_sa=False,
        ):
            return SpatialVideoTransformer(
                ch,
                num_heads,
                dim_head,
                depth=depth,
                context_dim=context_dim,
                time_context_dim=time_context_dim,
                dropout=dropout,
                ff_in=extra_ff_mix_layer,
                use_spatial_context=use_spatial_context,
                merge_strategy=merge_strategy,
                merge_factor=merge_factor,
                checkpoint=use_checkpoint,
                use_linear=use_linear_in_transformer,
                attn_mode=spatial_transformer_attn_type,
                disable_self_attn=disabled_sa,
                disable_temporal_crossattention=disable_temporal_crossattention,
                max_time_embed_period=max_ddpm_temb_period,
            )

        def get_resblock(
            merge_factor,
            merge_strategy,
            video_kernel_size,
            ch,
            time_embed_dim,
            dropout,
            out_ch,
            dims,
            use_checkpoint,
            use_scale_shift_norm,
            down=False,
            up=False,
        ):
            return VideoResBlock(
                merge_factor=merge_factor,
                merge_strategy=merge_strategy,
                video_kernel_size=video_kernel_size,
                channels=ch,
                emb_channels=time_embed_dim,
                dropout=dropout,
                out_channels=out_ch,
                dims=dims,
                use_checkpoint=use_checkpoint,
                use_scale_shift_norm=use_scale_shift_norm,
                down=down,
                up=up,
            )

        for level, mult in enumerate(channel_mult):
            for _ in range(num_res_blocks):
                layers = [
                    get_resblock(
                        merge_factor=merge_factor,
                        merge_strategy=merge_strategy,
                        video_kernel_size=video_kernel_size,
                        ch=ch,
                        time_embed_dim=time_embed_dim,
                        dropout=dropout,
                        out_ch=mult * model_channels,
                        dims=dims,
                        use_checkpoint=use_checkpoint,
                        use_scale_shift_norm=use_scale_shift_norm,
                    )
                ]
                ch = mult * model_channels
                if ds in attention_resolutions:
                    if num_head_channels == -1:
                        dim_head = ch // num_heads
                    else:
                        num_heads = ch // num_head_channels
                        dim_head = num_head_channels

                    layers.append(
                        get_attention_layer(
                            ch,
                            num_heads,
                            dim_head,
                            depth=transformer_depth[level],
                            context_dim=context_dim,
                            use_checkpoint=use_checkpoint,
                            disabled_sa=False,
                        )
                    )
                self.input_blocks.append(TimestepEmbedSequential(*layers))
                self._feature_size += ch
                input_block_chans.append(ch)
            if level != len(channel_mult) - 1:
                ds *= 2
                out_ch = ch
                self.input_blocks.append(
                    TimestepEmbedSequential(
                        get_resblock(
                            merge_factor=merge_factor,
                            merge_strategy=merge_strategy,
                            video_kernel_size=video_kernel_size,
                            ch=ch,
                            time_embed_dim=time_embed_dim,
                            dropout=dropout,
                            out_ch=out_ch,
                            dims=dims,
                            use_checkpoint=use_checkpoint,
                            use_scale_shift_norm=use_scale_shift_norm,
                            down=True,
                        )
                        if resblock_updown
                        else Downsample(
                            ch,
                            conv_resample,
                            dims=dims,
                            out_channels=out_ch,
                            third_down=time_downup,
                        )
                    )
                )
                ch = out_ch
                input_block_chans.append(ch)

                self._feature_size += ch

        if num_head_channels == -1:
            dim_head = ch // num_heads
        else:
            num_heads = ch // num_head_channels
            dim_head = num_head_channels

        self.middle_block = TimestepEmbedSequential(
            get_resblock(
                merge_factor=merge_factor,
                merge_strategy=merge_strategy,
                video_kernel_size=video_kernel_size,
                ch=ch,
                time_embed_dim=time_embed_dim,
                out_ch=None,
                dropout=dropout,
                dims=dims,
                use_checkpoint=use_checkpoint,
                use_scale_shift_norm=use_scale_shift_norm,
            ),
            get_attention_layer(
                ch,
                num_heads,
                dim_head,
                depth=transformer_depth_middle,
                context_dim=context_dim,
                use_checkpoint=use_checkpoint,
            ),
            get_resblock(
                merge_factor=merge_factor,
                merge_strategy=merge_strategy,
                video_kernel_size=video_kernel_size,
                ch=ch,
                out_ch=None,
                time_embed_dim=time_embed_dim,
                dropout=dropout,
                dims=dims,
                use_checkpoint=use_checkpoint,
                use_scale_shift_norm=use_scale_shift_norm,
            ),
        )
        self._feature_size += ch

        self.merger = Merger(
            merge_mode=merging_mode, input_channels=model_channels, frame_expansion=frame_expansion)

        conditioning_channels = 3 if downsample_controlnet_cond else 4
        block_out_channels = (320, 640, 1280, 1280)

        self.controlnet_cond_embedding = ControlNetConditioningEmbedding(
            conditioning_embedding_channels=block_out_channels[0],
            conditioning_channels=conditioning_channels,
            block_out_channels=conditioning_embedding_out_channels,
            downsample=downsample_controlnet_cond,
            final_3d_conv=condition_encoder.endswith("3DConv"),
            use_controlnet_mask=use_controlnet_mask,
            use_normalization=use_image_encoder_normalization,
        )

    def forward(
        self,
        x: th.Tensor,
        timesteps: th.Tensor,
        controlnet_cond: th.Tensor,
        context: Optional[th.Tensor] = None,
        y: Optional[th.Tensor] = None,
        time_context: Optional[th.Tensor] = None,
        num_video_frames: Optional[int] = None,
        num_video_frames_conditional: Optional[int] = None,
        image_only_indicator: Optional[th.Tensor] = None,
    ):
        assert (y is not None) == (
            self.num_classes is not None
        ), "must specify y if and only if the model is class-conditional -> no, relax this TODO"
        hs = []
        t_emb = timestep_embedding(
            timesteps, self.model_channels, repeat_only=False).to(x.dtype)

        emb = self.time_embed(t_emb)

        # TODO restrict y to [:self.num_frames] (conditonal frames)

        if self.num_classes is not None:
            assert y.shape[0] == x.shape[0]
            emb = emb + self.label_emb(y)

        controlnet_cond = self.controlnet_cond_embedding(controlnet_cond)

        h = x
        for idx, module in enumerate(self.input_blocks):
            h = module(
                h,
                emb,
                context=context,
                image_only_indicator=image_only_indicator,
                time_context=time_context,
                num_video_frames=num_video_frames,
            )
            if idx == 0:
                h = self.merger(h, controlnet_cond, num_video_frames=num_video_frames,
                                num_video_frames_conditional=num_video_frames_conditional)

            hs.append(h)
        h = self.middle_block(
            h,
            emb,
            context=context,
            image_only_indicator=image_only_indicator,
            time_context=time_context,
            num_video_frames=num_video_frames,
        )

        # 5. Control net blocks

        down_block_res_samples = hs

        mid_block_res_sample = h

        return (down_block_res_samples, mid_block_res_sample)

    @classmethod
    def from_unet(cls,
                  model: OpenAIWrapper,
                  merging_mode: str = "addition",
                  zero_conv_mode: str = "Identity",
                  frame_expansion: str = "none",
                  downsample_controlnet_cond: bool = True,
                  use_image_encoder_normalization: bool = False,
                  use_controlnet_mask: bool = False,
                  condition_encoder: str = "",
                  conditioning_embedding_out_channels: List[int] = None,

                  ):

        unet: VideoUNet = model.diffusion_model

        controlnet = cls(in_channels=unet.in_channels,
                         model_channels=unet.model_channels,
                         out_channels=unet.out_channels,
                         num_res_blocks=unet.num_res_blocks,
                         attention_resolutions=unet.attention_resolutions,
                         dropout=unet.dropout,
                         channel_mult=unet.channel_mult,
                         conv_resample=unet.conv_resample,
                         dims=unet.dims,
                         num_classes=unet.num_classes,
                         use_checkpoint=unet.use_checkpoint,
                         num_heads=unet.num_heads,
                         num_head_channels=unet.num_head_channels,
                         num_heads_upsample=unet.num_heads_upsample,
                         use_scale_shift_norm=unet.use_scale_shift_norm,
                         resblock_updown=unet.resblock_updown,
                         transformer_depth=unet.transformer_depth,
                         transformer_depth_middle=unet.transformer_depth_middle,
                         context_dim=unet.context_dim,
                         time_downup=unet.time_downup,
                         time_context_dim=unet.time_context_dim,
                         extra_ff_mix_layer=unet.extra_ff_mix_layer,
                         use_spatial_context=unet.use_spatial_context,
                         merge_strategy=unet.merge_strategy,
                         merge_factor=unet.merge_factor,
                         spatial_transformer_attn_type=unet.spatial_transformer_attn_type,
                         video_kernel_size=unet.video_kernel_size,
                         use_linear_in_transformer=unet.use_linear_in_transformer,
                         adm_in_channels=unet.adm_in_channels,
                         disable_temporal_crossattention=unet.disable_temporal_crossattention,
                         max_ddpm_temb_period=unet.max_ddpm_temb_period,  # up to here unet params
                         merging_mode=merging_mode,
                         zero_conv_mode=zero_conv_mode,
                         frame_expansion=frame_expansion,
                         downsample_controlnet_cond=downsample_controlnet_cond,
                         use_image_encoder_normalization=use_image_encoder_normalization,
                         use_controlnet_mask=use_controlnet_mask,
                         condition_encoder=condition_encoder,
                         conditioning_embedding_out_channels=conditioning_embedding_out_channels,
                         )
        controlnet: ControlNet

        return controlnet


def zero_module(module, reset=True):
    if reset:
        for p in module.parameters():
            nn.init.zeros_(p)
    return module