Local archive¶

Appliance-local archive control lives on client.archive and client.robot(...).archive. Cloud catalog reads use client.platform.archive.listBags(), listVideos(), and listImages() (see Cloud archive catalog).

core.getImage returns the next frame in memory for processing. archive.captureImage saves a JPEG to appliance disk. archive.captureVideo records an MP4 session with stop(). archive.recordBag / playBag control local rosbag recording and playback (see Local rosbag recording).

Image and video capture example:

Python

import time

from olo import Client
# @olo-playground: autofill-sim-topics

TOPIC = "/image"
MESSAGE_TYPE = "sensor_msgs/msg/Image"

with Client() as client:
    robot = client.robot()

    # Capture one JPEG from /image and save it on the appliance.
    snapshot = robot.archive.capture_image(topic=TOPIC, message_type=MESSAGE_TYPE)
    print(f"Saved image: {snapshot.file_path} ({snapshot.file_size} bytes)")

    # Record a 10s MP4 from the same topic.
    with robot.archive.capture_video(topic=TOPIC, message_type=MESSAGE_TYPE) as session:
        print(f"Recording video on {TOPIC} for 10s...")
        time.sleep(10)
        result = session.stop()
        print(f"Saved video: {result.file_path} ({result.frame_count} frames, {result.duration_seconds:.1f}s)")

    # List saved images and videos on the appliance.
    print("Images:", [item.filename for item in robot.archive.list_images()])
    print("Videos:", [item.filename for item in robot.archive.list_videos()])

TypeScript

import { connect } from "olo/node";
// @olo-playground: autofill-sim-topics

const TOPIC = "/image";
const MESSAGE_TYPE = "sensor_msgs/msg/Image";

const client = connect();
const robot = await client.robot();

// Capture one JPEG from /image and save it on the appliance.
const snapshot = await robot.archive.captureImage({ topic: TOPIC, messageType: MESSAGE_TYPE });
console.log(`Saved image: ${snapshot.filePath} (${snapshot.fileSize} bytes)`);

// Record a 10s MP4 from the same topic.
await using session = await robot.archive.captureVideo({
  topic: TOPIC,
  messageType: MESSAGE_TYPE,
});
console.log(`Recording video on ${TOPIC} for 10s...`);
await new Promise((r) => setTimeout(r, 10_000));
const result = await session.stop();
console.log(
  `Saved video: ${result.filePath} (${result.frameCount} frames, ${result.durationSeconds.toFixed(1)}s)`,
);

// List saved images and videos on the appliance.
console.log(
  "Images:",
  (await robot.archive.listImages()).map((item) => item.filename),
);
console.log(
  "Videos:",
  (await robot.archive.listVideos()).map((item) => item.filename),
);

Python API¶

Appliance-local archive (saved images, videos, and rosbags).

class olo.archive.Archive[source]¶

Bases: object

Sync wrapper for appliance-local archive control.

__init__(session)[source]¶

Parameters:: session (ChannelSession)
Return type:: None

class olo.archive.ArchiveHandle[source]¶

Bases: object

Namespace-scoped archive handle (symmetry with other robot modules).

__init__(client, namespace=None)[source]¶

Parameters:

client (Archive)
namespace (str | None)

Return type:

None

class olo.archive.BagInfo[source]¶

Bases: object

BagInfo(summary: olo.archive.types.BagSummary, topics: tuple[olo.archive.types.BagTopicInfo, …], compression: str)

__init__(summary, topics, compression)¶

Parameters:

summary (BagSummary)
topics (tuple[BagTopicInfo, ...])
compression (str)

Return type:

None

class olo.archive.BagSession[source]¶

Bases: object

Handle for an active bag recording session.

__init__(client, recording_id, filename, output_path)[source]¶

Parameters:

client (Archive)
recording_id (str)
filename (str)
output_path (str)

Return type:

None

class olo.archive.BagSessionTerminalEvent[source]¶

Bases: object

BagSessionTerminalEvent(type: Literal[‘completed’, ‘failed’], exit_code: int, total_bytes: int | None = None, error_message: str | None = None)

__init__(type, exit_code, total_bytes=None, error_message=None)¶

Parameters:

type (Literal['completed', 'failed'])
exit_code (int)
total_bytes (int | None)
error_message (str | None)

Return type:

None

class olo.archive.BagSummary[source]¶

Bases: object

BagSummary(name: str, total_bytes: int, file_count: int, created_at: datetime.datetime | None, updated_at: datetime.datetime | None, duration_seconds: float, message_count: int, storage_format: str)

__init__(name, total_bytes, file_count, created_at, updated_at, duration_seconds, message_count, storage_format)¶

Parameters:

name (str)
total_bytes (int)
file_count (int)
created_at (datetime | None)
updated_at (datetime | None)
duration_seconds (float)
message_count (int)
storage_format (str)

Return type:

None

class olo.archive.BagTopicInfo[source]¶

Bases: object

BagTopicInfo(name: str, type: str, message_count: int)

__init__(name, type, message_count)¶

Parameters:

name (str)
type (str)
message_count (int)

Return type:

None

class olo.archive.CaptureImageResult[source]¶

Bases: object

CaptureImageResult(filename: str, file_path: str, file_size: int, mime_type: str)

__init__(filename, file_path, file_size, mime_type)¶

Parameters:

filename (str)
file_path (str)
file_size (int)
mime_type (str)

Return type:

None

class olo.archive.ImageSummary[source]¶

Bases: object

ImageSummary(filename: str, file_path: str, file_size: int, mime_type: str, created_at: datetime.datetime | None, modified_at: datetime.datetime | None)

__init__(filename, file_path, file_size, mime_type, created_at, modified_at)¶

Parameters:

filename (str)
file_path (str)
file_size (int)
mime_type (str)
created_at (datetime | None)
modified_at (datetime | None)

Return type:

None

class olo.archive.PlaybackSession[source]¶

Bases: object

Handle for an active bag playback session.

__init__(client, playback_id, name)[source]¶

Parameters:

client (Archive)
playback_id (str)
name (str)

Return type:

None

class olo.archive.StopBagPlaybackResult[source]¶

Bases: object

StopBagPlaybackResult(playback_id: str)

__init__(playback_id)¶

Parameters:: playback_id (str)
Return type:: None

class olo.archive.StopBagRecordingResult[source]¶

Bases: object

StopBagRecordingResult(recording_id: str, filename: str, total_bytes: int)

__init__(recording_id, filename, total_bytes)¶

Parameters:

recording_id (str)
filename (str)
total_bytes (int)

Return type:

None

class olo.archive.StopVideoCaptureResult[source]¶

Bases: object

StopVideoCaptureResult(recording_id: str, filename: str, file_path: str, file_size: int, frame_count: int, duration_seconds: float, reason: str)

__init__(recording_id, filename, file_path, file_size, frame_count, duration_seconds, reason)¶

Parameters:

recording_id (str)
filename (str)
file_path (str)
file_size (int)
frame_count (int)
duration_seconds (float)
reason (str)

Return type:

None

class olo.archive.VideoSession[source]¶

Bases: object

Handle for an active video capture session.

__init__(client, recording_id, filename, file_path)[source]¶

Parameters:

client (Archive)
recording_id (str)
filename (str)
file_path (str)

Return type:

None

class olo.archive.VideoSummary[source]¶

Bases: object

VideoSummary(filename: str, file_path: str, file_size: int, created_at: datetime.datetime | None, modified_at: datetime.datetime | None, recording_id: str, topic: str, status: str)

__init__(filename, file_path, file_size, created_at, modified_at, recording_id, topic, status)¶

Parameters:

filename (str)
file_path (str)
file_size (int)
created_at (datetime | None)
modified_at (datetime | None)
recording_id (str)
topic (str)
status (str)

Return type:

None

TypeScript API¶

Migration from SDK v1¶

oloClient.videoRecording.startRecording(topic, opts) → robot.archive.captureVideo({ topic, ... })
oloClient.videoRecording.stopRecording(id) → session.stop()
Local rosbag recording → robot.archive.recordBag / playBag