E-RADIO / eradio_model.py
gheinrich's picture
Upload model
f134378
raw
history blame
53.9 kB
#!/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)