一、理论知识

参考NVIDIA的论文/pdf/1604.07316.pdf

这个实验的目的是使用卷积神经网络（CNN）将从前向摄像机得到的原始图像映射成自动驾驶汽车的驾驶命令。

这个过程需要先采集数据，原理是这样的：

一共有左、中、右3个摄像头，负责采集视频数据。在训练后我们只需要中间的摄像头采集的数据用于驾驶决策，那为什么需要3个采集训练数据呢，因为左右两个采集特定偏离中心的变化图像，这样只有来自人类驾驶员的数据是不足以用来训练的；网络还必须学习如何从任何错误中恢复，否则该汽车就将慢慢偏移道路。因此训练数据还扩充了额外的图像，这些图像显示了远离车道中心的偏离程度以及不同道路方向上的转动。左右两个摄像头采集负责特定偏离中心的变化图像，用这个方法来扩充额外图像数据

下面是训练系统框图。训练用图像被送入一个卷积神经网络，然后计算一个被预测出的转向命令。这个被转向命令会与该图像的期望命令相比较，卷积神经网络的权重就会被调整以使其实际输出更接近期望输出。

一旦训练完成，网络就能够从中摄像机的视频图像中生成转向命令

图像数据需要从RGB转成YUV格式，传入一个9 层网络架构的神经网络，其中包括一个归一化层、5 个卷积层和 3 个完全连接的层。

二、无人驾驶模拟环境搭建

这里我们使用Udacity的自动驾驶汽车模拟器，下载地址是/udacity/self-driving-car-sim

下载后运行程序选择Training Mode，就可以开始模拟驾驶了，玩得差不多了，就可以录制训练数据了

决定录制的时候在训练模式下点Record进行录制，第一次需要选择用于存放3个摄像头的图片与driving_log.csv数据

这是中间摄像头的图像

找两张分别是左、右摄像头的图像

好吧，看得出来我开得不好，其实我是让他帮我开的

三、训练程序

数据预处理

csv文件的各个字段含义是：center,left,right,steering,throttle,reverse,speed，我们需要的训练数据（center，left，right）和标签数据（steering）都在里面

#加载训练数据并将其分解为训练和验证集def load_data(self):#从csv读取数据data_df = pd.read_csv(os.path.join(os.getcwd(), self.data_path, 'driving_log.csv'),names=['center', 'left', 'right', 'steering', 'throttle', 'reverse', 'speed'])X = data_df[['center', 'left', 'right']].valuesy = data_df['steering'].values#随机划分训练集和测试集X_train, X_valid, y_train, y_valid = train_test_split(X, y, test_size=self.test_size, random_state=0)return X_train, X_valid, y_train, y_valid

图片的处理

首先需要取一部分图片进行增强处理（这里是60%的概率）

进行增强处理流程是：先平均概率随机取一个摄像头的图片，如果取得的不是中间的摄像头，对于的steering angle也进行相应的调整

#从中心、左或右随机选择一个图像，并进行调整def choose_image(self, center, left, right, steering_angle):choice = np.random.choice(3)if choice == 0:return self.load_image(left), steering_angle + 0.2elif choice == 1:return self.load_image(right), steering_angle - 0.2return self.load_image(center), steering_angle

接下来是进行随机反转，这个步骤是为了防止训练数据过少，比如都是偏左或都是右拐的数据，翻转的时候一样要翻转转向角

#随机反转图片def random_flip(self, image, steering_angle):if np.random.rand() < 0.5:image = cv2.flip(image, 1)steering_angle = -steering_anglereturn image, steering_angle

然后是使用cv2.warpAffine()函数随机平移变换，第三个参数是输出图像的大小。第二个参数是变换矩阵

#随机平移变换def random_translate(self, image, steering_angle, range_x, range_y):trans_x = range_x * (np.random.rand() - 0.5)trans_y = range_y * (np.random.rand() - 0.5)steering_angle += trans_x * 0.002trans_m = np.float32([[1, 0, trans_x], [0, 1, trans_y]])height, width = image.shape[:2]image = cv2.warpAffine(image, trans_m, (width, height))return image, steering_angle

在对图片随机加入阴影，并对路上的白线进行加强，这里放几个对比图能更直观感受到

中间为了方便处理图片，把RGB格式转换为HLS格式和HSV格式

#添加随机阴影def random_shadow(self, image):x1, y1 = self.IMAGE_WIDTH * np.random.rand(), 0x2, y2 = self.IMAGE_WIDTH * np.random.rand(), self.IMAGE_HEIGHT#xm, ym = np.mgrid[0:self.IMAGE_HEIGHT, 0:self.IMAGE_WIDTH]xm, ym = np.mgrid[0:image.shape[0], 0:image.shape[1]]mask = np.zeros_like(image[:, :, 1])mask[(ym - y1) * (x2 - x1) - (y2 - y1) * (xm - x1) > 0] = 1cond = mask == np.random.randint(2)s_ratio = np.random.uniform(low=0.2, high=0.5)hls = cv2.cvtColor(image, cv2.COLOR_RGB2HLS)hls[:, :, 1][cond] = hls[:, :, 1][cond] * s_ratioreturn cv2.cvtColor(hls, cv2.COLOR_HLS2RGB)#随机调整亮度def random_brightness(self, image):hsv = cv2.cvtColor(image, cv2.COLOR_RGB2HSV)ratio = 1.0 + 0.4 * (np.random.rand() - 0.5)hsv[:, :, 2] = hsv[:, :, 2] * ratioreturn cv2.cvtColor(hsv, cv2.COLOR_HSV2RGB)

对图像进行增强后，就是对图片进行最后整理：先去掉天空和汽车部分，由320*160变成200*66的小图，然后转换为YUV格式

#除去顶部的天空和底部的汽车正面def crop(self, image):return image[60:-25, :, :]#调整图像大小def resize(self, image):return cv2.resize(image, (self.IMAGE_WIDTH, self.IMAGE_HEIGHT), cv2.INTER_AREA)#转换RGB为YUV格式def rgb2yuv(self, image):return cv2.cvtColor(image, cv2.COLOR_RGB2YUV)

神经网络模型

好了，准备好数据接下来就是我们的模型了，按论文进行构建就可以了，稍微可以做些自己的调整，比如这里加了Dropout

#构建模型def build_model(self):model = Sequential()model.add(Lambda(lambda x: x / 127.5 - 1.0, input_shape=(self.IMAGE_HEIGHT, self.IMAGE_WIDTH, self.IMAGE_CHANNELS)))model.add(Conv2D(24, (5, 5), activation='elu', strides=2))model.add(Conv2D(36, (5, 5), activation='elu', strides=2))model.add(Conv2D(48, (5, 5), activation='elu', strides=2))model.add(Conv2D(64, (3, 3), activation='elu'))model.add(Conv2D(64, (3, 3), activation='elu'))model.add(Dropout(self.keep_prob))model.add(Flatten())model.add(Dense(100, activation='elu'))model.add(Dense(50, activation='elu'))model.add(Dense(10, activation='elu'))model.add(Dense(1))model.summary()return model

Layer (type) Output Shape Param #

=================================================================

lambda_1 (Lambda)(None, 66, 200, 3) 0

_________________________________________________________________

conv2d_1 (Conv2D)(None, 31, 98, 24) 1824

_________________________________________________________________

conv2d_2 (Conv2D)(None, 14, 47, 36) 21636

_________________________________________________________________

conv2d_3 (Conv2D)(None, 5, 22, 48) 43248

_________________________________________________________________

conv2d_4 (Conv2D)(None, 3, 20, 64) 27712

_________________________________________________________________

conv2d_5 (Conv2D)(None, 1, 18, 64) 36928

_________________________________________________________________

dropout_1 (Dropout)(None, 1, 18, 64) 0

_________________________________________________________________

flatten_1 (Flatten)(None, 1152) 0

_________________________________________________________________

dense_1 (Dense) (None, 100) 115300

_________________________________________________________________

dense_2 (Dense) (None, 50) 5050

_________________________________________________________________

dense_3 (Dense) (None, 10) 510

_________________________________________________________________

dense_4 (Dense) (None, 1) 11

=================================================================

Total params: 252,219

训练

可以开始训练了，这里用fit_generator，这样可以CPU实时图像增强同时并行在GPU训练模型

# CPU做图像的实时数据增强,并行在GPU训练模型model.fit_generator(self.batch_generator(X_train, y_train, True),steps_per_epoch = self.steps_per_epoch,epochs = self.epochs,#max_queue_size=1,validation_data=self.batch_generator(X_valid, y_valid, False),validation_steps=len(X_valid),callbacks=[checkpoint],verbose=1)

我尝试训练了50000次挑战了下山路，loss到了 0.0486左右就迫不及待的去测试去了

四、模拟自动驾驶

Python使用socketio通过4567端口连接到UDACITY模拟器，当收到数据后，就根据当前的中心摄像头图像进行预测转向，然后根据转向角度与速度计算油门

# 汽车的当前转向角#steering_angle = float(data["steering_angle"])# 汽车的油门#throttle = float(data["throttle"])# 当前的速度speed = float(data["speed"])# 中心摄像头image = Image.open(BytesIO(base64.b64decode(data["image"])))try:image = np.asarray(image) # from PIL image to numpy arrayimage = preprocess(image) # apply the preprocessingimage = np.array([image]) # the model expects 4D array# 预测图像的转向steering_angle = float(model.predict(image, batch_size=1))# 根据速度调整油门，如果大于最大速度就减速，如果小于最低速度就加加速if speed > MAX_SPEED:speed_limit = MIN_SPEED # slow downelse:speed_limit = MAX_SPEEDthrottle = 1.0 - steering_angle ** 2 - (speed / speed_limit) ** 2print('{} {} {}'.format(steering_angle, throttle, speed))send_control(steering_angle, throttle)except Exception as e:print(e)

在UDACITY模拟器里选择Autonomous Mode，运行起模拟驾驶程序就可以了

五、完整程序

训练程序，里面的参数可以随意实验调整看看

PowerMode_autopilot.py

# -*- coding: utf-8 -*-import pandas as pdimport cv2, osfrom sklearn.model_selection import train_test_splitfrom keras.models import Sequentialfrom keras.layers import Lambda, Conv2D, Dropout, Dense, Flattenfrom keras.callbacks import ModelCheckpointimport numpy as npimport matplotlib.image as mpimgfrom keras.optimizers import Adamfrom keras.models import load_modelclass PowerMode_autopilot:def __init__(self, data_path = 'data', learning_rate = 1.0e-4, keep_prob = 0.5 , batch_size = 40,save_best_only = True, test_size = 0.2, steps_per_epoch = 20000, epochs = 10):self.IMAGE_HEIGHT, self.IMAGE_WIDTH, self.IMAGE_CHANNELS = 66, 200, 3self.data_path = data_pathself.learning_rate = learning_rateself.keep_prob = keep_probself.save_best_only = save_best_onlyself.batch_size = batch_sizeself.test_size = test_sizeself.steps_per_epoch = steps_per_epochself.epochs = epochs#加载训练数据并将其分解为训练和验证集def load_data(self):#从csv读取数据data_df = pd.read_csv(os.path.join(os.getcwd(), self.data_path, 'driving_log.csv'),names=['center', 'left', 'right', 'steering', 'throttle', 'reverse', 'speed'])X = data_df[['center', 'left', 'right']].valuesy = data_df['steering'].values#随机划分训练集和测试集X_train, X_valid, y_train, y_valid = train_test_split(X, y, test_size=self.test_size, random_state=0)return X_train, X_valid, y_train, y_valid#构建模型def build_model(self):model = Sequential()model.add(Lambda(lambda x: x / 127.5 - 1.0, input_shape=(self.IMAGE_HEIGHT, self.IMAGE_WIDTH, self.IMAGE_CHANNELS)))model.add(Conv2D(24, (5, 5), activation='elu', strides=2))model.add(Conv2D(36, (5, 5), activation='elu', strides=2))model.add(Conv2D(48, (5, 5), activation='elu', strides=2))model.add(Conv2D(64, (3, 3), activation='elu'))model.add(Conv2D(64, (3, 3), activation='elu'))model.add(Dropout(self.keep_prob))model.add(Flatten())model.add(Dense(100, activation='elu'))model.add(Dense(50, activation='elu'))model.add(Dense(10, activation='elu'))model.add(Dense(1))model.summary()return model#加载图片def load_image(self, image_file):return mpimg.imread(os.path.join(self.data_path, image_file.strip()))# ---------增强处理-------#从中心、左或右随机选择一个图像，并进行调整def choose_image(self, center, left, right, steering_angle):choice = np.random.choice(3)if choice == 0:return self.load_image(left), steering_angle + 0.2elif choice == 1:return self.load_image(right), steering_angle - 0.2return self.load_image(center), steering_angle#随机反转图片def random_flip(self, image, steering_angle):if np.random.rand() < 0.5:image = cv2.flip(image, 1)steering_angle = -steering_anglereturn image, steering_angle#随机平移变换model.summary()def random_translate(self, image, steering_angle, range_x, range_y):trans_x = range_x * (np.random.rand() - 0.5)trans_y = range_y * (np.random.rand() - 0.5)steering_angle += trans_x * 0.002trans_m = np.float32([[1, 0, trans_x], [0, 1, trans_y]])height, width = image.shape[:2]image = cv2.warpAffine(image, trans_m, (width, height))return image, steering_angle#添加随机阴影def random_shadow(self, image):x1, y1 = self.IMAGE_WIDTH * np.random.rand(), 0x2, y2 = self.IMAGE_WIDTH * np.random.rand(), self.IMAGE_HEIGHT#xm, ym = np.mgrid[0:self.IMAGE_HEIGHT, 0:self.IMAGE_WIDTH]xm, ym = np.mgrid[0:image.shape[0], 0:image.shape[1]]mask = np.zeros_like(image[:, :, 1])mask[(ym - y1) * (x2 - x1) - (y2 - y1) * (xm - x1) > 0] = 1cond = mask == np.random.randint(2)s_ratio = np.random.uniform(low=0.2, high=0.5)hls = cv2.cvtColor(image, cv2.COLOR_RGB2HLS)hls[:, :, 1][cond] = hls[:, :, 1][cond] * s_ratioreturn cv2.cvtColor(hls, cv2.COLOR_HLS2RGB)#随机调整亮度def random_brightness(self, image):hsv = cv2.cvtColor(image, cv2.COLOR_RGB2HSV)ratio = 1.0 + 0.4 * (np.random.rand() - 0.5)hsv[:, :, 2] = hsv[:, :, 2] * ratioreturn cv2.cvtColor(hsv, cv2.COLOR_HSV2RGB)#产生一个增强图像和调整转向角def augument(self, center, left, right, steering_angle, range_x=100, range_y=10):image, steering_angle = self.choose_image(center, left, right, steering_angle)#随机选择一个图像，并进行调整image, steering_angle = self.random_flip(image, steering_angle)#翻转image, steering_angle = self.random_translate(image, steering_angle, range_x, range_y) #移动image = self.random_shadow(image) #加阴影image = self.random_brightness(image) #亮度return image, steering_angle#------图像预处理-------------#除去顶部的天空和底部的汽车正面def crop(self, image):return image[60:-25, :, :]#调整图像大小def resize(self, image):return cv2.resize(image, (self.IMAGE_WIDTH, self.IMAGE_HEIGHT), cv2.INTER_AREA)#转换RGB为YUV格式def rgb2yuv(self, image):return cv2.cvtColor(image, cv2.COLOR_RGB2YUV)#图像预处理def preprocess(self, image):image = self.crop(image)image = self.resize(image)image = self.rgb2yuv(image)return image#-------生成训练图像--------------#生成训练图像，给出图像路径和相关的转向角def batch_generator(self, image_paths, steering_angles, is_training):images = np.empty([self.batch_size, self.IMAGE_HEIGHT, self.IMAGE_WIDTH, self.IMAGE_CHANNELS])steers = np.empty(self.batch_size)while True:i = 0for index in np.random.permutation(image_paths.shape[0]):center, left, right = image_paths[index]steering_angle = steering_angles[index]# 0.6的概率做图像增强if is_training and np.random.rand() < 0.6:image, steering_angle = self.augument(center, left, right, steering_angle)else:image = self.load_image(center)#将图像和转向角度添加到批处理中images[i] = self.preprocess(image)steers[i] = steering_anglei += 1if i == self.batch_size:breakyield images, steers#训练数据def train_model(self, model, X_train, X_valid, y_train, y_valid):#用户每次epoch之后保存模型数据checkpoint = ModelCheckpoint('model-{epoch:03d}.h5',monitor='val_loss',verbose=0,# 如果save_best_only = True，则最近验证误差最好的模型数据会被保存下来save_best_only=self.save_best_only,mode='auto')pile(loss='mean_squared_error', optimizer=Adam(lr=self.learning_rate))# CPU做图像的实时数据增强,并行在GPU训练模型model.fit_generator(self.batch_generator(X_train, y_train, True),steps_per_epoch = self.steps_per_epoch,epochs = self.epochs,#max_queue_size=1,validation_data=self.batch_generator(X_valid, y_valid, False),validation_steps=len(X_valid),callbacks=[checkpoint],verbose=1)def main():autopilot = PowerMode_autopilot(data_path = 'data', learning_rate = 1.0e-4, keep_prob = 0.5 , batch_size = 40,save_best_only = True, test_size = 0.2, steps_per_epoch = 2000, epochs = 10)data = autopilot.load_data()model = autopilot.build_model()#model = load_model('model-xx.h5')#model.summary()autopilot.train_model(model, *data)if __name__ == '__main__':main()

自动驾驶程序

PowerDirve.py

# -*- coding: utf-8 -*-import os, cv2, socketio, base64, shutil, eventlet.wsgiimport numpy as npfrom keras.models import load_modelfrom flask import Flaskfrom PIL import Imagefrom io import BytesIOfrom datetime import datetimefrom keras.preprocessing.image import array_to_img# socketiosio = socketio.Server()# ------图像预处理-------------# 除去顶部的天空和底部的汽车正面def crop(image):return image[60:-25, :, :]# 调整图像大小def resize(image):return cv2.resize(image, (IMAGE_WIDTH, IMAGE_HEIGHT), cv2.INTER_AREA)# 转换RGB为YUV格式def rgb2yuv(image):return cv2.cvtColor(image, cv2.COLOR_RGB2YUV)# 图像预处理def preprocess(image):image = crop(image)image = resize(image)image = rgb2yuv(image)return image@sio.on('telemetry')def telemetry(sid, data):if data:# 汽车的当前转向角#steering_angle = float(data["steering_angle"])# 汽车的油门#throttle = float(data["throttle"])# 当前的速度speed = float(data["speed"])# 中心摄像头image = Image.open(BytesIO(base64.b64decode(data["image"])))try:image = np.asarray(image) # from PIL image to numpy arrayimage = preprocess(image) # apply the preprocessingimage = np.array([image]) # the model expects 4D array# 预测图像的转向steering_angle = float(model.predict(image, batch_size=1))# 根据速度调整油门，如果大于最大速度就减速，如果小于最低速度就加加速if speed > MAX_SPEED:speed_limit = MIN_SPEED # slow downelse:speed_limit = MAX_SPEEDthrottle = 1.0 - steering_angle ** 2 - (speed / speed_limit) ** 2print('{} {} {}'.format(steering_angle, throttle, speed))send_control(steering_angle, throttle)except Exception as e:print(e)# save frameif image_folder != '':timestamp = datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S_%f')[:-3]image_filename = os.path.join(image_folder, timestamp)array_to_img(image[0]).save('{}.jpg'.format(image_filename))else:sio.emit('manual', data={}, skip_sid=True)@sio.on('connect')def connect(sid, environ):print("connect ", sid)send_control(0, 0)def send_control(steering_angle, throttle):sio.emit("steer",data={'steering_angle': steering_angle.__str__(),'throttle': throttle.__str__()},skip_sid=True)if __name__ == '__main__':IMAGE_HEIGHT, IMAGE_WIDTH, IMAGE_CHANNELS = 66, 200, 3MAX_SPEED, MIN_SPEED = 25, 10# 载入模型model = load_model('model-xx.h5')image_folder = ''# 设定图片缓存目录if image_folder != '':if os.path.exists(image_folder):shutil.rmtree(image_folder)os.makedirs(image_folder)app = Flask(__name__)app = socketio.Middleware(sio, app)eventlet.wsgi.server(eventlet.listen(('', 4567)), app)

如果觉得《实战端到端深度学习模拟无人驾驶》对你有帮助，请点赞、收藏，并留下你的观点哦！