Apollo 功能安全管理器 (Safety Manager)

Python 实现的 Apollo Cyber 功能安全监控工具，适配 Apollo 10.0 园区版安全框架。

目录

• 功能安全框架
• Statistics 配置
• 功能
• Apollo 通道说明
• 依赖
• 环境配置
• 运行
• 配置
• 功能安全现有配置项
• 告警级别
• 降级级别

功能安全框架

开源 Apollo 主要面向学习型开发者使用，不能提供实车场景安全保障，园区版新增了功能安全模块检测系统的异常情况并执行风险规避措施，整体架构如下：

┌─────────────────────────────────────────────────────┐
│                    Statistics 层                      │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────┐ │
│  │ Resource Stat│  │ Process Stat │  │ Latency    │ │
│  │              │  │              │  │ Stat       │ │
│  └──────────────┘  └──────────────┘  └────────────┘ │
└─────────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────┐
│                 Safety Manager 层                    │
│                                                      │
│  ┌──────────┐    ┌──────────┐    ┌───────────────┐  │
│  │ Checker  │───▶│ Exporter │───▶│ Analyzer      │  │
│  │ (异常检测) │    │ (输出)   │    │ (规则匹配+过滤) │  │
│  └──────────┘    └──────────┘    └───────┬───────┘  │
│                                          │          │
│                                     ┌────▼─────┐   │
│                                     │Aggregator│   │
│                                     │ (聚合)   │   │
│                                     └────┬─────┘   │
│                                          │         │
│                                     ┌────▼─────┐   │
│                                     │ Decider  │   │
│                                     │ (决策)   │   │
│                                     └────┬─────┘   │
│                                          │         │
│                                     ┌────▼─────┐   │
│                                     │ Guardian │   │
│                                     │ (执行)   │   │
│                                     └──────────┘   │
└─────────────────────────────────────────────────────┘
                        │
                        ▼
              /apollo/safety/status
              /apollo/control (刹停指令)

整体分为两部分：

• statistics：负责完成一些信息数据统计，比如资源使用信息、进程存在信息、模块间消息延迟信息等；
• safety_manager：即功能安全模块，负责检测是否存在异常，并触发安全风险规避指令。

Safety Manager 的执行过程：状态检测 → 发送 → 聚合决策 → 执行

组件	说明
Checker	负责异常检测，每种 checker 是个独立的 Cyber 组件
Exporter	基础类库，checker 调用 exporter 输出零散的异常信息
Analyzer	解析 exporter 的输出信息，根据策略判断每个检测项的实时状态
Aggregator	用于消息聚合，输出聚合消息
Decider	根据所有检测项结果生成刹停策略（缓刹或急刹）
Guardian	读取刹停策略，输出最终的 control 指令

Statistics 配置

Statistics 层负责完成信息数据统计，通常使用默认配置即可。

Resource Statistics

文件路径：modules/statistics/resource/conf/resource_statistic_conf.pb.txt

配置说明：CPU、GPU、Mem 会自动检测，主要是配置检测的磁盘对象。

disk_conf {
  # 配置一个检测的磁盘，支持标准的正则表达式匹配
  # sda1, nvme0n1 and etc
  device: ["sd[a-z][0-9]*", "nvme[\\w]*", "ssd[\\w]*"]
}

输出说明：channel /apollo/statistics/resources

资源	输出内容
CPU	包括 CPU 使用率和温度
GPU	包括 GPU 使用率和温度
内存	包括内存使用量、总量、空闲量、使用率
磁盘	包括磁盘使用量、总量、空闲量、使用率、负载（Orin 上暂未支持）

Process Statistics

文件路径：modules/statistics/process/conf/process_statistic_conf.pb.txt

配置说明：Mainboard 启动的 Cyber 组件会自行检测，主要是配置其他程序。

general_process {
  # 支持正则表达式配置
  regex: [".*/apollo/scripts/.*", "cyber_recorder record .*"]
}

输出说明：channel /apollo/statistics/process

项目	说明
mainboard_dag	列出所有的用 mainboard 启动的 dag
mainboard_process_group	列出所有 mainboard 的 group 组，即 -p 指定的参数
general_process	用户自定义的程序状态

Latency Statistics

文件路径：modules/statistics/latency/conf/latency_statistic_gflags.conf

配置说明：主要是配置指定融合点云的 channel（因为每个用户输出的 channel 名可能不一致）。

# set your lidar compensator topic, default use FLAGS_pointcloud_16_topic
#--compensator_topic=

输出说明：channel /apollo/statistics/latency

包括补偿点云、perception、prediction、planning、control 多个模块的延迟信息，每个延迟项包括：

指标	说明
proc_ms	处理耗时毫秒数，包括 min、max、avg，以及计算采用的样本数量
chain_ms	主 lidar 读取时间到当前模块的延迟，包括 min、max、avg，以及计算采用的样本数量

功能

模块	说明	对应的官方 Checker
ChannelFreqChecker	检查通道频率/数据是否到达，解析 protobuf 验证字段有效性	channel_freq_checker
ProcessChecker	通过 cyber.NodeUtils.get_nodes() 监控 Apollo 节点运行状态	process_checker
ResourceChecker	监控系统 CPU、内存、磁盘使用率	resource_checker
BatteryChecker	监控系统电池电量	—
SystemHealthChecker	监控系统运行时间、负载、CPU 频率	—
Analyzer	规则匹配 + 历史过滤，抑制瞬时误报	analyzer
Aggregator	按模块聚合安全决策	aggregator
Decider	基于分数模型决定降级级别	decider
Guardian	执行控制动作（正常/限速/软停车/紧急停车）	guardian

Apollo 通道说明

输入通道（监控）

通道名称	消息类型	描述	默认频率
/apollo/perception/obstacles	apollo::perception::PerceptionObstacles	感知输出障碍物消息通道	10Hz
/apollo/planning	apollo::planning::ADCTrajectory	车辆规划轨迹线信息	10Hz
/apollo/prediction	apollo::prediction::PredictionObstacles	障碍物预测信息	10Hz
/apollo/localization/pose	apollo::localization::LocalizationEstimate	车辆定位信息	200Hz
/apollo/canbus/chassis	apollo::canbus::Chassis	车辆底盘信息（速度、油门、刹车、档位、灯光等）	100Hz
/apollo/perception/traffic_light	apollo::perception::TrafficLight	交通灯感知信息（颜色、ID 等）	10Hz
/apollo/control	apollo::control::ControlCommand	控制指令（方向盘、油门、刹车等）	100Hz
/apollo/guardian	apollo::guardian::GuardianCommand	安全指令	100Hz
/apollo/chassis_control	apollo::external_command::ChassisCommand	外部底盘命令	100Hz

传感器通道

通道名称	描述
/apollo/sensor/camera/CAM_FRONT/image	前摄像头
/apollo/sensor/camera/CAM_FRONT_RIGHT/image	右前摄像头
/apollo/sensor/camera/CAM_FRONT_LEFT/image	左前摄像头
/apollo/sensor/camera/CAM_BACK/image	后摄像头
/apollo/sensor/camera/CAM_BACK_LEFT/image	左后摄像头
/apollo/sensor/camera/CAM_BACK_RIGHT/image	右后摄像头

输出通道

通道名称	描述
/apollo/safety/status	功能安全状态通道（统一输出）

依赖

psutil
pyyaml

环境配置

在运行 Safety Manager 之前，需要设置 Apollo Cyber 环境变量：

export PYTHONPATH=/apollo_workspace/bazel-bin/cyber/python:/apollo/bazel-bin/modules:$PYTHONPATH

确保 libfastrtps 等共享库在 LD_LIBRARY_PATH 中。

运行

cd /apollo_workspace/safetymanager
python3 ./main.py

程序会持续运行，按 Ctrl+C 停止。

配置

所有配置文件位于 configs/ 目录：

配置文件	用途	对应官方配置
channel_freq_checker_conf.yaml	通道频率监控项（通道名、期望频率、阈值）	channel_freq_checker_conf.pb.txt
process_checker_conf.yaml	需要监控的 Apollo 节点列表	process_checker_conf.pb.txt
resource_checker_conf.yaml	CPU/内存/磁盘告警阈值	resource_checker_conf.pb.txt
rule_matcher_conf.yaml	分析器规则匹配配置	rule_matcher_conf.pb.txt
guardian_conf.yaml	降级决策阈值	guardian_conf.pb.txt

通道频率配置示例

items:
  - name: "/apollo/planning"
    level: "ERROR"
    expect_freq: 10           # 期望频率 (Hz)
    tags: ["planning", "critical"]

阈值通用配置说明

参数	说明
report_interval_ms	检查结果输出频率（毫秒）
filter.required_cycles / filter.total_cycles	过滤策略，如 3/5 表示 5 个周期内 3 次异常才判定为异常
level	报警级别：WARN=警告，ERROR=缓刹，FATAL=急刹
expect_freq	通道期望发布频率 (Hz)，用于自动计算超时阈值
tags	检测项标签，用于聚合和规则匹配

Checker 通用配置说明

各个 checker 的通用配置主要包括：

参数	说明
report_interval_ms	检查结果输出频率（毫秒）
filter_required_cycles / filter_total_cycles	过滤策略，如 3/5 表示 5 个周期内 3 次异常才判定为异常
duration_ms	异常信息持续多长时间才会判定为异常
status_timeout	异常消息经过多久会置为正常
level	报警级别：WARN=警告，ERROR=缓刹，FATAL=急刹

所有 checker 的最终结果都会输出到 /apollo/safety/status channel。

通道频率配置详解

每个 item 一行，指定通道名称、异常级别以及频率下限和上限：

item {
  name: "/apollo/perception/obstacles"
  level: ERROR
  expect_freq: 10
  lower_limit: 6
  upper_limit: 15
}

碰撞检测配置 (collision_checker)

配置警告、缓刹、急刹的阈值，以及选取的策略（敏感/普通/精确）：

参数	说明
warn_sec	碰撞预警时间（秒），发送警告状态
error_sec	碰撞预警时间（秒），舒适刹车
fatal_sec	碰撞预警时间（秒），紧急刹车
type	策略类型：SENSITIVE / NORMAL / ACCURATE

延迟检测配置 (latency_checker)

配置指定模块耗时或延迟阈值：

参数	说明
module	模块：LIDAR_COMPENSATOR, PERCEPTION, PREDICTION, PLANNING, CONTROL
type	类型：PROC_MS（处理耗时）, CHAIN_MS（链路延迟）
avg	平均延迟阈值（毫秒）
sample_size_lower_limit	采样数量下限，低于此值会被忽略

文件存在性配置 (file_exist_checker)

指定文件路径和报警级别：

item {
  name: "novatel_localization_extrinsics.yaml"
  path: "./modules/localization/msf/params/novatel_localization_extrinsics.yaml"
  level: FATAL
}

进程存在性配置 (process_checker)

检测 dag、mainboard 进程组、普通进程是否存在：

item {
  name: "Perception Process Group"
  type: MAINBOARD_PROCESS_GROUP
  value: "perception"
  level: FATAL
}

资源占用配置 (resource_checker)

配置 CPU、GPU、内存、磁盘的使用上限阈值：

cpu:
  usage_warning: 80
  usage_error: 90
memory:
  available_warning: 2000
  available_error: 1000
disk:
  device: "sda1"
  usage_warning: 80
  usage_error: 90

控制一致性配置 (control_checker)

检测自动驾驶控制指令和车体实际控制情况的偏差，通常使用默认配置即可。

定位状态配置 (localization_checker)

读取 localization status channel 状态返回，可配置等级映射关系：

状态	值
MeasureState::ERROR	1
MeasureState::CRITICAL_ERROR	2
MeasureState::FATAL_ERROR	3

功能安全现有配置项

WARN 只警告，ERROR 缓刹，FATAL 急刹 配置文件在 /apollo/modules/safety_manager/guardian/conf/guardian_conf.pb.txt 中 默认缓刹阈值 25，急刹阈值 50，可以修改。

通道频率检测

通道	描述	级别	阈值
/apollo/sensor/lidar16/back/PointCloud2	顶 lidar 频率	ERROR	lower: 6, upper: 15
/apollo/sensor/lidar16/left/PointCloud2	左 lidar 频率	ERROR	lower: 6, upper: 15
/apollo/sensor/lidar16/right/PointCloud2	右 lidar 频率	ERROR	lower: 6, upper: 15
/apollo/sensor/lidar16/compensator/PointCloud2	运动补偿 lidar 频率	ERROR	lower: 6, upper: 15
/apollo/perception/obstacles	感知输出频率	ERROR	lower: 6, upper: 15
/apollo/prediction	预测输出频率	ERROR	lower: 6, upper: 15
/apollo/planning	规划输出频率	ERROR	lower: 6, upper: 15
/apollo/control	控制输出频率	ERROR	lower: 80, upper: 150
/apollo/canbus/chassis	底盘输出频率	ERROR	lower: 80, upper: 150
/apollo/canbus/chassis_detail	底盘详情输出频率	ERROR	lower: 80, upper: 150
/apollo/guardian	功能安全控制输出频率	ERROR	lower: 80, upper: 150
/apollo/sensor/gnss/imu	GNSS 驱动 IMU 输出频率	ERROR	lower: 60, upper: 150
/apollo/localization/pose	定位输出频率	ERROR	lower: 80, upper: 150
/tf	TF 输出频率	ERROR	lower: 80, upper: 150

文件存在性检测

文件	描述	级别
novatel_localization_extrinsics.yaml	IMU 外参配置文件是否存在	FATAL
static_transform_conf.pb.txt	静态 TF 配置文件是否存在	FATAL

延迟检测

检测项	描述	级别	阈值
lidar compensator proc time	Lidar 驱动处理耗时	ERROR	avg: 150ms, samples: 10
lidar compensator chain time	Lidar 点云输入到运动补偿输出耗时	ERROR	avg: 150ms, samples: 10
perception proc time	感知程序处理耗时（点云输入到障碍物输出）	ERROR	avg: 200ms, samples: 10
perception chain time	Lidar 点云输入到感知障碍物输出耗时	ERROR	avg: 250ms, samples: 10
prediction proc time	预测程序处理耗时	ERROR	avg: 100ms, samples: 10
prediction chain time	Lidar 点云输入到预测障碍物输出耗时	ERROR	avg: 350ms, samples: 10
planning proc time	规划程序处理耗时	ERROR	avg: 100ms, samples: 10
planning chain time	Lidar 点云输入到规划结果输出耗时	ERROR	avg: 450ms, samples: 10
control proc time	控制程序处理耗时	ERROR	avg: 100ms, samples: 10
control chain time	Lidar 点云输入到控制结果输出耗时	ERROR	avg: 500ms, samples: 10

定位状态检测

检测项	描述	级别	说明
Localization	定位状态	根据 MSF Status	根据 status 阈值决定

进程存在性检测

检测项	描述	级别
Perception Process Group	感知进程	FATAL
Static Transform	TF 进程	FATAL
Localization Process Group	定位进程	FATAL
Planning	规划进程	FATAL
Control	控制进程	FATAL
Canbus	Canbus 进程	FATAL
Statistics Process Group	资源统计进程	ERROR
Recorder Process	落盘进程	ERROR
Data Clean Process	数据清理进程	WARN

资源占用检测

检测项	描述	阈值
CPU	CPU 资源占用	usage_warning: 80, usage_error: 90, temperature_warning: 75, temperature_error: 85
Memory	内存资源占用	usage_warning: 80, usage_error: 90, available_warning: 2000, available_error: 1000
Disk	磁盘资源占用	device: "nvme0n1p1", usage_warning: 80, usage_error: 90

碰撞检测

检测项	描述	级别	阈值
Collision	碰撞检测	WARN	warn_sec: 5.0 (发送警告状态)
Collision	碰撞检测	ERROR	error_sec: 4.0 (舒适刹车)
Collision	碰撞检测	FATAL	fatal_sec: 3.0 (紧急刹车)

控制一致性检测

检测项	描述	级别	说明
Control	控制一致性检测	根据配置项	参考配置文件，通常用默认配置

告警级别

级别	含义	分数
OK	正常	0
WARN	警告	5
ERROR	错误	15
FATAL	致命	30

降级级别

级别	名称	动作
L0	正常	自动驾驶正常运行
L1	限速	限速 30km/h，禁止变道
L2	软停车	减速靠边，请求接管
L3	紧急停车	最大减速度停车，双闪灯

使用方法

1. 修改 /apollo/modules/canbus/conf/canbus.conf 配置，注释掉最后一行 --noreceive_guardian，改成 #--noreceive_guardian
2. 根据需要调整 configs/ 下的配置文件
3. 启动 python3 ./main.py

使用方法

1. 修改 /apollo/modules/canbus/conf/canbus.conf 配置，注释掉最后一行 --noreceive_guardian，改成 #--noreceive_guardian
2. 根据需要调整 configs/ 下的配置文件
3. 启动 python3 ./main.py

自定义添加 Checker

功能安全的 exporter 对外提供了 callback 和主动发送两种接口，消息会统一输出到 /apollo/safety/status channel。

Python 版本

from checkers.base import BaseChecker
from exporter import SafetyExporter
from models import SafetyStatus, SafetyLevel

class CustomChecker(BaseChecker):
    def __init__(self, config: dict, exporter: SafetyExporter):
        super().__init__("CustomChecker", config, exporter)

    def run_once(self) -> SafetyStatus:
        # 自定义检测逻辑
        is_ok = check_something()
        if is_ok:
            return SafetyStatus(
                name="CustomCheck", source=self.name,
                level=SafetyLevel.OK, message="OK",
            )
        else:
            return SafetyStatus(
                name="CustomCheck", source=self.name,
                level=SafetyLevel.ERROR, message="Error detected",
            )

C++ 版本参考

#include <iostream>
#include "cyber/cyber.h"
#include "modules/safety_manager/exporter/exporter.h"

namespace safety_ns = apollo::safety_manager;

class GnssChecker {
public:
    explicit GnssChecker(const std::shared_ptr<apollo::cyber::Node>& node) : node_(node) {
        // set period: 1s
        exporter_ = std::make_shared<safety_ns::SafetyExporter>(node_, 1000);
        // set default filter condition
        exporter_->SetDefaultFilter(3, 3);
    }

    void Run() {
        exporter_->Add("Gnss Status", this, &GnssChecker::check);
    }

protected:
    virtual void check(safety_ns::SafetyStatusWrapper* status_wrapper) {
        // do some gnss status check
        status_wrapper->Summary(safety_ns::SafetyStatus_Status_OK, "OK");
    }

private:
    std::shared_ptr<apollo::cyber::Node> node_;
    std::shared_ptr<safety_ns::SafetyExporter> exporter_;
};

int main(int argc, char** argv) {
    apollo::cyber::Init(argv[0]);
    std::shared_ptr<apollo::cyber::Node> node = apollo::cyber::CreateNode("safety_exporter_example");

    // create safety status reader
    apollo::cyber::ReaderConfig reader_cfg;
    reader_cfg.channel_name = SAFETY_EXPORTER_TOPIC;
    reader_cfg.pending_queue_size = 10;
    node->CreateReader<safety_ns::SafetyStatus>(
            reader_cfg, [](const std::shared_ptr<safety_ns::SafetyStatus>& safety_status) {
                AINFO << "receive safety status: " << safety_status->DebugString();
            });

    // report status directly
    auto exporter = std::make_shared<safety_ns::SafetyExporter>(node);
    exporter->Report("CPU Usage", safety_ns::SafetyStatus_Status_WARN, "value 75%, threshold 70%");

    apollo::cyber::WaitForShutdown();
}