Datastore Parts API

Datastore parts handle recording driving data to tubs and managing tub datasets.

Tub (v2)

The primary datastore class for recording and reading driving data. Constructor:

Tub(base_path, inputs=[], types=[], metadata=[], max_catalog_len=1000, read_only=False)

base_path

str

Path to tub directory

inputs

list

default:"[]"

List of input channel names (e.g., [‘cam/image_array’, ‘user/angle’])

types

list

default:"[]"

List of data types corresponding to inputs (e.g., [‘image_array’, ‘float’])

metadata

list

default:"[]"

List of metadata key:value pairs

max_catalog_len

int

default:"1000"

Maximum catalog size before flushing

read_only

bool

default:"false"

Open tub in read-only mode

Supported Data Types:

'float': Floating point numbers
'int': Integers
'str': Strings
'boolean': Boolean values
'image_array': NumPy image arrays (saved as JPEG)
'gray16_array': 16-bit grayscale images (saved as PNG)
'nparray': Generic NumPy arrays (saved as list)
'list' or 'vector': Lists of values

Methods:

write_record

(record: dict) -> None

Write a single record to the tub

delete_records

(record_indexes: list) -> None

Delete specific records by index

delete_last_n_records

(n: int) -> None

Delete the last N records

restore_records

(record_indexes: list) -> None

Restore previously deleted records

() -> None

Close the tub and flush data

__iter__

() -> Iterator

Iterate over all records in the tub

__len__

() -> int

Get number of records in the tub

Usage Example:

from donkeycar.parts.tub_v2 import Tub
import numpy as np

# Create a new tub
tub = Tub(
    base_path='data/tub_1',
    inputs=['cam/image_array', 'user/angle', 'user/throttle'],
    types=['image_array', 'float', 'float']
)

# Write a record
img = np.random.randint(0, 255, (120, 160, 3), dtype=np.uint8)
record = {
    'cam/image_array': img,
    'user/angle': 0.5,
    'user/throttle': 0.3
}
tub.write_record(record)

# Close when done
tub.close()

TubWriter

Donkeycar part for writing records to a tub. Constructor:

TubWriter(base_path, inputs=[], types=[], metadata=[], max_catalog_len=1000)

base_path

str

Path to tub directory

inputs

list

List of input channel names

types

list

List of data types

metadata

list

Metadata key:value pairs

Methods:

run

(*args) -> int

Donkeycar part interface. Accepts values matching inputs list and writes record. Returns current record index.

shutdown

() -> None

Close the tub

Usage Example:

from donkeycar.parts.tub_v2 import TubWriter

# Create writer
tub_writer = TubWriter(
    base_path='data/tub_1',
    inputs=['cam/image_array', 'user/angle', 'user/throttle', 'user/mode'],
    types=['image_array', 'float', 'float', 'str']
)

# Add to vehicle
V.add(tub_writer,
      inputs=['cam/image_array', 'user/angle', 'user/throttle', 'user/mode'],
      outputs=['tub/num_records'],
      run_condition='recording')

TubHandler

Helper class for managing multiple tubs. Constructor:

TubHandler(path)

path

str

Base path containing tubs

Methods:

get_tub_list

(path: str) -> list

Get list of tub directories

next_tub_number

(path: str) -> int

Get next available tub number

create_tub_path

() -> str

Create a new tub path with incremented number and date

new_tub_writer

(inputs, types, user_meta) -> TubWriter

Create a new TubWriter with auto-generated path

Usage Example:

from donkeycar.parts.datastore import TubHandler

handler = TubHandler(path='data')

# Create new tub with auto-naming
tub_writer = handler.new_tub_writer(
    inputs=['cam/image_array', 'user/angle', 'user/throttle'],
    types=['image_array', 'float', 'float'],
    user_meta=['location:track1', 'driver:alice']
)

TubWiper

Part for deleting recent records during recording (delete bad data on the fly). Constructor:

TubWiper(tub, num_records=20)

tub

Tub or TubWriter

Tub to operate on

num_records

int

default:"20"

Number of records to delete when triggered

Methods:

run

(is_delete: bool) -> None

Delete records when trigger switches from False to True (debounced)

Usage Example:

from donkeycar.parts.tub_v2 import TubWriter, TubWiper

# Create tub writer
tub_writer = TubWriter('data/tub_1', inputs=['cam/image_array'], types=['image_array'])

# Create wiper
wiper = TubWiper(tub=tub_writer.tub, num_records=50)

# Add to vehicle
V.add(wiper, inputs=['delete_records'])

Legacy Tub (v1)

Older tub implementation (deprecated, use Tub v2 instead). Constructor:

Tub(path, inputs=None, types=None, user_meta=[])

Methods: Similar to v2 but with different internal implementation.

Record Structure

Record Format

Each record contains:

User-defined fields (inputs)
Private metadata:
- _timestamp_ms: Timestamp in milliseconds
- _index: Record index
- _session_id: Recording session ID

Example Record

{
  "cam/image_array": "123_cam-image_array_.jpg",
  "user/angle": 0.15,
  "user/throttle": 0.3,
  "user/mode": "user",
  "_timestamp_ms": 1234567890,
  "_index": 123,
  "_session_id": "abc123"
}

Tub Directory Structure

data/tub_1/
├── manifest.json          # Tub metadata and catalog
├── images/               # Image files
│   ├── 0_cam-image_array_.jpg
│   ├── 1_cam-image_array_.jpg
│   └── ...
└── deleted.json          # Deleted record indexes (optional)

Manifest File

The manifest.json contains:

Input channel names and types
Metadata
Current index
Session information
Record catalog

Example:

{
  "inputs": ["cam/image_array", "user/angle", "user/throttle"],
  "types": ["image_array", "float", "float"],
  "metadata": {},
  "current_index": 1234,
  "session_id": "abc123",
  "records": [
    {"_index": 0, "_timestamp_ms": 1234567890, ...},
    {"_index": 1, "_timestamp_ms": 1234567891, ...}
  ]
}

Configuration

Typical datastore configuration in myconfig.py:

# Tub Settings
TUB_PATH = 'data'
MAX_CATALOG_LEN = 1000

# Inputs to record
INPUTS = ['cam/image_array', 'user/angle', 'user/throttle', 'user/mode']
TYPES = ['image_array', 'float', 'float', 'str']

# Auto-delete settings  
NUM_RECORDS_TO_ERASE = 100

Integration Example

From templates like complete.py:

from donkeycar.parts.datastore import TubHandler
from donkeycar.parts.tub_v2 import TubWriter

# Create tub handler
tub_handler = TubHandler(path=cfg.DATA_PATH)

# Create new tub
inputs = ['cam/image_array', 'user/angle', 'user/throttle', 'user/mode']
types = ['image_array', 'float', 'float', 'str']

tub = tub_handler.new_tub_writer(inputs=inputs, types=types)

# Add to vehicle
V.add(tub,
      inputs=inputs,
      outputs=['tub/num_records'],
      run_condition='recording')

Reading Tubs for Training

from donkeycar.parts.tub_v2 import Tub

# Open existing tub
tub = Tub('data/tub_1', read_only=True)

# Iterate over records
for record in tub:
    img = record['cam/image_array']
    angle = record['user/angle']
    throttle = record['user/throttle']
    # ... use for training

# Get tub size
num_records = len(tub)

Core API

Parts API

Datastore Parts API

Tub (v2)

TubWriter

TubHandler

TubWiper

Legacy Tub (v1)

Record Structure

Record Format

Example Record

Tub Directory Structure

Manifest File

Configuration

Integration Example

Reading Tubs for Training

Build docs developers (and LLMs) love

Core API

Parts API

​Tub (v2)

​TubWriter

​TubHandler

​TubWiper

​Legacy Tub (v1)

​Record Structure

​Record Format

​Example Record

​Tub Directory Structure

​Manifest File

​Configuration

​Integration Example

​Reading Tubs for Training

Build docs developers (and LLMs) love

Tub (v2)

TubWriter

TubHandler

TubWiper

Legacy Tub (v1)

Record Structure

Record Format

Example Record

Tub Directory Structure

Manifest File

Configuration

Integration Example

Reading Tubs for Training