论文参考：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 nn
from paddle.optimizer import Adam
from paddle.callbacks import ModelCheckpoint, EarlyStopping
from 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.2
parser.add_argument('-t',
help='test size fraction',
dest='test_size',
type=float,
default=0.2)
# 添加 Dropout 概率参数，默认值为 0.5
parser.add_argument('-k',
help='drop out probability',
dest='keep_prob',
type=float,
default=0.5)
# 添加训练轮数参数，默认值为 100
parser.add_argument('-n',
help='number of epochs',
dest='nb_epoch',
type=int,
default=100)
# 添加批次大小参数，默认值为 40
parser.add_argument('-b',
help='batch size',
dest='batch_size',
type=int,
default=40)
# 添加学习率参数，默认值为 1.0e-4
parser.add_argument('-l',
help='learning rate',
dest='learning_rate',
type=float,
default=1.0e-4)
# 添加早停参数，默认值为 False
parser.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 Image
from 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_limit
if speed > speed_limit:
# 如果当前速度超过速度限制，则将速度限制降低到最小速度，以减速
speed_limit = MIN_SPEED
else:
# 如果当前速度低于速度限制，则将速度限制恢复到最大速度
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)
# 发送初始控制指令，将转向角度和油门值都设为 0
send_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
