當前位置：首頁 > 编程语言 > python >内容正文

python

ROS 创建msg和srv 编写发布者和订阅者节点编写服务端和客户端节点(python版本)

發布時間：2025/3/12 python 20 豆豆

生活随笔收集整理的這篇文章主要介紹了 ROS 创建msg和srv 编写发布者和订阅者节点编写服务端和客户端节点(python版本) 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

ROS 創建msg和srv 編寫發布者和訂閱者節點編寫服務端和客戶端節點-python版本

- - rosed
  - msg和srv
  - 創建msg
  - 使用rosmsg
  - 創建srv
  - 使用rossrv
  - 重新make一下軟件包
  - 編寫發布者節點
  - 發布者節點代碼解析
  - 編寫訂閱者節點
  - 訂閱者節點代碼解析
  - 構建節點
  - 運行發布者和訂閱者節點
  - 編寫服務節點
  - 編寫客戶端節點
  - 運行服務端和客戶端節點

rosed

利用它可以直接通過軟件包名編輯包中的文件，而無需鍵入完整路徑。

$ rosed roscpp Logger.msg

編輯roscpp軟件包中的Logger.msg文件。退出vim，按下鍵盤上的Esc，然后分別按下:q!

Tap補全，在不知道準確文件名的情況下，你也可以輕松地查看和編輯包中的所有文件。

$ rosed roscpp <tab><tab> jym@ubuntu:~$ rosed roscpp Empty.srv roscpp.cmake genmsg_cpp.py roscppConfig.cmake gensrv_cpp.py roscppConfig-version.cmake GetLoggers.srv roscpp-msg-extras.cmake Logger.msg roscpp-msg-paths.cmake msg_gen.py SetLoggerLevel.srv package.xml

msg和srv

msg（消息）：msg文件就是文本文件，用于描述ROS消息的字段。它們用于為不同編程語言編寫的消息生成源代碼。

srv（服務）：一個srv文件描述一個服務。它由兩部分組成：請求（request）和響應（response）。

創建msg

在之前創建的軟件包里定義一個新的消息。直接使用roscd beginner_tutorials，會報錯。顯示：

roscd: No such package/stack 'beginner_tutorials'

因為roscd只能切換到那些路徑已經包含在ROS_PACKAGE_PATH環境變量中的軟件包。

工作空間構建完成后，要將這個工作空間添加到ROS環境中，需要source一下生成的配置文件。

然后用roscd beginner_tutorials就沒問題了。

jym@ubuntu:~$ roscd beginner_tutorials roscd: No such package/stack 'beginner_tutorials' jym@ubuntu:~$ . ~/catkin_ws/devel/setup.bash jym@ubuntu:~$ roscd beginner_tutorials jym@ubuntu:~/catkin_ws/src/beginner_tutorials$

然后輸入：

$ mkdir msg $ echo "int64 num" > msg/Num.msg

那么就在msg文件夾里面生成了Num.msg文件。

然后打開package.xml，確保有以下兩行代碼，沒有的話加上。

確保msg文件能被轉換為C++、Python和其他語言的源代碼。

<build_depend>message_generation</build_depend><exec_depend>message_runtime</exec_depend>

在構建時，只需要message_generation，而在運行時，只需要message_runtime。

輸入下面指令可以直接打開CMakeLists.txt文件

$ beginner_tutorials CMakeLists.txt

按esc后再按i，可以進入編輯。

1.在CMakeLists.txt文件中，為已經存在里面的find_package調用添加message_generation依賴項

添加message_generation后的find_package如下所示。

find_package(catkin REQUIRED COMPONENTSroscpprospystd_msgsmessage_generation )

2.還要確保導出消息的運行時依賴關系：

找到catkin_package添加CATKIN_DEPENDS message_runtime

catkin_package(...CATKIN_DEPENDS message_runtime ......)

3.找到如下代碼塊：

# add_message_files( # FILES # Message1.msg # Message2.msg # )

刪除#符號來取消注釋，然后將Message*.msg替換為你的.msg文件名，就像下邊這樣：

add_message_files(FILESNum.msg )

手動添加.msg文件后，我們要確保CMake知道何時需要重新配置項目。

4.現在必須確保generate_messages()函數被調用：

取消下面幾行的注釋：添加任意你的消息用到的包含.msg文件的軟件包（本例中為std_msgs）

# generate_messages( # DEPENDENCIES # std_msgs # )

以上是創建消息的所有步驟。

使用rosmsg

通過rosmsg show命令看看ROS能否識別創建的消息。

輸入：rosmsg show beginner_tutorials/Num

可看到：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ rosmsg show beginner_tutorials/Num int64 num

上面的例子中，消息類型包含兩部分：

beginner_tutorials – 定義消息的軟件包
Num – 消息的名稱Num

如果不記得msg在哪個包中，也可以省略包名稱。$ rosmsg show Num

可看到：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ rosmsg show Num [beginner_tutorials/Num]: int64 num

創建srv

可從另一個包復制現有的srv定義，而不是手動創建新的srv。

roscp是一個實用的命令行工具，用于將文件從一個包復制到另一個包。

從rospy_tutorials包中復制一個服務：

$ roscp rospy_tutorials AddTwoInts.srv srv/AddTwoInts.srv

剩下的和創建msg類似：

1.打開package.xml，確保有以下兩行代碼：

<build_depend>message_generation</build_depend><exec_depend>message_runtime</exec_depend>

2.在CMakeLists.txt文件中，為已經存在里面的find_package調用添加message_generation依賴項。

find_package(catkin REQUIRED COMPONENTSroscpprospystd_msgsmessage_generation )

3.在package.xml中修改服務字段。

刪除#符號來取消注釋，然后將Service*.srv替換為你的.srv文件名，就像下邊這樣：

add_service_files(FILESAddTwoInts.srv )

以上就是創建服務的所有步驟。

使用rossrv

通過rossrv show命令看看ROS能否識別創建的服務。

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ rossrv show beginner_tutorials/AddTwoInts int64 a int64 b --- int64 sum

也可以在不指定包名的情況下找到這樣的服務。

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ rossrv show AddTwoInts [rospy_tutorials/AddTwoInts]: int64 a int64 b --- int64 sum[beginner_tutorials/AddTwoInts]: int64 a int64 b --- int64 sum

第一個是剛剛在beginner_tutorials包中創建的，第二個是之前rospy_tutorials包中已經存在的。

重新make一下軟件包

現在我們已經創建了一些新消息，所以需要重新make一下軟件包：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ roscd beginner_tutorials jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ cd ../.. jym@ubuntu:~/catkin_ws$ catkin_makejym@ubuntu:~/catkin_ws$ cd - /home/jym/catkin_ws/src/beginner_tutorials jym@ubuntu:~/catkin_ws/src/beginner_tutorials$

msg目錄中的任何.msg文件都將生成所有支持語言的代碼。

C++消息的頭文件將生成在~/catkin_ws/devel/include/beginner_tutorials/。Python腳本將創建在~/catkin_ws/devel/lib/python3/dist-packages/beginner_tutorials/msg。而Lisp文件則出現在~/catkin_ws/devel/share/common-lisp/ros/beginner_tutorials/msg/。

類似地，srv目錄中的任何.srv文件都將生成支持語言的代碼。對于C++，頭文件將生成在消息的頭文件的同一目錄中。對于Python和Lisp，會在msg目錄旁邊的srv目錄中。

編寫發布者節點

“節點”是連接到ROS網絡的可執行文件。

這里將創建talker（發布者）節點，該節點將不斷廣播消息。

1.將目錄切換到創建的beginner_tutorials包中：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ roscd beginner_tutorials

2.創建一個scripts目錄來存放Python腳本：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ mkdir scripts jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ cd scripts

3.在scripts目錄下創建talker.py文件：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ vim talker.py

輸入：

#!/usr/bin/env python # license removed for brevity import rospy from std_msgs.msg import Stringdef talker():pub = rospy.Publisher('chatter', String, queue_size=10)rospy.init_node('talker', anonymous=True)rate = rospy.Rate(10) # 10hzwhile not rospy.is_shutdown():hello_str = "hello world %s" % rospy.get_time()rospy.loginfo(hello_str)pub.publish(hello_str)rate.sleep()if __name__ == '__main__':try:talker()except rospy.ROSInterruptException:pass

然后給它執行權限：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ chmod +x talker.py

可以通過下面的命令查看和編輯這個文件：

$ rosed beginner_tutorials talker.py

4.將以下內容添加到CMakeLists.txt文件

catkin_install_python(PROGRAMS scripts/talker.pyDESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION} )

確保正確安裝Python腳本，并使用合適的Python解釋器。

發布者節點代碼解析

#!/usr/bin/env python

每個Python ROS節點的最開頭都有這個聲明。第一行確保腳本作為Python腳本執行。

import rospy from std_msgs.msg import String

如果要編寫ROS節點，則需要導入rospy。std_msgs.msg的導入則是為了使我們能夠重用std_msgs/String消息類型（即一個簡單的字符串容器）來發布。

pub = rospy.Publisher('chatter', String, queue_size=10)rospy.init_node('talker', anonymous=True)

這部分代碼定義了talker與其他ROS部分的接口。

pub = rospy.Publisher("chatter", String, queue_size=10)聲明該節點正在使用String消息類型發布到chatter話題。

這里的String實際上是std_msgs.msg.String類。

queue_size參數是在ROS Hydro及更新版本中新增的，用于在訂閱者接收消息的速度不夠快的情況下，限制排隊的消息數量。

下一行的rospy.init_node(NAME, ...)非常重要，因為它把該節點的名稱告訴了rospy。

只有rospy掌握了這一信息后，才會開始與ROS主節點進行通信。

在本例中，節點使用talker名稱。注意：名稱必須是基本名稱，不能包含任何斜杠。

anonymous = True會讓名稱末尾添加隨機數，來確保節點具有唯一的名稱。

rate = rospy.Rate(10) # 10hz

此行創建一個Rate對象rate。借助其方法sleep()，它提供了一種方便的方法，來以你想要的速率循環。它的參數是10，即表示希望它每秒循環10次。

while not rospy.is_shutdown():hello_str = "hello world %s" % rospy.get_time()rospy.loginfo(hello_str)pub.publish(hello_str)rate.sleep()

這個循環是一個相當標準的rospy結構：檢查rospy.is_shutdown()標志，然后執行代碼邏輯。

查看is_shutdown()以檢查程序是否應該退出（例如有Ctrl+C或其他）。

在本例中，代碼邏輯即對public .publish(hello_str)的調用，它將一個字符串發布到chatter話題。

循環的部分還調用了rate.sleep()，它在循環中可以用剛剛好的睡眠時間維持期望的速率。

此循環還調用了rospy.loginfo(str)，它有3個任務：打印消息到屏幕上；把消息寫入節點的日志文件中；寫入rosout。

rosout是一個方便的調試工具：可以使用rqt_console來拉取消息，而不必在控制臺窗口找節點的輸出。

try:talker()except rospy.ROSInterruptException:pass

除了標準的Python __main__檢查，它還會捕獲一個rospy.ROSInterruptException異常，當按下Ctrl+C或節點因其他原因關閉時，這一異常就會被rospy.sleep()和rospy.Rate.sleep()拋出。

編寫訂閱者節點

$ roscd beginner_tutorials/scripts/

2.創建listener.py文件

$ vim listener.py

輸入：

#!/usr/bin/env python import rospy from std_msgs.msg import Stringdef callback(data):rospy.loginfo(rospy.get_caller_id() + "I heard %s", data.data)def listener():# In ROS, nodes are uniquely named. If two nodes with the same# name are launched, the previous one is kicked off. The# anonymous=True flag means that rospy will choose a unique# name for our 'listener' node so that multiple listeners can# run simultaneously.rospy.init_node('listener', anonymous=True)rospy.Subscriber("chatter", String, callback)# spin() simply keeps python from exiting until this node is stoppedrospy.spin()if __name__ == '__main__':listener()

3.給它執行權限

$ chmod +x listener.py

4.編輯你CMakeLists.txt中的catkin_install_python()調用

可以用rosed打開CMakeLists.txt

rosed beginner_tutorials CMakeLists.txt

然后添加

catkin_install_python(PROGRAMS scripts/talker.py scripts/listener.pyDESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION} )

訂閱者節點代碼解析

listener.py的代碼類似于talker.py，只不過我們為訂閱消息引入了一種新的基于回調的機制。

rospy.init_node('listener', anonymous=True)rospy.Subscriber("chatter", String, callback)# spin() simply keeps python from exiting until this node is stoppedrospy.spin()

這聲明節點訂閱了chatter話題，類型是std_msgs.msgs.String。當接收到新消息時，callback函數被調用，消息作為第一個參數。

稍微更改了對rospy.init_node()的調用。添加了anonymous=True關鍵字參數。

ROS要求每個節點都有一個唯一的名稱，如果出現具有相同名稱的節點，則會與前一個節點發生沖突，這樣一來，出現故障的節點很容易地被踢出網絡。

anonymous=True標志會告訴rospy為節點生成唯一的名稱，這樣就很容易可以有多個listener.py一起運行。

rospy.spin()只是不讓節點退出，直到節點被明確關閉。

與roscpp不同，rospy.spin()不影響訂閱者回調函數，因為它們有自己的線程。

構建節點

使用CMake作為構建系統。即使是Python節點也必須使用它。這是為了確保能為創建的消息和服務自動生成Python代碼。

回到catkin工作空間，然后運行catkin_make：

$ cd ~/catkin_ws $ catkin_make

運行發布者和訂閱者節點

開三個終端：

第一個輸入：

$ roscore

第二個：

運行talker發布者節點

$ cd ~/catkin_ws $ source ./devel/setup.bash $ rosrun beginner_tutorials talker.py

第三個：

$ cd ~/catkin_ws $ source ./devel/setup.bash $ rosrun beginner_tutorials listener.py

可以通過

$ rosrun rqt_graph rqt_graph

看到當前運行的節點和話題

發布者節點發出：

[INFO] [1635421895.494006]: hello world 1635421895.4938853 [INFO] [1635421895.594515]: hello world 1635421895.5943954 [INFO] [1635421895.694447]: hello world 1635421895.6943207 [INFO] [1635421895.793802]: hello world 1635421895.7936826 [INFO] [1635421895.893810]: hello world 1635421895.893687 [INFO] [1635421895.993590]: hello world 1635421895.9934704 [INFO] [1635421896.094563]: hello world 1635421896.0942173 [INFO] [1635421896.194204]: hello world 1635421896.1938524 [INFO] [1635421896.294536]: hello world 1635421896.2941887 [INFO] [1635421896.394086]: hello world 1635421896.3939602 [INFO] [1635421896.493973]: hello world 1635421896.4936652 [INFO] [1635421896.594645]: hello world 1635421896.5942914 [INFO] [1635421896.694563]: hello world 1635421896.6942132 [INFO] [1635421896.794407]: hello world 1635421896.7942781 [INFO] [1635421896.894170]: hello world 1635421896.8938253 ^Cjym@ubuntu:~/catkin_ws$

訂閱者節點收到：

[INFO] [1635421896.198918]: /listener_8151_1635421497025I heard hello world 1635421896.1938524 [INFO] [1635421896.299254]: /listener_8151_1635421497025I heard hello world 1635421896.2941887 [INFO] [1635421896.397918]: /listener_8151_1635421497025I heard hello world 1635421896.3939602 [INFO] [1635421896.498270]: /listener_8151_1635421497025I heard hello world 1635421896.4936652 [INFO] [1635421896.599628]: /listener_8151_1635421497025I heard hello world 1635421896.5942914 [INFO] [1635421896.698885]: /listener_8151_1635421497025I heard hello world 1635421896.6942132 [INFO] [1635421896.795746]: /listener_8151_1635421497025I heard hello world 1635421896.7942781 [INFO] [1635421896.898501]: /listener_8151_1635421497025I heard hello world 1635421896.8938253

可以看到在發布者節點添加ctrl+c之后，訂閱者節點也不接收消息了。

編寫服務節點

前面已經創建了需要的服務AddTwoInts.srv。

接下來：

在beginner_tutorials包中創建scripts/add_two_ints_server.py文件并粘貼以下內容進去：

#!/usr/bin/env pythonfrom __future__ import print_functionfrom beginner_tutorials.srv import AddTwoInts,AddTwoIntsResponse import rospydef handle_add_two_ints(req):print("Returning [%s + %s = %s]"%(req.a, req.b, (req.a + req.b)))return AddTwoIntsResponse(req.a + req.b)def add_two_ints_server():rospy.init_node('add_two_ints_server')s = rospy.Service('add_two_ints', AddTwoInts, handle_add_two_ints)print("Ready to add two ints.")rospy.spin()if __name__ == "__main__":add_two_ints_server()

然后將以下內容添加到CMakeLists.txt文件。確保正確安裝Python腳本，并使用合適的Python解釋器。

catkin_install_python(PROGRAMS scripts/add_two_ints_server.pyDESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION} )

以上流程在linux上輸入的指令代碼：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials$ roscd beginner_tutorials/scripts/ jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ vim add_two_ints_server.py jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ chmod +x add_two_ints_server.py jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ rosed beginner_tutorials CMakeLists.txt

使用rospy編寫服務。使用init_node()聲明我們的節點，然后再聲明我們的服務：

s = rospy.Service('add_two_ints', AddTwoInts, handle_add_two_ints)

這聲明了一個名為add_two_ints的新服務，其服務類型為AddTwoInts。所有的請求（request）都傳遞給了handle_add_two_ints函數。handle_add_two_ints被AddTwoIntsRequest的實例調用，返回AddTwoIntsResponse實例。

就像訂閱者中的例子一樣，rospy.spin()可以防止代碼在服務關閉之前退出。

編寫客戶端節點

在beginner_tutorials包中創建scripts/add_two_ints_client.py文件并粘貼以下內容進去：

#!/usr/bin/env pythonfrom __future__ import print_functionimport sys import rospy from beginner_tutorials.srv import *def add_two_ints_client(x, y):rospy.wait_for_service('add_two_ints')try:add_two_ints = rospy.ServiceProxy('add_two_ints', AddTwoInts)resp1 = add_two_ints(x, y)return resp1.sumexcept rospy.ServiceException as e:print("Service call failed: %s"%e)def usage():return "%s [x y]"%sys.argv[0]if __name__ == "__main__":if len(sys.argv) == 3:x = int(sys.argv[1])y = int(sys.argv[2])else:print(usage())sys.exit(1)print("Requesting %s+%s"%(x, y))print("%s + %s = %s"%(x, y, add_two_ints_client(x, y)))

同樣將以下內容添加到CMakeLists.txt文件。

catkin_install_python(PROGRAMS scripts/add_two_ints_server.py scripts/add_two_ints_client.pyDESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION} )

以上流程在linux上輸入的指令代碼：

jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ roscd beginner_tutorials/scripts/ jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ vim add_two_ints_client.py jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ chmod +x add_two_ints_client.py jym@ubuntu:~/catkin_ws/src/beginner_tutorials/scripts$ rosed beginner_tutorials CMakeLists.txt

客戶端（用來調用服務）的代碼分析：

對于客戶端來說不需要調用init_node()。首先調用：

rospy.wait_for_service('add_two_ints')

這是一種很方便的方法，可以讓在add_two_ints服務可用之前一直阻塞。

add_two_ints = rospy.ServiceProxy('add_two_ints', AddTwoInts)

這里為服務的調用創建了一個句柄（handle）。

resp1 = add_two_ints(x, y)return resp1.sum

然后可以使用這個句柄，就像普通的函數一樣調用它。

因為已經將服務的類型聲明為AddTwoInts，它會生成AddTwoIntsRequest對象 (you’re free to pass in your own instead)。如果調用失敗，rospy.ServiceException將會拋出，所以應該弄一個合適的try/except部分。

使用CMake作為構建系統。為了確保能為創建的消息和服務自動生成Python代碼。

切換當前目錄到catkin工作空間，然后運行catkin_make：

$ cd ~/catkin_ws $ catkin_make

現在已經編寫了一個簡單的服務和客戶端

運行服務端和客戶端節點