Jetsonからラズパイにメッセージを届けてみる（ROS2/DDS）

シングルボードコンピュータはすべて同じ構成にしておきます。

Ubuntu + ROS2 コンテナ

以下をご参照ください。

上記の設定だと、それぞれのコンテナーは以下のコマンドで起動できます。

起動しておきます。

sudo docker start -i my_ros2

1	sudo docker start -i my_ros2

こういうネットワークでやってみます。

左上のNano(4GB)からpublishしますが、DOMAIN_IDは４GBとラズパイで同じにしておきます。

（DOMAIN_IDが同じでもtopic名が異なればメッセージは届きません）

ベースになるPublisherとSubscriberのコードはこんな感じです。

クライアント・ライブラリはPython版のrclpy、topic名はchatter

Publisher

参考ソースはROS2のDDSで通信テストに使った
talker.py

このソースではタイマー設定で断続的にpublishが実行されるので、１回だけpublishするようにします。

【publisher.py】

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

rclpy.init()

node = Node("talker")

pub = node.create_publisher(String, "chatter",10)

msg = String()

msg.data = 'Hello RaspberryPi'

pub.publish(msg)

node.destroy_node()

rclpy.shutdown()

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

rclpy.init()

node = Node("talker")

pub = node.create_publisher(String, "chatter",10)

msg = String()

msg.data = 'Hello RaspberryPi'

pub.publish(msg)

node.destroy_node()

rclpy.shutdown()

Subscriber

参考ソースはROS2のDDSで通信テストに使った
listener.py

単純化しました。

【subscriber.py】

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

rclpy.init()

node = Node("listener")

def chatter_callback(msg):
    print("{0}".format(msg))

node.create_subscription(String, "chatter", chatter_callback,10)

try:
    rclpy.spin(node)
except KeyboardInterrupt:
    pass

node.destroy_node()
rclpy.shutdown()

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

rclpy.init()

node = Node("listener")

def chatter_callback(msg):

print("{0}".format(msg))

node.create_subscription(String, "chatter", chatter_callback,10)

try:

rclpy.spin(node)

except KeyboardInterrupt:

pass

node.destroy_node()

rclpy.shutdown()

PublisherとSubscriberのテスト

ラズパイとNano（２GB）で待ち受けます（DDSモデルではこの待ち受けるというイメージは少し違う感じもします….間違っているわけではないけど正しくもない、言葉どおりなら「購読する」ですがこれでは分かってる人以外は分からない、お金の絡まないサブスク？）。

ラズパイではDOMAIN_IDは10にしておきます。

export ROS_DOMAIN_ID=10
python3 subscriber.py

1 2	export ROS_DOMAIN_ID=10 python3 subscriber.py

Nano（２GB）ではDOMAIN_IDは11にしておきます。

export ROS_DOMAIN_ID=11
python3 subscriber.py

1 2	export ROS_DOMAIN_ID=11 python3 subscriber.py

Nano（4GB）でDOMAIN_IDは10でPublishします。

export ROS_DOMAIN_ID=10
python3 publisher.py

1 2	export ROS_DOMAIN_ID=10 python3 publisher.py

DOMAIN_IDが同じラズパイではメッセージを受け取っています。

DOMAIN_IDの異なるJetson (２GB) ではメッセージは届きません（というか、スルー／無視しています）。

publishのテストはros2コマンドラインインターフェースからも実行できます。

まず先にPythonコンソールなどでpublisherの以下の部分を実行しておきます。

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
rclpy.init()
node = Node("talker")
pub = node.create_publisher(String, "chatter",10)
msg = String()

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

rclpy.init()

node = Node("talker")

pub = node.create_publisher(String, "chatter",10)

msg = String()

Subscriber として以下を実行

ros2 topic echo /chatter

1	ros2 topic echo /chatter

publisherでメッセージを送ります。

msg.data = "Hello";pub.publish(msg)

1	msg.data = "Hello";pub.publish(msg)

Subscriber側でメッセージが標準出力されます。

Publisherを後始末して終了

node.destroy_node()
rclpy.shutdown()ど

1 2	node.destroy_node() rclpy.shutdown()ど

同様にsubscribeのテストを実行する場合

メッセージをpublishするROS2コマンドはこんな感じ

ros2 topic pub /chatter std_msgs/String "data: Hello world"

1	ros2 topic pub /chatter std_msgs/String "data: Hello world"

Jetson からラズパイにメッセージとして「角度」を送ってサーボモーターのホーンを回転させます。

ラズパイでサーボモーターを制御するのはこのページをご参照ください

Suscriberのコードはこんな感じ

【subscriber_sg90.py】

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
import RPi.GPIO as GPIO
import time

PINNUMBER = 14 #GPIO14
FREQUENCE = 50 #Hz

GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(PINNUMBER, GPIO.OUT)
servo = GPIO.PWM(PINNUMBER, FREQUENCE)

rclpy.init()
#ノード名、便宜上listenerになっていますが、分かりやすく「servo_controller」などにしてもいいです
node = Node("listener")

def ServoSetCtrl(deg):
    #ホーンの位置を角度で設定
    duty = (12 - 2.5)/180 * deg + 2.5 + 0
    servo.start(duty)

def ServoCleanup():
    ServoSetCtrl(0)
    time.sleep(1)
    GPIO.setup(PINNUMBER, GPIO.IN)

def chatter_callback(msg):
    #メッセージ（角度）を取得
    mes = "{0}".format(msg)
    print(mes)
    ServoSetCtrl(int(msg.data))

node.create_subscription(String, "chatter", chatter_callback,10)

#initial position
ServoSetCtrl(0)

print("start")

try:
    rclpy.spin(node)
except KeyboardInterrupt:
    pass
finally:
    ServoCleanup()
    GPIO.cleanup()
    node.destroy_node()
    rclpy.shutdown()

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

import RPi.GPIO as GPIO

import time

PINNUMBER = 14 #GPIO14

FREQUENCE = 50 #Hz

GPIO.setwarnings(False)

GPIO.setmode(GPIO.BCM)

GPIO.setup(PINNUMBER, GPIO.OUT)

servo = GPIO.PWM(PINNUMBER, FREQUENCE)

rclpy.init()

#ノード名、便宜上listenerになっていますが、分かりやすく「servo_controller」などにしてもいいです

node = Node("listener")

def ServoSetCtrl(deg):

#ホーンの位置を角度で設定

duty = (12 - 2.5)/180 * deg + 2.5 + 0

servo.start(duty)

def ServoCleanup():

ServoSetCtrl(0)

time.sleep(1)

GPIO.setup(PINNUMBER, GPIO.IN)

def chatter_callback(msg):

#メッセージ（角度）を取得

mes = "{0}".format(msg)

print(mes)

ServoSetCtrl(int(msg.data))

node.create_subscription(String, "chatter", chatter_callback,10)

#initial position

ServoSetCtrl(0)

print("start")

try:

rclpy.spin(node)

except KeyboardInterrupt:

pass

finally:

ServoCleanup()

GPIO.cleanup()

node.destroy_node()

rclpy.shutdown()

Publisher のコード

【publisher_sg90.py】

引数で角度を渡します（０～１８０）

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

args = sys.argv
if len(args) < 2:
    deg = "0"
else:
    deg = sys.argv[1]

rclpy.init()

node = Node("talker")

pub = node.create_publisher(String, "chatter",10)

msg = String()
#引数（角度）をメッセージにセット
msg.data = deg
#publishを実行
pub.publish(msg)

node.destroy_node()

rclpy.shutdown()

import rclpy

from rclpy.node import Node

from std_msgs.msg import String

import sys

args = sys.argv

if len(args) < 2:

deg = "0"

else:

deg = sys.argv[1]

rclpy.init()

node = Node("talker")

pub = node.create_publisher(String, "chatter",10)

msg = String()

#引数（角度）をメッセージにセット

msg.data = deg

#publishを実行

pub.publish(msg)

node.destroy_node()

rclpy.shutdown()

まず、ラズパイ側でメッセージを待ちます。

export ROS_DOMAIN_ID=10
python3 subscriber_sg90.py

1 2	export ROS_DOMAIN_ID=10 python3 subscriber_sg90.py

Jetson からメッセージとして角度を送ります。

export ROS_DOMAIN_ID=10
python3 publisher_sg90.py 180

1 2	export ROS_DOMAIN_ID=10 python3 publisher_sg90.py 180

如何でしょう。モーターのギア音やらジッター現象の振動音でうるさいですが、ホーンがこんなふうに回転すると思います。

Appendix

ネットワーク上にどういうnodeがあるの？

ros2 node list

1	ros2 node list

ネットワーク上にどういうtopicがあるの？

ros2 topic list

1	ros2 topic list

FRONT

地図と画像のサイト

Jetsonからラズパイにメッセージを届けてみる（ROS2/DDS）

Be the first to comment

Leave a Reply コメントをキャンセル