AIMA开放平台文档中心AimDK A3-ultra

6. 二次开发快速上手

6. 二次开发快速上手

本章节将通过几个简单的示例程序演示如何快速跑起第一个二次开发程序，对此建立一个基本认知。

快速上手教程以二次开发程序部署在 HDU 上且使用 Python venv 虚拟环境为例。

6.1 前置条件

6.1.1 连接 HDU

  1. 通过WiFi 确认IP并连接

需要开发机和机器人连接到同一个无线网络，从 AimMaster 的设置-无线局域网界面上获取机器人 ip，实际为HDU的 WiFi的 ip，ssh 连接上后再通过 10.42.10.12 可跳转到 MDU

6.1.2 创建部署程序文件夹

推荐的程序部署目录为 /agibot/，机上磁盘清理模块会定时清理空间，该文件夹为白名单文件夹不会被清理。

程序、数据放置于其他路径可能造成分区写满、文件被误删等后果，请严格遵循使用 /agibot/ 目录的约定。

bash
mkdir -p /agibot/data/home/agi/Desktop

如遇权限问题，可考虑使用如下命令进行文件夹创建

bash
sudo mkdir -p /agibot/data/home/agi/Desktop
sudo chown -R agi:agi /agibot/data/home/agi

6.1.3 创建 Python 虚拟环境

bash
cd /agibot/data/home/agi/Desktop
python3 -m venv mydev
source mydev/bin/activate

后续示例都假设已创建虚拟环境并激活，不再重复此步骤。

6.2 查询静默模式程序运行示例

本示例采用 HTTP 请求查询静默模式。

直接在 ORIN 命令行运行如下脚本：

yaml
curl --location --request POST 'http://10.42.10.10:59301/rpc/aimdk.protocol.AgentControlService/GetVoiceEnable' \
     --header 'Content-Type: application/json' \
     --data-raw '{}'

返回结果应如下所示（具体字段含义在后续接口章节中有详细解释）：

python
{"header":{"code":"0","msg":"GetVoiceEnable successfully","trace_id":"","domin":""},"enable_voice":true}

6.3 获取机器人上肢关节状态示例

将以下代码在 HDU 上保存为 /agibot/data/home/agi/Desktop/joint_state.py 文件。

python
#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
from sensor_msgs.msg import JointState

class JointStateSubscriber(Node):
    def __init__(self, topic_name: str):
        super().__init__("joint_state_subscriber")

        self.topic_name = topic_name

        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST, depth=10, reliability=QoSReliabilityPolicy.BEST_EFFORT
        )

        self.subscription = self.create_subscription(JointState, topic_name, self.listener_callback, qos_profile)

    def listener_callback(self, msg: JointState):
        self.get_logger().info(f"=== Received {self.topic_name} ===")
        self.get_logger().info(f"  header: {msg.header}")
        self.get_logger().info(f"  name: {msg.name}")
        self.get_logger().info(f"  position: {msg.position}")
        self.get_logger().info(f"  velocity: {msg.velocity}")
        self.get_logger().info(f"  effort: {msg.effort}")

def main(args=None):
    rclpy.init(args=args)
    joint_state_node = JointStateSubscriber("/motion/control/arm_joint_state")
    try:
        rclpy.spin(joint_state_node)
    except KeyboardInterrupt:
        pass
    finally:
        joint_state_node.destroy_node()
        rclpy.shutdown()

if __name__ == "__main__":
    main()

然后执行如下命令

yaml
source /agibot/software/v0/entry/env/env.sh
python3 /agibot/data/home/agi/Desktop/joint_state.py

6.4 控制机器人动作示例

控制机器人运动需要将机器人的动作模式切换成 MOTION。为了切换机器人的运动控制模式，将如下代码在 HDU 上保存为 S_SetAction.py 文件。

python
#!/usr/bin/env python3

## 功能：设置 action

import json
import requests
from datetime import datetime

def create_header():
    now = datetime.utcnow()
    header = {
        "timestamp": {
            "seconds": int(now.timestamp()),
            "nanos": now.microsecond * 1000,
            "ms_since_epoch": int(now.timestamp() * 1000),
        },
        "control_source": "ControlSource_SAFE",
        "uuid": "",
        "trace_id": "user_McScript",
        "domin": "",
    }
    return header

def get_available_actions():
    url = f"http://10.42.10.12:56322/rpc/aimdk.protocol.MotionControlActionService/GetAvailableActions"
    headers = {'Content-Type': 'application/json'}
    response = requests.post(url, headers=headers, json={})
    response.raise_for_status()

    return response.json().get('commands', [])

def select_action(actions = None):
    try:
        if actions is None:
            actions = get_available_actions()

        if len(actions) == 0:
            print("No actions available.")
            exit(0)
        elif len(actions) == 1:
            return actions[0]["ext_action"]
        else:
            print("Please select an action:")
            for index, action in enumerate(actions, 1):
                print(f" {index:02d}: {action['ext_action']}")
            choice = input("Enter the number corresponding to the desired action: ")
            if choice.isdigit() and int(choice) >= 1 and int(choice) <= len(actions):
                return set_action(actions[int(choice) - 1])
            elif isinstance(choice, str) and choice == "q":
                exit(0)
            else:
                print("Invalid choice.")
                exit(1)
    except Exception as e:
        print(f"Error: {e}")
        exit(1)

def set_action(action):
    url = f"http://10.42.10.12:56322/rpc/aimdk.protocol.MotionControlActionService/SetAction"
    headers = {"Content-Type": "application/json"}
    payload = {
        "header": create_header(),
        "command": action,
    }
    response = requests.Session().post(url, headers=headers, json=payload)
    return response.json()


def pretty_print_json(json_data):
    print("Response:")
    print(json.dumps(json_data, indent=2, ensure_ascii=False))


def main():
    response = select_action()
    pretty_print_json(response)
    requests.Session().close()

if __name__ == "__main__":
    main()

通过如下命令进行机器人的运控模式切换，选择 PD_STAND 模式（模式代码为 09）：

bash
python3 /agibot/data/home/agi/Desktop/S_SetAction.py

接着切换至 MOTION 模式（模式代码为 05）：

bash
python3 /agibot/data/home/agi/Desktop/S_SetAction.py

下发关节控制命令前需关闭 motion_player 模块。该操作需要在 MDU 下进行，首先使用如下命令切换到 MDU：

bash
ssh agi@10.42.10.12

接着使用如下命令关闭 motion_player 模块：

bash
# 关闭 motion_player 命令
curl -i -H 'content-type:application/json' -X POST 'http://127.0.0.1:50080/json/stop_app' -d '{"app_name":"motion_player"}'
# 重启 motion_player 命令
curl -i -H 'content-type:application/json' -X POST 'http://127.0.0.1:50080/json/start_app' -d '{"app_name": "motion_player"}'

切换回 HDU 进行运动控制操作：

bash
ssh agi@10.42.10.10

将以下代码在 HDU 上保存为 /agibot/data/home/agi/Desktop/neck.py 文件。

python
#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
from sensor_msgs.msg import JointState


class NeckPublisher(Node):
    def __init__(self, neck_topic_name: str):
        super().__init__("neck_publisher")

        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=QoSReliabilityPolicy.BEST_EFFORT,
        )

        self.publisher = self.create_publisher(JointState, neck_topic_name, qos_profile)

        timer_period = 0.02
        self.timer = self.create_timer(timer_period, self.timer_callback)

    def timer_callback(self):
        msg = JointState()
        msg.header.stamp = self.get_clock().now().to_msg()

        # 仅支持位控
        msg.name = ["head_yaw_joint", "head_pitch_joint"]
        msg.position = [0.0, 0.5]

        # 以下字段无实际作用，但必须设置，否则可能导致运控 crash
        msg.velocity = [0.0, 0.0]
        msg.effort = [0.0, 0.0]

        self.publisher.publish(msg)


def main(args=None):
    rclpy.init(args=args)
    neck_publisher = NeckPublisher("/motion/control/neck_joint_command")
    try:
        rclpy.spin(neck_publisher)
    except KeyboardInterrupt:
        pass
    finally:
        neck_publisher.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

然后执行如下命令：

bash
source /opt/agibot/entry/env/env.sh
source /opt/ros/jazzy/setup.bash
python3 /agibot/data/home/agi/Desktop/neck.py

6.5 机器人 TTS 语音播报示例

将以下脚本保存为 tts_broadcast.sh 文件（对应 7.2.2 TTS 播报 RPC 接口）。

bash
#!/bin/bash

if [ $# -eq 0 ]; then
    echo "arg error, need text, ./tts_broadcast.sh '测试文本'"
    exit 0
fi

# 播放 tts 文本
# ./tts_broadcast.sh "测试文本"
curl -i \
    -H 'content-type:application/json' \
    -X POST 'http://10.42.10.10:59301/rpc/aimdk.protocol.TTSService/PlayTTS' \
    -d '{"text":"'$1'","priority_level":"INTERACTION_L6","domain":"example", "trace_id":"hafhjkqwjwefk", "is_interrupted":true}'

运行以下命令，双引号中的是测试文本：

bash
bash tts_broadcast.sh "你好，我是远征A3"