如何从本地Python脚本获取远程ROS任务脚本的实时回调数据？

完全可行，针对你的ROS固定翼场景，推荐以下几种低延迟实时数据传输方案，延迟均可控制在几十毫秒甚至更低，满足实时性需求：

方案一：跨机器ROS节点通信（最贴合现有ROS环境） #

ROS原生支持跨主机节点通信，无需额外引入第三方库，适配你的现有开发流程：

1. 远程电脑（运行mission_script.py的机器）配置

   • 先将回调函数中的数据封装为ROS话题发布：
     • 自定义ROS消息类型（比如FixedWingStatus.msg），放置在你的ROS功能包的msg目录下：

       float64 latitude
       float64 longitude
       float64 altitude
       float64 compass_angle
       float64 velocity
       float32 battery_level
       

     • 在package.xml和CMakeLists.txt中添加消息依赖配置，编译生成消息文件
     • 修改mission_script.py，在回调函数中创建ROS发布者，将数据打包为自定义消息并发布：

       import rospy
       from your_package.msg import FixedWingStatus
       
       # 初始化ROS节点
       rospy.init_node('fixed_wing_data_publisher')
       status_pub = rospy.Publisher('/fixed_wing/status', FixedWingStatus, queue_size=10)
       
       # 回调函数示例
       def data_callback(latitude, longitude, altitude, compass_angle, velocity, battery_level):
           status_msg = FixedWingStatus()
           status_msg.latitude = latitude
           status_msg.longitude = longitude
           status_msg.altitude = altitude
           status_msg.compass_angle = compass_angle
           status_msg.velocity = velocity
           status_msg.battery_level = battery_level
           status_pub.publish(status_msg)
       

   • 配置ROS环境变量：编辑~/.bashrc，添加export ROS_MASTER_URI=http://<远程电脑IP>:11311，执行source ~/.bashrc后启动roscore和mission_script.py节点

2. 本地电脑（地面站机器）配置

   • 配置ROS环境变量：编辑~/.bashrc，添加export ROS_MASTER_URI=http://<远程电脑IP>:11311和export ROS_IP=<本地电脑IP>，执行source ~/.bashrc
   • 编写ROS订阅节点，接收远程话题数据并更新Tkinter界面：

     import rospy
     from your_package.msg import FixedWingStatus
     import tkinter as tk
     
     root = tk.Tk()
     # 初始化界面控件...
     
     def update_ui(status_msg):
         # 更新Tkinter界面的逻辑，比如更新标签文本
         lat_label.config(text=f"纬度: {status_msg.latitude}")
         # 其他控件更新...
     
     def data_subscriber():
         rospy.Subscriber('/fixed_wing/status', FixedWingStatus, update_ui)
         rospy.spin()
     
     # 用子线程运行订阅逻辑，避免阻塞Tkinter主线程
     import threading
     sub_thread = threading.Thread(target=data_subscriber)
     sub_thread.daemon = True
     sub_thread.start()
     
     root.mainloop()
     

方案二：UDP套接字直接通信（轻量无依赖） #

如果不想依赖ROS框架，可直接用Python原生套接字实现低延迟数据传输（UDP比TCP延迟更低，适合实时场景）：

1. 远程电脑（mission_script.py）端

   import socket
   import json
   
   # 创建UDP套接字
   udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
   udp_socket.bind(('0.0.0.0', 5000))  # 绑定本地所有IP和端口5000
   local_ip = '<本地电脑IP>'  # 地面站电脑的IP
   local_port = 5001
   
   # 回调函数示例
   def data_callback(latitude, longitude, altitude, compass_angle, velocity, battery_level):
       data = {
           'latitude': latitude,
           'longitude': longitude,
           'altitude': altitude,
           'compass_angle': compass_angle,
           'velocity': velocity,
           'battery_level': battery_level
       }
       # 转为JSON字符串并发送
       udp_socket.sendto(json.dumps(data).encode('utf-8'), (local_ip, local_port))
   

2. 本地电脑（地面站）端

   import socket
   import json
   import tkinter as tk
   import threading
   
   root = tk.Tk()
   lat_label = tk.Label(root, text="纬度: ")
   lat_label.pack()
   # 其他界面控件...
   
   def receive_data():
       udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
       udp_socket.bind(('0.0.0.0', 5001))
       while True:
           data, _ = udp_socket.recvfrom(1024)
           data_dict = json.loads(data.decode('utf-8'))
           # 通过after方法在主线程更新界面
           root.after(0, lambda d=data_dict: update_ui(d))
   
   def update_ui(data_dict):
       lat_label.config(text=f"纬度: {data_dict['latitude']}")
       # 其他控件更新...
   
   # 启动接收线程
   recv_thread = threading.Thread(target=receive_data)
   recv_thread.daemon = True
   recv_thread.start()
   
   root.mainloop()
   

方案三：ZeroMQ低延迟消息通信（高性能易维护） #

ZeroMQ是专门针对高性能、低延迟场景设计的消息库，支持发布-订阅模式，比原生套接字更稳定易用：

1. 远程电脑（mission_script.py）端

   import zmq
   import json
   
   # 创建发布者
   context = zmq.Context()
   publisher = context.socket(zmq.PUB)
   publisher.bind("tcp://<远程电脑IP>:5555")
   
   # 回调函数示例
   def data_callback(latitude, longitude, altitude, compass_angle, velocity, battery_level):
       data = {
           'latitude': latitude,
           'longitude': longitude,
           'altitude': altitude,
           'compass_angle': compass_angle,
           'velocity': velocity,
           'battery_level': battery_level
       }
       publisher.send_json(data)
   

2. 本地电脑（地面站）端

   import zmq
   import tkinter as tk
   import threading
   
   root = tk.Tk()
   lat_label = tk.Label(root, text="纬度: ")
   lat_label.pack()
   # 其他界面控件...
   
   def receive_data():
       context = zmq.Context()
       subscriber = context.socket(zmq.SUB)
       subscriber.connect("tcp://<远程电脑IP>:5555")
       subscriber.subscribe(b'')  # 订阅所有消息
   
       while True:
           data_dict = subscriber.recv_json()
           root.after(0, lambda d=data_dict: update_ui(d))
   
   def update_ui(data_dict):
       lat_label.config(text=f"纬度: {data_dict['latitude']}")
       # 其他控件更新...
   
   # 启动接收线程
   recv_thread = threading.Thread(target=receive_data)
   recv_thread.daemon = True
   recv_thread.start()
   
   root.mainloop()
   

注意事项 #

• 确保远程和本地电脑处于同一局域网，或通过VPN打通网络，保证IP可达
• 使用ROS方案时，需保证两台电脑的ROS版本一致，避免兼容性问题
• 套接字/ZeroMQ方案中，若追求极致性能，可将JSON序列化改为二进制序列化（如struct模块）
• Tkinter界面更新必须在主线程执行，因此需用after方法触发更新，避免线程安全问题

内容的提问来源于stack exchange，提问作者user19726663