ROS 2 Humble Hawksbill 发布消息或速度

zhangrelay

发布于 2022-06-01 10:05:20

1K0

创建节点，这些节点以字符串消息/速度的形式通过主题发布。这里使用的例子是一个简单的发布系统；发布字符串或速度。

一个简单的hello ros如下所示：

import rclpy
from rclpy.node import Node

from std_msgs.msg import String


class MinimalPublisher(Node):

    def __init__(self):
        super().__init__('minimal_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = 'Hello ROS2 World: %d' % self.i
        self.publisher_.publish(msg)
        self.get_logger().info('Publishing: "%s"' % msg.data)
        self.i += 1


def main(args=None):
    rclpy.init(args=args)

    minimal_publisher = MinimalPublisher()

    rclpy.spin(minimal_publisher)

    # Destroy the node explicitly
    # (optional - otherwise it will be done automatically
    # when the garbage collector destroys the node object)
    minimal_publisher.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

注释后的第一行代码 import rclpy 以便可以使用它的 Node 类。

import rclpy from rclpy.node import Node

下一条语句导入节点用于构造其在主题上传递的数据的内置字符串消息类型。

from std_msgs.msg import String

这些线代表节点的依赖关系。回想一下，必须将依赖项添加到 package.xml

接下来，创建 MinimalPublisher 类，它继承自 Node（或者是 Node 的子类）。

class MinimalPublisher(Node):

以下是类的构造函数的定义。 super().__init__ 调用 Node 类的构造函数并给它你的节点名称，在本例中为 minimum_publisher。

create_publisher 声明节点通过名为 topic 的主题发布字符串类型的消息（从 std_msgs.msg 模块导入），并且“队列大小”为 10。队列大小是必需的 QoS（服务质量）设置，用于限制如果订阅者没有足够快地接收它们，则排队消息的数量。

接下来，创建一个带有回调的计时器，每 0.5 秒执行一次。 self.i 是回调中使用的计数器。

def __init__(self): super().__init__('minimal_publisher') self.publisher_ = self.create_publisher(String, 'topic', 10) timer_period = 0.5 # seconds self.timer = self.create_timer(timer_period, self.timer_callback) self.i = 0

timer_callback 创建一个附加了计数器值的消息，并使用 get_logger().info 将其发布到控制台。

def timer_callback(self): msg = String() msg.data = 'Hello ROS2 World: %d' % self.i self.publisher_.publish(msg) self.get_logger().info('Publishing: "%s"' % msg.data) self.i += 1

最后，定义了 main 函数。

def main(args=None): rclpy.init(args=args) minimal_publisher = MinimalPublisher() rclpy.spin(minimal_publisher) # 显式销毁节点 # (可选 - 否则会自动完成 # 当垃圾收集器销毁节点对象时） minimum_publisher.destroy_node() rclpy.shutdown()

首先初始化 rclpy 库，然后创建节点，然后它“spins”节点，以便调用其回调。

将其逐句更新为发布cmd_vel:

import rclpy
from rclpy.node import Node

# from std_msgs.msg import String
from geometry_msgs.msg import Twist



class MinimalPublisher(Node):

    def __init__(self):
        super().__init__('my_pub')
        self.publisher_ = self.create_publisher(Twist, 'turtle1/cmd_vel', 10)
        timer_period = 1  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        vel = Twist()
#        vel.data = 'My Hello ROS2 World: %d' % self.i
        vel.linear.x=1.0
        vel.angular.z=0.01*self.i
        self.publisher_.publish(vel)
        self.get_logger().info('Publishing: x"%f"' % vel.linear.x)
        self.get_logger().info('Publishing: z"%f"' % vel.angular.z)
        self.i += 1


def main(args=None):
    rclpy.init(args=args)

    minimal_publisher = MinimalPublisher()

    rclpy.spin(minimal_publisher)

    # Destroy the node explicitly
    # (optional - otherwise it will be done automatically
    # when the garbage collector destroys the node object)
    minimal_publisher.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

是不是非常简单。

测试一下效果：