[具身智能-597]：RDK X5 4G/5G 模组对接云端完整方案 + 可运行代码示例

一、整体通信架构说明

  1. 硬件：USB 4G/5G 模组插入 RDK X5 USB3.0，拨号后生成标准蜂窝网卡wwan0；
  2. 底层：QMI/MBIM/ECM/PPP 拨号建立蜂窝 TCP/IP 链路；
  3. 上云主流 4 种工业方案：
    • MQTT（物联网标准，推荐，低带宽、支持上下行指令）
    • TCP Socket（自定义私有协议，机器人私有数据透传）
    • HTTP/HTTPS（文件上传、日志上报、简单数据上报）
    • WebSocket（实时视频 / 高频状态推送）
  4. 核心特性：4G/5G 网卡属于标准 Linux 网络设备，业务代码无需区分以太网 / Wi‑Fi / 蜂窝网卡，系统路由自动控制出口。

前置环境准备

1. 安装依赖

bash

运行

sudo apt update # MQTT工具库 sudo apt install python3-paho-mqtt # HTTP请求库 pip3 install requests # 蜂窝拨号工具（移远QMI为例） sudo apt install libqmi-utils udhcpc

2. 4G/5G 拨号基础脚本4g_dial.sh

bash

运行

#!/bin/bash CDC_DEV="/dev/cdc-wdm0" WWAN_IF="wwan0" APN="cmnet" sudo ip link set $WWAN_IF down sudo uqmi -d $CDC_DEV --reset # 建立蜂窝会话 sudo uqmi -d $CDC_DEV --start-network apn=$APN # 获取IP sudo udhcpc -i $WWAN_IF # 默认流量走4G网卡 sudo route del default sudo route add default dev $WWAN_IF metric 10 echo "4G拨号完成，wwan信息：" ip a show $WWAN_IF

赋予权限：chmod +x 4g_dial.sh && ./4g_dial.sh

方案 1：MQTT 物联网上云（量产首选）

适用：设备状态上报、云端下发控制指令、AI 识别结果推送、低功耗小报文

Python 完整示例

python

运行

import paho.mqtt.client as mqtt import json import time import socket # ===================== 配置区 ===================== # 云端MQTT公网地址（阿里云/腾讯云/自建MQTT服务器） MQTT_BROKER = "xxx.mqtt.server.com" MQTT_PORT = 1883 # 设备唯一标识 DEVICE_ID = "rdk_x5_robot_001" # 上行：设备上报数据主题 UP_TOPIC = f"device/{DEVICE_ID}/upload" # 下行：云端下发控制指令主题 CMD_TOPIC = f"device/{DEVICE_ID}/cmd" # 连接回调 def on_connect(client, userdata, flags, rc): if rc == 0: print("MQTT连接云端成功") client.subscribe(CMD_TOPIC) else: print("MQTT连接失败，错误码：", rc) # 接收云端下发指令回调 def on_message(client, userdata, msg): cmd_data = json.loads(msg.payload.decode("utf-8")) print(f"收到云端控制指令：{cmd_data}") # 此处对接底盘/AI控制逻辑 if cmd_data["action"] == "stop": print("执行机器人停止动作") # 发布消息回调 def on_publish(client, userdata, mid): pass if __name__ == "__main__": # 创建MQTT客户端 client = mqtt.Client(client_id=DEVICE_ID) client.on_connect = on_connect client.on_message = on_message client.on_publish = on_publish # 连接云端（自动使用wwan0蜂窝网卡出口） client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60) client.loop_start() count = 0 while True: # 模拟设备上报数据：AI识别数量、信号、设备时间 payload = json.dumps({ "device_id": DEVICE_ID, "net_type": "4G/5G", "detect_num": count, "rssi": -62, "timestamp": time.time() }) # 上报云端 client.publish(UP_TOPIC, payload, qos=1) print("上报数据：", payload) count += 1 time.sleep(2)

方案 2：TCP Socket 私有协议透传（机器人私有业务）

适用：自定义二进制协议、底盘实时运动数据、高频原始数据流

Python TCP 客户端示例

python

运行

import socket import time import struct # 云端TCP服务地址 SERVER_IP = "123.45.67.89" SERVER_PORT = 9000 def tcp_client(): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) try: sock.connect((SERVER_IP, SERVER_PORT)) print("TCP连接云端服务器成功") except Exception as e: print("TCP连接失败：", e) sock.close() return while True: # 自定义二进制报文：2字节长度 + 4字节状态 + 校验 msg_len = 8 status_data = struct.pack("=HIII", msg_len, 0x01, 100, 200) sock.send(status_data) # 接收云端返回指令 recv_buf = sock.recv(1024) if recv_buf: print("云端返回数据：", recv_buf.hex()) time.sleep(0.1) if __name__ == "__main__": tcp_client()

方案 3：HTTP/HTTPS 日志、图片、文件上传

适用：AI 截图上传、运行日志上报、设备注册、简单表单数据

python

运行

import requests import json import time # 云端HTTP接口地址 HTTP_UPLOAD_URL = "http://xxx.server.com/api/device/upload" DEVICE_ID = "rdk_x5_robot_001" def http_report(): upload_data = { "device_id": DEVICE_ID, "signal": -65, "cpu_temp": 42.5, "detect_count": 128 } try: # POST上传JSON数据 resp = requests.post(HTTP_UPLOAD_URL, json=upload_data, timeout=10) if resp.status_code == 200: print("HTTP上报成功，云端返回：", resp.json()) else: print("HTTP上报失败，状态码：", resp.status_code) except Exception as e: print("网络异常：", e) if __name__ == "__main__": while True: http_report() time.sleep(5)

图片文件上传扩展（HTTP）

python

运行

def upload_image(img_path): files = {"img": open(img_path, "rb")} data = {"device_id": DEVICE_ID} resp = requests.post("http://xxx.server.com/api/img/upload", files=files, data=data) print("图片上传结果：", resp.text)

方案 4：ROS2（TogetheROS）机器人业务上云

场景：RDK AI 识别、底盘里程数据通过 4G 网卡上传远端上位机 / 云端

python

运行

#!/usr/bin/env python3 import rclpy from rclpy.node import Node from std_msgs.msg import String import json import time DEVICE_ID = "rdk_x5_ros_001" class Ros4GUploadNode(Node): def __init__(self): super().__init__("4g_cloud_upload_node") self.pub = self.create_publisher(String, "/cloud/upload_msg", 10) self.timer = self.create_timer(2.0, self.upload_cb) self.get_logger().info("4G ROS云端上报节点启动") def upload_cb(self): msg_data = json.dumps({ "device": DEVICE_ID, "network": "wwan0 4G/5G", "time": time.time() }) msg = String() msg.data = msg_data self.pub.publish(msg) self.get_logger().info(f"ROS上报云端数据：{msg_data}") def main(args=None): rclpy.init(args=args) node = Ros4GUploadNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == "__main__": main()

启动时绑定蜂窝网卡，保证 ROS 流量走 4G：

bash

运行

export FASTRTPS_DEFAULT_PROFILES_FILE=fastdds_wwan.xml ros2 run your_package 4g_cloud_upload_node

工程配套增强方案（工业稳定必备）

1. 断网自动重连机制（MQTT 示例改造）

在循环内增加网络检测，断开后重新拨号重连：

python

运行

def check_network(): try: # 公网连通性检测 socket.create_connection(("8.8.8.8", 53), timeout=2) return True except: return False # 上报前判断网络，断网执行重拨号 if not check_network(): print("4G网络断开，重新拨号") import os os.system("/home/hobot/4g_dial.sh") time.sleep(5)

2. 路由策略：指定业务仅走 4G 网卡

bash

运行

# 默认所有流量优先4G sudo route add default dev wwan0 metric 10 # 仅访问云端IP走4G，其余走以太网 sudo ip route add 123.45.67.0/24 dev wwan0

3. 开机自启动拨号 + 上云服务

systemd 同时托管拨号脚本与 Python 上云程序，断电自动恢复通信。

选型总结

  1. 物联网设备、远程控制、指令下发 →MQTT（推荐）
  2. 机器人底层私有二进制数据流、低延迟透传 →TCP Socket
  3. 日志、截图、文件批量上传 →HTTP/HTTPS
  4. TogetheROS 机器人分布式场景 →ROS2 + MQTT

关键注意事项

  1. 4G/5G 模组运营商多为内网 IP，云端主动连设备会失败，必须采用设备主动发起连接（客户端模式，以上全部代码均为客户端）；
  2. 5G 模组供电要求高，峰值 2A，外置独立电源防止掉线；
  3. 增加心跳包机制，云端 30s 无心跳判定设备离线；
  4. 传输二进制大文件建议分包，搭配断点续传逻辑。