Adding Cameras

Adamo encodes camera frames with a hardware H.264 encoder and publishes them as named tracks. You attach as many tracks as you want — one per camera — and they run in parallel.

There are two input methods:

1. V4L2 — direct device capture from /dev/videoN. Use this for USB webcams, RealSense, and any V4L2-compatible camera.
2. Shared memory (iceoryx2) — frames produced by another process on the same host (a custom driver, a perception pipeline, an external bridge), or pushed in from your own SDK code via the caller-fed video() track.

Those sources can carry raw pixels such as BGRA, NV12, or I420. They can also carry compressed Motion JPEG frames by setting the source format to mjpeg.

USB / V4L2 Cameras

For USB webcams, Intel RealSense, and any V4L2-compatible device.

Python

import adamo

robot = adamo.Robot(api_key="ak_...", name="my-arm")
robot.attach_video(
    "main",
    device="/dev/video0",
    width=1280,
    height=720,
    fps=30,
    bitrate_kbps=4000,
)
robot.run()

Rust

use adamo::Robot;

fn main() -> adamo::Result<()> {
    let mut robot = Robot::new_default("ak_...", Some("my-arm"))?;
    robot.attach_v4l2("main", "/dev/video0", 1280, 720, 30, 4000, false)?;
    robot.run()
}

C

adamo_robot_attach_video_v4l2(
    robot, "main", "/dev/video0",
    /* width  */ 1280, /* height */ 720,
    /* fps    */ 30,   /* bitrate_kbps */ 4000,
    /* stereo */ false);

Finding the device

ls /dev/video*
v4l2-ctl --list-devices
v4l2-ctl -d /dev/video0 --list-formats     # what pixel formats it supports

On Jetson the SDK auto-detects the V4L2 pixel format (YUY2, NV12, …) so the encoder uses the optimal GPU colourspace path.

If the camera exposes an MJPEG mode, request it with pixel_format="mjpeg":

robot.attach_video(
    "usb_mjpeg",
    device="/dev/video0",
    width=1280,
    height=720,
    fps=30,
    bitrate_kbps=4000,
    pixel_format="mjpeg",
)

Multiple Cameras

Attach one track per camera. They run independently — different fps and bitrates per track is fine.

Python

robot.attach_video("front", device="/dev/video0", width=1920, height=1080, fps=60, bitrate_kbps=8000)
robot.attach_video("rear",  device="/dev/video2", width=1280, height=720,  fps=15, bitrate_kbps=2000)

Rust

robot.attach_v4l2("front", "/dev/video0", 1920, 1080, 60, 8000, false)?;
robot.attach_v4l2("rear",  "/dev/video2", 1280, 720,  15, 2000, false)?;

C

adamo_robot_attach_video_v4l2(robot, "front", "/dev/video0", 1920, 1080, 60, 8000, false);
adamo_robot_attach_video_v4l2(robot, "rear",  "/dev/video2", 1280, 720,  15, 2000, false);

Shared Memory (iceoryx2)

Use shared memory when frames come from somewhere V4L2 can’t see — a custom driver, a Python perception loop, an external bridge, or your own SDK code rendering frames in process.

Two patterns: consume an existing iceoryx2 service produced by another process, or publish frames yourself from a tight loop.

Consume an existing SHM service

Use this when another process already owns the camera or sensor and can publish raw frames into iceoryx2. The general flow is:

1. Your camera process captures or generates one complete raw frame.
2. It publishes that frame as a single iox2.Slice[ctypes.c_uint8] sample.
3. Adamo attaches to the same iceoryx2 service with robot.attach_video(..., shm=...).
4. The native Adamo pipeline reads the SHM frames, hardware-encodes them, and streams the video track.

The producer and Adamo bridge must agree on the iceoryx2 service name, frame width, frame height, pixel format, and frame rate. The SHM payload is raw frame bytes only. Do not prepend timestamps, headers, or metadata. For 1280x720 BGRA, each sample must contain exactly 1280 * 720 * 4 = 3,686,400 bytes.

Python

Run the publisher and Adamo bridge as separate processes on the same host:

export ADAMO_API_KEY="ak_..."
python zed_left_to_shm.py
python shm_to_adamo.py

The publisher below is a ZED-specific example: it captures the left image from the ZED SDK and publishes it as raw BGRA frames. For other camera types, keep the iceoryx2 publishing shape the same and replace only the frame acquisition code.

zed_left_to_shm.py

import ctypes

import iceoryx2 as iox2
import numpy as np
import pyzed.sl as sl

SERVICE = "camera/zed/left"
FPS = 60

params = sl.InitParameters()
params.camera_resolution = sl.RESOLUTION.HD720
params.camera_fps = FPS
params.depth_mode = sl.DEPTH_MODE.NONE

camera = sl.Camera()
status = camera.open(params)
if status != sl.ERROR_CODE.SUCCESS:
    raise RuntimeError(f"ZED open failed: {status}")

info = camera.get_camera_information()
resolution = info.camera_configuration.resolution
WIDTH = int(resolution.width)
HEIGHT = int(resolution.height)
PIXEL_FORMAT = "BGRA"
FRAME_SIZE = WIDTH * HEIGHT * 4

node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
service = (
    node.service_builder(iox2.ServiceName.new(SERVICE))
    .publish_subscribe(iox2.Slice[ctypes.c_uint8])
    .enable_safe_overflow(True)
    .subscriber_max_buffer_size(2)
    .open_or_create()
)
publisher = service.publisher_builder().initial_max_slice_len(FRAME_SIZE).create()

runtime = sl.RuntimeParameters()
image = sl.Mat()

while True:
    if camera.grab(runtime) != sl.ERROR_CODE.SUCCESS:
        continue

    camera.retrieve_image(image, sl.VIEW.LEFT)
    frame = np.ascontiguousarray(image.get_data())
    payload = frame.tobytes()
    if len(payload) != FRAME_SIZE:
        continue

    sample = publisher.loan_slice_uninit(FRAME_SIZE)
    ctypes.memmove(sample.payload_ptr, payload, FRAME_SIZE)
    sample.assume_init().send()

shm_to_adamo.py

import os

import adamo

robot = adamo.Robot(
    api_key=os.environ["ADAMO_API_KEY"],
    name="my-robot",
)

robot.attach_video(
    "main",
    shm="camera/zed/left",
    width=1280,
    height=720,
    pixel_format="BGRA",
    fps=60,
    bitrate_kbps=4000,
)

robot.run()

The Adamo side is camera-agnostic. It only needs the service name and frame layout that your publisher uses.

If the shared-memory service publishes complete JPEG frames instead of raw pixels, set pixel_format="mjpeg":

robot.attach_video(
    "head",
    shm="head_cam_mjpeg",
    width=1280,
    height=720,
    fps=30,
    bitrate_kbps=4000,
    pixel_format="mjpeg",
)

Rust

The Rust SDK does not yet expose a consume-side attach_shm helper. Use the publish-frames-yourself pattern below, or wrap the producer in Python.

C

The C SDK does not yet expose a consume-side helper. Use the publish-frames-yourself pattern below, or wrap the producer in Python.

Republish SHM frames to ROS 2

You can also bridge the same iceoryx2 service into ROS 2. This is useful when Adamo is the low-latency video transport, but another local ROS node still needs the raw camera stream.

The flow is:

1. A camera-specific producer publishes raw frames to iceoryx2.
2. A ROS republisher subscribes to that SHM service.
3. The republisher copies each raw frame into a sensor_msgs/Image message.

This single example assumes 1280x720 BGRA frames from camera/zed/left. For other camera types, change the constants at the top to match your producer.

source /opt/ros/humble/setup.bash
python shm_to_ros.py

shm_to_ros.py

import ctypes
import time

import iceoryx2 as iox2
import rclpy
from sensor_msgs.msg import Image

SERVICE = "camera/zed/left"
TOPIC = "/zed/left/image_raw"
WIDTH = 1280
HEIGHT = 720
ENCODING = "bgra8"
BYTES_PER_PIXEL = 4
FRAME_ID = "zed_left_camera"

EXPECTED_SIZE = WIDTH * HEIGHT * BYTES_PER_PIXEL
STEP = WIDTH * BYTES_PER_PIXEL

iox_node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
service = (
    iox_node.service_builder(iox2.ServiceName.new(SERVICE))
    .publish_subscribe(iox2.Slice[ctypes.c_uint8])
    .enable_safe_overflow(True)
    .subscriber_max_buffer_size(2)
    .open_or_create()
)
subscriber = service.subscriber_builder().create()

rclpy.init()
node = rclpy.create_node("shm_image_republisher")
publisher = node.create_publisher(Image, TOPIC, 10)
node.get_logger().info(
    f"Republishing SHM '{SERVICE}' to ROS topic '{TOPIC}' "
    f"as {WIDTH}x{HEIGHT} {ENCODING}"
)

try:
    while rclpy.ok():
        sample = subscriber.receive()
        if sample is None:
            rclpy.spin_once(node, timeout_sec=0.0)
            time.sleep(0.0005)
            continue

        try:
            payload = bytes(sample.payload())
        finally:
            delete = getattr(sample, "delete", None)
            if delete is not None:
                delete()

        if len(payload) != EXPECTED_SIZE:
            node.get_logger().warn(
                f"dropping SHM sample: expected {EXPECTED_SIZE} bytes, "
                f"got {len(payload)}"
            )
            continue

        msg = Image()
        msg.header.stamp = node.get_clock().now().to_msg()
        msg.header.frame_id = FRAME_ID
        msg.height = HEIGHT
        msg.width = WIDTH
        msg.encoding = ENCODING
        msg.is_bigendian = 0
        msg.step = STEP
        msg.data = payload
        publisher.publish(msg)
        rclpy.spin_once(node, timeout_sec=0.0)
except KeyboardInterrupt:
    pass
finally:
    node.destroy_node()
    rclpy.shutdown()

Publish frames yourself

The SDK gives you a track handle; you push raw pixel buffers into it. Adamo allocates a private iceoryx2 service for the track and routes the frames into the encoder.

Python

import cv2
import adamo

robot = adamo.Robot(api_key="ak_...", name="my-arm")
track = robot.video("overlay", width=1280, height=720, pixel_format="BGRA",
                    fps=30, bitrate_kbps=4000)

cap = cv2.VideoCapture(0)
while True:
    ok, frame = cap.read()
    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2BGRA)
    track.send(frame)

The encoder pipeline auto-starts on the first track.send().

track.send() is for fixed-size raw pixel buffers. For variable-size MJPEG frames, publish complete JPEG frames to an iceoryx2 service and consume that service with attach_video(..., shm=..., pixel_format="mjpeg").

Rust

use adamo::Robot;

fn main() -> adamo::Result<()> {
    let mut robot = Robot::new_default("ak_...", Some("my-arm"))?;
    let mut track = robot.video("overlay", 1280, 720, "BGRA", 30, 4000, false)?;
    std::thread::spawn(move || {
        loop {
            let frame: Vec<u8> = render_next_frame();  // your code
            track.send(&frame).unwrap();
        }
    });
    robot.run()
}

C

adamo_video_track_t *t = adamo_robot_video(
    robot, "overlay", 1280, 720, "BGRA", 30, 4000, false);
// push frames in a loop:
adamo_video_track_send(t, frame_bytes, frame_len);

Supported data types

Format	Payload size	Notes
BGRA, RGBA, BGRX, RGBX	4 bytes per pixel	Most convenient for OpenCV / numpy.
RGB, BGR	3 bytes per pixel
YUY2, UYVY	2 bytes per pixel
I420, NV12	1.5 bytes per pixel	Most efficient raw formats for the encoder.
MJPEG, MJPG, JPEG, image/jpeg	Variable	Compressed Motion JPEG. Each source sample must contain one complete JPEG frame. Supported for V4L2, GStreamer, and consume-side SHM sources.

MJPEG sources are decoded and re-encoded into the configured output codec. On Jetson, Adamo avoids the Jetson nvjpegdec path for non-16-aligned MJPEG dimensions because that decoder can corrupt the bottom of frames; the stream still uses the Jetson hardware H.264 encoder after decode.

Tuning Quality

Two knobs cover almost everything:

• bitrate_kbps — higher means better picture and more bandwidth. Start at 2000–4000 kbps for 720p, 6000–8000 for 1080p.
• fps — 60 if your camera supports it, 30 otherwise. On constrained links drop to 15.

If video looks blocky, raise the bitrate before anything else. If bandwidth is the bottleneck, lower fps first.

Encoder selection

Adamo picks the best available hardware H.264 encoder for the host at startup:

Host	Encoder used
Jetson (Orin, Thor, …)	On-chip NVENC
x86 with NVIDIA discrete GPU	NVENC
Intel / AMD integrated graphics	VA-API
macOS	VideoToolbox
Anything else	CPU (x264)

To see what the SDK picked, check the log line printed at connect.