dnbcd-busuclm / models /vmunet /LGF_VSS.py
congdanh99's picture
Upload models/vmunet/LGF_VSS.py with huggingface_hub
67c5be7 verified
Raw
History Blame Contribute Delete
24.7 kB
import time
import math
from functools import partial
from typing import Optional, Callable
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.utils.checkpoint as checkpoint
from einops import rearrange, repeat
from timm.models.layers import DropPath, to_2tuple, trunc_normal_
try:
from mamba_ssm.ops.selective_scan_interface import selective_scan_fn, selective_scan_ref
except:
pass
try:
from selective_scan import selective_scan_fn as selective_scan_fn_v1
from selective_scan import selective_scan_ref as selective_scan_ref_v1
except:
pass
DropPath.__repr__ = lambda self: f"timm.DropPath({self.drop_prob})"
class PatchEmbed2D(nn.Module):
def __init__(self, patch_size=4, in_chans=3, embed_dim=128, norm_layer=None, **kwargs):
super().__init__()
if isinstance(patch_size, int):
patch_size = (patch_size, patch_size)
self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=patch_size)
if norm_layer is not None:
self.norm = norm_layer(embed_dim)
else:
self.norm = None
def forward(self, x):
x = self.proj(x).permute(0, 2, 3, 1)
if self.norm is not None:
x = self.norm(x)
return x
class PatchMerging2D(nn.Module):
def __init__(self, dim, norm_layer=nn.LayerNorm):
super().__init__()
self.down = nn.Sequential(
Permute(0,3,1,2),
nn.PixelUnshuffle(2),
nn.Conv2d(in_channels=4*dim, out_channels=2*dim, kernel_size=1),
Permute(0,2,3,1),
nn.LayerNorm(2*dim)
)
def forward(self, x):
return self.down(x)
class DownSample(nn.Module):
def __init__(self, in_channels, out_channels, scale):
super().__init__()
self.conv = nn.Conv2d(in_channels, out_channels, 3, scale, 1, bias=False)
def forward(self, x):
x = self.conv(x.permute(0, 3, 1, 2))
return x.permute(0, 2, 3, 1)
class PatchExpand2D(nn.Module):
def __init__(self, dim, dim_scale=2, norm_layer=nn.LayerNorm):
super().__init__()
self.dim = dim * 2
self.dim_scale = dim_scale
self.expand = nn.Linear(self.dim, dim_scale*self.dim, bias=False)
self.norm = norm_layer(self.dim // dim_scale)
def forward(self, x):
B, H, W, C = x.shape
x = self.expand(x)
x = rearrange(x, 'b h w (p1 p2 c)-> b (h p1) (w p2) c', p1=self.dim_scale, p2=self.dim_scale, c=C//self.dim_scale)
x= self.norm(x)
return x
class UpSample(nn.Module):
def __init__(self, in_channels, out_channels, scale_factory):
super().__init__()
self.scale_factory = scale_factory
self.conv = nn.Conv2d(in_channels, out_channels, 3, 1, 1, bias=False)
def forward(self, x):
x = self.conv(x.permute(0, 3, 1, 2))
x = nn.functional.interpolate(x, scale_factor=self.scale_factory, mode='bilinear')
return x.permute(0, 2, 3, 1)
class Final_Expand(nn.Module):
def __init__(self, dim, dim_scale=4, norm_layer=nn.LayerNorm):
super().__init__()
self.dim = dim
self.dim_scale = dim_scale
self.expand = nn.Linear(self.dim, dim_scale*self.dim, bias=False)
self.norm = norm_layer(self.dim // dim_scale)
def forward(self, x):
B, H, W, C = x.shape
x = self.expand(x)
x = rearrange(x, 'b h w (p1 p2 c)-> b (h p1) (w p2) c', p1=self.dim_scale, p2=self.dim_scale, c=C//self.dim_scale)
x= self.norm(x)
return x
class Permute(nn.Module):
def __init__(self, *dims):
super().__init__()
self.dims = dims
def forward(self, x):
return x.permute(*self.dims)
class SS2D(nn.Module):
def __init__(
self,
d_model,
d_state=16,
d_conv=3,
expand=2,
dt_rank="auto",
dt_min=0.001,
dt_max=0.1,
dt_init="random",
dt_scale=1.0,
dt_init_floor=1e-4,
dropout=0.,
conv_bias=True,
bias=False,
device=None,
dtype=None,
drop_path_rate = 0.1,
**kwargs,
):
factory_kwargs = {"device": device, "dtype": dtype}
super().__init__()
self.d_model = d_model
self.d_state = d_state
self.d_inner = self.d_model
self.dt_rank = math.ceil(self.d_model / 16) if dt_rank == "auto" else dt_rank
self.conv2d = nn.Conv2d(
in_channels=self.d_inner,
out_channels=self.d_inner,
groups=self.d_inner,
bias=conv_bias,
kernel_size=d_conv,
padding=(d_conv - 1) // 2,
**factory_kwargs,
)
self.act = nn.SiLU()
self.conv1x1 = nn.Sequential(
nn.Conv2d(self.d_inner, self.d_inner, 1, padding=0, groups=self.d_inner),
Permute(0,2,3,1),
nn.LayerNorm(self.d_inner),
Permute(0,3,1,2),
nn.SiLU()
)
self.conv3x3 = nn.Sequential(
nn.Conv2d(self.d_inner, self.d_inner, 3, padding=1, groups=self.d_inner),
Permute(0,2,3,1),
nn.LayerNorm(self.d_inner),
Permute(0,3,1,2),
nn.SiLU()
)
self.conv5x5 = nn.Sequential(
nn.Conv2d(self.d_inner, self.d_inner, 5, padding=2, groups=self.d_inner),
Permute(0,2,3,1),
nn.LayerNorm(self.d_inner),
Permute(0,3,1,2),
nn.SiLU()
)
self.gate = nn.Sequential(
nn.AdaptiveAvgPool2d(1),
nn.Dropout(0.1),
Permute(0,2,3,1),
nn.Linear(self.d_inner, self.d_inner),
nn.SiLU(),
Permute(0,3,1,2),
nn.Conv2d(self.d_inner, 4*self.d_inner, 1),
nn.Softmax(dim=1)
)
self.x_proj = (
nn.Linear(self.d_inner, (self.dt_rank + self.d_state * 2), bias=False, **factory_kwargs),
nn.Linear(self.d_inner, (self.dt_rank + self.d_state * 2), bias=False, **factory_kwargs),
nn.Linear(self.d_inner, (self.dt_rank + self.d_state * 2), bias=False, **factory_kwargs),
nn.Linear(self.d_inner, (self.dt_rank + self.d_state * 2), bias=False, **factory_kwargs),
)
self.x_proj_weight = nn.Parameter(torch.stack([t.weight for t in self.x_proj], dim=0))
del self.x_proj
self.dt_projs = (
self.dt_init(self.dt_rank, self.d_inner, dt_scale, dt_init, dt_min, dt_max, dt_init_floor, **factory_kwargs),
self.dt_init(self.dt_rank, self.d_inner, dt_scale, dt_init, dt_min, dt_max, dt_init_floor, **factory_kwargs),
self.dt_init(self.dt_rank, self.d_inner, dt_scale, dt_init, dt_min, dt_max, dt_init_floor, **factory_kwargs),
self.dt_init(self.dt_rank, self.d_inner, dt_scale, dt_init, dt_min, dt_max, dt_init_floor, **factory_kwargs),
)
self.dt_projs_weight = nn.Parameter(torch.stack([t.weight for t in self.dt_projs], dim=0))
self.dt_projs_bias = nn.Parameter(torch.stack([t.bias for t in self.dt_projs], dim=0))
del self.dt_projs
self.A_logs = self.A_log_init(self.d_state, self.d_inner, copies=4, merge=True)
self.Ds = self.D_init(self.d_inner, copies=4, merge=True)
self.forward_core = self.forward_corev0
self.out_norm = nn.LayerNorm(self.d_inner)
self.out_proj = nn.Linear(self.d_inner, self.d_model, bias=bias, **factory_kwargs)
self.dropout = nn.Dropout(dropout) if dropout > 0. else None
self.drop_path = DropPath(drop_path_rate) if drop_path_rate > 0 else nn.Identity()
@staticmethod
def dt_init(dt_rank, d_inner, dt_scale=1.0, dt_init="random", dt_min=0.001, dt_max=0.1, dt_init_floor=1e-4, **factory_kwargs):
dt_proj = nn.Linear(dt_rank, d_inner, bias=True, **factory_kwargs)
dt_init_std = dt_rank**-0.5 * dt_scale
if dt_init == "constant":
nn.init.constant_(dt_proj.weight, dt_init_std)
elif dt_init == "random":
nn.init.uniform_(dt_proj.weight, -dt_init_std, dt_init_std)
else:
raise NotImplementedError
dt = torch.exp(
torch.rand(d_inner, **factory_kwargs) * (math.log(dt_max) - math.log(dt_min))
+ math.log(dt_min)
).clamp(min=dt_init_floor)
inv_dt = dt + torch.log(-torch.expm1(-dt))
with torch.no_grad():
dt_proj.bias.copy_(inv_dt)
dt_proj.bias._no_reinit = True
return dt_proj
@staticmethod
def A_log_init(d_state, d_inner, copies=1, device=None, merge=True):
A = repeat(
torch.arange(1, d_state + 1, dtype=torch.float32, device=device),
"n -> d n",
d=d_inner,
).contiguous()
A_log = torch.log(A)
if copies > 1:
A_log = repeat(A_log, "d n -> r d n", r=copies)
if merge:
A_log = A_log.flatten(0, 1)
A_log = nn.Parameter(A_log)
A_log._no_weight_decay = True
return A_log
@staticmethod
def D_init(d_inner, copies=1, device=None, merge=True):
D = torch.ones(d_inner, device=device)
if copies > 1:
D = repeat(D, "n1 -> r n1", r=copies)
if merge:
D = D.flatten(0, 1)
D = nn.Parameter(D)
D._no_weight_decay = True
return D
def forward_corev0(self, x: torch.Tensor):
self.selective_scan = selective_scan_fn
B, C, H, W = x.shape
L = H * W
K = 4
x_hwwh = torch.stack([x.view(B, -1, L), torch.transpose(x, dim0=2, dim1=3).contiguous().view(B, -1, L)], dim=1).view(B, 2, -1, L)
xs = torch.cat([x_hwwh, torch.flip(x_hwwh, dims=[-1])], dim=1)
x_dbl = torch.einsum("b k d l, k c d -> b k c l", xs.view(B, K, -1, L), self.x_proj_weight)
dts, Bs, Cs = torch.split(x_dbl, [self.dt_rank, self.d_state, self.d_state], dim=2)
dts = torch.einsum("b k r l, k d r -> b k d l", dts.view(B, K, -1, L), self.dt_projs_weight)
xs = xs.float().view(B, -1, L)
dts = dts.contiguous().float().view(B, -1, L)
Bs = Bs.float().view(B, K, -1, L)
Cs = Cs.float().view(B, K, -1, L)
Ds = self.Ds.float().view(-1)
As = -torch.exp(self.A_logs.float()).view(-1, self.d_state)
dt_projs_bias = self.dt_projs_bias.float().view(-1)
out_y = self.selective_scan(
xs, dts,
As, Bs, Cs, Ds, z=None,
delta_bias=dt_projs_bias,
delta_softplus=True,
return_last_state=False,
).view(B, K, -1, L)
assert out_y.dtype == torch.float
inv_y = torch.flip(out_y[:, 2:4], dims=[-1]).view(B, 2, -1, L)
wh_y = torch.transpose(out_y[:, 1].view(B, -1, W, H), dim0=2, dim1=3).contiguous().view(B, -1, L)
invwh_y = torch.transpose(inv_y[:, 1].view(B, -1, W, H), dim0=2, dim1=3).contiguous().view(B, -1, L)
return out_y[:, 0], inv_y[:, 0], wh_y, invwh_y
def forward(self, x: torch.Tensor, **kwargs):
B, H, W, C = x.shape
z = x
x = x.permute(0, 3, 1, 2).contiguous()
x = self.act(self.conv2d(x))
y1, y2, y3, y4 = self.forward_core(x)
assert y1.dtype == torch.float32
y = y1 + y2 + y3 + y4
y = torch.transpose(y, dim0=1, dim1=2).contiguous().view(B, H, W, -1)
c1 = self.conv1x1(x)
c3 = self.conv3x3(x)
c5 = self.conv5x5(x)
weights = self.gate(x)
w_m, w_c1, w_c3, w_c5 = weights.chunk(4, dim=1)
fused = w_m * y.permute(0,3,1,2) + w_c3 * c3 + w_c5 * c5 + w_c1 * c1
out = x + self.drop_path(fused)
y = self.out_norm(out.permute(0,2,3,1))
y = y * F.silu(z)
if self.dropout is not None:
y = self.dropout(y)
return y
class LCF_VSSBlock(nn.Module):
def __init__(
self,
hidden_dim: int = 0,
drop_path: float = 0,
norm_layer: Callable[..., torch.nn.Module] = partial(nn.LayerNorm, eps=1e-6),
attn_drop_rate: float = 0,
d_state: int = 16,
**kwargs,
):
super().__init__()
self.ln_1 = norm_layer(hidden_dim)
self.self_attention = SS2D(d_model=hidden_dim, dropout=attn_drop_rate, d_state=d_state, **kwargs)
self.drop_path = DropPath(drop_path)
def forward(self, input: torch.Tensor):
x = input + self.drop_path(self.self_attention(self.ln_1(input)))
return x
class LGF_VSSLayer(nn.Module):
def __init__(
self,
dim,
depth,
attn_drop=0.,
drop_path=0.,
norm_layer=nn.LayerNorm,
downsample=None,
use_checkpoint=False,
d_state=16,
**kwargs,
):
super().__init__()
self.dim = dim
self.use_checkpoint = use_checkpoint
self.blocks = nn.ModuleList([
LCF_VSSBlock(
hidden_dim=dim,
drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path,
norm_layer=norm_layer,
attn_drop_rate=attn_drop,
d_state=d_state,
)
for i in range(depth)])
if True:
def _init_weights(module: nn.Module):
for name, p in module.named_parameters():
if name in ["out_proj.weight"]:
p = p.clone().detach_()
nn.init.kaiming_uniform_(p, a=math.sqrt(5))
self.apply(_init_weights)
if downsample is not None:
self.downsample = downsample(dim=dim, norm_layer=norm_layer)
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 LGF_VSSLayer_up(nn.Module):
def __init__(
self,
dim,
depth,
attn_drop=0.,
drop_path=0.,
norm_layer=nn.LayerNorm,
upsample=None,
use_checkpoint=False,
d_state=16,
**kwargs,
):
super().__init__()
self.use_checkpoint = use_checkpoint
self.blocks = nn.ModuleList([
LCF_VSSBlock(
hidden_dim=dim,
drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path,
norm_layer=norm_layer,
attn_drop_rate=attn_drop,
d_state=d_state,
)
for i in range(depth)])
if True:
def _init_weights(module: nn.Module):
for name, p in module.named_parameters():
if name in ["out_proj.weight"]:
p = p.clone().detach_()
nn.init.kaiming_uniform_(p, a=math.sqrt(5))
self.apply(_init_weights)
if upsample is not None:
self.upsample = upsample(dim=dim, norm_layer=norm_layer)
else:
self.upsample = None
def forward(self, x):
if self.upsample is not None:
x = self.upsample(x)
for blk in self.blocks:
if self.use_checkpoint:
x = checkpoint.checkpoint(blk, x)
else:
x = blk(x)
return x
class AttentionGate(nn.Module):
def __init__(self, in_channels, gate_channels, inter_channels=None):
super().__init__()
if inter_channels is None:
inter_channels = in_channels // 2
self.W_g = nn.Sequential(
nn.Conv2d(gate_channels, inter_channels, kernel_size=1),
nn.BatchNorm2d(inter_channels)
)
self.W_x = nn.Sequential(
nn.Conv2d(in_channels, inter_channels, kernel_size=3, padding=1),
nn.BatchNorm2d(inter_channels)
)
self.psi = nn.Sequential(
nn.Conv2d(inter_channels, 1, kernel_size=1),
nn.BatchNorm2d(1),
nn.Sigmoid()
)
self.channel_att = nn.Sequential(
nn.AdaptiveAvgPool2d(1),
nn.Conv2d(inter_channels, inter_channels//8, 1),
nn.ReLU(),
nn.Conv2d(inter_channels//8, inter_channels, 1),
nn.Sigmoid()
)
def forward(self, x, g):
x = x.permute(0, 3, 1, 2)
_, _, H, W = x.shape
g = g.permute(0, 3, 1, 2)
g_resized = F.interpolate(g, size=(H, W), mode='bilinear', align_corners=False)
g_conv = self.W_g(g_resized)
x_conv = self.W_x(x)
spatial_att = torch.sigmoid(g_conv + x_conv)
channel_att = self.channel_att(spatial_att)
att = spatial_att * channel_att
return x * self.psi(att)
class MCFB(nn.Module):
def __init__(self, enc_dims=[96, 192, 384, 768]):
super().__init__()
self.norm = nn.ModuleList([nn.LayerNorm(dim) for dim in enc_dims])
self.enc_dims = enc_dims
feat_num = len(enc_dims)
self.dec_trans = nn.ModuleList()
for i in range(feat_num):
self.trans = nn.ModuleList()
for j in range(feat_num):
if j > i:
tmp = UpSample(enc_dims[j], enc_dims[i] // 4, 2 ** (j - i))
elif j == i:
tmp = DownSample(enc_dims[j], enc_dims[i] // 4, 1)
else:
tmp = DownSample(enc_dims[j], enc_dims[i] // 4, 2 ** (i - j))
self.trans.append(tmp)
self.dec_trans.append(self.trans)
self.blks = nn.ModuleList()
for i in range(feat_num):
self.blks.append(LCF_VSSBlock(
hidden_dim=enc_dims[i],
drop_path=0.,
norm_layer=nn.LayerNorm,
attn_drop_rate=0.,
d_state=16,
))
self.attention_gates = nn.ModuleList()
for i in range(len(enc_dims)):
gates = nn.ModuleList()
for j in range(len(enc_dims)):
if j != i:
gates.append(AttentionGate(enc_dims[j], enc_dims[i]))
self.attention_gates.append(gates)
def forward(self, enc_list: list, dec_list: list, dec_idx):
full_scale_list = []
target_dim = self.enc_dims[dec_idx]
Nnum = 0
for i, feat in enumerate(enc_list[:]):
if i != dec_idx:
feat = self.attention_gates[dec_idx][Nnum](feat, enc_list[dec_idx]).permute(0, 2, 3, 1)
Nnum = Nnum + 1
full_scale_list.append(feat)
full_scale_list = [self.dec_trans[dec_idx][i](feat) for i, feat in enumerate(full_scale_list)]
output = torch.concat(full_scale_list, dim=3)
output = self.blks[dec_idx](output)
output += dec_list[-1]
return self.norm[dec_idx](output)
class LGF_VSS(nn.Module):
def __init__(self, patch_size=4, in_chans=3, num_classes=1000, depths=[2, 2, 9, 2], depths_decoder=[2, 9, 2, 2],
dims=[96, 192, 384, 768], dims_decoder=[768, 384, 192, 96], d_state=16, drop_rate=0., attn_drop_rate=0., drop_path_rate=0.1,
norm_layer=nn.LayerNorm, patch_norm=True,
use_checkpoint=False,
use_fullScaleSkip=False,
**kwargs):
super().__init__()
self.num_classes = num_classes
self.num_layers = len(depths)
if isinstance(dims, int):
dims = [int(dims * 2 ** i_layer) for i_layer in range(self.num_layers)]
self.embed_dim = dims[0]
self.num_features = dims[-1]
self.dims = dims
self.patch_embed = PatchEmbed2D(patch_size=patch_size, in_chans=in_chans, embed_dim=self.embed_dim,
norm_layer=norm_layer if patch_norm else None)
self.ape = False
if self.ape:
self.patches_resolution = self.patch_embed.patches_resolution
self.absolute_pos_embed = nn.Parameter(torch.zeros(1, *self.patches_resolution, self.embed_dim))
trunc_normal_(self.absolute_pos_embed, std=.02)
self.pos_drop = nn.Dropout(p=drop_rate)
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, sum(depths))]
dpr_decoder = [x.item() for x in torch.linspace(0, drop_path_rate, sum(depths_decoder))][::-1]
self.layers = nn.ModuleList()
for i_layer in range(self.num_layers):
layer = LGF_VSSLayer(
dim=dims[i_layer],
depth=depths[i_layer],
d_state=math.ceil(dims[0] / 6) if d_state is None else d_state,
drop=drop_rate,
attn_drop=attn_drop_rate,
drop_path=dpr[sum(depths[:i_layer]):sum(depths[:i_layer + 1])],
norm_layer=norm_layer,
downsample=PatchMerging2D if (i_layer < self.num_layers - 1) else None,
use_checkpoint=use_checkpoint,
)
self.layers.append(layer)
self.layers_up = nn.ModuleList()
for i_layer in range(self.num_layers):
layer = LGF_VSSLayer_up(
dim=dims_decoder[i_layer],
depth=depths_decoder[i_layer],
d_state=math.ceil(dims[0] / 6) if d_state is None else d_state,
drop=drop_rate,
attn_drop=attn_drop_rate,
drop_path=dpr_decoder[sum(depths_decoder[:i_layer]):sum(depths_decoder[:i_layer + 1])],
norm_layer=norm_layer,
upsample=PatchExpand2D if (i_layer != 0) else None,
use_checkpoint=use_checkpoint,
)
self.layers_up.append(layer)
self.final_up = Final_Expand(dim=dims_decoder[-1], dim_scale=4, norm_layer=norm_layer)
self.final_conv = nn.Conv2d(dims_decoder[-1]//4, num_classes, 1)
self.fullScaleSkip = MCFB(dims) if use_fullScaleSkip else None
self.decoder_heads = nn.ModuleList([
nn.Conv2d(dim, num_classes, 1) for dim in dims[::-1]
])
self.apply(self._init_weights)
def _init_weights(self, m: nn.Module):
if isinstance(m, nn.Linear):
trunc_normal_(m.weight, std=.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(self):
return {'absolute_pos_embed'}
@torch.jit.ignore
def no_weight_decay_keywords(self):
return {'relative_position_bias_table'}
def forward_features(self, x):
skip_list = []
x = self.patch_embed(x)
if self.ape:
x = x + self.absolute_pos_embed
x = self.pos_drop(x)
for layer in self.layers:
skip_list.append(x)
x = layer(x)
return x, skip_list
def forward_features_up(self, x, skip_list):
for inx, layer_up in enumerate(self.layers_up):
if inx == 0:
x = layer_up(x)
else:
x = layer_up(x+skip_list[-inx])
return x
def forward_features_up2(self, x, enc_list):
dec_list = []
dec_outputs = []
for i, layer_up in enumerate(self.layers_up):
x1 = layer_up(x)
dec_list.append(x1)
dec_outputs.append(self.decoder_heads[i](x1.permute(0, 3, 1, 2)).permute(0, 2, 3, 1))
x = self.fullScaleSkip(enc_list, dec_list, self.num_layers - i - 1)
return x, dec_outputs
def forward_final(self, x):
x = self.final_up(x)
x = x.permute(0,3,1,2)
x = self.final_conv(x)
return x
def forward_backbone(self, x):
x = self.patch_embed(x)
if self.ape:
x = x + self.absolute_pos_embed
x = self.pos_drop(x)
for layer in self.layers:
x = layer(x)
return x
def forward(self, x):
dec_outputs = None
x, skip_list = self.forward_features(x)
if self.fullScaleSkip is None:
x = self.forward_features_up(x, skip_list)
else:
x, dec_outputs = self.forward_features_up2(x, skip_list)
x = self.forward_final(x)
return x, dec_outputs