【OpenAI】Python：基于 Gym-CarRacing 的自动驾驶项目（1） | 前置知识介绍 | 项目环境准备 | 手把手带你一步步实现

> conda env create –n autodriving
> conda activate autodirving
> conda install python
> pip install pytorch

conda

conda create -n python=2.7

conda activate [文件名]

gym.make()         环境配置。
env.reset()        设置初始变量。
env.step(action)   执行动作并返回以下观察值、补偿、是否结束。

cd {{installed gym path}}

pip install -e .[box2d]
pip install matplotlib
pip install scipy
pip install pyglet
pip install pygame

python -m pip install Box2D

cd gym/envs/box2d

import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import find_peaks
from scipy.interpolate import splprep, splev
from scipy.optimize import minimize
import time
class LaneDetection:
    '''
    Lane detection module using edge detection and b-spline fitting
    args: 
        cut_size (cut_size=65) cut the image at the front of the car
        spline_smoothness (default=10)
        gradient_threshold (default=14)
        distance_maxima_gradient (default=3)
    '''
    def __init__(self, cut_size=65, spline_smoothness=10, gradient_threshold=14, distance_maxima_gradient=3):
        self.car_position = np.array([48,0])
        self.spline_smoothness = spline_smoothness
        self.cut_size = cut_size
        self.gradient_threshold = gradient_threshold
        self.distance_maxima_gradient = distance_maxima_gradient
        self.lane_boundary1_old = 0
        self.lane_boundary2_old = 0
    def cut_gray(self, state_image_full):
        '''
        ##### TODO #####
        This function should cut the image at the front end of the car (e.g. pixel row 65) 
        and translate to gray scale
        input:
            state_image_full 96x96x3
        output:
            gray_state_image 65x96x1
        '''
        return gray_state_image[::-1] 
    def edge_detection(self, gray_image):
        '''
        ##### TODO #####
        In order to find edges in the gray state image, 
        this function should derive the absolute gradients of the gray state image.
        Derive the absolute gradients using numpy for each pixel. 
        To ignore small gradients, set all gradients below a threshold (self.gradient_threshold) to zero. 
        input:
            gray_state_image 65x96x1
        output:
            gradient_sum 65x96x1
        '''
        return gradient_sum
    def find_maxima_gradient_rowwise(self, gradient_sum):
        '''
        ##### TODO #####
        This function should output arguments of local maxima for each row of the gradient image.
        You can use scipy.signal.find_peaks to detect maxima. 
        Hint: Use distance argument for a better robustness.
        input:
            gradient_sum 65x96x1
        output:
            maxima (np.array) shape : (Number_maxima, 2)
        '''
        return argmaxima
    def find_first_lane_point(self, gradient_sum):
        '''
        Find the first lane_boundaries points above the car.
        Special cases like just detecting one lane_boundary or more than two are considered. 
        Even though there is space for improvement ;) 
        input:
            gradient_sum 65x96x1
        output: 
            lane_boundary1_startpoint
            lane_boundary2_startpoint
            lanes_found  true if lane_boundaries were found
        '''
        # Variable if lanes were found or not
        lanes_found = False
        row = 0
        # loop through the rows
        while not lanes_found:
            # Find peaks with min distance of at least 3 pixel 
            argmaxima = find_peaks(gradient_sum[row],distance=3)[0]
            # if one lane_boundary is found
            if argmaxima.shape[0] == 1:
                lane_boundary1_startpoint = np.array([[argmaxima[0],  row]])
                if argmaxima[0] < 48:
                    lane_boundary2_startpoint = np.array([[0,  row]])
                else: 
                    lane_boundary2_startpoint = np.array([[96,  row]])
                lanes_found = True
            # if 2 lane_boundaries are found
            elif argmaxima.shape[0] == 2:
                lane_boundary1_startpoint = np.array([[argmaxima[0],  row]])
                lane_boundary2_startpoint = np.array([[argmaxima[1],  row]])
                lanes_found = True
            # if more than 2 lane_boundaries are found
            elif argmaxima.shape[0] > 2:
                # if more than two maxima then take the two lanes next to the car, regarding least square
                A = np.argsort((argmaxima - self.car_position[0])**2)
                lane_boundary1_startpoint = np.array([[argmaxima[A[0]],  0]])
                lane_boundary2_startpoint = np.array([[argmaxima[A[1]],  0]])
                lanes_found = True
            row += 1
            # if no lane_boundaries are found
            if row == self.cut_size:
                lane_boundary1_startpoint = np.array([[0,  0]])
                lane_boundary2_startpoint = np.array([[0,  0]])
                break
        return lane_boundary1_startpoint, lane_boundary2_startpoint, lanes_found
    def lane_detection(self, state_image_full):
        '''
        ##### TODO #####
        This function should perform the road detection 
        args:
            state_image_full [96, 96, 3]
        out:
            lane_boundary1 spline
            lane_boundary2 spline
        '''
        # to gray
        gray_state = self.cut_gray(state_image_full)
        # edge detection via gradient sum and thresholding
        gradient_sum = self.edge_detection(gray_state)
        maxima = self.find_maxima_gradient_rowwise(gradient_sum)
        # first lane_boundary points
        lane_boundary1_points, lane_boundary2_points, lane_found = self.find_first_lane_point(gradient_sum)
        # if no lane was found,use lane_boundaries of the preceding step
        if lane_found:
            ##### TODO #####
            #  in every iteration: 
            # 1- find maximum/edge with the lowest distance to the last lane boundary point 
            # 2- append maximum to lane_boundary1_points or lane_boundary2_points
            # 3- delete maximum from maxima
            # 4- stop loop if there is no maximum left 
            #    or if the distance to the next one is too big (>=100)
            # lane_boundary 1
            # lane_boundary 2
            ################
            ##### TODO #####
            # spline fitting using scipy.interpolate.splprep 
            # and the arguments self.spline_smoothness
            # 
            # if there are more lane_boundary points points than spline parameters 
            # else use perceding spline
            if lane_boundary1_points.shape[0] > 4 and lane_boundary2_points.shape[0] > 4:
                # Pay attention: the first lane_boundary point might occur twice
                # lane_boundary 1
                # lane_boundary 2
            else:
                lane_boundary1 = self.lane_boundary1_old
                lane_boundary2 = self.lane_boundary2_old
            ################
        else:
            lane_boundary1 = self.lane_boundary1_old
            lane_boundary2 = self.lane_boundary2_old
        self.lane_boundary1_old = lane_boundary1
        self.lane_boundary2_old = lane_boundary2
        # output the spline
        return lane_boundary1, lane_boundary2
    def plot_state_lane(self, state_image_full, steps, fig, waypoints=[]):
        '''
        Plot lanes and way points
        '''
        # evaluate spline for 6 different spline parameters.
        t = np.linspace(0, 1, 6)
        lane_boundary1_points_points = np.array(splev(t, self.lane_boundary1_old))
        lane_boundary2_points_points = np.array(splev(t, self.lane_boundary2_old))
        plt.gcf().clear()
        plt.imshow(state_image_full[::-1])
        plt.plot(lane_boundary1_points_points[0], lane_boundary1_points_points[1]+96-self.cut_size, linewidth=5, color='orange')
        plt.plot(lane_boundary2_points_points[0], lane_boundary2_points_points[1]+96-self.cut_size, linewidth=5, color='orange')
        if len(waypoints):
            plt.scatter(waypoints[0], waypoints[1]+96-self.cut_size, color='white')
        plt.axis('off')
        plt.xlim((-0.5,95.5))
        plt.ylim((-0.5,95.5))
        plt.gca().axes.get_xaxis().set_visible(False)
        plt.gca().axes.get_yaxis().set_visible(False)
        fig.canvas.flush_events()

import gym
from gym.envs.box2d.car_racing import CarRacing
import pygame
from lane_detection import LaneDetection
import matplotlib.pyplot as plt
import numpy as np
import pyglet
from pyglet import gl
from pyglet.window import key
# action variables
action = np.array([0.0, 0.0, 0.0])
def register_input():
    for event in pygame.event.get():
        if event.type == pygame.KEYDOWN:
            if event.key == pygame.K_LEFT:
                action[0] = -1.0
            if event.key == pygame.K_RIGHT:
                action[0] = +1.0
            if event.key == pygame.K_UP:
                action[1] = +0.5
            if event.key == pygame.K_DOWN:
                action[2] = +0.8  # set 1.0 for wheels to block to zero rotation
            if event.key == pygame.K_r:
                global retry
                retry = True
            if event.key == pygame.K_s:
                global record
                record = True
            if event.key == pygame.K_q:
                global quit
                quit = True
        if event.type == pygame.KEYUP:
            if event.key == pygame.K_LEFT and action[0] < 0.0:
                action[0] = 0
            if event.key == pygame.K_RIGHT and action[0] > 0.0:
                action[0] = 0
            if event.key == pygame.K_UP:
                action[1] = 0
            if event.key == pygame.K_DOWN:
                action[2] = 0
# init environement
env = CarRacing()
env.render()
env.reset()
# define variables
total_reward = 0.0
steps = 0
retry = False
quit = False
# init modules of the pipeline
LD_module = LaneDetection()
# init extra plot
fig = plt.figure()
plt.ion()
plt.show()
while not quit:
    env.reset()
    retry = False
    while True:
        # perform step
        register_input()
        s, r, done, speed= env.step(action)
        # lane detection
        splines = LD_module.lane_detection(s)
        # reward
        total_reward += r
        # outputs during training
        if steps % 2 == 0 or done:
            print("\naction " + str(["{:+0.2f}".format(x) for x in action]))
            print("step {} total_reward {:+0.2f}".format(steps, total_reward))
            LD_module.plot_state_lane(s, steps, fig)
        steps += 1
        env.render()
        if done or retry or quit: break
env.close()

📌 [ 笔者 ]   王亦优
📃 [ 更新 ]   2022.12.29
❌ [ 勘误 ]   /* 暂无 */
📜 [ 声明 ]   由于作者水平有限，本文有错误和不准确之处在所难免，
              本人也很想知道这些错误，恳望读者批评指正！