【ROS2】トピック

ROS2トピックとは？

​トピックは、ROS2における一方向の非同期的なデータストリームです。この仕組みは「出版/購読モデル（Publish/Subscribe Model）」に基づいています。

• ​パブリッシャー (Publisher): 特定のトピックに対して、メッセージ（データ）を送信するノード（ROS2の実行プログラムの単位）。

• ​サブスクライバー (Subscriber): 特定のトピックを購読し、そこへパブリッシュされたメッセージを受信するノード。

• ​トピック (Topic): パブリッシャーとサブスクライバーを繋ぐ「チャンネル」のようなもので、名前で識別されます。同じトピック名とメッセージ型を持つノード同士が通信できます。

​このモデルの利点は、パブリッシャーとサブスクライバーが互いに存在を知る必要がない点です。パブリッシャーはただデータを送り、サブスクライバーはただデータを受け取るだけです。これにより、システム全体のモジュール性が高まり、柔軟でスケーラブルなロボットアプリケーションの構築が可能になります。

​トピック利用時の注意点

​トピックを効果的に利用するためには、いくつかの重要な点に注意する必要があります。

• ​命名規則:

  • ​明確で階層的な名前: トピック名は、そのデータが何であるかを明確に示すべきです。例えば、/robot/front_camera/image_raw のように階層構造を持たせると、データの出所が分かりやすくなります。

  • ​小文字とアンダースコア: ROSの慣習として、トピック名は小文字のスネークケース（例: laser_scan）で命名します。

  • ​グローバルとプライベート: 先頭に / を付けるとグローバルな名前になり、付けないとノードの名前空間に属します。これにより、同じノードを複数起動した際のトピック名の衝突を防げます。

• ​QoS (Quality of Service):

  • ​ROS2の大きな特徴の一つがQoS設定です。これにより、通信の信頼性、耐久性、履歴の深さなどを細かく調整できます。

  • ​信頼性 (Reliability): RELIABLE（信頼性が高いが、パフォーマンスに影響が出る可能性）と BEST_EFFORT（最善を尽くすが、メッセージの損失を許容）があります。センサーデータのような高頻度で最新の値が重要な場合は BEST_EFFORT、設定値のような必ず届いてほしいデータには RELIABLE を選択します。

  • ​パブリッシャーとサブスクライバーのQoS設定は互換性がなければ通信が確立しません。 ros2 topic info -v <トピック名> コマンドでQoSの互換性を確認できます。

• ​メッセージの型:

  • ​パブリッシャーとサブスクライバーは、全く同じメッセージ型を使用する必要があります。異なる型のメッセージを送受信しようとしても、通信は行われません。

  • ​ros2 topic info <トピック名> で、そのトピックが使用しているメッセージ型を確認できます。

​C++による使い方

​C++でトピックを利用するには、rclcppライブラリを使用します。ここでは、簡単な文字列メッセージを送受信する例を示します。

​パブリッシャー (C++)

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"
#include <chrono>

using namespace std::chrono_literals;

class MinimalPublisher : public rclcpp::Node
{
public:
  MinimalPublisher()
  : Node("minimal_publisher"), count_(0)
  {
    // "topic" という名前で std_msgs::msg::String 型のメッセージを配信
    // QoSの履歴の深さは10に設定
    publisher_ = this->create_publisher<std_msgs::msg::String>("topic", 10);
    
    // 500msごとに timer_callback を呼び出すタイマーを設定
    timer_ = this->create_wall_timer(
      500ms, std::bind(&MinimalPublisher::timer_callback, this));
  }

private:
  void timer_callback()
  {
    auto message = std_msgs::msg::String();
    message.data = "Hello, world! " + std::to_string(count_++);
    RCLCPP_INFO(this->get_logger(), "Publishing: '%s'", message.data.c_str());
    publisher_->publish(message);
  }
  rclcpp::TimerBase::SharedPtr timer_;
  rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;
  size_t count_;
};

int main(int argc, char * argv[])
{
  rclcpp::init(argc, argv);
  rclcpp::spin(std::make_shared<MinimalPublisher>());
  rclcpp::shutdown();
  return 0;
}

サブスクライバー (C++)

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"

using std::placeholders::_1;

class MinimalSubscriber : public rclcpp::Node
{
public:
  MinimalSubscriber()
  : Node("minimal_subscriber")
  {
    // "topic" という名前のトピックを購読
    // メッセージを受信すると topic_callback が呼び出される
    subscription_ = this->create_subscription<std_msgs::msg::String>(
      "topic", 10, std::bind(&MinimalSubscriber::topic_callback, this, _1));
  }

private:
  void topic_callback(const std_msgs::msg::String & msg) const
  {
    RCLCPP_INFO(this->get_logger(), "I heard: '%s'", msg.data.c_str());
  }
  rclcpp::Subscription<std_msgs::msg::String>::SharedPtr subscription_;
};

int main(int argc, char * argv[])
{
  rclcpp::init(argc, argv);
  rclcpp::spin(std::make_shared<MinimalSubscriber>());
  rclcpp::shutdown();
  return 0;
}

Pythonによる使い方

​Pythonでは、rclpyライブラリを使用します。C++と同様の機能をよりシンプルに記述できます。

​パブリッシャー (Python)

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class MinimalPublisher(Node):

    def __init__(self):
        super().__init__('minimal_publisher')
        # "topic" という名前で String 型のメッセージを配信
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello, world! {self.i}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.i += 1


def main(args=None):
    rclpy.init(args=args)
    minimal_publisher = MinimalPublisher()
    rclpy.spin(minimal_publisher)
    minimal_publisher.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

サブスクライバー (Python)

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class MinimalSubscriber(Node):

    def __init__(self):
        super().__init__('minimal_subscriber')
        # "topic" という名前のトピックを購読
        self.subscription = self.create_subscription(
            String,
            'topic',
            self.listener_callback,
            10)
        self.subscription  # prevent unused variable warning

    def listener_callback(self, msg):
        self.get_logger().info(f'I heard: "{msg.data}"')


def main(args=None):
    rclpy.init(args=args)
    minimal_subscriber = MinimalSubscriber()
    rclpy.spin(minimal_subscriber)
    minimal_subscriber.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()