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展开，如下图所示（概念描述）：

1. MQTT代理（Broker）：作为消息中转站，通常与Home Assistant安装在同一服务器上（例如使用Mosquitto add-on）。
2. Home Assistant：订阅来自Reachy Mini的主题（Topic），以获取其状态；同时向特定主题发布消息，来向Reachy Mini发送指令。
3. Reachy Mini：运行一个自定义的Python桥接程序。这个程序需要：
   • 使用paho-mqtt或asyncio-mqtt库连接到MQTT Broker。
   • 订阅Home Assistant发送指令的主题（例如reachy_mini/command）。
   • 根据指令，调用Reachy SDK（reachy_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。

1. 安装Mosquitto Broker：
   • 进入Home Assistant的“配置” -> “加载项” -> “加载项商店”。
   • 搜索“Mosquitto broker”，点击进入并安装。安装完成后，不要立即启动。
2. 配置Mosquitto：
   • 在Mosquitto add-on的“配置”标签页，我们通常需要设置用户名和密码以增强安全性。一个简单的配置如下：

     logins: - username: reachy password: your_secure_password_here anonymous: false customize: active: false

   • 将your_secure_password_here替换为强密码。记下这个用户名和密码，后续Reachy Mini的连接会用到。
   • 在“网络”标签页，确保端口映射正确（例如，1883端口用于MQTT，9001端口用于WebSocket，后者可用于一些高级前端工具）。
3. 启动并集成：
   • 启动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环境。

1. 确保Python版本：Reachy SDK通常要求Python 3.7或更高版本。使用python3 --version确认。
2. 安装Reachy SDK：这是控制机械臂的核心。

   pip install reachy-sdk

   实操心得：建议在虚拟环境（venv或conda）中操作，避免包冲突。特别是如果你的系统还运行其他机器人应用。

3. 安装MQTT客户端库：我们将使用paho-mqtt，它稳定且文档丰富。

   pip install paho-mqtt

4. 构建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_id="reachy_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(host=self.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}, duration=2.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(f"reachy_mini/state/arm/{arm_side}/joints", json.dumps(joint_states), retain=True) 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(f"reachy_mini/state/{attribute}", json.dumps({"value": value}), retain=True) def _publish_availability(self, available: bool): """发布设备可用性。""" status = "online" if available else "offline" self.mqtt_client.publish("reachy_mini/status", status, retain=True) 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_port=1883, mqtt_user="reachy", mqtt_pass="your_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）很重要，防止多个运动指令同时执行导致冲突。
• retain=True参数在发布状态时使用，确保新上线的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 = f"homeassistant/script/reachy_mini/go_to_rest/config" self.mqtt_client.publish(discovery_topic, json.dumps(rest_script_config), retain=True) # 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 = f"reachy_mini_{arm_side}_{joint}_pos" sensor_config = { "name": f"Reachy Mini {arm_side} {joint.replace('_', ' ').title()} Position", "unique_id": sensor_unique_id, "state_topic": f"reachy_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 = f"homeassistant/sensor/{sensor_unique_id}/config" self.mqtt_client.publish(discovery_topic, json.dumps(sensor_config), retain=True) # 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 = f"homeassistant/button/reachy_mini/send_command/config" self.mqtt_client.publish(discovery_topic, json.dumps(trigger_button_config), retain=True) print("自动发现配置已发布。")

发布这些配置后，重启Home Assistant或在“配置”->“设备与服务”中点击“MQTT”集成下的“配置”，然后“重新发现设备”，你应该就能在设备列表中看到“Reachy Mini”设备及其下属的传感器和按钮实体了。

重要提示：自动发现消息必须设置retain=True，这样当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. 实战避坑与性能优化心得

在开发和测试这个集成项目时，我遇到了几个典型问题，这里分享解决方案：

1. 网络延迟与运动卡顿：MQTT通信和网络处理会引入微小延迟。在快速连续发送运动指令时，可能导致运动不流畅。解决方案：在桥接程序中实现一个简单的指令队列（queue.Queue），由一个单独的线程顺序处理运动指令，而不是在MQTT回调函数中直接执行耗时运动。同时，适当增加goto函数的duration参数，使运动更平滑。

2. 关节角度单位与范围：Reachy SDK使用的关节角度单位是弧度（rad），而Home Assistant的传感器默认显示度数（°），直接显示弧度值不直观。解决方案：在桥接程序发布状态前进行单位转换（degrees = radians * 180 / π），或者在Home Assistant的传感器配置中使用value_template进行转换，例如"value_template": "{{ (value_json.left_shoulder_pitch * 180 / 3.14159) | round(2) }}"。

3. 自动发现实体混乱或重复：如果修改了自动发现配置（如unique_id）并重新发布，HA可能会创建重复实体。解决方案：清理旧实体最有效的方法是，在停止桥接程序后，手动删除HA中对应的实体和设备，然后删除MQTT Broker上保留的（retained）自动发现消息。可以使用MQTT客户端向该发现主题发布一个空消息（payload为空）并设置retain=True来清除。然后重启桥接程序发布新配置。

4. Reachy SDK连接不稳定：有时网络波动会导致SDK连接断开。解决方案：在桥接程序中增加重连逻辑。在_connect_reachy方法外围添加重试循环，并在主循环中定期检查连接状态，如果断开则尝试重连，并重新发布可用性状态。

5. 安全强化：上述示例使用了明文密码。在生产环境中，务必将密码等敏感信息存储在环境变量或配置文件中，并确保配置文件不被提交到版本库。可以考虑使用Home Assistant的“密钥”功能来管理MQTT密码。

将Reachy Mini集成到Home Assistant，不仅仅是增加了一个可遥控的玩具，而是构建了一个“具身智能”的初级原型。它让自动化从虚拟的数字世界延伸到了真实的物理世界，能够执行“拿起”、“放下”、“指向”、“按压”等动作。你可以继续扩展这个桥接程序，例如集成Reachy的摄像头进行简单的视觉识别（如通过OpenCV识别特定物体），然后将识别结果通过MQTT发送给HA，触发更复杂的自动化流程。或者，为Reachy Mini创建更复杂的“服务”，在HA中直接调用如reachy.pick_up_object这样的高级抽象指令，由桥接程序分解为一系列底层关节运动。这个项目的天花板，取决于你的想象力和对这两个平台的理解深度。