blog

ROS2 데이터 파이프라인 구축 및 디버깅

NEXTREE.

• 12 min read
ROS2 데이터 파이프라인 구축 및 디버깅

산업용 IoT나 로보틱스 환경에서 시스템을 설계하다 보면, 필연적으로 수많은 센서 데이터와 제어 명령을 실시간으로 처리해야 하는 과제에 직면하게 됩니다. ROS2(Robot Operating System 2)는 이러한 문제를 해결하는 데 활용될 수 있습니다.

ROS2는 이름에 '운영체제(OS)'가 포함되어 있지만, 윈도우나 리눅스 같은 실제 운영체제가 아니라 로봇 응용 프로그램을 쉽게 개발할 수 있도록 돕는 소프트웨어 프레임워크입니다. 이름에 '로봇'이 들어가 있지만, 그 본질은 강력한 비동기 분산 통신 아키텍처입니다. 효과적인 센서 데이터 수집에 활용되는 ROS2의 데이터 흐름을 이해한 뒤, 이를 어떻게 설계하고 디버깅하는지 알아보겠습니다.

1. ROS2 통신: DDS와 비동기 Pub/Sub 패턴

ROS2 아키텍처를 이해하기 위해 가장 먼저 알아야 할 개념은 DDS(Data Distribution Service)입니다. DDS는 분산 시스템 환경에서 기기 간에 데이터를 빠르고 안정적으로 교환하기 위해 만들어진 국제 산업 표준 통신 미들웨어입니다.

기존의 전통적인 웹 서비스가 주로 클라이언트의 요청에 서버가 응답하는 동기식 REST API에 의존했다면, 실시간 센서 데이터를 다루는 환경에서는 이러한 방식이 엄청난 병목을 초래합니다. 반면 DDS는 클라이언트가 서버에 매번 데이터를 요청할 필요 없이, 데이터가 생성될 때마다 필요한 곳에 즉시 전달하는 발행-구독 모델을 사용합니다. 이를 통해 수많은 센서가 동시다발적으로 쏟아내는 방대한 데이터를 지연 없이 실시간으로 처리할 수 있습니다. DDS 표준의 특징은 다음과 같습니다.

1.1 Pub/Sub 구조

  • 생산자와 소비자의 분리: 데이터를 생성하는 Publisher(예: 온도 센서 노드)와 데이터를 소비하는 Subscriber(예: 데이터베이스 저장 노드, UI 렌더링 노드)는 서로의 존재를 알 필요가 없습니다.

  • 유연한 확장성: 단순히 특정 토픽을 중간 매개체로 데이터를 던지고 받기 때문에, 시스템 가동 중에도 새로운 모니터링 노드를 추가하거나 기존 노드를 제거하는 것이 매우 자유롭습니다. 이는 Command와 Query의 책임을 분리하여 시스템의 유연성을 극대화하는 최신 소프트웨어 아키텍처 패턴들과도 일맥상통합니다.

1.2 QoS (Quality of Service)

DDS의 또 다른 강력함은 QoS입니다. 네트워크 상태가 불안정할 때 데이터를 무조건 다 보낼 것인지, 아니면 최신 데이터만 빠르게 보낼 것인지 등을 노드 단위로 세밀하게 설정할 수 있어 산업 현장의 요구사항에 유연하게 대응할 수 있습니다.

2. 센서 데이터 모델링: 비즈니스 도메인 최적화

ROS2를 사용하기 위해서 가장 먼저 필요한 것은 데이터 규격을 정의하는 것입니다. ROS2에서는 .msg 파일을 통해 데이터 인터페이스를 정의합니다. 데이터 구조 관리의 편의성을 위해 메시지 전용 패키지와 실행 로직 패키지를 분리하는 것이 기본입니다.

2.1 메시지 패키지 생성

터미널에서 워크스페이스(예: ~/ros2_ws/src)로 이동하여 메시지 전용 패키지를 생성합니다. 메시지 패키지는 의존성 관리를 위해 보통 C++ 빌드 시스템인 ament_cmake를 사용합니다.

ros2 pkg create --build-type ament_cmake iiot_interfaces

2.2 Custom Message 정의

퍼블리셔와 서브스크라이버는 서로 존재를 모르지만, 메시지를 주고받을 때 사용될 데이터의 형태는 알고 있어야합니다. 실제 비즈니스 도메인에서 사용할 법한 메타데이터(디바이스 ID, 상태 코드 등)를 포함하여 데이터 타입을 정의합니다.

# interfaces/msg/SensorData.msg
string device_id # 장비 고유 식별자 (예: "CNC_Machine_01")
int32 status_code # 장비 상태 코드 (0: 정상, 1: 경고, 2: 에러)
float64 temperature # 섭씨 온도
float64 vibration_hz # 진동 주파수 (Hz)

2.3 패키지 설정 파일 업데이트

ROS2 빌드 시스템(colcon)이 방금 만든 .msg 파일을 인식하고 Python/C++ 코드로 변환할 수 있도록 설정 파일들을 수정해야 합니다.

[ package.xml 수정 ]

<buildtool_depend>
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
</buildtool_depend>

 [ CMakeLists.txt 수정 ]

find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/SensorData.msg"
)

이후 colcon build --packages-select iiot_interfaces 명령어로 메시지 패키지를 먼저 빌드합니다.

3. Pub/Sub 노드 구현

데이터 규격이 준비되었으니, 실제로 데이터를 생산하고 소비하는 파이썬 노드를 작성해야합니다.

3.1 로직 패키지 생성

위에서 정의한 데이터 타입을 활용하여 실제 데이터 처리 로직을 담당할 새로운 파이썬 패키지를 생성하는 명령어를 실행합니다.

ros2 pkg create --build-type ament_python ${패키지이름} --dependencies rclpy interfaces

3.2 Publisher 노드: 센서 데이터 발행

- sensor_pub.py 파일 생성: 실제 하드웨어 센서나 IIoT 장비(예: CNC 머신)에서 발생하는 데이터를 읽어와 ROS2 네트워크 상의 특정 토픽으로 계속해서 발행하는 역할을 하는 스크립트입니다.

import rclpy
from rclpy.node import Node
import random
from iiot_interfaces.msg import SensorData

class SensorPublisher(Node):
    def __init__(self):
        super().__init__('sensor_publisher_node')
        self.publisher_ = self.create_publisher(SensorData, '/machine_data', 10)
        timer_period = 0.5
        self.timer = self.create_timer(timer_period, self.publish_sensor_data)
        self.get_logger().info('센서 퍼블리셔 노드가 시작되었습니다.')

    def publish_sensor_data(self):
        msg = SensorData()
        msg.device_id = "CNC_Machine_01"
        msg.status_code = 0 if random.random() > 0.1 else 1
        msg.temperature = round(random.uniform(40.0, 85.0), 2)
        msg.vibration_hz = round(random.uniform(10.0, 100.0), 2)

        self.publisher_.publish(msg)

def main(args=None):
    rclpy.init(args=args)
    node = SensorPublisher()

    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info('사용자에 의해 종료되었습니다.')
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

3.3 Subscriber 노드: 데이터 수신 및 가공

- data_sub.py 파일 생성: 퍼블리셔가 네트워크 상의 토픽(/machine_data)으로 열심히 던져준 센서 데이터를 수신하여 검사하고 활용하는 역할을 합니다.

import rclpy
from rclpy.node import Node
from iiot_interfaces.msg import SensorData

class DataProcessor(Node):
    def __init__(self):
        super().__init__('data_processor_node')

        self.subscription = self.create_subscription(
            SensorData,
            '/machine_data',
            self.process_data,
            10
        )
        self.get_logger().info('데이터 프로세서 노드가 구독을 시작했습니다.')

    def process_data(self, msg):
        if msg.status_code != 0 or msg.temperature > 80.0:
            self.get_logger().warn(f'🚨 경고! [{msg.device_id}] 온도: {msg.temperature}°C (초과)')
        else:
            self.get_logger().info(f'정상: [{msg.device_id}] 온도: {msg.temperature}°C')

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

if __name__ == '__main__':
    main()

3.4 setup.py Entry Point 등록

두 파이썬 스크립트가 실행될 수 있도록 setup.py에 등록하고 워크스페이스를 다시 colcon build 합니다.

entry_points={
    'console_scripts': [
        'sensor_pub = data_pipeline.sensor_pub:main',
        'data_sub = data_pipeline.data_sub:main',
    ],
},

4. Troubleshooting 및 디버깅 방법

개발 환경에서는 코드가 완벽해 보여도 네트워크 격리 문제, 메시지 타입 불일치 등으로 데이터가 흐르지 않는 경우가 흔합니다. ROS2는 이를 추적할 수 있는 강력한 CLI도구를 제공합니다.

터미널을 여러 개 열어 각 노드를 실행(ros2 run data_pipeline sensor_pub, ros2 run data_pipeline data_sub)해 둔 상태에서 아래의 툴들을 활용해 시스템을 진단합니다.

4.1 노드 목록 및 정보 확인

- ros2 node list: 현재 네트워크에 어떤 노드들이 살아있는지 확인하는 명령어입니다.

$ ros2 node list
/sensor_publisher_node
/data_processor_node

- ros2 node info: 특정 노드가 어떤 토픽을 Publish/Subscribe 하고 있는지 상세 정보를 보는 명령어입니다. 노드가 예상치 못한 동작을 할 때 가장 먼저 확인해야합니다.

$ ros2 node info /sensor_publisher_node
/sensor_publisher_node
  Publishers:
    /machine_data: iiot_interfaces/msg/SensorData
  Subscribers:
    ...

4.2 토픽 상태 확인

- ros2 topic list: 데이터를 보내기 위한 토픽이 생성되었는지 확인하는 명령어입니다. (-t 옵션: 메시지 타입 확인 가능)

$ ros2 topic list -t
/machine_data [iiot_interfaces/msg/SensorData]
/parameter_events [rcl_interfaces/msg/ParameterEvent]

- ros2 topic info: 해당 토픽에 몇 개의 퍼블리셔와 서브스크라이버가 붙어있는지 확인하는 명령어입니다. 만약 데이터가 안 온다면, 여기서 카운트가 0으로 표시되는지 확인해야 합니다.

$ ros2 topic info /machine_data
Type: iiot_interfaces/msg/SensorData
Publisher count: 1
Subscription count: 1

4.3 데이터 수신 확인

- ros2 topic echo: Subscriber 코드를 짜기 전에, 현재 토픽에 실제 데이터가 제대로 흘러가고 있는지 Raw 데이터 상태를 볼 수 있는 명령어입니다.

$ ros2 topic echo /machine_data
device_id: CNC_Machine_01
status_code: 0
temperature: 65.42
vibration_hz: 42.11
---

4.4 다이어그램을 통한 파악

- rqt_graph: 터미널 창에 rqt_graph를 입력하면 GUI 창이 뜹니다. 현재 작동 중인 모든 노드와 토픽들의 연결 상태를 다이어그램으로 그려줍니다. 시스템 아키텍처가 의도한 대로 구성되었는지 한눈에 파악할 수 있는 가장 직관적인 도구입니다.

5. 결론: 분산 아키텍처로의 도약

지금까지 ROS2를 사용하기 위한 커스텀 메시지를 설계하고, Python으로 Pub/Sub 노드를 구현하여 디버깅하는 전체 사이클을 확인해 보았습니다.

ROS2의 핵심은 DDS 기반의 비동기 통신을 통해 시스템의 각 모듈을 독립적으로 분리하는 데 있습니다. 이를 통해 센서 데이터를 수집하는 퍼블리셔와 이를 가공하는 서브스크라이버가 서로의 상태에 얽매이지 않고 안정적으로 데이터를 주고받을 수 있습니다.

이러한 ROS2의 아키텍처는 단순히 "센서 데이터를 어떻게 보낼까?"에 대한 해답을 넘어, 향후 시스템이 비대해졌을 때 어떻게 모듈들을 안전하게 분리하고 확장할 것인지에 대한 방법을 제시합니다.

Rookie

Site footer