第三节:场景管理器入门——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 场景生命周期与调试技巧

一个场景从加载到结束,会经历这几个阶段:

  1. 初始化:加载地图、生成车辆、设置初始状态
  2. 等待触发:监听条件(距离、时间、速度等)
  3. 执行动作:触发后执行预定义动作
  4. 结束条件:达到结束条件(超时、碰撞、完成)
  5. 清理:销毁车辆、重置状态

调试建议:

  • --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