topic通信の作り方(Python)

ROS1のようにtopicを使った通信のプログラムを作る方法を紹介します。

Publisherの作成

はじめにデータを送信するpublisherというのものを作ります。

まずは下記コマンドで新しくパッケージを作ります。

一応テストなので~/ros2_ws/src/testsというディレクトリを作っておきます。

$ mkdir -p ~/ros2_ws/src/tests

$ cd ~/ros2_ws/src/tests

$ ros2 pkg create --build-type ament_python --node-name publisher_node mypublisher

--build-type でpythonを使うので「ament_python」を指定します。

--node-nameで自分の好きなパッケージ名とノード名を書きます。最初にノード名、スペースを空けてからパッケージ名を記載します。

今回は最初にtopicを発信するノードを作るので、ノード名を「publisher_node」、パッケージ名を「mypublisher」としました。

mypublisherというディレクトリができているはずなので移動します。

$ cd mypublisher

$ ls 

上記のlsを入力するとディレクトリ内のファイルとディレクトリの一覧が出てきます。

修正するのは、setup.pyとpublisherディレクトリ内のpublisher_node.pyです。

まず、setup.pyから修正しましょう。geditというテキストエディタを下記で使っていますが好きなエディタで修正してください。

$ gedit setup.py

上記を入力すると下記のようなソースを確認できると思います。

修正するところだけ「#」の後ろにコメントを加えています。

from setuptools import setup

package_name = 'mypublisher' # パッケージ名は自動で入力されている。

setup(

    name=package_name,

    version='0.0.0', # バージョンを変えたいときはここの数字を変更する

    packages=[package_name],

    data_files=[

        ('share/ament_index/resource_index/packages',

            ['resource/' + package_name]),

        ('share/' + package_name, ['package.xml']),

    ],

    install_requires=['setuptools'],

    zip_safe=True,

    maintainer='takubo', # 保守する人の名前

    maintainer_email='takubo@***.ac.jp', # 保守する人のメールアドレス

    description='TODO: Package description', # パッケージの説明

    license='TODO: License declaration', # ライセンス形態

    tests_require=['pytest'],

    entry_points={ # プログラムの開始場所の指定

        'console_scripts': [

            'publisher_node = mypublisher.publisher_node:main' # 開始場所がここに自動的に入っている。publisher_node.pyのmainが最初と指定されている。他のノードファイルをパッケージに手動で増やしたときは、次の行に追加する必要がある。

        ],

    },

)

次に、publisher_node.pyを編集します。

$ gedit ~/ros2_ws/src/tests/mypublisher/mypublisher/publisher_node.py

開くと下記のようにほとんど何も書いていないので書き換えます。

def main():

    print('Hi from publisher.')

if __name__ == '__main__':

    main()

追記した後の内容は下記のとおりです。

#publihse_node.pyの内容

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

class Publisher(Node):

    def __init__(self):

        super().__init__('mypublisher')

        self.publisher_ = self.create_publisher(String, 'topic', 10) # topicというトピック名でデータを送信。バッファサイズは10。

        timer_period = 0.5  # 0.5秒ごとに送信する設定

        self.timer = self.create_timer(timer_period, self.timer_callback)

        self.i = 0

    def timer_callback(self):

        msg = String()

        msg.data = 'Hello World: %d' % self.i

        self.publisher_.publish(msg) # 送信

        self.get_logger().info('Publishing: "%s"' % msg.data) # 送信したデータをコンソールに表示

        self.i += 1

def main(args=None):

    print('送信開始')

    rclpy.init(args=args) # 初期化

    mypublisher = Publisher() # ノードの生成

    rclpy.spin(mypublisher) # ノードの処理開始

    mypublisher.destroy_node() # ノードの破壊

    rclpy.shutdown() # ノードの終了

if __name__ == '__main__':

    main()

書き終わって保存したらコンパイルして実行してみる。

$ cd ~/ros2_ws

$ colcon build

$ source ~/ros2_ws/install/setup.bash

$ ros2 run mypublisher publisher_node

最後の行を実行すると"Hello World: 数字"のメッセージが0.5秒ごとに表示されるはず。

これだけだと送信しているだけで情報が通信できているかわからないので、受信側のプログラムを作る。