# -------------------------------------------------------- # TinyViT Model Architecture # Copyright (c) 2022 Microsoft # Adapted from LeViT and Swin Transformer # LeViT: (https://github.com/facebookresearch/levit) # Swin: (https://github.com/microsoft/swin-transformer) # Build the TinyViT Model # -------------------------------------------------------- import collections import itertools import math import warnings import torch import torch.nn as nn import torch.nn.functional as F import torch.utils.checkpoint as checkpoint from typing import Tuple def _ntuple(n): def parse(x): if isinstance(x, collections.abc.Iterable) and not isinstance(x, str): return x return tuple(itertools.repeat(x, n)) return parse to_2tuple = _ntuple(2) def _trunc_normal_(tensor, mean, std, a, b): # Cut & paste from PyTorch official master until it's in a few official releases - RW # Method based on https://people.sc.fsu.edu/~jburkardt/presentations/truncated_normal.pdf def norm_cdf(x): # Computes standard normal cumulative distribution function return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0 if (mean < a - 2 * std) or (mean > b + 2 * std): warnings.warn( "mean is more than 2 std from [a, b] in nn.init.trunc_normal_. " "The distribution of values may be incorrect.", stacklevel=2, ) # Values are generated by using a truncated uniform distribution and # then using the inverse CDF for the normal distribution. # Get upper and lower cdf values l = norm_cdf((a - mean) / std) u = norm_cdf((b - mean) / std) # Uniformly fill tensor with values from [l, u], then translate to # [2l-1, 2u-1]. tensor.uniform_(2 * l - 1, 2 * u - 1) # Use inverse cdf transform for normal distribution to get truncated # standard normal tensor.erfinv_() # Transform to proper mean, std tensor.mul_(std * math.sqrt(2.0)) tensor.add_(mean) # Clamp to ensure it's in the proper range tensor.clamp_(min=a, max=b) return tensor def trunc_normal_(tensor, mean=0.0, std=1.0, a=-2.0, b=2.0): # type: (Tensor, float, float, float, float) -> Tensor r"""Fills the input Tensor with values drawn from a truncated normal distribution. The values are effectively drawn from the normal distribution :math:`\mathcal{N}(\text{mean}, \text{std}^2)` with values outside :math:`[a, b]` redrawn until they are within the bounds. The method used for generating the random values works best when :math:`a \leq \text{mean} \leq b`. NOTE: this impl is similar to the PyTorch trunc_normal_, the bounds [a, b] are applied while sampling the normal with mean/std applied, therefore a, b args should be adjusted to match the range of mean, std args. Args: tensor: an n-dimensional `torch.Tensor` mean: the mean of the normal distribution std: the standard deviation of the normal distribution a: the minimum cutoff value b: the maximum cutoff value Examples: >>> w = torch.empty(3, 5) >>> nn.init.trunc_normal_(w) """ with torch.no_grad(): return _trunc_normal_(tensor, mean, std, a, b) def drop_path( x, drop_prob: float = 0.0, training: bool = False, scale_by_keep: bool = True ): """Drop paths (Stochastic Depth) per sample (when applied in main path of residual blocks). This is the same as the DropConnect impl I created for EfficientNet, etc networks, however, the original name is misleading as 'Drop Connect' is a different form of dropout in a separate paper... See discussion: https://github.com/tensorflow/tpu/issues/494#issuecomment-532968956 ... I've opted for changing the layer and argument names to 'drop path' rather than mix DropConnect as a layer name and use 'survival rate' as the argument. """ if drop_prob == 0.0 or not training: return x keep_prob = 1 - drop_prob shape = (x.shape[0],) + (1,) * ( x.ndim - 1 ) # work with diff dim tensors, not just 2D ConvNets random_tensor = x.new_empty(shape).bernoulli_(keep_prob) if keep_prob > 0.0 and scale_by_keep: random_tensor.div_(keep_prob) return x * random_tensor class TimmDropPath(nn.Module): """Drop paths (Stochastic Depth) per sample (when applied in main path of residual blocks).""" def __init__(self, drop_prob: float = 0.0, scale_by_keep: bool = True): super(TimmDropPath, self).__init__() self.drop_prob = drop_prob self.scale_by_keep = scale_by_keep def forward(self, x): return drop_path(x, self.drop_prob, self.training, self.scale_by_keep) def extra_repr(self): return f"drop_prob={round(self.drop_prob,3):0.3f}" class Conv2d_BN(torch.nn.Sequential): def __init__( self, a, b, ks=1, stride=1, pad=0, dilation=1, groups=1, bn_weight_init=1 ): super().__init__() self.add_module( "c", torch.nn.Conv2d(a, b, ks, stride, pad, dilation, groups, bias=False) ) bn = torch.nn.BatchNorm2d(b) torch.nn.init.constant_(bn.weight, bn_weight_init) torch.nn.init.constant_(bn.bias, 0) self.add_module("bn", bn) @torch.no_grad() def fuse(self): c, bn = self._modules.values() w = bn.weight / (bn.running_var + bn.eps) ** 0.5 w = c.weight * w[:, None, None, None] b = bn.bias - bn.running_mean * bn.weight / (bn.running_var + bn.eps) ** 0.5 m = torch.nn.Conv2d( w.size(1) * self.c.groups, w.size(0), w.shape[2:], stride=self.c.stride, padding=self.c.padding, dilation=self.c.dilation, groups=self.c.groups, ) m.weight.data.copy_(w) m.bias.data.copy_(b) return m class DropPath(TimmDropPath): def __init__(self, drop_prob=None): super().__init__(drop_prob=drop_prob) self.drop_prob = drop_prob def __repr__(self): msg = super().__repr__() msg += f"(drop_prob={self.drop_prob})" return msg class PatchEmbed(nn.Module): def __init__(self, in_chans, embed_dim, resolution, activation): super().__init__() img_size: Tuple[int, int] = to_2tuple(resolution) self.patches_resolution = (img_size[0] // 4, img_size[1] // 4) self.num_patches = self.patches_resolution[0] * self.patches_resolution[1] self.in_chans = in_chans self.embed_dim = embed_dim n = embed_dim self.seq = nn.Sequential( Conv2d_BN(in_chans, n // 2, 3, 2, 1), activation(), Conv2d_BN(n // 2, n, 3, 2, 1), ) def forward(self, x): return self.seq(x) class MBConv(nn.Module): def __init__(self, in_chans, out_chans, expand_ratio, activation, drop_path): super().__init__() self.in_chans = in_chans self.hidden_chans = int(in_chans * expand_ratio) self.out_chans = out_chans self.conv1 = Conv2d_BN(in_chans, self.hidden_chans, ks=1) self.act1 = activation() self.conv2 = Conv2d_BN( self.hidden_chans, self.hidden_chans, ks=3, stride=1, pad=1, groups=self.hidden_chans, ) self.act2 = activation() self.conv3 = Conv2d_BN(self.hidden_chans, out_chans, ks=1, bn_weight_init=0.0) self.act3 = activation() self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() def forward(self, x): shortcut = x x = self.conv1(x) x = self.act1(x) x = self.conv2(x) x = self.act2(x) x = self.conv3(x) x = self.drop_path(x) x += shortcut x = self.act3(x) return x class PatchMerging(nn.Module): def __init__(self, input_resolution, dim, out_dim, activation): super().__init__() self.input_resolution = input_resolution self.dim = dim self.out_dim = out_dim self.act = activation() self.conv1 = Conv2d_BN(dim, out_dim, 1, 1, 0) stride_c = 2 if out_dim == 320 or out_dim == 448 or out_dim == 576: stride_c = 1 self.conv2 = Conv2d_BN(out_dim, out_dim, 3, stride_c, 1, groups=out_dim) self.conv3 = Conv2d_BN(out_dim, out_dim, 1, 1, 0) def forward(self, x): if x.ndim == 3: H, W = self.input_resolution B = len(x) # (B, C, H, W) x = x.view(B, H, W, -1).permute(0, 3, 1, 2) x = self.conv1(x) x = self.act(x) x = self.conv2(x) x = self.act(x) x = self.conv3(x) x = x.flatten(2).transpose(1, 2) return x class ConvLayer(nn.Module): def __init__( self, dim, input_resolution, depth, activation, drop_path=0.0, downsample=None, use_checkpoint=False, out_dim=None, conv_expand_ratio=4.0, ): super().__init__() self.dim = dim self.input_resolution = input_resolution self.depth = depth self.use_checkpoint = use_checkpoint # build blocks self.blocks = nn.ModuleList( [ MBConv( dim, dim, conv_expand_ratio, activation, drop_path[i] if isinstance(drop_path, list) else drop_path, ) for i in range(depth) ] ) # patch merging layer if downsample is not None: self.downsample = downsample( input_resolution, dim=dim, out_dim=out_dim, activation=activation ) else: self.downsample = None def forward(self, x): for blk in self.blocks: if self.use_checkpoint: x = checkpoint.checkpoint(blk, x) else: x = blk(x) if self.downsample is not None: x = self.downsample(x) return x class Mlp(nn.Module): def __init__( self, in_features, hidden_features=None, out_features=None, act_layer=nn.GELU, drop=0.0, ): super().__init__() out_features = out_features or in_features hidden_features = hidden_features or in_features self.norm = nn.LayerNorm(in_features) self.fc1 = nn.Linear(in_features, hidden_features) self.fc2 = nn.Linear(hidden_features, out_features) self.act = act_layer() self.drop = nn.Dropout(drop) def forward(self, x): x = self.norm(x) x = self.fc1(x) x = self.act(x) x = self.drop(x) x = self.fc2(x) x = self.drop(x) return x class Attention(torch.nn.Module): def __init__( self, dim, key_dim, num_heads=8, attn_ratio=4, resolution=(14, 14), ): super().__init__() # (h, w) assert isinstance(resolution, tuple) and len(resolution) == 2 self.num_heads = num_heads self.scale = key_dim**-0.5 self.key_dim = key_dim self.nh_kd = nh_kd = key_dim * num_heads self.d = int(attn_ratio * key_dim) self.dh = int(attn_ratio * key_dim) * num_heads self.attn_ratio = attn_ratio h = self.dh + nh_kd * 2 self.norm = nn.LayerNorm(dim) self.qkv = nn.Linear(dim, h) self.proj = nn.Linear(self.dh, dim) points = list(itertools.product(range(resolution[0]), range(resolution[1]))) N = len(points) attention_offsets = {} idxs = [] for p1 in points: for p2 in points: offset = (abs(p1[0] - p2[0]), abs(p1[1] - p2[1])) if offset not in attention_offsets: attention_offsets[offset] = len(attention_offsets) idxs.append(attention_offsets[offset]) self.attention_biases = torch.nn.Parameter( torch.zeros(num_heads, len(attention_offsets)) ) self.register_buffer( "attention_bias_idxs", torch.LongTensor(idxs).view(N, N), persistent=False ) @torch.no_grad() def train(self, mode=True): super().train(mode) if mode and hasattr(self, "ab"): del self.ab else: self.register_buffer( "ab", self.attention_biases[:, self.attention_bias_idxs], persistent=False, ) def forward(self, x): # x (B,N,C) B, N, _ = x.shape # Normalization x = self.norm(x) qkv = self.qkv(x) # (B, N, num_heads, d) q, k, v = qkv.view(B, N, self.num_heads, -1).split( [self.key_dim, self.key_dim, self.d], dim=3 ) # (B, num_heads, N, d) q = q.permute(0, 2, 1, 3) k = k.permute(0, 2, 1, 3) v = v.permute(0, 2, 1, 3) attn = (q @ k.transpose(-2, -1)) * self.scale + ( self.attention_biases[:, self.attention_bias_idxs] if self.training else self.ab ) attn = attn.softmax(dim=-1) x = (attn @ v).transpose(1, 2).reshape(B, N, self.dh) x = self.proj(x) return x class TinyViTBlock(nn.Module): r"""TinyViT Block. Args: dim (int): Number of input channels. input_resolution (tuple[int, int]): Input resolution. num_heads (int): Number of attention heads. window_size (int): Window size. mlp_ratio (float): Ratio of mlp hidden dim to embedding dim. drop (float, optional): Dropout rate. Default: 0.0 drop_path (float, optional): Stochastic depth rate. Default: 0.0 local_conv_size (int): the kernel size of the convolution between Attention and MLP. Default: 3 activation: the activation function. Default: nn.GELU """ def __init__( self, dim, input_resolution, num_heads, window_size=7, mlp_ratio=4.0, drop=0.0, drop_path=0.0, local_conv_size=3, activation=nn.GELU, ): super().__init__() self.dim = dim self.input_resolution = input_resolution self.num_heads = num_heads assert window_size > 0, "window_size must be greater than 0" self.window_size = window_size self.mlp_ratio = mlp_ratio self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() assert dim % num_heads == 0, "dim must be divisible by num_heads" head_dim = dim // num_heads window_resolution = (window_size, window_size) self.attn = Attention( dim, head_dim, num_heads, attn_ratio=1, resolution=window_resolution ) mlp_hidden_dim = int(dim * mlp_ratio) mlp_activation = activation self.mlp = Mlp( in_features=dim, hidden_features=mlp_hidden_dim, act_layer=mlp_activation, drop=drop, ) pad = local_conv_size // 2 self.local_conv = Conv2d_BN( dim, dim, ks=local_conv_size, stride=1, pad=pad, groups=dim ) def forward(self, x): H, W = self.input_resolution B, L, C = x.shape assert L == H * W, "input feature has wrong size" res_x = x if H == self.window_size and W == self.window_size: x = self.attn(x) else: x = x.view(B, H, W, C) pad_b = (self.window_size - H % self.window_size) % self.window_size pad_r = (self.window_size - W % self.window_size) % self.window_size padding = pad_b > 0 or pad_r > 0 if padding: x = F.pad(x, (0, 0, 0, pad_r, 0, pad_b)) pH, pW = H + pad_b, W + pad_r nH = pH // self.window_size nW = pW // self.window_size # window partition x = ( x.view(B, nH, self.window_size, nW, self.window_size, C) .transpose(2, 3) .reshape(B * nH * nW, self.window_size * self.window_size, C) ) x = self.attn(x) # window reverse x = ( x.view(B, nH, nW, self.window_size, self.window_size, C) .transpose(2, 3) .reshape(B, pH, pW, C) ) if padding: x = x[:, :H, :W].contiguous() x = x.view(B, L, C) x = res_x + self.drop_path(x) x = x.transpose(1, 2).reshape(B, C, H, W) x = self.local_conv(x) x = x.view(B, C, L).transpose(1, 2) x = x + self.drop_path(self.mlp(x)) return x def extra_repr(self) -> str: return ( f"dim={self.dim}, input_resolution={self.input_resolution}, num_heads={self.num_heads}, " f"window_size={self.window_size}, mlp_ratio={self.mlp_ratio}" ) class BasicLayer(nn.Module): """A basic TinyViT layer for one stage. Args: dim (int): Number of input channels. input_resolution (tuple[int]): Input resolution. depth (int): Number of blocks. num_heads (int): Number of attention heads. window_size (int): Local window size. mlp_ratio (float): Ratio of mlp hidden dim to embedding dim. drop (float, optional): Dropout rate. Default: 0.0 drop_path (float | tuple[float], optional): Stochastic depth rate. Default: 0.0 downsample (nn.Module | None, optional): Downsample layer at the end of the layer. Default: None use_checkpoint (bool): Whether to use checkpointing to save memory. Default: False. local_conv_size: the kernel size of the depthwise convolution between attention and MLP. Default: 3 activation: the activation function. Default: nn.GELU out_dim: the output dimension of the layer. Default: dim """ def __init__( self, dim, input_resolution, depth, num_heads, window_size, mlp_ratio=4.0, drop=0.0, drop_path=0.0, downsample=None, use_checkpoint=False, local_conv_size=3, activation=nn.GELU, out_dim=None, ): super().__init__() self.dim = dim self.input_resolution = input_resolution self.depth = depth self.use_checkpoint = use_checkpoint # build blocks self.blocks = nn.ModuleList( [ TinyViTBlock( dim=dim, input_resolution=input_resolution, num_heads=num_heads, window_size=window_size, mlp_ratio=mlp_ratio, drop=drop, drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path, local_conv_size=local_conv_size, activation=activation, ) for i in range(depth) ] ) # patch merging layer if downsample is not None: self.downsample = downsample( input_resolution, dim=dim, out_dim=out_dim, activation=activation ) else: self.downsample = None def forward(self, x): for blk in self.blocks: if self.use_checkpoint: x = checkpoint.checkpoint(blk, x) else: x = blk(x) if self.downsample is not None: x = self.downsample(x) return x def extra_repr(self) -> str: return f"dim={self.dim}, input_resolution={self.input_resolution}, depth={self.depth}" class LayerNorm2d(nn.Module): def __init__(self, num_channels: int, eps: float = 1e-6) -> None: super().__init__() self.weight = nn.Parameter(torch.ones(num_channels)) self.bias = nn.Parameter(torch.zeros(num_channels)) self.eps = eps def forward(self, x: torch.Tensor) -> torch.Tensor: u = x.mean(1, keepdim=True) s = (x - u).pow(2).mean(1, keepdim=True) x = (x - u) / torch.sqrt(s + self.eps) x = self.weight[:, None, None] * x + self.bias[:, None, None] return x class TinyViT(nn.Module): def __init__( self, img_size=224, in_chans=3, num_classes=1000, embed_dims=[96, 192, 384, 768], depths=[2, 2, 6, 2], num_heads=[3, 6, 12, 24], window_sizes=[7, 7, 14, 7], mlp_ratio=4.0, drop_rate=0.0, drop_path_rate=0.1, use_checkpoint=False, mbconv_expand_ratio=4.0, local_conv_size=3, layer_lr_decay=1.0, ): super().__init__() self.img_size = img_size self.num_classes = num_classes self.depths = depths self.num_layers = len(depths) self.mlp_ratio = mlp_ratio activation = nn.GELU self.patch_embed = PatchEmbed( in_chans=in_chans, embed_dim=embed_dims[0], resolution=img_size, activation=activation, ) patches_resolution = self.patch_embed.patches_resolution self.patches_resolution = patches_resolution # stochastic depth dpr = [ x.item() for x in torch.linspace(0, drop_path_rate, sum(depths)) ] # stochastic depth decay rule # build layers self.layers = nn.ModuleList() for i_layer in range(self.num_layers): kwargs = dict( dim=embed_dims[i_layer], input_resolution=( patches_resolution[0] // (2 ** (i_layer - 1 if i_layer == 3 else i_layer)), patches_resolution[1] // (2 ** (i_layer - 1 if i_layer == 3 else i_layer)), ), # input_resolution=(patches_resolution[0] // (2 ** i_layer), # patches_resolution[1] // (2 ** i_layer)), depth=depths[i_layer], drop_path=dpr[sum(depths[:i_layer]) : sum(depths[: i_layer + 1])], downsample=PatchMerging if (i_layer < self.num_layers - 1) else None, use_checkpoint=use_checkpoint, out_dim=embed_dims[min(i_layer + 1, len(embed_dims) - 1)], activation=activation, ) if i_layer == 0: layer = ConvLayer( conv_expand_ratio=mbconv_expand_ratio, **kwargs, ) else: layer = BasicLayer( num_heads=num_heads[i_layer], window_size=window_sizes[i_layer], mlp_ratio=self.mlp_ratio, drop=drop_rate, local_conv_size=local_conv_size, **kwargs, ) self.layers.append(layer) # Classifier head self.norm_head = nn.LayerNorm(embed_dims[-1]) self.head = ( nn.Linear(embed_dims[-1], num_classes) if num_classes > 0 else torch.nn.Identity() ) # init weights self.apply(self._init_weights) self.set_layer_lr_decay(layer_lr_decay) self.neck = nn.Sequential( nn.Conv2d( embed_dims[-1], 256, kernel_size=1, bias=False, ), LayerNorm2d(256), nn.Conv2d( 256, 256, kernel_size=3, padding=1, bias=False, ), LayerNorm2d(256), ) def set_layer_lr_decay(self, layer_lr_decay): decay_rate = layer_lr_decay # layers -> blocks (depth) depth = sum(self.depths) lr_scales = [decay_rate ** (depth - i - 1) for i in range(depth)] # print("LR SCALES:", lr_scales) def _set_lr_scale(m, scale): for p in m.parameters(): p.lr_scale = scale self.patch_embed.apply(lambda x: _set_lr_scale(x, lr_scales[0])) i = 0 for layer in self.layers: for block in layer.blocks: block.apply(lambda x: _set_lr_scale(x, lr_scales[i])) i += 1 if layer.downsample is not None: layer.downsample.apply(lambda x: _set_lr_scale(x, lr_scales[i - 1])) assert i == depth for m in [self.norm_head, self.head]: m.apply(lambda x: _set_lr_scale(x, lr_scales[-1])) for k, p in self.named_parameters(): p.param_name = k def _check_lr_scale(m): for p in m.parameters(): assert hasattr(p, "lr_scale"), p.param_name self.apply(_check_lr_scale) def _init_weights(self, m): if isinstance(m, nn.Linear): trunc_normal_(m.weight, std=0.02) if isinstance(m, nn.Linear) and m.bias is not None: nn.init.constant_(m.bias, 0) elif isinstance(m, nn.LayerNorm): nn.init.constant_(m.bias, 0) nn.init.constant_(m.weight, 1.0) @torch.jit.ignore def no_weight_decay_keywords(self): return {"attention_biases"} def forward_features(self, x): # x: (N, C, H, W) x = self.patch_embed(x) x = self.layers[0](x) start_i = 1 for i in range(start_i, len(self.layers)): layer = self.layers[i] x = layer(x) B, _, C = x.size() x = x.view(B, 64, 64, C) x = x.permute(0, 3, 1, 2) x = self.neck(x) return x def forward(self, x): x = self.forward_features(x) # x = self.norm_head(x) # x = self.head(x) return x