【ROS2講座③】パッケージの作成とビルド【Python】

mkdir -p ~/ros2_ws/src

cd ~/ros2_ws

git clone https://github.com/ros2/examples src/examples -b humble

source /opt/ros/humble/setup.bash

colcon build --symlink-install

colcon build --executor sequential

ros2_ws/
├── src/          ← ソースコード
├── build/        ← 中間ファイル（コンパイル結果）
├── install/      ← 実行に必要なファイル（実行バイナリなど）
├── log/          ← ビルド時のログ情報

source install/setup.bash

ros2 run examples_rclcpp_minimal_subscriber subscriber_member_function

ros2 run examples_rclcpp_minimal_publisher publisher_member_function

codium --no-sandbox --disable-gpu

cd src

ros2 pkg create --build-type ament_python --license Apache-2.0 py_pubsub

# rclpy は ROS 2 の Python クライアントライブラリ
import rclpy
# Node クラスは ROS 2 ノードを作るための基本クラス
from rclpy.node import Node

# std_msgs の String 型メッセージを使う
from std_msgs.msg import String

# MinimalPublisher クラス：ROSノードの本体となるクラス
class MinimalPublisher(Node):

    def __init__(self):
        # 親クラス（Node）を初期化。ノード名は "minimal_publisher"
        super().__init__('minimal_publisher')

        # パブリッシャーを作成。型は String、トピック名は 'topic'、キューサイズは10
        self.publisher_ = self.create_publisher(String, 'topic', 10)

        # 0.5秒ごとに timer_callback() を呼び出すタイマーを設定
        timer_period = 0.5  # 秒単位
        self.timer = self.create_timer(timer_period, self.timer_callback)

        # カウント用の変数 i（メッセージに含める番号）
        self.i = 0

    # タイマーで定期的に呼び出される関数
    def timer_callback(self):
        # メッセージオブジェクトを作成（型はString）
        msg = String()
        # メッセージの中身を "Hello World: i" という文字列に設定
        msg.data = 'Hello World: %d' % self.i

        # メッセージをパブリッシュ（送信）
        self.publisher_.publish(msg)

        # ログとして送信したメッセージ内容を表示（画面に出力）
        self.get_logger().info('Publishing: "%s"' % msg.data)

        # カウントアップ（次回メッセージ用）
        self.i += 1

# ノードを起動する main 関数
def main(args=None):
    # ROS 2 の Pythonライブラリを初期化（ノードなどを使う準備）
    rclpy.init(args=args)

    # MinimalPublisher ノードのインスタンスを作成
    minimal_publisher = MinimalPublisher()

    # ノードを実行状態にする（Ctrl+C で止めるまでループ）
    rclpy.spin(minimal_publisher)

    # ノードを明示的に破棄（メモリ解放・終了処理）
    minimal_publisher.destroy_node()
    # ROS 2 をシャットダウン（終了処理）
    rclpy.shutdown()

# スクリプトが直接実行されたとき、main() を呼び出す
if __name__ == '__main__':
    main()

# rclpy は ROS 2 の Python クライアントライブラリ
import rclpy
# Node クラスは ROS 2 ノードを定義するための基本クラス
from rclpy.node import Node

# std_msgs パッケージから、文字列メッセージ型（String）をインポート
from std_msgs.msg import String


# MinimalSubscriber クラス：ROS ノードの受信側ノード（サブスクライバー）を定義
class MinimalSubscriber(Node):

    def __init__(self):
        # 親クラス Node を初期化。ノード名は 'minimal_subscriber'
        super().__init__('minimal_subscriber')

        # サブスクリプション（購読）を作成
        # 'topic' という名前のトピックから String 型メッセージを購読
        # メッセージを受信すると listener_callback 関数が呼ばれる
        self.subscription = self.create_subscription(
            String,              # 受け取るメッセージ型（std_msgs.msg.String）
            'topic',             # 購読するトピック名
            self.listener_callback,  # コールバック関数（受信時に実行）
            10)                  # キューサイズ（メッセージがたまる最大数）

        # 変数を使わないと警告が出るので、それを回避するためのダミー参照
        self.subscription

    # メッセージを受け取ったときに呼ばれる関数（コールバック）
    def listener_callback(self, msg):
        # msg.data に入っている文字列をログ出力（ターミナルに表示）
        self.get_logger().info('I heard: "%s"' % msg.data)


# main 関数：このノードを実行するときのエントリーポイント
def main(args=None):
    # rclpy の初期化（ROS 2 通信を開始できるようにする）
    rclpy.init(args=args)

    # MinimalSubscriber ノードのインスタンスを作成
    minimal_subscriber = MinimalSubscriber()

    # ノードを実行状態にして、メッセージを受信し続ける（Ctrl+Cで停止）
    rclpy.spin(minimal_subscriber)

    # ノードを明示的に破棄（省略可能だが、きれいに終了させたい場合は書く）
    minimal_subscriber.destroy_node()
    # ROS 2 の通信をシャットダウン
    rclpy.shutdown()


# このスクリプトが直接実行された場合、main() を呼び出す
if __name__ == '__main__':
    main()

<exec_depend>rclpy</exec_depend>
<exec_depend>std_msgs</exec_depend>

entry_points={
    'console_scripts': [
        'publisher = py_pubsub.publisher_member_function:main',
        'subscriber = py_pubsub.subscriber_member_function:main',
    ],
},

cd ~/ros2_ws

colcon build --packages-select py_pubsub

source install/setup.bash

ros2 run py_pubsub publisher

source install/setup.bash

ros2 run py_pubsub subscriber

from std_msgs.msg import Int32

msg = Int32()
msg.data = 42  # 任意の整数値を代入

self.publisher_ = self.create_publisher(Int32, 'count', 10)

self.subscription = self.create_subscription(
    Int32,
    'count',
    self.listener_callback,
    10)