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

ping <机器人IP>


ssh <user>@<机器人IP>


source /opt/ros/humble/setup.bash
ros2 node list


ros2 service list | grep aimdk

1
2
3

source /opt/ros/humble/setup.bash
python3 --version                      
python3 -c "import aimdk_msgs; print('aimdk_msgs OK')"

1
2
3
4
5
6
7
8
9
10
11
12
13
14

source /opt/ros/humble/setup.bash



cd ./aimdk/
colcon build


source /opt/ros/humble/setup.bash
source install/local_setup.bash


ros2 run py_examples get_mc_action

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

"""查询机器人上部署的灵创动作资源列表"""

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()

1
2
3
4

res.header          
res.header.header   
res.header.header.code   
res.header.message