car_model.py

# -*- coding: utf-8 -*-
"""CAR-MODEL.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1ie2fR9mQTDn-yMa2rD2aDmnki39yZfPN
"""

! git clone https://github.com/ssiwach8888/SD_Cars_Training_Data

! ls DATA

# ! pip3 install imgaug

import numpy as np
import ntpath
import keras
from keras.layers import Dense, Dropout, Flatten, MaxPooling2D, Convolution2D
from keras.models import Sequential
import matplotlib.pyplot as plt
from keras.optimizers import Adam
import os
import pandas as pd
import random
import cv2
from sklearn.utils import shuffle
from sklearn.model_selection import train_test_split
import matplotlib.image as mpimg
from imgaug import augmenters as iaa

dir = 'DATA'

list = ['center', 'left', 'right', 'steering', 'throttle', 'reverse', 'speed']
data = pd.read_csv(os.path.join(dir, 'driving_log.csv'),names=list)
pd.set_option('display.max_colwidth', -1)
data.head()

def remove_path(path):
  head, tail = ntpath.split(path)
  return tail

data['center'] = data['center'].apply(remove_path)
data['left'] = data['left'].apply(remove_path)
data['right'] = data['right'].apply(remove_path)

data.head()
print(data['steering'])

n_bins = 25
samples = 500
hist, bins = np.histogram(data['steering'], n_bins)
# x and y
# Now Process our bins to plot a bar graph
print('Bins',bins)
print('hist',hist)
center = (bins[:-1] + bins[1:]) * 0.5
plt.bar(center, hist, width=0.05)
plt.plot((np.min(data['steering']), np.max(data['steering'])), (samples, samples))

list = []
print('Total Data ', len(data))
for j in range(n_bins):
  list_ = []
  for i in range(len(data['steering'])):
    if data['steering'][i]>=bins[j] and data['steering'][i]<=bins[j+1]:
      # it is use to check if the data we iterate is belong
      # to curent bin or not
      # if angle fall b/w two bins then it belong to the same bin
      # our Bins var store the range of bin
      list_.append(i)
  list_ = shuffle(list_)
  list_ = list_[samples:]
  list.extend(list_)

print('processed data ',len(list))

data.drop(data.index[list], inplace=True)
print('Remaining ',len(data))

# now plot our Graph again
hist, bins = np.histogram(data['steering'], n_bins)
plt.bar(center, hist, width=0.05)
plt.plot((np.min(data['steering']), np.max(data['steering'])), (samples, samples))

# From out Data_frame extract center images which is capture
# by our car's Center camera and alse extract there steer angle

print(data.iloc[1])

def load_data(path_var, data_frame):
  img_path = []
  steer_angle = []
  for i in range(len(data_frame)):
    index = data_frame.iloc[i]
    center = index[0]
    left = index[1]
    right = index[2]
    angle = index[3]
    '''img_path.extend([os.path.join(path_var, center.strip()),
                     os.path.join(path_var, left.strip()),
                     os.path.join(path_var, right.strip())])
    steer_angle.extend([float(index[3]),
                        float(index[3]),
                              float(index[3])])'''
    img_path.append(os.path.join(path_var, center.strip()))
    steer_angle.append(float(angle))
    img_path.append(os.path.join(path_var, left.strip()))
    #steer_angle.append(float(angle)+0.15)
    steer_angle.append(float(angle)+0.2)
    img_path.append(os.path.join(path_var,right.strip()))
    steer_angle.append(float(angle)-0.2)
    #steer_angle.append(float(angle)-0.15)

  return np.asarray(img_path), np.asarray(steer_angle)

img_path, angle = load_data(dir + '/IMG/IMG', data)
print(img_path.shape)
print(angle.shape)

print(img_path)
print(angle)

# split our Data into Training and validation set

train_data, val_data, train_label, val_label = train_test_split(
    img_path, angle, test_size=0.2, random_state=0
)
print(train_data.shape)
print(train_label.shape)
print(val_data.shape)
print(val_label.shape)

#print(train_data)
#print(train_label)

# plot bar graph of both data_sets to test if data split
# uniformly or not

fig, axs = plt.subplots(1, 2, figsize=(12, 4))
axs[0].hist(train_label, bins=n_bins, width=0.05, color='red')
axs[0].set_title('Training')
axs[1].hist(val_label, bins=n_bins, width=0.05, color='blue')

# Data Augmentation Technique
# it use to generate variety of data_sets
# and help our model to generalize new Data_sets better

# for this we use imgaug library

# 1st is Zoom
def zoom(img):
  zoom = iaa.Affine(scale=(1, 1.3))
  return zoom.augment_image(img)

# 2nd is panning _> horizontal and vertical translation of image
def panning(img):
  pan = iaa.Affine(translate_percent={
      'x':(-0.1, 0.1), 'y':(-0.1, 0.1)
  })
  return pan.augment_image(img)

# 3rd is Altering Brightness
def alter_bright(img):
  bright = iaa.Multiply((0.2, 1.2))
  return bright.augment_image(img)

# 4th is flipping: it also need steering angle becoz by flipping
# image it also change its angle which we use as label
# and IT is most IMPORTANT technique to Make our data balance
def flip(img, steer_angle):
  img = cv2.flip(img, 1)
  # 0->verticle, 1->horizontal, and -1 ->combine
  steer_angle = -steer_angle
  return img, steer_angle

# NOW the most important thing is we don't apply
# all techniques on a image it reduce the variety of our images
# so we use a simple if else logic to 50% apply every 
# DATA augumentation

def random_aug(img, steer_angle):
  img = mpimg.imread(img)
  if np.random.rand()<0.5:
    img = zoom(img)
  if np.random.rand()<0.5:
    img = panning(img)
  if np.random.rand()<0.5:
    img = alter_bright(img)
  if np.random.rand()<0.5:
    img, steer_angle = flip(img, steer_angle)
  return img, steer_angle

# image preProcessing
# Display a random image using matplotlib image 

temp = train_data[random.randint(0, len(train_data)-1)]
temp = mpimg.imread(temp)
temp = plt.imshow(temp)
plt.show(temp)

# by display our image we decide which type of preprocssing we 
# have to done on our image

def preprocess(img):
  # crop
  # img[height, width, channel]
  img = img[60:135,:,:]
  # change colorspace to (y,u,v)
  img = cv2.cvtColor(img, cv2.COLOR_BGR2YUV)
  # Gaussian Blurr to remove noise
  img = cv2.GaussianBlur(img, (3,3),0)
  # Resize our image according to our NN Model Input Node
  img = cv2.resize(img, (200, 66))
  # Normalize
  img = img/255
  return img

# now Make our Batch Generator
# it is a yield method which preserve its state
'''def batch_generator(img, angle, batch_size, is_training):
  while True:
    batch_img = []
    batch_angel = []
    for i in range(batch_size):
      index = random.randint(0, len(img)-1)
      temp = img[index]
      temp_angle = angle[index]
      if is_training:
        temp, temp_angle = random_aug(temp, temp_angle)
      else:
        temp = mpimg.imread(temp)
      img = preprocess(temp)
      batch_img.append(img)
      batch_angel.append(temp_angle)
    yield (np.asarray(batch_img), np.asarray(batch_angel))'''

def batch_generator(image_paths, steering_ang, batch_size, istraining):
  
  while True:
    batch_img = []
    batch_steering = []
    
    for i in range(batch_size):
      random_index = random.randint(0, len(image_paths) - 1)
      
      if istraining:
        im, steering = random_aug(image_paths[random_index], steering_ang[random_index])
     
      else:
        im = mpimg.imread(image_paths[random_index])
        steering = steering_ang[random_index]
      
      im = preprocess(im)
      batch_img.append(im)
      batch_steering.append(steering)
    yield (np.asarray(batch_img), np.asarray(batch_steering))

# Now test our fit generator Function
# use next() method it call a iterator and and retrive next() item for it

batch_test , batch_test_label = next(batch_generator(train_data, train_label, 1, 1))

plt.imshow(batch_test[0])
print(batch_test_label[0])

batch_test , batch_test_label = next(batch_generator(train_data, train_label, 1, 0))
plt.imshow(batch_test[0])
print(batch_test_label[0])

# check our Preprocessed image

temp = train_data[random.randint(0, len(train_data)-1)]
temp = mpimg.imread(temp)
temp = preprocess(temp)
temp = plt.imshow(temp)
plt.show(temp)

# Nvidia NN Model:
def nvidia_model():
  model = Sequential()
  model.add(Convolution2D(24, (5, 5), strides=(2, 2), input_shape=(66, 200, 3), activation='elu'))
  model.add(Convolution2D(36, (5, 5), strides=(2, 2), activation='elu'))
  model.add(Convolution2D(48, (5, 5), strides=(2, 2), activation='elu'))
  model.add(Convolution2D(64, (3, 3), activation='elu'))
  
  model.add(Convolution2D(64, (3, 3), activation='elu'))
  #model.add(Dropout(0.5))
  
  
  model.add(Flatten())
  
  model.add(Dense(100, activation = 'elu'))
#   model.add(Dropout(0.5))
  
  model.add(Dense(50, activation = 'elu'))
#   model.add(Dropout(0.5))
  
  model.add(Dense(10, activation = 'elu'))
#   model.add(Dropout(0.5))

  model.add(Dense(1))
  
  optimizer = Adam(lr=1e-4)
  model.compile(loss='mse', optimizer=optimizer)
  return model

model_obj = nvidia_model()
print(model_obj.summary())

history = model_obj.fit(batch_generator(train_data, train_label, 100, 1),
                        steps_per_epoch=300, 
                        epochs=10,
                        validation_data=batch_generator(val_data, val_label, 100, 0),
                        validation_steps=200,
                        verbose=1,
                        shuffle = 1)

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.legend(['training' , 'validation'])
plt.title('Loss')
plt.xlabel('epochs')

model_obj.save('model.h5')
from google.colab import files
files.download('model.h5')