archs/Restormer.py

## Restormer: Efficient Transformer for High-Resolution Image Restoration
## Syed Waqas Zamir, Aditya Arora, Salman Khan, Munawar Hayat, Fahad Shahbaz Khan, and Ming-Hsuan Yang
## https://arxiv.org/abs/2111.09881


import torch
import torch.nn as nn
import torch.nn.functional as F
from pdb import set_trace as stx
import numbers

from einops import rearrange



##########################################################################
## Layer Norm

def to_3d(x):
    return rearrange(x, 'b c h w -> b (h w) c')

def to_4d(x,h,w):
    return rearrange(x, 'b (h w) c -> b c h w',h=h,w=w)

class BiasFree_LayerNorm(nn.Module):
    def __init__(self, normalized_shape):
        super(BiasFree_LayerNorm, self).__init__()
        if isinstance(normalized_shape, numbers.Integral):
            normalized_shape = (normalized_shape,)
        normalized_shape = torch.Size(normalized_shape)

        assert len(normalized_shape) == 1

        self.weight = nn.Parameter(torch.ones(normalized_shape))
        self.normalized_shape = normalized_shape

    def forward(self, x):
        sigma = x.var(-1, keepdim=True, unbiased=False)
        return x / torch.sqrt(sigma+1e-5) * self.weight

class WithBias_LayerNorm(nn.Module):
    def __init__(self, normalized_shape):
        super(WithBias_LayerNorm, self).__init__()
        if isinstance(normalized_shape, numbers.Integral):
            normalized_shape = (normalized_shape,)
        normalized_shape = torch.Size(normalized_shape)

        assert len(normalized_shape) == 1

        self.weight = nn.Parameter(torch.ones(normalized_shape))
        self.bias = nn.Parameter(torch.zeros(normalized_shape))
        self.normalized_shape = normalized_shape

    def forward(self, x):
        mu = x.mean(-1, keepdim=True)
        sigma = x.var(-1, keepdim=True, unbiased=False)
        return (x - mu) / torch.sqrt(sigma+1e-5) * self.weight + self.bias


class LayerNorm(nn.Module):
    def __init__(self, nf, LayerNorm_type):
        super(LayerNorm, self).__init__()
        if LayerNorm_type =='BiasFree':
            self.body = BiasFree_LayerNorm(nf)
        else:
            self.body = WithBias_LayerNorm(nf)

    def forward(self, x):
        h, w = x.shape[-2:]
        return to_4d(self.body(to_3d(x)), h, w)



##########################################################################
## Gated-Dconv Feed-Forward Network (GDFN)
class FeedForward(nn.Module):
    def __init__(self, nf, ffn_expansion_factor, bias):
        super(FeedForward, self).__init__()

        hidden_features = int(nf*ffn_expansion_factor)

        self.project_in = nn.Conv2d(nf, hidden_features*2, kernel_size=1, bias=bias)

        self.dwconv = nn.Conv2d(hidden_features*2, hidden_features*2, kernel_size=3, stride=1, padding=1, groups=hidden_features*2, bias=bias)

        self.project_out = nn.Conv2d(hidden_features, nf, kernel_size=1, bias=bias)

    def forward(self, x):
        x = self.project_in(x)
        x1, x2 = self.dwconv(x).chunk(2, dim=1)
        x = F.gelu(x1) * x2
        x = self.project_out(x)
        return x



##########################################################################
## Multi-DConv Head Transposed Self-Attention (MDTA)
class Attention(nn.Module):
    def __init__(self, nf, num_heads, bias):
        super(Attention, self).__init__()
        self.num_heads = num_heads
        self.temperature = nn.Parameter(torch.ones(num_heads, 1, 1))

        self.qkv = nn.Conv2d(nf, nf*3, kernel_size=1, bias=bias)
        self.qkv_dwconv = nn.Conv2d(nf*3, nf*3, kernel_size=3, stride=1, padding=1, groups=nf*3, bias=bias)
        self.project_out = nn.Conv2d(nf, nf, kernel_size=1, bias=bias)
        
        self.emb_layers = nn.Sequential(
            nn.Conv2d(1, nf, 1, 1, 0, bias=True),
            nn.SiLU(inplace=True),
            nn.Conv2d(nf, nf, 1, 1, 0, bias=True),
            nn.SiLU(inplace=True),
            nn.Conv2d(nf, nf*3*2, 1, 1, 0, bias=True)
        )


    def forward(self, x, t):
        """
        x: [B, C, H, W]
        t: [B, 1, 1, 1] 或 [B, 1, H, W] 的时间信号
        """
        b,c,h,w = x.shape

        qkv = self.qkv_dwconv(self.qkv(x))
        if len(t.shape) > 0 and t.shape[-1] != 1:
            t = F.interpolate(t, size=x.shape[2:], mode='bilinear', align_corners=False)
        scale, shift = self.emb_layers(t).chunk(2, dim=1)
        qkv = qkv * (1+scale) + shift
        q,k,v = qkv.chunk(3, dim=1)   
        
        q = rearrange(q, 'b (head c) h w -> b head c (h w)', head=self.num_heads)
        k = rearrange(k, 'b (head c) h w -> b head c (h w)', head=self.num_heads)
        v = rearrange(v, 'b (head c) h w -> b head c (h w)', head=self.num_heads)

        q = torch.nn.functional.normalize(q, dim=-1)
        k = torch.nn.functional.normalize(k, dim=-1)

        attn = (q @ k.transpose(-2, -1)) * self.temperature
        attn = attn.softmax(dim=-1)

        out = (attn @ v)
        
        out = rearrange(out, 'b head c (h w) -> b (head c) h w', head=self.num_heads, h=h, w=w)

        out = self.project_out(out)
        return out



##########################################################################
class TransformerBlock(nn.Module):
    def __init__(self, nf, num_heads, ffn_expansion_factor, bias, LayerNorm_type):
        super(TransformerBlock, self).__init__()

        self.norm1 = LayerNorm(nf, LayerNorm_type)
        self.attn = Attention(nf, num_heads, bias)
        self.norm2 = LayerNorm(nf, LayerNorm_type)
        self.ffn = FeedForward(nf, ffn_expansion_factor, bias)

    def forward(self, x, t):
        x = x + self.attn(self.norm1(x), t)
        x = x + self.ffn(self.norm2(x))

        return x



##########################################################################
## Overlapped image patch embedding with 3x3 Conv
class OverlapPatchEmbed(nn.Module):
    def __init__(self, in_c=3, embed_dim=48, bias=False):
        super(OverlapPatchEmbed, self).__init__()

        self.proj = nn.Conv2d(in_c, embed_dim, kernel_size=3, stride=1, padding=1, bias=bias)

    def forward(self, x):
        x = self.proj(x)

        return x



##########################################################################
## Resizing modules
class Downsample(nn.Module):
    def __init__(self, n_feat):
        super(Downsample, self).__init__()

        self.body = nn.Sequential(nn.Conv2d(n_feat, n_feat//2, kernel_size=3, stride=1, padding=1, bias=False),
                                  nn.PixelUnshuffle(2))

    def forward(self, x):
        return self.body(x)

class Upsample(nn.Module):
    def __init__(self, n_feat):
        super(Upsample, self).__init__()

        self.body = nn.Sequential(nn.Conv2d(n_feat, n_feat*2, kernel_size=3, stride=1, padding=1, bias=False),
                                  nn.PixelShuffle(2))

    def forward(self, x):
        return self.body(x)

##########################################################################
##---------- Restormer -----------------------
class Restormer(nn.Module):
    def __init__(self, args={}):
        super().__init__()
        self.args = args
        
        # 设置默认参数
        self.in_nc = in_nc = self.args.get('in_nc', 3)
        self.out_nc = out_nc = self.args.get('out_nc', 3)
        self.nf = nf = self.args.get('nf', 48)
        self.nb = nb = self.args.get('nb', [4, 6, 6, 8])
        self.num_refinement_blocks = num_refinement_blocks = self.args.get('num_refinement_blocks', 4)
        self.heads = heads = self.args.get('heads', [1, 2, 4, 8])
        self.ffn_expansion_factor = ffn_expansion_factor = self.args.get('ffn_expansion_factor', 2.66)
        self.bias = bias = self.args.get('bias', False)
        self.LayerNorm_type = LayerNorm_type = self.args.get('LayerNorm_type', 'WithBias')
        self.nframes = nframes = self.args.get('nframes', 1)
        self.res = self.args.get('res', True)
        self.rectified = self.args.get('rectified', False)
        self.vpred = self.args.get('vpred', False)

        self.patch_embed = OverlapPatchEmbed(in_nc*nframes, nf)

        # Encoder blocks
        self.encoder_level1 = nn.Sequential(*[
            TransformerBlock(
                nf=nf,
                num_heads=heads[0],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[0])
        ])

        # Downsample Level 1 → Level 2
        self.down1_2 = Downsample(nf)
        self.encoder_level2 = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**1),
                num_heads=heads[1],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[1])
        ])

        # Downsample Level 2 → Level 3
        self.down2_3 = Downsample(int(nf * 2**1))
        self.encoder_level3 = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**2),
                num_heads=heads[2],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[2])
        ])

        # Downsample Level 3 → Level 4
        self.down3_4 = Downsample(int(nf * 2**2))
        self.latent = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**3),
                num_heads=heads[3],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[3])
        ])

        # Decoder blocks
        # Upsample Level 4 → Level 3
        self.up4_3 = Upsample(int(nf * 2**3))
        self.reduce_chan_level3 = nn.Conv2d(
            int(nf * 2**3),
            int(nf * 2**2),
            kernel_size=1,
            bias=bias
        )
        self.decoder_level3 = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**2),
                num_heads=heads[2],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[2])
        ])

        # Upsample Level 3 → Level 2
        self.up3_2 = Upsample(int(nf * 2**2))
        self.reduce_chan_level2 = nn.Conv2d(
            int(nf * 2**2),
            int(nf * 2**1),
            kernel_size=1,
            bias=bias
        )
        self.decoder_level2 = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**1),
                num_heads=heads[1],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[1])
        ])

        # Upsample Level 2 → Level 1
        self.up2_1 = Upsample(int(nf * 2**1))
        self.decoder_level1 = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**1),
                num_heads=heads[0],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(nb[0])
        ])

        # Refinement and output
        self.refinement = nn.Sequential(*[
            TransformerBlock(
                nf=int(nf * 2**1),
                num_heads=heads[0],
                ffn_expansion_factor=ffn_expansion_factor,
                bias=bias,
                LayerNorm_type=LayerNorm_type
            ) for i in range(num_refinement_blocks)
        ])

        self.output = nn.Conv2d(
            int(nf * 2**1),
            out_nc,
            kernel_size=3,
            stride=1,
            padding=1,
            bias=bias
        )

    def layer_by_layer(self, modules, x, t):
        # 手动处理需要t的模块
        for layer in modules:
            if isinstance(layer, TransformerBlock):
                x = layer(x, t)  # 传递t
            else:
                x = layer(x)
        return x

    def forward(self, inp_img, t):
        if self.rectified:
            t = t / (1 + t)# sigma->t
            inp_img = inp_img * (1-t)

        # Encoder
        inp_enc_level1 = self.patch_embed(inp_img)
        out_enc_level1 = self.layer_by_layer(self.encoder_level1, inp_enc_level1, t)
        
        inp_enc_level2 = self.down1_2(out_enc_level1)
        out_enc_level2 = self.layer_by_layer(self.encoder_level2, inp_enc_level2, t)

        inp_enc_level3 = self.down2_3(out_enc_level2)
        out_enc_level3 = self.layer_by_layer(self.encoder_level3, inp_enc_level3, t)

        inp_enc_level4 = self.down3_4(out_enc_level3)        
        latent = self.layer_by_layer(self.latent, inp_enc_level4, t)
        
        # Decoder
        inp_dec_level3 = self.up4_3(latent)
        inp_dec_level3 = torch.cat([inp_dec_level3, out_enc_level3], 1)
        inp_dec_level3 = self.reduce_chan_level3(inp_dec_level3)
        out_dec_level3 = self.layer_by_layer(self.decoder_level3, inp_dec_level3, t)

        inp_dec_level2 = self.up3_2(out_dec_level3)
        inp_dec_level2 = torch.cat([inp_dec_level2, out_enc_level2], 1)
        inp_dec_level2 = self.reduce_chan_level2(inp_dec_level2)
        out_dec_level2 = self.layer_by_layer(self.decoder_level2, inp_dec_level2, t)

        inp_dec_level1 = self.up2_1(out_dec_level2)
        inp_dec_level1 = torch.cat([inp_dec_level1, out_enc_level1], 1)
        out_dec_level1 = self.layer_by_layer(self.decoder_level1, inp_dec_level1, t)
        
        # Refinement
        out_dec_level1 = self.layer_by_layer(self.refinement, out_dec_level1, t)

        out = self.output(out_dec_level1)

        if not self.vpred:
            if self.res: out += inp_img
            if self.rectified: out /= (1-t)

        return out