端到端自动驾驶——cnn网络搭建

论文参考：https://arxiv.org/abs/1604.07316
demo

今天主要来看一个如何通过图像直接到控制的自动驾驶端到端的项目，首先需要配置好我的仿真环境，下载软件udacity：
https://d17h27t6h515a5.cloudfront.net/topher/2016/November/5831f3a4_simulator-windows-64/simulator-windows-64.zip

现在好的解压即可

运行时打开终端，然后将文件拖入终端中运行

选择合适的窗口大小

成功进入界面

然后配置一下python环境，对于已经有conda环境后，直接

conda create -n cnn python=3.8

conda activate cnn

安装以下依赖：

astor==0.8.1
bidict==0.21.2
certifi==2021.5.30
charset-normalizer==2.0.4
click==8.0.1
colorama==0.4.4
cycler==0.10.0
decorator==5.0.9
dnspython==1.16.0
eventlet==0.31.1
Flask==2.0.1
gast==0.3.3
greenlet==1.1.0
idna==3.2
itsdangerous==2.0.1
Jinja2==3.0.1
joblib==1.0.1
kiwisolver==1.3.1
MarkupSafe==2.0.1
matplotlib==3.4.2
numpy==1.19.3
opencv-python==4.5.3.56
paddlepaddle==2.1.2
pandas==1.3.1
Pillow==8.3.1
protobuf==3.17.3
pyparsing==2.4.7
python-dateutil==2.8.2
python-engineio==3.13.0
python-socketio==4.6.1
pytz==2021.1
requests==2.26.0
scikit-learn==0.24.2
scipy==1.7.1
six==1.16.0
threadpoolctl==2.2.0
urllib3==1.26.6
Werkzeug==2.0.1

代码

git clone https://github.com/chan-yuu/end_to_end_ws.git

数据集可以自己驾驶udacity的车辆收集，也可以直接下载下来训练师需要指定文件夹的路径，这里我使用的是一个csv文件来指定我的文件夹
主要文件放置的结构要写入csv文件中，分别是左中右的摄像头图片路径，方向盘，油门，刹车，速度信息。
打开仿真环境，进入training模式

通过键盘即可控制车辆运行。
可以点击record记录这个过程中的数据，之后就能自动生成需要的数据集内容

python train.py -d xxx.csv

我仔细研读了一下这个代码

import paddle
import argparse
import numpy as np
import paddle.nn as nnfrom paddle.optimizer import Adam
from paddle.callbacks import ModelCheckpoint, EarlyStoppingfrom car.model import build_model
from car.utils import CarDataset, load_data# 设置随机种子，确保结果可复现
np.random.seed(0)def train_model(model, args, X_train, X_valid, y_train, y_valid):"""Train the model"""# 创建一个模型检查点回调，用于在训练过程中保存模型checkpoint = ModelCheckpoint(save_dir=args.save_dir)# 创建一个早停回调，当监控的指标（这里是损失值）在一定轮数（patience）内没有改善时，停止训练earlystopping = EarlyStopping(monitor='loss',mode='min',  # 监控损失值，希望其越小越好patience=10,  # 允许损失值在 10 个 epoch 内没有改善verbose=1,  # 打印早停信息min_delta=0,  # 损失值的最小改善量baseline=None,  # 基线值，这里不使用save_best_model=True)  # 保存最佳模型# 根据命令行参数决定是否使用早停回调if args.early_stop:cbs = [checkpoint, earlystopping]else:cbs = [checkpoint]# 创建 Adam 优化器，用于更新模型的参数opt = Adam(learning_rate=args.learning_rate, parameters=model.parameters())# 将模型包装为 paddle.Model 对象，方便进行训练和评估model = paddle.Model(model)# 配置模型的损失函数和优化器model.prepare(loss=nn.MSELoss(), optimizer=opt)# 创建训练数据集对象train_dataset = CarDataset(args.data_dir, X_train, y_train, True)# 创建验证数据集对象val_dataset = CarDataset(args.data_dir, X_valid, y_valid, False)# 开始训练模型model.fit(train_data=train_dataset,  # 训练数据集eval_data=val_dataset,  # 验证数据集epochs=args.nb_epoch,  # 训练的轮数batch_size=args.batch_size,  # 每个批次的样本数量save_dir=args.save_dir,  # 模型保存的目录callbacks=cbs,  # 回调函数列表verbose=1)  # 打印训练进度信息def s2b(s):"""Converts a string to boolean value"""# 将字符串转换为小写s = s.lower()# 判断字符串是否表示真return s == 'true' or s == 'yes' or s == 'y' or s == '1'def main():"""Load train/validation data set and train the model"""# 创建命令行参数解析器parser = argparse.ArgumentParser(description='Behavioral Cloning Training Program')# 添加数据目录参数，默认值为 'data'parser.add_argument('-d',help='data directory',dest='data_dir',type=str,default='data')# 添加模型保存目录参数，默认值为 'save'parser.add_argument('-s',help='save directory',dest='save_dir',type=str,default='save')# 添加测试集大小比例参数，默认值为 0.2parser.add_argument('-t',help='test size fraction',dest='test_size',type=float,default=0.2)# 添加 Dropout 概率参数，默认值为 0.5parser.add_argument('-k',help='drop out probability',dest='keep_prob',type=float,default=0.5)# 添加训练轮数参数，默认值为 100parser.add_argument('-n',help='number of epochs',dest='nb_epoch',type=int,default=100)# 添加批次大小参数，默认值为 40parser.add_argument('-b',help='batch size',dest='batch_size',type=int,default=40)# 添加学习率参数，默认值为 1.0e-4parser.add_argument('-l',help='learning rate',dest='learning_rate',type=float,default=1.0e-4)# 添加早停参数，默认值为 Falseparser.add_argument('-e',help='early stop',dest='early_stop',type=bool,default=False)# 解析命令行参数args = parser.parse_args()print('-' * 30)print('Parameters')print('-' * 30)# 打印所有命令行参数for key, value in vars(args).items():print('{:<20} := {}'.format(key, value))print('-' * 30)# 加载训练数据和验证数据data = load_data(args)# 构建模型model = build_model(args.keep_prob)# 调用训练函数进行模型训练train_model(model, args, *data)if __name__ == '__main__':# 程序入口，调用 main 函数main()

训练结束后可以得到对应的模型

使用这个模型进行测试
打开仿真软件的auto模式

此时是无法记录的，然后我可以加载模型并驾驶车辆

python drive.py ./pretrained_models/model_paddle_test2.pdparams

同样，自己写一边这个代码更容易理解：

import os
import base64
import paddle
import shutil
import argparse
import socketio
import eventlet
import numpy
import eventlet.wsgi
from PIL import Imagefrom io import BvtesIO
from flask import Flask
from datatime import datatime
from car.model import build_model
from car.utils import preprocess# 创建一个socket.IO服务器
sio = socket.Server()
app = Flask(__name__)
# 初始化模型变量，用于后续加载模型
model = None
# 初始化上一帧图像数组变量，用于记录上一帧的图像数据
prev_image_array = None# 定义最大速度和最小速度
MAX_SPEED = 25
MIN_SPEED = 10# 初始化速度限制为最大速度
speed_limit = MAX_SPEED# 定义一个事件处理函数，当接收到 'telemetry' 事件时触发
@sio.on('telemetry')
def telemetry(sid, data):if data:# 从接收到的数据中提取当前汽车的转向角度steering_angle = float(data["steering_angle"])# 从接收到的数据中提取当前汽车的油门值throttle = float(data["throttle"])# 从接收到的数据中提取当前汽车的速度speed = float(data["speed"])# 从接收到的数据中提取当前汽车中心摄像头的图像，并将其解码为 PIL 图像对象image = Image.open(BytesIO(base64.b64decode(data["image"])))# 如果指定了图像保存文件夹，则保存当前帧图像if args.image_folder != '':# 生成当前时间戳，用于作为图像文件名timestamp = datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S_%f')[:-3]# 构建图像文件的完整路径image_filename = os.path.join(args.image_folder, timestamp)# 保存图像为 JPEG 格式image.save('{}.jpg'.format(image_filename))try:# 将 PIL 图像对象转换为 NumPy 数组image = np.asarray(image)# 对图像进行预处理，例如裁剪、归一化等操作image = preprocess(image)# 为图像数组添加一个维度，使其成为 4D 数组，以满足模型输入要求image = np.array([image])# 使用模型对图像进行预测，得到转向角度的预测值steering_angle = model(image.astype('float32') / 127.5 - 1.0).item()# 根据当前速度调整速度限制和油门值global speed_limitif speed > speed_limit:# 如果当前速度超过速度限制，则将速度限制降低到最小速度，以减速speed_limit = MIN_SPEEDelse:# 如果当前速度低于速度限制，则将速度限制恢复到最大速度speed_limit = MAX_SPEED# 根据转向角度和速度计算油门值throttle = 1.0 - steering_angle**2 - (speed / speed_limit)**2# 打印当前的转向角度、油门值和速度print(f'steering_angle={steering_angle:.3f}, throttle={throttle:.3f}, speed={speed:.3f}')# 发送控制指令，包括转向角度和油门值send_control(steering_angle, throttle)except Exception as e:# 打印异常信息print(e)else:# 如果没有接收到数据，则发送手动控制指令sio.emit('manual', data={}, skip_sid=True)# 定义一个事件处理函数，当有新的客户端连接时触发
@sio.on('connect')
def connect(sid, environ):# 打印连接信息print("connect ", sid)# 发送初始控制指令，将转向角度和油门值都设为 0send_control(0, 0)# 定义一个函数，用于发送控制指令
def send_control(steering_angle, throttle):# 向客户端发送 'steer' 事件，包含转向角度和油门值sio.emit("steer",data={'steering_angle': steering_angle.__str__(),'throttle': throttle.__str__()},skip_sid=True)if __name__ == '__main__':# 创建一个命令行参数解析器parser = argparse.ArgumentParser(description='Remote Driving')# 添加一个必需的命令行参数，用于指定模型文件的路径parser.add_argument('model',type=str,help='Path to model h5 file. Model should be on the same path.')# 添加一个可选的命令行参数，用于指定图像保存文件夹的路径parser.add_argument('image_folder',type=str,nargs='?',default='',help='Path to image folder. This is where the images from the run will be saved.')# 解析命令行参数args = parser.parse_args()# 构建模型model = build_model()# 加载模型的参数params = paddle.load(args.model)# 将加载的参数设置到模型中model.set_dict(params)# 将模型转换为静态图模式，以提高推理速度model = paddle.jit.to_static(model)# 将模型设置为评估模式model.eval()# 如果指定了图像保存文件夹if args.image_folder != '':# 打印创建图像文件夹的信息print("Creating image folder at {}".format(args.image_folder))# 如果文件夹不存在，则创建它if not os.path.exists(args.image_folder):os.makedirs(args.image_folder)else:# 如果文件夹已存在，则先删除它，再重新创建shutil.rmtree(args.image_folder)os.makedirs(args.image_folder)# 打印记录运行信息print("RECORDING THIS RUN ...")else:# 如果没有指定图像保存文件夹，则打印不记录运行信息print("NOT RECORDING THIS RUN ...")# 使用 Socket.IO 中间件包装 Flask 应用app = socketio.Middleware(sio, app)# 使用 eventlet 启动一个 WSGI 服务器，监听 4567 端口eventlet.wsgi.server(eventlet.listen(('', 4567)), app)

即可实现基于视觉的自动驾驶功能。后面这篇文章还会继续完善论文中的一些观点和代码的一些学习过程。
https://github.com/naokishibuya/car-behavioral-cloning?tab=readme-ov-file