基于carla和python的自动驾驶仿真系列3

760 阅读 0 评论 502 点赞

我是靠谱客的博主快乐菠萝，最近开发中收集的这篇文章主要介绍基于carla和python的自动驾驶仿真系列3，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

欢迎来到卡拉自动驾驶汽车的Python编程教程的第3部分。在本教程中，我们将利用Carla API的知识，尝试将这个问题转化为一个强化学习问题。
在OpenAI开创了强化学习环境和解决方案的开放源码之后，我们就有了一种接近强化学习环境的标准化方法。
这里的想法是，您的环境将有一个step方法，它返回:observation, reward, done, extra_info，以及一个reset方法，它将基于某种done标志重新启动环境。
我们所需要做的就是创建代码来表示这个。我们将从卡拉经常进口的东西开始:

import glob
import os
import sys
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

接下来，我们将创建环境的类，我们将其命名为CarEnv。在模型、代理和环境的脚本顶部添加一些常量会很方便，所以我要做的第一件事就是创建我们的第一个常量，即SHOW PREVIEW：

SHOW_PREVIEW = False

现在让我们为环境类设置一些初始值:

class CarEnv:
SHOW_CAM = SHOW_PREVIEW
STEER_AMT = 1.0
im_width = IMG_WIDTH
im_height = IMG_HEIGHT
actor_list = []
front_camera = None
collision_hist = []

我认为这些都是不言自明的，但是SHOW_CAM是我们是否要显示预览的关键。为了调试的目的，查看一个可能很有用，但您不一定想要一直显示一个，因为执行所有这些可能会耗费大量资源。
STEER_AMT是我们想要应用在转向上的。目前，这是一个全面的转变。后来我们可能会发现，最好是控制的力度小一点，也许做一些累积的事情……等。现在，全力驾驶!
将使用collision_hist，因为碰撞传感器报告事故历史。基本上，如果这个列表中有任何东西，我们会说我们碰撞了。对于init方法。你在之前的教程中看到的所有东西:

def __init__(self):
self.client = carla.Client('localhost', 2000)
self.client.set_timeout(2.0)
# Once we have a client we can retrieve the world that is currently
# running.
self.world = self.client.get_world()
# The world contains the list blueprints that we can use for adding new
# actors into the simulation.
blueprint_library = self.world.get_blueprint_library()
# Now let's filter all the blueprints of type 'vehicle' and choose one
# at random.
#print(blueprint_library.filter('vehicle'))
self.model_3 = blueprint_library.filter('model3')[0]

接下来，我们将创建reset方法。


def reset(self):
self.collision_hist = []
self.actor_list = []
self.transform = random.choice(self.world.get_map().get_spawn_points())
self.vehicle = self.world.spawn_actor(self.model_3, self.transform)
self.actor_list.append(self.vehicle)
self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
self.rgb_cam.set_attribute('image_size_x', f'{self.im_width}')
self.rgb_cam.set_attribute('image_size_y', f'{self.im_height}')
self.rgb_cam.set_attribute('fov', '110')
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)) # initially passing some commands seems to help with time. Not sure why.

这里你看到的所有东西，没有什么新的或新奇的，只是变成OOP。接下来，我们将碰撞传感器添加到这个方法中:


time.sleep(4) # sleep to get things started and to not detect a collision when the car spawns/falls from sky.
colsensor = self.world.get_blueprint_library().find('sensor.other.collision')
self.colsensor = self.world.spawn_actor(colsensor, transform, attach_to=self.vehicle)
self.actor_list.append(self.colsensor)
self.colsensor.listen(lambda event: self.collision_data(event))

我们在这里开始了四秒钟的睡眠，因为汽车真的“掉入”了模拟器。通常情况下，当汽车撞到地面时，我们会得到一个碰撞记录。而且，最初，它会花一些时间让这些传感器初始化和返回值，所以我们会用一个安全可靠的4秒。为了防止需要更长的时间，我们可以做以下事情:


while self.front_camera is None:
time.sleep(0.01)

但是我们不能这么做，因为我们需要确定赛车在衍生时已经从天上掉下来了。最后，让我们记录下这一集的实际开始时间，确保没有使用刹车和油门，并返回我们的第一个观察结果:


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

完整的代码为我们的CarEnv到目前为止是:

class CarEnv:
SHOW_CAM = SHOW_PREVIEW
STEER_AMT = 1.0
im_width = IMG_WIDTH
im_height = IMG_HEIGHT
actor_list = []
front_camera = None
collision_hist = []
def __init__(self):
self.client = carla.Client('localhost', 2000)
self.client.set_timeout(2.0)
# Once we have a client we can retrieve the world that is currently
# running.
self.world = self.client.get_world()
# The world contains the list blueprints that we can use for adding new
# actors into the simulation.
blueprint_library = self.world.get_blueprint_library()
# Now let's filter all the blueprints of type 'vehicle' and choose one
# at random.
#print(blueprint_library.filter('vehicle'))
self.model_3 = blueprint_library.filter('model3')[0]
def reset(self):
self.collision_hist = []
self.actor_list = []
self.transform = random.choice(self.world.get_map().get_spawn_points())
self.vehicle = self.world.spawn_actor(self.model_3, self.transform)
self.actor_list.append(self.vehicle)
self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
self.rgb_cam.set_attribute('image_size_x', f'{self.im_width}')
self.rgb_cam.set_attribute('image_size_y', f'{self.im_height}')
self.rgb_cam.set_attribute('fov', '110')
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))
time.sleep(4) # sleep to get things started and to not detect a collision when the car spawns/falls from sky.
colsensor = self.world.get_blueprint_library().find('sensor.other.collision')
self.colsensor = self.world.spawn_actor(colsensor, transform, attach_to=self.vehicle)
self.actor_list.append(self.colsensor)
self.colsensor.listen(lambda event: self.collision_data(event))
while self.front_camera is None:
time.sleep(0.01)
self.episode_start = time.time()
self.vehicle.apply_control(carla.VehicleControl(brake=0.0, throttle=0.0))
return self.front_camera

现在，让我们添加collision_data和process_img方法:


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

现在我们需要用step方法。该方法采取一个动作，然后按照通常的强化学习范式返回observation, reward, done, any_extra_info 。开始:


def step(self, action):
'''
For now let's just pass steer left, center, right?
0, 1, 2
'''
if action == 0:
self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=0))
if action == 1:
self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=-1*self.STEER_AMT))
if action == 2:
self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=1*self.STEER_AMT))

上面展示了我们如何根据传递给我们的数值动作来采取一个动作，现在我们只需要处理observation, possible collision, and reward:


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

我们得到了飞行器的速度，从速度转换为千米每小时。我这样做是为了避免探员在一个狭小的圈子里开车。如果我们需要一定的速度来获得奖励，这应该有希望抑制它。
接下来，我们会查看是否已经耗尽了我们的回合时间，然后我们会返回所有的东西。这样，我们的环境就完成了!
完整的代码到这一点:

import glob
import os
import sys
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
class CarEnv:
SHOW_CAM = SHOW_PREVIEW
STEER_AMT = 1.0
im_width = IMG_WIDTH
im_height = IMG_HEIGHT
actor_list = []
front_camera = None
collision_hist = []
def __init__(self):
self.client = carla.Client('localhost', 2000)
self.client.set_timeout(2.0)
# Once we have a client we can retrieve the world that is currently
# running.
self.world = self.client.get_world()
# The world contains the list blueprints that we can use for adding new
# actors into the simulation.
blueprint_library = self.world.get_blueprint_library()
# Now let's filter all the blueprints of type 'vehicle' and choose one
# at random.
#print(blueprint_library.filter('vehicle'))
self.model_3 = blueprint_library.filter('model3')[0]
def reset(self):
self.collision_hist = []
self.actor_list = []
self.transform = random.choice(self.world.get_map().get_spawn_points())
self.vehicle = self.world.spawn_actor(self.model_3, self.transform)
self.actor_list.append(self.vehicle)
self.rgb_cam = self.world.get_blueprint_library().find('sensor.camera.rgb')
self.rgb_cam.set_attribute('image_size_x', f'{self.im_width}')
self.rgb_cam.set_attribute('image_size_y', f'{self.im_height}')
self.rgb_cam.set_attribute('fov', '110')
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))
time.sleep(4) # sleep to get things started and to not detect a collision when the car spawns/falls from sky.
colsensor = self.world.get_blueprint_library().find('sensor.other.collision')
self.colsensor = self.world.spawn_actor(colsensor, transform, attach_to=self.vehicle)
self.actor_list.append(self.colsensor)
self.colsensor.listen(lambda event: self.collision_data(event))
while self.front_camera is None:
time.sleep(0.01)
self.episode_start = time.time()
self.vehicle.apply_control(carla.VehicleControl(brake=0.0, throttle=0.0))
return self.front_camera
def collision_data(self, event):
self.collision_hist.append(event)
def process_img(self, image):
i = np.array(image.raw_data)
#np.save("iout.npy", i)
i2 = i.reshape((self.im_height, self.im_width, 4))
i3 = i2[:, :, :3]
if self.SHOW_CAM:
cv2.imshow("",i3)
cv2.waitKey(1)
self.front_camera = i3
def step(self, action):
'''
For now let's just pass steer left, center, right?
0, 1, 2
'''
if action == 0:
self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=0))
if action == 1:
self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=-1*self.STEER_AMT))
if 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