跳转到内容
SMI Developer Center

FX 基于 AI 的 PRPD 分类器概述

概述

自 BATCAM FX 固件 v1.0.3c 起，本公司自主研发的基于 AI 的局部放电分类功能已集成到固件中。该功能利用 CUDA 与 ONNX Runtime，设计为可在设备内实时对局部放电类型进行分类。

PRPD（Phase-Resolved Partial Discharge）是针对电源相位分析局部放电信号的局部放电诊断方法，BATCAM FX 的 AI 分类器以该 PRPD 数据作为输入判别局部放电类型。

本文档说明该功能的用户如何接收并利用基于 AI 的分类结果。 固件支持的分类结果共 4 种（Corona、Floating、Surface、Noise），并按各 LPoint Audio 通道（CH0~CH2）分别提供独立的结果。

类别说明
corona电晕放电（Corona discharge）类型
floating悬浮电极（Floating electrode）放电类型
surface表面（沿面，Surface）放电类型
noise非局部放电的背景噪声·信号

分类结果通过 ROS 2 与 WebSocket 两个通道提供。针对各 LPoint Audio 通道，提供所支持的各局部放电类型的 logit score，用户可基于此推断分类结果。

💡 注意： 本文档以 BATCAM FX 为准。BATCAM FX2 的 AI 功能请参阅 设备管理器 — 人工智能 文档。

前提条件

  • 固件版本：BATCAM FX 固件 v1.0.3c 及以上。ROS 2 输出在 v1.0.3b 中首次添加，包含 WebSocket 输出在内的全部功能自 v1.0.3c 起提供。详细的变更历史请参阅 固件发行说明。

  • 版本确认方法：在浏览器中访问 {CAMERA_IP}/setting/status 页面，或确认 REST API GET /setting/status 响应中的 firmware_version 值。（参阅 FX API 演练场）

  • 接收端 ROS 2 环境：BATCAM FX 的 ROS 2 功能基于 ROS 2 Humble Hawksbill，消息包 fx-stream-msgs 以在 Humble 上使用为前提，在 Python 3.10.16、ROS 2 Jazzy（Ubuntu 24.04）上完成了构建与验证。（参阅 ROS 概述）

  • ROS_DOMAIN_ID 一致：相机的默认 ROS_DOMAIN_ID 为 123。接收端环境也必须使用相同的值，话题才可见。（参阅 ROS Domain ID 设置）

数据接收

ROS 2 话题

BATCAM FX 以相机的硬件 ID 为基准创建 PRPD 分类结果话题。

项目值
话题名称/fx_{hardware_id}/prpd
消息类型fx_stream_msgs/msg/Prpd
消息大小52 B/条消息
平均带宽约 220 B/s

Prpd 消息针对 4 个类别分别提供包含各通道（CH0~CH2）logit 值的长度为 3 的数组。

字段类型说明
noisefloat32[3]通道 CH0~CH2 的 noise 类别 logit（无量纲 score）
coronafloat32[3]通道 CH0~CH2 的 corona 类别 logit（无量纲 score）
floatingfloat32[3]通道 CH0~CH2 的 floating 类别 logit（无量纲 score）
surfacefloat32[3]通道 CH0~CH2 的 surface 类别 logit（无量纲 score）

logit 值是未经归一化的 score，正值和负值均有可能，因此必须基于相对大小进行解释。

完整的 ROS 2 话题列表与消息定义整理在 ROS 概述 的消息定义段落中，消息格式的原始定义可在 SMI OSS - fx-stream-msgs 仓库中确认。

WebSocket (Protobuf) 流

自固件 v1.0.3c 起，相同的分类结果也以基于 WebSocket 的 Protobuf 流形式提供。在订阅通道中以 Subscribe ID 3 接收 Prpd 事件，以 BATCAM FX 为准约以 4 Hz 发送。

WebSocket 连接、订阅、解析全过程的 Python 示例请参阅 Python Protobuf 示例 文档。该文档中也一并整理了 FX 与 FX2 之间的差异。

分类方法

用户针对各通道，在 4 个类别的 logit 值中选择具有最大值的项（argmax）。 此外，如果希望计算各结果的概率，则需要从各通道中取出与该项对应的值并进行 softmax。

通道noise (logit)corona (logit)floating (logit)surface (logit)分类结果 (argmax)概率 (softmax)
ch0-0.962.98-3.550.30corona0.918 ( 91.8 % )
ch14.485.88-5.45-1.51corona0.802 ( 80.2 % )
ch2-6.54-1.390.730.45floating0.533 ( 53.3 % )

💡 解释注意： softmax 概率是 4 个类别之间的相对置信度。像示例中的 ch2 那样最高概率计算得较低（53.3 %）的情况，意味着 floating（0.73）与 surface（0.45）的 logit 差异较小，两个类别处于竞争状态。是否采纳分类结果的判断标准（阈值）会因应用领域和现场条件而异，因此请根据所使用系统的需求进行定义并应用。

Softmax

softmax 函数将 logit 向量转换为概率分布，其一般公式如下：

softmax(xi)=exi∑j=1Kexj\text{softmax}(x_i) = \frac{e^{x_i}}{\sum_{j=1}^{K} e^{x_j}}

但是，当 logit 值过大或过小时，可能会发生 上溢（Overflow） 或 下溢（Underflow） ，因此通常按如下归一化形式进行计算：

softmax(xi)=exi−max⁡(x)∑j=1Kexj−max⁡(x)\text{softmax}(x_i) = \frac{e^{x_i - \max(x)}}{\sum_{j=1}^{K} e^{x_j - \max(x)}}

将下面的公式用 Python 实现时，如下例所示：

Python 实现

import numpy as np


def softmax(logits):
    logits = np.array(logits)
    exps = np.exp(logits - np.max(logits)) # 稳定化的 softmax
    return exps / np.sum(exps) # 返回概率值

Common Lisp 实现

(defun softmax (logits)
  "接收列表或向量 LOGITS，返回 softmax 概率向量"
  (let* ((vec (coerce logits 'vector))
         (maxval (reduce #'max vec))
         (exps (map 'vector (lambda (x) (exp (- x maxval))) vec))
         (sum   (reduce #'+ exps)))
    (map 'vector (lambda (x) (/ x sum)) exps)))

完整示例：rclpy 订阅节点

以下是订阅 PRPD 话题并将各通道的分类结果与概率输出为日志的完整 Python（rclpy）节点。可保存为单个文件（prpd_subscriber.py）并直接运行。

💡 注意： 代码中的话题名称 /fx_276730383020104/prpd 为示例，必须根据所使用相机的硬件 ID 进行更改。实际的话题名称可通过 ros2 topic list 命令确认。

import numpy as np
import rclpy
from rclpy.node import Node


from fx_stream_msgs.msg import Prpd


# 与 Prpd.msg 字段顺序相同的标签定义
LABELS = ['noise', 'corona', 'floating', 'surface']


def softmax(logits):
    logits = np.array(logits)
    exps = np.exp(logits - np.max(logits))  # 稳定化的 softmax
    return exps / np.sum(exps)


class PrpdSubscriber(Node):
    def __init__(self):
        super().__init__('prpd_subscriber')
        # 请根据所使用相机的硬件 ID 更改话题名称。
        self.subscription = self.create_subscription(
            Prpd,
            '/fx_276730383020104/prpd',
            self.listener_callback,
            10)


    def listener_callback(self, msg: Prpd):
        results = []


        # 以 3 通道 (CH0 ~ CH2) 为准
        for ch in range(3):
            logits = [
                msg.noise[ch],
                msg.corona[ch],
                msg.floating[ch],
                msg.surface[ch],
            ]
            probs = softmax(logits)
            idx = int(np.argmax(probs))
            results.append({
                'channel': f'CH{ch}',
                'label': LABELS[idx],
                'probability': round(float(probs[idx]), 3),
            })


        # 日志输出
        for result in results:
            self.get_logger().info(
                f"{result['channel']}: {result['label']} "
                f"({result['probability'] * 100:.1f}%)"
            )


def main(args=None):
    rclpy.init(args=args)
    node = PrpdSubscriber()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

运行方法

在 source 了 ROS 2 环境以及构建了 fx-stream-msgs 消息包的工作空间之后运行。必须使用与相机相同的 ROS_DOMAIN_ID（默认值 123）。

Terminal window
source /opt/ros/humble/setup.bash
# 同时 source 构建了 fx-stream-msgs 消息包的工作空间。
export ROS_DOMAIN_ID=123
python3 prpd_subscriber.py

预期输出

假设接收到上述“分类方法”中的示例 logit 值，则会输出如下日志：

[INFO] [1748324429.123456789] [prpd_subscriber]: CH0: corona (91.8%)
[INFO] [1748324429.123512345] [prpd_subscriber]: CH1: corona (80.2%)
[INFO] [1748324429.123556789] [prpd_subscriber]: CH2: floating (53.3%)

Common Lisp 参考实现

这是相同分类逻辑（softmax + argmax）的 Common Lisp 参考实现。消息接收部分需根据所使用的环境进行对接，以下代码假设 msg 中 (noise corona floating surface) 各自包含长度为 3 的向量。

; Define labels for the discharge types
(defparameter *labels* #("noise" "corona" "floating" "surface"))


; Softmax function
(defun softmax (logits)
  "接收列表或向量 LOGITS，返回 softmax 概率向量"
  (let* ((vec (coerce logits 'vector))
         (maxval (reduce #'max vec))
         (exps (map 'vector (lambda (x) (exp (- x maxval))) vec))
         (sum   (reduce #'+ exps)))
    (map 'vector (lambda (x) (/ x sum)) exps)))


(defun classify-channel (noise corona floating surface)
  "接收单个通道的对数值，返回标签/概率"
  (let* ((logits (vector noise corona floating surface))
         (probs  (softmax logits))
         (idx    (position (reduce #'max probs) probs)))
    (list :label (aref *labels* idx)
          :probability (aref probs idx))))


(defun listener-callback (msg)
  "假设 msg 的 (noise corona floating surface) 各自为长度 3 的向量"
  (loop for ch from 0 below 3
        for result = (classify-channel
                      (aref (getf msg :noise) ch)
                      (aref (getf msg :corona) ch)
                      (aref (getf msg :floating) ch)
                      (aref (getf msg :surface) ch))
        do (format t "CH~A: ~A (~,1F%)~%"
                   ch
                   (getf result :label)
                   (* (getf result :probability) 100))))

故障排除 (Troubleshooting)

  • 看不到 prpd 话题时：

    • 确认固件版本是否为 v1.0.3c 及以上。在浏览器中访问 {CAMERA_IP}/setting/status，或确认 GET /setting/status 的 firmware_version 值，如果是旧版本，请通过 Web 更新 或 Shell 更新 更新固件。

    • 确认接收端的 ROS_DOMAIN_ID 是否与相机一致。相机的默认值为 123，在固件 1.0.3 及以上版本中可通过 REST API（/ros/domain）查询和变更。（参阅 FX API 演练场、ROS Domain ID 设置）

    • 使用 ros2 topic list 再次确认 /fx_{hardware_id}/prpd 话题是否出现。

  • 发生 ModuleNotFoundError: No module named 'fx_stream_msgs' 错误时：

    • 这是尚未 source 构建了 fx-stream-msgs 消息包的工作空间的状态。请在 source 工作空间的 setup.bash 之后重新运行。

  • softmax 计算中出现 overflow/underflow 警告时：

    • 不要直接对 logit 值取指数，而应使用上述“Softmax”段落中的归一化形式（减去 np.max 后再取指数）。

  • 最高概率计算得较低时：

    • 这并非错误，而是意味着两个以上类别的 logit 相近而处于竞争状态。请查看上述“分类方法”中的解释注意事项。

相关文档