#!/usr/bin/env python3 # Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved. # # NVIDIA CORPORATION and its licensors retain all intellectual property # and proprietary rights in and to this software, related documentation # and any modifications thereto. Any use, reproduction, disclosure or # distribution of this software and related documentation without an express # license agreement from NVIDIA CORPORATION is strictly prohibited. # Created by Pavlo Molchanov, LPR - DL Efficiency Research team # based on Fastervit1 from LPR import torch import torch.nn as nn from timm.models.registry import register_model from timm.models.layers import trunc_normal_, DropPath, LayerNorm2d import numpy as np import torch.nn.functional as F from .block import C2f TRT = False # should help for TRT import pickle global bias_indx bias_indx = -1 DEBUG = False def pixel_unshuffle(data, factor=2): # performs nn.PixelShuffle(factor) in reverse, torch has some bug for ONNX and TRT, so doing it manually B, C, H, W = data.shape return ( data.view(B, C, factor, H // factor, factor, W // factor) .permute(0, 1, 2, 4, 3, 5) .reshape(B, -1, H // factor, W // factor) ) class SwiGLU(nn.Module): # should be more advanced, but doesnt improve results so far def forward(self, x): x, gate = x.chunk(2, dim=-1) return F.silu(gate) * x def window_partition(x, window_size): """ Args: x: (B, C, H, W) window_size: window size Returns: windows - local window features (num_windows*B, window_size*window_size, C) (Hp, Wp) - the size of the padded image """ B, C, H, W = x.shape if window_size == 0 or (window_size == H and window_size == W): windows = x.flatten(2).transpose(1, 2) Hp, Wp = H, W else: pad_h = (window_size - H % window_size) % window_size pad_w = (window_size - W % window_size) % window_size if pad_h > 0 or pad_w > 0: x = F.pad(x, (0, pad_w, 0, pad_h, 0, 0, 0, 0)) Hp, Wp = H + pad_h, W + pad_w x = x.view(B, C, Hp // window_size, window_size, Wp // window_size, window_size) windows = x.permute(0, 2, 4, 3, 5, 1).reshape(-1, window_size * window_size, C) return windows, (Hp, Wp) class Conv2d_BN(nn.Module): """ Conv2d + BN layer with folding capability to speed up inference """ def __init__( self, a, b, kernel_size=1, stride=1, padding=0, dilation=1, groups=1, bn_weight_init=1, bias=False, ): super().__init__() self.conv = torch.nn.Conv2d( a, b, kernel_size, stride, padding, dilation, groups, bias=False ) if 1: self.bn = torch.nn.BatchNorm2d(b) torch.nn.init.constant_(self.bn.weight, bn_weight_init) torch.nn.init.constant_(self.bn.bias, 0) def forward(self, x): x = self.conv(x) x = self.bn(x) return x @torch.no_grad() def switch_to_deploy(self): # return 1 if not isinstance(self.bn, nn.Identity): c, bn = self.conv, self.bn 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 self.conv.weight.data.copy_(w) self.conv.bias = nn.Parameter(b) self.bn = nn.Identity() def window_reverse(windows, window_size, H, W, pad_hw): """ Args: windows: local window features (num_windows*B, window_size, window_size, C) window_size: Window size H: Height of image W: Width of image pad_w - a tuple of image passing used in windowing step Returns: x: (B, C, H, W) """ # print(f"window_reverse, windows.shape {windows.shape}") Hp, Wp = pad_hw if window_size == 0 or (window_size == H and window_size == W): B = int(windows.shape[0] / (Hp * Wp / window_size / window_size)) x = windows.transpose(1, 2).view(B, -1, H, W) else: B = int(windows.shape[0] / (Hp * Wp / window_size / window_size)) x = windows.view( B, Hp // window_size, Wp // window_size, window_size, window_size, -1 ) x = x.permute(0, 5, 1, 3, 2, 4).reshape(B, windows.shape[2], Hp, Wp) if Hp > H or Wp > W: x = x[:, :, :H, :W,].contiguous() return x class PosEmbMLPSwinv2D(nn.Module): def __init__( self, window_size, pretrained_window_size, num_heads, seq_length, no_log=False ): super().__init__() self.window_size = window_size self.num_heads = num_heads # mlp to generate continuous relative position bias self.cpb_mlp = nn.Sequential( nn.Linear(2, 512, bias=True), nn.ReLU(inplace=True), nn.Linear(512, num_heads, bias=False), ) # get relative_coords_table relative_coords_h = torch.arange( -(self.window_size[0] - 1), self.window_size[0], dtype=torch.float32 ) relative_coords_w = torch.arange( -(self.window_size[1] - 1), self.window_size[1], dtype=torch.float32 ) relative_coords_table = ( torch.stack(torch.meshgrid([relative_coords_h, relative_coords_w])) .permute(1, 2, 0) .contiguous() .unsqueeze(0) ) # 1, 2*Wh-1, 2*Ww-1, 2 if pretrained_window_size[0] > 0: relative_coords_table[:, :, :, 0] /= pretrained_window_size[0] - 1 relative_coords_table[:, :, :, 1] /= pretrained_window_size[1] - 1 else: relative_coords_table[:, :, :, 0] /= self.window_size[0] - 1 relative_coords_table[:, :, :, 1] /= self.window_size[1] - 1 if not no_log: relative_coords_table *= 8 # normalize to -8, 8 relative_coords_table = ( torch.sign(relative_coords_table) * torch.log2(torch.abs(relative_coords_table) + 1.0) / np.log2(8) ) self.register_buffer("relative_coords_table", relative_coords_table) # get pair-wise relative position index for each token inside the window coords_h = torch.arange(self.window_size[0]) coords_w = torch.arange(self.window_size[1]) coords = torch.stack(torch.meshgrid([coords_h, coords_w])) # 2, Wh, Ww coords_flatten = torch.flatten(coords, 1) # 2, Wh*Ww relative_coords = ( coords_flatten[:, :, None] - coords_flatten[:, None, :] ) # 2, Wh*Ww, Wh*Ww relative_coords = relative_coords.permute( 1, 2, 0 ).contiguous() # Wh*Ww, Wh*Ww, 2 relative_coords[:, :, 0] += self.window_size[0] - 1 # shift to start from 0 relative_coords[:, :, 1] += self.window_size[1] - 1 relative_coords[:, :, 0] *= 2 * self.window_size[1] - 1 relative_position_index = relative_coords.sum(-1) # Wh*Ww, Wh*Ww self.register_buffer("relative_position_index", relative_position_index) self.grid_exists = False self.deploy = False relative_bias = torch.zeros(1, num_heads, seq_length, seq_length) self.seq_length = seq_length self.register_buffer("relative_bias", relative_bias) # for EMA def switch_to_deploy(self): self.deploy = True self.grid_exists = True def forward(self, input_tensor): # for efficiency, we want this forward to be folded into a single operation (sum) # if resolution stays the same, then we dont need to recompute MLP layers # # to dynamically adjust patch size over the step # if not (input_tensor.shape[1:] == self.relative_bias.shape[1:]): # self.grid_exists = False if self.training: self.grid_exists = False if self.deploy and self.grid_exists: input_tensor += self.relative_bias return input_tensor if not self.grid_exists: self.grid_exists = True relative_position_bias_table = self.cpb_mlp( self.relative_coords_table ).view(-1, self.num_heads) relative_position_bias = relative_position_bias_table[ self.relative_position_index.view(-1) ].view( self.window_size[0] * self.window_size[1], self.window_size[0] * self.window_size[1], -1, ) # Wh*Ww,Wh*Ww,nH relative_position_bias = relative_position_bias.permute( 2, 0, 1 ).contiguous() # nH, Wh*Ww, Wh*Ww relative_position_bias = 16 * torch.sigmoid(relative_position_bias) self.relative_bias = relative_position_bias.unsqueeze(0) input_tensor += self.relative_bias return input_tensor class GRAAttentionBlock(nn.Module): def __init__( self, window_size, dim_in, dim_out, num_heads, drop_path=0.0, qk_scale=None, qkv_bias=False, norm_layer=nn.LayerNorm, layer_scale=None, use_swiglu=True, subsample_ratio=1, dim_ratio=1, conv_base=False, do_windowing=True, multi_query=False, ) -> None: super().__init__() dim = dim_in # conv_base = True SHUFFLE = True SHUFFLE = False self.do_windowing = do_windowing if do_windowing: if SHUFFLE: self.downsample_op = ( torch.nn.PixelUnshuffle(subsample_ratio) if subsample_ratio > 1 else torch.nn.Identity() ) self.downsample_mixer = ( nn.Conv2d( dim_in * (subsample_ratio * subsample_ratio), dim_in * (dim_ratio), kernel_size=1, stride=1, padding=0, bias=False, ) if dim * dim_ratio != dim * subsample_ratio * subsample_ratio else torch.nn.Identity() ) else: if conv_base: self.downsample_op = ( nn.Conv2d( dim_in, dim_out, kernel_size=subsample_ratio, stride=subsample_ratio, ) if subsample_ratio > 1 else nn.Identity() ) self.downsample_mixer = nn.Identity() else: self.downsample_op = ( nn.AvgPool2d( kernel_size=subsample_ratio, stride=subsample_ratio ) if subsample_ratio > 1 else nn.Identity() ) self.downsample_mixer = ( Conv2d_BN(dim_in, dim_out, kernel_size=1, stride=1) if subsample_ratio > 1 else nn.Identity() ) if do_windowing: if SHUFFLE: self.upsample_mixer = ( nn.Conv2d( dim_in * dim_ratio, dim_in * (subsample_ratio * subsample_ratio), kernel_size=1, stride=1, padding=0, bias=False, ) if dim * dim_ratio != dim * subsample_ratio * subsample_ratio else torch.nn.Identity() ) self.upsample_op = ( torch.nn.PixelShuffle(subsample_ratio) if subsample_ratio > 1 else torch.nn.Identity() ) else: if conv_base: self.upsample_mixer = nn.Identity() self.upsample_op = ( nn.ConvTranspose2d( dim_in, dim_out, kernel_size=subsample_ratio, stride=subsample_ratio, ) if subsample_ratio > 1 else nn.Identity() ) else: self.upsample_mixer = ( nn.Upsample(scale_factor=subsample_ratio, mode="nearest") if subsample_ratio > 1 else nn.Identity() ) self.upsample_op = ( Conv2d_BN( dim_in, dim_out, kernel_size=1, stride=1, padding=0, bias=False, ) if subsample_ratio > 1 else nn.Identity() ) self.window_size = window_size self.norm1 = norm_layer(dim_in) if DEBUG: print( f"GRAAttentionBlock: input_resolution: , window_size: {window_size}, dim_in: {dim_in}, dim_out: {dim_out}, num_heads: {num_heads}, drop_path: {drop_path}, qk_scale: {qk_scale}, qkv_bias: {qkv_bias}, layer_scale: {layer_scale}" ) self.attn = WindowAttention( dim_in, num_heads=num_heads, qkv_bias=qkv_bias, qk_scale=qk_scale, resolution=window_size, seq_length=window_size ** 2, dim_out=dim_in, multi_query=multi_query, ) if DEBUG: print( f"Attention: dim_in: {dim_in}, num_heads: {num_heads}, qkv_bias: {qkv_bias}, qk_scale: {qk_scale}, resolution: {window_size}, seq_length: {window_size**2}, dim_out: {dim_in}" ) print(f"drop_path: {drop_path}, layer_scale: {layer_scale}") self.drop_path1 = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() use_layer_scale = layer_scale is not None and type(layer_scale) in [int, float] self.gamma1 = ( nn.Parameter(layer_scale * torch.ones(dim_in)) if use_layer_scale else 1 ) ### mlp layer mlp_ratio = 4 self.norm2 = norm_layer(dim_in) mlp_hidden_dim = int(dim_in * mlp_ratio) activation = nn.GELU if not use_swiglu else SwiGLU mlp_hidden_dim = ( int((4 * dim_in * 1 / 2) / 64) * 64 if use_swiglu else mlp_hidden_dim ) self.mlp = Mlp( in_features=dim_in, hidden_features=mlp_hidden_dim, act_layer=activation, use_swiglu=use_swiglu, ) self.gamma2 = ( nn.Parameter(layer_scale * torch.ones(dim_in)) if layer_scale else 1 ) self.drop_path2 = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() if DEBUG: print( f"MLP layer: dim_in: {dim_in}, dim_out: {dim_in}, mlp_hidden_dim: {mlp_hidden_dim}" ) print(f"drop_path: {drop_path}, layer_scale: {layer_scale}") def forward(self, x): skip_connection = x if self.do_windowing: # performing windowing if required x = self.downsample_op(x) x = self.downsample_mixer(x) if self.window_size > 0: H, W = x.shape[2], x.shape[3] x, pad_hw = window_partition(x, self.window_size) # window attention x = x + self.drop_path1(self.gamma1 * self.attn(self.norm1(x))) # mlp layer x = x + self.drop_path2(self.gamma2 * self.mlp(self.norm2(x))) if self.do_windowing: if self.window_size > 0: x = window_reverse(x, self.window_size, H, W, pad_hw) x = self.upsample_mixer(x) x = self.upsample_op(x) if ( x.shape[2] != skip_connection.shape[2] or x.shape[3] != skip_connection.shape[3] ): x = torch.nn.functional.pad( x, ( 0, -x.shape[3] + skip_connection.shape[3], 0, -x.shape[2] + skip_connection.shape[2], ), ) # need to add skip connection because downsampling and upsampling will break residual connection # 0.5 is needed to make sure that the skip connection is not too strong # in case of no downsample / upsample we can show that 0.5 compensates for the residual connection x = 0.5 * x + 0.5 * skip_connection return x class MultiResolutionAttention(nn.Module): """ MultiResolutionAttention (MRA) module The idea is to use multiple attention blocks with different resolution Feature maps are downsampled / upsampled for each attention block on different blocks Every attention block supports """ def __init__( self, window_size, sr_ratio, dim, dim_ratio, num_heads, do_windowing=True, layer_scale=1e-5, norm_layer=nn.LayerNorm, drop_path=0, qkv_bias=False, qk_scale=1.0, use_swiglu=True, multi_query=False, conv_base=False, ) -> None: """ Args: input_resolution: input image resolution window_size: window size compression_ratio: compression ratio max_depth: maximum depth of the GRA module """ super().__init__() depth = len(sr_ratio) self.attention_blocks = nn.ModuleList() for i in range(depth): subsample_ratio = sr_ratio[i] if len(window_size) > i: window_size_local = window_size[i] else: window_size_local = window_size[0] self.attention_blocks.append( GRAAttentionBlock( window_size=window_size_local, dim_in=dim, dim_out=dim, num_heads=num_heads, qkv_bias=qkv_bias, qk_scale=qk_scale, norm_layer=norm_layer, layer_scale=layer_scale, drop_path=drop_path, use_swiglu=use_swiglu, subsample_ratio=subsample_ratio, dim_ratio=dim_ratio, do_windowing=do_windowing, multi_query=multi_query, conv_base=conv_base, ), ) def forward(self, x): for attention_block in self.attention_blocks: x = attention_block(x) return x class Mlp(nn.Module): """ Multi-Layer Perceptron (MLP) block """ def __init__( self, in_features, hidden_features=None, out_features=None, act_layer=nn.GELU, use_swiglu=True, drop=0.0, ): """ Args: in_features: input features dimension. hidden_features: hidden features dimension. out_features: output features dimension. act_layer: activation function. drop: dropout rate. """ super().__init__() out_features = out_features or in_features hidden_features = hidden_features or in_features self.fc1 = nn.Linear( in_features, hidden_features * (2 if use_swiglu else 1), bias=False ) self.act = act_layer() self.fc2 = nn.Linear(hidden_features, out_features, bias=False) # self.drop = GaussianDropout(drop) def forward(self, x): x_size = x.size() x = x.view(-1, x_size[-1]) x = self.fc1(x) x = self.act(x) # x = self.drop(x) x = self.fc2(x) # x = self.drop(x) x = x.view(x_size) return x class Downsample(nn.Module): """ Down-sampling block Pixel Unshuffle is used for down-sampling, works great accuracy - wise but takes 10% more TRT time """ def __init__( self, dim, shuffle=False, ): """ Args: dim: feature size dimension. shuffle: idea with keep_dim: bool argument for maintaining the resolution. """ super().__init__() dim_out = 2 * dim if shuffle: self.norm = lambda x: pixel_unshuffle(x, factor=2) self.reduction = Conv2d_BN(dim * 4, dim_out, 1, 1, 0, bias=False) else: # removed layer norm for better, in this formulation we are getting 10% better speed # LayerNorm for high resolution inputs will be a pain as it pools over the entire spatial dimension self.norm = nn.Identity() self.reduction = Conv2d_BN(dim, dim_out, 3, 2, 1, bias=False) def forward(self, x): x = self.norm(x) x = self.reduction(x) return x class PatchEmbed(nn.Module): """ Patch embedding block """ def __init__(self, in_chans=3, in_dim=64, dim=96, shuffle_down=False): """ Args: in_chans: number of input channels. in_dim: intermediate feature size dimension to speed up stem. dim: final stem channel number shuffle_down: use PixelUnshuffle for down-sampling, effectively increases the receptive field """ super().__init__() # shuffle_down = False if not shuffle_down: self.proj = nn.Identity() self.conv_down = nn.Sequential( Conv2d_BN(in_chans, in_dim, 3, 2, 1, bias=False), nn.ReLU(), Conv2d_BN(in_dim, dim, 3, 2, 1, bias=False), nn.ReLU(), ) else: self.proj = lambda x: pixel_unshuffle(x, factor=4) # self.conv_down = nn.Sequential(Conv2d_BN(in_chans*16, in_dim, 3, 1, 1), # nn.SiLU(), # Conv2d_BN(in_dim, dim, 3, 1, 1), # nn.SiLU(), # ) self.conv_down = nn.Sequential( Conv2d_BN(in_chans * 16, dim, 3, 1, 1), nn.ReLU(), ) def forward(self, x): x = self.proj(x) x = self.conv_down(x) return x class ConvBlock(nn.Module): """ Convolutional block, used in first couple of stages Experimented with plan resnet-18 like modules, they are the best in terms of throughput Experimented with RepVGG, dont see significant improvement in accuracy Finally, YOLOv8 idea seem to work fine (resnet-18 like block with squeezed feature dimension, and feature concatendation at the end) """ def __init__( self, dim, drop_path=0.0, layer_scale=None, kernel_size=3, rep_vgg=False ): super().__init__() self.rep_vgg = rep_vgg if not rep_vgg: self.conv1 = Conv2d_BN( dim, dim, kernel_size=kernel_size, stride=1, padding=1 ) self.act1 = nn.GELU() else: self.conv1 = RepVGGBlock( dim, dim, kernel_size=kernel_size, stride=1, padding=1, groups=1 ) if not rep_vgg: self.conv2 = Conv2d_BN( dim, dim, kernel_size=kernel_size, stride=1, padding=1 ) else: self.conv2 = RepVGGBlock( dim, dim, kernel_size=kernel_size, stride=1, padding=1, groups=1 ) self.layer_scale = layer_scale if layer_scale is not None and type(layer_scale) in [int, float]: self.gamma = nn.Parameter(layer_scale * torch.ones(dim)) self.layer_scale = True else: self.layer_scale = False self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() def forward(self, x): input = x if not self.rep_vgg: x = self.conv1(x) x = self.act1(x) x = self.conv2(x) else: x = self.conv1(x) x = self.conv2(x) if self.layer_scale: x = x * self.gamma.view(1, -1, 1, 1) x = input + self.drop_path(x) return x class WindowAttention(nn.Module): # Windowed Attention from SwinV2 # use a MLP trick to deal with various input image resolutions, then fold it to improve speed # tested multi-querry attention, but it is not as good as full attention: # look into palm: https://github.com/lucidrains/PaLM-pytorch/blob/main/palm_pytorch/palm_pytorch.py # single kv attention, mlp in parallel (didnt improve speed) def __init__( self, dim, num_heads=8, qkv_bias=False, qk_scale=None, resolution=0, seq_length=0, dim_out=None, multi_query=False, ): # taken from EdgeViT and tweaked with attention bias. super().__init__() if not dim_out: dim_out = dim self.multi_query = multi_query self.num_heads = num_heads head_dim = dim // num_heads self.head_dim = dim // num_heads self.dim_internal = dim self.scale = qk_scale or head_dim ** -0.5 if not multi_query: if TRT: self.q = nn.Linear(dim, dim, bias=qkv_bias) self.k = nn.Linear(dim, dim, bias=qkv_bias) self.v = nn.Linear(dim, dim, bias=qkv_bias) else: self.qkv = nn.Linear(dim, dim * 3, bias=qkv_bias) else: self.qkv = nn.Linear(dim, dim + 2 * self.head_dim, bias=qkv_bias) self.proj = nn.Linear(dim, dim_out, bias=False) # attention positional bias self.pos_emb_funct = PosEmbMLPSwinv2D( window_size=[resolution, resolution], pretrained_window_size=[resolution, resolution], num_heads=num_heads, seq_length=seq_length, ) self.resolution = resolution def forward(self, x): B, N, C = x.shape if not self.multi_query: if TRT: q = ( self.q(x) .reshape(B, -1, self.num_heads, C // self.num_heads) .permute(0, 2, 1, 3) ) k = ( self.k(x) .reshape(B, -1, self.num_heads, C // self.num_heads) .permute(0, 2, 1, 3) ) v = ( self.v(x) .reshape(B, -1, self.num_heads, C // self.num_heads) .permute(0, 2, 1, 3) ) else: qkv = ( self.qkv(x) .reshape(B, -1, 3, self.num_heads, C // self.num_heads) .permute(2, 0, 3, 1, 4) ) q, k, v = qkv[0], qkv[1], qkv[2] else: qkv = self.qkv(x) (q, k, v) = qkv.split( [self.dim_internal, self.head_dim, self.head_dim], dim=2 ) q = q.reshape(B, -1, self.num_heads, C // self.num_heads).permute( 0, 2, 1, 3 ) k = k.reshape(B, -1, 1, C // self.num_heads).permute(0, 2, 1, 3) v = v.reshape(B, -1, 1, C // self.num_heads).permute(0, 2, 1, 3) attn = (q @ k.transpose(-2, -1)) * self.scale attn = self.pos_emb_funct(attn) attn = attn.softmax(dim=-1) x = (attn @ v).transpose(1, 2).reshape(B, -1, C) x = self.proj(x) return x class FasterViTLayer(nn.Module): """ fastervitlayer """ def __init__( self, dim, depth, num_heads, window_size, conv=False, downsample=True, mlp_ratio=4.0, qkv_bias=False, qk_scale=None, norm_layer=nn.LayerNorm, drop_path=0.0, layer_scale=None, layer_scale_conv=None, sr_dim_ratio=1, sr_ratio=1, multi_query=False, use_swiglu=True, rep_vgg=False, yolo_arch=False, downsample_shuffle=False, conv_base=False, ): """ Args: dim: feature size dimension. depth: number of layers in each stage. input_resolution: input image resolution. window_size: window size in each stage. downsample: bool argument for down-sampling. mlp_ratio: MLP ratio. num_heads: number of heads in each stage. qkv_bias: bool argument for query, key, value learnable bias. qk_scale: bool argument to scaling query, key. drop: dropout rate. attn_drop: attention dropout rate. drop_path: drop path rate. norm_layer: normalization layer. layer_scale: layer scaling coefficient. """ super().__init__() self.conv = conv self.yolo_arch = False if conv: if not yolo_arch: self.blocks = nn.ModuleList( [ ConvBlock( dim=dim, drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path, layer_scale=layer_scale_conv, rep_vgg=rep_vgg, ) for i in range(depth) ] ) else: self.blocks = C2f(dim, dim, n=depth, shortcut=True, e=0.5) self.yolo_arch = True else: if not isinstance(window_size, list): window_size = [window_size] self.window_size = window_size[0] self.do_single_windowing = True if not isinstance(sr_ratio, list): sr_ratio = [sr_ratio] if any([sr != 1 for sr in sr_ratio]) or len(set(window_size)) > 1: self.do_single_windowing = False do_windowing = True else: self.do_single_windowing = True do_windowing = False self.blocks = nn.ModuleList() for i in range(depth): self.blocks.append( MultiResolutionAttention( window_size=window_size, sr_ratio=sr_ratio, dim=dim, dim_ratio=sr_dim_ratio, num_heads=num_heads, norm_layer=norm_layer, drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path, layer_scale=layer_scale, qkv_bias=qkv_bias, qk_scale=qk_scale, use_swiglu=use_swiglu, do_windowing=do_windowing, multi_query=multi_query, conv_base=conv_base, ) ) self.transformer = not conv self.downsample = ( None if not downsample else Downsample(dim=dim, shuffle=downsample_shuffle) ) def forward(self, x): B, C, H, W = x.shape if self.transformer and self.do_single_windowing: H, W = x.shape[2], x.shape[3] x, pad_hw = window_partition(x, self.window_size) if not self.yolo_arch: for bn, blk in enumerate(self.blocks): x = blk(x) else: x = self.blocks(x) if self.transformer and self.do_single_windowing: x = window_reverse(x, self.window_size, H, W, pad_hw) if self.downsample is None: return x, x return self.downsample(x), x # changing to output pre downsampled features class FasterViT(nn.Module): """ FasterViT """ def __init__( self, dim, in_dim, depths, window_size, mlp_ratio, num_heads, drop_path_rate=0.2, in_chans=3, num_classes=1000, qkv_bias=False, qk_scale=None, layer_scale=None, layer_scale_conv=None, layer_norm_last=False, sr_ratio=[1, 1, 1, 1], max_depth=-1, conv_base=False, use_swiglu=False, multi_query=False, norm_layer=nn.LayerNorm, rep_vgg=False, drop_uniform=False, yolo_arch=False, shuffle_down=False, downsample_shuffle=False, return_full_features=False, full_features_head_dim=128, neck_start_stage=1, use_neck=False, **kwargs, ): """ Args: dim: feature size dimension. depths: number of layers in each stage. window_size: window size in each stage. mlp_ratio: MLP ratio. num_heads: number of heads in each stage. drop_path_rate: drop path rate. in_chans: number of input channels. num_classes: number of classes. qkv_bias: bool argument for query, key, value learnable bias. qk_scale: bool argument to scaling query, key. drop_rate: dropout rate. attn_drop_rate: attention dropout rate. norm_layer: normalization layer. layer_scale: layer scaling coefficient. return_full_features: output dense features as well as logits full_features_head_dim: number of channels in the dense features head neck_start_stage: a stage id to start full feature neck. Model has 4 stages, indix starts with 0 for 224 resolution, the output of the stage before downsample: stage 0: 56x56, stage 1: 28x28, stage 2: 14x14, stage 3: 7x7 use_neck: even for summarization embedding use neck """ super().__init__() num_features = int(dim * 2 ** (len(depths) - 1)) self.num_classes = num_classes self.patch_embed = PatchEmbed( in_chans=in_chans, in_dim=in_dim, dim=dim, shuffle_down=shuffle_down ) # set return_full_features true if we want to return full features from all stages self.return_full_features = return_full_features self.use_neck = use_neck dpr = [x.item() for x in torch.linspace(0, drop_path_rate, sum(depths))] if drop_uniform: dpr = [drop_path_rate for x in range(sum(depths))] if not isinstance(max_depth, list): max_depth = [max_depth] * len(depths) self.levels = nn.ModuleList() for i in range(len(depths)): conv = True if (i == 0 or i == 1) else False level = FasterViTLayer( dim=int(dim * 2 ** i), depth=depths[i], num_heads=num_heads[i], window_size=window_size[i], mlp_ratio=mlp_ratio, qkv_bias=qkv_bias, qk_scale=qk_scale, conv=conv, drop_path=dpr[sum(depths[:i]) : sum(depths[: i + 1])], downsample=(i < 3), layer_scale=layer_scale, layer_scale_conv=layer_scale_conv, sr_ratio=sr_ratio[i], use_swiglu=use_swiglu, multi_query=multi_query, norm_layer=norm_layer, rep_vgg=rep_vgg, yolo_arch=yolo_arch, downsample_shuffle=downsample_shuffle, conv_base=conv_base, ) self.levels.append(level) if self.return_full_features or self.use_neck: # create feature projection layers for segmentation output self.neck_features_proj = nn.ModuleList() self.neck_start_stage = neck_start_stage upsample_ratio = 1 for i in range(len(depths)): level_n_features_output = int(dim * 2 ** i) if self.neck_start_stage > i: continue if ( upsample_ratio > 1 ) or full_features_head_dim != level_n_features_output: feature_projection = nn.Sequential() # feature_projection.add_module("norm",LayerNorm2d(level_n_features_output)) #slow, but better if 0: # Train: 0 [1900/10009 ( 19%)] Loss: 6.113 (6.57) Time: 0.548s, 233.40/s (0.549s, 233.04/s) LR: 1.000e-05 Data: 0.015 (0.013) feature_projection.add_module( "norm", nn.BatchNorm2d(level_n_features_output) ) # fast, but worse feature_projection.add_module( "dconv", nn.ConvTranspose2d( level_n_features_output, full_features_head_dim, kernel_size=upsample_ratio, stride=upsample_ratio, ), ) else: # pixel shuffle based upsampling # Train: 0 [1950/10009 ( 19%)] Loss: 6.190 (6.55) Time: 0.540s, 236.85/s (0.548s, 233.38/s) LR: 1.000e-05 Data: 0.015 (0.013) feature_projection.add_module( "norm", nn.BatchNorm2d(level_n_features_output) ) # fast, but worse feature_projection.add_module( "conv", nn.Conv2d( level_n_features_output, full_features_head_dim * upsample_ratio * upsample_ratio, kernel_size=1, stride=1, ), ) feature_projection.add_module( "upsample_pixelshuffle", nn.PixelShuffle(upsample_ratio) ) else: feature_projection = nn.Sequential() feature_projection.add_module( "norm", nn.BatchNorm2d(level_n_features_output) ) self.neck_features_proj.append(feature_projection) if i > 0 and self.levels[i - 1].downsample is not None: upsample_ratio *= 2 num_features = ( full_features_head_dim if (self.return_full_features or self.use_neck) else num_features ) self.num_features = num_features self.norm = ( LayerNorm2d(num_features) if layer_norm_last else nn.BatchNorm2d(num_features) ) self.avgpool = nn.AdaptiveAvgPool2d(1) self.head = ( nn.Linear(num_features, num_classes) if num_classes > 0 else nn.Identity() ) self.apply(self._init_weights) # pass 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) elif isinstance(m, LayerNorm2d): nn.init.constant_(m.bias, 0) nn.init.constant_(m.weight, 1.0) elif isinstance(m, nn.BatchNorm2d): nn.init.ones_(m.weight) nn.init.zeros_(m.bias) @torch.jit.ignore def no_weight_decay_keywords(self): return {"rpb"} def forward_features(self, x): x = self.patch_embed(x) full_features = None for il, level in enumerate(self.levels): x, pre_downsample_x = level(x) if self.return_full_features or self.use_neck: if self.neck_start_stage > il: continue if full_features is None: full_features = self.neck_features_proj[il - self.neck_start_stage]( pre_downsample_x ) else: # upsample torch tensor x to match full_features size, and add to full_features feature_projection = self.neck_features_proj[ il - self.neck_start_stage ](pre_downsample_x) if ( feature_projection.shape[2] != full_features.shape[2] or feature_projection.shape[3] != full_features.shape[3] ): feature_projection = torch.nn.functional.pad( feature_projection, ( 0, -feature_projection.shape[3] + full_features.shape[3], 0, -feature_projection.shape[2] + full_features.shape[2], ), ) full_features += feature_projection # x = self.norm(full_features if (self.return_full_features or self.use_neck) else x) x = self.norm(x) # new version for x = self.avgpool(x) x = torch.flatten(x, 1) if not self.return_full_features: return x, None return x, full_features def forward(self, x): x, full_features = self.forward_features(x) x = self.head(x) if full_features is not None: return x, full_features return x def switch_to_deploy(self): """ A method to perform model self-compression merges BN into conv layers converts MLP relative positional bias into precomputed buffers """ for level in [self.patch_embed, self.levels, self.head]: for module in level.modules(): if hasattr(module, "switch_to_deploy"): module.switch_to_deploy() @register_model def fastervit2_small(pretrained=False, **kwargs): # , model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=96, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [1, 2], 1], use_swiglu=False, downsample_shuffle=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_tiny(pretrained=False, **kwargs): # , model = FasterViT( depths=[1, 3, 4, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=80, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, downsample_shuffle=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base_fullres1(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=1024, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base_fullres2(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=512, neck_start_stage=1, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base_fullres3(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=256, neck_start_stage=1, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base_fullres4(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=256, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_base_fullres5(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=512, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model # pyt: 1934, 4202 TRT @register_model def fastervit2_large(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128 + 64, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_large_fullres(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[None, None, [7, 7], 7], dim=192, in_dim=64, mlp_ratio=4, drop_path_rate=0.0, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=1536, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_large_fullres_ws8(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[None, None, [8, 8], 8], dim=192, in_dim=64, mlp_ratio=4, drop_path_rate=0.0, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=1536, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_large_fullres_ws16(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[None, None, [16, 16], 16], dim=192, in_dim=64, mlp_ratio=4, drop_path_rate=0.0, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=1536, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_large_fullres_ws32(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[None, None, [32, 32], 32], dim=192, in_dim=64, mlp_ratio=4, drop_path_rate=0.0, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, conv_base=True, use_neck=True, full_features_head_dim=1536, neck_start_stage=2, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model # pyt: 897 @register_model def fastervit2_xlarge(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128 + 128 + 64, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model # pyt: @register_model def fastervit2_huge(pretrained=False, **kwargs): model = FasterViT( depths=[3, 3, 5, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=128 + 128 + 128 + 64, in_dim=64, mlp_ratio=4, drop_path_rate=0.2, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_xtiny(pretrained=False, **kwargs): # , model = FasterViT( depths=[1, 3, 4, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=64, in_dim=64, mlp_ratio=4, drop_path_rate=0.1, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, downsample_shuffle=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_xxtiny_5(pretrained=False, **kwargs): # , model = FasterViT( depths=[1, 3, 4, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=48, in_dim=64, mlp_ratio=4, drop_path_rate=0.05, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, downsample_shuffle=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def fastervit2_xxxtiny(pretrained=False, **kwargs): # , model = FasterViT( depths=[1, 3, 4, 5], num_heads=[2, 4, 8, 16], window_size=[8, 8, [7, 7], 7], dim=32, in_dim=32, mlp_ratio=4, drop_path_rate=0.0, sr_ratio=[1, 1, [2, 1], 1], use_swiglu=False, downsample_shuffle=False, yolo_arch=True, shuffle_down=False, **kwargs, ) if pretrained: model.load_state_dict(torch.load(pretrained)) return model @register_model def eradio(pretrained=False, **kwargs): return fastervit2_large_fullres_ws16(pretrained=pretrained, **kwargs)