3. Secondary Development Interface Overview

Most modules in the above software system use AimRT as the underlying framework. AimRT is a robot runtime framework comparable to ROS2, open-sourced by Zhiyuan Robotics in 2024/09. AimRT provides two basic underlying communication methods: Channel and RPC (for more details, see the AimRT official documentation). Each communication method can correspond to multiple different communication backends.

To reduce the cognitive burden for secondary development, we provide two simple and general-purpose interfaces: HTTP JSON RPC and ROS2 Topic. HTTP JSON RPC is suitable for low-frequency, many-to-one calls, while ROS2 Topic is suitable for high-frequency, many-to-many calls. The following diagram shows the communication relationship between the two.

Regarding programming languages, there are no restrictions for secondary development. As long as you can send and receive HTTP requests, you can call all HTTP JSON RPC interfaces; as long as you support ROS 2 Jazzy (based on Fast DDS), you can call all native interfaces by subscribing and publishing ROS 2 Topics.

Here are two concrete interface call examples to help clarify the interface usage:

3.1 HTTP JSON RPC

Below is a command-line RPC request example for querying silent mode:

bash

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

Explanation:

10.42.10.10 is the IP address of the robot's HDU in the internal network segment; 10.42.10.12 is the IP address of the MDU in the same segment; and 10.42.10.11 is the IP address of the ADU in the same segment. The interface URL in the documentation will clearly indicate which controller should be accessed. For remote calls, replace it with the reachable IP of the corresponding device in the current network (you can check it in AimMaster or by running ip a on the target device).
59301 is the HTTP port of the corresponding service, usually provided in the detailed interface example.
content-type:application/json is the request header, indicating the request body is in JSON format. This header must be included.
aimdk.protocol.AgentControlService/GetVoiceEnable is the RPC interface name for querying silent mode. The specific names of each interface can be found in the corresponding documentation and examples.
'{}' is the request body. Here, an empty object means all fields use default values. If you need to pass specific fields, you can provide a JSON string, e.g., '{"bms_state": 1}' means passing the field bms_state with value 1 (for example only; refer to the documentation and examples for actual field names and values).

3.2 ROS2 Topic

3.2.1 Environment Variable Setup

Topic interfaces use native ROS2 topics. You must set some environment variables before calling any ROS2 Topic interface, as follows:

bash

 source /agibot/software/v0/entry/env/env.sh

3.2.2 QoS Settings

Also note that the default QoS for onboard topics is:

yaml

  history: keep_last
  depth: 10
  reliability: best_effort

When using, keep the QoS consistent to ensure smooth communication.

3.2.3 Example Usage

Below is an example script for subscribing to joint state ROS2 Topic.

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)
    # Arm joint state: /motion/control/arm_joint_state
    # Neck joint state: /motion/control/neck_joint_state
    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()