ROS2-话题学习

强烈推荐教程：

《ROS 2机器人开发从入门到实践》3.2.2订阅小说并合成语音_哔哩哔哩_bilibili

构建功能包

# create package demo_python_pkg

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

 

自己写的代码放在./demo_python_pkg/demo_python_pkg目录下

发布者

import rclpy
from rclpy.node import Node
import requests
from example_interfaces.msg import String
from queue import Queueclass NovelPubNode(Node):def __init__(self, node_name):super().__init__(node_name)self.novels_queue = Queue()self.get_logger().info("NovelPubNode has been started")self.novel_pub = self.create_publisher(String, "novel_topic", 10)self.timer = self.create_timer(5,self.timer_callback)def timer_callback(self):if not self.novels_queue.empty():line = self.novels_queue.get()msg = String()msg.data = lineself.novel_pub.publish(msg)self.get_logger().info(f"Published novel: {line}")       def download_novel(self, url):response = requests.get(url)response.encoding = 'utf-8'text = response.textfor text_line in text.splitlines():self.novels_queue.put(text_line)self.get_logger().info(f"Downloaded novel: {text}")def main():rclpy.init()node = NovelPubNode("novel_pub_node")node.download_novel("http://0.0.0.0:8000/novel_1.txt")rclpy.spin(node)node.shutdown()

 

订阅者

import time
import espeakng
import rclpy
from rclpy.node import Node
from example_interfaces.msg import String
from queue import Queue
import threadingclass NovelSubNode(Node):def __init__(self, node_name):super().__init__(node_name)self.get_logger().info("NovelSubNode has been created!")self.novels_queue = Queue()self.create_subscription(String, "novel_topic", self.novel_callback, 10)self.say_thread = threading.Thread(target=self.say)self.say_thread.start()def novel_callback(self, msg):self.novels_queue.put(msg.data)def say(self):engine = espeakng.Speaker()engine.voice = "zh"while rclpy.ok():if not self.novels_queue.empty():novel = self.novels_queue.get()engine.say(novel)print("Said: " + novel)engine.wait()else:time.sleep(1)def main():rclpy.init()node = NovelSubNode("novel_sub_node")rclpy.spin(node)rclpy.shutdown()

代码完成后，配置setup.py文件

格式为："名字 = 包名.文件名:函数名"

在根目录运行以下终端命令

# build package demo_python_pkg

colcon build

# source setup.bash

source install/setup.bash

运行以上命令后，得到build、install、log文件夹

可执行的节点文件在以下文件夹

./install/demo_python_pkg/lib/demo_python_pkg

运行节点命令

ros2 run demo_python_pkg python_pub_node

ros2 run demo_python_pkg python_sub_node

 

其他常用命令

# check if node is running

ros2 node list

# check if topic is published

ros2 topic list

# check topic content

ros2 topic echo /novel_topic

# check topic speed

ros2 topic hz /novel_topic

# check if service is available

ros2 service list