# by:迪菲赫尔曼
import warnings
import torch
from torch import nn
("ignore")
"""
This code is mainly the deformation process of our DSConv
"""
class DSConv():
def __init__(self, in_ch, out_ch, kernel_size, extend_scope, morph,
if_offset):
"""
动态蛇形卷积
:param in_ch: 输入通道
:param out_ch: 输出通道
:param kernel_size: 卷积核的大小
:param extend_scope: 扩展范围（默认为此方法的1）
:param morph: 卷积核的形态主要分为两种类型，沿x轴（0）和沿y轴（1）（详细信息请参阅论文）
:param if_offset: 是否需要变形，如果为False，则是标准卷积核
"""
super(DSConv, self).__init__()
# use the <offset_conv> to learn the deformable offset
# offset_conv: 学习可变形偏移的卷积层
self.offset_conv = nn.Conv2d(in_ch, 2 * kernel_size, 3, padding=1)
= nn.BatchNorm2d(2 * kernel_size)
self.kernel_size = kernel_size
# two types of the DSConv (along x-axis and y-axis)
# dsc_conv_x 和 dsc_conv_y：两种动态蛇形卷积层，分别沿x轴和y轴。
self.dsc_conv_x = nn.Conv2d(
in_ch,
out_ch,
kernel_size=(kernel_size, 1),
stride=(kernel_size, 1),
padding=0,
)
self.dsc_conv_y = nn.Conv2d(
in_ch,
out_ch,
kernel_size=(1, kernel_size),
stride=(1, kernel_size),
padding=0,
)
# gn：组归一化层
= (out_ch // 4, out_ch)
= (inplace=True)
# extend_scope：扩展范围
self.extend_scope = extend_scope
# morph：卷积核形态的类型
= morph
# if_offset：指示是否需要变形的布尔值
self.if_offset = if_offset
def forward(self, f):
offset = self.offset_conv(f)
offset = (offset)
# We need a range of deformation between -1 and 1 to mimic the snake's swing
offset = (offset)
input_shape =
dsc = DSC(input_shape, self.kernel_size, self.extend_scope, )
deformed_feature = dsc.deform_conv(f, offset, self.if_offset)
if == 0:
x = self.dsc_conv_x(deformed_feature.type())
x = (x)
x = (x)
return x
else:
x = self.dsc_conv_y(deformed_feature.type())
x = (x)
x = (x)
return x
# Core code, for ease of understanding, we mark the dimensions of input and output next to the code
class DSC(object):
def __init__(self, input_shape, kernel_size, extend_scope, morph):
self.num_points = kernel_size
= input_shape[2]
= input_shape[3]
= morph
self.extend_scope = extend_scope # offset (-1 ~ 1) * extend_scope
# define feature map shape
"""
B: Batch size C: Channel W: Width H: Height
"""
self.num_batch = input_shape[0]
self.num_channels = input_shape[1]
"""
input: offset [B,2*K,W,H] K: Kernel size (2*K: 2D image, deformation contains <x_offset> and <y_offset>)
output_x: [B,1,W,K*H] coordinate map
output_y: [B,1,K*W,H] coordinate map
"""
def _coordinate_map_3D(self, offset, if_offset):
"""
1.输入为偏移 (offset) 和是否需要偏移 (if_offset)。
2.根据输入特征图的形状、卷积核大小、扩展范围以及形态类型，生成二维坐标映射。
3.如果形态类型为0，表示沿x轴，生成y坐标映射；如果形态类型为1，表示沿y轴，生成x坐标映射。
4.根据偏移和扩展范围调整坐标映射。
5.返回生成的坐标映射。
"""
device =
# offset
y_offset, x_offset = (offset, self.num_points, dim=1)
y_center = (0, ).repeat([])
y_center = y_center.reshape(, )
y_center = y_center.permute(1, 0)
y_center = y_center.reshape([-1, , ])
y_center = y_center.repeat([self.num_points, 1, 1]).float()
y_center = y_center.unsqueeze(0)
x_center = (0, ).repeat([])
x_center = x_center.reshape(, )
x_center = x_center.permute(0, 1)
x_center = x_center.reshape([-1, , ])
x_center = x_center.repeat([self.num_points, 1, 1]).float()
x_center = x_center.unsqueeze(0)
if == 0:
"""
Initialize the kernel and flatten the kernel
y: only need 0
x: -num_points//2 ~ num_points//2 (Determined by the kernel size)
!!! The related PPT will be submitted later, and the PPT will contain the whole changes of each step
"""
y = (0, 0, 1)
x = (
-int(self.num_points // 2),
int(self.num_points // 2),
int(self.num_points),
)
y, x = (y, x)
y_spread = (-1, 1)
x_spread = (-1, 1)
y_grid = y_spread.repeat([1, * ])
y_grid = y_grid.reshape([self.num_points, , ])
y_grid = y_grid.unsqueeze(0) # [B*K*K, W,H]
x_grid = x_spread.repeat([1, * ])
x_grid = x_grid.reshape([self.num_points, , ])
x_grid = x_grid.unsqueeze(0) # [B*K*K, W,H]
y_new = y_center + y_grid
x_new = x_center + x_grid
y_new = y_new.repeat(self.num_batch, 1, 1, 1).to(device)
x_new = x_new.repeat(self.num_batch, 1, 1, 1).to(device)
y_offset_new = y_offset.detach().clone()
if if_offset:
y_offset = y_offset.permute(1, 0, 2, 3)
y_offset_new = y_offset_new.permute(1, 0, 2, 3)
center = int(self.num_points // 2)
# The center position remains unchanged and the rest of the positions begin to swing
# This part is quite simple. The main idea is that "offset is an iterative process"
y_offset_new[center] = 0
for index in range(1, center):
y_offset_new[center + index] = (y_offset_new[center + index - 1] + y_offset[center + index])
y_offset_new[center - index] = (y_offset_new[center - index + 1] + y_offset[center - index])
y_offset_new = y_offset_new.permute(1, 0, 2, 3).to(device)
y_new = y_new.add(y_offset_new.mul(self.extend_scope))
y_new = y_new.reshape(
[self.num_batch, self.num_points, 1, , ])
y_new = y_new.permute(0, 3, 1, 4, 2)
y_new = y_new.reshape([
self.num_batch, self.num_points * , 1 *
])
x_new = x_new.reshape(
[self.num_batch, self.num_points, 1, , ])
x_new = x_new.permute(0, 3, 1, 4, 2)
x_new = x_new.reshape([
self.num_batch, self.num_points * , 1 *
])
return y_new, x_new
else:
"""
Initialize the kernel and flatten the kernel
y: -num_points//2 ~ num_points//2 (Determined by the kernel size)
x: only need 0
"""
y = (
-int(self.num_points // 2),
int(self.num_points // 2),
int(self.num_points),
)
x = (0, 0, 1)
y, x = (y, x)
y_spread = (-1, 1)
x_spread = (-1, 1)
y_grid = y_spread.repeat([1, * ])
y_grid = y_grid.reshape([self.num_points, , ])
y_grid = y_grid.unsqueeze(0)
x_grid = x_spread.repeat([1, * ])
x_grid = x_grid.reshape([self.num_points, , ])
x_grid = x_grid.unsqueeze(0)
y_new = y_center + y_grid
x_new = x_center + x_grid
y_new = y_new.repeat(self.num_batch, 1, 1, 1)
x_new = x_new.repeat(self.num_batch, 1, 1, 1)
y_new = y_new.to(device)
x_new = x_new.to(device)
x_offset_new = x_offset.detach().clone()
if if_offset:
x_offset = x_offset.permute(1, 0, 2, 3)
x_offset_new = x_offset_new.permute(1, 0, 2, 3)
center = int(self.num_points // 2)
x_offset_new[center] = 0
for index in range(1, center):
x_offset_new[center + index] = (x_offset_new[center + index - 1] + x_offset[center + index])
x_offset_new[center - index] = (x_offset_new[center - index + 1] + x_offset[center - index])
x_offset_new = x_offset_new.permute(1, 0, 2, 3).to(device)
x_new = x_new.add(x_offset_new.mul(self.extend_scope))
y_new = y_new.reshape(
[self.num_batch, 1, self.num_points, , ])
y_new = y_new.permute(0, 3, 1, 4, 2)
y_new = y_new.reshape([
self.num_batch, 1 * , self.num_points *
])
x_new = x_new.reshape(
[self.num_batch, 1, self.num_points, , ])
x_new = x_new.permute(0, 3, 1, 4, 2)
x_new = x_new.reshape([
self.num_batch, 1 * , self.num_points *
])
return y_new, x_new
"""
input: input feature map [N,C,D,W,H]；coordinate map [N,K*D,K*W,K*H]
output: [N,1,K*D,K*W,K*H] deformed feature map
"""
def _bilinear_interpolate_3D(self, input_feature, y, x):
"""
1.输入为输入特征图 (input_feature)、y坐标映射 (y) 和x坐标映射 (x)。
2.进行三维双线性插值，获取变形后的特征。
3.返回插值得到的变形特征。
"""
device = input_feature.device
y = ([-1]).float()
x = ([-1]).float()
zero = ([]).int()
max_y = - 1
max_x = - 1
# find 8 grid locations
y0 = (y).int()
y1 = y0 + 1
x0 = (x).int()
x1 = x0 + 1
# clip out coordinates exceeding feature map volume
y0 = (y0, zero, max_y)
y1 = (y1, zero, max_y)
x0 = (x0, zero, max_x)
x1 = (x1, zero, max_x)
input_feature_flat = input_feature.flatten()
input_feature_flat = input_feature_flat.reshape(
self.num_batch, self.num_channels, , )
input_feature_flat = input_feature_flat.permute(0, 2, 3, 1)
input_feature_flat = input_feature_flat.reshape(-1, self.num_channels)
dimension = *
base = (self.num_batch) * dimension
base = ([-1, 1]).float()
repeat = ([self.num_points * *
]).unsqueeze(0)
repeat = repeat.float()
base = (base, repeat)
base = ([-1])
base = (device)
base_y0 = base + y0 *
base_y1 = base + y1 *
# top rectangle of the neighbourhood volume
index_a0 = base_y0 - base + x0
index_c0 = base_y0 - base + x1
# bottom rectangle of the neighbourhood volume
index_a1 = base_y1 - base + x0
index_c1 = base_y1 - base + x1
# get 8 grid values
value_a0 = input_feature_flat[index_a0.type(torch.int64)].to(device)
value_c0 = input_feature_flat[index_c0.type(torch.int64)].to(device)
value_a1 = input_feature_flat[index_a1.type(torch.int64)].to(device)
value_c1 = input_feature_flat[index_c1.type(torch.int64)].to(device)
# find 8 grid locations
y0 = (y).int()
y1 = y0 + 1
x0 = (x).int()
x1 = x0 + 1
# clip out coordinates exceeding feature map volume
y0 = (y0, zero, max_y + 1)
y1 = (y1, zero, max_y + 1)
x0 = (x0, zero, max_x + 1)
x1 = (x1, zero, max_x + 1)
x0_float = x0.float()
x1_float = x1.float()
y0_float = y0.float()
y1_float = y1.float()
vol_a0 = ((y1_float - y) * (x1_float - x)).unsqueeze(-1).to(device)
vol_c0 = ((y1_float - y) * (x - x0_float)).unsqueeze(-1).to(device)
vol_a1 = ((y - y0_float) * (x1_float - x)).unsqueeze(-1).to(device)
vol_c1 = ((y - y0_float) * (x - x0_float)).unsqueeze(-1).to(device)
outputs = (value_a0 * vol_a0 + value_c0 * vol_c0 + value_a1 * vol_a1 +
value_c1 * vol_c1)
if == 0:
outputs = ([
self.num_batch,
self.num_points * ,
1 * ,
self.num_channels,
])
outputs = (0, 3, 1, 2)
else:
outputs = ([
self.num_batch,
1 * ,
self.num_points * ,
self.num_channels,
])
outputs = (0, 3, 1, 2)
return outputs
def deform_conv(self, input, offset, if_offset):
"""
1.输入为原始特征图 (input)、偏移 (offset) 和是否需要偏移 (if_offset)。
2.调用 _coordinate_map_3D 方法获取坐标映射。
3.调用 _bilinear_interpolate_3D 方法进行双线性插值，得到变形后的特征。
4.返回变形后的特征。
"""
y, x = self._coordinate_map_3D(offset, if_offset)
deformed_feature = self._bilinear_interpolate_3D(input, y, x)
return deformed_feature
#---------------------------------YOLOv5 专用部分↓---------------------------------
class DSConv_Bottleneck():
# DSConv bottleneck
def __init__(self, c1, c2, shortcut=True, g=1, e=0.5): # ch_in, ch_out, shortcut, groups, expansion
super().__init__()
c_ = int(c2 * e) # hidden channels
self.cv1 = Conv(c1, c_, 1, 1)
self.cv2 = Conv(c_, c2, 3, 1, g=g)
= shortcut and c1 == c2
= DSConv(c2, c2, 3, 1, 1, True)
def forward(self, x):
return x + (self.cv2(self.cv1(x))) if else (self.cv2(self.cv1(x)))
class DSConv_C3():
# DSConv Bottleneck with 3 convolutions
def __init__(self, c1, c2, n=1, shortcut=True, g=1, e=0.5): # ch_in, ch_out, number, shortcut, groups, expansion
super().__init__()
c_ = int(c2 * e) # hidden channels
self.cv1 = Conv(c1, c_, 1, 1)
self.cv2 = Conv(c1, c_, 1, 1)
self.cv3 = Conv(2 * c_, c2, 1) # act=FReLU(c2)
= (*(DSConv_Bottleneck(c_, c_, shortcut, g, e=1.0) for _ in range(n)))
def forward(self, x):
return self.cv3((((self.cv1(x)), self.cv2(x)), dim=1))
#---------------------------------YOLOv5 专用部分↑---------------------------------