通过 WebSocket 实时获取 ROS 机器人数据

John Doe Lv1
  2025-07-25 19:31:20 2025-07-25 19:31:20 Created   2025-07-27 13:06:23 2025-07-27 13:06:23 Updated   

从零开始：通过 WebSocket 实时获取 ROS 机器人数据 (小白终极教程)

前言

本教程旨在帮助初学者解决一个常见的需求：如何从一台运行ROS（机器人操作系统）的机器人上，将实时数据（例如IMU、里程计、摄像头等）传输到另一台普通的电脑上，并用Python等通用编程语言进行处理。

我们将使用rosbridge这个强大的工具，它就像一座桥梁，将ROS复杂的通信协议转换成大家所熟知的WebSocket协议，让非ROS程序也能轻松与机器人交互。

核心架构

在开始之前，我们先理清整个数据流动的路径，这会让你对接下来的步骤有更清晰的认识：

1
2
3
机器人端 (数据发送方) → 客户端 (数据接收方)

机器人节点 (如/base_driver) → 发布 /imu 话题 → ROS Master (核心) → rosbridge_server (运行在机器人上) → 局域网 (Wi-Fi) → 你的电脑 (运行Python脚本) → 成功接收数据

这里的关键点是：rosbridge_server必须和你的机器人节点运行在同一台设备上！

第一部分：机器人端配置 (以 IP: 192.168.9.32 为例)

这部分的所有操作都在你的机器人或装有ROS的主控电脑上进行。

第1步：确定并记录机器人的IP地址

这是所有网络通信的基础。打开一个终端，输入以下命令之一：

1
2
3
ifconfig
# 或者
ip addr

在输出的信息中找到你的主网络接口（通常是wlan0, wlp3s0等无线网卡，或是eth0, enp4s0等有线网卡），记下它的inet地址。在本案例中，我们确定了机器人IP是192.168.9.32。

第2步：安装rosbridge_server

如果你的机器人上没有安装过，请在终端中运行：

1
2
sudo apt-get update
sudo apt-get install ros-noetic-rosbridge-server

(请将noetic替换成你对应的ROS版本，如melodic)

第3步：启动并验证数据源（IMU话题）

确保你的机器人底盘驱动或传感器节点已经正常运行。这是数据的源头，必须先启动。

启动机器人节点 (请替换成你自己的命令):

1
roslaunch my_robot_control robot_start.launch

验证 /imu 话题是否正常发布 (非常关键的检查步骤):
打开一个新的终端，输入：

1
rostopic info /imu

如果你看到类似下面的输出，特别是Publishers:部分不为空，说明数据源正常。

1
2
3
4
5
Type: sensor_msgs/Imu

Publishers: 
 * /base_driver (http://192.168.9.32:38583/)
...

第4步：配置并启动rosbridge_server

找到并编辑launch文件，将其监听的IP地址配置为机器人的IP。

1
2
3
roscd rosbridge_server/launch/
sudo cp rosbridge_websocket.launch rosbridge_websocket.launch.backup # 备份是个好习惯
sudo gedit rosbridge_websocket.launch

在打开的文件中，找到 <param name="address" ... /> 这一行，将其value修改为你在第1步中查到的机器人IP地址。

1
<param name="address" value="192.168.9.32" />

为何要这样做？
默认情况下rosbridge可能只监听本地127.0.0.1，修改为局域网IP后，网络内的其他设备才能访问到它。

启动rosbridge_server:

1
roslaunch rosbridge_server rosbridge_websocket.launch

此时，你的机器人端已经准备就绪。它正在发布 /imu 数据，并且rosbridge正在将这些数据通过 ws://192.168.9.32:9090 这个地址向全网络广播。

第二部分：客户端配置 (你的电脑)

这部分操作在你自己的电脑（例如笔记本）上进行。

第1步：准备Python环境

安装pip (如果还没有):

1
2
sudo apt-get update
sudo apt-get install python3-pip

安装websocket-client库:

1
pip3 install websocket-client

第2步：编写并配置Python脚本

在你电脑上创建一个名为test.py的文件，将以下代码完整地复制进去：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import websocket
import json
import time

def on_message(ws, message):
    """当接收到消息时此函数被调用"""
    try:
        data = json.loads(message)
        # 检查消息中是否包含 'msg' 字段，这通常是rosbridge话题数据的结构
        if "msg" in data:
            imu_data = data['msg']
            # 提取并打印线加速度数据
            if 'linear_acceleration' in imu_data:
                x = imu_data['linear_acceleration']['x']
                y = imu_data['linear_acceleration']['y']
                z = imu_data['linear_acceleration']['z']
                timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                print(f"[{timestamp}] [IMU Data] x={x:.4f}, y={y:.4f}, z={z:.4f}")
    except json.JSONDecodeError as e:
        print(f"[JSON解析错误] {e}: {message}")

def on_error(ws, error):
    """当发生错误时此函数被调用"""
    print(f"[WebSocket错误]: {error}")

def on_close(ws, close_status_code, close_msg):
    """当连接关闭时此函数被调用"""
    print("WebSocket连接已关闭")
    print(f"状态码: {close_status_code}, 消息: {close_msg}")

def on_open(ws):
    """当连接建立时此函数被调用"""
    print("WebSocket连接已建立")
    # 构建订阅消息
    subscribe_message = {
        "op": "subscribe",
        "topic": "/imu",  # 订阅的话题名称
        "throttle_rate": 100,  # 限制消息发布频率为100ms
        "queue_length": 10,    # 队列长度为10
        "compression": "cbor"  # 使用CBOR压缩以减少带宽
    }
    # 发送订阅请求
    ws.send(json.dumps(subscribe_message))
    print(f"已订阅话题: /imu")

if __name__ == "__main__":
    # 设为True可以看到详细的底层通信日志，用于调试
    websocket.enableTrace(False)
  
    # --- !!! 请务必修改为你机器人端的真实IP地址 !!! ---
    ws_url = "ws://192.168.9.32:9090"
  
    print(f"正在连接到 {ws_url}...")
  
    # 创建WebSocketApp对象并设置回调函数
    ws = websocket.WebSocketApp(
        ws_url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
  
    # 设置断线重连机制
    reconnect_interval = 5  # 断线后等待5秒重试
    while True:
        try:
            # 启动WebSocket客户端
            ws.run_forever()
        except Exception as e:
            print(f"发生异常: {e}")
            print(f"{reconnect_interval}秒后尝试重连...")
            time.sleep(reconnect_interval)

第3步：运行脚本，见证奇迹

在test.py文件所在的目录打开终端，运行：

1
python3 test.py

如果一切顺利，你将看到：

1
2
3
4
5
6
正在连接到 ws://192.168.9.32:9090...
WebSocket连接已建立
已订阅话题: /imu
[2023-01-01 12:00:00] [IMU Data] x=0.0123, y=0.0456, z=0.0789
[2023-01-01 12:00:01] [IMU Data] x=0.0124, y=0.0455, z=0.0790
...

恭喜你！你已经成功地从ROS机器人获取了实时数据！

第三部分：常见问题与排错 (FAQ)

问题1：Python脚本报错 ConnectionRefusedError (连接被拒绝)

  • 原因A: 你忘记在机器人端启动rosbridge_server。
    解决方法: 在机器人端运行 roslaunch rosbridge_server rosbridge_websocket.launch
  • 原因B: test.py里的ws_url IP地址填错了，不是机器人的IP。
    解决方法: 重新检查机器人IP地址并更新脚本
  • 原因C: 机器人或你的电脑开启了防火墙，阻止了9090端口的通信。
    解决方法: 在机器人端关闭防火墙或开放9090端口
    1
    2
    3
    sudo ufw disable  # 关闭防火墙（不推荐在生产环境使用）
    # 或者开放9090端口
    sudo ufw allow 9090/tcp

问题2：rosbridge终端报错 Cannot infer topic type… not yet advertised

  • 唯一原因: 你没有启动发布/imu话题的机器人节点(/base_driver)，或者该节点启动后崩溃了。
    解决方法: 请务必先用 rostopic info /imu 确认数据源正常。检查机器人节点是否正确启动，查看节点日志排查问题。

问题3：脚本能连接，也显示订阅成功，但就是不打印 [IMU Data]

  • 原因A: 话题名称写错了。
    解决方法: 请检查test.py中 "topic": "/imu" 是否与 rostopic list 显示的完全一致。
  • 原因B: 机器人节点虽然启动了，但由于某种原因停止发布数据了。
    解决方法: 可以在机器人端用 rostopic echo /imu 监听一下，看看是否有数据滚动。
  • 原因C: JSON解析错误导致数据处理失败。
    解决方法: 打开调试模式 websocket.enableTrace(True) 查看详细的消息内容，检查数据格式是否符合预期。

问题4：数据接收不稳定，经常断线

  • 原因: 网络连接不稳定或ROS消息发布频率过高。
    解决方法:

    1. 优化网络环境，确保机器人和客户端在同一局域网且信号良好
    2. 降低ROS话题发布频率
    3. 调整订阅参数，增加队列长度和节流率：
    1
    2
    3
    4
    5
    6
    7
    subscribe_message = {
        "op": "subscribe",
        "topic": "/imu",
        "throttle_rate": 200,  # 增加节流率到200ms
        "queue_length": 50,    # 增加队列长度到50
        "compression": "cbor"
    }

第四部分：进阶应用

1. 订阅多个话题

如果你需要同时订阅多个话题，只需在on_open函数中发送多个订阅消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def on_open(ws):
    print("WebSocket连接已建立")
  
    # 订阅IMU话题
    ws.send(json.dumps({
        "op": "subscribe",
        "topic": "/imu",
        "throttle_rate": 100
    }))
  
    # 订阅里程计话题
    ws.send(json.dumps({
        "op": "subscribe",
        "topic": "/odom",
        "throttle_rate": 100
    }))
  
    print("已订阅话题: /imu, /odom")

2. 数据可视化

使用matplotlib库可以实时绘制IMU数据曲线：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import numpy as np

# 初始化图形
fig, ax = plt.subplots()
x_data, y_data, z_data = [], [], []
line_x, = ax.plot([], [], 'r-', label='X')
line_y, = ax.plot([], [], 'g-', label='Y')
line_z, = ax.plot([], [], 'b-', label='Z')
ax.set_xlim(0, 100)
ax.set_ylim(-10, 10)
ax.legend()
ax.set_title('IMU加速度数据实时显示')
ax.set_xlabel('时间')
ax.set_ylabel('加速度 (m/s²)')

def update(frame):
    line_x.set_data(range(len(x_data)), x_data)
    line_y.set_data(range(len(y_data)), y_data)
    line_z.set_data(range(len(z_data)), z_data)
    if len(x_data) > 100:
        ax.set_xlim(len(x_data)-100, len(x_data))
    return line_x, line_y, line_z

def on_message(ws, message):
    try:
        data = json.loads(message)
        if "msg" in data and 'linear_acceleration' in data['msg']:
            imu_data = data['msg']
            x = imu_data['linear_acceleration']['x']
            y = imu_data['linear_acceleration']['y']
            z = imu_data['linear_acceleration']['z']
          
            # 更新数据
            x_data.append(x)
            y_data.append(y)
            z_data.append(z)
          
            # 限制数据点数量，避免内存溢出
            if len(x_data) > 200:
                x_data.pop(0)
                y_data.pop(0)
                z_data.pop(0)
              
    except json.JSONDecodeError as e:
        print(f"[JSON解析错误] {e}: {message}")

# 创建动画
ani = FuncAnimation(fig, update, interval=100)

# 启动WebSocket和图形显示
if __name__ == "__main__":
    # ... (保持原有代码不变)
  
    # 在单独线程中启动图形显示
    import threading
    thread = threading.Thread(target=plt.show)
    thread.daemon = True
    thread.start()
  
    # 启动WebSocket客户端
    ws.run_forever()

3. 发送命令到ROS

除了订阅话题接收数据，你还可以通过WebSocket向ROS发布消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def send_command(ws, topic, message_type, data):
    """向ROS发布消息"""
    publish_msg = {
        "op": "publish",
        "topic": topic,
        "type": message_type,
        "msg": data
    }
    ws.send(json.dumps(publish_msg))

# 示例：发布速度命令
def on_open(ws):
    print("WebSocket连接已建立")
    # 订阅IMU话题
    ws.send(json.dumps({
        "op": "subscribe",
        "topic": "/imu"
    }))
  
    # 发布速度命令
    send_command(
        ws, 
        "/cmd_vel", 
        "geometry_msgs/Twist", 
        {
            "linear": {"x": 0.1, "y": 0.0, "z": 0.0},
            "angular": {"x": 0.0, "y": 0.0, "z": 0.2}
        }
    )

通过这些步骤和代码，你不仅可以获取ROS机器人的实时数据，还可以实现数据可视化和机器人控制，为开发更复杂的机器人应用打下基础。

  • Title: 通过 WebSocket 实时获取 ROS 机器人数据
  • Author: John Doe
  • Created at : 2025-07-25 19:31:20
  • Updated at : 2025-07-27 13:06:23
  • Link: https://redefine.ohevan.com/2025/07/25/通过-WebSocket-实时获取-ROS-机器人数据/
  • License: This work is licensed under CC BY-NC-SA 4.0.
Comments
On this page
通过 WebSocket 实时获取 ROS 机器人数据
  1. 从零开始：通过 WebSocket 实时获取 ROS 机器人数据 (小白终极教程)
    1. 前言
    2. 核心架构
    3. 第一部分：机器人端配置 (以 IP: 192.168.9.32 为例)
      1. 第1步：确定并记录机器人的IP地址
      2. 第2步：安装rosbridge_server
      3. 第3步：启动并验证数据源（IMU话题）
      4. 第4步：配置并启动rosbridge_server
    4. 第二部分：客户端配置 (你的电脑)
      1. 第1步：准备Python环境
      2. 第2步：编写并配置Python脚本
      3. 第3步：运行脚本，见证奇迹
    5. 第三部分：常见问题与排错 (FAQ)
      1. 问题1：Python脚本报错 ConnectionRefusedError (连接被拒绝)
      2. 问题2：rosbridge终端报错 Cannot infer topic type… not yet advertised
      3. 问题3：脚本能连接，也显示订阅成功，但就是不打印 [IMU Data]
      4. 问题4：数据接收不稳定，经常断线
    6. 第四部分：进阶应用
      1. 1. 订阅多个话题
      2. 2. 数据可视化
      3. 3. 发送命令到ROS