利用树莓派和神经网络，实现智能小车
2018-09-07 本文已影响0人
SJTU_JORY
大众实习任务
修改英伟达的端对端神经网络，在树莓派上实现智能小车
1.图像处理
# -*- coding: utf-8 -*-
"""
Created on Tue Jul 17 13:38:23 2018
@author: 51207
"""
import os
from PIL import Image
import numpy as np
import xlrd
import pickle
def loadimg_infile(imgdir):
imgname=[]
imgname_sorted=[]
filename=os.listdir(imgdir)
for file in filename:
if 'jpg' in file:
imgname.append(imgdir+file)
s=len(imgname)
for i in range(s):
imgname_sorted.append(imgdir+'/img_'+str(i)+'.jpg')
return imgname_sorted
def imglist_to_nparray(imglist):
result=np.zeros((len(imglist),90,160,3),dtype=np.uint8)
i=0
for img in imglist:
image=np.array(Image.open(img))
result[i]=image
i+=1
return result
imgdir='D:/mywork/aidrive/intern/smallcar/data/shoudong/image3'
x=loadimg_infile(imgdir)
X=imglist_to_nparray(x)
ydata=imgdir+'/ydata.xlsx'
workbook=xlrd.open_workbook(ydata)
sheet=workbook.sheet_by_name('Sheet1')
Y=np.array(sheet.col_values(0),dtype=np.uint8)
with open('datashoudong3.pkl','wb') as fp:
pickle.dump([X,Y],fp)
2.模型训练
# -*- coding: utf-8 -*-
"""
Created on Tue Jul 3 10:43:21 2018
@author: 51207
"""
import os
import pickle
import matplotlib
from matplotlib.pyplot import imshow
file_path="D:/mywork/aidrive/intern/smallcar/data/alldata.pkl"
#提取数据
with open(file_path,'rb') as f:
X,Y=pickle.load(f)
#划分数据集
import numpy as np
def unison_shuffled_copies(X,Y):
assert len(X)==len(Y)
p=np.random.permutation(len(X))
return X[p], Y[p]
shuffled_X,shuffled_Y=unison_shuffled_copies(X,Y)
test_cutoff=int(len(X)* .8)
val_cutoff=test_cutoff+int(len(X)* .1)
train_X,train_Y=shuffled_X[:test_cutoff],shuffled_Y[:test_cutoff]
val_X, val_Y = shuffled_X[test_cutoff:val_cutoff], shuffled_Y[test_cutoff:val_cutoff]
test_X, test_Y = shuffled_X[val_cutoff:], shuffled_Y[val_cutoff:]
'''
#增强训练集
X_flipped = np.array([np.fliplr(i) for i in train_X])
Y_flipped = np.array([-i for i in train_Y])
train_X = np.concatenate([train_X, X_flipped])
train_Y = np.concatenate([train_Y, Y_flipped])
'''
#建模型
from keras.models import Model, load_model
from keras.layers import Input, Convolution2D, MaxPooling2D, Activation, Dropout, Flatten, Dense
img_in=Input(shape=(90,160,3),name='img_in')
angle_in=Input(shape=(1,),name='angle_in')
x=Convolution2D(8,3,3)(img_in)
x=Activation('relu')(x)
x=MaxPooling2D(pool_size=(2,2))(x)
x=Convolution2D(16,3,3)(x)
x=Activation('relu')(x)
x=MaxPooling2D(pool_size=(2,2))(x)
x=Convolution2D(32,3,3)(x)
x=Activation('relu')(x)
x=MaxPooling2D(pool_size=(2,2))(x)
merged=Flatten()(x)
x=Dense(256)(merged)
x=Activation('linear')(x)
x=Dropout(.2)(x)
angle_out=Dense(1,name='angle_out')(x)
model=Model(input=[img_in],output=[angle_out])
model.compile(optimizer='adam',loss='mean_squared_error')
model.summary()
from keras import callbacks
model_path = os.path.expanduser('~/best_autopilot.hdf5')
save_best = callbacks.ModelCheckpoint(model_path, monitor='val_loss', verbose=1,save_best_only=True, mode='min')
early_stop = callbacks.EarlyStopping(monitor='val_loss', min_delta=0, patience=5,verbose=0, mode='auto')
callbacks_list = [save_best, early_stop]
model.fit(train_X, train_Y, batch_size=50, nb_epoch=20, validation_data=(val_X, val_Y), callbacks=callbacks_list)
import pandas as pd
model = load_model(model_path)
test_P = model.predict(test_X)
test_P = test_P.reshape((test_P.shape[0],))
df = pd.DataFrame({'predicted':test_P, 'actual':test_Y})
ax = df.plot.scatter('predicted', 'actual')
P = model.predict(X[:150])
#predict outputs nested arrays so we need to reshape to plot.
P = P.reshape((P.shape[0],))
ax = pd.DataFrame({'predicted':P, 'actual':Y[:150]}).plot()
ax.set_ylabel("steering angle")
3.树莓派上运行
# -*- coding: utf-8 -*
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BOARD)
GPIO.setup(32, GPIO.IN) # right US Echo
GPIO.setup(36, GPIO.OUT) # right US Trig
GPIO.setup(15, GPIO.IN) # front US Echo
GPIO.setup(38, GPIO.OUT) # front US Trig
GPIO.setup(37, GPIO.OUT,initial = GPIO.HIGH) # red1
GPIO.setup(35, GPIO.OUT,initial = GPIO.HIGH) # red2
GPIO.setup(33, GPIO.OUT,initial = GPIO.LOW) # yellow_right
GPIO.setup(31, GPIO.OUT,initial = GPIO.LOW) # yellow_left
GPIO.setup(11, GPIO.OUT,initial = GPIO.HIGH) # Beep
from picamera.array import PiRGBArray
from picamera import PiCamera
import serial
import time
import cv2
import numpy as np
from keras.models import load_model
ser = serial.Serial("/dev/ttyUSB0", 9600)
def set_picamera():
camera = PiCamera()
camera.resolution = (640,360)
camera.framerate = 30
camera.brightness=60
camera.shutter_speed = 10000
camera.exposure_mode = 'night'
camera.iso=800
time.sleep(5)
g = camera.awb_gains
camera.awb_mode='off'
camera.awb_gains = g
rawCapture = PiRGBArray(camera,size=(640,360))
return camera,rawCapture
def start_Beep():
for i in range(0,5):
GPIO.output(11,GPIO.LOW)
time.sleep(0.1)
GPIO.output(11,GPIO.HIGH)
time.sleep(0.1)
def get_angle(imageCap):
result=np.zeros((1,90,160,3),dtype=np.uint8)
image=np.array(imageCap)
result[0]=image
return int(model.predict(result)[0])
def normalize_angle(angle):
return (int(angle/5.0+0.5))*5
if __name__ == '__main__':
try:
model_path='ml_model_retrain.hdf5'
#model_path='ml_model_pbs.hdf5'
model = load_model(model_path)
#start_Beep()
camera,raw = set_picamera()
for frame in camera.capture_continuous(raw,format = "bgr",use_video_port=True):
img = frame.array
imageCap = cv2.resize(img,(160,90))
angle=get_angle(imageCap)
#angle=normalize_angle(get_angle(imageCap))
ser.write("1,13,"+str(angle))
except KeyboardInterrupt:
if ser != None:
str1 = "1,0,90"
n = ser.write(str1)
time.sleep(0.2)
print ("\nfinish")
ser.close()
未附上图像采集等其他程序
最终效果：
https://pan.baidu.com/s/1dZfu7dOQvmXW-ESvyUX0tQ
d1420180907_165649.gif
