以下链接是个人关于DG-Net(行人重识别ReID)所有见解,如有错误欢迎大家指出,我会第一时间纠正。有兴趣的朋友可以加微信:a944284742相互讨论技术。若是帮助到了你什么,一定要记得点赞!因为这是对我最大的鼓励。
行人重识别0-00:DG-GAN(ReID)-目录-史上最新最全:https://blog.csdn.net/weixin_43013761/article/details/102364512
代码引导
我们依旧回到trainer.py文件,找到如下代码:
#我们使用的是ft_netAB,是代码中Ea编码的过程,也就得到 ap code的过程
# ID_stride,外观编码器池化层的stride
if not 'ID_stride' in hyperparameters.keys():
hyperparameters['ID_stride'] = 2
# hyperparameters['ID_style']默认为'AB',论文中的Ea编码器
if hyperparameters['ID_style']=='PCB':
self.id_a = PCB(ID_class)
elif hyperparameters['ID_style']=='AB':
# 这是我们执行的模型,注意的是,id_a返回两个x(表示身份),获得f,具体介绍看函数内部
self.id_a = ft_netAB(ID_class, stride = hyperparameters['ID_stride'], norm=hyperparameters['norm_id'], pool=hyperparameters['pool'])
else:
self.id_a = ft_net(ID_class, norm=hyperparameters['norm_id'], pool=hyperparameters['pool']) # return 2048 now
# 这里进行的是浅拷贝,所以我认为他们的权重是一起的,可以理解为一个
self.id_b = self.id_a
可以很明确的知道,其和Ea编码相关的操作,都在ft_netAB这个类中,那么我们就来看看把。
ft_netAB详解
# Define the AB Model(Ea)
class ft_netAB(nn.Module):
"""
论文是这样描述Ea的:外观编码器)使用的是基于ResNet50预训练的ImageNet模型,
并且移除了全局平均池化层和全连接层,然后添加了一个合适的max pooling层去输出ap code a(2048x4x1),
然后通过全连接层, 被映射到primary feature f_prim和fine_grained feature f_fine
具体怎么卷积,怎么池化我就不介绍了,注意X开了两个分支,表示身份预测。
"""
def __init__(self, class_num, norm=False, stride=2, droprate=0.5, pool='avg'):
super(ft_netAB, self).__init__()
model_ft = models.resnet50(pretrained=True)
self.part = 4
if pool=='max':
model_ft.partpool = nn.AdaptiveMaxPool2d((self.part,1))
model_ft.avgpool = nn.AdaptiveMaxPool2d((1,1))
else:
model_ft.partpool = nn.AdaptiveAvgPool2d((self.part,1))
model_ft.avgpool = nn.AdaptiveAvgPool2d((1,1))
self.model = model_ft
if stride == 1:
self.model.layer4[0].downsample[0].stride = (1,1)
self.model.layer4[0].conv2.stride = (1,1)
# 对身份进行预测,结合论文的Figure 2,有两个身份
self.classifier1 = ClassBlock(2048, class_num, 0.5)
self.classifier2 = ClassBlock(2048, class_num, 0.75)
# x[batch,3,256,128]
def forward(self, x):
"""
下面这一段都是为了ap code,其中包含了
[身份信息]+ [衣服+鞋子+手机+包包等],还没有进行分离
"""
x = self.model.conv1(x)
x = self.model.bn1(x)
x = self.model.relu(x)
x = self.model.maxpool(x)
x = self.model.layer1(x)
x = self.model.layer2(x)
x = self.model.layer3(x)
x = self.model.layer4(x)
# 这里进行分离,获得[衣服+鞋子+手机+包包等]
# f[batch_size, 2048, 4, 1]
f = self.model.partpool(x)
# 相当于resize[batch_size, 2048, 4]
f = f.view(f.size(0),f.size(1)*self.part)
# 这个值后续不再计算梯度
f = f.detach() # no gradient
# 这里进行分离,分离出身份信息
# x[batch_size, 2048, 1, 1]
x = self.model.avgpool(x)
# x[batch_size,2048]
x = x.view(x.size(0), x.size(1))
# 身份信息又进行分离,分离出主要身份信息,和细致身份信息,
# 同时我们对身份的鉴别,也是这两个综合起来考虑的。
# x1[batch_size, class_num] = [batch_size, 751]
x1 = self.classifier1(x)
# x2[batch_size, class_num] = [batch_size, 751]
x2 = self.classifier2(x)
x=[]
x.append(x1)
x.append(x2)
return f, x
我先带大家看论文图中的一句话:
他说re-id鉴别器是别嵌入在生成模块中的,和编码器Ea是共用的。也就是说,编码器,不仅仅是编码器,其还是ReID行人从识别的模型(着重注意),代码注释也比较详细,就不讲解了。主要注意一个点,就这里进行了两次分离:
第一次分离:apcode 分离成 x[身份信息], f[衣服+鞋子+手机+包包等]信息的分离。
第二次分离:x[身份信息]分离成,主要身份信息,以及细致身份信息
至于他们分离的原理,当然是loss的定义了,后续有详细的分析。
到目前为止,下图红框:
的部分,已经全部讲解完成了,接下来我们看看鉴别器,看看是什么东西。
鉴别器-图片真假鉴别
首先还是回到trainer.py文件,找到如下代码:
# 鉴别器,行人重识别,这里使用的是一个多尺寸的鉴别器,大概就是说,对图片进行几次缩放,并且对每次缩放都会预测,计算总的损失
# 经过网络3个元素,分别大小为[batch_size,1,64,32], [batch_size,1,32,16], [batch_size,1,16,8]
self.dis_a = MsImageDis(3, hyperparameters['dis'], fp16 = False) # discriminator for domain a
self.dis_b = self.dis_a # discriminator for domain b
也算是简单明了,首先 ,要注意的是,这里的鉴别,是鉴别图像的真假,不是身份的鉴别,身份的鉴别是包含在Ea编码器中的。现在进入这个MsImageDis类看看把
class MsImageDis(nn.Module):
"""
论文是这样描述鉴别器的:D(鉴别器)跟随现在流行的多尺度缩放 PatchGAN。
我们采用了不同尺寸的图像进行输入,64x32,128x64,256x128。
也应用了梯度惩罚,当更新D(鉴别器)到稳定的过程中
"""
# Multi-scale discriminator architecture
# input_dim表示输入通道数,默认为3
def __init__(self, input_dim, params, fp16):
super(MsImageDis, self).__init__()
self.n_layer = params['n_layer'] # 鉴别器的层数
self.gan_type = params['gan_type'] # GAN loss [lsgan/nsgan],默认为lsgan
self.dim = params['dim'] # 最后一层的filters数目
self.norm = params['norm'] # 正则化方式,可选择[none/bn/in/ln],默认为none
self.activ = params['activ'] # 激活函数,可选[relu/lrelu/prelu/selu/tanh],默认为lrelu
self.num_scales = params['num_scales'] # 图片缩放的次数,默认为3
self.pad_type = params['pad_type']
self.LAMBDA = params['LAMBDA'] # 正则化的一个超参数
self.non_local = params['non_local'] # 非本地的层数,不知道啥玩意,不用管他
self.n_res = params['n_res'] # 跳跃链接的层数
self.input_dim = input_dim # 这个应该是图片的通道数目,默认为3
self.fp16 = fp16
# 类似于一个下采样的操作
self.downsample = nn.AvgPool2d(3, stride=2, padding=[1, 1], count_include_pad=False)
if not self.gan_type == 'wgan':
self.cnns = nn.ModuleList()
for _ in range(self.num_scales):
Dis = self._make_net()
Dis.apply(weights_init('gaussian'))
self.cnns.append(Dis)
else:
self.cnn = self.one_cnn()
def _make_net(self):
dim = self.dim
cnn_x = []
cnn_x += [Conv2dBlock(self.input_dim, dim, 1, 1, 0, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
cnn_x += [Conv2dBlock(dim, dim, 3, 1, 1, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
cnn_x += [Conv2dBlock(dim, dim, 3, 2, 1, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
for i in range(self.n_layer - 1):
dim2 = min(dim*2, 512)
cnn_x += [Conv2dBlock(dim, dim, 3, 1, 1, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
cnn_x += [Conv2dBlock(dim, dim2, 3, 2, 1, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
dim = dim2
if self.non_local>1:
cnn_x += [NonlocalBlock(dim)]
for i in range(self.n_res):
cnn_x += [ResBlock(dim, norm=self.norm, activation=self.activ, pad_type=self.pad_type, res_type='basic')]
if self.non_local>0:
cnn_x += [NonlocalBlock(dim)]
cnn_x += [nn.Conv2d(dim, 1, 1, 1, 0)]
cnn_x = nn.Sequential(*cnn_x)
return cnn_x
def one_cnn(self):
dim = self.dim
cnn_x = []
cnn_x += [Conv2dBlock(self.input_dim, dim, 4, 2, 1, norm='none', activation=self.activ, pad_type=self.pad_type)]
for i in range(5):
dim2 = min(dim*2, 512)
cnn_x += [Conv2dBlock(dim, dim2, 4, 2, 1, norm=self.norm, activation=self.activ, pad_type=self.pad_type)]
dim = dim2
cnn_x += [nn.Conv2d(dim, 1, (4,2), 1, 0)]
cnn_x = nn.Sequential(*cnn_x)
return cnn_x
# x[4, 3, 256, 128],注意,输出的结果有3个,因为三个尺寸进行预测
def forward(self, x):
if not self.gan_type == 'wgan':
outputs = []
for model in self.cnns:
outputs.append(model(x))
x = self.downsample(x)
else:
outputs = self.cnn(x)
outputs = torch.squeeze(outputs)
# 这里一起3个元素,分别大小为[batch_size,1,64,32], [batch_size,1,32,16], [batch_size,1,16,8]
return outputs
def calc_dis_loss(self, model, input_fake, input_real):
"""
该loss为了训练D,即鉴别器本身
:param model: 为自己本身MsImageDis
:param input_fake: 输入假图片,也就是合成的图片
:param input_real: 输入真图片,训练集里面的图片
:return:
"""
# calculate the loss to train D
input_real.requires_grad_()
# 这里一起3个元素,分别大小为[batch_size, 1,64,32], [batch_size, 1,32,16], [batch_size, 1,16,8]
outs0 = model.forward(input_fake)
# 这里一起3个元素,分别大小为[batch_size, 1,64,32], [batch_size, 1,32,16], [batch_size, 1,16,8]
outs1 = model.forward(input_real)
loss = 0
reg = 0
Drift = 0.001
LAMBDA = self.LAMBDA
# 默认gan_type = 'lsgan',即没有执行这里
if self.gan_type == 'wgan':
loss += torch.mean(outs0) - torch.mean(outs1)
# progressive gan
loss += Drift*( torch.sum(outs0**2) + torch.sum(outs1**2))
#alpha = torch.FloatTensor(input_fake.shape).uniform_(0., 1.)
#alpha = alpha.cuda()
#differences = input_fake - input_real
#interpolates = Variable(input_real + (alpha*differences), requires_grad=True)
#dis_interpolates = self.forward(interpolates)
#gradient_penalty = self.compute_grad2(dis_interpolates, interpolates).mean()
#reg += LAMBDA*gradient_penalty
reg += LAMBDA* self.compute_grad2(outs1, input_real).mean() # I suggest Lambda=0.1 for wgan
loss = loss + reg
return loss, reg
for it, (out0, out1) in enumerate(zip(outs0, outs1)):
# 默认gan_type == 'lsgan',最小二乘损失方式,主要解决生成图像不稳定的问题
if self.gan_type == 'lsgan':
loss += torch.mean((out0 - 0)**2) + torch.mean((out1 - 1)**2)
# regularization
reg += LAMBDA* self.compute_grad2(out1, input_real).mean()
elif self.gan_type == 'nsgan':
all0 = Variable(torch.zeros_like(out0.data).cuda(), requires_grad=False)
all1 = Variable(torch.ones_like(out1.data).cuda(), requires_grad=False)
loss += torch.mean(F.binary_cross_entropy(F.sigmoid(out0), all0) +
F.binary_cross_entropy(F.sigmoid(out1), all1))
reg += LAMBDA* self.compute_grad2(F.sigmoid(out1), input_real).mean()
else:
assert 0, "Unsupported GAN type: {}".format(self.gan_type)
loss = loss+reg
return loss, reg
def calc_gen_loss(self, model, input_fake):
"""
:param model: 为自己本身MsImageDis
:param input_fake: 输入假的图片
:return:
"""
# calculate the loss to train G
# 生成图片,初一这里的输出还是有3个尺寸
outs0 = model.forward(input_fake)
loss = 0
Drift = 0.001
# 该处不执行,因为gan_type == 'lsgan'
if self.gan_type == 'wgan':
loss += -torch.mean(outs0)
# progressive gan
loss += Drift*torch.sum(outs0**2)
return loss
# 同理我们使用的是gan_type == 'lsgan'
for it, (out0) in enumerate(outs0):
if self.gan_type == 'lsgan':
loss += torch.mean((out0 - 1)**2) * 2 # LSGAN
elif self.gan_type == 'nsgan':
all1 = Variable(torch.ones_like(out0.data).cuda(), requires_grad=False)
loss += torch.mean(F.binary_cross_entropy(F.sigmoid(out0), all1))
else:
assert 0, "Unsupported GAN type: {}".format(self.gan_type)
return loss
# 计算梯度
def compute_grad2(self, d_out, x_in):
batch_size = x_in.size(0)
# 这是一个对输出自动求导数的函数,这里表示对outputs=d_out.sum()求inputs=x_in的导数
grad_dout = torch.autograd.grad(
outputs=d_out.sum(), inputs=x_in,
create_graph=True, retain_graph=True, only_inputs=True
)[0]
grad_dout2 = grad_dout.pow(2)
assert(grad_dout2.size() == x_in.size())
reg = grad_dout2.view(batch_size, -1).sum(1)
return reg
代码也是比较详了,该处要注意的是,我们输入图的是的是[batch_size,3,256,128]得到的是三个特征向量[batch_size, 1,64,32], [batch_size, 1,32,16], [batch_size, 1,16,8],是需要一起计算损失的。
计算损失的有两个函数,分别为calc_dis_loss(),calc_gen_loss()。下小节我们就拿这两个损失函数开刀吧,今天大家好好休息,明天继续嗨皮!别忘记点赞奥,毕竟都被我一直忽悠到这里来了嘛。