Spaces:
Paused
Paused
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from diffusers import ModelMixin | |
| from .warplayer import warp | |
| device = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| def conv(in_planes, out_planes, kernel_size=3, stride=1, padding=1, dilation=1): | |
| return nn.Sequential( | |
| nn.Conv2d(in_planes, out_planes, kernel_size=kernel_size, stride=stride, | |
| padding=padding, dilation=dilation, bias=True), | |
| nn.PReLU(out_planes) | |
| ) | |
| def conv_bn(in_planes, out_planes, kernel_size=3, stride=1, padding=1, dilation=1): | |
| return nn.Sequential( | |
| nn.Conv2d(in_planes, out_planes, kernel_size=kernel_size, stride=stride, | |
| padding=padding, dilation=dilation, bias=False), | |
| nn.BatchNorm2d(out_planes), | |
| nn.PReLU(out_planes) | |
| ) | |
| def convert(param): | |
| return { | |
| k.replace("module.", ""): v | |
| for k, v in param.items() | |
| if "module." in k | |
| } | |
| class IFBlock(nn.Module): | |
| def __init__(self, in_planes, c=64): | |
| super(IFBlock, self).__init__() | |
| self.conv0 = nn.Sequential( | |
| conv(in_planes, c//2, 3, 2, 1), | |
| conv(c//2, c, 3, 2, 1), | |
| ) | |
| self.convblock0 = nn.Sequential( | |
| conv(c, c), | |
| conv(c, c) | |
| ) | |
| self.convblock1 = nn.Sequential( | |
| conv(c, c), | |
| conv(c, c) | |
| ) | |
| self.convblock2 = nn.Sequential( | |
| conv(c, c), | |
| conv(c, c) | |
| ) | |
| self.convblock3 = nn.Sequential( | |
| conv(c, c), | |
| conv(c, c) | |
| ) | |
| self.conv1 = nn.Sequential( | |
| nn.ConvTranspose2d(c, c//2, 4, 2, 1), | |
| nn.PReLU(c//2), | |
| nn.ConvTranspose2d(c//2, 4, 4, 2, 1), | |
| ) | |
| self.conv2 = nn.Sequential( | |
| nn.ConvTranspose2d(c, c//2, 4, 2, 1), | |
| nn.PReLU(c//2), | |
| nn.ConvTranspose2d(c//2, 1, 4, 2, 1), | |
| ) | |
| def forward(self, x, flow, scale=1): | |
| x = F.interpolate(x, scale_factor= 1. / scale, mode="bilinear", align_corners=False, recompute_scale_factor=False) | |
| flow = F.interpolate(flow, scale_factor= 1. / scale, mode="bilinear", align_corners=False, recompute_scale_factor=False) * 1. / scale | |
| feat = self.conv0(torch.cat((x, flow), 1)) | |
| feat = self.convblock0(feat) + feat | |
| feat = self.convblock1(feat) + feat | |
| feat = self.convblock2(feat) + feat | |
| feat = self.convblock3(feat) + feat | |
| flow = self.conv1(feat) | |
| mask = self.conv2(feat) | |
| flow = F.interpolate(flow, scale_factor=scale, mode="bilinear", align_corners=False, recompute_scale_factor=False) * scale | |
| mask = F.interpolate(mask, scale_factor=scale, mode="bilinear", align_corners=False, recompute_scale_factor=False) | |
| return flow, mask | |
| class IFNet(ModelMixin): | |
| def __init__(self, ckpt_path="checkpoints/flownet.pkl"): | |
| super(IFNet, self).__init__() | |
| self.block0 = IFBlock(7+4, c=90) | |
| self.block1 = IFBlock(7+4, c=90) | |
| self.block2 = IFBlock(7+4, c=90) | |
| self.block_tea = IFBlock(10+4, c=90) | |
| if ckpt_path is not None: | |
| self.load_state_dict(convert(torch.load(ckpt_path, map_location ='cpu'))) | |
| def inference(self, img0, img1, scale=1.0): | |
| imgs = torch.cat((img0, img1), 1) | |
| scale_list = [4/scale, 2/scale, 1/scale] | |
| flow, mask, merged = self.forward(imgs, scale_list) | |
| return merged[2] | |
| def forward(self, x, scale_list=[4, 2, 1], training=False): | |
| if training == False: | |
| channel = x.shape[1] // 2 | |
| img0 = x[:, :channel] | |
| img1 = x[:, channel:] | |
| flow_list = [] | |
| merged = [] | |
| mask_list = [] | |
| warped_img0 = img0 | |
| warped_img1 = img1 | |
| flow = (x[:, :4]).detach() * 0 | |
| mask = (x[:, :1]).detach() * 0 | |
| loss_cons = 0 | |
| block = [self.block0, self.block1, self.block2] | |
| for i in range(3): | |
| f0, m0 = block[i](torch.cat((warped_img0[:, :3], warped_img1[:, :3], mask), 1), flow, scale=scale_list[i]) | |
| f1, m1 = block[i](torch.cat((warped_img1[:, :3], warped_img0[:, :3], -mask), 1), torch.cat((flow[:, 2:4], flow[:, :2]), 1), scale=scale_list[i]) | |
| flow = flow + (f0 + torch.cat((f1[:, 2:4], f1[:, :2]), 1)) / 2 | |
| mask = mask + (m0 + (-m1)) / 2 | |
| mask_list.append(mask) | |
| flow_list.append(flow) | |
| warped_img0 = warp(img0, flow[:, :2]) | |
| warped_img1 = warp(img1, flow[:, 2:4]) | |
| merged.append((warped_img0, warped_img1)) | |
| ''' | |
| c0 = self.contextnet(img0, flow[:, :2]) | |
| c1 = self.contextnet(img1, flow[:, 2:4]) | |
| tmp = self.unet(img0, img1, warped_img0, warped_img1, mask, flow, c0, c1) | |
| res = tmp[:, 1:4] * 2 - 1 | |
| ''' | |
| for i in range(3): | |
| mask_list[i] = torch.sigmoid(mask_list[i]) | |
| merged[i] = merged[i][0] * mask_list[i] + merged[i][1] * (1 - mask_list[i]) | |
| # merged[i] = torch.clamp(merged[i] + res, 0, 1) | |
| return flow_list, mask_list[2], merged | |