紧急通知:禁止一切关于政治&VPN翻墙等话题,发现相关帖子会立马删除封号
小鱼 ROS 2 新书上线!点击链接查看, 新书配套视频点击链接查看。
提问前必看的发帖注意事项—— 提问前必看!不符合要求的问题拒绝回答!!
社区使用指南—如何添加标签修改密码
小鱼 ROS 2 新书上线!点击链接查看, 新书配套视频点击链接查看。
提问前必看的发帖注意事项—— 提问前必看!不符合要求的问题拒绝回答!!
社区使用指南—如何添加标签修改密码
ROS2 一个节点能同时发布和订阅多个话题吗?
-
如题,在用jetson nano + pico做个机器人,过程中发现 一个节点中无法同时发布和订阅不同的话题,代码 如下:
pico_info这个话题收不到回调,这个pico_info发布时成功了,没报错的import rclpy
import jsonfrom rclpy.node import Node
from std_msgs.msg import String
from controller.service import init_server
from controller.run_data import *class ControllerNode(Node):
def __init__(self): super().__init__('controller_Node') self.command_publisher = self.create_publisher(String, 'controller', 10) self.create_subscription(String, 'pico_info', self.pico_info_callback, 6) init_server(self.command_publisher) def pico_info_callback(self, msg): global voltage12v print('Recive message: "%s"' % msg.data) # msg.data = '{"target": "jarvis", "voltage12v": 11.32}' rec_json = json.loads(str(msg.data)) voltage12v = int(rec_json['voltage12v']) set_voltage12v(voltage12v)
def main(args=None):
rclpy.init(args=args) controller_Node = ControllerNode() rclpy.spin(controller_Node) controller_Node.destroy_node() rclpy.shutdown()
if name == 'main':
main() -
@7493416 有没有可能没有发布
-
@小鱼 感谢大佬的回复,大佬的意思就是说ROS2是支持同时发布和订阅多个话题,是吧?是的话,我再排查一下消息是否发布成功;以下是发布消息节点的发布相关代码
class LowerNode(Node): def __init__(self): super().__init__('motion_node') self.pico_publisher = self.create_publisher(String, 'pico_info', 10) set_pico_publisher(self.pico_publisher) self.create_subscription(String, 'recognition_obj', self.recognition_obj_callback, 10) self.create_subscription(String, 'controller', self.controller_callback, 10) uart_init() _thread.start_new_thread(recv_data, (self.pico_publisher,))
import serial as serial import time import random from motion.utils import command_util from std_msgs.msg import String from motion.run_data import * my_uart = serial.Serial(port="/dev/ttyTHS1", baudrate=115200) def uart_init(): if not my_uart.is_open: my_uart.open() def send_data(s_data): s_data = s_data + "\n" print('Send data: ' + s_data) my_uart.write(s_data.encode()) time.sleep(0.3) def recv_data(pico_publisher): while True: r_data = my_uart.readline().decode() print('recv data: ' + r_data) if len(r_data) > 0: s_msg = String() # 要发布的消息类型 s_msg.data = str(r_data) pico_publisher.publish(s_msg) time.sleep(0.2)
-
话题应该是发布成功了的。难道是接受问题?
-
@7493416 肯定支持的,是否发布成功用 ros2 topic echo 检查