第三节:场景管理器入门——ScenarioRunner框架、基本场景定义与运行
好,咱们今天来聊聊 ScenarioRunner。说实话,这是 Carla 里我最喜欢的模块之一。为什么?因为它把「写场景」这件事,从一堆杂乱无章的 Python 脚本,变成了结构清晰的工程化流程。
我个人习惯把 ScenarioRunner 理解成一个「导演」。你写剧本(场景定义),它负责调度演员(车辆、行人)、控制灯光、触发事件。你想想看,如果没有这个导演,你要手动控制每一帧的车辆位置、速度、碰撞检测……那得多累。
3.1 ScenarioRunner 框架概览
ScenarioRunner 是 Carla 官方提供的一个高级框架。它基于 Python,核心思想是:把测试场景抽象成可复用的模块。
它的架构其实不复杂,我画个简图给你看:
用户定义场景(OpenSCENARIO / Python API)
↓
ScenarioRunner 解析器
↓
行为树(Behavior Tree)执行引擎
↓
Carla 仿真核心(车辆、传感器、地图)
嗯,这里要注意:底层用的是行为树。不是状态机,不是有限状态机,是行为树。为什么选它?因为行为树天然支持并行、条件分支、回退。我在项目中遇到过用状态机写复杂场景,写到后面 if-else 嵌套到怀疑人生。换成行为树后,逻辑清晰多了。
核心组件一览:
- ScenarioManager:场景管理器,负责加载、运行、停止场景
- BehaviorAgent:行为代理,控制车辆执行具体动作(跟车、变道、紧急制动)
- RoutePlanner:路径规划器,生成全局路径
- ScenarioParser:场景解析器,把 OpenSCENARIO 或 Python 定义转成行为树
3.2 基本场景定义:两种方式
定义场景有两种主流方式。我建议你两种都掌握,因为实际项目中经常混用。
方式一:OpenSCENARIO 格式(推荐用于复杂场景)
OpenSCENARIO 是 ASAM 标准,说白了就是一个 XML 文件。它定义了「谁在什么时候做什么」。举个例子:
<?xml version="1.0" encoding="UTF-8"?>
<OpenSCENARIO>
<FileHeader author="deep3321" description="前车急刹场景"/>
<Storyboard>
<Init>
<Actions>
<Private action="teleport" actor="ego_vehicle" x="100" y="0" z="0"/>
<Private action="teleport" actor="target_vehicle" x="50" y="0" z="0"/>
</Actions>
</Init>
<Story name="紧急制动">
<Act>
<Maneuver>
<Event name="开始减速">
<Action>
<Private action="longitudinal">
<SpeedAction>
<Target>0</Target>
<Transition>3</Transition>
</SpeedAction>
</Private>
</Action>
</Event>
</Maneuver>
</Act>
</Story>
</Storyboard>
</OpenSCENARIO>
这个场景很简单:主车在 100 米处,目标车在 50 米处。然后目标车在 3 秒内减速到 0。嗯,典型的「前车急刹」场景。
我的小技巧:写 OpenSCENARIO 时,先用 Python 脚本快速验证逻辑,再转成 XML。因为 XML 调试起来比较痛苦,你想想看,少了一个标签就得从头解析。
方式二:Python API 定义(适合快速原型)
如果你不想碰 XML,直接用 Python 也行。ScenarioRunner 提供了 Python 接口:
import carla
from scenario_runner import ScenarioRunner
# 初始化
runner = ScenarioRunner(
host='localhost',
port=2000,
timeout=10.0
)
# 定义场景
scenario_config = {
'name': 'emergency_brake',
'actors': [
{'type': 'vehicle', 'id': 'ego', 'spawn': carla.Transform(carla.Location(100, 0, 0))},
{'type': 'vehicle', 'id': 'target', 'spawn': carla.Transform(carla.Location(50, 0, 0))}
],
'triggers': [
{'condition': 'distance_less_than', 'actor1': 'ego', 'actor2': 'target', 'value': 30}
],
'actions': [
{'actor': 'target', 'type': 'speed_change', 'target_speed': 0, 'duration': 3}
]
}
# 运行
runner.run_scenario(scenario_config)
这段代码干了什么?它定义了两个车,设置了一个触发条件(距离小于 30 米时触发),然后执行动作(目标车减速到 0)。
我曾经踩过的坑:用 Python API 定义场景时,spawn 的位置一定要确保在地图上。有一次我随手写了个 Location(0, 0, 0),结果车直接掉到地底下去了。嗯,后来我习惯先打印地图的 spawn points 再写坐标。
3.3 运行场景:命令行与 API
运行场景有两种方式。我平时调试用命令行,集成测试用 API。
命令行运行
# 运行 OpenSCENARIO 场景
python scenario_runner.py --scenario scenarios/emergency_brake.xosc
# 运行 Python 场景
python scenario_runner.py --scenario scenarios/emergency_brake.py
# 带参数运行
python scenario_runner.py --scenario scenarios/emergency_brake.xosc --reload --sync
参数说明:
| 参数 | 作用 |
|---|---|
--reload |
重新加载地图(建议每次运行都加,避免残留) |
--sync |
同步模式(帧率更稳定,适合录制数据) |
--timeout |
超时时间(默认 60 秒) |
API 集成运行
如果你想把场景跑在自动化测试流水线里,用 API 更合适:
from scenario_runner import ScenarioRunner
runner = ScenarioRunner()
result = runner.run_scenario('scenarios/emergency_brake.xosc')
if result.success:
print("场景通过!")
else:
print(f"场景失败:{result.error_message}")
我记得有一次做回归测试,用 API 批量跑了 200 个场景,结果发现 3 个场景因为地图版本不一致挂了。嗯,从那以后我每次跑之前都会检查地图版本。
3.4 场景生命周期与调试技巧
一个场景从加载到结束,会经历这几个阶段:
- 初始化:加载地图、生成车辆、设置初始状态
- 等待触发:监听条件(距离、时间、速度等)
- 执行动作:触发后执行预定义动作
- 结束条件:达到结束条件(超时、碰撞、完成)
- 清理:销毁车辆、重置状态
调试建议:
- 用
--debug参数启动,会打印每个行为树节点的状态 - 在场景里加
print语句,虽然土但有效 - 用 Carla 的 spectator 视角观察,比看日志直观多了
你可能会问:「为什么我的场景跑着跑着就卡住了?」
大概率是触发条件没满足。比如你设了距离小于 10 米触发,但两车始终保持 15 米距离。嗯,检查一下初始位置和速度设置。
3.5 实战:写一个最简单的场景
咱们动手写一个。目标:主车匀速行驶,前车在 5 秒后突然变道到主车前方。
# simple_cut_in.py
import carla
from scenario_runner import ScenarioRunner
def main():
runner = ScenarioRunner()
# 定义场景
scenario = {
'name': 'cut_in_test',
'actors': [
{
'type': 'vehicle',
'id': 'ego',
'spawn': carla.Transform(carla.Location(200, 0, 0)),
'speed': 10 # 10 m/s
},
{
'type': 'vehicle',
'id': 'cut_in_car',
'spawn': carla.Transform(carla.Location(150, -3.5, 0)), # 相邻车道
'speed': 12 # 稍快一点
}
],
'triggers': [
{'condition': 'time_elapsed', 'value': 5.0}
],
'actions': [
{
'actor': 'cut_in_car',
'type': 'lane_change',
'target_lane': 0, # 主车道
'duration': 2.0
}
],
'end_conditions': [
{'condition': 'collision', 'actor': 'ego'},
{'condition': 'time_elapsed', 'value': 15.0}
]
}
result = runner.run_scenario(scenario)
print(f"场景结果:{'通过' if result.success else '失败'}")
if __name__ == '__main__':
main()
这个场景里,我故意让 cut_in_car 速度比主车快 2 m/s。为什么?因为现实中变道车辆通常会加速。你想想看,如果它比主车慢,变道过来就是急刹,那场景就变成「前车急刹」了,不是「切入」了。
我的经验:写场景时,一定要想清楚「这个场景要测试什么」。是测试感知的响应时间?还是测试规划的避让策略?目标不同,参数设置完全不同。
好了,这一节的内容就到这。ScenarioRunner 的核心就是「定义-触发-执行」这个循环。你掌握了这个,后面那些复杂的多车交互、天气突变、传感器故障场景,都是在这个基础上加条件、加动作而已。
下一节咱们聊聊怎么用 ScenarioRunner 生成「边缘场景」——就是那些让自动驾驶系统头疼的 corner case。
公众号:蓝海资料掘金营,微信deep3321