上周我打通了Milk-V Duo S和机器人的串口控制的脉络同时制作了两个动作组下蹲、自由飞翔。阶段状态环境搭建✅ 完成网络配置✅ 完成串口控制✅ 完成动作调度✅ 完成序列执行控制✅ 完成时间控制✅ 完成部分动作组制作✅ 完成线程执行锁✅ 完成云端接入⏳ 下一步复杂动作组制作⏳ 下一步…⏳ 下一步本周我将进行Milk-V Duo S开发板与云端服务器进行通信模块的开发,接入云端以从云端接收数据从而进一步实现后端执行层与AI分析层之间的连接 。云端接入经过我的调研找到了两种方案如下第一种通过MQTT建立与云端的通信让机器人从“本地脚本控制”跃迁到“云端实时指挥”核心就是在 Milk‑V Duo S 上架起一座 MQTT 消息桥。我本周开发的通信模块mqtt_control.py扮演了云端指令的接收者和执行调度器。我们使用paho.mqtt.client连接到 MQTT 服务器订阅robot/control主题一旦收到 JSON 格式的动作序列如{actions: [1, 7, 6, 2, 0]}就立刻调用之前已经调试成熟的run_sequence函数驱动真实的舵机表演。提示词请生成一个 Python 脚本文件保存为 mqtt_control.py用于在 Milk-V Duo S 上通过 MQTT 接收云端指令并控制机器人执行动作序列。要求如下 - 导入 paho.mqtt.client、json、threading 模块。 - 从 seq_control 模块导入 RobotController、run_sequence、ACTION_DURATION。 串口默认 /dev/ttyS0波特率 115200。 - 定义一个全局锁用于保护串口通信。 - 定义 connect 回调函数 - 定义 message 回调函数 - 将接收到的消息载荷用 json.loads 解析为字典 data。 - 打印 收到云端指令: 和 data。 - 否则使用 threading.Thread 启动一个新线程目标函数为 run_sequence - 创建 MQTT 客户端对象将 connect 和 message 分别赋值给 client.on_connect 和 client.on_message。 - 设置云服务器地址 SERVER 47.93.27.196 PORT 1883 - 打印 等待云端指令...然后调用 client.loop_forever() 进入消息循环。 请输出完整的 Python 代码。LLM产出codeimport paho.mqtt.client as mqtt import json import threading from seq_control import RobotController, run_sequence, ACTION_DURATION # ---------- 机器人控制器初始化 ---------- robot RobotController(port/dev/ttyS0, baudrate115200) # ---------- 锁保护串口通信 ---------- lock threading.Lock() def on_connect(client, userdata, flags, rc): print(fMQTT 连接成功状态码: {rc}) client.subscribe(robot/control) def on_message(client, userdata, msg): try: data json.loads(msg.payload.decode()) print(收到云端指令:, data) sequence data.get(actions, []) if not sequence: print(指令中无动作序列忽略) return # 在新线程中执行动作序列避免阻塞 MQTT 消息接收 threading.Thread( targetrun_sequence, args(robot, sequence), daemonTrue ).start() except Exception as e: print(解析或执行错误:, e) # ---------- MQTT 客户端配置 ---------- client mqtt.Client() client.on_connect on_connect client.on_message on_message # 你的云服务器地址MQTT 默认端口 1883 SERVER 47.93.27.196 PORT 1883 client.connect(SERVER, PORT, 60) print(等待云端指令...) client.loop_forever()这里最关键的升级是多线程执行。因为动作组里有长达 72 秒的舞蹈如果直接在on_message回调里运行机器人会在这段时间内完全“失聪”无法响应任何新指令。我们通过threading.Thread将动作序列放到后台线程并设置为 Daemon 模式这样主 MQTT 循环不会被阻塞随时可以接收新的云端命令甚至能够被更新、中断或重新编排。配合seq_control.py中已有的threading.Lock两个线程对串口的访问也得到了保护不会出现指令交错。至此机器人不再只是一段预设脚本的播放器而是一个可以实时接收、解析、执行云端动作指令的智能终端为后续接入 NLP 情绪分析、远程控制等场景铺平了道路。第二种通过HTTP轮询建立与云端的通信提示词生成一个 HTTP 轮询机器人控制脚本 poll_control.py 从 seq_control 导入 RobotController 和 run_sequence。 服务器地址 http://47.93.27.196:8765客户端 ID 为 milk_duos_001 每 2 秒轮询一次 /poll/id 获取动作序列 有新的 command_id 时执行 run_sequence 并向 /ack/id 发送确认。 包含异常处理和 CtrlC 停止。 输出完整Python代码.LLM产出codeimport requests import time from robot_control import RobotController, run_sequence # 配置 SERVER http://47.93.27.196:8765 CLIENT_ID milk_duos_001 POLL_INTERVAL 2 robot RobotController() def poll_action(): url f{SERVER}/poll/{CLIENT_ID} try: r requests.get(url, timeout5) return r.json() except Exception as e: print(轮询失败:, e) return None def ack_action(command_id): url f{SERVER}/ack/{CLIENT_ID} try: requests.post(url, json{command_id: command_id}) print(已确认:, command_id) except Exception as e: print(确认失败:, e) def main(): print( Duo S 已连接云端开始轮询...) last_id None while True: result poll_action() if result: # 有动作 if result.get(action_sequence): cmd_id result.get(command_id) # 防止重复执行 if cmd_id ! last_id: sequence result[action_sequence] print(收到动作:, sequence) # ⭐ 核心执行真实机器人动作 run_sequence(robot, sequence) # ⭐ 执行完告诉云端 ack_action(cmd_id) last_id cmd_id elif result.get(status) no_action: pass else: print(未知响应:, result) time.sleep(POLL_INTERVAL) if __name__ __main__: try: main() except KeyboardInterrupt: robot.stop() robot.close()采用HTTP 轮询的方式让机器人从云端获取动作指令。Doo S 周期性地向服务器发起 GET 请求查询当前是否有待执行的动作序列如果返回了新的command_id就调用已有的run_sequence执行真实机器人动作执行完毕再通过 POST 请求向服务器确认已完成。这种方式实现简单无需额外的中间件仅靠 HTTP 就能在有限资源下跑通。但它本质上是一种“拉”模式机器人每隔 2 秒主动查一次动作的响应会有最多 2 秒的延迟不适合需要即时反应的交互场景。不过在 Duo S 这种嵌入式板卡上该方案不依赖持久的连接对网络波动容忍度更高且容易调试和扩展很适合初期原型验证。对比MQTTHTTP 轮询消息模式“推” 模式云端有新指令立刻推送给机器人“拉” 模式由机器人定时主动查询云端指令实时性实时性高指令几乎无延迟存在固定查询间隔如 2 秒响应不够及时连接依赖需要维持长连接对网络稳定性要求高断网需重连每次请求独立、无状态网络适应性强资源开销长连接省流量但客户端需维护心跳频繁发起请求即便无新指令也会持续消耗带宽和 CPU适用场景交互频繁、要求低延迟的云控场景指令稀疏、开发早期快速验证场景根据测试情况外加与组长和组员的交流我们在测试过程中发现HTTP轮询的查询间隔2s延迟并不明显于是决定先使用HTTP轮询的方式快速进行项目原型验证。云端Duo S机器人端联调1.确认云服务器状态正常curl http://47.93.27.196:8765/health{status:ok}表明云服务器状态正常且Duo S已经成功连上云服务器2.Duo S在终端运行同时云端发送情绪curl -X POST http://47.93.27.196:8765/emotion \ -H Content-Type: application/json \ -d {\content\:\我人工智能导论只考了95分差一分就能拿到A了,现在没有达成目标感觉好难过\}刚开始我们是失败状态的后来发现犯了一个低级错误Duo S wifi配置自动执行文件的默认账号密码之前在借给队友时候修改过了然后现在还没修改为我的热点。到目录/etc/wpa_supplicant.conf文件 修改完成之后才实现HTTP正常连接同时云服务器再次发送emotion演示视频2