6. Quick Start for Secondary Development
6. Quick Start for Secondary Development
This section demonstrates how to quickly run your first secondary development program through several simple example programs, to establish a basic understanding.
The quick start tutorial takes the example of deploying the secondary development program on the HDU and using a Python venv virtual environment.
6.1 Prerequisites
6.1.1 Connect to HDU
- Confirm IP and connect via WiFi
The development machine and the robot need to be connected to the same wireless network. Obtain the robot's IP from the WLAN interface in AimMaster's settings (this is the WiFi IP of the HDU). After connecting via SSH, you can jump to the MDU via 10.42.10.12.
6.1.2 Create Deployment Program Folder
The recommended program deployment directory is /agibot/. The onboard disk cleanup module will periodically clean up space, but this folder is whitelisted and will not be cleaned.
Placing programs or data in other paths may cause the partition to fill up or files to be accidentally deleted. Please strictly follow the convention of using the /agibot/ directory.
mkdir -p /agibot/data/home/agi/Desktop
If you encounter permission issues, you can use the following commands to create the folder:
sudo mkdir -p /agibot/data/home/agi/Desktop
sudo chown -R agi:agi /agibot/data/home/agi
6.1.3 Create Python Virtual Environment
cd /agibot/data/home/agi/Desktop
python3 -m venv mydev
source mydev/bin/activate
All subsequent examples assume that the virtual environment has been created and activated, and this step will not be repeated.
6.2 Example: Query Silent Mode Program
This example uses an HTTP request to query silent mode.
Run the following script directly in the ORIN command line:
curl --location --request POST 'http://10.42.10.10:59301/rpc/aimdk.protocol.AgentControlService/GetVoiceEnable' \
--header 'Content-Type: application/json' \
--data-raw '{}'
The returned result should be as follows (the meaning of each field is explained in detail in later interface sections):
{"header":{"code":"0","msg":"GetVoiceEnable successfully","trace_id":"","domin":""},"enable_voice":true}
6.3 Example: Get Robot Upper Limb Joint State
Save the following code as /agibot/data/home/agi/Desktop/joint_state.py on the HDU.
#!/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()
Then execute the following commands:
source /agibot/software/v0/entry/env/env.sh
python3 /agibot/data/home/agi/Desktop/joint_state.py
6.4 Example: Control Robot Motion
To control robot motion, switch the robot action mode to MOTION. To switch the robot motion control mode, save the following code as S_SetAction.py on the HDU.
#!/usr/bin/env python3
## Function: set 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()
Run the following command to switch the robot motion control mode and select the PD_STAND mode (mode code 09):
python3 /agibot/data/home/agi/Desktop/S_SetAction.py
Then switch to MOTION mode (mode code 05):
python3 /agibot/data/home/agi/Desktop/S_SetAction.py
Before sending joint control commands, stop the motion_player module. This operation must be performed on the MDU. First, switch to the MDU:
ssh agi@10.42.10.12
Then use the following commands to stop motion_player:
# Stop 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"}'
# Restart 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"}'
Switch back to the HDU for motion control:
ssh agi@10.42.10.10
Save the following code as /agibot/data/home/agi/Desktop/neck.py on the HDU.
#!/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()
# Position control only
msg.name = ["head_yaw_joint", "head_pitch_joint"]
msg.position = [0.0, 0.5]
# These fields have no actual effect, but must be set to avoid motion control crashes
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()
Then run the following commands:
source /opt/agibot/entry/env/env.sh
source /opt/ros/jazzy/setup.bash
python3 /agibot/data/home/agi/Desktop/neck.py
6.5 Example: Robot TTS Broadcast
Save the following script as tts_broadcast.sh (corresponding to 7.2.2 TTS Playback RPC Interface).
#!/bin/bash
if [ $# -eq 0 ]; then
echo "arg error, need text, ./tts_broadcast.sh 'test text'"
exit 0
fi
# Play TTS text
# ./tts_broadcast.sh "test text"
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}'
Run the following command. The text inside the double quotes is the test content:
bash tts_broadcast.sh "Hello, I am Expedition A3"