Akash Chowdhury
e9ef5dcea7
- Load Visual Chronometer model with device selection (auto/cpu/cuda) - Predict PhyFPS from video frames using sliding window analysis - Batch prediction with per-segment FPS breakdown - SDPA-optimized attention for better speed and memory efficiency - Auto-downloads model checkpoint from HuggingFace on first use - Progress bar and detailed terminal logging - Input tooltips and node descriptions
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""2+1D Video Autoencoder with temporal 1D CNN - encoder path only for inference."""
|
|
|
|
import math
|
|
import torch
|
|
import torch.nn as nn
|
|
from einops import rearrange
|
|
|
|
from .ae_modules import Normalize, nonlinearity
|
|
from .attention import (
|
|
normalization, zero_module, conv_nd,
|
|
RelativePosition, QKVAttention, CrossAttention as CrossAttentionBase,
|
|
default, exists,
|
|
)
|
|
from .distributions import DiagonalGaussianDistribution
|
|
|
|
|
|
class TemporalConvLayer(nn.Module):
|
|
def __init__(self, in_channels, out_channels):
|
|
super().__init__()
|
|
self.norm = Normalize(in_channels)
|
|
self.conv = torch.nn.Conv3d(
|
|
in_channels, out_channels,
|
|
kernel_size=(3, 3, 3), stride=1, padding=(1, 1, 1),
|
|
)
|
|
nn.init.constant_(self.conv.weight, 0)
|
|
nn.init.constant_(self.conv.bias, 0)
|
|
|
|
def forward(self, x):
|
|
h = self.norm(x)
|
|
h = nonlinearity(h)
|
|
h = self.conv(h)
|
|
return h
|
|
|
|
|
|
class ResnetBlock2plus1D(nn.Module):
|
|
def __init__(self, *, in_channels, out_channels=None, conv_shortcut=False, dropout, temb_channels=512, **kwargs):
|
|
super().__init__()
|
|
self.in_channels = in_channels
|
|
out_channels = in_channels if out_channels is None else out_channels
|
|
self.out_channels = out_channels
|
|
self.use_conv_shortcut = conv_shortcut
|
|
|
|
self.norm1 = Normalize(in_channels)
|
|
self.conv1 = torch.nn.Conv3d(in_channels, out_channels, kernel_size=(1, 3, 3), stride=1, padding=(0, 1, 1))
|
|
self.conv1_tmp = TemporalConvLayer(out_channels, out_channels)
|
|
|
|
if temb_channels > 0:
|
|
self.temb_proj = torch.nn.Linear(temb_channels, out_channels)
|
|
|
|
self.norm2 = Normalize(out_channels)
|
|
self.dropout = torch.nn.Dropout(dropout)
|
|
self.conv2 = torch.nn.Conv3d(out_channels, out_channels, kernel_size=(1, 3, 3), stride=1, padding=(0, 1, 1))
|
|
self.conv2_tmp = TemporalConvLayer(out_channels, out_channels)
|
|
|
|
if self.in_channels != self.out_channels:
|
|
if self.use_conv_shortcut:
|
|
self.conv_shortcut = torch.nn.Conv3d(in_channels, out_channels, kernel_size=(1, 3, 3), stride=1, padding=(0, 1, 1))
|
|
else:
|
|
self.nin_shortcut = torch.nn.Conv3d(in_channels, out_channels, kernel_size=(1, 1, 1), stride=1, padding=(0, 0, 0))
|
|
self.conv3_tmp = TemporalConvLayer(out_channels, out_channels)
|
|
|
|
def forward(self, x, temb=None, mask_temporal=False):
|
|
h = x
|
|
h = self.norm1(h)
|
|
h = nonlinearity(h)
|
|
h = self.conv1(h)
|
|
if not mask_temporal:
|
|
h = self.conv1_tmp(h) + h
|
|
|
|
if temb is not None:
|
|
h = h + self.temb_proj(nonlinearity(temb))[:, :, None, None]
|
|
|
|
h = self.norm2(h)
|
|
h = nonlinearity(h)
|
|
h = self.dropout(h)
|
|
h = self.conv2(h)
|
|
if not mask_temporal:
|
|
h = self.conv2_tmp(h) + h
|
|
|
|
if self.in_channels != self.out_channels:
|
|
if self.use_conv_shortcut:
|
|
x = self.conv_shortcut(x)
|
|
else:
|
|
x = self.nin_shortcut(x)
|
|
if not mask_temporal:
|
|
x = self.conv3_tmp(x) + x
|
|
|
|
return x + h
|
|
|
|
|
|
class AttnBlock3D(nn.Module):
|
|
def __init__(self, in_channels):
|
|
super().__init__()
|
|
self.in_channels = in_channels
|
|
self.norm = Normalize(in_channels)
|
|
self.q = torch.nn.Conv3d(in_channels, in_channels, kernel_size=1, stride=1, padding=0)
|
|
self.k = torch.nn.Conv3d(in_channels, in_channels, kernel_size=1, stride=1, padding=0)
|
|
self.v = torch.nn.Conv3d(in_channels, in_channels, kernel_size=1, stride=1, padding=0)
|
|
self.proj_out = torch.nn.Conv3d(in_channels, in_channels, kernel_size=1, stride=1, padding=0)
|
|
|
|
def forward(self, x):
|
|
h_ = self.norm(x)
|
|
q = self.q(h_)
|
|
k = self.k(h_)
|
|
v = self.v(h_)
|
|
|
|
b, c, t, h, w = q.shape
|
|
# SDPA expects (batch, num_heads, seq_len, head_dim) — use 1 head with dim=c
|
|
q = rearrange(q, "b c t h w -> (b t) 1 (h w) c")
|
|
k = rearrange(k, "b c t h w -> (b t) 1 (h w) c")
|
|
v = rearrange(v, "b c t h w -> (b t) 1 (h w) c")
|
|
|
|
h_ = torch.nn.functional.scaled_dot_product_attention(q, k, v)
|
|
|
|
h_ = rearrange(h_, "(b t) 1 (h w) c -> b c t h w", b=b, t=t, h=h)
|
|
h_ = self.proj_out(h_)
|
|
return x + h_
|
|
|
|
|
|
class TemporalAttention(nn.Module):
|
|
def __init__(self, channels, num_heads=1, num_head_channels=-1, max_temporal_length=64):
|
|
super().__init__()
|
|
if num_head_channels == -1:
|
|
self.num_heads = num_heads
|
|
else:
|
|
self.num_heads = channels // num_head_channels
|
|
|
|
self.norm = normalization(channels)
|
|
self.qkv = zero_module(conv_nd(1, channels, channels * 3, 1))
|
|
self.attention = QKVAttention(self.num_heads)
|
|
self.relative_position_k = RelativePosition(
|
|
num_units=channels // self.num_heads, max_relative_position=max_temporal_length
|
|
)
|
|
self.relative_position_v = RelativePosition(
|
|
num_units=channels // self.num_heads, max_relative_position=max_temporal_length
|
|
)
|
|
self.proj_out = zero_module(conv_nd(1, channels, channels, 1))
|
|
|
|
def forward(self, x, mask=None):
|
|
b, c, t, h, w = x.shape
|
|
out = rearrange(x, "b c t h w -> (b h w) c t")
|
|
qkv = self.qkv(self.norm(out))
|
|
len_q = qkv.size()[-1]
|
|
k_rp = self.relative_position_k(len_q, len_q)
|
|
v_rp = self.relative_position_v(len_q, len_q)
|
|
out = self.attention(qkv, rp=(k_rp, v_rp))
|
|
out = self.proj_out(out)
|
|
out = rearrange(out, "(b h w) c t -> b c t h w", b=b, h=h, w=w)
|
|
return x + out
|
|
|
|
|
|
class Downsample2plus1D(nn.Module):
|
|
def __init__(self, in_channels, with_conv, temp_down):
|
|
super().__init__()
|
|
self.with_conv = with_conv
|
|
self.temp_down = temp_down
|
|
if self.with_conv:
|
|
self.conv = torch.nn.Conv3d(in_channels, in_channels, kernel_size=(1, 3, 3), stride=(1, 2, 2), padding=0)
|
|
|
|
def forward(self, x, mask_temporal=False):
|
|
if self.with_conv:
|
|
pad = (0, 1, 0, 1, 0, 0)
|
|
x = torch.nn.functional.pad(x, pad, mode="constant", value=0)
|
|
x = self.conv(x)
|
|
return x
|
|
|
|
|
|
class Encoder2plus1D(nn.Module):
|
|
def __init__(
|
|
self, *, ch, out_ch, temporal_down_factor, ch_mult=(1, 2, 4, 8),
|
|
num_res_blocks, attn_resolutions, dropout=0.0, resamp_with_conv=True,
|
|
in_channels, resolution, z_channels, double_z=True, **ignore_kwargs,
|
|
):
|
|
super().__init__()
|
|
self.ch = ch
|
|
self.temb_ch = 0
|
|
self.num_resolutions = len(ch_mult)
|
|
self.n_temporal_down = int(math.log2(temporal_down_factor))
|
|
self.num_res_blocks = num_res_blocks
|
|
|
|
self.conv_in = torch.nn.Conv3d(in_channels, self.ch, kernel_size=(1, 3, 3), stride=1, padding=(0, 1, 1))
|
|
|
|
curr_res = resolution
|
|
in_ch_mult = (1,) + tuple(ch_mult)
|
|
self.down = nn.ModuleList()
|
|
|
|
for i_level in range(self.num_resolutions):
|
|
block = nn.ModuleList()
|
|
attn = nn.ModuleList()
|
|
block_in = ch * in_ch_mult[i_level]
|
|
block_out = ch * ch_mult[i_level]
|
|
for i_block in range(self.num_res_blocks):
|
|
block.append(
|
|
ResnetBlock2plus1D(in_channels=block_in, out_channels=block_out, temb_channels=self.temb_ch, dropout=dropout)
|
|
)
|
|
block_in = block_out
|
|
down = nn.Module()
|
|
down.block = block
|
|
down.attn = attn
|
|
if i_level != self.num_resolutions - 1:
|
|
temp_down = i_level <= self.n_temporal_down - 1
|
|
down.downsample = Downsample2plus1D(block_in, resamp_with_conv, temp_down)
|
|
curr_res = curr_res // 2
|
|
self.down.append(down)
|
|
|
|
self.mid = nn.Module()
|
|
self.mid.block_1 = ResnetBlock2plus1D(in_channels=block_in, out_channels=block_in, temb_channels=self.temb_ch, dropout=dropout)
|
|
self.mid.attn_1 = AttnBlock3D(block_in)
|
|
self.mid.attn_1_tmp = TemporalAttention(block_in, num_heads=1)
|
|
self.mid.block_2 = ResnetBlock2plus1D(in_channels=block_in, out_channels=block_in, temb_channels=self.temb_ch, dropout=dropout)
|
|
|
|
self.norm_out = Normalize(block_in)
|
|
self.conv_out = torch.nn.Conv3d(
|
|
block_in, 2 * z_channels if double_z else z_channels,
|
|
kernel_size=(1, 3, 3), stride=1, padding=(0, 1, 1),
|
|
)
|
|
|
|
def forward(self, x, text_embeddings=None, text_attn_mask=None, mask_temporal=False):
|
|
temb = None
|
|
hs = [self.conv_in(x)]
|
|
for i_level in range(self.num_resolutions):
|
|
for i_block in range(self.num_res_blocks):
|
|
h = self.down[i_level].block[i_block](hs[-1], temb, mask_temporal)
|
|
if len(self.down[i_level].attn) > 0:
|
|
h = h + self.down[i_level].attn[i_block](h, context=text_embeddings, mask=text_attn_mask)
|
|
hs.append(h)
|
|
if i_level != self.num_resolutions - 1:
|
|
hs.append(self.down[i_level].downsample(hs[-1], mask_temporal))
|
|
h = hs[-1]
|
|
h = self.mid.block_1(h, temb, mask_temporal)
|
|
h = self.mid.attn_1(h)
|
|
if not mask_temporal:
|
|
h = self.mid.attn_1_tmp(h)
|
|
h = self.mid.block_2(h, temb, mask_temporal)
|
|
h = self.norm_out(h)
|
|
h = nonlinearity(h)
|
|
h = self.conv_out(h)
|
|
return h
|
|
|
|
|
|
class AutoencoderKL2plus1D_1dcnn(nn.Module):
|
|
"""2+1D Video Autoencoder with temporal 1D CNN. Inference-only version."""
|
|
|
|
def __init__(
|
|
self,
|
|
ddconfig,
|
|
ppconfig,
|
|
lossconfig=None,
|
|
embed_dim=0,
|
|
use_quant_conv=True,
|
|
ckpt_path=None,
|
|
**kwargs,
|
|
):
|
|
super().__init__()
|
|
self.use_quant_conv = use_quant_conv
|
|
|
|
# 2+1D encoder (the only component used for FPS prediction)
|
|
self.encoder = Encoder2plus1D(**ddconfig)
|
|
|
|
if use_quant_conv:
|
|
assert embed_dim
|
|
self.embed_dim = embed_dim
|
|
self.quant_conv = torch.nn.Conv3d(2 * ddconfig["z_channels"], 2 * embed_dim, 1)
|
|
|
|
if ckpt_path is not None:
|
|
self._init_from_ckpt(ckpt_path)
|
|
|
|
def _init_from_ckpt(self, path):
|
|
sd = torch.load(path, map_location="cpu", weights_only=False)
|
|
try:
|
|
sd = sd["state_dict"]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
self.load_state_dict(sd, strict=False)
|
|
|
|
def freeze(self):
|
|
self.eval()
|
|
for param in self.parameters():
|
|
param.requires_grad = False
|