Datastore Parts API

Data recording and tub management APIs

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

close · () -> 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)