Streaming Example

This example demonstrates how to programmatically control streaming on your Aria Gen2 device using the Python SDK. You'll learn how to start streaming, implement custom callbacks for real-time data processing, and optionally record streaming data to VRS files.

Quick Start

Run the streaming example script:

python ~/Downloads/projectaria_client_sdk_samples_gen2/device_streaming.py

To save streaming data to a VRS file:

python ~/Downloads/projectaria_client_sdk_samples_gen2/device_streaming.py --record-to-vrs /path

---

What This Example Does

The script performs the following operations:

1. Connects to the device
2. Configures streaming settings (profile, interface)
3. Starts streaming on the device
4. Sets up a streaming receiver with custom callbacks
5. Processes real-time data from all sensors (cameras, IMU, audio, machine perception)
6. Optionally records streaming data to a VRS file

---

Code Walkthrough

Step 1: Import Required Modules

import argparse
import signal
import sys
import aria.sdk_gen2 as sdk_gen2
import aria.stream_receiver as receiver
from projectaria_tools.core.mps import EyeGaze, hand_tracking
from projectaria_tools.core.sensor_data import (
    AudioData, AudioDataRecord, FrontendOutput, ImageData, ImageDataRecord, MotionData,
)

Key Modules:

• aria.sdk_gen2: Main SDK for device control
• aria.stream_receiver: Receives and processes streaming data
• projectaria_tools.core: Data structures for sensor data and machine perception

---

Step 2: Parse Command-Line Arguments

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--record-to-vrs",
        dest="record_to_vrs",
        type=str,
        default="",
        required=False,
        help="Output directory to save the received streaming into VRS",
    )
    return parser.parse_args()

Available Options:

• --record-to-vrs: Optional path to save streaming data as a VRS file
• If not specified, data is only processed in callbacks (not saved)

Important: Data drops can occur with poor streaming connections. The saved VRS file will reflect any data drops that occurred during streaming.

---

Step 3: Connect to Device and Start Streaming

# Create device client
device_client = sdk_gen2.DeviceClient()

# Establish connection to the device
config = sdk_gen2.DeviceClientConfig()
device_client.set_client_config(config)
device = device_client.connect()

# Set recording config with profile name
streaming_config = sdk_gen2.HttpStreamingConfig()
streaming_config.profile_name = "profile9"
device.set_streaming_config(streaming_config)

# Start and stop recording
device.start_streaming()

Streaming Configuration:

• profile_name: Use mp_streaming_demo for smooth visualization with all machine perception data
• streaming_interface: Current support:
  • USB_NCM: USB connection (default)

---

Step 4: Define Data Callbacks

Callbacks are functions that process data as it arrives from the device. Here's how to implement callbacks for each data type:

Image Callback

Processes all camera streams: RGB (1x), SLAM (4x), and Eye Tracking (2x):

def image_callback(image_data: ImageData, image_record: ImageDataRecord):
    """Called for each image frame from any camera."""
    image_array = image_data.to_numpy_array()
    timestamp_ns = image_record.capture_timestamp_ns

    print(f"Received image: shape={image_array.shape}, timestamp={timestamp_ns} ns")

    # Example: Process the image
    # - image_array is a numpy array you can process with OpenCV, PIL, etc.
    # - image_record contains metadata like timestamp, camera ID, exposure

---

Audio Callback

Processes audio data from all 8 microphone channels:

def audio_callback(audio_data: AudioData, audio_record: AudioDataRecord, num_channels: int):
    """Called for each audio data packet."""
    num_samples = len(audio_data.data)
    num_timestamps = len(audio_record.capture_timestamps_ns)

    print(f"Received audio: samples={num_samples}, timestamps={num_timestamps}, channels={num_channels}")

    # Example: Process audio
    # - audio_data.data contains the raw audio samples
    # - audio_record.capture_timestamps_ns contains per-sample timestamps

---

IMU Callback

Processes high-frequency IMU data from both IMU sensors at 800Hz:

def imu_callback(imu_data: MotionData, sensor_label: str):
    """Called for each IMU data sample from imu-left or imu-right."""
    accel = imu_data.accel_msec2  # Acceleration in m/s²
    gyro = imu_data.gyro_radsec   # Gyroscope in rad/s

    print(f"Received {sensor_label}: accel={accel}, gyro={gyro}")

    # Example: Process IMU data
    # - Use for motion tracking.

---

Eye Gaze Callback

Processes eye tracking data with gaze direction and depth:

def eyegaze_callback(eyegaze_data: EyeGaze):
    """Called for each eye gaze estimate."""
    timestamp_sec = eyegaze_data.tracking_timestamp.total_seconds()
    yaw_rad = eyegaze_data.yaw
    pitch_rad = eyegaze_data.pitch
    depth_m = eyegaze_data.depth

    print(f"Eye Gaze: timestamp={timestamp_sec}s, yaw={yaw_rad:.3f}, pitch={pitch_rad:.3f}, depth={depth_m:.3f}m")

    # Example: Use eye gaze data
    # - Track where user is looking
    # - Estimate focus depth
    # - Build attention maps

---

Hand Tracking Callback

Processes hand pose estimates for both left and right hands:

def handtracking_callback(handtracking_data: hand_tracking.HandTrackingResult):
    """Called for each hand tracking estimate."""
    timestamp_sec = handtracking_data.tracking_timestamp.total_seconds()
    print(f"Hand Tracking: timestamp={timestamp_sec}s")

    # Process left hand
    if handtracking_data.left_hand is not None:
        left_hand = handtracking_data.left_hand
        print(f"  Left hand confidence: {left_hand.confidence:.3f}")
        print(f"  Left wrist position: {left_hand.get_wrist_position_device()}")
        print(f"  Left palm position: {left_hand.get_palm_position_device()}")
        if left_hand.wrist_and_palm_normal_device is not None:
            normals = left_hand.wrist_and_palm_normal_device
            print(f"  Left wrist normal: {normals.wrist_normal_device}")
            print(f"  Left palm normal: {normals.palm_normal_device}")
    else:
        print("  Left hand: No data")


    # Process right hand (similar to left hand)
    if handtracking_data.right_hand is not None:
        right_hand = handtracking_data.right_hand
        print(f"  Right hand confidence: {right_hand.confidence:.3f}")
        print(f"  Right wrist position: {right_hand.get_wrist_position_device()}")
        print(f"  Right palm position: {right_hand.get_palm_position_device()}")
        if right_hand.wrist_and_palm_normal_device is not None:
            normals = right_hand.wrist_and_palm_normal_device
            print(f"  Right wrist normal: {normals.wrist_normal_device}")
            print(f"  Right palm normal: {normals.palm_normal_device}")
    else:
        print("  Right hand: No data")

---

VIO Callback

Processes Visual-Inertial Odometry data (6-DOF pose estimates):

def vio_callback(vio_data: FrontendOutput):
    """Called for each VIO pose estimate."""
    timestamp_ns = vio_data.capture_timestamp_ns
    rotation = vio_data.transform_odometry_bodyimu.rotation().log()
    translation = vio_data.transform_odometry_bodyimu.translation()

    print(f"VIO: timestamp={timestamp_ns}ns, rotation={rotation}, translation={translation}")

    # Example: Use VIO data
    # - Track device position and orientation
    # - Build 3D maps
    # - Enable AR applications

---

Step 5: Set Up the Streaming Receiver

The streaming receiver listens for data and dispatches it to your callbacks:

def setup_streaming_receiver(device, record_to_vrs):
    """Configure and start the streaming receiver."""
    # Configure server
    config = sdk_gen2.HttpServerConfig()
    config.address = "0.0.0.0"  # Listen on all interfaces
    config.port = 6768           # Default streaming port

    # Create receiver
    stream_receiver = receiver.StreamReceiver()
    stream_receiver.set_server_config(config)

    # Optional: Record to VRS
    if record_to_vrs != "":
        stream_receiver.record_to_vrs(record_to_vrs)

    # Register all callbacks
    stream_receiver.register_slam_callback(image_callback)
    stream_receiver.register_rgb_callback(image_callback)
    stream_receiver.register_audio_callback(audio_callback)
    stream_receiver.register_eye_gaze_callback(eyegaze_callback)
    stream_receiver.register_hand_pose_callback(handtracking_callback)
    stream_receiver.register_vio_callback(vio_callback)

    # Start receiving data
    stream_receiver.start_server()

    return stream_receiver

Important Configuration Notes:

• Port 6768: Ensure this port is open and not blocked by firewall
• VPN: Disable VPN to allow streaming data to be received
• 0.0.0.0: Listens on all network interfaces (required for device to connect)

---

Step 6: Main Function

Tie everything together:

if __name__ == "__main__":
    args = parse_args()

    # Connect and start streaming
    device = device_streaming()

    # Set up receiver with callbacks
    stream_receiver = setup_streaming_receiver(device, args.record_to_vrs)

    # Keep running until interrupted
    print("Streaming... Press Ctrl+C to stop")
    try:
        signal.pause()  # Wait for interrupt signal
    except KeyboardInterrupt:
        print("\nStopping streaming...")
        device.stop_streaming()
        print("Streaming stopped")

---

Complete Example Code

Here's the full streaming script structure:

import argparse
import time

import aria.sdk_gen2 as sdk_gen2
import aria.stream_receiver as receiver

from projectaria_tools.core.mps import EyeGaze, hand_tracking, OpenLoopTrajectoryPose
from projectaria_tools.core.sensor_data import (
    AudioData,
    AudioDataRecord,
    FrontendOutput,
    ImageData,
    ImageDataRecord,
    MotionData,
)

# Set up the device client to initiate connection to the device
device_client = sdk_gen2.DeviceClient()


def device_streaming():
    # Set up the device client config to specify the device to be connected to e.g. device serial number.
    # If nothing is specified, the first device in the list of connected devices will be connected to
    config = sdk_gen2.DeviceClientConfig()
    device_client.set_client_config(config)
    device = device_client.connect()

    # Set recording config with profile name
    streaming_config = sdk_gen2.HttpStreamingConfig()
    streaming_config.profile_name = "profile9"
    device.set_streaming_config(streaming_config)

    # Start and stop recording
    device.start_streaming()
    return device


def image_callback(image_data: ImageData, image_record: ImageDataRecord):
    print(
        f"Received image data of size {image_data.to_numpy_array().shape} with timestamp {image_record.capture_timestamp_ns} ns"
    )


def audio_callback(
    audio_data: AudioData, audio_record: AudioDataRecord, num_channels: int
):
    print(
        f"Received audio data with {len(audio_data.data)} samples and {len(audio_record.capture_timestamps_ns)} timestamps and num channels {num_channels}"
    )


def imu_callback(imu_data: MotionData, sensor_label: str):
    print(
        f"Received {sensor_label} accel data {imu_data.accel_msec2} and gyro {imu_data.gyro_radsec}"
    )


def eyegaze_callback(eyegaze_data: EyeGaze):
    print(
        f"Received EyeGaze data at timestamp {eyegaze_data.tracking_timestamp.total_seconds()} sec "
        f"with yaw={eyegaze_data.yaw:.3f} rad, pitch={eyegaze_data.pitch:.3f} rad, "
        f"depth={eyegaze_data.depth:.3f} m"
    )


def handtracking_callback(handtracking_data: hand_tracking.HandTrackingResult):
    print(
        f"Received HandTracking data at timestamp {handtracking_data.tracking_timestamp.total_seconds()} sec"
    )

    # Check left hand data
    if handtracking_data.left_hand is not None:
        left_hand = handtracking_data.left_hand
        print(f"  Left hand confidence: {left_hand.confidence:.3f}")
        print(f"  Left wrist position: {left_hand.get_wrist_position_device()}")
        print(f"  Left palm position: {left_hand.get_palm_position_device()}")
        if left_hand.wrist_and_palm_normal_device is not None:
            normals = left_hand.wrist_and_palm_normal_device
            print(f"  Left wrist normal: {normals.wrist_normal_device}")
            print(f"  Left palm normal: {normals.palm_normal_device}")
    else:
        print("  Left hand: No data")

    # Check right hand data
    if handtracking_data.right_hand is not None:
        right_hand = handtracking_data.right_hand
        print(f"  Right hand confidence: {right_hand.confidence:.3f}")
        print(f"  Right wrist position: {right_hand.get_wrist_position_device()}")
        print(f"  Right palm position: {right_hand.get_palm_position_device()}")
        if right_hand.wrist_and_palm_normal_device is not None:
            normals = right_hand.wrist_and_palm_normal_device
            print(f"  Right wrist normal: {normals.wrist_normal_device}")
            print(f"  Right palm normal: {normals.palm_normal_device}")
    else:
        print("  Right hand: No data")


def vio_callback(vio_data: FrontendOutput):
    print(
        f"Received VIO data at timestamp {vio_data.capture_timestamp_ns} with transform_odometry_bodyimu: {vio_data.transform_odometry_bodyimu.rotation().log()} and {vio_data.transform_odometry_bodyimu.translation()} ns"
    )


def calib_callback(calib_json_str: str):
    print(f"Received calibration: {calib_json_str}")


def setup_streaming_receiver(device, record_to_vrs):
    # setup the server to receive streaming data from the device
    # IP address : 0.0.0.0 means that the server is listening on all available interfaces
    # Port : 6768 is the port number that the server is listening on
    config = sdk_gen2.HttpServerConfig()
    config.address = "0.0.0.0"
    config.port = 6768

    # setup the receiver
    stream_receiver = receiver.StreamReceiver()
    stream_receiver.set_server_config(config)
    if record_to_vrs != "":
        stream_receiver.record_to_vrs(record_to_vrs)

    # register callbacks for each type of data
    stream_receiver.register_slam_callback(image_callback)
    stream_receiver.register_rgb_callback(image_callback)
    stream_receiver.register_audio_callback(audio_callback)
    stream_receiver.register_eye_gaze_callback(eyegaze_callback)
    stream_receiver.register_hand_pose_callback(handtracking_callback)
    stream_receiver.register_vio_callback(vio_callback)

    # start the server
    stream_receiver.start_server()

    time.sleep(10)

    # stop streaming and terminate the server
    device.stop_streaming()

    time.sleep(2)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--record-to-vrs",
        dest="record_to_vrs",
        type=str,
        default="",
        required=False,
        help="Output directory to save the received streaming into VRS",
    )

    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    # setup device to start streaming
    device = device_streaming()

    # setup streaming receiver to receive streaming data with callbacks
    setup_streaming_receiver(device, args.record_to_vrs)

---

Usage Examples

Basic Streaming with Console Output

python ~/Downloads/projectaria_client_sdk_samples_gen2/device_streaming.py

What happens:

• Connects to device and starts streaming
• Prints real-time data from all sensors to console
• Data is NOT saved

---

Streaming and Recording to VRS

python ~/Downloads/projectaria_client_sdk_samples_gen2/device_streaming.py \
    --record-to-vrs ~/Downloads/streaming_capture.vrs

What happens:

• Streams data with real-time callbacks
• Simultaneously records all data to VRS file
• VRS file can be played back later with aria_rerun_viewer

---

Troubleshooting

No Data in Callbacks

Problem: Streaming starts but callbacks are never called.

Solutions:

1. Check port 6768 is open:

   # On Linux/macOS
   sudo lsof -i :6768

2. Disable VPN:

   • VPNs block streaming data
   • Disconnect from VPN/Lighthouse

3. Check firewall:

   • Ensure firewall allows port 6768
   • Add exception for Python script

4. Verify server address:

   • Must be 0.0.0.0 to listen on all interfaces

5. Run aria_doctor:

   aria_doctor

---

High Data Drop Rate

Problem: VRS file shows many dropped frames.

Solutions:

• Close other applications consuming bandwidth
• Check system resources (CPU, memory)
• Reduce callback processing time
• Increase message queue size

---

HTTP Streaming to Remote Server

This section demonstrates how to stream data to a remote server running at a specific IP address and port. This is useful when you want to run your streaming receiver on a different machine than the one connected to the Aria device.

Overview

Remote streaming allows you to:

• Run the receiver on a dedicated server: Process streaming data on a more powerful machine
• Stream across the network: Send data to a receiver running on a remote IP address
• Secure communication: Use TLS certificates for encrypted streaming

Workflow Overview

To set up remote streaming, follow these steps in order:

1. Install certificates on the device (outputs a cert name)
2. Copy certificates to the remote server
3. Start the streaming server on the remote machine
4. Start streaming on the device with the cert name

Important

Always start streaming with the correct certificate name or install certificates before starting the streaming server. Starting streaming without specifying the cert name will default to installing new certificates, which would invalidate any old certificates already copied to the server.

---

Step 1: Install Streaming Certificates

On the machine connected to the Aria Gen2 device via USB, install the streaming certificates:

# Install certificates on the device
aria_gen2 streaming install-certs

This command:

• Generates new streaming certificates
• Installs the publisher certificate on the Aria Gen2 device
• Saves the subscriber certificates locally to ~/.aria/streaming-certs/persistent/
• Outputs the cert name that you'll use when starting streaming

Example output:

[AriaGen2SDK:Device][INFO]: Installing local streaming certs under: "/home/xxx/.aria/streaming-certs/persistent"
[AriaGen2SDK:Device][INFO]: Request installing streaming certs on device: cert_1769712774692532015
[AriaGen2Cli:App][INFO]: successfully installed streaming cert.

Note the cert name (e.g., cert_1769712774692532015) - you'll need it in Step 4.

Certificate Files

After installation, the following files are created in ~/.aria/streaming-certs/persistent/:

File	Description
root_ca.pem	Root CA certificate that verifies the device's publisher certificate
subscriber.pem	Subscriber certificate for the streaming server
subscriber-key.pem	Subscriber private key for the streaming server
publisher-cert-name	Contains the certificate name

---

Step 2: Copy Certificates to Remote Server

Copy the certificate directory to your remote server:

# Copy certificates to the remote server
scp -r ~/.aria/streaming-certs/persistent/ user@<remote-server-ip>:~/.aria/streaming-certs/persistent/

---

Step 3: Start the Streaming Server

On the remote server, start the streaming server with the copied certificates.

Python SDK Configuration

import os
import aria.sdk_gen2 as sdk_gen2
import aria.stream_receiver as receiver

def setup_remote_streaming_server(certs_dir):
    """Configure streaming receiver for remote connections with TLS certificates."""

    # Certificate paths from copied directory
    root_ca_path = os.path.join(certs_dir, "root_ca.pem")
    subscriber_cert_path = os.path.join(certs_dir, "subscriber.pem")
    subscriber_key_path = os.path.join(certs_dir, "subscriber-key.pem")

    # Configure server with remote IP and port
    config = sdk_gen2.HttpServerConfig()
    config.address = "0.0.0.0"  # Listen on all interfaces for remote connections
    config.port = 6768           # Default streaming port

    # Configure TLS with subscriber certificates
    config.ca_root = root_ca_path
    config.certificate.cert = subscriber_cert_path
    config.certificate.key = subscriber_key_path

    # Create and configure the receiver
    stream_receiver = receiver.StreamReceiver(
        enable_image_decoding=True, enable_raw_stream=False
    )
    stream_receiver.set_server_config(config)

    # Register callbacks (same as local streaming)
    stream_receiver.register_slam_callback(image_callback)
    stream_receiver.register_rgb_callback(image_callback)
    stream_receiver.register_audio_callback(audio_callback)
    stream_receiver.register_eye_gaze_callback(eyegaze_callback)
    stream_receiver.register_hand_pose_callback(handtracking_callback)
    stream_receiver.register_vio_callback(vio_callback)

    # Start the server
    stream_receiver.start_server()

    return stream_receiver

---

Step 4: Start Streaming on Device

On the machine connected to the Aria device, start streaming with the cert name from Step 1:

# Start streaming to the remote server
aria_gen2 streaming start \
    --url https://<remote-server-ip>:<remote-server-port> \
    --streaming-cert-name <cert-name-from-step-1>

Options:

• --url: The URL of the remote streaming server (e.g., https://192.168.1.100:6768)
• --streaming-cert-name: The cert name output from Step 1 (e.g., cert_1767647798330102000)

---

Network Configuration

Ensure the following for remote streaming:

1. Firewall: Open port 6768 (or your custom port) on the remote server
2. Network connectivity: The Aria Gen2 device must be able to reach the remote server IP
3. Certificate matching: The publisher certificate on the device must match the subscriber certificates on the server (issued from the same Nebula CA)

Troubleshooting Remote Streaming

Connection refused:

• Verify the server is running and listening on the correct port
• Check firewall rules on the server

Certificate errors:

• Ensure all certificate files exist in the specified directory
• Verify you installed certs (aria_gen2 streaming install-certs) before copying them to the server
• If you ran streaming start after copying certs, new certs will been installed - reinstall and recopy
• Check that the cert name used in --streaming-cert-name matches the one from install-certs output

No data received:

• Confirm the device started streaming successfully
• Check network connectivity between device and server
• Verify the streaming URL is correct

---

Next Steps

• Learn about text-to-speech commands
• Review all Python SDK examples
• Explore the Streaming Guide for CLI streaming
• Check Troubleshooting for common issues