import glob
import warnings
import yaml
warnings.filterwarnings('ignore')
import argparse
import os
import random
import shutil
import time
import warnings
import torch
import numpy as np
import apex
from apex import amp
import torch.nn as nn
import torch.nn.parallel
import torch.npu
import torch.backends.cudnn as cudnn
import torch.distributed as dist
import torch.optim
import torch.multiprocessing as mp
import torch.utils.data
import torch.utils.data.distributed
import torchvision.transforms as transforms
import torchvision.datasets as datasets
import models.resnet_0_6_0 as resnet_0_6_0
import moxing as mox
import argparse
import time
import yaml
import os
import logging
from collections import OrderedDict
from contextlib import suppress
from datetime import datetime
import glob
import shutil
torch.multiprocessing.set_sharing_strategy('file_system')
parser = argparse.ArgumentParser(description='PyTorch ImageNet Training')
parser.add_argument('--data', default="",metavar='DIR',
help='path to dataset')
parser.add_argument('--output', default="",metavar='DIR',
help='pth和onnx文件')
parser.add_argument('--onnx', default=True, type=bool, metavar='M',
help='momentum')
parser.add_argument('-j', '--workers', default=4, type=int, metavar='N',
help='number of data loading workers (default: 4)')
parser.add_argument('--epochs', default=20, type=int, metavar='N',
help='number of total epochs to run')
parser.add_argument('--start-epoch', default=0, type=int, metavar='N',
help='manual epoch number (useful on restarts)')
parser.add_argument('-b', '--batch-size', default=256, type=int,
metavar='N',
help='mini-batch size (default: 256), this is the total '
'batch size of all GPUs on the current node when '
'using Data Parallel or Distributed Data Parallel')
parser.add_argument('--lr', '--learning-rate', default=0.1, type=float,
metavar='LR', help='initial learning rate', dest='lr')
parser.add_argument('--momentum', default=0.9, type=float, metavar='M',
help='momentum')
parser.add_argument('--wd', '--weight-decay', default=1e-4, type=float,
metavar='W', help='weight decay (default: 1e-4)',
dest='weight_decay')
parser.add_argument('-p', '--print-freq', default=10, type=int,
metavar='N', help='print frequency (default: 10)')
parser.add_argument('--resume', default='./checkpoint.pth.tar', type=str, metavar='PATH',
help='path to latest checkpoint (default: none)')
parser.add_argument('-e', '--evaluate', dest='evaluate', action='store_true',
help='evaluate model on validation set')
parser.add_argument('--pretrained', dest='pretrained', action='store_true',
help='use pre-trained model')
parser.add_argument('--world-size', default=-1, type=int,
help='number of nodes for distributed training')
parser.add_argument('--rank', default=-1, type=int,
help='node rank for distributed training')
parser.add_argument('--dist-url', default='tcp://224.66.41.62:23456', type=str,
help='url used to set up distributed training')
parser.add_argument('--dist-backend', default='nccl', type=str,
help='distributed backend')
parser.add_argument('--seed', default=None, type=int,
help='seed for initializing training. ')
parser.add_argument('--gpu', default=None, type=int,
help='GPU id to use.')
parser.add_argument('--multiprocessing-distributed', action='store_true',
help='Use multi-processing distributed training to launch '
'N processes per node, which has N GPUs. This is the '
'fastest way to use PyTorch for either single node or '
'multi node data parallel training')
parser.add_argument('--device', default='npu', type=str, help='npu or gpu')
parser.add_argument('--addr', default='10.136.181.115',
type=str, help='master addr')
parser.add_argument('--device_list', default='0,1,2,3,4,5,6,7',
type=str, help='device id list')
parser.add_argument('--amp', default=False, action='store_true',
help='use amp to train the model')
parser.add_argument('--loss-scale', default=1024., type=float,
help='loss scale using in amp, default -1 means dynamic')
parser.add_argument('--opt-level', default='O2', type=str,
help='loss scale using in amp, default -1 means dynamic')
parser.add_argument('--prof', default=False, action='store_true',
help='use profiling to evaluate the performance of model')
parser.add_argument('--warm_up_epochs', default=5, type=int,
help='warm up')
best_acc1 = 0
def device_id_to_process_device_map(device_list):
devices = device_list.split(",")
devices = [int(x) for x in devices]
devices.sort()
process_device_map = dict()
for process_id, device_id in enumerate(devices):
process_device_map[process_id] = device_id
return process_device_map
def main():
args = parser.parse_args()
print(args.device_list)
os.environ['MASTER_ADDR'] = args.addr
os.environ['MASTER_PORT'] = '29688'
if args.seed is not None:
random.seed(args.seed)
torch.manual_seed(args.seed)
cudnn.deterministic = True
warnings.warn('You have chosen to seed training. '
'This will turn on the CUDNN deterministic setting, '
'which can slow down your training considerably! '
'You may see unexpected behavior when restarting '
'from checkpoints.')
if args.gpu is not None:
warnings.warn('You have chosen a specific GPU. This will completely '
'disable data parallelism.')
if args.dist_url == "env://" and args.world_size == -1:
args.world_size = int(os.environ["WORLD_SIZE"])
args.distributed = args.world_size > 1 or args.multiprocessing_distributed
args.process_device_map = device_id_to_process_device_map(args.device_list)
if args.device == 'npu':
ngpus_per_node = len(args.process_device_map)
else:
if args.distributed:
ngpus_per_node = torch.cuda.device_count()
else:
ngpus_per_node = 1
print('ngpus_per_node:', ngpus_per_node)
if args.multiprocessing_distributed:
args.world_size = ngpus_per_node * args.world_size
mp.spawn(main_worker, nprocs=ngpus_per_node,
args=(ngpus_per_node, args))
else:
main_worker(args.gpu, ngpus_per_node, args)
def main_worker(gpu, ngpus_per_node, args):
global best_acc1
args.gpu = args.process_device_map[gpu]
if args.gpu is not None:
print("Use GPU: {} for training".format(args.gpu))
if args.distributed:
if args.dist_url == "env://" and args.rank == -1:
args.rank = int(os.environ["RANK"])
if args.multiprocessing_distributed:
args.rank = args.rank * ngpus_per_node + gpu
if args.device == 'npu':
dist.init_process_group(backend=args.dist_backend,
world_size=args.world_size, rank=args.rank)
else:
dist.init_process_group(backend=args.dist_backend, init_method=args.dist_url,
world_size=args.world_size, rank=args.rank)
if args.pretrained:
print("=> using pre-trained model resnet152")
model = resnet_0_6_0.resnet152(pretrained=False, progress=True)
pretrained_dict = \
torch.load("checkpoint.pth.tar", map_location="cpu")["state_dict"]
model.load_state_dict({k.replace('module.', ''): v for k, v in pretrained_dict.items()})
if "fc.weight" in pretrained_dict:
pretrained_dict.pop('fc.weight')
pretrained_dict.pop('fc.bias')
if "module.fc.weight" in pretrained_dict:
pretrained_dict.pop('module.fc.weight')
pretrained_dict.pop('module.fc.bias')
for param in model.parameters():
param.requires_grad = False
model.fc = nn.Linear(2048, 1000)
model.load_state_dict(pretrained_dict, strict=False)
else:
print("=> creating model resnet152")
model = resnet_0_6_0.resnet152()
if args.distributed:
if args.gpu is not None:
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
torch.npu.set_device(loc)
model = model.to(loc)
else:
torch.cuda.set_device(args.gpu)
model.cuda(args.gpu)
args.batch_size = int(args.batch_size / args.world_size)
args.workers = int((args.workers + ngpus_per_node - 1) / ngpus_per_node)
else:
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
model = model.to(loc)
else:
model.cuda()
print("[gpu id:", args.gpu, "]",
"============================test args.gpu is not None else==========================")
elif args.gpu is not None:
print("[gpu id:", args.gpu, "]",
"============================test elif args.gpu is not None:==========================")
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
torch.npu.set_device(args.gpu)
model = model.to(loc)
else:
torch.cuda.set_device(args.gpu)
model = model.cuda(args.gpu)
else:
print("[gpu id:", args.gpu, "]", "============================test 1==========================")
print("[gpu id:", args.gpu, "]", "============================test 3==========================")
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
else:
print("before : model = torch.nn.DataParallel(model).cuda()")
optimizer = apex.optimizers.NpuFusedSGD(model.parameters(), args.lr,
momentum=args.momentum,
weight_decay=args.weight_decay)
if args.amp:
model, optimizer = amp.initialize(
model, optimizer, opt_level=args.opt_level, loss_scale=args.loss_scale)
if args.distributed:
if args.gpu is not None:
if args.pretrained:
model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[args.gpu], broadcast_buffers=False,
find_unused_parameters=True)
else:
model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[args.gpu], broadcast_buffers=False)
else:
print("[gpu id:", args.gpu, "]",
"============================test args.gpu is not None else==========================")
model = torch.nn.parallel.DistributedDataParallel(model)
elif args.gpu is not None:
print("[gpu id:", args.gpu, "]",
"============================test elif args.gpu is not None:==========================")
else:
print("[gpu id:", args.gpu, "]", "============================test 1==========================")
print("[gpu id:", args.gpu, "]", "============================test 3==========================")
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
model = torch.nn.DataParallel(model).to(loc)
else:
model = torch.nn.DataParallel(model).cuda()
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
criterion = nn.CrossEntropyLoss().to(loc)
else:
criterion = nn.CrossEntropyLoss().cuda(args.gpu)
if args.resume:
if os.path.isfile(args.resume):
print("=> loading checkpoint '{}'".format(args.resume))
if args.gpu is None:
checkpoint = torch.load(args.resume)
else:
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
else:
loc = 'cuda:{}'.format(args.gpu)
checkpoint = torch.load(args.resume, map_location=loc)
args.start_epoch = checkpoint['epoch']
best_acc1 = checkpoint['best_acc1']
if args.gpu is not None:
best_acc1 = best_acc1.to(args.gpu)
model.load_state_dict(checkpoint['state_dict'])
optimizer.load_state_dict(checkpoint['optimizer'])
if args.amp:
amp.load_state_dict(checkpoint['amp'])
print("=> loaded checkpoint '{}' (epoch {})"
.format(args.resume, checkpoint['epoch']))
else:
print("=> no checkpoint found at '{}'".format(args.resume))
cudnn.benchmark = True
real_path = '/cache/data_url'
if not os.path.exists(real_path):
os.makedirs(real_path)
pth_path = '/cache/output'
if not os.path.exists(pth_path):
os.makedirs(pth_path)
mox.file.copy_parallel(args.data, real_path)
print("training data finish copy to %s." % real_path)
traindir = os.path.join(real_path, 'train')
valdir = os.path.join(real_path, 'val')
normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
train_dataset = datasets.ImageFolder(
traindir,
transforms.Compose([
transforms.RandomResizedCrop(224),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
normalize,
]))
if args.distributed:
train_sampler = torch.utils.data.distributed.DistributedSampler(
train_dataset)
else:
train_sampler = None
train_loader = torch.utils.data.DataLoader(
train_dataset, batch_size=args.batch_size, shuffle=(
train_sampler is None),
num_workers=args.workers, pin_memory=False, sampler=train_sampler, drop_last=True)
val_loader = torch.utils.data.DataLoader(
datasets.ImageFolder(valdir, transforms.Compose([
transforms.Resize(256),
transforms.CenterCrop(224),
transforms.ToTensor(),
normalize,
])),
batch_size=args.batch_size, shuffle=True,
num_workers=args.workers, pin_memory=False, drop_last=True)
if args.evaluate:
validate(val_loader, model, criterion, args, ngpus_per_node)
return
if args.prof:
profiling(train_loader, model, criterion, optimizer, args)
return
start_time = time.time()
for epoch in range(args.start_epoch, args.epochs):
if args.distributed:
train_sampler.set_epoch(epoch)
adjust_learning_rate(optimizer, epoch, args)
train(train_loader, model, criterion, optimizer, epoch, args, ngpus_per_node)
acc1 = validate(val_loader, model, criterion, args, ngpus_per_node)
is_best = acc1 > best_acc1
if(epoch==args.start_epoch):
is_best = True
best_acc1 = max(acc1, best_acc1)
if args.device == 'npu' and args.gpu == 0 and epoch == 4:
print("Complete 4 epoch training, take time:{}h".format(round((time.time() - start_time) / 3600.0, 2)))
pth2onnx_path = args.output
print(pth2onnx_path)
if not args.multiprocessing_distributed or (args.multiprocessing_distributed
and args.rank % ngpus_per_node == 0):
if args.amp:
print(is_best)
save_checkpoint({
'epoch': epoch + 1,
'arch': 'resnet152',
'state_dict': model.state_dict(),
'best_acc1': best_acc1,
'optimizer': optimizer.state_dict(),
'amp': amp.state_dict(),
}, is_best,pth_path)
else:
save_checkpoint({
'epoch': epoch + 1,
'arch': 'resnet152',
'state_dict': model.state_dict(),
'best_acc1': best_acc1,
'optimizer': optimizer.state_dict(),
}, is_best,pth_path)
mox.file.copy_parallel(pth_path, pth2onnx_path)
CACHE_TRAINING_URL = "/cache/training"
if args.onnx:
os.makedirs(CACHE_TRAINING_URL, exist_ok=True)
print("abspath:", os.path.abspath(pth_path))
print("output_dir:", os.listdir(pth_path))
shutil.copy(os.path.join(os.path.abspath(pth_path), 'model_best.pth.tar'), CACHE_TRAINING_URL)
pth_pattern = os.path.join(CACHE_TRAINING_URL, 'model_best.pth.tar')
print("pth_pattern:", os.path.abspath(pth_pattern))
pth_file_list = glob.glob(pth_pattern)
if not pth_file_list:
print(f"can't find pth {pth_pattern}")
pth_file = pth_file_list[0]
onnx_path = pth_file.split(".")[0] + '.onnx'
convert(pth_file, onnx_path)
mox.file.copy_parallel(CACHE_TRAINING_URL, pth2onnx_path)
print("CACHE_TRAINING_URL:", os.listdir(CACHE_TRAINING_URL))
def proc_node_module(checkpoint, attr_name):
new_model_state = OrderedDict()
for k, v in checkpoint[attr_name].items():
if(k[0: 7] == "module."):
name = k[7:]
else:
name = k[0:]
new_model_state[name] = v
return new_model_state
def convert(pth_file_path, onnx_file_path):
checkpoint = torch.load(pth_file_path, map_location='cpu')
checkpoint['state_dict'] = proc_node_module(checkpoint, 'state_dict')
model = resnet_0_6_0.resnet152(pretrained=False, progress=True)
model.load_state_dict(checkpoint['state_dict'], False)
model.eval()
input_names = ["image"]
output_names = ["output1"]
dummy_input = torch.randn(2, 3, 224, 224)
torch.onnx.export(model, dummy_input, onnx_file_path, input_names=input_names, output_names=output_names, opset_version=11)
def profiling(data_loader, model, criterion, optimizer, args):
model.train()
def update(model, images, target, optimizer):
output = model(images)
loss = criterion(output, target)
if args.amp:
with amp.scale_loss(loss, optimizer) as scaled_loss:
scaled_loss.backward()
else:
loss.backward()
optimizer.zero_grad()
optimizer.step()
for step, (images, target) in enumerate(data_loader):
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
images = images.to(loc, non_blocking=True).to(torch.float)
target = target.to(torch.int32).to(loc, non_blocking=True)
else:
images = images.cuda(args.gpu, non_blocking=True)
target = target.cuda(args.gpu, non_blocking=True)
if step < 5:
update(model, images, target, optimizer)
else:
if args.device == 'npu':
with torch.autograd.profiler.profile(use_npu=True) as prof:
update(model, images, target, optimizer)
else:
with torch.autograd.profiler.profile(use_cuda=True) as prof:
update(model, images, target, optimizer)
break
prof.export_chrome_trace("output.prof")
def train(train_loader, model, criterion, optimizer, epoch, args, ngpus_per_node):
batch_time = AverageMeter('Time', ':6.3f')
data_time = AverageMeter('Data', ':6.3f')
losses = AverageMeter('Loss', ':.4e')
top1 = AverageMeter('Acc@1', ':6.2f')
top5 = AverageMeter('Acc@5', ':6.2f')
progress = ProgressMeter(
len(train_loader),
[batch_time, data_time, losses, top1, top5],
prefix="Epoch: [{}]".format(epoch))
model.train()
end = time.time()
for i, (images, target) in enumerate(train_loader):
data_time.update(time.time() - end)
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
images = images.to(loc, non_blocking=True).to(torch.float)
target = target.to(torch.int32).to(loc, non_blocking=True)
else:
images = images.cuda(args.gpu, non_blocking=True)
target = target.cuda(args.gpu, non_blocking=True)
output = model(images)
loss = criterion(output, target)
acc1, acc5 = accuracy(output, target, topk=(1, 5))
losses.update(loss.item(), images.size(0))
top1.update(acc1[0], images.size(0))
top5.update(acc5[0], images.size(0))
optimizer.zero_grad()
if args.amp:
with amp.scale_loss(loss, optimizer) as scaled_loss:
scaled_loss.backward()
else:
loss.backward()
optimizer.step()
if args.device == 'npu':
torch.npu.synchronize()
cost_time = time.time() - end
batch_time.update(cost_time)
end = time.time()
if i % args.print_freq == 0:
if not args.multiprocessing_distributed or (args.multiprocessing_distributed
and args.rank % ngpus_per_node == 0):
progress.display(i)
if not args.multiprocessing_distributed or (args.multiprocessing_distributed
and args.rank % ngpus_per_node == 0):
print("[npu id:", args.gpu, "]", "batch_size:", args.world_size * args.batch_size,
'Time: {:.3f}'.format(batch_time.avg), '* FPS@all {:.3f}'.format(
args.batch_size * args.world_size))
def validate(val_loader, model, criterion, args, ngpus_per_node):
batch_time = AverageMeter('Time', ':6.3f')
losses = AverageMeter('Loss', ':.4e')
top1 = AverageMeter('Acc@1', ':6.2f')
top5 = AverageMeter('Acc@5', ':6.2f')
progress = ProgressMeter(
len(val_loader),
[batch_time, losses, top1, top5],
prefix='Test: ')
model.eval()
with torch.no_grad():
end = time.time()
for i, (images, target) in enumerate(val_loader):
if args.gpu is not None:
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
images = images.to(loc).to(torch.float)
else:
images = images.cuda(args.gpu, non_blocking=True)
if args.device == 'npu':
loc = 'npu:{}'.format(args.gpu)
target = target.to(torch.int32).to(loc, non_blocking=True)
else:
target = target.cuda(args.gpu, non_blocking=True)
output = model(images)
loss = criterion(output, target)
acc1, acc5 = accuracy(output, target, topk=(1, 5))
losses.update(loss.item(), images.size(0))
top1.update(acc1[0], images.size(0))
top5.update(acc5[0], images.size(0))
cost_time = time.time() - end
batch_time.update(cost_time)
end = time.time()
if i % args.print_freq == 0:
if not args.multiprocessing_distributed or (args.multiprocessing_distributed
and args.rank % ngpus_per_node == 0):
progress.display(i)
if i % args.print_freq == 0:
if not args.multiprocessing_distributed or (args.multiprocessing_distributed
and args.rank % ngpus_per_node == 0):
print("[gpu id:", args.gpu, "]", '[AVG-ACC] * Acc@1 {top1.avg:.3f} Acc@5 {top5.avg:.3f}'
.format(top1=top1, top5=top5))
return top1.avg
def save_checkpoint(state, is_best, filename='checkpoint.pth.tar'):
filename_check = os.path.join(filename, 'checkpoint.pth.tar')
print("ahead save")
torch.save(state, filename_check)
print("ahead best")
if is_best:
filename_best = os.path.join(filename, 'model_best.pth.tar')
shutil.copyfile(filename_check, filename_best)
class AverageMeter(object):
"""Computes and stores the average and current value"""
def __init__(self, name, fmt=':f', start_count_index=2):
self.name = name
self.fmt = fmt
self.reset()
self.start_count_index = start_count_index
def reset(self):
self.val = 0
self.avg = 0
self.sum = 0
self.count = 0
def update(self, val, n=1):
if self.count == 0:
self.N = n
self.val = val
self.count += n
if self.count > (self.start_count_index * self.N):
self.sum += val * n
self.avg = self.sum / (self.count - self.start_count_index * self.N)
def __str__(self):
fmtstr = '{name} {val' + self.fmt + '} ({avg' + self.fmt + '})'
return fmtstr.format(**self.__dict__)
class ProgressMeter(object):
def __init__(self, num_batches, meters, prefix=""):
self.batch_fmtstr = self._get_batch_fmtstr(num_batches)
self.meters = meters
self.prefix = prefix
def display(self, batch):
entries = [self.prefix + self.batch_fmtstr.format(batch)]
entries += [str(meter) for meter in self.meters]
print('\t'.join(entries))
def _get_batch_fmtstr(self, num_batches):
num_digits = len(str(num_batches // 1))
fmt = '{:' + str(num_digits) + 'd}'
return '[' + fmt + '/' + fmt.format(num_batches) + ']'
def adjust_learning_rate(optimizer, epoch, args):
"""Sets the learning rate to the initial LR decayed by 10 every 30 epochs"""
if args.warm_up_epochs > 0 and epoch < args.warm_up_epochs:
lr = args.lr * ((epoch + 1) / (args.warm_up_epochs + 1))
else:
alpha = 0
cosine_decay = 0.5 * (
1 + np.cos(np.pi * (epoch - args.warm_up_epochs) / (args.epochs - args.warm_up_epochs)))
decayed = (1 - alpha) * cosine_decay + alpha
lr = args.lr * decayed
print("=> Epoch[%d] Setting lr: %.4f" % (epoch, lr))
for param_group in optimizer.param_groups:
param_group['lr'] = lr
def accuracy(output, target, topk=(1,)):
"""Computes the accuracy over the k top predictions for the specified values of k"""
with torch.no_grad():
maxk = max(topk)
batch_size = target.size(0)
_, pred = output.topk(maxk, 1, True, True)
pred = pred.t()
correct = pred.eq(target.view(1, -1).expand_as(pred))
res = []
for k in topk:
correct_k = correct[:k].reshape(-1).float().sum(0, keepdim=True)
res.append(correct_k.mul_(100.0 / batch_size))
return res
if __name__ == '__main__':
main()