智元灵犀X2机器人二开02:端侧环境与第一个 AIMDK 程序

本篇产出：一个可运行的 ROS 2 节点，并沉淀出一个后续所有端侧代码都会复用的 service 重试封装。
前置：第 01 篇；一台灵犀 X2（或可用仿真环境）；一台 Linux 开发机。

---

一、上电与安全准备

按官方《开机指南》操作。上电后机器人自动启动系统，默认进入自然语音交互模式，内置交互模块会占用音频输入输出流——如果后续要做自研语音（第 05、06 篇），需要先关闭它。

三个工具按优先级准备：

工具	用途	必要性
遥控手柄	随时接管运动控制	调试运动类程序的必备兜底
Agibot Go APP	连接机器人、查看状态、基础配置	首次上手建议使用
手册/关机流程	按官方《关机指南》操作	不要直接断电

⚠️ 调试任何涉及运动的程序前，先确认环境安全开阔、能随时急停、手柄在手边。

---

二、网络接入与链路自检

二开程序部署在开发计算单元上，通过千兆以太网（RJ45）访问 Orin NX 与 RK3588 或连接到机器人所在的同一网络。

graph LR
    DEV["开发机"] -->|"RJ45 / 交换机"| DCU["X2 开发计算单元<br/>Orin NX / RK3588"]
    DCU -.->|"可访问，禁止部署"| PC1["运控计算单元 PC1<br/>10.0.1.40"]

链路自检按顺序执行，任一步失败就停下排查，不要跳步：

# 1. 网络可达，机器人IP可通过手机APP链接后查看
ping <机器人IP>

# 2. 远程登录（若开放）
ssh <user>@<机器人IP>

# 3. ROS 2 环境与节点
source /opt/ros/humble/setup.bash
ros2 node list

# 4. 确认 AIMDK 服务已注册
ros2 service list | grep aimdk

第 4 步是判定依据：能列出 /aimdk_5Fmsgs/srv/... 开头的服务，说明 ROS 2 通信链路已通，可以进入下一步。

---

三、环境安装

二开环境可使用机器人上的开发机或自行准备Ubuntu 22.04+安装ROS2 Humble+Aimdk。

可以参考下面的步骤验证是否安装好:

source /opt/ros/humble/setup.bash
python3 --version                      # 需 Python 3 环境
python3 -c "import aimdk_msgs; print('aimdk_msgs OK')"

---

四、跑通官方示例

在写自己的代码之前，先用官方示例确认环境。官方提供 Python 与 C++ 两套示例，覆盖控制模块与交互模块。

1. wait_for_service 是否超时
2. 调用是否返回 code == 0
3. 打印出的数据结构长什么样

# 设置环境变量 (开发计算单元直接运行模式可跳过本步)
source /opt/ros/humble/setup.bash

# 构建 SDK
# 假设拷贝并解压后到 X2 AimDK 项层目录名称为 aimdk
cd ./aimdk/
colcon build

# 设置环境变量
source /opt/ros/humble/setup.bash
source install/local_setup.bash

# Python 示例
ros2 run py_examples get_mc_action

---

五、第一个自己的节点

下面我们实现一个节点：查询机器人上部署的灵创动作资源列表。

覆盖 ROS 2 节点创建、service 调用，以及跨板 service 的重试封装——第 01 篇提到的已知缺陷，在这里一次性解决，后续所有端侧代码都能复用这个封装。

1.完整代码

#!/usr/bin/env python3
"""查询机器人上部署的灵创动作资源列表"""

import rclpy
from rclpy.node import Node
from aimdk_msgs.srv import GetRobotResources

SERVICE_NAME = "/aimdk_5Fmsgs/srv/GetRobotResources"


class ResourceQuery(Node):
    def __init__(self):
        super().__init__("resource_query")
        self.cli = self.create_client(GetRobotResources, SERVICE_NAME)

    def call_with_retry(self, req, timeout_sec: float = 3.0, max_retry: int = 3):
        """带重试的 service 调用。

        跨板通信存在偶发失败，官方要求必须加保护机制。
        三类失败需要分别处理：服务未就绪、调用超时、业务返回码非 0。
        """
        last_err = None

        for attempt in range(1, max_retry + 1):
            if not self.cli.wait_for_service(timeout_sec=timeout_sec):
                last_err = "service not ready"
                self.get_logger().warn(
                    f"[{attempt}/{max_retry}] 服务未就绪，重试")
                continue

            future = self.cli.call_async(req)
            rclpy.spin_until_future_complete(self, future, timeout_sec=timeout_sec)
            res = future.result()

            if res is None:
                last_err = "timeout / no response"
                self.get_logger().warn(
                    f"[{attempt}/{max_retry}] 无响应，重试")
                continue

            code = res.header.header.code
            if code != 0:
                # 业务层失败：重试无意义，直接返回
                self.get_logger().error(
                    f"业务错误 code={code}, msg={res.header.message}")
                return None

            return res

        self.get_logger().error(f"重试耗尽：{last_err}")
        return None


def main():
    rclpy.init()
    node = ResourceQuery()
    try:
        res = node.call_with_retry(GetRobotResources.Request())
        if res is None:
            node.get_logger().error("获取资源失败")
            return

        resources = res.robot_resources
        print(f"\n共加载 {len(resources)} 个灵创动作资源：\n")
        for i, item in enumerate(resources, 1):
            print(f"{i:>3}. {item.current_version.name}")
            print(f"     resource_key = {item.resource_key}")
            print(f"     version      = {item.current_version.version}")
        print()
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

2.重试

这里区分了三类性质不同的失败：

失败类型	判定条件	处理策略
服务未就绪	wait_for_service 超时	重试——服务可能尚未启动
调用超时	future.result() 为 None	重试——跨板链路抖动
业务失败	header.header.code != 0	不重试——参数或资源问题，重试不会变好

第三类是很多人容易写错的地方：把所有非零返回都拿去重试，白白拉长失败耗时。

3.返回值处理

注意 res.header.header.code 这个双层结构：

res.header          # 消息头
res.header.header   # 业务状态头
res.header.header.code   # 0 = 成功
res.header.message       # 失败原因

外层 header 是 AIMDK 消息的统一封装，内层 header 承载业务状态。第 03 篇的灵创动作播放、导航调用都会复用这个结构。

六、故障对照表

现象	根因	处理
wait_for_service 持续超时	服务名拼错，或对应模块未启动	用 ros2 service list | grep aimdk 对照
偶发调用失败、重试即恢复	跨板通信抖动（已知缺陷）	统一走 call_with_retry，禁止裸调
future.result() 返回 None	超时或链路异常	判空后重试，不要直接取属性
返回 code != 0	业务层失败（如资源不存在）	读 header.message 定位，不要重试
进程无法正常退出	rclpy 生命周期未收尾	用 try/finally 保证 destroy_node() + rclpy.shutdown()

---

总结

• 端侧开发三件套：ROS 2 Humble + aimdk_msgs + 开发计算单元。
• 命名规范：服务 /aimdk_5Fmsgs/srv/、话题 /aima/、消息包 aimdk_msgs。
• 跨板 service 必须包一层重试，且要区分”服务未就绪 / 调用超时 / 业务失败”三类——第三类不应重试。
• 返回码为双层结构：res.header.header.code。
• 环境自检用一下命令查询ros2服务 ros2 service list | grep aimdk。

下一篇进入端侧接口：控制模块七类接口 + 交互模块三类接口，最后编排一个”走一段 → 播灵创动作 → 说一句话”的动作序列。

---

声明

• 端侧SDK文档
• 灵心SDK文档

本文首发智元AIMA平台，并在博客同步