1. 项目缘起为什么要把Reachy Mini接入Home Assistant如果你和我一样既对机器人技术着迷又是个智能家居的深度用户那么你肯定想过一个问题能不能让我的机器人助手也成为智能家居生态的一部分比如让Reachy Mini这个灵巧的桌面机械臂在检测到门口有人时自动帮你把客厅的灯打开或者在你准备开始远程会议时自动帮你调整好桌面的摄像头角度和灯光亮度。这听起来像是科幻电影里的场景但通过Home Assistant这个强大的开源家庭自动化平台我们完全可以将Reachy Mini从一个独立的机器人设备转变为一个能够感知环境、执行复杂自动化任务的智能“家庭成员”。Reachy Mini是一款由Pollen Robotics开发的开源、模块化桌面机械臂以其友好的Python SDK和相对亲民的价格成为了机器人爱好者和研究者的热门选择。它拥有7个自由度末端执行器灵活非常适合进行抓取、示教、人机交互等任务。然而它的能力通常被限制在本地网络或直接编程控制中。Home Assistant则是一个将不同品牌、不同协议的智能设备统一管理、并实现自动化联动的中心大脑。将两者结合本质上是为Reachy Mini打开了通往整个智能家居世界的大门让它能够基于丰富的环境传感器数据如人体移动、温度、光照、门磁状态来触发动作或者将其自身的状态如是否正在执行任务、关节角度反馈给家庭自动化系统从而实现更高阶的、上下文感知的自动化。这个集成的核心价值在于场景融合与能力扩展。对于机器人开发者这提供了一种低成本验证服务机器人家庭应用场景的途径对于智能家居极客这是将物理交互能力引入自动化流程的绝佳实验场。接下来我将详细拆解从零开始实现这一集成的完整路径涵盖原理、通信方案选型、详细配置步骤以及我趟过的那些坑。2. 核心通信架构设计MQTT连接两个世界的桥梁要实现Reachy Mini与Home Assistant的对话首先要解决通信协议问题。Home Assistant支持多种集成方式如直接API集成、通过Add-on安装特定集成等。但对于Reachy Mini这类没有官方集成的设备MQTT协议是最通用、最灵活且最推荐的选择。MQTT是一种轻量级的发布/订阅消息传输协议特别适合物联网设备。在Home Assistant中MQTT是一个核心组件任何能够连接MQTT broker并按照一定格式发布消息的设备都可以被Home Assistant自动发现并作为实体Entity进行管理。因此我们的系统架构将围绕MQTT展开如下图所示概念描述MQTT代理Broker作为消息中转站通常与Home Assistant安装在同一服务器上例如使用Mosquitto add-on。Home Assistant订阅来自Reachy Mini的主题Topic以获取其状态同时向特定主题发布消息来向Reachy Mini发送指令。Reachy Mini运行一个自定义的Python桥接程序。这个程序需要使用paho-mqtt或asyncio-mqtt库连接到MQTT Broker。订阅Home Assistant发送指令的主题例如reachy_mini/command。根据指令调用Reachy SDKreachy_sdk控制机械臂运动。将机械臂的状态如是否忙碌、关节角度、电池电量等发布到特定的状态主题例如reachy_mini/state。为什么选择MQTT而不是直接的HTTP API原因有三一是解耦双方只需认识Broker无需知道对方IP和端口网络配置更简单二是实时性MQTT的发布/订阅模式非常适合实时控制与状态同步三是Home Assistant原生支持通过MQTT自动发现功能可以近乎零配置地在UI中生成控制卡片。注意确保你的网络环境中MQTT Broker的端口默认1883是可达的。如果Home Assistant运行在Docker或虚拟机中需要注意网络模式配置。3. 环境准备与基础组件部署在开始编写代码之前我们需要搭建好基础环境。这里假设你已经有一台运行了Home Assistant的设备如Raspberry Pi、旧电脑或NAS并且已经初始化完成。3.1 在Home Assistant中安装并配置MQTT Broker最简便的方式是通过Home Assistant的官方插件库安装“Mosquitto broker” add-on。安装Mosquitto Broker进入Home Assistant的“配置” - “加载项” - “加载项商店”。搜索“Mosquitto broker”点击进入并安装。安装完成后不要立即启动。配置Mosquitto在Mosquitto add-on的“配置”标签页我们通常需要设置用户名和密码以增强安全性。一个简单的配置如下logins: - username: reachy password: your_secure_password_here anonymous: false customize: active: false将your_secure_password_here替换为强密码。记下这个用户名和密码后续Reachy Mini的连接会用到。在“网络”标签页确保端口映射正确例如1883端口用于MQTT9001端口用于WebSocket后者可用于一些高级前端工具。启动并集成启动Mosquitto add-on并勾选“自动启动”和“监视”选项。然后进入Home Assistant的“配置” - “设备与服务” - “集成”。点击“添加集成”搜索“MQTT”并添加。在配置时Broker地址填写localhost或127.0.0.1因为Mosquitto与HA同机端口1883并输入上一步设置的用户名和密码。至此Home Assistant端的MQTT服务就绪。你可以通过安装一个简单的MQTT测试客户端如MQTT Explorer来验证Broker是否正常工作。3.2 Reachy Mini侧Python环境与依赖安装在运行Reachy Mini的计算机上通常是连接着Reachy的树莓派或PC我们需要准备Python环境。确保Python版本Reachy SDK通常要求Python 3.7或更高版本。使用python3 --version确认。安装Reachy SDK这是控制机械臂的核心。pip install reachy-sdk实操心得建议在虚拟环境venv或conda中操作避免包冲突。特别是如果你的系统还运行其他机器人应用。安装MQTT客户端库我们将使用paho-mqtt它稳定且文档丰富。pip install paho-mqtt4. 构建MQTT桥接程序从指令到动作这是整个集成的核心代码部分。我们将编写一个Python脚本作为Reachy Mini的“大脑”翻译来自Home Assistant的MQTT消息为具体的机械臂动作。4.1 程序骨架与连接逻辑首先我们创建主程序框架处理MQTT连接、订阅和消息回调。#!/usr/bin/env python3 Reachy Mini to Home Assistant MQTT Bridge 用于将Reachy Mini的控制和状态集成到Home Assistant。 import json import time import threading from typing import Dict, Any import paho.mqtt.client as mqtt from reachy_sdk import ReachySDK from reachy_sdk.trajectory import goto class ReachyMiniHABridge: def __init__(self, mqtt_broker: str, mqtt_port: int, mqtt_user: str, mqtt_pass: str, reachy_ip: str): 初始化桥接器。 :param mqtt_broker: MQTT代理地址 :param mqtt_port: MQTT代理端口 :param mqtt_user: MQTT用户名 :param mqtt_pass: MQTT密码 :param reachy_ip: Reachy Mini的IP地址 self.mqtt_broker mqtt_broker self.mqtt_port mqtt_port self.mqtt_user mqtt_user self.mqtt_pass mqtt_pass self.reachy_ip reachy_ip # MQTT客户端 self.mqtt_client mqtt.Client(client_idreachy_mini_bridge) self.mqtt_client.username_pw_set(self.mqtt_user, self.mqtt_pass) self.mqtt_client.on_connect self._on_mqtt_connect self.mqtt_client.on_message self._on_mqtt_message # Reachy SDK 客户端 self.reachy None self._connect_reachy() # 状态跟踪 self.is_busy False self.current_joints {} def _connect_reachy(self): 连接到Reachy Mini机器人。 try: print(f正在连接Reachy Mini ({self.reachy_ip})...) # 注意实际连接可能需要根据你的网络配置调整 self.reachy ReachySDK(hostself.reachy_ip) # 开机所有关节电机 self.reachy.turn_on(all) time.sleep(2) # 等待电机上电完成 print(Reachy Mini 连接成功。) # 发布初始状态 self._publish_availability(True) self._publish_all_states() except Exception as e: print(f连接Reachy Mini失败: {e}) self._publish_availability(False) def _on_mqtt_connect(self, client, userdata, flags, rc): MQTT连接成功回调。 if rc 0: print(成功连接到MQTT Broker。) # 订阅命令主题 client.subscribe(reachy_mini/command/#) # 发布自动发现信息详见下一节 self._publish_auto_discovery() else: print(f连接MQTT Broker失败返回码: {rc}) def _on_mqtt_message(self, client, userdata, msg): 处理收到的MQTT消息。 topic msg.topic payload msg.payload.decode() print(f收到消息: {topic} - {payload}) # 根据主题分发处理 if topic reachy_mini/command/set: self._handle_command(payload) elif topic.startswith(reachy_mini/command/arm/): # 处理手臂特定命令例如 reachy_mini/command/arm/left/goto self._handle_arm_command(topic, payload) def _handle_command(self, payload: str): 处理通用命令。 try: cmd json.loads(payload) action cmd.get(action) if action go_to_rest: self._go_to_rest_position() elif action get_status: self._publish_all_states() # 可以扩展更多命令... except json.JSONDecodeError as e: print(f命令JSON解析错误: {e}) def _handle_arm_command(self, topic: str, payload: str): 处理手臂运动命令。 if self.is_busy: print(机器人正忙忽略新命令。) return # 解析主题例如reachy_mini/command/arm/left/goto parts topic.split(/) arm_side parts[4] # left 或 right command parts[5] # goto try: target_positions json.loads(payload) # 应是一个关节角度字典 self._move_arm_to_positions(arm_side, target_positions) except Exception as e: print(f处理手臂命令时出错: {e}) def _move_arm_to_positions(self, arm_side: str, positions: Dict[str, float]): 控制单臂运动到指定位置。 self.is_busy True self._publish_state(is_busy, True) try: arm getattr(self.reachy, f{arm_side}_arm) # 获取目标关节对象 target_joints {} for joint_name, angle in positions.items(): if hasattr(arm, joint_name): target_joints[getattr(arm, joint_name)] angle else: print(f警告关节 {joint_name} 在 {arm_side}_arm 上不存在。) # 执行运动 goto({**target_joints}, duration2.0) # 2秒内完成运动 time.sleep(2.2) # 稍作等待确保运动完成 # 更新并发布当前关节状态 self._update_and_publish_joint_states(arm_side) except Exception as e: print(f机械臂运动出错: {e}) finally: self.is_busy False self._publish_state(is_busy, False) def _go_to_rest_position(self): 让机械臂回到休息位置。 # 这里需要定义你的“休息”位姿例如所有关节为0度 rest_positions { shoulder_pitch: 0.0, shoulder_roll: 0.0, arm_yaw: 0.0, elbow_pitch: 0.0, forearm_yaw: 0.0, wrist_pitch: 0.0, wrist_roll: 0.0, } # 可以依次移动左臂和右臂或只移动一个 self._move_arm_to_positions(left, rest_positions) # 如果需要移动右臂 self._move_arm_to_positions(right, rest_positions) def _update_and_publish_joint_states(self, arm_side: str): 读取并发布指定手臂的关节角度。 arm getattr(self.reachy, f{arm_side}_arm) joint_states {} for joint in [shoulder_pitch, shoulder_roll, arm_yaw, elbow_pitch, forearm_yaw, wrist_pitch, wrist_roll]: joint_obj getattr(arm, joint) joint_states[f{arm_side}_{joint}] joint_obj.present_position # 发布到MQTT例如主题reachy_mini/state/arm/left/joints self.mqtt_client.publish(freachy_mini/state/arm/{arm_side}/joints, json.dumps(joint_states), retainTrue) def _publish_all_states(self): 发布所有状态信息。 self._publish_state(is_busy, self.is_busy) self._publish_state(is_connected, self.reachy is not None) # 发布左右臂关节状态 if self.reachy: for side in [left, right]: self._update_and_publish_joint_states(side) def _publish_state(self, attribute: str, value: Any): 发布单个状态属性。 self.mqtt_client.publish(freachy_mini/state/{attribute}, json.dumps({value: value}), retainTrue) def _publish_availability(self, available: bool): 发布设备可用性。 status online if available else offline self.mqtt_client.publish(reachy_mini/status, status, retainTrue) def _publish_auto_discovery(self): 向Home Assistant发布自动发现配置。此部分内容较多将在下一节详述。 pass # 占位4.2节实现 def run(self): 启动桥接器主循环。 try: self.mqtt_client.connect(self.mqtt_broker, self.mqtt_port, 60) # 在后台线程运行MQTT网络循环 self.mqtt_client.loop_start() # 主线程可以做一些其他事情或者简单循环保持运行 while True: # 定期发布心跳或状态 self._publish_availability(self.reachy is not None) time.sleep(30) except KeyboardInterrupt: print(正在关闭...) self.mqtt_client.loop_stop() if self.reachy: self.reachy.turn_off(all) print(已关闭。) if __name__ __main__: # 配置参数请根据你的实际环境修改 BRIDGE ReachyMiniHABridge( mqtt_broker你的HA主机IP, # 例如 192.168.1.100 mqtt_port1883, mqtt_userreachy, mqtt_passyour_secure_password_here, reachy_ip你的Reachy Mini IP # 例如 192.168.1.200 ) BRIDGE.run()这段代码搭建了一个稳固的框架。关键点在于_on_mqtt_message是消息路由器根据主题将指令分发给不同的处理函数。_move_arm_to_positions是核心执行函数它将收到的目标角度字典转化为对Reachy SDKgoto函数的调用。状态管理is_busy很重要防止多个运动指令同时执行导致冲突。retainTrue参数在发布状态时使用确保新上线的Home Assistant能立刻获取到最新状态。4.2 实现Home Assistant自动发现Auto Discovery为了让Home Assistant自动识别并创建Reachy Mini的控制实体我们需要利用MQTT的自动发现功能。这需要我们的桥接程序在启动时向特定的发现主题发布配置消息。以创建一个控制左臂的“脚本”实体Script Entity为例它可以在HA的自动化中调用def _publish_auto_discovery(self): 发布自动发现配置。 # 设备信息 device_info { identifiers: [reachy_mini_001], name: Reachy Mini, manufacturer: Pollen Robotics, model: Reachy Mini, } # 1. 创建一个用于触发“回到休息位置”的脚本实体 rest_script_config { name: Reachy Mini Go to Rest, unique_id: reachy_mini_script_go_to_rest, command_topic: reachy_mini/command/set, payload_press: json.dumps({action: go_to_rest}), device: device_info, } discovery_topic fhomeassistant/script/reachy_mini/go_to_rest/config self.mqtt_client.publish(discovery_topic, json.dumps(rest_script_config), retainTrue) # 2. 为左臂的每个关节创建一个数字输入实体用于设置目标角度 arm_side left joints [shoulder_pitch, shoulder_roll, arm_yaw, elbow_pitch, forearm_yaw, wrist_pitch, wrist_roll] for joint in joints: sensor_unique_id freachy_mini_{arm_side}_{joint}_pos sensor_config { name: fReachy Mini {arm_side} {joint.replace(_, ).title()} Position, unique_id: sensor_unique_id, state_topic: freachy_mini/state/arm/{arm_side}/joints, unit_of_measurement: °, value_template: f{{{{ value_json.{arm_side}_{joint} }}}}, device_class: None, device: device_info, } discovery_topic fhomeassistant/sensor/{sensor_unique_id}/config self.mqtt_client.publish(discovery_topic, json.dumps(sensor_config), retainTrue) # 3. 创建一个命令实体来控制左臂运动例如通过服务调用发送一组角度 # 这里我们可以创建一个“按钮”实体当按下时发布一个包含所有关节角度的复杂命令。 # 但更灵活的方式是在HA中创建一个“脚本”或“模板传感器”来组合这些角度然后通过一个MQTT发布动作发送。 # 为简化我们先创建一个用于调试的“触发”按钮。 trigger_button_config { name: Reachy Mini Send Arm Command, unique_id: reachy_mini_button_send_command, command_topic: reachy_mini/command/arm/left/goto, payload_press: json.dumps({shoulder_pitch: 10.0, elbow_pitch: -45.0}), # 示例载荷 device: device_info, } discovery_topic fhomeassistant/button/reachy_mini/send_command/config self.mqtt_client.publish(discovery_topic, json.dumps(trigger_button_config), retainTrue) print(自动发现配置已发布。)发布这些配置后重启Home Assistant或在“配置”-“设备与服务”中点击“MQTT”集成下的“配置”然后“重新发现设备”你应该就能在设备列表中看到“Reachy Mini”设备及其下属的传感器和按钮实体了。重要提示自动发现消息必须设置retainTrue这样当Home Assistant重启后它仍然能从Broker获取到设备配置信息。5. 在Home Assistant中创建自动化与仪表盘当实体成功添加后乐趣才真正开始。我们可以在Home Assistant中创建丰富的自动化场景和直观的仪表盘。5.1 构建自动化让Reachy Mini与环境互动假设我们有一个安装在门厅的人体传感器binary_sensor.front_door_motion。我们想实现当检测到有人移动且环境光线较暗sensor.lux_sensor 50时让Reachy Mini挥动右臂模拟开灯动作同时真正打开客厅的灯。在HA的“配置”-“自动化与场景”-“创建自动化”中使用可视化编辑器或YAML模式alias: 有人进门且光线暗时Reachy挥手并开灯 description: trigger: - platform: state entity_id: binary_sensor.front_door_motion to: on condition: - condition: numeric_state entity_id: sensor.lux_sensor below: 50 action: - service: button.press target: entity_id: button.reachy_mini_send_command data: # 注意这里需要自定义按钮的payload或者更好的方式是直接调用script - service: light.turn_on target: entity_id: light.living_room_ceiling mode: single但更优雅的方式是在HA中创建一个“脚本”Script专门用于执行挥手的复杂动作序列然后在自动化中调用这个脚本。5.2 创建控制仪表盘在HA的“概览”仪表盘编辑模式下你可以添加卡片来控制Reachy Mini按钮卡片关联我们创建的script.reachy_mini_go_to_rest一键归位。实体卡片显示各个关节的角度传感器sensor.reachy_mini_left_shoulder_pitch_position等。手动命令输入卡片通过MQTT集成或“辅助元素”创建可以让你临时发送自定义的JSON指令到reachy_mini/command/arm/left/goto主题用于调试和特殊动作编排。一个更高级的用法是使用picture-elements卡片在一张Reachy Mini的图片上为每个关节创建滑块input_number实体通过自动化将这些滑块的值组合成JSON命令发送给Reachy实现可视化拖拽控制。6. 实战避坑与性能优化心得在开发和测试这个集成项目时我遇到了几个典型问题这里分享解决方案网络延迟与运动卡顿MQTT通信和网络处理会引入微小延迟。在快速连续发送运动指令时可能导致运动不流畅。解决方案在桥接程序中实现一个简单的指令队列queue.Queue由一个单独的线程顺序处理运动指令而不是在MQTT回调函数中直接执行耗时运动。同时适当增加goto函数的duration参数使运动更平滑。关节角度单位与范围Reachy SDK使用的关节角度单位是弧度rad而Home Assistant的传感器默认显示度数°直接显示弧度值不直观。解决方案在桥接程序发布状态前进行单位转换degrees radians * 180 / π或者在Home Assistant的传感器配置中使用value_template进行转换例如value_template: {{ (value_json.left_shoulder_pitch * 180 / 3.14159) | round(2) }}。自动发现实体混乱或重复如果修改了自动发现配置如unique_id并重新发布HA可能会创建重复实体。解决方案清理旧实体最有效的方法是在停止桥接程序后手动删除HA中对应的实体和设备然后删除MQTT Broker上保留的retained自动发现消息。可以使用MQTT客户端向该发现主题发布一个空消息payload为空并设置retainTrue来清除。然后重启桥接程序发布新配置。Reachy SDK连接不稳定有时网络波动会导致SDK连接断开。解决方案在桥接程序中增加重连逻辑。在_connect_reachy方法外围添加重试循环并在主循环中定期检查连接状态如果断开则尝试重连并重新发布可用性状态。安全强化上述示例使用了明文密码。在生产环境中务必将密码等敏感信息存储在环境变量或配置文件中并确保配置文件不被提交到版本库。可以考虑使用Home Assistant的“密钥”功能来管理MQTT密码。将Reachy Mini集成到Home Assistant不仅仅是增加了一个可遥控的玩具而是构建了一个“具身智能”的初级原型。它让自动化从虚拟的数字世界延伸到了真实的物理世界能够执行“拿起”、“放下”、“指向”、“按压”等动作。你可以继续扩展这个桥接程序例如集成Reachy的摄像头进行简单的视觉识别如通过OpenCV识别特定物体然后将识别结果通过MQTT发送给HA触发更复杂的自动化流程。或者为Reachy Mini创建更复杂的“服务”在HA中直接调用如reachy.pick_up_object这样的高级抽象指令由桥接程序分解为一系列底层关节运动。这个项目的天花板取决于你的想象力和对这两个平台的理解深度。