""" A liberated of WU's MAE pretrained backbone import sys sys.path.append('/data/joncrall/dvc-repos/smart_expt_dvc/models/wu/MAE-2023-02-09') import pred_features import liberator lib = liberator.Liberator() lib.add_dynamic(pred_features.ViT) lib.expand(['pred_features']) print(lib.current_sourcecode()) """ from einops.layers.torch import Rearrange import torch.nn as nn import torch from einops import rearrange from einops import repeat class PreNorm(nn.Module): def __init__(self, dim, fn): super().__init__() self.norm = nn.LayerNorm(dim) self.fn = fn def forward(self, x, **kwargs): return self.fn(self.norm(x), **kwargs) class FeedForward(nn.Module): def __init__(self, dim, hidden_dim, dropout=0.): super().__init__() self.net = nn.Sequential( nn.Linear(dim, hidden_dim), nn.GELU(), nn.Dropout(dropout), nn.Linear(hidden_dim, dim), nn.Dropout(dropout) ) def forward(self, x, mask): return self.net(x) class Attention(nn.Module): def __init__(self, dim, heads=8, dim_head=64, dropout=0.): super().__init__() inner_dim = dim_head * heads project_out = not (heads == 1 and dim_head == dim) self.heads = heads self.scale = dim_head ** -0.5 self.attend = nn.Softmax(dim=-1) self.dropout = nn.Dropout(dropout) self.to_qkv = nn.Linear(dim, inner_dim * 3, bias=False) self.to_out = nn.Sequential( nn.Linear(inner_dim, dim), nn.Dropout(dropout) ) if project_out else nn.Identity() def forward(self, x, mask): qkv = self.to_qkv(x).chunk(3, dim=-1) q, k, v = map( lambda t: rearrange( t, 'b n (h d) -> b h n d', h=self.heads), qkv) #dots = torch.matmul(q, k.transpose(-1, -2)) * self.scale dots = torch.einsum('ijkd,ijld->ijkl', q, k) * self.scale attn = torch.einsum('ijkl,ijkl->ijkl', dots, repeat(mask, 'b n -> b h n d', h=self.heads, d=dots.shape[-1])) attn = self.attend(dots) attn = self.dropout(attn) out = torch.matmul(attn, v) out = rearrange(out, 'b h n d -> b n (h d)') return self.to_out(out) def pair(t): return t if isinstance(t, tuple) else (t, t) class Transformer(nn.Module): """ Example: >>> dim = 4 >>> depth = 3 >>> heads = 2 >>> dim_head = 2 >>> mlp_dim = 2 >>> self = Transformer(dim, depth, heads, dim_head, mlp_dim) >>> x = torch.rand(2, 3, dim) >>> mask = torch.rand(2, 3) > 0 >>> out = self.forward(x, mask) >>> print(f'out.shape={out.shape}') """ def __init__(self, dim, depth, heads, dim_head, mlp_dim, dropout=0.): super().__init__() self.layers = nn.ModuleList([]) for _ in range(depth): self.layers.append(nn.ModuleList([ PreNorm( dim, Attention( dim, heads=heads, dim_head=dim_head, dropout=dropout)), PreNorm(dim, FeedForward(dim, mlp_dim, dropout=dropout)) ])) def forward(self, x, mask): for attn, ff in self.layers: x = attn(x, mask=mask) + x x = ff(x, mask=mask) + x return x class ViT(nn.Module): def __init__(self, *, image_size, image_patch_size, frames, frame_patch_size, dim, depth, heads, mlp_dim, channels=6, dim_head=64, dropout=0., emb_dropout=0.): super().__init__() image_height, image_width = pair(image_size) patch_height, patch_width = pair(image_patch_size) assert image_height % patch_height == 0 and image_width % patch_width == 0, 'Image dimensions must be divisible by the patch size.' assert frames % frame_patch_size == 0, 'Frames must be divisible by frame patch size' num_patches = (image_height // patch_height) * \ (image_width // patch_width) * (frames // frame_patch_size) patch_dim = channels * patch_height * patch_width * frame_patch_size #assert pool in {'cls', 'mean'}, 'pool type must be either cls (cls token) or mean (mean pooling)' self.to_patch_embedding = nn.Sequential( Rearrange( 'b (f pf) c (h p1) (w p2) -> b (f h w) (p1 p2 pf c)', p1=patch_height, p2=patch_width, pf=frame_patch_size), nn.Linear(patch_dim, dim), ) self.pos_embedding = nn.Parameter(torch.randn(1, num_patches + 1, dim)) #self.cls_token = nn.Parameter(torch.randn(1, 1, dim)) self.dropout = nn.Dropout(emb_dropout) self.transformer = Transformer( dim, depth, heads, dim_head, mlp_dim, dropout) #self.pool = pool #self.to_latent = nn.Identity() # self.mlp_head = nn.Sequential( # nn.LayerNorm(dim), #nn.Linear(dim, num_classes) # ) def forward(self, video): x = self.to_patch_embedding(video) b, n, _ = x.shape #cls_tokens = repeat(self.cls_token, '1 1 d -> b 1 d', b = b) #x = torch.cat((cls_tokens, x), dim=1) x += self.pos_embedding[:, :(n + 1)] x = self.dropout(x) x = self.transformer(x) #x = x.mean(dim = 1) if self.pool == 'mean' else x[:, 0] #x = self.to_latent(x) return x def wu_backbone(): """ from torch_liberator import Pretrained ckpt_fpath = '/home/joncrall/remote/toothbrush/data/dvc-repos/smart_expt_dvc/models/wu/MAE-2023-02-09/goldenMae-epoch=07-val_loss=0.23.ckpt' initializer = Pretrained(ckpt_fpath, association='embedding') vit = wu_backbone() result = initializer.forward(vit) """ vit = ViT( image_size=128, image_patch_size=4, frames=4, frame_patch_size=2, dim=16, depth=12, heads=12, mlp_dim=1024, dropout=0.1 ) return vit