Dataset loading for ML¶

Load a Foxglove dataset version into PyTorch or Ray Data. Choose the topics to read and define how to convert each episode into training samples.

Warning

The foxglove.datasets APIs, including the PyTorch and Ray Data adapters and EpisodeReader, are experimental and unstable. They may change in backward-incompatible ways as we continue development and incorporate user feedback.

Installation¶

Install the extra for your framework, or both to use Ray with PyTorch:

pip install 'foxglove-sdk[torch]'      # PyTorch
pip install 'foxglove-sdk[ray]'        # Ray Data
pip install 'foxglove-sdk[torch,ray]'  # Both

PyTorch¶

Set FOXGLOVE_API_TOKEN before running this example. It reads JSON messages with speed and steering fields; adapt read_episode to your data.

from foxglove.datasets.torch import read_dataset
from torch.utils.data import DataLoader

def read_episode(episode):
    for schema, channel, message, decoded in episode.iter_messages():
        yield {"speed": decoded["speed"], "steering": decoded["steering"]}

if __name__ == "__main__":
    dataset = read_dataset(
        "ds_123", version=7, topics=["/training/measurements"],
        read_episode=read_episode,
    )
    loader = DataLoader(
        dataset, batch_size=32, num_workers=2,
        multiprocessing_context="spawn", prefetch_factor=1,
    )
    for batch in loader:
        print(batch)

The result is an IterableDataset. Your callback can yield tensors, tuples, dictionaries, or objects handled by a custom collate_fn.

Ray Data¶

Use the same callback with the Ray adapter:

from foxglove.datasets.ray import read_dataset

dataset = read_dataset(
    "ds_123", version=7, topics=["/training/measurements"],
    read_episode=read_episode, concurrency=2,
)
for batch in dataset.iter_torch_batches(batch_size=32):
    print(batch)

The result is a ray.data.Dataset. Callbacks must yield mappings with consistent column types, such as scalars or NumPy arrays. Use map_batches for further processing. concurrency limits the number of simultaneous read tasks.

Object storage¶

The default source="foxglove" downloads messages through Foxglove. For customer-managed indexed storage, pass source="object_storage" to either adapter. Foxglove supplies episode metadata and recording locations; workers read MCAP files directly with their cloud credentials. The same callback works for both sources:

dataset = read_dataset(
    "ds_123", version=7, topics=["/training/measurements"],
    read_episode=read_episode, source="object_storage",
)

Important

Object storage reads require the API and client to provide recording locations with location.scheme (s3, gs, or az).

The SDK selects S3, GCS, or Azure storage from the location’s scheme. S3 bucket regions are discovered automatically and cached; Azure uses the location’s azure_storage_account_name. The framework extras include pyarrow. Configure native cloud credentials on each worker, such as environment variables or the machine’s IAM role, and grant read access to the original objects. The Foxglove API token does not grant bucket access. Run compute near your storage to avoid cross-region transfer.

Object storage reads require immutable MCAP recordings with summaries, chunk indexes, and message indexes. Missing locations, permissions, unsupported formats, and missing indexes raise errors; there is no fallback to Foxglove downloads. Query-optimized sites that do not retain original objects should use the default download path.

Custom factories¶

Use client_factory for custom Foxglove authentication or configuration. Without a factory, the SDK creates a client using FOXGLOVE_API_TOKEN. Define factories at module scope so workers can serialize them:

import os
from foxglove.client import Client

def make_client():
    return Client(token=os.environ["MY_FOXGLOVE_TOKEN"])

dataset = read_dataset(
    "ds_123", version=7, topics=["/training/measurements"],
    read_episode=read_episode, client_factory=make_client,
)

For custom storage or caching, pass object_store_factory. A factory creates one store in each consuming process; its open method receives an ObjectLocation and returns a seekable binary handle. For example:

import pyarrow.fs as fs
from foxglove.datasets import ObjectLocation

class S3Store:
    def __init__(self):
        self.filesystem = fs.S3FileSystem(region="us-east-1")

    def open(self, location: ObjectLocation):
        return self.filesystem.open_input_file(
            f"{location.bucket}/{location.path}"
        )

dataset = read_dataset(
    "ds_123", version=7, topics=["/training/measurements"],
    read_episode=read_episode, source="object_storage", object_store_factory=S3Store,
)

A custom object_store_factory requires source="object_storage". Use the exact bucket and object key from the location, which may differ from the display filename. Create live filesystem clients inside the factory; do not capture open files or short-lived credentials. A custom store can return a seekable fsspec binary handle with bounded caching. Use multiprocessing_context="spawn" for PyTorch workers.

The built-in S3 store discovers bucket regions through AWS. For S3-compatible services such as MinIO, custom endpoints, or networks that cannot reach the region-discovery endpoint, use a custom factory. Configure its fs.S3FileSystem with an explicit region and, when needed, endpoint_override="minio.example.com:9000". This bypasses built-in region discovery. Configure TLS and credentials for that endpoint in the factory.

Performance and iteration behavior:

Planning requests recording locations with the paginated episode metadata; it does not open objects. Storage clients are created once per worker iteration or Ray read task, after episode sharding. Object storage workers do not need to call the Foxglove API, and the planning client factory is not serialized.
MCAP indexes select chunks by topic and the episode’s inclusive time window, extended by lookback when requested. A selected chunk may contain other topics, so the reader still fetches and decompresses that whole chunk. The SDK uses a 64 KiB buffer to coalesce small header and summary reads.
Recordings are merged in log-time order before decoding. Equivalent schemas and channels share episode-local IDs; conflicting file-local IDs are remapped. Original timestamps and payload bytes are preserved. Files close as each recording is exhausted, including errors and early termination. Recordings outside the requested time range close after their summaries are read.
Before the first sample, each recording’s summary is read and the first matching chunk from each overlapping recording is loaded to establish global timestamp order. Startup latency therefore grows with the number of recordings.
Memory includes the indexes and active decompressed chunks of the episode’s recordings, decoded samples, and framework prefetch buffers. It is not bounded by Ray’s output block target alone. Tune Ray concurrency or PyTorch num_workers and prefetch_factor against your workload and storage limits.
Every new iteration reads again. Sparse windows benefit from range reads; repeated dense epochs may benefit from a bounded worker-local disk cache in your store. Whole-file caches fetch the entire recording and can increase first-sample latency and transfer. Keep recordings immutable and manage cache size explicitly.

Message formats¶

JSON decoding is included by default. For Protobuf or ROS, install mcap-protobuf-support, mcap-ros1-support, or mcap-ros2-support on every worker.

For custom decoding, create MCAP decoder factories inside your callback and pass them to episode.iter_messages(decoder_factories=[...]). Messages arrive in log-time order as (schema, channel, message, decoded_message) tuples.

See python/foxglove-sdk-examples/dataset-training for a complete example.

H.264 video¶

Install foxglove-sdk[video,torch] or foxglove-sdk[video,ray], plus the MCAP decoder for your recording (see Message formats).

Pass decode_h264 as the episode callback to decode every selected camera topic. Using equally sized camera images:

from foxglove.datasets.torch import read_dataset, to_image_tensor
from foxglove.datasets.video import decode_h264
from torch.utils.data import DataLoader

if __name__ == "__main__":
    dataset = read_dataset(
        "ds_123", version=7,
        topics=["/camera/front", "/camera/wrist"],
        read_episode=decode_h264,
    )
    loader = DataLoader(
        dataset, batch_size=16, num_workers=2,
        multiprocessing_context="spawn", prefetch_factor=1,
    )
    for batch in loader:
        images = to_image_tensor(batch)  # [B, 3, H, W]
        print(batch["topic"], images.shape)

Each sample is one camera frame: an RGB uint8 NumPy array in [H, W, 3] order, its topic, and original timestamps in nanoseconds. Frames from different cameras are not synchronized or paired. Match timestamps in your own callback to assemble multi-camera samples. For Ray, change the adapter import and use dataset.iter_rows().

Within a callback, decode_h264(episode, topic="/camera/front") reads just that camera in a separate download. The topic must be among those selected by read_dataset. Without topic, all selected cameras share one download.

Important limits:

CPU decoding of CompressedVideo H.264 Annex B only: one image per message, no B-frames, and an initial IDR keyframe containing SPS/PPS.
Missing video messages can cause undetected image corruption until the next keyframe, even when decoding raises no error.
Decoding needs earlier frames. The default five-second lookback searches only recordings attached to the episode. Missing history raises VideoDecodeError; increase the lookback or attach the missing recording. Compressed history is buffered and decoded only from the latest usable keyframe. Only frames within the episode’s inclusive MCAP log-time window are emitted.
Without topic, non-video messages are discarded. An explicit non-video topic raises VideoDecodeError. Read other sensors separately with episode.iter_messages(topics=["/joint_states"]).
Resize images as needed before batching. to_image_tensor(sample) from foxglove.datasets.torch accepts channels-last RGB NumPy images or tensors, including batches, and moves the color channels before height and width. Call it once, before or after DataLoader; it preserves dtype and device. Normalize and move to your training device as needed.

To request ten seconds of history, pass read_episode=read_video to read_dataset using this callback, which also works with spawned workers:

from datetime import timedelta
from functools import partial

read_video = partial(decode_h264, lookback=timedelta(seconds=10))

Consume the iterator fully to receive buffered frames. The episode closes streams automatically, including on errors or early termination.

API reference¶

PyTorch¶

foxglove.datasets.torch.read_dataset(dataset_id, *, version, topics, read_episode, source='foxglove', client_factory=None, object_store_factory=None, rank=None, world_size=None)¶

Plan a topic-filtered dataset; download messages only during iteration. :rtype: IterableDataset[TypeVar(_T)]

Warning

This API is experimental and unstable. It may change in backward-incompatible ways as we continue development and incorporate user feedback.

read_episode yields samples, such as tensors, tuples, or dictionaries. Use DataLoader’s collate_fn for custom sample types. Only metadata is fetched here. Each new iteration rereads data. The default client reads FOXGLOVE_API_TOKEN from the environment; override client_factory for custom authentication or endpoints.

source="object_storage" reads indexed MCAPs from S3, GCS, or Azure using recording locations and the worker’s cloud credentials. No filesystem configuration is needed. Missing locations, unsupported schemes, and access failures raise errors without falling back to Foxglove downloads. Use DataLoader’s spawn multiprocessing context with cloud storage.

object_store_factory overrides built-in storage and requires source="object_storage". Factories and callbacks used by workers must be pickleable. The API client factory is used only during object storage planning and is not serialized; in default mode it also runs in each consuming process.

Distributed rank and world size are captured here, before DataLoader workers start, or can be supplied together explicitly. Episodes can yield unequal sample counts: use an uneven-input training policy such as DDP’s join context. DataLoader shuffling and DistributedSampler do not apply to this iterable.

Ray Data¶

foxglove.datasets.ray.read_dataset(dataset_id, *, version, topics, read_episode, source='foxglove', client_factory=None, object_store_factory=None, concurrency=None)¶

Return a native Ray dataset of callback-produced sample mappings.

Mappings (including TypedDict samples) are materialized as dictionaries for Ray. :rtype: Dataset

Warning

This API is experimental and unstable. It may change in backward-incompatible ways as we continue development and incorporate user feedback.

Only episode metadata is fetched during planning. Workers stream selected topics and execute read_episode. Factories and callbacks must be serializable and their dependencies available on every worker. Use consistent column types containing scalars, NumPy arrays, or other Arrow-compatible values.

The default client uses FOXGLOVE_API_TOKEN. Set source="object_storage" to read indexed MCAPs directly from S3, GCS, or Azure. The SDK selects the filesystem from recording locations and uses each worker’s cloud credentials. Missing locations, unsupported schemes, and access failures raise errors without falling back to Foxglove downloads.

client_factory overrides API authentication or endpoints. object_store_factory overrides built-in storage and requires source="object_storage". Factories used by workers must be serializable. For object storage the API client factory is only used during planning and is not serialized. No storage clients or open files are serialized in the read tasks.

concurrency caps concurrent read tasks. Block construction follows Ray’s DataContext.target_max_block_size. Ray can combine blocks and buffer multiple episodes before delivering output. This target is not a memory ceiling: large samples, prefetching, and concurrent tasks can exceed it. Setting it to None allows an entire task to be buffered. Read task retries rerun callbacks, so callbacks should not perform external side effects.

Episode reader¶

class foxglove.datasets.EpisodeReader¶

An episode scoped to the topics selected by read_dataset.

Warning

This API is experimental and unstable. It may change in backward-incompatible ways as we continue development and incorporate user feedback.

Instances are supplied to the user’s read_episode callback and remain usable only until that callback’s iterator finishes or is closed. Message tuples are the client’s (schema, channel, message, decoded_message) values. Create custom message decoders inside the callback and pass them to iter_messages(). Media decoding (such as H.264 to images) is a separate processing step.

__init__(episode, topics, client, object_store=None)¶

property end_time: datetime¶: End of the episode’s time window.

property id: str¶: Episode ID.

iter_messages(*, decoder_factories=None, topics=None, lookback=datetime.timedelta(0))¶

Stream the selected topics in log-time order without buffering the episode.

Each invocation opens a new stream on first iteration. Repeated invocations redownload data. Active streams are closed when the callback ends. Message context and stream order are preserved for stateful processing.

Parameters:

decoder_factories (Sequence[DecoderFactory] | None (default: None)) – MCAP decoder factories used for message deserialization. None uses the client’s default decoders; an explicit list replaces them. Construct factories inside the callback so their state stays local to the worker and episode. This does not decode media payloads into images.
topics (Sequence[str] | None (default: None)) – Optional nonempty subset of the dataset topics. Filtering happens on the server or through MCAP indexes for object storage; separate calls open separate downloads.
lookback (timedelta (default: datetime.timedelta(0))) – Additional history to request before the episode start, for stateful media decoding. Must be nonnegative. Only recordings attached to the episode are searched; history may be unavailable. Consumers must exclude lookback messages from their training samples.

Return type:

Generator[tuple[Schema | None, Channel, Message, Any], None, None]

Returns:

Tuples of schema (possibly None), channel, raw MCAP message, and decoded message payload.

property metadata: Mapping[str, Any]¶: User-defined episode metadata.

property start_time: datetime¶: Start of the episode’s time window.

Object storage¶

class foxglove.datasets.ObjectLocation¶

Location of an original recording in customer-managed storage.

path is the full object key, not the recording’s display filename. scheme identifies the cloud provider: s3, gs, or az for built-in storage access. Custom stores may also accept other or absent schemes. For Azure, bucket is the container and azure_storage_account_name identifies its account. These APIs are experimental and unstable.

__init__(bucket, path, azure_storage_account_name=None, scheme=None)¶

class foxglove.datasets.ObjectStore¶

Worker-local storage access using the customer’s credentials.

Implementations may use PyArrow, fsspec, or another random-access filesystem. These APIs are experimental and unstable.

__init__(*args, **kwargs)¶

open(location)¶

Open a fresh seekable binary file; the SDK owns and closes this handle.

The handle must support readinto as well as read and seek. The SDK buffers small reads in 64 KiB blocks. Use range reads and bounded caching, not a whole-object download. Files must remain immutable for the duration of a dataset read.

Return type:: IO[bytes]

Video decoding¶

foxglove.datasets.video.decode_h264(episode, *, topic=None, lookback=datetime.timedelta(seconds=5), decoder_factories=None)¶

Yield RGB frames from an episode. Requires foxglove-sdk[video].

Warning

This API is experimental and unstable. It may change in backward-incompatible ways as we continue development and incorporate user feedback.

Use directly as read_episode=decode_h264 with PyTorch or Ray, or call inside an episode callback. Each invocation opens one download and maintains independent decoder state per MCAP channel. The episode closes its iterators on exit; call close() to release one earlier.

Parameters:

episode (EpisodeReader) – Supplies the inclusive MCAP log-time window for output frames.
topic (str | None (default: None)) – One selected dataset topic, or None for all selected topics. Non-video schemas are ignored only when topic is None.
lookback (timedelta (default: datetime.timedelta(seconds=5))) – Nonnegative history budget, default five seconds. Searches only attached recordings. Missing initialization history raises VideoDecodeError. Only history since the latest self-contained keyframe is decoded; older slice data is not checked for corruption.
decoder_factories (Sequence[DecoderFactory] | None (default: None)) – Optional MCAP deserializers created in the callback. None uses the client’s defaults.

Return type:

Generator[VideoSample, None, None]

Returns:

VideoSample dictionaries, ordered per channel, without camera synchronization. Consume to exhaustion to receive buffered frames.

Raises:

VideoDecodeError – Unsupported or invalid video, or incomplete decoding. Supports CompressedVideo H.264 Annex B, one image per message, no B-frames, with an initial IDR keyframe containing SPS/PPS. Missing packets can cause undetected corruption until the next keyframe.

class foxglove.datasets.video.VideoSample¶

An RGB image and the metadata of its original compressed message.

Warning

This API is experimental and unstable. It may change in backward-incompatible ways as we continue development and incorporate user feedback.

image is a writable, contiguous uint8 NumPy array of shape [H, W, 3]. All timestamps are integer nanoseconds since the Unix epoch. timestamp_ns is the video’s capture timestamp; log_time_ns determines episode membership. frame_id identifies the camera’s coordinate frame, not a sequence number.

exception foxglove.datasets.video.VideoDecodeError¶

Video data cannot be decoded completely and safely.

__init__(*args, **kwargs)¶

__new__(**kwargs)¶

foxglove.datasets.torch.to_image_tensor(frame)¶

Convert channels-last images to a contiguous channels-first tensor.

Accepts a sample or batch whose image is a NumPy array or Tensor shaped [..., H, W, 3]; returns [..., 3, H, W]. Preserves dtype and device. Call once, either in the episode callback or after DataLoader. Resizing, normalization, and device placement remain up to the training pipeline.

Return type:: Tensor

Dataset loading for ML¶

Installation¶

PyTorch¶

Ray Data¶

Object storage¶

Custom factories¶

Message formats¶

H.264 video¶

API reference¶

PyTorch¶

Ray Data¶

Episode reader¶

Object storage¶

Video decoding¶

Foxglove SDK

Navigation

Related Topics