感知模块

from typing import Annotated

import cv2
import numpy as np
import typer
from x2robot import connect


def main(
    server: Annotated[
        str,
        typer.Option(
            help="SDK Server 地址，例如：192.168.10.1:50051"
        ),
    ] = "localhost:50051",
):
    robot = connect(f"x2://{server}")
    image = robot.head_camera.get_rgb_image()

    if not image or not image.data:
        raise RuntimeError("未获取到头部 RGB 图像数据")

    image_array = np.frombuffer(
        bytes(image.data),
        dtype=np.uint8,
    )
    frame = cv2.imdecode(
        image_array,
        cv2.IMREAD_COLOR,
    )

    if frame is None:
        raise RuntimeError(
            f"无法解码头部 RGB 图像，"
            f"format={image.format!r}"
        )

    window_name = "Head RGB"
    try:
        cv2.imshow(window_name, frame)
        print("按任意键关闭图像窗口。")
        cv2.waitKey(0)
    finally:
        cv2.destroyAllWindows()


if __name__ == "__main__":
    typer.run(main)

python3 head_rgb_image.py \
  --server 192.168.10.1:50051

image = robot.head_camera.get_depth_image()

images = robot.head_camera.get_rgb_video_stream()
for image in images:
    # 处理响应
    pass

images = robot.head_camera.get_depth_video_stream()
for image in images:
    # 处理响应
    pass

image = robot.head_camera.get_left_eye_image()

images = robot.head_camera.get_left_eye_video_stream()
for image in images:
    # 处理响应
    pass

image = robot.left_arm_camera.get_raw_image()

images = robot.left_arm_camera.get_video_stream()

for image in images:
    print(
        f"format={image.format}, "
        f"bytes={len(image.data)}"
    )

image = robot.left_arm_camera.get_elbow_image()

images = robot.left_arm_camera.get_elbow_video_stream()
for image in images:
    # 处理响应
    pass

image = robot.right_arm_camera.get_raw_image()

images = robot.right_arm_camera.get_video_stream()

for image in images:
    print(
        f"format={image.format}, "
        f"bytes={len(image.data)}"
    )

image = robot.right_arm_camera.get_elbow_image()

images = robot.right_arm_camera.get_elbow_video_stream()
for image in images:
    # 处理响应
    pass

scan_data = robot.radar.get_laser_scan()

print(
    f"Angle Range: "
    f"[{scan_data.angle_min:.3f}, "
    f"{scan_data.angle_max:.3f}] rad"
)
print(
    f"Angle Increment: "
    f"{scan_data.angle_increment:.6f} rad"
)
print(
    f"Time Increment: "
    f"{scan_data.time_increment:.9f} s"
)

for scan_data in robot.radar.get_laser_scan_stream():
    print(
        f"Angle Range: "
        f"[{scan_data.angle_min:.3f}, "
        f"{scan_data.angle_max:.3f}] rad"
    )
    print(
        f"Angle Increment: "
        f"{scan_data.angle_increment:.6f} rad"
    )
    print(
        f"Time Increment: "
        f"{scan_data.time_increment:.9f} s"
    )

imu_data = robot.imu.get_chassis_imu()

print(f"Orientation Quaternion: {imu_data.orientation}")
print(f"Angular Velocity: {imu_data.angular_velocity}")
print(f"Linear Acceleration: {imu_data.linear_acceleration}")

for imu_data in robot.imu.get_chassis_imu_stream():
    print(f"Orientation Quaternion: {imu_data.orientation}")
    print(f"Angular Velocity: {imu_data.angular_velocity}")
    print(f"Linear Acceleration: {imu_data.linear_acceleration}")

range_data = robot.tof.get_chassis_tof_1()

print(f"Distance: {range_data.range:.3f} m")
print(f"Radiation Type: {range_data.radiation_type}")
print(
    f"Field of View: {range_data.field_of_view:.3f} rad "
    f"({math.degrees(range_data.field_of_view):.1f}°)"
)
print(f"Min Range: {range_data.min_range:.2f} m")
print(f"Max Range: {range_data.max_range:.2f} m")

for range_data in robot.tof.get_chassis_tof_1_stream():
    print(f"Distance: {range_data.range:.3f} m")
    print(f"Radiation Type: {range_data.radiation_type}")
    print(
        f"Field of View: {range_data.field_of_view:.3f} rad "
        f"({math.degrees(range_data.field_of_view):.1f}°)"
    )
    print(f"Min Range: {range_data.min_range:.2f} m")
    print(f"Max Range: {range_data.max_range:.2f} m")

import math

range_data = (
    robot.ultrasonic
    .get_chassis_ultrasonic_1()
)

print(f"Distance: {range_data.range:.3f} m")
print(f"Radiation Type: {range_data.radiation_type}")
print(
    f"Field of View: {range_data.field_of_view:.3f} rad "
    f"({math.degrees(range_data.field_of_view):.1f}°)"
)
print(f"Min Range: {range_data.min_range:.2f} m")
print(f"Max Range: {range_data.max_range:.2f} m")

import math

for range_data in (
    robot.ultrasonic
    .get_chassis_ultrasonic_1_stream()
):
    print(f"Distance: {range_data.range:.3f} m")
    print(f"Radiation Type: {range_data.radiation_type}")
    print(
        f"Field of View: {range_data.field_of_view:.3f} rad "
        f"({math.degrees(range_data.field_of_view):.1f}°)"
    )
    print(f"Min Range: {range_data.min_range:.2f} m")
    print(f"Max Range: {range_data.max_range:.2f} m")

from x2robot import connect

robot = connect("x2://192.168.10.1:50051")

msg = robot.depth_points.get_chassis_depth_points(
    timeout=5
)

print(f"width={msg.width}, height={msg.height}")
print(
    f"point_step={msg.point_step}, "
    f"row_step={msg.row_step}"
)
print(
    "fields="
    f"{[field.name for field in msg.fields]}"
)
print(f"data_bytes={len(msg.data)}")
print(f"is_dense={msg.is_dense}")
print(f"is_bigendian={msg.is_bigendian}")

frame_id = (
    msg.header.frame_id
    if msg.header is not None
    else "<not provided>"
)
print(f"frame_id={frame_id}")

from x2robot import connect

robot = connect("x2://192.168.10.1:50051")

try:
    for msg in (
        robot.depth_points
        .get_chassis_depth_points_stream(
            timeout=None
        )
    ):
        print(
            f"width={msg.width}, "
            f"height={msg.height}, "
            f"data_bytes={len(msg.data)}"
        )
except KeyboardInterrupt:
    print("Depth point stream stopped.")

# 单次读取和解析
python3 examples/depth_points.py   --action single   --server 192.168.10.1:50051

# 流式读取和解析
python3 examples/depth_points.py   --action stream   --server 192.168.10.1:50051