文章目录
在上一篇文章中完成了前期的准备工作，见链接：
SeaFormer实战：使用SeaFormer实现图像分类任务（一）
这篇主要是讲解如何训练和测试
训练部分
完成上面的步骤后，就开始train脚本的编写，新建train.py
导入项目使用的库
在train.py导入
import json
import os
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.nn.parallel
import torch.optim as optim
import torch.utils.data
import torch.utils.data.distributed
import torchvision.transforms as transforms
from timm.utils import accuracy, AverageMeter, ModelEma
from sklearn.metrics import classification_report
from timm.data.mixup import Mixup
from timm.loss import SoftTargetCrossEntropy
from models.seaformer import SeaFormer_T
from torch.autograd import Variable
from torchvision import datasets
torch.backends.cudnn.benchmark = False
import warnings
warnings.filterwarnings("ignore")
os.environ['CUDA_VISIBLE_DEVICES']="0,1"
os.environ[‘CUDA_VISIBLE_DEVICES’]=“0,1” 选择显卡，index从0开始，比如一台机器上有8块显卡，我们打算使用前两块显卡训练，设置为“0,1”,同理如果打算使用第三块和第六块显卡训练，则设置为“2,5”。
设置随机因子
def seed_everything(seed=42):
os.environ['PYHTONHASHSEED'] = str(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.backends.cudnn.deterministic = True
设置了固定的随机因子，再次训练的时候就可以保证图片的加载顺序不会发生变化。
设置全局参数
if __name__ == '__main__':
#创建保存模型的文件夹
file_dir = 'checkpoints/seaformer/'
if os.path.exists(file_dir):
print('true')
os.makedirs(file_dir,exist_ok=True)
else:
os.makedirs(file_dir)
# 设置全局参数
model_lr = 1e-4
BATCH_SIZE = 16
EPOCHS = 300
DEVICE = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
use_amp = True # 是否使用混合精度
use_dp = True #是否开启dp方式的多卡训练
classes = 12
resume =None
CLIP_GRAD = 5.0
Best_ACC = 0 #记录最高得分
use_ema=True
model_ema_decay=0.9998
start_epoch=1
seed=1
seed_everything(seed)
设置存放权重文件的文件夹，如果文件夹存在删除再建立。
接下来，设置全局参数，比如：学习率、BatchSize、epoch等参数，判断环境中是否存在GPU，如果没有则使用CPU。
注：建议使用GPU，CPU太慢了。
参数的详细解释：
file_dir = 'checkpoints/seaformer'
这是存放seaformer模型的路径。
图像预处理与增强
# 数据预处理7
transform = transforms.Compose([
transforms.RandomRotation(10),
transforms.GaussianBlur(kernel_size=(5,5),sigma=(0.1, 3.0)),
transforms.ColorJitter(brightness=0.5, contrast=0.5, saturation=0.5),
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.44127703, 0.4712498, 0.43714803], std= [0.18507297, 0.18050247, 0.16784933])
])
transform_test = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.44127703, 0.4712498, 0.43714803], std= [0.18507297, 0.18050247, 0.16784933])
])
mixup_fn = Mixup(
mixup_alpha=0.8, cutmix_alpha=1.0, cutmix_minmax=None,
prob=0.1, switch_prob=0.5, mode='batch',
label_smoothing=0.1, num_classes=classes)
数据处理和增强比较简单，加入了随机10度的旋转、高斯模糊、色彩饱和度明亮度的变化、Mixup等比较常用的增强手段，做了Resize和归一化。
这里注意下Resize的大小，由于选用的seaformer模型输入是224×224的大小，所以要Resize为224×224。
读取数据
# 读取数据
dataset_train = datasets.ImageFolder('data/train', transform=transform)
dataset_test = datasets.ImageFolder("data/val", transform=transform_test)
with open('class.txt', 'w') as file:
file.write(str(dataset_train.class_to_idx))
with open('class.json', 'w', encoding='utf-8') as file:
file.write(json.dumps(dataset_train.class_to_idx))
# 导入数据
train_loader = torch.utils.data.DataLoader(dataset_train, batch_size=BATCH_SIZE, pin_memory=True,shuffle=True,drop_last=True)
test_loader = torch.utils.data.DataLoader(dataset_test, batch_size=BATCH_SIZE, pin_memory=True,shuffle=False)
-
使用pytorch默认读取数据的方式，然后将dataset_train.class_to_idx打印出来，预测的时候要用到。
-
对于train_loader ，drop_last设置为True，因为使用了Mixup数据增强，必须保证每个batch里面的图片个数为偶数（不能为零），如果最后一个batch里面的图片为奇数，则会报错，所以舍弃最后batch的迭代。pin_memory设置为True，可以加快运行速度。
-
将dataset_train.class_to_idx保存到txt文件或者json文件中。
class_to_idx的结果：
{'Black-grass': 0, 'Charlock': 1, 'Cleavers': 2, 'Common Chickweed': 3, 'Common wheat': 4, 'Fat Hen': 5, 'Loose Silky-bent': 6, 'Maize': 7, 'Scentless Mayweed': 8, 'Shepherds Purse': 9, 'Small-flowered Cranesbill': 10, 'Sugar beet': 11}
设置Loss
# 实例化模型并且移动到GPU
criterion_train = SoftTargetCrossEntropy()
criterion_val = torch.nn.CrossEntropyLoss()
设置loss函数，训练的loss为：SoftTargetCrossEntropy，验证的loss：nn.CrossEntropyLoss()。
设置模型
#设置模型
model_ft = SeaFormer_T(pretrained=True)
num_fr=model_ft.linear.in_features
model_ft.linear=nn.Linear(num_fr,classes)
print(model_ft)
if resume:
model=torch.load(resume)
print(model['state_dict'].keys())
model_ft.load_state_dict(model['state_dict'])
Best_ACC=model['Best_ACC']
start_epoch=model['epoch']+1
model_ft.to(DEVICE)
- 设置模型为SeaFormer_T，pretrained设置为true,表示加载预训练模型，修改head层，将将输出classes设置为12。
- 如果resume为True，则加载模型接着resume指向的模型接着训练，使用模型里的Best_ACC初始化Best_ACC，使用epoch参数初始化start_epoch.
设置优化器和学习率调整算法
# 选择简单暴力的Adam优化器，学习率调低
optimizer = optim.AdamW(model_ft.parameters(),lr=model_lr)
cosine_schedule = optim.lr_scheduler.CosineAnnealingLR(optimizer=optimizer, T_max=20, eta_min=1e-6)
- 优化器设置为adamW。
- 学习率调整策略选择为余弦退火。
设置混合精度，DP多卡，EMA
if use_amp:
scaler = torch.cuda.amp.GradScaler()
if torch.cuda.device_count() > 1 and use_dp:
print("Let's use", torch.cuda.device_count(), "GPUs!")
model_ft = torch.nn.DataParallel(model_ft)
if use_ema:
model_ema = ModelEma(
model_ft,
decay=model_ema_decay,
device=DEVICE,
resume=resume)
else:
model_ema=None
- use_amp为True，则开启混合精度训练,声明pytorch自带的混合精度 torch.cuda.amp.GradScaler()。
- 检测可用显卡的数量，如果大于1，并且开启多卡训练的情况下，则要用torch.nn.DataParallel加载模型，开启多卡训练。
- 如果使用ema，则注册ema
注：torch.nn.DataParallel方式，默认不能开启混合精度训练的，如果想要开启混合精度训练，则需要在模型的forward前面加上@autocast()函数。
如果不开启混合精度则要将@autocast()去掉，否则loss一直试nan。
定义训练和验证函数
训练函数
# 定义训练过程
def train(model, device, train_loader, optimizer, epoch,model_ema):
model.train()
loss_meter = AverageMeter()
acc1_meter = AverageMeter()
acc5_meter = AverageMeter()
total_num = len(train_loader.dataset)
print(total_num, len(train_loader))
for batch_idx, (data, target) in enumerate(train_loader):
data, target = data.to(device, non_blocking=True), Variable(target).to(device, non_blocking=True)
samples, targets = mixup_fn(data, target)
output = model(samples)
optimizer.zero_grad()
if use_amp:
with torch.cuda.amp.autocast():
loss = torch.nan_to_num(criterion_train(output, targets))
scaler.scale(loss).backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), CLIP_GRAD)
# Unscales gradients and calls
# or skips optimizer.step()
scaler.step(optimizer)
# Updates the scale for next iteration
scaler.update()
else:
loss = criterion_train(output, targets)
loss.backward()
# torch.nn.utils.clip_grad_norm_(model.parameters(), CLIP_GRAD)
optimizer.step()
if model_ema is not None:
model_ema.update(model)
torch.cuda.synchronize()
lr = optimizer.state_dict()['param_groups'][0]['lr']
loss_meter.update(loss.item(), target.size(0))
acc1, acc5 = accuracy(output, target, topk=(1, 5))
loss_meter.update(loss.item(), target.size(0))
acc1_meter.update(acc1.item(), target.size(0))
acc5_meter.update(acc5.item(), target.size(0))
if (batch_idx + 1) % 10 == 0:
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}\tLR:{:.9f}'.format(
epoch, (batch_idx + 1) * len(data), len(train_loader.dataset),
100. * (batch_idx + 1) / len(train_loader), loss.item(), lr))
ave_loss =loss_meter.avg
acc = acc1_meter.avg
print('epoch:{}\tloss:{:.2f}\tacc:{:.2f}'.format(epoch, ave_loss, acc))
return ave_loss, acc
训练的主要步骤：
否则，直接反向传播求梯度。torch.nn.utils.clip_grad_norm_函数执行梯度裁剪，防止梯度爆炸。
7、如果use_ema为True，则执行model_ema的updata函数，更新模型。
8、 torch.cuda.synchronize()，等待上面所有的操作执行完成。
9、接下来，更新loss，ACC1，ACC5的值。
等待一个epoch训练完成后，计算平均loss和平均acc
验证函数
# 验证过程
@torch.no_grad()
def val(model, device, test_loader):
global Best_ACC
model.eval()
loss_meter = AverageMeter()
acc1_meter = AverageMeter()
acc5_meter = AverageMeter()
total_num = len(test_loader.dataset)
print(total_num, len(test_loader))
val_list = []
pred_list = []
for data, target in test_loader:
for t in target:
val_list.append(t.data.item())
data, target = data.to(device,non_blocking=True), target.to(device,non_blocking=True)
output = model(data)
loss = criterion_val(output, target)
_, pred = torch.max(output.data, 1)
for p in pred:
pred_list.append(p.data.item())
acc1, acc5 = accuracy(output, target, topk=(1, 5))
loss_meter.update(loss.item(), target.size(0))
acc1_meter.update(acc1.item(), target.size(0))
acc5_meter.update(acc5.item(), target.size(0))
acc = acc1_meter.avg
print('\nVal set: Average loss: {:.4f}\tAcc1:{:.3f}%\tAcc5:{:.3f}%\n'.format(
loss_meter.avg, acc, acc5_meter.avg))
if acc > Best_ACC:
if isinstance(model, torch.nn.DataParallel):
torch.save(model.module, file_dir + '/' + 'best.pth')
else:
torch.save(model, file_dir + '/' + 'best.pth')
Best_ACC = acc
if isinstance(model, torch.nn.DataParallel):
state = {
'epoch': epoch,
'state_dict': model.module.state_dict(),
'Best_ACC':Best_ACC
}
if use_ema:
state['state_dict_ema']=model.module.state_dict()
torch.save(state, file_dir + "/" + 'model_' + str(epoch) + '_' + str(round(acc, 3)) + '.pth')
else:
state = {
'epoch': epoch,
'state_dict': model.state_dict(),
'Best_ACC': Best_ACC
}
if use_ema:
state['state_dict_ema']=model.state_dict()
torch.save(state, file_dir + "/" + 'model_' + str(epoch) + '_' + str(round(acc, 3)) + '.pth')
return val_list, pred_list, loss_meter.avg, acc
验证集和训练集大致相似，主要步骤：
4、本次epoch循环完成后，求得本次epoch的acc、loss。
5、接下来是保存模型的逻辑
如果ACC比Best_ACC高，则保存best模型
判断模型是否为DP方式训练的模型。
6、接下来保存每个epoch的模型。
判断模型是否为DP方式训练的模型。
注意：对于每个epoch的模型只保存了state_dict参数，没有保存整个模型文件。
调用训练和验证方法
# 训练与验证
is_set_lr = False
log_dir = {}
train_loss_list, val_loss_list, train_acc_list, val_acc_list, epoch_list = [], [], [], [], []
if resume and os.path.isfile(file_dir+"result.json"):
with open(file_dir+'result.json', 'r', encoding='utf-8') as file:
logs = json.load(file)
train_acc_list = logs['train_acc']
train_loss_list = logs['train_loss']
val_acc_list = logs['val_acc']
val_loss_list = logs['val_loss']
epoch_list = logs['epoch_list']
for epoch in range(start_epoch, EPOCHS + 1):
epoch_list.append(epoch)
log_dir['epoch_list'] = epoch_list
train_loss, train_acc = train(model_ft, DEVICE, train_loader, optimizer, epoch,model_ema)
train_loss_list.append(train_loss)
train_acc_list.append(train_acc)
log_dir['train_acc'] = train_acc_list
log_dir['train_loss'] = train_loss_list
if use_ema:
val_list, pred_list, val_loss, val_acc = val(model_ema.ema, DEVICE, test_loader)
else:
val_list, pred_list, val_loss, val_acc = val(model_ft, DEVICE, test_loader)
val_loss_list.append(val_loss)
val_acc_list.append(val_acc)
log_dir['val_acc'] = val_acc_list
log_dir['val_loss'] = val_loss_list
log_dir['best_acc'] = Best_ACC
with open(file_dir + '/result.json', 'w', encoding='utf-8') as file:
file.write(json.dumps(log_dir))
print(classification_report(val_list, pred_list, target_names=dataset_train.class_to_idx))
if epoch < 600:
cosine_schedule.step()
else:
if not is_set_lr:
for param_group in optimizer.param_groups:
param_group["lr"] = 1e-6
is_set_lr = True
fig = plt.figure(1)
plt.plot(epoch_list, train_loss_list, 'r-', label=u'Train Loss')
# 显示图例
plt.plot(epoch_list, val_loss_list, 'b-', label=u'Val Loss')
plt.legend(["Train Loss", "Val Loss"], loc="upper right")
plt.xlabel(u'epoch')
plt.ylabel(u'loss')
plt.title('Model Loss ')
plt.savefig(file_dir + "/loss.png")
plt.close(1)
fig2 = plt.figure(2)
plt.plot(epoch_list, train_acc_list, 'r-', label=u'Train Acc')
plt.plot(epoch_list, val_acc_list, 'b-', label=u'Val Acc')
plt.legend(["Train Acc", "Val Acc"], loc="lower right")
plt.title("Model Acc")
plt.ylabel("acc")
plt.xlabel("epoch")
plt.savefig(file_dir + "/acc.png")
plt.close(2)
调用训练函数和验证函数的主要步骤：
运行以及结果查看
完成上面的所有代码就可以开始运行了。点击右键，然后选择“run train.py”即可，运行结果如下：
在每个epoch测试完成之后，打印验证集的acc、recall等指标。
SeaFormer测试结果：
测试
测试，我们采用一种通用的方式。
测试集存放的目录如下图：
SeaFormer_demo
├─test
│ ├─1.jpg
│ ├─2.jpg
│ ├─3.jpg
│ ├ ......
└─test.py
import torch.utils.data.distributed
import torchvision.transforms as transforms
from PIL import Image
from torch.autograd import Variable
import os
classes = ('Black-grass', 'Charlock', 'Cleavers', 'Common Chickweed',
'Common wheat', 'Fat Hen', 'Loose Silky-bent',
'Maize', 'Scentless Mayweed', 'Shepherds Purse', 'Small-flowered Cranesbill', 'Sugar beet')
transform_test = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.51819474, 0.5250407, 0.4945761], std=[0.24228974, 0.24347611, 0.2530049])
])
DEVICE = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
model=torch.load('checkpoints/SeaFormer/best.pth',map_location='cpu')
model.eval()
model.to(DEVICE)
path = 'test/'
testList = os.listdir(path)
for file in testList:
img = Image.open(path + file)
img = transform_test(img)
img.unsqueeze_(0)
img = Variable(img).to(DEVICE)
out = model(img)
# Predict
_, pred = torch.max(out.data, 1)
print('Image Name:{},predict:{}'.format(file, classes[pred.data.item()]))
测试的主要逻辑：
运行结果：
热力图可视化展示
新建脚本cam_image.py,插入如下代码：
import argparse
import os
import cv2
import numpy as np
import torch
from pytorch_grad_cam import GradCAM, \
ScoreCAM, \
GradCAMPlusPlus, \
AblationCAM, \
XGradCAM, \
EigenCAM, \
EigenGradCAM, \
LayerCAM, \
FullGrad
from pytorch_grad_cam import GuidedBackpropReLUModel
from pytorch_grad_cam.utils.image import show_cam_on_image, \
deprocess_image, \
preprocess_image
from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
import timm
from torch.autograd import Variable
def reshape_transform_resmlp(tensor, height=14, width=14):
result = tensor.reshape(tensor.size(0),
height, width, tensor.size(2))
result = result.transpose(2, 3).transpose(1, 2)
return result
def reshape_transform_swin(tensor, height=7, width=7):
result = tensor.reshape(tensor.size(0),
height, width, tensor.size(2))
# Bring the channels to the first dimension,
# like in CNNs.
result = result.transpose(2, 3).transpose(1, 2)
return result
def reshape_transform_vit(tensor, height=14, width=14):
result = tensor[:, 1:, :].reshape(tensor.size(0),
height, width, tensor.size(2))
# Bring the channels to the first dimension,
# like in CNNs.
result = result.transpose(2, 3).transpose(1, 2)
return result
def get_args():
parser = argparse.ArgumentParser()
parser.add_argument('--use-cuda', action='store_true', default=False,
help='Use NVIDIA GPU acceleration')
parser.add_argument(
'--image-path',
type=str,
default="./test/0bf7bfb05.png",
help='Input image path')
parser.add_argument(
'--output-image-path',
type=str,
default=None,
help='Output image path')
parser.add_argument(
'--model',
type=str,
default='seaformer',
help='model name')
parser.add_argument('--aug_smooth', action='store_true',
help='Apply test time augmentation to smooth the CAM')
parser.add_argument(
'--eigen_smooth',
action='store_true',
help='Reduce noise by taking the first principle componenet'
'of cam_weights*activations')
parser.add_argument('--method', type=str, default='gradcam++',
choices=['gradcam', 'gradcam++',
'scorecam', 'xgradcam',
'ablationcam', 'eigencam',
'eigengradcam', 'layercam', 'fullgrad'],
help='Can be gradcam/gradcam++/scorecam/xgradcam'
'/ablationcam/eigencam/eigengradcam/layercam')
args = parser.parse_args()
args.use_cuda = args.use_cuda and torch.cuda.is_available()
if args.use_cuda:
print('Using GPU for acceleration')
else:
print('Using CPU for computation')
return args
if __name__ == '__main__':
args = get_args()
methods = \
{"gradcam": GradCAM,
"scorecam": ScoreCAM,
"gradcam++": GradCAMPlusPlus,
"ablationcam": AblationCAM,
"xgradcam": XGradCAM,
"eigencam": EigenCAM,
"eigengradcam": EigenGradCAM,
"layercam": LayerCAM,
"fullgrad": FullGrad}
DEVICE = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
model = torch.load('checkpoints/seaformer/best.pth', map_location='cpu')
reshape_transform = None
print(model.trans2)
if 'seaformer' in args.model:
target_layers = [model.trans2] # [model.network[-1][-2]]
print(target_layers)
model.eval()
model.to(DEVICE)
img_path = args.image_path
if args.image_path:
img_path = args.image_path
else:
import requests
image_url = 'http://146.48.86.29/edge-mac/imgs/n02123045/ILSVRC2012_val_00023779.JPEG'
img_path = image_url.split('/')[-1]
if os.path.exists(img_path):
img_data = requests.get(image_url).content
with open(img_path, 'wb') as handler:
handler.write(img_data)
if args.output_image_path:
save_name = args.output_image_path
else:
img_type = img_path.split('.')[-1]
it_len = len(img_type)
save_name = img_path.split('/')[-1][:-(it_len + 1)]
save_name = save_name + '_' + args.model + '.' + img_type
img = cv2.imread(img_path, 1)
img = cv2.resize(img, (224, 224), interpolation=cv2.INTER_AREA)
if args.model == 'resize':
cv2.imwrite(save_name, img)
else:
rgb_img = img[:, :, ::-1]
rgb_img = np.float32(rgb_img) / 255
input_tensor = Variable(preprocess_image(rgb_img,
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]), requires_grad=True).to(DEVICE)
targets = None
cam_algorithm = methods[args.method]
with cam_algorithm(model=model,
target_layers=target_layers,
use_cuda=args.use_cuda,
reshape_transform=reshape_transform,
) as cam:
cam.batch_size = 1
grayscale_cam = cam(input_tensor=input_tensor,
targets=targets,
aug_smooth=args.aug_smooth,
eigen_smooth=args.eigen_smooth)
grayscale_cam = grayscale_cam[0, :]
cam_image = show_cam_on_image(rgb_img, grayscale_cam, use_rgb=True)
cam_image = cv2.cvtColor(cam_image, cv2.COLOR_RGB2BGR)
cv2.imwrite(save_name, cam_image)
对get_args函数的参数进行设置：
- use-cuda:是否使用cuda，如果在没有GPU的电脑上调试时，将其设置为False。
- image-path：待测图片的路径，这个是必填项。
- model：必填项，默认值：InternImage。
效果如下图所示：
完整的代码
完整的代码：
https://download.csdn.net/download/hhhhhhhhhhwwwwwwwwww/87742059?spm=1001.2014.3001.5503