Add SGM-VFI (CVPR 2024) frame interpolation support
SGM-VFI combines local flow estimation with sparse global matching (GMFlow) to handle large motion and occlusion-heavy scenes. Adds 3 new nodes: Load SGM-VFI Model, SGM-VFI Interpolate, SGM-VFI Segment Interpolate. Architecture files vendored from MCG-NJU/SGM-VFI with device-awareness fixes (no hardcoded .cuda()), relative imports, and debug code removed. README updated with model comparison table. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
459
sgm_vfi_arch/feature_extractor.py
Normal file
459
sgm_vfi_arch/feature_extractor.py
Normal file
@@ -0,0 +1,459 @@
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import math
|
||||
from timm.models.layers import DropPath, to_2tuple, trunc_normal_
|
||||
from .position import PositionEmbeddingSine
|
||||
|
||||
def window_partition(x, window_size):
|
||||
B, H, W, C = x.shape
|
||||
x = x.view(B, H // window_size[0], window_size[0], W // window_size[1], window_size[1], C)
|
||||
windows = (
|
||||
x.permute(0, 1, 3, 2, 4, 5).contiguous().view(-1, window_size[0] * window_size[1], C)
|
||||
)
|
||||
return windows
|
||||
|
||||
|
||||
def window_reverse(windows, window_size, H, W):
|
||||
nwB, N, C = windows.shape
|
||||
windows = windows.view(-1, window_size[0], window_size[1], C)
|
||||
B = int(nwB / (H * W / window_size[0] / window_size[1]))
|
||||
x = windows.view(
|
||||
B, H // window_size[0], W // window_size[1], window_size[0], window_size[1], -1
|
||||
)
|
||||
x = x.permute(0, 1, 3, 2, 4, 5).contiguous().view(B, H, W, -1)
|
||||
return x
|
||||
|
||||
|
||||
def pad_if_needed(x, size, window_size):
|
||||
n, h, w, c = size
|
||||
pad_h = math.ceil(h / window_size[0]) * window_size[0] - h
|
||||
pad_w = math.ceil(w / window_size[1]) * window_size[1] - w
|
||||
if pad_h > 0 or pad_w > 0: # center-pad the feature on H and W axes
|
||||
img_mask = torch.zeros((1, h + pad_h, w + pad_w, 1)) # 1 H W 1
|
||||
h_slices = (
|
||||
slice(0, pad_h // 2),
|
||||
slice(pad_h // 2, h + pad_h // 2),
|
||||
slice(h + pad_h // 2, None),
|
||||
)
|
||||
w_slices = (
|
||||
slice(0, pad_w // 2),
|
||||
slice(pad_w // 2, w + pad_w // 2),
|
||||
slice(w + pad_w // 2, None),
|
||||
)
|
||||
cnt = 0
|
||||
for h in h_slices:
|
||||
for w in w_slices:
|
||||
img_mask[:, h, w, :] = cnt
|
||||
cnt += 1
|
||||
|
||||
mask_windows = window_partition(
|
||||
img_mask, window_size
|
||||
) # nW, window_size*window_size, 1
|
||||
mask_windows = mask_windows.squeeze(-1)
|
||||
attn_mask = mask_windows.unsqueeze(1) - mask_windows.unsqueeze(2)
|
||||
attn_mask = attn_mask.masked_fill(
|
||||
attn_mask != 0, float(-100.0)
|
||||
).masked_fill(attn_mask == 0, float(0.0))
|
||||
return nn.functional.pad(
|
||||
x,
|
||||
(0, 0, pad_w // 2, pad_w - pad_w // 2, pad_h // 2, pad_h - pad_h // 2),
|
||||
), attn_mask
|
||||
return x, None
|
||||
|
||||
|
||||
def depad_if_needed(x, size, window_size):
|
||||
n, h, w, c = size
|
||||
pad_h = math.ceil(h / window_size[0]) * window_size[0] - h
|
||||
pad_w = math.ceil(w / window_size[1]) * window_size[1] - w
|
||||
if pad_h > 0 or pad_w > 0: # remove the center-padding on feature
|
||||
return x[:, pad_h // 2: pad_h // 2 + h, pad_w // 2: pad_w // 2 + w, :].contiguous()
|
||||
return x
|
||||
|
||||
|
||||
class Mlp(nn.Module):
|
||||
def __init__(self, in_features, hidden_features=None, out_features=None, act_layer=nn.GELU, drop=0.):
|
||||
super().__init__()
|
||||
out_features = out_features or in_features
|
||||
hidden_features = hidden_features or in_features
|
||||
self.fc1 = nn.Linear(in_features, hidden_features)
|
||||
self.dwconv = DWConv(hidden_features)
|
||||
self.act = act_layer()
|
||||
self.fc2 = nn.Linear(hidden_features, out_features)
|
||||
self.drop = nn.Dropout(drop)
|
||||
self.relu = nn.ReLU(inplace=True)
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
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)
|
||||
elif isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def forward(self, x, H, W):
|
||||
x = self.fc1(x)
|
||||
x = self.dwconv(x, H, W)
|
||||
x = self.act(x)
|
||||
x = self.drop(x)
|
||||
x = self.fc2(x)
|
||||
x = self.drop(x)
|
||||
return x
|
||||
|
||||
|
||||
class InterFrameAttention(nn.Module):
|
||||
def __init__(self, dim, num_heads=8, qkv_bias=False, qk_scale=None, attn_drop=0., proj_drop=0.):
|
||||
super().__init__()
|
||||
assert dim % num_heads == 0, f"dim {dim} should be divided by num_heads {num_heads}."
|
||||
|
||||
self.dim = dim
|
||||
self.num_heads = num_heads
|
||||
head_dim = dim // num_heads
|
||||
self.scale = qk_scale or head_dim ** -0.5
|
||||
|
||||
self.q = nn.Linear(dim, dim, bias=qkv_bias)
|
||||
self.kv = nn.Linear(dim, dim * 2, bias=qkv_bias)
|
||||
self.attn_drop = nn.Dropout(attn_drop)
|
||||
self.proj = nn.Linear(dim, dim)
|
||||
self.proj_drop = nn.Dropout(proj_drop)
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
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)
|
||||
elif isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def forward(self, x1, x2, H, W, mask=None):
|
||||
B, N, C = x1.shape
|
||||
q = self.q(x1).reshape(B, N, self.num_heads, C // self.num_heads).permute(0, 2, 1, 3)
|
||||
kv = self.kv(x2).reshape(B, -1, 2, self.num_heads, C // self.num_heads).permute(2, 0, 3, 1, 4)
|
||||
k, v = kv[0], kv[1]
|
||||
attn = (q @ k.transpose(-2, -1)) * self.scale
|
||||
|
||||
if mask is not None:
|
||||
nW = mask.shape[0] # mask: nW, N, N
|
||||
attn = attn.view(B // nW, nW, self.num_heads, N, N) + mask.unsqueeze(
|
||||
1
|
||||
).unsqueeze(0)
|
||||
attn = attn.view(-1, self.num_heads, N, N)
|
||||
attn = attn.softmax(dim=-1)
|
||||
else:
|
||||
attn = attn.softmax(dim=-1)
|
||||
|
||||
attn = self.attn_drop(attn)
|
||||
x = (attn @ v).transpose(1, 2).reshape(B, N, C)
|
||||
x = self.proj(x)
|
||||
x = self.proj_drop(x)
|
||||
return x
|
||||
|
||||
|
||||
class MotionFormerBlock(nn.Module):
|
||||
def __init__(self, dim, num_heads, window_size=0, shift_size=0, mlp_ratio=4., bidirectional=True,
|
||||
qkv_bias=False, qk_scale=None, drop=0., attn_drop=0.,
|
||||
drop_path=0., act_layer=nn.GELU, norm_layer=nn.LayerNorm, ):
|
||||
super().__init__()
|
||||
self.window_size = window_size
|
||||
if not isinstance(self.window_size, (tuple, list)):
|
||||
self.window_size = to_2tuple(window_size)
|
||||
self.shift_size = shift_size
|
||||
if not isinstance(self.shift_size, (tuple, list)):
|
||||
self.shift_size = to_2tuple(shift_size)
|
||||
self.bidirectional = bidirectional
|
||||
self.norm1 = norm_layer(dim)
|
||||
self.attn = InterFrameAttention(
|
||||
dim,
|
||||
num_heads=num_heads, qkv_bias=qkv_bias, qk_scale=qk_scale,
|
||||
attn_drop=attn_drop, proj_drop=drop)
|
||||
self.drop_path = DropPath(drop_path) if drop_path > 0. else nn.Identity()
|
||||
self.norm2 = norm_layer(dim)
|
||||
mlp_hidden_dim = int(dim * mlp_ratio)
|
||||
self.mlp = Mlp(in_features=dim, hidden_features=mlp_hidden_dim, act_layer=act_layer, drop=drop)
|
||||
# BEGIN: absolute pos_embed, beneficial to local information extraction in our experiments
|
||||
self.pos_embed = PositionEmbeddingSine(dim // 2)
|
||||
# END: absolute pos_embed
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
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)
|
||||
elif isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def forward(self, x, H, W, B, self_att=False):
|
||||
x = x.view(2 * B, H, W, -1)
|
||||
x_pad, mask = pad_if_needed(x, x.size(), self.window_size)
|
||||
|
||||
if self.shift_size[0] or self.shift_size[1]:
|
||||
_, H_p, W_p, C = x_pad.shape
|
||||
x_pad = torch.roll(x_pad, shifts=(-self.shift_size[0], -self.shift_size[1]), dims=(1, 2))
|
||||
|
||||
if hasattr(self, 'HW') and self.HW.item() == H_p * W_p:
|
||||
shift_mask = self.attn_mask
|
||||
else:
|
||||
shift_mask = torch.zeros((1, H_p, W_p, 1)) # 1 H W 1
|
||||
h_slices = (slice(0, -self.window_size[0]),
|
||||
slice(-self.window_size[0], -self.shift_size[0]),
|
||||
slice(-self.shift_size[0], None))
|
||||
w_slices = (slice(0, -self.window_size[1]),
|
||||
slice(-self.window_size[1], -self.shift_size[1]),
|
||||
slice(-self.shift_size[1], None))
|
||||
cnt = 0
|
||||
for h in h_slices:
|
||||
for w in w_slices:
|
||||
shift_mask[:, h, w, :] = cnt
|
||||
cnt += 1
|
||||
|
||||
mask_windows = window_partition(shift_mask, self.window_size).squeeze(-1)
|
||||
shift_mask = mask_windows.unsqueeze(1) - mask_windows.unsqueeze(2)
|
||||
shift_mask = shift_mask.masked_fill(shift_mask != 0,
|
||||
float(-100.0)).masked_fill(shift_mask == 0,
|
||||
float(0.0))
|
||||
|
||||
if mask is not None:
|
||||
shift_mask = shift_mask.masked_fill(mask != 0,
|
||||
float(-100.0))
|
||||
self.register_buffer("attn_mask", shift_mask)
|
||||
self.register_buffer("HW", torch.Tensor([H_p * W_p]))
|
||||
else:
|
||||
shift_mask = mask
|
||||
|
||||
if shift_mask is not None:
|
||||
shift_mask = shift_mask.to(x_pad.device)
|
||||
|
||||
_, Hw, Ww, C = x_pad.shape
|
||||
x_win = window_partition(x_pad, self.window_size)
|
||||
|
||||
nwB = x_win.shape[0]
|
||||
x_norm = self.norm1(x_win)
|
||||
# BEGIN: absolute pos embed, beneficial to local information extraction in our experiments
|
||||
x_norm = x_norm.view(nwB, self.window_size[0], self.window_size[1], C).permute(0, 3, 1, 2)
|
||||
ape = self.pos_embed(x_norm)
|
||||
x_norm = x_norm + ape
|
||||
x_norm = x_norm.permute(0, 2, 3, 1).view(nwB, self.window_size[0] * self.window_size[1], C)
|
||||
# END: absolute pos embed
|
||||
|
||||
if self_att is False:
|
||||
x_reverse = torch.cat([x_norm[nwB // 2:], x_norm[:nwB // 2]])
|
||||
x_appearence = self.attn(x_norm, x_reverse, H, W, shift_mask)
|
||||
else:
|
||||
x_appearence = self.attn(x_norm, x_norm, H, W, shift_mask)
|
||||
|
||||
x_norm = x_norm + self.drop_path(x_appearence)
|
||||
|
||||
x_back = x_norm
|
||||
x_back_win = window_reverse(x_back, self.window_size, Hw, Ww)
|
||||
|
||||
if self.shift_size[0] or self.shift_size[1]:
|
||||
x_back_win = torch.roll(x_back_win, shifts=(self.shift_size[0], self.shift_size[1]), dims=(1, 2))
|
||||
|
||||
x = depad_if_needed(x_back_win, x.size(), self.window_size).view(2 * B, H * W, -1)
|
||||
|
||||
x = x + self.drop_path(self.mlp(self.norm2(x), H, W))
|
||||
return x
|
||||
|
||||
|
||||
class ConvBlock(nn.Module):
|
||||
def __init__(self, in_dim, out_dim, depths=2, act_layer=nn.PReLU):
|
||||
super().__init__()
|
||||
layers = []
|
||||
for i in range(depths):
|
||||
if i == 0:
|
||||
layers.append(nn.Conv2d(in_dim, out_dim, 3, 1, 1))
|
||||
else:
|
||||
layers.append(nn.Conv2d(out_dim, out_dim, 3, 1, 1))
|
||||
layers.extend([
|
||||
act_layer(out_dim),
|
||||
])
|
||||
self.conv = nn.Sequential(*layers)
|
||||
|
||||
def _init_weights(self, m):
|
||||
if isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def forward(self, x):
|
||||
x = self.conv(x)
|
||||
return x
|
||||
|
||||
|
||||
class OverlapPatchEmbed(nn.Module):
|
||||
def __init__(self, patch_size=7, stride=4, in_chans=3, embed_dim=768):
|
||||
super().__init__()
|
||||
patch_size = to_2tuple(patch_size)
|
||||
|
||||
self.patch_size = patch_size
|
||||
self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=stride,
|
||||
padding=(patch_size[0] // 2, patch_size[1] // 2))
|
||||
self.norm = nn.LayerNorm(embed_dim)
|
||||
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
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)
|
||||
elif isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def forward(self, x):
|
||||
x = self.proj(x)
|
||||
_, _, H, W = x.shape
|
||||
x = x.flatten(2).transpose(1, 2)
|
||||
x = self.norm(x)
|
||||
|
||||
return x, H, W
|
||||
|
||||
|
||||
class MotionFormer(nn.Module):
|
||||
def __init__(self, in_chans=3, embed_dims=None, num_heads=None,
|
||||
mlp_ratios=None, qkv_bias=True, qk_scale=None, drop_rate=0.,
|
||||
attn_drop_rate=0., drop_path_rate=0., norm_layer=nn.LayerNorm,
|
||||
depths=None, window_sizes=None, **kwarg):
|
||||
super().__init__()
|
||||
self.depths = depths
|
||||
self.num_stages = len(embed_dims)
|
||||
|
||||
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, sum(depths))] # stochastic depth decay rule
|
||||
cur = 0
|
||||
|
||||
self.conv_stages = self.num_stages - len(num_heads)
|
||||
|
||||
for i in range(self.num_stages):
|
||||
if i == 0:
|
||||
block = ConvBlock(in_chans, embed_dims[i], depths[i])
|
||||
else:
|
||||
if i < self.conv_stages:
|
||||
patch_embed = nn.Sequential(
|
||||
nn.Conv2d(embed_dims[i - 1], embed_dims[i], 3, 2, 1),
|
||||
nn.PReLU(embed_dims[i])
|
||||
)
|
||||
block = ConvBlock(embed_dims[i], embed_dims[i], depths[i])
|
||||
else:
|
||||
patch_embed = OverlapPatchEmbed(patch_size=3,
|
||||
stride=2,
|
||||
in_chans=embed_dims[i - 1],
|
||||
embed_dim=embed_dims[i])
|
||||
|
||||
block = nn.ModuleList([MotionFormerBlock(
|
||||
dim=embed_dims[i], num_heads=num_heads[i - self.conv_stages],
|
||||
window_size=window_sizes[i - self.conv_stages],
|
||||
shift_size=0 if (j % 2) == 0 else window_sizes[i - self.conv_stages] // 2,
|
||||
mlp_ratio=mlp_ratios[i - self.conv_stages], qkv_bias=qkv_bias, qk_scale=qk_scale,
|
||||
drop=drop_rate, attn_drop=attn_drop_rate, drop_path=dpr[cur + j], norm_layer=norm_layer)
|
||||
for j in range(depths[i])])
|
||||
|
||||
norm = norm_layer(embed_dims[i])
|
||||
setattr(self, f"norm{i + 1}", norm)
|
||||
setattr(self, f"patch_embed{i + 1}", patch_embed)
|
||||
cur += depths[i]
|
||||
|
||||
setattr(self, f"block{i + 1}", block)
|
||||
|
||||
self.cor = {}
|
||||
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
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)
|
||||
elif isinstance(m, nn.Conv2d):
|
||||
fan_out = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
|
||||
fan_out //= m.groups
|
||||
m.weight.data.normal_(0, math.sqrt(2.0 / fan_out))
|
||||
if m.bias is not None:
|
||||
m.bias.data.zero_()
|
||||
|
||||
def get_cor(self, shape, device):
|
||||
k = (str(shape), str(device))
|
||||
if k not in self.cor:
|
||||
tenHorizontal = torch.linspace(-1.0, 1.0, shape[2], device=device).view(
|
||||
1, 1, 1, shape[2]).expand(shape[0], -1, shape[1], -1).permute(0, 2, 3, 1)
|
||||
tenVertical = torch.linspace(-1.0, 1.0, shape[1], device=device).view(
|
||||
1, 1, shape[1], 1).expand(shape[0], -1, -1, shape[2]).permute(0, 2, 3, 1)
|
||||
self.cor[k] = torch.cat([tenHorizontal, tenVertical], -1).to(device)
|
||||
return self.cor[k]
|
||||
|
||||
def forward(self, x1, x2):
|
||||
B = x1.shape[0]
|
||||
x = torch.cat([x1, x2], 0)
|
||||
appearence_features = []
|
||||
xs = []
|
||||
for i in range(self.num_stages):
|
||||
patch_embed = getattr(self, f"patch_embed{i + 1}", None)
|
||||
block = getattr(self, f"block{i + 1}", None)
|
||||
norm = getattr(self, f"norm{i + 1}", None)
|
||||
if i < self.conv_stages:
|
||||
if i > 0:
|
||||
x = patch_embed(x)
|
||||
x = block(x)
|
||||
xs.append(x)
|
||||
else:
|
||||
x, H, W = patch_embed(x)
|
||||
for j in range(len(block)):
|
||||
x = block[j](x, H, W, B, self_att=False)
|
||||
xs.append(x.reshape(2 * B, H, W, -1).permute(0, 3, 1, 2).contiguous())
|
||||
x = norm(x)
|
||||
x = x.reshape(2 * B, H, W, -1).permute(0, 3, 1, 2).contiguous()
|
||||
appearence_features.append(x)
|
||||
return appearence_features
|
||||
|
||||
|
||||
class DWConv(nn.Module):
|
||||
def __init__(self, dim):
|
||||
super(DWConv, self).__init__()
|
||||
self.dwconv = nn.Conv2d(dim, dim, 3, 1, 1, bias=True, groups=dim)
|
||||
|
||||
def forward(self, x, H, W):
|
||||
B, N, C = x.shape
|
||||
x = x.transpose(1, 2).reshape(B, C, H, W).contiguous()
|
||||
x = self.dwconv(x)
|
||||
x = x.reshape(B, C, -1).transpose(1, 2)
|
||||
|
||||
return x
|
||||
|
||||
|
||||
def feature_extractor(**kargs):
|
||||
model = MotionFormer(**kargs)
|
||||
return model
|
||||
Reference in New Issue
Block a user