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

一、整体通信架构说明硬件USB 4G/5G 模组插入 RDK X5 USB3.0拨号后生成标准蜂窝网卡wwan0底层QMI/MBIM/ECM/PPP 拨号建立蜂窝 TCP/IP 链路上云主流 4 种工业方案MQTT物联网标准推荐低带宽、支持上下行指令TCP Socket自定义私有协议机器人私有数据透传HTTP/HTTPS文件上传、日志上报、简单数据上报WebSocket实时视频 / 高频状态推送核心特性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 udhcpc2. 4G/5G 拨号基础脚本4g_dial.shbash运行#!/bin/bash CDC_DEV/dev/cdc-wdm0 WWAN_IFwwan0 APNcmnet 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方案 1MQTT 物联网上云量产首选适用设备状态上报、云端下发控制指令、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 fdevice/{DEVICE_ID}/upload # 下行云端下发控制指令主题 CMD_TOPIC fdevice/{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_idDEVICE_ID) client.on_connect on_connect client.on_message on_message client.on_publish on_publish # 连接云端自动使用wwan0蜂窝网卡出口 client.connect(MQTT_BROKER, MQTT_PORT, keepalive60) 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, qos1) print(上报数据, payload) count 1 time.sleep(2)方案 2TCP 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()方案 3HTTP/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, jsonupload_data, timeout10) 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)图片文件上传扩展HTTPpython运行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, filesfiles, datadata) print(图片上传结果, resp.text)方案 4ROS2TogetheROS机器人业务上云场景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(fROS上报云端数据{msg_data}) def main(argsNone): rclpy.init(argsargs) node Ros4GUploadNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ __main__: main()启动时绑定蜂窝网卡保证 ROS 流量走 4Gbash运行export FASTRTPS_DEFAULT_PROFILES_FILEfastdds_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), timeout2) 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 wwan03. 开机自启动拨号 上云服务systemd 同时托管拨号脚本与 Python 上云程序断电自动恢复通信。选型总结物联网设备、远程控制、指令下发 →MQTT推荐机器人底层私有二进制数据流、低延迟透传 →TCP Socket日志、截图、文件批量上传 →HTTP/HTTPSTogetheROS 机器人分布式场景 →ROS2 MQTT关键注意事项4G/5G 模组运营商多为内网 IP云端主动连设备会失败必须采用设备主动发起连接客户端模式以上全部代码均为客户端5G 模组供电要求高峰值 2A外置独立电源防止掉线增加心跳包机制云端 30s 无心跳判定设备离线传输二进制大文件建议分包搭配断点续传逻辑。