编程笔记

lifelong learning & practice makes perfect

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

发表于 更新于 分类于
本文字数： 4.7k 阅读时长 ≈ 4 分钟
本文介绍灵犀X2二次开发的整体架构，包括端侧AIMDK和云端AgentSDK，并给出了需求判断依据和系统全景。

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

一、上电与安全准备

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

三个工具按优先级准备：

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

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

二、网络接入与链路自检

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

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

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

1
2
3
4
5
6
7
8
9
10
11
12
# 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。

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

1
2
3
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. 打印出的数据结构长什么样
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 设置环境变量 (开发计算单元直接运行模式可跳过本步)
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.完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
#!/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.重试

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

call_with_retry 重试流程 带重试的 ROS 2 服务调用流程：等待服务就绪、异步调用、校验返回值与业务错误码，失败时按重试次数循环。 call_with_retry(req) wait_for_service 超时？ call_async spin_until_future_complete future.result() 为 None？ header.header.code == 0？ 返回 res 立即返回 None 重试无意义 attempt < max_retry？ 重试耗尽，返回 None 是（服务未就绪） 是（调用超时） 是 否 否 是 否 否
失败类型 判定条件 处理策略
服务未就绪 wait_for_service 超时 重试——服务可能尚未启动
调用超时 future.result() 为 None 重试——跨板链路抖动
业务失败 header.header.code != 0 不重试——参数或资源问题，重试不会变好

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

3.返回值处理

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

1
2
3
4
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。

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

声明

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

欢迎关注我的其它发布渠道