Default Go2 camera: should we use /frontvideostream or /api/videohub/request? Thank you ^_^

[AlexWUrobot]

Check this before filling out

• [Yes] I have checked that this question has not been previously resolved.

What is your question? What do you need help with?

Dear Developers,

May I ask how to subscribe to the image_raw topic from the Unitree Go2?

I have installed and sourced the following package:
https://github.com/unitreerobotics/unitree_ros2

Currently, I can see the following:

ros2 topic info /frontvideostream
Type: unitree_go/msg/Go2FrontVideoData
Publisher count: 25
Subscription count: 1

And the message structure:

ros2 interface show unitree_go/msg/Go2FrontVideoData
uint64 time_frame
uint8[] video720p
uint8[] video360p
uint8[] video180p

I also checked the topic frequency:

ros2 topic hz /frontvideostream
average rate: 33.777
    min: 0.000s max: 0.113s std dev: 0.03392s window: 34
average rate: 31.029
    min: 0.000s max: 0.113s std dev: 0.03367s window: 64

However, when I run my Python script, I am not receiving any messages:

python3 go2_camera_viewer.py
running the code......

Could you please advise on the correct way to subscribe to the image_raw feed from the Go2 camera?

Thank you so much !

Here is my code (go2_camera_viewer.py)

import rclpy
from rclpy.node import Node
from unitree_go.msg import Go2FrontVideoData
import cv2
import numpy as np

class FrontVideoSubscriber(Node):
    def __init__(self):
        super().__init__('front_video_subscriber')
        self.subscription = self.create_subscription(
            Go2FrontVideoData,
            '/frontvideostream',
            self.listener_callback,
            10
        )

    def listener_callback(self, msg: Go2FrontVideoData):
        print("receive msg ")
        # Pick the resolution you want
        if len(msg.video720p) > 0:
            np_arr = np.frombuffer(msg.video720p, dtype=np.uint8)
            frame = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)
            if frame is not None:
                cv2.imshow("Go2 Front Camera (720p)", frame)
                cv2.waitKey(1)
            else:
                self.get_logger().warn("Failed to decode 720p frame")
        else:
            self.get_logger().warn("Empty video frame received")

def main(args=None):
    print("running the code ...... ")
    rclpy.init(args=args)
    node = FrontVideoSubscriber()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()
    cv2.destroyAllWindows()

if __name__ == '__main__':
    main()