笔者Ultra-Fast-Lane-Detection专栏链接🔗导航：
笔者的上一篇博客链接1，可以作为读者的车道线检测初探索。
此外笔者重新更新一篇车道线检测（Ultra-Fast-Lane-Detection）完整复现思路（主要针对自定义数据集）
内容包括：环境搭建、数据准备、模型训练、模型测试
首先下载项目源码：Ultra-Fast-Lane-Detection源码地址
git clone https://github.com/cfzd/Ultra-Fast-Lane-Detection
cd Ultra-Fast-Lane-Detection
Windows系统下则直接手动下载解压，项目文件中的INSTALL.md中有详细的环境搭建步骤，可以作为环境搭建的参考
笔者基于Anaconda环境搭建指令总结如下：
（Anaconda的安装教程可以参考笔者的博客【Anaconda】windows安装Anaconda）
conda create -n lane(自定义) python=3.7 -y
conda activate lane # 激活环境
# 参考官网：https://pytorch.org/get-started/locally/
pip install torch==1.9.1+cu111 torchvision==0.10.1+cu111 torchaudio===0.9.1 -f https://download.pytorch.org/whl/torch_stable.html
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 若存在个别库安装缓慢，则可以选中单独安装
# 特别注意setuptools版本不能过高
pip install setuptools==59.5.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
上述简单的几个步骤就完成了环境的搭建
使用labelme数据标注工具实现数据集的准备
人工标注完数据结构如下：
datasets
├─json(标签文件)
│ 202105270040-right00109.json
│ 202105270040-right00111.json
│ 202105270040-right00161.json
│
└─pic(图像数据)
202105270040-right00109.jpg
202105270040-right00111.jpg
202105270040-right00161.jpg
笔者的标注图像如下（简单的用两条直线标注轨道线）：
创建脚本/utils_alian/instance.py，（utils_alian为笔者创建的文件夹，用于存放自定义的脚本）其中代码如下：
"""
2022.4.17
author:alian
function:
根据labelme标注的文件生成实例掩膜
"""
import argparse
import glob
import json
import os
import os.path as ops
import cv2
import numpy as np
def init_args():
parser = argparse.ArgumentParser()
parser.add_argument('--img_dir', type=str, help='The origin path of image')
parser.add_argument('--json_dir', type=str, help='The origin path of json')
return parser.parse_args()
# 获得原始图像（.png），二值化图像以及实例分割图像
def process_json_file(img_path, json_path, instance_dst_dir):
"""
:param img_path: 原始图像路径
:param json_path: 标签文件路径
:param instance_dst_dir:实例图像保存路径
:return:
"""
assert ops.exists(img_path), '{:s} not exist'.format(img_path)
image = cv2.imread(img_path, cv2.IMREAD_COLOR)
instance_image = np.zeros([image.shape[0], image.shape[1]], np.uint8)
with open(json_path, 'r',encoding='utf8') as file:
info_dict = json.load(file)
for ind,info in enumerate(info_dict['shapes']):
contours = info['points']
contours = np.array(contours,dtype=int)
# 绘制多边形
cv2.fillPoly(instance_image, [contours], (ind+1, ind+1, ind+1)) # * 50 + 20
instance_image_path = img_path.replace(os.path.dirname(img_path),instance_dst_dir)
cv2.imwrite(instance_image_path.replace('.jpg','.png'), instance_image) # 实例分割图像
# 训练库构建
def process_tusimple_dataset(img_dir,json_dir):
"""
:param json_dir: 标签文件路径
:param img_dir: 原始图像路径
:return:
"""
gt_instance_dir = ops.join(os.path.dirname(img_dir), 'label') # 与原始图片文件夹在同级父目录下
os.makedirs(gt_instance_dir, exist_ok=True)
for img_path in glob.glob('{:s}/*.jpg'.format(img_dir)):
json_path = img_path.replace(img_dir,json_dir).replace('.jpg','.json')
process_json_file(img_path, json_path, gt_instance_dir)
return
if __name__ == '__main__':
img_dir = r'./pic' # 原始图片路径
json_dir = r'./json' # 标注文件路径
process_tusimple_dataset(img_dir,json_dir)
生成实例图片如下：
上篇博客【Lane】 Ultra-Fast-Lane-Detection 复现中提到：由于数据集的类别数与指定的类别数不一致导致训练出错，解决方法：
验证标签数目与训练时指定的数量一致，创建脚本utils_alian/check_label_num.py，代码如下：
"""
2022.4.17
author:alian
function:
核查.json标签数量
"""
import numpy as np
import scipy.special
import os
import cv2
path = r'C:\Users\ZNJT\Desktop\yuanshi\yuanshi\label' # 实例图像的路径
a = os.listdir(path)
print(a)
for i in a:
img = cv2.imread(os.path.join(path,i),-1)
print(set((img.flatten())))
终端显示如下：
0默认为背景标签，上述显示说明车道线数有2个，代码运行完数据集目录结构如下：
datasets
├─json(标签文件)
│ 202105270040-right00109.json
│ 202105270040-right00111.json
│ 202105270040-right00161.json
│
├─label
│ 202105270040-right00109.png
│ 202105270040-right00111.png
│ 202105270040-right00161.png
│
└─pic(图像数据)
202105270040-right00109.jpg
202105270040-right00111.jpg
202105270040-right00161.jpg
笔者对源码进行修改，主要针对自定义的数据集，删除了开源数据集的部分，修改的脚本文件如图所示
在项目中创建文件夹utils_alian，将自定义的脚本文件放在其中
创建脚本utils_alian/config.py，代码如下：
"""
2022.4.20
author:alian
function:
训练参数配置
"""
import argparse,datetime,os
def get_args(): # 配置训练参数
parser = argparse.ArgumentParser()
parser.add_argument('--source', type=str, default='./dataset/',help='数据库路径')
parser.add_argument('--log_path', type=str, default='./logs/', help='模型保存路径')
parser.add_argument('--epoch', type=int, default=100, help='训练轮数')
parser.add_argument('--batch_size', type=int, default=32, help='')
parser.add_argument('--device', default='', help='cuda device, i.e. 0 or 0,1,2,3 or cpu')
parser.add_argument('--optimizer', type=str, default='Adam', help='优化器：[SGD,Adam]')
parser.add_argument('--learning_rate', type=float, default=4e-4, help='学习率')
parser.add_argument('--weight_decay', type=float, default=1e-4, help='权重衰减系数')
parser.add_argument('--momentum', type=float, default=0.9, help='动量')
parser.add_argument('--scheduler', type=str, default='cos', help='调度器:[multi, cos]')
parser.add_argument('--gamma', type=float, default=0.1, help='模型预热')
parser.add_argument('--warmup', type=str, default='linear', help='模型预热')
parser.add_argument('--warmup_iters', type=int, default=100, help='模型预热')
parser.add_argument('--backbone', type=str, default='18', help='网络骨干')
parser.add_argument('--use_aux', type=bool, default=True, help='是否使用语义标签')
parser.add_argument('--griding_num', type=int, default=100, help='网格列数')
parser.add_argument('--row_anchor', type=int, default=56, help='行锚框')
parser.add_argument('--num_lanes', type=int, default=2, help='车道数')
parser.add_argument('--sim_loss_w', type=int, default=0, help='loss')
parser.add_argument('--shp_loss_w', type=int, default=0, help='loss')
parser.add_argument('--resume', type=str, default=None, help='继续训练')
parser.add_argument('--auto_backup', action='store_true', help='automatically backup current code in the log path')
parser.add_argument('--distributed',type=bool, default=False, help='分布式训练')
opt = parser.parse_args()
return opt
def get_work_dir(opt): # 模型保存路径
now = datetime.datetime.now().strftime('%m%d_%H%M') # 获得当前时间
hyper_param_str = '_lr_%1.0e_b_%d' % (opt.learning_rate, opt.batch_size)
work_dir = os.path.join(opt.log_path, now+hyper_param_str)
return work_dir
修改如下源码文件： utils_alian/dataset_alian.py
自定义的训练数据集加载函数，代码如下：
"""
2022.4.20
author:alian
function：
Ultra-Fast-Lane-Detection：https://github.com/cfzd/Ultra-Fast-Lane-Detection
加载自定义数据集
"""
import torch
from PIL import Image
import pdb
import numpy as np
import glob
from data.mytransforms import find_start_pos
import torch.utils.data
def loader_func(path):
return Image.open(path)
# 加载自定义的数据集
class ClsDataset(torch.utils.data.Dataset):
def __init__(self, path, img_transform=None, target_transform=None, simu_transform=None, griding_num=50,
load_name=False,
row_anchor=None, use_aux=False, segment_transform=None, num_lanes=4):
super(ClsDataset, self).__init__()
self.img_transform = img_transform #
self.target_transform = target_transform
self.segment_transform = segment_transform #
self.simu_transform = simu_transform #
self.path = path
self.griding_num = griding_num # 网格列数：100
self.load_name = load_name
self.use_aux = use_aux # 是否使用语义标签
self.num_lanes = num_lanes # 车道线的数量 2
self.row_anchor = row_anchor # 锚框出现的行位置，默认总行数为288
self.row_anchor.sort()
def __getitem__(self, index): # 获得图像数组
labels = glob.glob('%s/label/*.png'%self.path)
label_path = labels[index]
label = loader_func(label_path) # loader_func读取图像函数
imgs = glob.glob('%s/pic/*.jpg'%self.path)
img_path = imgs[index]
img = loader_func(img_path) # 读取图像
if self.simu_transform is not None:
img, label = self.simu_transform(img, label)
#
lane_pts = self._get_index(label) # 获得包含车道线的坐标[车道线数，56，2]（在原始图像上的，还未网格化）
# 网格化，将包含车道线的列值进行网格化
w, h = img.size
cls_label = self._grid_pts(lane_pts, self.griding_num, w)
if self.img_transform is not None:
img = self.img_transform(img)
# make the coordinates to classification label
if self.use_aux:
assert self.segment_transform is not None
seg_label = self.segment_transform(label)
aa = seg_label[:,33]
return img, cls_label, seg_label
if self.load_name:
return img, cls_label
return img, cls_label # img：[3,288,800],cls_label:[56,2],seg_label:[36,100]
def __len__(self):
return len(glob.glob('%s/label/*.png'%self.path))
def _grid_pts(self, pts, num_cols, w): # 获得网格数，返回[56,2]
"""
pts:包含车道线的坐标[车道线，56，2]
num_cols:网格列数
w:图片的列数（像素单位）
function：
将图片按指定的列数划分网格数，并获得车道线所在的网格位置
"""
# pts : numlane, n, 2
num_lane, n, n2 = pts.shape # [2,56,2]
col_sample = np.linspace(0, w - 1, num_cols) # 列间隔（在w=1920的原始图像上，分成100列，则每列的间隔为）
# 在w=1920的原始图像上，分成100列，则包含车道线的坐标落在那一列上
assert n2 == 2
to_pts = np.zeros((n, num_lane)) # [56,2]
for i in range(num_lane):
pti = pts[i, :, 1] # 包含车道线坐标的列值
to_pts[:, i] = np.asarray(
[int(pt // (col_sample[1] - col_sample[0])) if pt != -1 else num_cols for pt in pti])
return to_pts.astype(int)
def _get_index(self, label): # 获取包含车道线的坐标，返回[车道线数,56,2]
"""
label:1920*1080
"""
w, h = label.size
# 行锚框映射[64,68...284]-->[240,255,...1065]
if h != 288: # 若图像高度不为预设的288，则将行锚框row_anchors等比缩放到原始图像尺寸中[0,288]-->[0,1080]
scale_f = lambda x: int((x * 1.0 / 288) * h) # 定义一种匿名函数
sample_tmp = list(map(scale_f, self.row_anchor)) # 根据函数和自变量做映射
# 第一步骤：得到all_idx[2,56,2]的数组，包含车道线像素的坐标（在原始图像上的）
all_idx = np.zeros((self.num_lanes, len(sample_tmp), 2)) # [2,56,2]
# 实质是：获取车道线所在的坐标位置
for i, r in enumerate(sample_tmp): # 遍历行锚框[240,255,...1065]
label_r = np.asarray(label)[int(round(r))] # 获取该行的灰度值
for lane_idx in range(1, self.num_lanes + 1): # 遍历车道标签id，车道标签[1,2,...,n] 0一般为背景标签
pos = np.where(label_r == lane_idx)[0]
if len(pos) == 0:
all_idx[lane_idx - 1, i, 0] = r
all_idx[lane_idx - 1, i, 1] = -1
continue # 继续下一步的循环，不再往下走
pos = np.mean(pos) # 车道线具有一定的宽度，取列均值
all_idx[lane_idx - 1, i, 0] = r # 获取行值
all_idx[lane_idx - 1, i, 1] = pos # 获取列值
# 第2步骤all_idx_cp：下面的步骤为数据增强：将车道线坐标（在原始图像上的）延申至边缘
all_idx_cp = all_idx.copy()
for i in range(self.num_lanes):
if np.all(all_idx_cp[i, :, 1] == -1): # 若没找到车道线则继续找，找到了就执行下一步
continue
# if there is no lane
valid = all_idx_cp[i, :, 1] != -1 # 获得包含车道线的索引:仅包含true和false的列表
# get all valid lane points' index
valid_idx = all_idx_cp[i, valid, :] # 获得包含车道线的点
# get all valid lane points
if valid_idx[-1, 0] == all_idx_cp[0, -1, 0]:
# 若最底部的有效车道线点在图像边缘，则不进行延申操作（图像增强）
continue
if len(valid_idx) < 6: # 有效的车道线点至少6个
continue
valid_idx_half = valid_idx[len(valid_idx) // 2:, :] # 取一半的有效点
p = np.polyfit(valid_idx_half[:, 0], valid_idx_half[:, 1], deg=1) # 用一次函数进行拟合，返回一次函数的系数
start_line = valid_idx_half[-1, 0] # 最底部的有效车道线点
pos = find_start_pos(all_idx_cp[i, :, 0], start_line) + 1
fitted = np.polyval(p, all_idx_cp[i, pos:, 0]) # 根据系数和自变量返回应变量数组
fitted = np.array([-1 if y < 0 or y > w - 1 else y for y in fitted])
assert np.all(all_idx_cp[i, pos:, 1] == -1) # 条件为True时执行
all_idx_cp[i, pos:, 1] = fitted
if -1 in all_idx[:, :, 0]: # 所有的行坐标
pdb.set_trace() # 程序终止
return all_idx_cp # [4,56,2]
创建脚本 utils_alian/dataloader_alian.py，代码如下：
"""
2022.4.20
author:alian
function：
自定义数据加载器
"""
# 导入库
import torch
import torchvision.transforms as transforms
import torch.utils.data
# 导入项目源码中的文件
import data.mytransforms as mytransforms
from data.constant import tusimple_row_anchor
from utils_alian.dataset_alian import ClsDataset
def get_train_loader(batch_size, data_root, griding_num, use_aux, distributed, num_lanes):
target_transform = transforms.Compose([
mytransforms.FreeScaleMask((288, 800)), # 图像缩放功能
mytransforms.MaskToTensor(),
])
segment_transform = transforms.Compose([
mytransforms.FreeScaleMask((36, 100)),
mytransforms.MaskToTensor(),
])
img_transform = transforms.Compose([
transforms.Resize((288, 800)),
transforms.ToTensor(),
transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225)),
])
simu_transform = mytransforms.Compose2([
mytransforms.RandomRotate(6),
mytransforms.RandomUDoffsetLABEL(100),
mytransforms.RandomLROffsetLABEL(200)
])
# 自定义数据集
train_dataset = ClsDataset(data_root,
img_transform=img_transform, target_transform=target_transform,
simu_transform=simu_transform,
griding_num=griding_num,
row_anchor=tusimple_row_anchor,
segment_transform=segment_transform, use_aux=use_aux, num_lanes=num_lanes)
if distributed:
sampler = torch.utils.data.distributed.DistributedSampler(train_dataset)
else:
sampler = torch.utils.data.RandomSampler(train_dataset)
train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=batch_size, sampler=sampler, num_workers=0)
return train_loader
class SeqDistributedSampler(torch.utils.data.distributed.DistributedSampler): # 分布式采样
'''
Change the behavior of DistributedSampler to sequential distributed sampling.
The sequential sampling helps the stability of multi-thread testing, which needs multi-thread file io.
Without sequentially sampling, the file io on thread may interfere other threads.
'''
def __init__(self, dataset, num_replicas=None, rank=None, shuffle=False):
super().__init__(dataset, num_replicas, rank, shuffle)
def __iter__(self):
g = torch.Generator()
g.manual_seed(self.epoch)
if self.shuffle:
indices = torch.randperm(len(self.dataset), generator=g).tolist()
else:
indices = list(range(len(self.dataset)))
# add extra samples to make it evenly divisible
indices += indices[:(self.total_size - len(indices))]
assert len(indices) == self.total_size
num_per_rank = int(self.total_size // self.num_replicas)
# sequential sampling
indices = indices[num_per_rank * self.rank : num_per_rank * (self.rank + 1)]
assert len(indices) == self.num_samples
return iter(indices)
创建脚本train_alian.py，该脚本文件直接放在项目文件夹下
"""
2022.4.20
author：alian
车道线训练代码
"""
import os
os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
import torch, os, datetime,time
import torch.backends.cudnn
# 导入项目源码中的文件
from model.model import parsingNet
from utils_alian.dataloader_alian import get_train_loader # 自定义
from utils.dist_utils import dist_print, dist_tqdm
from utils.factory import get_metric_dict, get_loss_dict, get_optimizer, get_scheduler
from utils.metrics import update_metrics, reset_metrics
from utils.common import save_model, cp_projects
from utils.common import get_logger
from utils_alian.config import get_args,get_work_dir
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
def inference(net, data_label, use_aux): # 推理
if use_aux:
img, cls_label, seg_label = data_label
img, cls_label, seg_label = img.cuda(), cls_label.long().cuda(), seg_label.long().cuda()
cls_out, seg_out = net(img)
return {'cls_out': cls_out, 'cls_label': cls_label, 'seg_out':seg_out, 'seg_label': seg_label}
else:
img, cls_label = data_label
img, cls_label = img.cuda(), cls_label.long().cuda()
cls_out = net(img)
return {'cls_out': cls_out, 'cls_label': cls_label}
def resolve_val_data(results, use_aux): # 测试
results['cls_out'] = torch.argmax(results['cls_out'], dim=1)
if use_aux:
results['seg_out'] = torch.argmax(results['seg_out'], dim=1)
return results
def calc_loss(loss_dict, results, logger, global_step): # 计算损失函数
loss = 0 # 初始化损失
for i in range(len(loss_dict['name'])):
data_src = loss_dict['data_src'][i]
datas = [results[src] for src in data_src]
loss_cur = loss_dict['op'][i](*datas)
if global_step % 20 == 0:
logger.add_scalar('loss/'+loss_dict['name'][i], loss_cur, global_step)
loss += loss_cur * loss_dict['weight'][i] # 损失=当前损失*损失权重
return loss
def train(net, data_loader, loss_dict, optimizer, scheduler,logger, epoch, metric_dict, use_aux):
net.train()
progress_bar = dist_tqdm(train_loader) # 进度条
t_data_0 = time.time() # 时间
for b_idx, data_label in enumerate(progress_bar):
t_data_1 = time.time() # 时间
reset_metrics(metric_dict)
global_step = epoch * len(data_loader) + b_idx # 全局的步数
t_net_0 = time.time() # 时间
results = inference(net, data_label, use_aux)
loss = calc_loss(loss_dict, results, logger, global_step) # 构建损失函数
optimizer.zero_grad() # 梯度归零
loss.backward() # 损失反向传播（计算梯度）
optimizer.step() # 权重更新
scheduler.step(global_step)
t_net_1 = time.time() # 时间
results = resolve_val_data(results, use_aux)
update_metrics(metric_dict, results)
if global_step % 20 == 0:
for me_name, me_op in zip(metric_dict['name'], metric_dict['op']):
logger.add_scalar('metric/' + me_name, me_op.get(), global_step=global_step)
logger.add_scalar('meta/lr', optimizer.param_groups[0]['lr'], global_step=global_step)
if hasattr(progress_bar,'set_postfix'):
kwargs = {me_name: '%.3f' % me_op.get() for me_name, me_op in zip(metric_dict['name'], metric_dict['op'])}
progress_bar.set_postfix(loss='%.3f' % float(loss),
data_time = '%.3f' % float(t_data_1 - t_data_0),
net_time = '%.3f' % float(t_net_1 - t_net_0),
**kwargs)
t_data_0 = time.time()
if __name__ == "__main__": # 实例化
torch.backends.cudnn.benchmark = True # 若为 True，则cuDNN对多个卷积算法进行基准测试并选择最快的
opt = get_args()
work_dir = get_work_dir(opt) # 创建权重保存路径
distributed = False # 是否分布式训练
if 'WORLD_SIZE' in os.environ:
distributed = int(os.environ['WORLD_SIZE']) > 1
if distributed:
import torch.distributed
torch.cuda.set_device(0)
torch.distributed.init_process_group(backend='nccl', init_method='env://')
dist_print(datetime.datetime.now().strftime('[%Y/%m/%d %H:%M:%S]') + ' start training...')
dist_print(opt)
assert opt.backbone in ['18','34','50','101','152','50next','101next','50wide','101wide']
# 加载数据集
train_loader = get_train_loader(opt.batch_size, opt.source, opt.griding_num, opt.use_aux, distributed, opt.num_lanes)
# 加载模型
net = parsingNet(pretrained = True, backbone=opt.backbone,cls_dim=(opt.griding_num+1,opt.row_anchor, opt.num_lanes),use_aux=opt.use_aux).cuda()
if distributed: # 分布式数据并行
net = torch.nn.parallel.DistributedDataParallel(net, device_ids = [opt.local_rank])
# 构建优化器
optimizer = get_optimizer(net, opt)
if opt.resume is not None: # 继续训练
dist_print('==> Resume model from ' + opt.resume)
resume_dict = torch.load(opt.resume, map_location='cpu')
net.load_state_dict(resume_dict['model'])
if 'optimizer' in resume_dict.keys():
optimizer.load_state_dict(resume_dict['optimizer'])
resume_epoch = int(os.path.split(opt.resume)[1][2:5]) + 1
else:
resume_epoch = 0
scheduler = get_scheduler(optimizer, opt, len(train_loader))
dist_print(len(train_loader))
metric_dict = get_metric_dict(opt)
loss_dict = get_loss_dict(opt)
logger = get_logger(work_dir, opt)
cp_projects(opt.auto_backup, work_dir)
# 开始训练
for epoch in range(resume_epoch, opt.epoch):
# 传入参数： 网络，训练数据集，损失字典，优化器，调度器，
train(net, train_loader, loss_dict, optimizer, scheduler,logger, epoch, metric_dict, opt.use_aux) # 训练循环
save_model(net, optimizer, epoch, work_dir, distributed) # 模型保存
logger.close()
1）在终端执行如下指令开始训练：
python train_alian.py --source /data3/205b/Alian/Ultra-Fast-Lane-Detection/subway --log_path /data3/205b/Alian/Ultra-Fast-Lane-Detection/model --epoch 100 --batch_size 32 --num_lanes 2
2）在pycharm中运行
在train_alian.py页面右键，选择设置参数，如下图所示
接着，再运行train_alian.py文件
终端显示如下，则说明训练正常：
综上，就是Ultra-Fast-Lane-Detection自定义数据训练的全部过程
自训练完的模型测试部分，将在下一篇博客中详细说明。