|
import torch
|
|
import numpy as np
|
|
import torch.nn as nn
|
|
from typing import List
|
|
from torch import Tensor
|
|
import torch.nn.functional as F
|
|
from torchvision.models.resnet import (
|
|
BasicBlock,
|
|
model_urls,
|
|
load_state_dict_from_url,
|
|
conv1x1,
|
|
conv3x3,
|
|
)
|
|
|
|
device = torch.device("cuda")
|
|
|
|
|
|
class CustomResNet(nn.Module):
|
|
def __init__(
|
|
self,
|
|
layers: List[int],
|
|
block=BasicBlock,
|
|
zero_init_residual=False,
|
|
groups=1,
|
|
num_classes=1000,
|
|
width_per_group=64,
|
|
replace_stride_with_dilation=None,
|
|
norm_layer=None,
|
|
):
|
|
|
|
super().__init__()
|
|
|
|
if norm_layer is None:
|
|
self._norm_layer = nn.BatchNorm2d
|
|
|
|
self.inplanes = 64
|
|
self.dilation = 1
|
|
|
|
if replace_stride_with_dilation is None:
|
|
|
|
|
|
replace_stride_with_dilation = [False, False, False]
|
|
|
|
if len(replace_stride_with_dilation) != 3:
|
|
raise ValueError(
|
|
"replace_stride_with_dilation should be None "
|
|
f"or a 3-element tuple, got {replace_stride_with_dilation}"
|
|
)
|
|
|
|
self.groups = groups
|
|
self.base_width = width_per_group
|
|
|
|
self.conv1 = nn.Conv2d(
|
|
3, self.inplanes, kernel_size=7, stride=2, padding=3, bias=False
|
|
)
|
|
self.bn1 = self._norm_layer(self.inplanes)
|
|
self.relu = nn.ReLU(inplace=True)
|
|
self.maxpool = nn.MaxPool2d(kernel_size=3, stride=(2, 1), padding=1)
|
|
self.layer1 = self._make_layer(block, 64, layers[0])
|
|
self.layer2 = self._make_layer(
|
|
block, 128, layers[1], stride=(2, 1), dilate=replace_stride_with_dilation[0]
|
|
)
|
|
self.layer3 = self._make_layer(
|
|
block, 256, layers[2], stride=(2, 2), dilate=replace_stride_with_dilation[1]
|
|
)
|
|
self.layer4 = self._make_layer(
|
|
block, 512, layers[3], stride=(2, 1), dilate=replace_stride_with_dilation[2]
|
|
)
|
|
self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
|
|
self.fc = nn.Linear(512 * block.expansion, num_classes)
|
|
|
|
for m in self.modules():
|
|
if isinstance(m, nn.Conv2d):
|
|
nn.init.kaiming_normal_(m.weight, mode="fan_out", nonlinearity="relu")
|
|
elif isinstance(m, (nn.BatchNorm2d, nn.GroupNorm)):
|
|
nn.init.constant_(m.weight, 1)
|
|
nn.init.constant_(m.bias, 0)
|
|
|
|
|
|
|
|
|
|
if zero_init_residual:
|
|
for m in self.modules():
|
|
if isinstance(m, BasicBlock):
|
|
nn.init.constant_(m.bn2.weight, 0)
|
|
|
|
def _make_layer(
|
|
self,
|
|
block,
|
|
planes,
|
|
blocks,
|
|
stride=1,
|
|
dilate=False,
|
|
) -> nn.Sequential:
|
|
norm_layer = self._norm_layer
|
|
downsample = None
|
|
previous_dilation = self.dilation
|
|
if dilate:
|
|
self.dilation *= stride
|
|
stride = 1
|
|
if stride != 1 or self.inplanes != planes * block.expansion:
|
|
downsample = nn.Sequential(
|
|
conv1x1(self.inplanes, planes * block.expansion, stride),
|
|
norm_layer(planes * block.expansion),
|
|
)
|
|
|
|
layers = []
|
|
layers.append(
|
|
block(
|
|
self.inplanes,
|
|
planes,
|
|
stride,
|
|
downsample,
|
|
self.groups,
|
|
self.base_width,
|
|
previous_dilation,
|
|
norm_layer,
|
|
)
|
|
)
|
|
self.inplanes = planes * block.expansion
|
|
for _ in range(1, blocks):
|
|
layers.append(
|
|
block(
|
|
self.inplanes,
|
|
planes,
|
|
groups=self.groups,
|
|
base_width=self.base_width,
|
|
dilation=self.dilation,
|
|
norm_layer=norm_layer,
|
|
)
|
|
)
|
|
|
|
return nn.Sequential(*layers)
|
|
|
|
def _forward_impl(self, x: Tensor) -> Tensor:
|
|
|
|
x = self.conv1(x)
|
|
x = self.bn1(x)
|
|
x = self.relu(x)
|
|
x = self.maxpool(x)
|
|
|
|
x = self.layer1(x)
|
|
x = self.layer2(x)
|
|
x = self.layer3(x)
|
|
x = self.layer4(x)
|
|
return x
|
|
|
|
def forward(self, x: Tensor) -> Tensor:
|
|
return self._forward_impl(x)
|
|
|
|
|
|
def _resnet(layers: List[int], pretrained=True) -> CustomResNet:
|
|
model = CustomResNet(layers)
|
|
|
|
if pretrained:
|
|
model.load_state_dict(load_state_dict_from_url(model_urls["resnet34"]))
|
|
|
|
return model
|
|
|
|
|
|
def resnet34(*, pretrained=True) -> CustomResNet:
|
|
"""ResNet-34 from `Deep Residual Learning for Image Recognition <https://arxiv.org/pdf/1512.03385.pdf>`__.
|
|
Args:
|
|
weights (:class:`~torchvision.models.ResNet34_Weights`, optional): The
|
|
pretrained weights to use. See
|
|
:class:`~torchvision.models.ResNet34_Weights` below for
|
|
more details, and possible values. By default, no pre-trained
|
|
weights are used.
|
|
progress (bool, optional): If True, displays a progress bar of the
|
|
download to stderr. Default is True.
|
|
**kwargs: parameters passed to the ``torchvision.models.resnet.ResNet``
|
|
base class. Please refer to the `source code
|
|
<https://github.com/pytorch/vision/blob/main/torchvision/models/resnet.py>`_
|
|
for more details about this class.
|
|
.. autoclass:: torchvision.models.ResNet34_Weights
|
|
:members:
|
|
"""
|
|
|
|
return _resnet([3, 4, 6, 3], pretrained=pretrained)
|
|
|
|
|
|
class ResNetFeatureExtractor(nn.Module):
|
|
"""
|
|
Defines Base ResNet-34 feature extractor
|
|
"""
|
|
|
|
def __init__(self, pretrained=True):
|
|
"""
|
|
---------
|
|
Arguments
|
|
---------
|
|
pretrained : bool (default=True)
|
|
boolean to indicate whether to use a pretrained resnet model or not
|
|
"""
|
|
super().__init__()
|
|
self.output_channels = 512
|
|
self.resnet34 = resnet34(pretrained=pretrained)
|
|
|
|
def forward(self, x):
|
|
block1 = self.resnet34.conv1(x)
|
|
block1 = self.resnet34.bn1(block1)
|
|
block1 = self.resnet34.relu(block1)
|
|
|
|
block2 = self.resnet34.maxpool(block1)
|
|
block2 = self.resnet34.layer1(block2)
|
|
block3 = self.resnet34.layer2(block2)
|
|
block4 = self.resnet34.layer3(block3)
|
|
resnet_features = self.resnet34.layer4(block4)
|
|
|
|
|
|
return resnet_features
|
|
|
|
|
|
|
|
|
|
|
|
class TPS_SpatialTransformerNetwork(nn.Module):
|
|
"""Rectification Network of RARE, namely TPS based STN"""
|
|
|
|
def __init__(self, num_fiducial_points, I_size, I_r_size, I_channel_num=1):
|
|
"""Based on RARE TPS
|
|
input:
|
|
batch_I: Batch Input Image [batch_size x I_channel_num x I_height x I_width]
|
|
I_size : (height, width) of the input image I
|
|
I_r_size : (height, width) of the rectified image I_r
|
|
I_channel_num : the number of channels of the input image I
|
|
output:
|
|
batch_I_r: rectified image [batch_size x I_channel_num x I_r_height x I_r_width]
|
|
"""
|
|
super(TPS_SpatialTransformerNetwork, self).__init__()
|
|
self.num_fiducial_points = num_fiducial_points
|
|
self.I_size = I_size
|
|
self.I_r_size = I_r_size
|
|
self.I_channel_num = I_channel_num
|
|
self.LocalizationNetwork = LocalizationNetwork(
|
|
self.num_fiducial_points, self.I_channel_num
|
|
)
|
|
self.GridGenerator = GridGenerator(self.num_fiducial_points, self.I_r_size)
|
|
|
|
def forward(self, batch_I):
|
|
batch_C_prime = self.LocalizationNetwork(batch_I)
|
|
build_P_prime = self.GridGenerator.build_P_prime(
|
|
batch_C_prime
|
|
)
|
|
build_P_prime_reshape = build_P_prime.reshape(
|
|
[build_P_prime.size(0), self.I_r_size[0], self.I_r_size[1], 2]
|
|
)
|
|
|
|
if torch.__version__ > "1.2.0":
|
|
batch_I_r = F.grid_sample(
|
|
batch_I,
|
|
build_P_prime_reshape,
|
|
padding_mode="border",
|
|
align_corners=True,
|
|
)
|
|
else:
|
|
batch_I_r = F.grid_sample(
|
|
batch_I, build_P_prime_reshape, padding_mode="border"
|
|
)
|
|
|
|
return batch_I_r
|
|
|
|
|
|
class LocalizationNetwork(nn.Module):
|
|
"""Localization Network of RARE, which predicts C' (K x 2) from I (I_width x I_height)"""
|
|
|
|
def __init__(self, num_fiducial_points, I_channel_num):
|
|
super(LocalizationNetwork, self).__init__()
|
|
self.num_fiducial_points = num_fiducial_points
|
|
self.I_channel_num = I_channel_num
|
|
self.conv = nn.Sequential(
|
|
nn.Conv2d(
|
|
in_channels=self.I_channel_num,
|
|
out_channels=64,
|
|
kernel_size=3,
|
|
stride=1,
|
|
padding=1,
|
|
bias=False,
|
|
),
|
|
nn.BatchNorm2d(64),
|
|
nn.ReLU(True),
|
|
nn.MaxPool2d(2, 2),
|
|
nn.Conv2d(64, 128, 3, 1, 1, bias=False),
|
|
nn.BatchNorm2d(128),
|
|
nn.ReLU(True),
|
|
nn.MaxPool2d(2, 2),
|
|
nn.Conv2d(128, 256, 3, 1, 1, bias=False),
|
|
nn.BatchNorm2d(256),
|
|
nn.ReLU(True),
|
|
nn.MaxPool2d(2, 2),
|
|
nn.Conv2d(256, 512, 3, 1, 1, bias=False),
|
|
nn.BatchNorm2d(512),
|
|
nn.ReLU(True),
|
|
nn.AdaptiveAvgPool2d(1),
|
|
)
|
|
|
|
self.localization_fc1 = nn.Sequential(nn.Linear(512, 256), nn.ReLU(True))
|
|
self.localization_fc2 = nn.Linear(256, self.num_fiducial_points * 2)
|
|
|
|
|
|
self.localization_fc2.weight.data.fill_(0)
|
|
""" see RARE paper Fig. 6 (a) """
|
|
ctrl_pts_x = np.linspace(-1.0, 1.0, int(num_fiducial_points / 2))
|
|
ctrl_pts_y_top = np.linspace(0.0, -1.0, num=int(num_fiducial_points / 2))
|
|
ctrl_pts_y_bottom = np.linspace(1.0, 0.0, num=int(num_fiducial_points / 2))
|
|
ctrl_pts_top = np.stack([ctrl_pts_x, ctrl_pts_y_top], axis=1)
|
|
ctrl_pts_bottom = np.stack([ctrl_pts_x, ctrl_pts_y_bottom], axis=1)
|
|
initial_bias = np.concatenate([ctrl_pts_top, ctrl_pts_bottom], axis=0)
|
|
self.localization_fc2.bias.data = (
|
|
torch.from_numpy(initial_bias).float().view(-1)
|
|
)
|
|
|
|
def forward(self, batch_I):
|
|
"""
|
|
input: batch_I : Batch Input Image [batch_size x I_channel_num x I_height x I_width]
|
|
output: batch_C_prime : Predicted coordinates of fiducial points for input batch [batch_size x F x 2]
|
|
"""
|
|
batch_size = batch_I.size(0)
|
|
features = self.conv(batch_I).view(batch_size, -1)
|
|
batch_C_prime = self.localization_fc2(self.localization_fc1(features)).view(
|
|
batch_size, self.num_fiducial_points, 2
|
|
)
|
|
return batch_C_prime
|
|
|
|
|
|
class GridGenerator(nn.Module):
|
|
"""Grid Generator of RARE, which produces P_prime by multipling T with P"""
|
|
|
|
def __init__(self, num_fiducial_points, I_r_size):
|
|
"""Generate P_hat and inv_delta_C for later"""
|
|
super(GridGenerator, self).__init__()
|
|
self.eps = 1e-6
|
|
self.I_r_height, self.I_r_width = I_r_size
|
|
self.num_fiducial_points = num_fiducial_points
|
|
self.C = self._build_C(self.num_fiducial_points)
|
|
self.P = self._build_P(self.I_r_width, self.I_r_height)
|
|
|
|
self.register_buffer(
|
|
"inv_delta_C",
|
|
torch.tensor(
|
|
self._build_inv_delta_C(self.num_fiducial_points, self.C)
|
|
).float(),
|
|
)
|
|
self.register_buffer(
|
|
"P_hat",
|
|
torch.tensor(
|
|
self._build_P_hat(self.num_fiducial_points, self.C, self.P)
|
|
).float(),
|
|
)
|
|
|
|
|
|
|
|
|
|
def _build_C(self, F):
|
|
"""Return coordinates of fiducial points in I_r; C"""
|
|
ctrl_pts_x = np.linspace(-1.0, 1.0, int(F / 2))
|
|
ctrl_pts_y_top = -1 * np.ones(int(F / 2))
|
|
ctrl_pts_y_bottom = np.ones(int(F / 2))
|
|
ctrl_pts_top = np.stack([ctrl_pts_x, ctrl_pts_y_top], axis=1)
|
|
ctrl_pts_bottom = np.stack([ctrl_pts_x, ctrl_pts_y_bottom], axis=1)
|
|
C = np.concatenate([ctrl_pts_top, ctrl_pts_bottom], axis=0)
|
|
return C
|
|
|
|
def _build_inv_delta_C(self, F, C):
|
|
"""Return inv_delta_C which is needed to calculate T"""
|
|
hat_C = np.zeros((F, F), dtype=float)
|
|
for i in range(0, F):
|
|
for j in range(i, F):
|
|
r = np.linalg.norm(C[i] - C[j])
|
|
hat_C[i, j] = r
|
|
hat_C[j, i] = r
|
|
np.fill_diagonal(hat_C, 1)
|
|
hat_C = (hat_C**2) * np.log(hat_C)
|
|
|
|
delta_C = np.concatenate(
|
|
[
|
|
np.concatenate([np.ones((F, 1)), C, hat_C], axis=1),
|
|
np.concatenate([np.zeros((2, 3)), np.transpose(C)], axis=1),
|
|
np.concatenate([np.zeros((1, 3)), np.ones((1, F))], axis=1),
|
|
],
|
|
axis=0,
|
|
)
|
|
inv_delta_C = np.linalg.inv(delta_C)
|
|
return inv_delta_C
|
|
|
|
def _build_P(self, I_r_width, I_r_height):
|
|
I_r_grid_x = (
|
|
np.arange(-I_r_width, I_r_width, 2) + 1.0
|
|
) / I_r_width
|
|
I_r_grid_y = (
|
|
np.arange(-I_r_height, I_r_height, 2) + 1.0
|
|
) / I_r_height
|
|
P = np.stack(
|
|
np.meshgrid(I_r_grid_x, I_r_grid_y), axis=2
|
|
)
|
|
return P.reshape([-1, 2])
|
|
|
|
def _build_P_hat(self, F, C, P):
|
|
n = P.shape[0]
|
|
P_tile = np.tile(
|
|
np.expand_dims(P, axis=1), (1, F, 1)
|
|
)
|
|
C_tile = np.expand_dims(C, axis=0)
|
|
P_diff = P_tile - C_tile
|
|
rbf_norm = np.linalg.norm(P_diff, ord=2, axis=2, keepdims=False)
|
|
rbf = np.multiply(np.square(rbf_norm), np.log(rbf_norm + self.eps))
|
|
P_hat = np.concatenate([np.ones((n, 1)), P, rbf], axis=1)
|
|
return P_hat
|
|
|
|
def build_P_prime(self, batch_C_prime):
|
|
"""Generate Grid from batch_C_prime [batch_size x F x 2]"""
|
|
batch_size = batch_C_prime.size(0)
|
|
batch_inv_delta_C = self.inv_delta_C.repeat(batch_size, 1, 1)
|
|
batch_P_hat = self.P_hat.repeat(batch_size, 1, 1)
|
|
batch_C_prime_with_zeros = torch.cat(
|
|
(batch_C_prime, torch.zeros(batch_size, 3, 2).float().to(device)), dim=1
|
|
)
|
|
batch_T = torch.bmm(
|
|
batch_inv_delta_C, batch_C_prime_with_zeros
|
|
)
|
|
batch_P_prime = torch.bmm(batch_P_hat, batch_T)
|
|
return batch_P_prime
|
|
|
|
|
|
"""
|
|
########################################
|
|
######## Pyramid Pooling Block #########
|
|
########################################
|
|
class PyramidPool(nn.Module):
|
|
def __init__(self, pool_kernel_size, in_channels, out_channels):
|
|
super().__init__()
|
|
self.pool_kernel_size = pool_kernel_size
|
|
self.avg_pool_block = nn.Sequential(
|
|
nn.AvgPool2d((1, self.pool_kernel_size), stride=(1, self.pool_kernel_size)),
|
|
nn.Conv2d(in_channels, out_channels, kernel_size=1, stride=1, padding="same", bias=False),
|
|
nn.BatchNorm2d(out_channels),
|
|
nn.ELU(inplace=True),
|
|
)
|
|
|
|
for m in self.modules():
|
|
if isinstance(m, nn.Conv2d):
|
|
nn.init.xavier_normal_(m.weight)
|
|
elif isinstance(m, (nn.BatchNorm2d, nn.GroupNorm)):
|
|
nn.init.constant_(m.weight, 1)
|
|
nn.init.constant_(m.bias, 0)
|
|
|
|
def forward(self, x):
|
|
_, _, in_height, in_width = x.size()
|
|
x = self.avg_pool_block(x)
|
|
x = F.interpolate(x, size=(in_height, in_width), mode="bilinear")
|
|
return x
|
|
|
|
|
|
class PyramidPoolBlock(nn.Module):
|
|
def __init__(self, pyramid_pool_kernel_sizes=[4, 8, 16, 32], num_channels=512):
|
|
super().__init__()
|
|
pp_out_channels = 256
|
|
self.pyramid_pool_layers = nn.ModuleList([PyramidPool(pool_kernel_size=k, in_channels=num_channels, out_channels=pp_out_channels) for k in pyramid_pool_kernel_sizes])
|
|
self.final_layer = nn.Sequential(
|
|
nn.Conv2d((num_channels + (pp_out_channels * len(self.pyramid_pool_layers))), num_channels, (1, 5), stride=1, padding="same"),
|
|
nn.BatchNorm2d(num_channels),
|
|
nn.ELU(inplace=True),
|
|
nn.Dropout(p=0.1),
|
|
)
|
|
|
|
def forward(self, input):
|
|
pp_outputs = []
|
|
for pp_layer in self.pyramid_pool_layers:
|
|
pp_output = pp_layer(input)
|
|
pp_outputs.append(pp_output)
|
|
pp_outputs.append(input)
|
|
x = torch.cat(pp_outputs, dim=1)
|
|
x = self.final_layer(x)
|
|
return x
|
|
"""
|
|
|