当前位置：首页 > news >正文

在PyCharm中复现LaneNet车道线检测模型

news 2025/7/23 19:04:53

在PyCharm中复现LaneNet车道线检测模型

1. 引言

1.1 车道线检测的重要性

车道线检测是自动驾驶和高级驾驶辅助系统(ADAS)中的关键技术之一。准确的车道线检测能够帮助车辆保持在车道内行驶，为路径规划和车辆控制提供重要信息。随着自动驾驶技术的发展，车道线检测算法的准确性和实时性要求越来越高。

1.2 LaneNet模型概述

LaneNet是由Tulyakov等人提出的一种基于深度学习的端到端车道线检测模型。与传统方法相比，LaneNet采用了新颖的双分支网络结构：

实例分割分支：负责将车道线像素从背景中分离出来
嵌入分支：为每个车道线像素分配一个嵌入向量，使得相同车道线的像素在嵌入空间中距离较近，不同车道线的像素距离较远

这种双分支结构使得LaneNet能够处理可变数量的车道线，并准确区分不同的车道实例。

1.3 项目目标

本文旨在PyCharm开发环境中完整复现LaneNet模型，包括：

搭建模型架构
实现训练流程
准备和预处理数据集
进行模型评估
可视化检测结果

2. 环境配置

2.1 PyCharm环境设置

首先需要在PyCharm中创建新的Python项目：

打开PyCharm，选择"Create New Project"
指定项目位置和Python解释器（建议使用Python 3.7或更高版本）
创建完成后，在项目中新建以下目录结构：

lanenet_pycharm/
├── configs/          # 配置文件
├── data/             # 数据集
├── model/            # 模型代码
├── utils/            # 工具函数
├── train.py          # 训练脚本
├── test.py           # 测试脚本
└── evaluate.py       # 评估脚本

2.2 依赖库安装

在PyCharm的Terminal中运行以下命令安装所需依赖：

pip install tensorflow-gpu==2.4.1
pip install opencv-python
pip install numpy
pip install matplotlib
pip install scikit-learn
pip install scikit-image
pip install tqdm

或者通过PyCharm的Package管理界面安装这些包。

2.3 GPU配置（可选）

如果使用GPU加速训练，需要确保：

已安装合适的NVIDIA驱动程序
已安装CUDA和cuDNN（与TensorFlow版本匹配）
在PyCharm中正确配置了GPU环境

可以通过以下代码验证TensorFlow是否能检测到GPU：

import tensorflow as tf
print(tf.config.list_physical_devices('GPU'))

3. 数据集准备

3.1 数据集选择

LaneNet原始论文使用了TuSimple车道线检测数据集。我们将使用同样的数据集进行复现：

TuSimple数据集包含在不同交通和光照条件下拍摄的高速公路车道图像
数据集包括训练集、验证集和测试集
每张图像都标注了车道线的位置

3.2 数据集下载与预处理

从TuSimple官网下载数据集并解压到data/tusimple目录
实现数据预处理脚本utils/data_processor.py：

import os
import json
import cv2
import numpy as np
from tqdm import tqdmclass TuSimpleProcessor:def __init__(self, dataset_dir):self.dataset_dir = dataset_dirself.train_set = os.path.join(dataset_dir, 'train_set')self.test_set = os.path.join(dataset_dir, 'test_set')def process_annotation(self, json_file):with open(json_file, 'r') as f:annotations = json.load(f)samples = []for anno in tqdm(annotations, desc='Processing annotations'):raw_file = anno['raw_file']lanes = anno['lanes']y_samples = anno['h_samples']# 创建二进制分割图seg_img = np.zeros((720, 1280), dtype=np.uint8)for lane in lanes:points = [(x, y) for x, y in zip(lane, y_samples) if x >= 0]if len(points) > 1:cv2.polylines(seg_img, [np.array(points, np.int32)], isClosed=False, color=1, thickness=5)# 创建实例图instance_img = np.zeros((720, 1280), dtype=np.uint8)for i, lane in enumerate(lanes, 1):points = [(x, y) for x, y in zip(lane, y_samples) if x >= 0]if len(points) > 1:cv2.polylines(instance_img, [np.array(points, np.int32)], isClosed=False, color=i, thickness=5)samples.append({'image_path': os.path.join(self.dataset_dir, raw_file),'seg_label': seg_img,'instance_label': instance_img})return samplesdef prepare_dataset(self):train_json = os.path.join(self.train_set, 'label_data_0313.json')val_json = os.path.join(self.train_set, 'label_data_0531.json')test_json = os.path.join(self.test_set, 'label_data_0601.json')train_samples = self.process_annotation(train_json)val_samples = self.process_annotation(val_json)test_samples = self.process_annotation(test_json)return train_samples, val_samples, test_samples

3.3 数据增强

为了提高模型泛化能力，实现以下数据增强方法：

import random
import cv2
import numpy as npclass LaneNetAugmentor:def __init__(self):self.augmentations = [self.random_brightness,self.random_contrast,self.random_shadow,self.random_horizontal_shift,self.random_vertical_shift,self.random_rotation,self.random_blur]def __call__(self, image, seg_label, instance_label):# 随机选择几种增强方法aug_methods = random.sample(self.augmentations, k=random.randint(0, 4))for method in aug_methods:image, seg_label, instance_label = method(image, seg_label, instance_label)return image, seg_label, instance_labeldef random_brightness(self, image, seg_label, instance_label):if random.random() < 0.5:hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)h, s, v = cv2.split(hsv)adjust = random.uniform(0.7, 1.3)v = np.clip(v * adjust, 0, 255).astype(np.uint8)hsv = cv2.merge((h, s, v))image = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)return image, seg_label, instance_labeldef random_contrast(self, image, seg_label, instance_label):if random.random() < 0.5:alpha = random.uniform(0.8, 1.2)image = np.clip(image * alpha, 0, 255).astype(np.uint8)return image, seg_label, instance_label# 其他增强方法实现类似...

4. LaneNet模型实现

4.1 模型架构概述

LaneNet采用双分支网络结构：

编码器：共享的骨干网络（通常使用ENet或ResNet）
解码器：
- 二进制分割分支
- 实例嵌入分支

4.2 骨干网络实现

我们使用轻量级的ENet作为骨干网络：

import tensorflow as tf
from tensorflow.keras import layers, modelsclass ENetEncoder(tf.keras.Model):def __init__(self):super(ENetEncoder, self).__init__()# 初始块self.initial_block = InitialBlock()# Stage 1self.stage1_bottleneck1 = Bottleneck(16, downsample=True, dropout_rate=0.01)self.stage1_bottleneck2 = Bottleneck(64, dropout_rate=0.01)self.stage1_bottleneck3 = Bottleneck(64, dropout_rate=0.01)self.stage1_bottleneck4 = Bottleneck(64, dropout_rate=0.01)# Stage 2self.stage2_bottleneck1 = Bottleneck(128, downsample=True, dropout_rate=0.1)self.stage2_bottleneck2 = Bottleneck(128)self.stage2_bottleneck3 = Bottleneck(128, dilated=2)self.stage2_bottleneck4 = Bottleneck(128, asymmetric=5)self.stage2_bottleneck5 = Bottleneck(128, dilated=4)self.stage2_bottleneck6 = Bottleneck(128)self.stage2_bottleneck7 = Bottleneck(128, dilated=8)self.stage2_bottleneck8 = Bottleneck(128, asymmetric=5)self.stage2_bottleneck9 = Bottleneck(128, dilated=16)def call(self, inputs, training=None):x = self.initial_block(inputs, training=training)# Stage 1x, max_indices1 = self.stage1_bottleneck1(x, training=training)x = self.stage1_bottleneck2(x, training=training)x = self.stage1_bottleneck3(x, training=training)x = self.stage1_bottleneck4(x, training=training)# Stage 2x, max_indices2 = self.stage2_bottleneck1(x, training=training)x = self.stage2_bottleneck2(x, training=training)x = self.stage2_bottleneck3(x, training=training)x = self.stage2_bottleneck4(x, training=training)x = self.stage2_bottleneck5(x, training=training)x = self.stage2_bottleneck6(x, training=training)x = self.stage2_bottleneck7(x, training=training)x = self.stage2_bottleneck8(x, training=training)x = self.stage2_bottleneck9(x, training=training)return x, max_indices1, max_indices2

4.3 解码器实现

实现双分支解码器：

class LaneNetDecoder(tf.keras.Model):def __init__(self, num_classes=2, embedding_dim=4):super(LaneNetDecoder, self).__init__()# 共享的解码器部分self.upsample1 = layers.UpSampling2D(size=(2, 2))self.conv1 = layers.Conv2D(64, (3, 3), padding='same', activation='relu')self.upsample2 = layers.UpSampling2D(size=(2, 2))self.conv2 = layers.Conv2D(32, (3, 3), padding='same', activation='relu')# 二进制分割分支self.seg_upsample = layers.UpSampling2D(size=(2, 2))self.seg_conv = layers.Conv2D(num_classes, (1, 1), padding='same', activation='softmax')# 实例嵌入分支self.embedding_upsample = layers.UpSampling2D(size=(2, 2))self.embedding_conv = layers.Conv2D(embedding_dim, (1, 1), padding='same')def call(self, inputs, training=None):x = self.upsample1(inputs)x = self.conv1(x)x = self.upsample2(x)x = self.conv2(x)# 分割分支seg_output = self.seg_upsample(x)seg_output = self.seg_conv(seg_output)# 嵌入分支embedding_output = self.embedding_upsample(x)embedding_output = self.embedding_conv(embedding_output)return seg_output, embedding_output

4.4 完整的LaneNet模型

将编码器和解码器组合成完整的LaneNet：

class LaneNet(tf.keras.Model):def __init__(self, num_classes=2, embedding_dim=4):super(LaneNet, self).__init__()self.encoder = ENetEncoder()self.decoder = LaneNetDecoder(num_classes, embedding_dim)def call(self, inputs, training=None):# 编码器x, max_indices1, max_indices2 = self.encoder(inputs, training=training)# 解码器seg_output, embedding_output = self.decoder(x, training=training)return seg_output, embedding_output

5. 损失函数实现

5.1 二进制分割损失

使用加权交叉熵损失处理类别不平衡问题：

class BinarySegLoss(tf.keras.losses.Loss):def __init__(self, class_weights=[1.0, 10.0], name='binary_seg_loss'):super(BinarySegLoss, self).__init__(name=name)self.class_weights = class_weightsdef call(self, y_true, y_pred):# y_true: [batch, H, W, 1]# y_pred: [batch, H, W, num_classes]# 将y_true转换为one-hot编码y_true_onehot = tf.one_hot(tf.squeeze(y_true, axis=-1), depth=y_pred.shape[-1], dtype=tf.float32)# 计算交叉熵cross_entropy = -tf.reduce_sum(y_true_onehot * tf.math.log(tf.clip_by_value(y_pred, 1e-10, 1.0)),axis=-1)# 应用类别权重weights = tf.reduce_sum(y_true_onehot * self.class_weights, axis=-1)weighted_loss = cross_entropy * weightsreturn tf.reduce_mean(weighted_loss)

5.2 实例嵌入损失

使用判别损失函数（Discriminative Loss）来学习像素嵌入：

class DiscriminativeLoss(tf.keras.losses.Loss):def __init__(self, delta_var=0.5, delta_dist=1.5, norm=2, alpha=1.0, beta=1.0, gamma=0.001,name='discriminative_loss'):super(DiscriminativeLoss, self).__init__(name=name)self.delta_var = delta_varself.delta_dist = delta_distself.norm = normself.alpha = alphaself.beta = betaself.gamma = gammadef call(self, y_true, y_pred):"""y_true: [batch, H, W, 1] 实例标签图y_pred: [batch, H, W, embedding_dim] 嵌入向量"""batch_size = tf.shape(y_pred)[0]height = tf.shape(y_pred)[1]width = tf.shape(y_pred)[2]embedding_dim = tf.shape(y_pred)[3]# 展平所有维度y_true_flat = tf.reshape(y_true, [batch_size * height * width])y_pred_flat = tf.reshape(y_pred, [batch_size * height * width, embedding_dim])# 获取唯一的实例IDinstance_ids, _ = tf.unique(y_true_flat)instance_ids = instance_ids[instance_ids != 0]  # 移除背景# 如果没有实例，返回0损失if tf.equal(tf.size(instance_ids), 0):return tf.constant(0.0, dtype=tf.float32)# 计算每个实例的均值向量def compute_means(id_val):mask = tf.equal(y_true_flat, id_val)vectors = tf.boolean_mask(y_pred_flat, mask)mean = tf.reduce_mean(vectors, axis=0)return meanmeans = tf.map_fn(compute_means, instance_ids, dtype=tf.float32)# 计算方差项def compute_var_term(id_val, mean):mask = tf.equal(y_true_flat, id_val)vectors = tf.boolean_mask(y_pred_flat, mask)diff = tf.norm(vectors - mean, ord=self.norm, axis=1)diff = tf.maximum(diff - self.delta_var, 0.0)return tf.reduce_mean(tf.square(diff))var_terms = tf.map_fn(lambda x: compute_var_term(x[0], x[1]),(instance_ids, means),dtype=tf.float32)var_loss = tf.reduce_mean(var_terms)# 计算距离项n_instances = tf.size(instance_ids)if n_instances > 1:# 计算所有均值对之间的距离means_a = tf.tile(tf.expand_dims(means, 1), [1, n_instances, 1])means_b = tf.tile(tf.expand_dims(means, 0), [n_instances, 1, 1])diff = means_a - means_bdist = tf.norm(diff, ord=self.norm, axis=2)# 计算距离损失c_dist = 2 * self.delta_dist - distc_dist = tf.maximum(c_dist, 0.0)dist_loss = tf.reduce_mean(tf.square(c_dist))else:dist_loss = tf.constant(0.0, dtype=tf.float32)# 计算正则化项reg_loss = tf.reduce_mean(tf.norm(means, ord=self.norm, axis=1))# 组合损失total_loss = (self.alpha * var_loss + self.beta * dist_loss + self.gamma * reg_loss)return total_loss

5.3 总损失函数

class LaneNetLoss(tf.keras.losses.Loss):def __init__(self, seg_loss_weight=1.0, embedding_loss_weight=0.01, name='lanenet_loss'):super(LaneNetLoss, self).__init__(name=name)self.seg_loss = BinarySegLoss()self.embedding_loss = DiscriminativeLoss()self.seg_loss_weight = seg_loss_weightself.embedding_loss_weight = embedding_loss_weightdef call(self, y_true, y_pred):# y_true: (binary_label, instance_label)# y_pred: (binary_pred, embedding_pred)binary_label, instance_label = y_truebinary_pred, embedding_pred = y_predseg_loss = self.seg_loss(binary_label, binary_pred)embedding_loss = self.embedding_loss(instance_label, embedding_pred)total_loss = (self.seg_loss_weight * seg_loss + self.embedding_loss_weight * embedding_loss)return total_loss

6. 训练流程实现

6.1 数据管道

使用TensorFlow的Dataset API构建高效的数据管道：

class LaneNetDataLoader:def __init__(self, dataset_path, batch_size=8, input_size=(512, 256)):self.dataset_path = dataset_pathself.batch_size = batch_sizeself.input_size = input_sizeself.augmentor = LaneNetAugmentor()def _parse_sample(self, sample):# 读取图像image = tf.io.read_file(sample['image_path'])image = tf.image.decode_jpeg(image, channels=3)image = tf.image.convert_image_dtype(image, tf.float32)# 读取标签seg_label = tf.convert_to_tensor(sample['seg_label'], dtype=tf.uint8)instance_label = tf.convert_to_tensor(sample['instance_label'], dtype=tf.uint8)# 调整大小image = tf.image.resize(image, self.input_size)seg_label = tf.image.resize(tf.expand_dims(seg_label, -1), self.input_size, method='nearest')instance_label = tf.image.resize(tf.expand_dims(instance_label, -1), self.input_size, method='nearest')# 归一化image = (image - 0.5) * 2.0  # [-1, 1]return image, (tf.squeeze(seg_label), tf.squeeze(instance_label))def _augment_sample(self, image, seg_label, instance_label):# 将Tensor转换为numpy进行增强def _augment(image_np, seg_np, instance_np):return self.augmentor(image_np, seg_np, instance_np)image_aug, seg_aug, instance_aug = tf.numpy_function(_augment,[image, seg_label, instance_label],[tf.float32, tf.uint8, tf.uint8])# 设置形状image_aug.set_shape(image.shape)seg_aug.set_shape(seg_label.shape)instance_aug.set_shape(instance_label.shape)return image_aug, seg_aug, instance_augdef get_dataset(self, samples, shuffle=True, augment=True):# 创建数据集dataset = tf.data.Dataset.from_tensor_slices(samples)if shuffle:dataset = dataset.shuffle(len(samples))# 解析样本dataset = dataset.map(self._parse_sample, num_parallel_calls=tf.data.AUTOTUNE)# 数据增强if augment:dataset = dataset.map(self._augment_sample,num_parallel_calls=tf.data.AUTOTUNE)# 批处理dataset = dataset.batch(self.batch_size)dataset = dataset.prefetch(tf.data.AUTOTUNE)return dataset

6.2 训练循环

实现自定义训练循环以更好地控制训练过程：

class LaneNetTrainer:def __init__(self, model, train_dataset, val_dataset, optimizer, loss_fn, log_dir='logs', ckpt_dir='checkpoints'):self.model = modelself.train_dataset = train_datasetself.val_dataset = val_datasetself.optimizer = optimizerself.loss_fn = loss_fn# 设置日志和检查点self.summary_writer = tf.summary.create_file_writer(log_dir)self.ckpt = tf.train.Checkpoint(model=model, optimizer=optimizer)self.ckpt_manager = tf.train.CheckpointManager(self.ckpt, ckpt_dir, max_to_keep=5)# 指标self.train_loss = tf.keras.metrics.Mean(name='train_loss')self.val_loss = tf.keras.metrics.Mean(name='val_loss')self.seg_accuracy = tf.keras.metrics.Accuracy(name='seg_accuracy')@tf.functiondef train_step(self, images, labels):binary_labels, instance_labels = labelswith tf.GradientTape() as tape:# 前向传播binary_pred, embedding_pred = self.model(images, training=True)# 计算损失total_loss = self.loss_fn((binary_labels, instance_labels),(binary_pred, embedding_pred))# 计算梯度并更新权重gradients = tape.gradient(total_loss, self.model.trainable_variables)self.optimizer.apply_gradients(zip(gradients, self.model.trainable_variables))# 更新指标self.train_loss(total_loss)binary_pred_labels = tf.argmax(binary_pred, axis=-1)self.seg_accuracy(tf.reshape(binary_labels, [-1]),tf.reshape(binary_pred_labels, [-1]))return total_loss@tf.functiondef val_step(self, images, labels):binary_labels, instance_labels = labels# 前向传播binary_pred, embedding_pred = self.model(images, training=False)# 计算损失total_loss = self.loss_fn((binary_labels, instance_labels),(binary_pred, embedding_pred))# 更新指标self.val_loss(total_loss)return total_lossdef train(self, epochs, initial_epoch=0):best_val_loss = float('inf')for epoch in range(initial_epoch, epochs):# 重置指标self.train_loss.reset_states()self.val_loss.reset_states()self.seg_accuracy.reset_states()# 训练循环for images, labels in self.train_dataset:self.train_step(images, labels)# 验证循环for val_images, val_labels in self.val_dataset:self.val_step(val_images, val_labels)# 记录日志with self.summary_writer.as_default():tf.summary.scalar('train_loss', self.train_loss.result(), step=epoch)tf.summary.scalar('val_loss', self.val_loss.result(), step=epoch)tf.summary.scalar('seg_accuracy', self.seg_accuracy.result(), step=epoch)# 打印进度template = 'Epoch {}, Loss: {:.4f}, Val Loss: {:.4f}, Accuracy: {:.2%}'print(template.format(epoch + 1,self.train_loss.result(),self.val_loss.result(),self.seg_accuracy.result()))# 保存检查点if self.val_loss.result() < best_val_loss:best_val_loss = self.val_loss.result()self.ckpt_manager.save()print(f'Checkpoint saved at epoch {epoch + 1}')

6.3 训练配置与启动

创建训练脚本train.py：

import os
from model.lanenet import LaneNet
from model.losses import LaneNetLoss
from utils.data_loader import LaneNetDataLoader
from utils.data_processor import TuSimpleProcessor
from trainers.lanenet_trainer import LaneNetTrainer
import tensorflow as tfdef main():# 配置参数config = {'batch_size': 8,'input_size': (512, 256),'learning_rate': 1e-3,'epochs': 100,'dataset_path': 'data/tusimple','log_dir': 'logs/lanenet','ckpt_dir': 'checkpoints/lanenet'}# 准备数据集processor = TuSimpleProcessor(config['dataset_path'])train_samples, val_samples, _ = processor.prepare_dataset()data_loader = LaneNetDataLoader(config['dataset_path'],batch_size=config['batch_size'],input_size=config['input_size'])train_dataset = data_loader.get_dataset(train_samples, shuffle=True, augment=True)val_dataset = data_loader.get_dataset(val_samples, shuffle=False, augment=False)# 初始化模型model = LaneNet()# 优化器和损失函数optimizer = tf.keras.optimizers.Adam(learning_rate=config['learning_rate'])loss_fn = LaneNetLoss()# 创建训练器trainer = LaneNetTrainer(model=model,train_dataset=train_dataset,val_dataset=val_dataset,optimizer=optimizer,loss_fn=loss_fn,log_dir=config['log_dir'],ckpt_dir=config['ckpt_dir'])# 恢复检查点（如果存在）if os.path.exists(config['ckpt_dir']):trainer.ckpt.restore(tf.train.latest_checkpoint(config['ckpt_dir']))print(f"Restored from {tf.train.latest_checkpoint(config['ckpt_dir'])}")# 开始训练trainer.train(epochs=config['epochs'])if __name__ == '__main__':main()

7. 模型评估与推理

7.1 评估指标实现

实现TuSimple数据集官方评估指标：

import numpy as npclass LaneEval:@staticmethoddef get_intersection_ratio(pred, gt):"""计算预测车道线和真实车道线的交并比"""pred = np.array(pred)gt = np.array(gt)# 插值以获得更密集的点pred_interp = LaneEval.interpolate_lane(pred)gt_interp = LaneEval.interpolate_lane(gt)# 计算距离矩阵dist_matrix = np.sqrt((pred_interp[:, np.newaxis, 0] - gt_interp[np.newaxis, :, 0])**2 +(pred_interp[:, np.newaxis, 1] - gt_interp[np.newaxis, :, 1])**2)# 找到匹配点min_dist = np.min(dist_matrix, axis=1)matched = min_dist <= 5  # 5像素阈值if np.sum(matched) == 0:return 0.0ratio = np.sum(matched) / len(pred_interp)return ratio@staticmethoddef interpolate_lane(lane):"""对车道线点进行插值以获得更密集的点"""if len(lane) < 2:return lanex = lane[:, 0]y = lane[:, 1]# 移除重复的y值unique_y = np.unique(y)if len(unique_y) != len(y):# 对每个y值取x的平均值x_new = []for y_val in unique_y:x_new.append(np.mean(x[y == y_val]))x = np.array(x_new)y = unique_y# 插值f = interp1d(y, x, kind='linear', fill_value='extrapolate')y_interp = np.arange(y.min(), y.max() + 1)x_interp = f(y_interp)return np.column_stack((x_interp, y_interp))@staticmethoddef evaluate(pred_lanes, gt_lanes):"""评估预测车道线与真实车道线的匹配情况"""# 计算每个预测车道线与真实车道线的最大交并比ratios = []for pred in pred_lanes:max_ratio = 0for gt in gt_lanes:ratio = LaneEval.get_intersection_ratio(pred, gt)if ratio > max_ratio:max_ratio = ratioratios.append(max_ratio)# 计算准确率和假阳性率accuracy = np.mean([1 if r > 0.5 else 0 for r in ratios])fp = np.mean([1 if r <= 0.5 else 0 for r in ratios])return accuracy, fp

7.2 后处理与车道线聚类

将实例嵌入转换为车道线实例：

import numpy as np
import cv2
from sklearn.cluster import MeanShiftclass LanePostprocessor:def __init__(self, bandwidth=1.5, min_samples=100):self.bandwidth = bandwidthself.min_samples = min_samplesdef process(self, binary_pred, embedding_pred):"""处理模型输出，得到车道线实例参数:binary_pred: [H, W] 二值分割图embedding_pred: [H, W, embedding_dim] 嵌入向量返回:List of lanes, 每个lane是Nx2的数组"""# 获取车道线像素lane_pixels = np.argwhere(binary_pred == 1)if len(lane_pixels) == 0:return []# 获取对应的嵌入向量embeddings = embedding_pred[lane_pixels[:, 0], lane_pixels[:, 1]]# 使用MeanShift聚类clustering = MeanShift(bandwidth=self.bandwidth, min_bin_freq=self.min_samples)clustering.fit(embeddings)labels = clustering.labels_# 按聚类结果分组unique_labels = np.unique(labels)lanes = []for label in unique_labels:# 获取当前cluster的像素坐标cluster_pixels = lane_pixels[labels == label]if len(cluster_pixels) < self.min_samples:continue# 对车道线进行拟合lane = self.fit_lane(cluster_pixels)if lane is not None:lanes.append(lane)return lanesdef fit_lane(self, pixels):"""使用多项式拟合车道线"""if len(pixels) < 10:return None# 按y坐标排序sorted_idx = np.argsort(pixels[:, 0])y = pixels[sorted_idx, 0]x = pixels[sorted_idx, 1]# 使用二阶多项式拟合try:coeffs = np.polyfit(y, x, 2)except:return None# 生成拟合点y_min, y_max = np.min(y), np.max(y)y_range = np.arange(y_min, y_max + 1)x_fit = np.polyval(coeffs, y_range)return np.column_stack((x_fit, y_range))

7.3 推理脚本

创建测试脚本test.py：

import cv2
import numpy as np
import tensorflow as tf
from model.lanenet import LaneNet
from utils.postprocess import LanePostprocessor
from utils.visualization import draw_lanesdef load_model(ckpt_dir):model = LaneNet()ckpt = tf.train.Checkpoint(model=model)latest_ckpt = tf.train.latest_checkpoint(ckpt_dir)if latest_ckpt:ckpt.restore(latest_ckpt)print(f"Restored from {latest_ckpt}")else:raise ValueError("No checkpoint found")return modeldef preprocess_image(image, input_size=(512, 256)):# 调整大小并归一化image = cv2.resize(image, (input_size[1], input_size[0]))image = image.astype(np.float32) / 255.0image = (image - 0.5) * 2.0  # [-1, 1]return np.expand_dims(image, axis=0)def postprocess_output(binary_pred, embedding_pred):# 二值化分割结果binary_pred = np.argmax(binary_pred, axis=-1)[0]# 后处理得到车道线postprocessor = LanePostprocessor()lanes = postprocessor.process(binary_pred, embedding_pred[0])return lanesdef main():# 配置ckpt_dir = 'checkpoints/lanenet'input_size = (512, 256)test_image_path = 'data/test_images/test.jpg'# 加载模型model = load_model(ckpt_dir)# 读取测试图像image = cv2.imread(test_image_path)image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)original_size = image.shape[:2]# 预处理input_image = preprocess_image(image, input_size)# 推理binary_pred, embedding_pred = model.predict(input_image)# 后处理lanes = postprocess_output(binary_pred, embedding_pred)# 可视化result_image = draw_lanes(image, lanes, original_size, input_size)# 显示结果cv2.imshow('Result', cv2.cvtColor(result_image, cv2.COLOR_RGB2BGR))cv2.waitKey(0)cv2.destroyAllWindows()# 保存结果cv2.imwrite('data/test_images/result.jpg', cv2.cvtColor(result_image, cv2.COLOR_RGB2BGR))if __name__ == '__main__':main()

7.4 可视化工具

实现可视化函数utils/visualization.py：

import cv2
import numpy as npdef draw_lanes(image, lanes, original_size, input_size):"""在图像上绘制检测到的车道线参数:image: 原始图像lanes: 检测到的车道线列表original_size: 原始图像大小 (H, W)input_size: 模型输入大小 (H, W)返回:绘制了车道线的图像"""# 调整大小比例h_ratio = original_size[0] / input_size[0]w_ratio = original_size[1] / input_size[1]# 创建副本vis_image = image.copy()# 定义颜色colors = [(255, 0, 0),    # 红色(0, 255, 0),    # 绿色(0, 0, 255),    # 蓝色(255, 255, 0),  # 青色(255, 0, 255),  # 品红(0, 255, 255)   # 黄色]# 绘制每条车道线for i, lane in enumerate(lanes):if len(lane) < 2:continue# 调整坐标到原始图像大小lane[:, 0] = lane[:, 0] * w_ratiolane[:, 1] = lane[:, 1] * h_ratio# 转换为整数坐标lane = lane.astype(np.int32)# 绘制车道线color = colors[i % len(colors)]for j in range(1, len(lane)):cv2.line(vis_image, tuple(lane[j-1]), tuple(lane[j]), color, thickness=5)return vis_image

8. 模型优化与调试

8.1 常见问题与解决方案

在复现LaneNet过程中可能会遇到以下问题：

训练不稳定
- 解决方案：调整学习率，增加梯度裁剪，使用更小的batch size
实例嵌入不收敛
- 解决方案：调整Discriminative Loss的超参数，特别是delta_var和delta_dist
过拟合
- 解决方案：增加数据增强，添加Dropout层，使用权重正则化
推理速度慢
- 解决方案：使用更轻量的骨干网络（如ENet而非ResNet），减小输入尺寸

8.2 性能优化技巧

混合精度训练

from tensorflow.keras import mixed_precision
policy = mixed_precision.Policy('mixed_float16')
mixed_precision.set_global_policy(policy)

使用TensorRT加速推理

# 转换模型为TensorRT格式
conversion_params = tf.experimental.tensorrt.ConversionParams(precision_mode='FP16',maximum_cached_engines=16
)
converter = tf.experimental.tensorrt.Converter(input_saved_model_dir='saved_model',conversion_params=conversion_params
)
converter.convert()
converter.save('tensorrt_model')