CoordAtt

① 项目背景

• 1.Mobile Network设计的最新研究成果表明，通道注意力（例如，SE注意力）对于提升模型性能具有显著效果，但它们通常会忽略位置信息，而位置信息对于生成空间选择性attention maps是非常重要。
• 2.因此在本文中，作者通过将位置信息嵌入到通道注意力中提出了一种新颖的移动网络注意力机制，将其称为“Coordinate Attention”。与通过2维全局池化将特征张量转换为单个特征向量的通道注意力不同，coordinate注意力将通道注意力分解为两个1维特征编码过程，分别沿2个空间方向聚合特征。
• 3.这样，可以沿一个空间方向捕获远程依赖关系，同时可以沿另一空间方向保留精确的位置信息。然后将生成的特征图分别编码为一对方向感知和位置敏感的attention map，可以将其互补地应用于输入特征图，以增强关注对象的表示。

论文地址：https://arxiv.org/abs/2103.02907

2.1 解压缩数据集

我们将网上获取的数据集以压缩包的方式上传到aistudio数据集中，并加载到我们的项目内。

在使用之前我们进行数据集压缩包的一个解压。

!unzip -oq /home/aistudio/data/data69664/Images.zip -d work/dataset


import paddle
import numpy as np
from typing import Callable
#参数配置
config_parameters = {
"class_dim": 16, #分类数
"target_path":"/home/aistudio/work/",
'train_image_dir': '/home/aistudio/work/trainImages',
'eval_image_dir': '/home/aistudio/work/evalImages',
'epochs':100,
'batch_size': 32,
'lr': 0.01
}


2.2 划分数据集

接下来我们使用标注好的文件进行数据集类的定义，方便后续模型训练使用。

import os
import shutil
train_dir = config_parameters['train_image_dir']
eval_dir = config_parameters['eval_image_dir']
paths = os.listdir('work/dataset/Images')
if not os.path.exists(train_dir):
os.mkdir(train_dir)
if not os.path.exists(eval_dir):
os.mkdir(eval_dir)
for path in paths:
imgs_dir = os.listdir(os.path.join('work/dataset/Images', path))
target_train_dir = os.path.join(train_dir,path)
target_eval_dir = os.path.join(eval_dir,path)
if not os.path.exists(target_train_dir):
os.mkdir(target_train_dir)
if not os.path.exists(target_eval_dir):
os.mkdir(target_eval_dir)
for i in range(len(imgs_dir)):
if ' ' in imgs_dir[i]:
new_name = imgs_dir[i].replace(' ', '_')
else:
new_name = imgs_dir[i]
target_train_path = os.path.join(target_train_dir, new_name)
target_eval_path = os.path.join(target_eval_dir, new_name)
if i % 5 == 0:
shutil.copyfile(os.path.join(os.path.join('work/dataset/Images', path), imgs_dir[i]), target_eval_path)
else:
shutil.copyfile(os.path.join(os.path.join('work/dataset/Images', path), imgs_dir[i]), target_train_path)
print('finished train val split!')


finished train val split!


2.3 数据集定义与数据集展示

2.3.1 数据集展示

我们先看一下解压缩后的数据集长成什么样子,对比分析经典模型在Caltech101抽取16类mini版数据集上的效果


import os
import random
from matplotlib import pyplot as plt
from PIL import Image
imgs = []
paths = os.listdir('work/dataset/Images')
for path in paths:
img_path = os.path.join('work/dataset/Images', path)
if os.path.isdir(img_path):
img_paths = os.listdir(img_path)
img = Image.open(os.path.join(img_path, random.choice(img_paths)))
imgs.append((img, path))
f, ax = plt.subplots(4, 4, figsize=(12,12))
for i, img in enumerate(imgs[:16]):
ax[i//4, i%4].imshow(img[0])
ax[i//4, i%4].axis('off')
ax[i//4, i%4].set_title('label: %s' % img[1])
plt.show()


2.3.2 导入数据集的定义实现

#数据集的定义
class Dataset(paddle.io.Dataset):
"""
步骤一：继承paddle.io.Dataset类
"""
def __init__(self, transforms: Callable, mode: str ='train'):
"""
步骤二：实现构造函数，定义数据读取方式
"""
super(Dataset, self).__init__()

self.mode = mode
self.transforms = transforms
train_image_dir = config_parameters['train_image_dir']
eval_image_dir = config_parameters['eval_image_dir']
train_data_folder = paddle.vision.DatasetFolder(train_image_dir)
eval_data_folder = paddle.vision.DatasetFolder(eval_image_dir)

if self.mode == 'train':
self.data = train_data_folder
elif self.mode == 'eval':
self.data = eval_data_folder
def __getitem__(self, index):
"""
步骤三：实现__getitem__方法，定义指定index时如何获取数据，并返回单条数据（训练数据，对应的标签）
"""
data = np.array(self.data[index][0]).astype('float32')
data = self.transforms(data)
label = np.array([self.data[index][1]]).astype('int64')

return data, label

def __len__(self):
"""
步骤四：实现__len__方法，返回数据集总数目
"""
return len(self.data)


from paddle.vision import transforms as T
#数据增强
transform_train =T.Compose([T.Resize((256,256)),
#T.RandomVerticalFlip(10),
#T.RandomHorizontalFlip(10),
T.RandomRotation(10),
T.Transpose(),
T.Normalize(mean=[0, 0, 0], # 像素值归一化
std =[255, 255, 255]), # transforms.ToTensor(), # transpose操作 + (img / 255),并且数据结构变为PaddleTensor
T.Normalize(mean=[0.50950350, 0.54632660, 0.57409690],# 减均值 除标准差
std= [0.26059777, 0.26041326, 0.29220656])# 计算过程：output[channel] = (input[channel] - mean[channel]) / std[channel]
])
transform_eval =T.Compose([ T.Resize((256,256)),
T.Transpose(),
T.Normalize(mean=[0, 0, 0], # 像素值归一化
std =[255, 255, 255]), # transforms.ToTensor(), # transpose操作 + (img / 255),并且数据结构变为PaddleTensor
T.Normalize(mean=[0.50950350, 0.54632660, 0.57409690],# 减均值 除标准差
std= [0.26059777, 0.26041326, 0.29220656])# 计算过程：output[channel] = (input[channel] - mean[channel]) / std[channel]
])


2.3.3 实例化数据集类

根据所使用的数据集需求实例化数据集类，并查看总样本量。


train_dataset =Dataset(mode='train',transforms=transform_train)
eval_dataset =Dataset(mode='eval', transforms=transform_eval )
#数据异步加载
train_loader = paddle.io.DataLoader(train_dataset,
places=paddle.CUDAPlace(0),
batch_size=32,
shuffle=True,
#num_workers=2,
#use_shared_memory=True
)
eval_loader = paddle.io.DataLoader (eval_dataset,
places=paddle.CUDAPlace(0),
batch_size=32,
#num_workers=2,
#use_shared_memory=True
)
print('训练集样本量: {}，验证集样本量: {}'.format(len(train_loader), len(eval_loader)))


训练集样本量: 45，验证集样本量: 12

③ 模型选择和开发

3.1 对比网络构建

本次我们选取了经典的卷积神经网络resnet50，vgg19,mobilenet_v2来进行实验比较。

network = paddle.vision.models.vgg19(num_classes=16)
#模型封装
model = paddle.Model(network)
#模型可视化
model.summary((-1, 3,256 , 256))


network = paddle.vision.models.resnet50(num_classes=16)
#模型封装
model2 = paddle.Model(network)
#模型可视化
model2.summary((-1, 3,256 , 256))


3.2 对比网络训练

#优化器选择
class SaveBestModel(paddle.callbacks.Callback):
def __init__(self, target=0.5, path='work/best_model', verbose=0):
self.target = target
self.epoch = None
self.path = path
def on_epoch_end(self, epoch, logs=None):
self.epoch = epoch
def on_eval_end(self, logs=None):
if logs.get('acc') > self.target:
self.target = logs.get('acc')
self.model.save(self.path)
print('best acc is {} at epoch {}'.format(self.target, self.epoch))
callback_visualdl = paddle.callbacks.VisualDL(log_dir='work/vgg19')
callback_savebestmodel = SaveBestModel(target=0.5, path='work/best_model')
callbacks = [callback_visualdl, callback_savebestmodel]
base_lr = config_parameters['lr']
epochs = config_parameters['epochs']
def make_optimizer(parameters=None):
momentum = 0.9
learning_rate= paddle.optimizer.lr.CosineAnnealingDecay(learning_rate=base_lr, T_max=epochs, verbose=False)
weight_decay=paddle.regularizer.L2Decay(0.0001)
optimizer = paddle.optimizer.Momentum(
learning_rate=learning_rate,
momentum=momentum,
weight_decay=weight_decay,
parameters=parameters)
return optimizer
optimizer = make_optimizer(model.parameters())
model.prepare(optimizer,
paddle.nn.CrossEntropyLoss(),
paddle.metric.Accuracy())
model.fit(train_loader,
eval_loader,
epochs=100,
batch_size=1, # 是否打乱样本集
callbacks=callbacks,
verbose=1) # 日志展示格式


3.3 Coordinate Attention注意力机制

3.3.1 CA模块的介绍

一个coordinate attention块可以被看作是一个计算单元，旨在增强Mobile Network中特征的表达能力。它可以将任何中间特征张量作为输入并通过转换输出了与张量具有相同size同时具有增强表征的作用。

图1 CA模块细节示意图

import paddle
from paddle.fluid.layers.nn import transpose
import paddle.nn as nn
import math
import paddle.nn.functional as F
class h_sigmoid(nn.Layer):
def __init__(self):
super(h_sigmoid, self).__init__()
self.relu = nn.ReLU6()
def forward(self, x):
return self.relu(x + 3) / 6
class h_swish(nn.Layer):
def __init__(self):
super(h_swish, self).__init__()
self.sigmoid = h_sigmoid()
def forward(self, x):
return x * self.sigmoid(x)
class CoordAtt(nn.Layer):
def __init__(self, inp, oup, reduction=32):
super(CoordAtt, self).__init__()
self.pool_h = nn.AdaptiveAvgPool2D((None, 1))
self.pool_w = nn.AdaptiveAvgPool2D((1, None))
self.sigmoid = nn.Sigmoid()
mip = max(8, inp // reduction)
self.conv1 = nn.Conv2D(inp, mip, kernel_size=1, stride=1, padding=0)
self.bn1 = nn.BatchNorm2D(mip)
self.act = h_swish()

self.conv_h = nn.Conv2D(mip, oup, kernel_size=1, stride=1, padding=0)
self.conv_w = nn.Conv2D(mip, oup, kernel_size=1, stride=1, padding=0)

def forward(self, x):
identity = x
n,c,h,w = x.shape
x_h = self.pool_h(x)
x_w = transpose(self.pool_w(x),[0, 1, 3, 2])
y = paddle.concat([x_h, x_w], axis=2)
y = self.conv1(y)
y = self.bn1(y)
y = self.act(y)

x_h, x_w = paddle.split(y, [h, w], axis=2)
x_w = transpose(x_w,[0, 1, 3, 2])
a_h = self.sigmoid(self.conv_w(x_h))
a_w = self.sigmoid(self.conv_w(x_w))
out = identity * a_w * a_h
return out
if __name__ == '__main__':
x = paddle.randn(shape=[1, 16, 64, 128]) # b, c, h, w
ca_model = CoordAtt(inp=16,oup=16)
y = ca_model(x)
print(y.shape)


W1115 23:29:01.694252 143 device_context.cc:362] Please NOTE: device: 0, GPU Compute Capability: 7.0, Driver API Version: 10.1, Runtime API Version: 10.1
W1115 23:29:01.698771 143 device_context.cc:372] device: 0, cuDNN Version: 7.6.
[1, 16, 64, 128]
/opt/conda/envs/python35-paddle120-env/lib/python3.7/site-packages/paddle/nn/layer/norm.py:648: UserWarning: When training, we now always track global mean and variance.
"When training, we now always track global mean and variance.")


3.3.2 注意力多尺度特征融合卷积神经网络的搭建

import paddle.nn.functional as F
# 构建模型（Inception层）
class Inception(paddle.nn.Layer):
def __init__(self, in_channels, c1, c2, c3, c4):
super(Inception, self).__init__()
# 路线1，卷积核1x1
self.route1x1_1 = paddle.nn.Conv2D(in_channels, c1, kernel_size=1)
# 路线2，卷积层1x1、卷积层3x3
self.route1x1_2 = paddle.nn.Conv2D(in_channels, c2[0], kernel_size=1)
self.route3x3_2 = paddle.nn.Conv2D(c2[0], c2[1], kernel_size=3, padding=1)
# 路线3，卷积层1x1、卷积层5x5
self.route1x1_3 = paddle.nn.Conv2D(in_channels, c3[0], kernel_size=1)
self.route5x5_3 = paddle.nn.Conv2D(c3[0], c3[1], kernel_size=5, padding=2)
# 路线4，池化层3x3、卷积层1x1
self.route3x3_4 = paddle.nn.MaxPool2D(kernel_size=3, stride=1, padding=1)
self.route1x1_4 = paddle.nn.Conv2D(in_channels, c4, kernel_size=1)
def forward(self, x):
route1 = F.relu(self.route1x1_1(x))
route2 = F.relu(self.route3x3_2(F.relu(self.route1x1_2(x))))
route3 = F.relu(self.route5x5_3(F.relu(self.route1x1_3(x))))
route4 = F.relu(self.route1x1_4(self.route3x3_4(x)))
out = [route1, route2, route3, route4]
return paddle.concat(out, axis=1) # 在通道维度(axis=1)上进行连接
# 构建 BasicConv2d 层
def BasicConv2d(in_channels, out_channels, kernel, stride=1, padding=0):
layer = paddle.nn.Sequential(
paddle.nn.Conv2D(in_channels, out_channels, kernel, stride, padding),
paddle.nn.BatchNorm2D(out_channels, epsilon=1e-3),
paddle.nn.ReLU())
return layer
# 搭建网络
class TowerNet(paddle.nn.Layer):
def __init__(self, in_channel, num_classes):
super(TowerNet, self).__init__()
self.b1 = paddle.nn.Sequential(
BasicConv2d(in_channel, out_channels=64, kernel=3, stride=2, padding=1),
paddle.nn.MaxPool2D(2, 2))
self.b2 = paddle.nn.Sequential(
BasicConv2d(64, 128, kernel=3, padding=1),
paddle.nn.MaxPool2D(2, 2))
self.b3 = paddle.nn.Sequential(
BasicConv2d(128, 256, kernel=3, padding=1),
paddle.nn.MaxPool2D(2, 2),
CoordAtt(256,256))
self.b4 = paddle.nn.Sequential(
BasicConv2d(256, 256, kernel=3, padding=1),
paddle.nn.MaxPool2D(2, 2),
CoordAtt(256,256))
self.b5 = paddle.nn.Sequential(
Inception(256, 64, (64, 128), (16, 32), 32),
paddle.nn.MaxPool2D(2, 2),
CoordAtt(256,256),
Inception(256, 64, (64, 128), (16, 32), 32),
paddle.nn.MaxPool2D(2, 2),
CoordAtt(256,256),
Inception(256, 64, (64, 128), (16, 32), 32))
self.AvgPool2D=paddle.nn.AvgPool2D(2)
self.flatten=paddle.nn.Flatten()
self.b6 = paddle.nn.Linear(256, num_classes)
def forward(self, x):
x = self.b1(x)
x = self.b2(x)
x = self.b3(x)
x = self.b4(x)
x = self.b5(x)
x = self.AvgPool2D(x)
x = self.flatten(x)
x = self.b6(x)
return x



model = paddle.Model(TowerNet(3, config_parameters['class_dim']))
model.summary((-1, 3, 256, 256))

④改进模型的训练和优化器的选择

#优化器选择
class SaveBestModel(paddle.callbacks.Callback):
def __init__(self, target=0.5, path='work/best_model', verbose=0):
self.target = target
self.epoch = None
self.path = path
def on_epoch_end(self, epoch, logs=None):
self.epoch = epoch
def on_eval_end(self, logs=None):
if logs.get('acc') > self.target:
self.target = logs.get('acc')
self.model.save(self.path)
print('best acc is {} at epoch {}'.format(self.target, self.epoch))
callback_visualdl = paddle.callbacks.VisualDL(log_dir='work/CA_Inception_Net')
callback_savebestmodel = SaveBestModel(target=0.5, path='work/best_model')
callbacks = [callback_visualdl, callback_savebestmodel]
base_lr = config_parameters['lr']
epochs = config_parameters['epochs']
def make_optimizer(parameters=None):
momentum = 0.9
learning_rate= paddle.optimizer.lr.CosineAnnealingDecay(learning_rate=base_lr, T_max=epochs, verbose=False)
weight_decay=paddle.regularizer.L2Decay(0.0002)
optimizer = paddle.optimizer.Momentum(
learning_rate=learning_rate,
momentum=momentum,
weight_decay=weight_decay,
parameters=parameters)
return optimizer
optimizer = make_optimizer(model.parameters())


model.prepare(optimizer,
paddle.nn.CrossEntropyLoss(),
paddle.metric.Accuracy())


model.fit(train_loader,
eval_loader,
epochs=100,
batch_size=1, # 是否打乱样本集
callbacks=callbacks,
l.parameters())


model.prepare(optimizer,
paddle.nn.CrossEntropyLoss(),
paddle.metric.Accuracy())


model.fit(train_loader,
eval_loader,
epochs=100,
batch_size=1, # 是否打乱样本集
callbacks=callbacks,
verbose=1) # 日志展示格式

⑤模型训练效果展示

在增加了CA模块的注意力机制后，性能有了较大幅度的提升。

• 1.项目中的注意力残差卷积网络CA-Inception-Net模型时采取了学习率分段衰减的方式，对比实验模型采取了同样的方式进行训练。改进的注意力多尺度特征融合卷积神经网络CA-Inception-Net在SRM模块以及残差模块下有了对分类能力的提高。

• 2.在调整模型结构的过程中，重新改进了Inception的结构以及Conv模块的数量，小伙伴们后期可以增大L2正则化项系数和数据增强来抑制过拟合，模型的准确度应该还会增加。