Video Module#

This module provides a high-level interface for accessing camera frames and streaming video from the robot or an attached webcam. Users should interact only with VideoModule and should not use lower-level camera or streaming classes directly.

class go2.modules.video.video_module.VideoModule(camera_source: CameraSource | CameraGroup, stream_config: StreamConfig = StreamConfig(width=640, height=480, fps=30, host='0.0.0.0', port=8080))[source]#

Bases: DogModule

VideoModule provides a simple API for:
  • Accessing camera frames

  • Streaming video to a browser via WebRTC

Users should not access or construct this class directly. Rather, they should access it through the Go2Controller instance.

Parameters:
  • camera_source (CameraSource or CameraGroup) –

    • A CameraSource instance provided by CameraSourceFactory

    • A CameraGroup instance provided by CameraSourceFactory

  • stream_config (StreamConfig) –

    • A StreamConfig instance provided by the user

Notes

get_frames() FrameResult | MultiFrameResult[source]#

Retrieve the latest frame(s) from all cameras. FrameResult objects will be empty if no frame was currently available.

Returns:

  • FrameResult – If constructed with a single CameraSource.

  • MultiFrameResult – If constructed with a CameraGroup.

get_stream_server_local_ip() str[source]#

Get the local IP address of the stream server.

Raises:

RuntimeError – If the stream server is not running.

get_stream_server_port() int[source]#

Get the port number of the stream server.

Raises:

RuntimeError – If the stream server is not running.

get_stream_url() str[source]#

Return the full WebRTC stream URL for this machine.

Raises:

RuntimeError – If the server has not been started.

send_frame(frame: ndarray) None[source]#

Send a video frame to connected streaming clients.

Parameters:

frame (numpy.ndarray) – An image frame to stream (typically obtained from get_frames()).

Raises:

RuntimeError – If the stream server has not been started.

start_stream_server() None[source]#

Start the video streaming server.

This launches a local WebRTC server that allows video frames to be viewed in a browser. Only clients connected to the same subnet as the robot can access the stream.

Raises:

RuntimeError – If the stream server has already been started.

class go2.modules.video.sources.camera_source_factory.CameraSourceFactory[source]#

Bases: object

Factory class for creating camera sources.

CameraSourceFactory provides a simple, safe way for users to select which camera hardware to use without needing to understand low-level camera implementations.

The returned objects are compatible with VideoModule and should not be used directly.

Notes

Users should not instantiate camera source classes directly. Always use this factory instead.

static create_camera_group(sources: Dict[str, CameraSource]) CameraGroup[source]#

Create a named group of camera sources for multi-camera setups.

Parameters:

sources (Dict[str, CameraSource]) – Mapping of camera name to camera source. Names are used to key frames in the returned MultiFrameResult.

static create_depth_camera() CameraSource[source]#

Create an RGB-Depth camera source using an Intel RealSense D345i.

This camera provides both color and depth frames. The returned camera source automatically aligns depth data to the color frame.

The returned FrameResult includes:
  • color: BGR image as a NumPy uint8 array

  • depth: Z16 aligned depth image as a NumPy uint16 array (D345i depth values in millimeters)

Notes

Requires an Intel RealSense camera and the RealSense SDK.

static create_native_camera() CameraSource[source]#

Create a camera source backed by the robot’s internal camera.

This camera is typically used when running code directly on the robot hardware.

The returned frame contains a BGR color image as a NumPy uint8 array.

static create_opencv_camera(camera_index: int = 0) CameraSource[source]#

Create a camera source using OpenCV’s VideoCapture.

This is the recommended option for:
  • USB webcams

  • Laptop cameras

  • Development on personal machines

The returned frame contains a BGR color image as a NumPy uint8 array.

Parameters:

camera_index (int, optional) – Index of the OpenCV camera (default is 0).

static create_virtual_camera() CameraSource[source]#

Create an RGB-Depth camera source inside of the robot simulation.

This camera provides both color and depth frames. The returned camera source automatically aligns depth data to the color frame.

The returned FrameResult includes:
  • color: BGR image as a NumPy uint8 array

  • depth: Depth image as a NumPy uint16 array

Notes

Runs only within the Mujoco Simulation.

class go2.modules.video.frame_result.FrameStatus(value)[source]#

Bases: Enum

Describes the state of a FrameResult object.

OK = 1#

Frame data is present and usable

PENDING = 2#

Source initialised but no frame was currently available in a frame buffer

class go2.modules.video.frame_result.FrameResult(status: FrameStatus = FrameStatus.PENDING, color: ndarray | None = None, depth: ndarray | None = None)[source]#

Bases: object

Holds one captured moment from a single camera source.

color: ndarray | None = None#

BGR color image (H, W, 3). Always present for color/RGB cameras.

classmethod color_and_depth(color: ndarray, depth: ndarray) FrameResult[source]#

Source provides both channels (e.g. RealSense aligned frames).

Notes

Static factory method for internal use only. Should not be called by the user.

classmethod color_only(color: ndarray) FrameResult[source]#

Source is RGB-only; depth will never be present.

Notes

Static factory method for internal use only. Should not be called by the user.

depth: ndarray | None = None#

Depth image (H, W) in uint16. Only present for depth cameras.

classmethod depth_only(depth: ndarray) FrameResult[source]#

Source is depth-only; color will never be present.

Notes

Static factory method for internal use only. Should not be called by the user.

has_color() bool[source]#

True if a valid rgb frame is present.

has_depth() bool[source]#

True if a valid depth frame is present.

is_fully_valid() bool[source]#

True only when status is OK and all channels are present.

classmethod pending() FrameResult[source]#

Source is initialised but the capture buffer hasn’t filled yet.

Notes

Static factory method for internal use only. Should not be called by the user.

status: FrameStatus = 2#
class go2.modules.video.frame_result.MultiFrameResult(frames: Dict[str, FrameResult])[source]#

Bases: object

Holds frame results from a group of named cameras.

frames: Dict[str, FrameResult]#

Maps camera name to its latest FrameResult, or an empty FrameResult if that camera had no frame ready this cycle.

is_fully_valid() bool[source]#

True only when every slot has all valid frames

pending_slots() list[str][source]#

Slots that are initialised but haven’t produced a frame yet.

valid_frames() dict[str, FrameResult][source]#

Only slots whose FrameResult.is_valid is True.

class go2.modules.video.streaming.stream_config.StreamConfig(width: int = 640, height: int = 480, fps: int = 30, host: str = '0.0.0.0', port: int = 8080)[source]#

Bases: object

Configration object that can be passed into VideoModule upon module addition.

fps: int = 30#

Target broadcast frame rate (default 30).

height: int = 480#

Height of the broadcast stream in pixels (default 480).

host: str = '0.0.0.0'#

The hostname or IP to bind the server (default “0.0.0.0”).

port: int = 8080#

HTTP port for the stream server (default 8080).

width: int = 640#

Width of the broadcast stream in pixels (default 640).