Local rosbag recording

Record a rosbag on the appliance, inspect its topics, then play it back with pause and resume. Use robot.archive.recordBag and robot.archive.playBag.

Python

import time
from concurrent.futures import ThreadPoolExecutor

from olo import Client

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

    bag_session = None
    playback = None
    try:
        # Record selected ROS topics for 10 seconds.
        bag_session = robot.archive.record_bag(topics=["cmd_vel"])
        print(f"Recording bag {bag_session.filename} for 10s")
        time.sleep(10)
        bag_result = bag_session.stop()
        bag_session = None
        print(f"Saved {bag_result.filename} ({bag_result.total_bytes} bytes)")

        # Inspect the saved bag and list its topics.
        info = robot.archive.inspect_bag(bag_result.filename)
        summary = info.summary
        print(
            f"Bag: {summary.message_count} messages, "
            f"{summary.duration_seconds:.1f}s, {summary.storage_format or 'unknown'} format"
        )
        for topic in info.topics:
            print(f"  {topic.name}: {topic.message_count} messages")

        # Play the bag, exercise playback controls, and wait for completion.
        # system_time avoids a second /clock publisher when the sim already publishes one.
        with ThreadPoolExecutor(max_workers=1) as pool:
            playback = robot.archive.play_bag(bag_result.filename, clock_type="system_time")
            # Subscribe before control/demo sleeps so fast bags cannot finish unseen.
            playback_done = pool.submit(playback.watch)
            print(f"Playing back {bag_result.filename}")
            time.sleep(5)
            playback.pause()
            print("Paused")
            time.sleep(2)
            playback.resume()
            print("Resumed — waiting for bag to finish")
            finished = playback_done.result()
        print(f"Playback {finished.type} (exit {finished.exit_code})")
        playback = None
    finally:
        if bag_session is not None:
            bag_session.stop()
        if playback is not None:
            playback.stop()

TypeScript

import { connect } from "olo/node";

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

let bagSession: Awaited<ReturnType<typeof robot.archive.recordBag>> | null = null;
let playback: Awaited<ReturnType<typeof robot.archive.playBag>> | null = null;
try {
  // Record selected ROS topics for 10 seconds.
  bagSession = await robot.archive.recordBag({ topics: ["cmd_vel"] });
  console.log(`Recording bag ${bagSession.filename} for 10s`);
  await new Promise((r) => setTimeout(r, 10_000));
  const bagResult = await bagSession.stop();
  bagSession = null;
  console.log(`Saved ${bagResult.filename} (${bagResult.totalBytes} bytes)`);

  // Inspect the saved bag and list its topics.
  const info = await robot.archive.inspectBag(bagResult.filename);
  const { summary } = info;
  console.log(
    `Bag: ${summary.messageCount} messages, ${summary.durationSeconds.toFixed(1)}s, ${summary.storageFormat || "unknown"} format`,
  );
  for (const topic of info.topics) {
    console.log(`  ${topic.name}: ${topic.messageCount} messages`);
  }

  // Play the bag, exercise playback controls, and wait for completion.
  // system_time avoids a second /clock publisher when the sim already publishes one.
  playback = await robot.archive.playBag(bagResult.filename, { clockType: "system_time" });
  // Subscribe before control/demo sleeps so fast bags cannot finish unseen.
  const playbackDone = playback.watch();
  console.log(`Playing back ${bagResult.filename}`);
  await new Promise((r) => setTimeout(r, 5_000));
  await playback.pause();
  console.log("Paused");
  await new Promise((r) => setTimeout(r, 2_000));
  await playback.resume();
  console.log("Resumed — waiting for bag to finish");
  const finished = await playbackDone;
  console.log(`Playback ${finished.type} (exit ${finished.exitCode})`);
  playback = null;
} finally {
  if (bagSession) await bagSession.stop();
  if (playback) await playback.stop();
}