Reinforcement Learning Environment for CARLA Autonomous Self-Driving Car Using Python

In the last tutorial, we introduced the concept of reinforcement learning and gave an example of how it is used in an autonomous self-driving car.

In this tutorial, we are going to take our knowledge of reinforcement learning and CARLA to add some artificial intelligence to our autonomous self-driving car.

We will begin by setting up the environment in which the reinforcement agent will carry out its operations.

For those who are new to reinforcement learning, there are different ways to go about coding a reinforcement learning environment; however, there is also a kind of standardized way of going about it, which was pioneered by Open AI.

Reinforcement Learning Environment with CARLA

We will create a method called step(), which processes an action and returns an observation, a reward, a boolean flag that indicates when it's done, and any extra info we might need to know, and a reset method that will restart the environment when we want to end or restart the program.

Next, we will create a reset() method that will restart the environment when we want to end or restart the program.

Create a file named autonomous-car-tutorial2.py. We will copy some of the codes from autonomous-car-tutorial.py into this new file.

import glob
import os
import sys
import random
import time
import numpy as np
import cv2

try:
    sys.path.append(glob.glob('../carla/dist/carla-*%d.%d-%s.egg' % (
        sys.version_info.major,
        sys.version_info.minor,
        'win-amd64' if os.name == 'nt' else 'linux-x86_64'))[0])
except IndexError:
    pass
import carla

We've already gone through how the code above operates in the previous tutorial. We will basically just add new codes to it.

We will add the SHOW_PREVIEW constant below to indicate whether we want to show the camera from CARLA or not. As we begin, the value will be set to false to avoid maxing out computing resources. The IMG_WIDTH and IMG_HEIGHT constants are pretty much descriptive.

SHOW_PREVIEW = False
IMG_WIDTH = 640
IMG_HEIGHT = 480

Next, we will create a class for the environment called CarEnvironment.

class CarEnvironment:

We will add some constants to this class to define some default functionalities.

    SHOW_CAM = SHOW_PREVIEW
    STEER_AMT = 1.0
    img_width = IMG_WIDTH
    img_height = IMG_HEIGHT
    front_camera = None

Most of the codes above are self-explanatory if you have been following the tutorial series from the beginning. The STEER_AMT constant indicates how the car turns. It can be set to -1.0, 0, or 1.0 for full left turn, straight or full right turn. You can add more functionalities when you understand the basics of how it works.

Next, we will define an __init__() method in the class. If you are familiar with python, the __init__() method wouldn't be new to you.

To refresh your memory, __init__ is a reserved method in python classes. It is known as a constructor in object-oriented concepts. This method is called when an object is created from the class and allows the class to initialize the attributes of that class.

    def __init__(self):
        self.client = carla.Client('localhost', 2000)
        self.client.set_timeout(2.0)
        self.world = self.client.get_world()
        self.blueprint_library = self.world.get_blueprint_library()
        self.model_3 = self.blueprint_library.filter('model3')[0]

In the code above, firstly, we set a client. When we have a client, we can retrieve the world that is currently running.

The world contains the list of blueprints that we can use for adding new actors into the simulation. Next, we filtered the Tesla Model 3 car from the blueprint library of vehicles.

If you desire it, you can also filter through all the blueprints of type 'vehicle' and choose one at random using the code blueprint_library.filter('vehicle').

Before we continue, below is the full code for code autonomous-car-tutorial2.py up to this point.

import glob
import os
import sys
import random
import time
import numpy as np
import cv2

try:
    sys.path.append(glob.glob('../carla/dist/carla-*%d.%d-%s.egg' % (
        sys.version_info.major,
        sys.version_info.minor,
        'win-amd64' if os.name == 'nt' else 'linux-x86_64'))[0])
except IndexError:
    pass
import carla

SHOW_PREVIEW = False
IMG_WIDTH = 640
IMG_HEIGHT = 480

class CarEnvironment:
    SHOW_CAM = SHOW_PREVIEW
    STEER_AMT = 1.0
    img_width = IMG_WIDTH
    img_height = IMG_HEIGHT
    front_camera = None

    def __init__(self):
        self.client = carla.Client('localhost', 2000)
        self.client.set_timeout(2.0)
        self.world = self.client.get_world()
        self.blueprint_library = self.world.get_blueprint_library()
        self.tesla_model3 = self.blueprint_library.filter('model3')[0]

Creating the reset method for the Autonomous Self-Driving Car Environment

Firstly, we will work on the reset method for the reinforcement learning environment. This method handles the collision, vehicle spawn, camera span zones, camera sensor, and collision, and then returns the observation.

    def reset(self):
        self.collision_hist = []
        self.actor_list = []

        #Vehicle Spawn and and location set up
        self.transform = random.choice(self.world.get_map().get_spawn_points())
        self.vehicle = self.world.spawn_actor(self.tesla_model3, self.transform)
        self.actor_list.append(self.vehicle)

        #RGB Camera set up
        self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
        self.rgb_cam.set_attribute('image_size_x', f'{self.img_width}')
        self.rgb_cam.set_attribute('image_size_y', f'{self.img_height}')
        self.rgb_cam.set_attribute('fov', '110')

        #Vehicle sensor set up
        transform = carla.Transform(carla.Location(x=2.5, z=0.7))
        self.sensor = self.world.spawn_actor(self.rgb_cam, transform, attach_to=self.vehicle)
        self.actor_list.append(self.sensor)
        self.sensor.listen(lambda data: self.process_img(data))
        self.vehicle.apply_control(carla.VehicleControl(throttle=0.0, brake=0.0))

def reset(self): method takes the self as a parameter. In essence, self represents an instance of the CarEnvironment class. By using the self keyword as a parameter we can access the attributes and methods of the CarEnvironment class in python.

If you've been following this tutorial series from the beginning, you will be familiar with most of the codes above, we simply just converted it to object-oriented programming format. However, let's go through some important things.

The process_img method is the same as the one in the previous tutorials. We will simply copy it into this class and make some minor modifications to fit in the class.

def process_img(self, image):
        i = np.array(image.raw_data)
        #np.save("iout.npy", i)
        i2 = i.reshape((self.img_height, self.img_width, 4))
        i3 = i2[:, :, :3]
        if self.SHOW_CAM:
            cv2.imshow("",i3)
            cv2.waitKey(1)
        self.front_camera = i3

Next, we will add some more codes to the def reset(self): method that will get the collision sensor working.

        #Collision sensor set up
        time.sleep(4)
        collision_sensor = self.world.get_blueprint_library().find('sensor.other.collision')
        self.collision_sensor = self.world.spawn_actor(collision_sensor, transform, attach_to=self.vehicle)
        self.actor_list.append(self.collision_sensor)
        self.collision_sensor.listen(lambda event: self.collision_data(event))

In the code block above, we simply set a time (4 seconds) for the car to rest before it begins moving. This is because the car drops into the CARLA spawn point like in Grand Theft game, and when it hits the ground, a collision is usually registered.

Also, when that happens, it can take some time for these sensors to initialize and return values.

Then, using the CARLA blueprint library function, we got a blueprint for a collision sensor, attached it to the car, and added it to the list of actors.

The last line basically instructs the sensor to listen for collision events after the initial 4 seconds of sleep time by calling a collision_data() method.

The collision_data() method does not exist yet because we haven't created it.

Finally, for the reset() method, let's log the actual starting time for the training event (each training event will be called an episode), make sure brake and throttle aren't being used, and return our first value, which is the camera.

    self.episode_start = time.time()
    self.vehicle.apply_control(carla.VehicleControl(brake=0.0, throttle=0.0))
    return self.front_camera

Below is the full code for the reset() method.

    def reset(self):
        self.collision_hist = []
        self.actor_list = []

        #Vehicle Spawn and and location set up
        self.transform = random.choice(self.world.get_map().get_spawn_points())
        self.vehicle = self.world.spawn_actor(self.tesla_model3, self.transform)
        self.actor_list.append(self.vehicle)

        #RGB Camera set up
        self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
        self.rgb_cam.set_attribute('image_size_x', f'{self.img_width}')
        self.rgb_cam.set_attribute('image_size_y', f'{self.img_height}')
        self.rgb_cam.set_attribute('fov', '110')

        #Vehicle sensor set up
        transform = carla.Transform(carla.Location(x=2.5, z=0.7))
        self.sensor = self.world.spawn_actor(self.rgb_cam, transform, attach_to=self.vehicle)
        self.actor_list.append(self.sensor)
        self.sensor.listen(lambda data: self.process_img(data))
        self.vehicle.apply_control(carla.VehicleControl(throttle=0.0, brake=0.0))

        #Collision sensor set up
        time.sleep(4)
        collision_sensor = self.world.get_blueprint_library().find('sensor.other.collision')
        self.collision_sensor = self.world.spawn_actor(collision_sensor, transform, attach_to=self.vehicle)
        self.actor_list.append(self.collision_sensor)
        self.collision_sensor.listen(lambda event: self.collision_data(event))
     
        #episodes set up
        self.episode_start = time.time()
        self.vehicle.apply_control(carla.VehicleControl(brake=0.0, throttle=0.0))

        #The value returned
        return self.front_camera

Now, let's create the collision_data method and add it to the class.

Python Code

def collision_data(self, event):
    self.collision_hist.append(event)

Creating the step method for the Autonomous Self-driving Car Environment

Now, we will work on the step method for the reinforcement learning environment. This method takes a specific action and then returns the observation, reward, completion indicator, and any extra information that may be required based on the standard reinforcement learning paradigm.

    def step(self, action):
        if action == 0:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=-1*self.STEER_AMT))
        elif action == 1:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer= 0))
        elif action == 2:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=1*self.STEER_AMT))

The code block above shows how we sort out an action based on the numerical value passed using multiple if statements.

Next, we will need to handle the observation, possible collision, and reward:

At this point, we will be using the math package in python. To do that, add the code below at the top of the file where the other package imports are located.

    import math

Next, We complete the code for the step method.

        v = self.vehicle.get_velocity()
        kmh = int(3.6 * math.sqrt(v.x**2 + v.y**2 + v.z**2))

        if len(self.collision_hist) != 0:
            done = True
            reward = -200
        elif kmh < 50:
            done = False
            reward = -1
        else:
            done = False
            reward = 1

        if self.episode_start + SECONDS_PER_EPISODE < time.time():
            done = True

        return self.front_camera, reward, done, None

In the code above, we get the vehicle's speed, convert it from velocity to Kilometer per hour(KMH). This is done to prevent the reinforcement learning agent from driving in a tight circle. If we require a certain speed to get a reward, this should hopefully discourage it.

Next, we check to see if we've exhausted our episode time, then we return all the data.

Also, you will notice we introduced the constant SECONDS_PER_EPISODE. You need to add the code at the top where we placed other constants.

        SECONDS_PER_EPISODE = 15

Full code for the Reinforcement Learning Environment of our Autonomous Self-Driving Car Using CARLA

import glob
import os
import sys
import random
import time
import numpy as np
import cv2
import math

try:
    sys.path.append(glob.glob('../carla/dist/carla-*%d.%d-%s.egg' % (
        sys.version_info.major,
        sys.version_info.minor,
        'win-amd64' if os.name == 'nt' else 'linux-x86_64'))[0])
except IndexError:
    pass
import carla

SHOW_PREVIEW = False
IMG_WIDTH = 640
IMG_HEIGHT = 480
SECONDS_PER_EPISODE = 15

class CarEnvironment:
    SHOW_CAM = SHOW_PREVIEW
    STEER_AMT = 1.0
    img_width = IMG_WIDTH
    img_height = IMG_HEIGHT
    front_camera = None

    def __init__(self):
        self.client = carla.Client('localhost', 2000)
        self.client.set_timeout(2.0)
        self.world = self.client.get_world()
        self.blueprint_library = self.world.get_blueprint_library()
        self.tesla_model3 = self.blueprint_library.filter('model3')[0]

    def reset(self):
        self.collision_hist = []
        self.actor_list = []

        #Vehicle Spawn and and location set up
        self.transform = random.choice(self.world.get_map().get_spawn_points())
        self.vehicle = self.world.spawn_actor(self.tesla_model3, self.transform)
        self.actor_list.append(self.vehicle)

        #RGB Camera set up
        self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
        self.rgb_cam.set_attribute('image_size_x', f'{self.img_width}')
        self.rgb_cam.set_attribute('image_size_y', f'{self.img_height}')
        self.rgb_cam.set_attribute('fov', '110')

        #Vehicle sensor set up
        transform = carla.Transform(carla.Location(x=2.5, z=0.7))
        self.sensor = self.world.spawn_actor(self.rgb_cam, transform, attach_to=self.vehicle)
        self.actor_list.append(self.sensor)
        self.sensor.listen(lambda data: self.process_img(data))
        self.vehicle.apply_control(carla.VehicleControl(throttle=0.0, brake=0.0))

        #Collision sensor set up
        time.sleep(4)
        collision_sensor = self.world.get_blueprint_library().find('sensor.other.collision')
        self.collision_sensor = self.world.spawn_actor(collision_sensor, transform, attach_to=self.vehicle)
        self.actor_list.append(self.collision_sensor)
        self.collision_sensor.listen(lambda event: self.collision_data(event))
     
        #episodes set up
        self.episode_start = time.time()
        self.vehicle.apply_control(carla.VehicleControl(brake=0.0, throttle=0.0))

        #The value returned
        return self.front_camera

    def process_img(self, image):
        i = np.array(image.raw_data)
        #np.save("iout.npy", i)
        i2 = i.reshape((self.img_height, self.img_width, 4))
        i3 = i2[:, :, :3]
        if self.SHOW_CAM:
            cv2.imshow("",i3)
            cv2.waitKey(1)
        self.front_camera = i3

    def collision_data(self, event):
        self.collision_hist.append(event)

    def step(self, action):
        if action == 0:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=-1*self.STEER_AMT))
        elif action == 1:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer= 0))
        elif action == 2:
            self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=1*self.STEER_AMT))

        v = self.vehicle.get_velocity()
        kmh = int(3.6 * math.sqrt(v.x**2 + v.y**2 + v.z**2))

        if len(self.collision_hist) != 0:
            done = True
            reward = -200
        elif kmh < 50:
            done = False
            reward = -1
        else:
            done = False
            reward = 1

        if self.episode_start + SECONDS_PER_EPISODE < time.time():
            done = True

        return self.front_camera, reward, done, None

With that, we're done with creating our reinforcement learning environment!

Wrap Off

In the next tutorial, we will work on the reinforcement learning agent of our autonomous self-driving car.

If you run into errors or unable to complete this tutorial, feel free to contact us anytime, and we will instantly resolve it. You can also request clarification, download this tutorial as a pdf, or report bugs using the buttons below.