컨텐츠로 건너뛰기
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()

실행 방법

ROS 2 환경과 fx-stream-msgs 메시지 패키지를 빌드한 워크스페이스를 소싱한 뒤 실행합니다. 카메라와 동일한 ROS_DOMAIN_ID(기본값 123)를 사용해야 합니다.

Terminal window
source /opt/ros/humble/setup.bash
# 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 값을 확인하고, 구버전인 경우 웹 업데이트 또는 셸 업데이트로 펌웨어를 업데이트합니다.

    • 수신측의 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' 오류가 발생하는 경우:

    • fx-stream-msgs 메시지 패키지를 빌드한 워크스페이스가 소싱되지 않은 상태입니다. 워크스페이스의 setup.bash를 소싱한 뒤 다시 실행합니다.

  • softmax 계산에서 overflow/underflow 경고가 발생하는 경우:

    • logit 값을 그대로 지수화하지 말고, 위 “Softmax” 문단의 정규화된 형태(np.max를 뺀 후 지수화)를 사용합니다.

  • 최고 확률이 낮게 계산되는 경우:

    • 오류가 아니라 두 개 이상의 클래스 logit이 비슷하여 경합하고 있다는 의미입니다. 위 “분류 방법”의 해석 참고를 확인하십시오.

관련 문서