blog

ROS2データパイプラインの構築とデバッグ

NEXTREE.

• 6 min read
ROS2データパイプラインの構築とデバッグ

産業用 IoT やロボティクス環境でシステムを設計していると、必然的に多数のセンサーデータや制御命令をリアルタイムで処理しなければならない課題に直面します。ROS2(Robot Operating System 2)は、このような問題の解決に活用できます。

ROS2には「オペレーティングシステム(OS)」という名前が含まれていますが、WindowsやLinuxのような実際のオペレーティングシステムではなく、ロボットアプリケーションを容易に開発できるよう支援するソフトウェアフレームワークです。名前に「ロボット」と入っていますが、その本質は強力な非同期分散通信アーキテクチャです。効果的なセンサーデータ収集に活用されるROS2のデータフローを理解したうえで、これをどのように設計し、デバッグするのかを見ていきましょう。

1. ROS2通信：DDSと非同期Pub/Subパターン

ROS2アーキテクチャを理解するうえで、最初に知っておくべき概念はDDS(Data Distribution Service)です。DDSは、分散システム環境でデバイス間のデータを高速かつ安定的に交換するために作られた国際産業標準の通信ミドルウェアです。

従来の一般的なWebサービスが、主にクライアントのリクエストにサーバーが応答する同期型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ノードの実装

データ仕様の準備ができたら、実際にデータを生成・消費するPythonノードを作成する必要があります。

3.1 ロジックパッケージの作成

上で定義したデータ型を利用して、実際のデータ処理ロジックを担当する新しいPythonパッケージを作成するコマンドを実行します。

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の登録

2つのPythonスクリプトを実行できるよう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