Skip to content

I/O Utilities

Reading and writing episode data, converting recordings to LeRobot datasets, image conversion, and reading body-model pickles without chumpy. The datasets and recording guide explains the recording workflow.

File handlers

The recorder manager writes episodes through a file handler, chosen by name from FILE_HANDLER_REGISTRY (HDF5 or NPZ). EpisodeData is the in-memory container of one recorded trajectory.

Base class and registry for dataset file handlers.

Attributes

FILE_HANDLER_REGISTRY module-attribute 

FILE_HANDLER_REGISTRY = Registry('FILE_HANDLER')

EPISODE_ATTRS module-attribute 

EPISODE_ATTRS: tuple[str, ...] = ('seed', 'success', 'env_id')

DEMO_NAME_RE module-attribute 

DEMO_NAME_RE = re.compile('^demo_(\\d+)$')

Classes

FileHandlerBase 

FileHandlerBase(options: T)

Bases: ConfigurableMixin[FileHandlerOptions]

Abstract class for handling dataset files.

Source code in eden/utils/common.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def __init__(self, options: T) -> None:
    self._uid = gs.UID()
    self._options: T = options

    mirror_options_onto(self, self._options)

    # Deferred, not module-scope — see the NOTE above this module's imports.
    from eden.options._metadata import OPTION_METADATA_KEYS

    # check if there are unused parameters in the options. ``_extra_option_params`` are legitimate
    # options-only fields (not mirrored as attrs), so they are expected to be "unused" here.
    unused_params = set(self._options.keys()) - set(self.get_parameter_names())
    tolerated = {"name", "class_name"} | set(OPTION_METADATA_KEYS) | self._extra_option_params
    # ``params`` is a declared framework field on function-capable term options. An *empty* one is
    # legitimately unused when the same options class configures a class-based term; a non-empty
    # ``params`` on a class-based term would be silently ignored, so it stays invalid.
    if not getattr(self._options, "params", None):
        tolerated.add("params")
    invalid_params = unused_params - tolerated
    if invalid_params:
        # Name the term, not only the class: for a function term the class is the shared
        # ``*FuncWrapper``, which cannot tell the author which of a dozen terms is wrong.
        # Same wording as the construction-time check (``ManagerTermOptions._validate_param_names``).
        term_name = getattr(self._options, "name", None)
        subject = f"term '{term_name}' ({type(self).__name__})" if term_name else type(self).__name__
        raise ValueError(
            invalid_parameters_message(subject, invalid_params, self.get_configure_specs(), self._retired_params)
        )

Methods:

resolve_path abstractmethod 
resolve_path(file_path: str) -> str

Return the concrete on-disk path used by this handler.

Source code in eden/utils/file_handler/base.py 
25
26
27
28
@abstractmethod
def resolve_path(self, file_path: str) -> str:
    """Return the concrete on-disk path used by this handler."""
    ...
open abstractmethod 
open(file_path: str, mode: str = 'r', env_cfg: dict | None = None)

Open a file. If env_cfg is provided, implementations may warn on mismatch.

Source code in eden/utils/file_handler/base.py 
30
31
32
33
@abstractmethod
def open(self, file_path: str, mode: str = "r", env_cfg: dict | None = None):
    """Open a file. If `env_cfg` is provided, implementations may warn on mismatch."""
    ...
create abstractmethod 
create(file_path: str, env_cfg: dict)

Create a new file.

Source code in eden/utils/file_handler/base.py 
35
36
37
38
@abstractmethod
def create(self, file_path: str, env_cfg: dict):
    """Create a new file."""
    ...
write_episode abstractmethod 
write_episode(episode: EpisodeData)

Write episode data to the file.

Source code in eden/utils/file_handler/base.py 
40
41
42
43
@abstractmethod
def write_episode(self, episode: EpisodeData):
    """Write episode data to the file."""
    ...
flush abstractmethod 
flush()

Flush the file.

Source code in eden/utils/file_handler/base.py 
45
46
47
48
@abstractmethod
def flush(self):
    """Flush the file."""
    ...
close abstractmethod 
close()

Close the file.

Source code in eden/utils/file_handler/base.py 
50
51
52
53
@abstractmethod
def close(self):
    """Close the file."""
    ...
load_episode abstractmethod 
load_episode(episode_name: str, device: str = 'cpu') -> EpisodeData | None

Load episode data from the file.

Source code in eden/utils/file_handler/base.py 
55
56
57
58
@abstractmethod
def load_episode(self, episode_name: str, device: str = "cpu") -> EpisodeData | None:
    """Load episode data from the file."""
    ...
get_num_episodes abstractmethod 
get_num_episodes() -> int

Get number of episodes in the file.

Source code in eden/utils/file_handler/base.py 
60
61
62
63
@abstractmethod
def get_num_episodes(self) -> int:
    """Get number of episodes in the file."""
    ...

EpisodeData container for recorded trajectories.

Classes

EpisodeData 

EpisodeData()

Class to store episode data.

Source code in eden/utils/file_handler/episode_data.py 
24
25
26
27
28
29
def __init__(self) -> None:
    self.data: dict = {}
    self.seed: int | None = None
    self.env_id: int | None = None
    self.success: bool | None = None
    self._next_index_dict: dict[str, int] = {}

Attributes

data instance-attribute 
data: dict = {}
seed instance-attribute 
seed: int | None = None
env_id instance-attribute 
env_id: int | None = None
success instance-attribute 
success: bool | None = None

Methods:

is_empty 
is_empty()

Check if the episode data is empty.

Source code in eden/utils/file_handler/episode_data.py 
31
32
33
def is_empty(self):
    """Check if the episode data is empty."""
    return not bool(self.data)
add 
add(key: str, value: Tensor | dict)

Add a key-value pair to the dataset.

The key can be nested by using the "/" character. For example: "obs/joint_pos".

PARAMETER DESCRIPTION
key

The key name.

TYPE: str

value

The corresponding value, either a tensor or a (possibly nested) dict.

TYPE: Tensor or dict

Source code in eden/utils/file_handler/episode_data.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def add(self, key: str, value: torch.Tensor | dict):
    """Add a key-value pair to the dataset.

    The key can be nested by using the "/" character.
    For example: "obs/joint_pos".

    Parameters
    ----------
    key : str
        The key name.
    value : torch.Tensor or dict
        The corresponding value, either a tensor or a (possibly nested) dict.
    """
    if isinstance(value, dict):
        for sub_key, sub_value in value.items():
            self.add(f"{key}/{sub_key}", sub_value)
        return

    self._append_owned(key, value.clone())
get 
get(key: str, index: int | None = None, sequential: bool = False)

General getter for any key in the episode data.

PARAMETER DESCRIPTION
key

Hierarchical key, e.g. 'states/obs', 'actions' or 'rewards'.

TYPE: str

index

Specific timestep to retrieve.

TYPE: int DEFAULT: None

sequential

If True, use and increment a per-key pointer.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
Tensor or dict or None

The tensor (or dict of tensors) at the requested timestep, or None if the key is missing or the index is out of range.

Examples:

>>> # Random access
>>> state5 = episode.get("states/obs/joint_pos", index=5)
>>> action3 = episode.get("actions", index=3)
>>> # Sequential access
>>> next_state = episode.get("states/obs", sequential=True)
>>> next_action = episode.get("actions", sequential=True)
Source code in eden/utils/file_handler/episode_data.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get(self, key: str, index: int | None = None, sequential: bool = False):
    """General getter for any key in the episode data.

    Parameters
    ----------
    key : str
        Hierarchical key, e.g. ``'states/obs'``, ``'actions'`` or ``'rewards'``.
    index : int, optional
        Specific timestep to retrieve.
    sequential : bool, optional
        If True, use and increment a per-key pointer.

    Returns
    -------
    torch.Tensor or dict or None
        The tensor (or dict of tensors) at the requested timestep, or None if
        the key is missing or the index is out of range.

    Examples
    --------
    >>> # Random access
    >>> state5 = episode.get("states/obs/joint_pos", index=5)
    >>> action3 = episode.get("actions", index=3)
    >>> # Sequential access
    >>> next_state = episode.get("states/obs", sequential=True)
    >>> next_action = episode.get("actions", sequential=True)
    """
    data_pointer = self._resolve_key(key)
    if data_pointer is None:
        return None

    # Determine the index
    if sequential:
        idx = self._next_index_dict.get(key, 0)
        self._next_index_dict[key] = idx + 1
    elif index is not None:
        idx = index
    else:
        idx = 0

    # Access the data
    if isinstance(data_pointer, (list, torch.Tensor)):
        if idx >= len(data_pointer):
            return None
        return data_pointer[idx] if isinstance(data_pointer, list) else data_pointer[idx, ...]
    elif isinstance(data_pointer, dict):
        return {subkey: self.get(f"{key}/{subkey}", index=idx) for subkey in data_pointer}
    return None
size 
size(key: str)

Return the size of the data at the given key.

PARAMETER DESCRIPTION
key

Hierarchical key, e.g. 'states/obs', 'actions' or 'rewards'.

TYPE: str

RETURNS DESCRIPTION
int or dict or None

Size of the stored value, a dict of sizes for nested keys, or None if the key is missing.
Source code in eden/utils/file_handler/episode_data.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def size(self, key: str):
    """Return the size of the data at the given key.

    Parameters
    ----------
    key : str
        Hierarchical key, e.g. ``'states/obs'``, ``'actions'`` or ``'rewards'``.

    Returns
    -------
    int or dict or None
        Size of the stored value, a dict of sizes for nested keys, or None if
        the key is missing.
    """
    data_pointer = self._resolve_key(key)
    if data_pointer is None:
        return None

    if isinstance(data_pointer, (list, torch.Tensor)):
        return len(data_pointer)
    elif isinstance(data_pointer, dict):
        return {subkey: self.size(f"{key}/{subkey}") for subkey in data_pointer}
    return 0
reset 
reset(clear_data: bool = False)
Source code in eden/utils/file_handler/episode_data.py 
158
159
160
161
def reset(self, clear_data: bool = False):
    self._next_index_dict = {}
    if clear_data:
        self.data = {}
pre_export 
pre_export()

Stack lists into tensors in-place. Destructive — do not call add() after this.

Source code in eden/utils/file_handler/episode_data.py 
163
164
165
def pre_export(self):
    """Stack lists into tensors in-place. Destructive — do not call add() after this."""
    _stack_recursive(self.data, in_place=True)
pre_export_copy 
pre_export_copy()

Return a new EpisodeData with lists stacked into tensors.

Source code in eden/utils/file_handler/episode_data.py 
167
168
169
170
171
172
173
174
def pre_export_copy(self):
    """Return a new EpisodeData with lists stacked into tensors."""
    exported = EpisodeData()
    exported.seed = self.seed
    exported.env_id = self.env_id
    exported.success = self.success
    exported.data = _stack_recursive(self.data, in_place=False)
    return exported

HDF5 dataset file handler.

Attributes

Classes

HDF5FileHandler 

HDF5FileHandler()

Bases: FileHandlerBase

HDF5 dataset file handler for storing and loading episode data.

The format of the stored structure dataset_file.hdf5 └── data/ # _hdf5_data_group ├── demo_0/ # Episode group │ ├── actions # Dataset (Tensor saved as NumPy array) │ ├── states # Nested dict -> datasets │ │ └── obs # Dataset │ └── ... # Any other keys │ ├── attrs: │ │ seed: int │ │ success: bool │ │ env_id: int ├── demo_1/ │ └── ... └── ... # Additional episodes attrs of data group: env_cfg: str (JSON) # Serialized environment configuration total: int # Number of episodes written

Initialize the HDF5 dataset file handler.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
46
47
48
49
50
51
52
53
54
55
def __init__(self):
    """Initialize the HDF5 dataset file handler."""
    self._hdf5_file_stream = None
    self._hdf5_data_group = None
    # _next_demo_id is the next writable demo index; _demo_count is the actual
    # episode count. They diverge when a resumed dataset has gaps in numbering
    # (e.g. demo_0, demo_2 after a crash → next_id=3, count=2), and keeping both means
    # the next write never collides and get_num_episodes() never over-reports.
    self._next_demo_id = 0
    self._demo_count = 0

Attributes

demo_count property 
demo_count: int

Methods:

resolve_path 
resolve_path(file_path: str) -> str
Source code in eden/utils/file_handler/hdf5_file_handler.py 
57
58
59
60
def resolve_path(self, file_path: str) -> str:
    if not file_path.endswith(".hdf5"):
        file_path += ".hdf5"
    return file_path
open 
open(file_path: str, mode: str = 'r', env_cfg: dict | None = None)

Open an existing dataset file.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def open(self, file_path: str, mode: str = "r", env_cfg: dict | None = None):
    """Open an existing dataset file."""
    if self._hdf5_file_stream is not None:
        raise RuntimeError("HDF5 dataset file stream is already in use")
    file_path = self.resolve_path(file_path)
    self._hdf5_file_stream = h5py.File(file_path, mode)

    if "data" not in self._hdf5_file_stream:
        raise RuntimeError("Invalid dataset file: missing 'data' group")

    self._hdf5_data_group = self._hdf5_file_stream["data"]
    existing_indices = [
        int(m.group(1)) for m in (_DEMO_NAME_RE.fullmatch(name) for name in self._hdf5_data_group.keys()) if m
    ]
    self._next_demo_id = max(existing_indices) + 1 if existing_indices else 0
    self._demo_count = len(existing_indices)

    if env_cfg is not None and "env_cfg" in self._hdf5_data_group.attrs:
        try:
            stored_cfg = json.loads(self._hdf5_data_group.attrs["env_cfg"])
        except (ValueError, json.JSONDecodeError):
            stored_cfg = None
        if stored_cfg is not None and stored_cfg != env_cfg:
            warnings.warn(
                f"Resumed HDF5 dataset '{file_path}' was recorded with a different env_cfg; "
                "mixing episodes from different configs may break downstream consumers.",
                stacklevel=2,
            )
create 
create(file_path: str, env_cfg: dict)

Create a new dataset file.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def create(self, file_path: str, env_cfg: dict):
    """Create a new dataset file."""
    if self._hdf5_file_stream is not None:
        raise RuntimeError("HDF5 dataset file stream is already in use")
    file_path = self.resolve_path(file_path)
    os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)
    self._hdf5_file_stream = h5py.File(file_path, "w")

    # set up a data group in the file
    self._hdf5_data_group = self._hdf5_file_stream.create_group("data")
    self._hdf5_data_group.attrs["env_cfg"] = json.dumps(env_cfg)
    self._hdf5_data_group.attrs["total"] = 0
    self._next_demo_id = 0
    self._demo_count = 0
get_episode_names 
get_episode_names() -> Iterable[str]

Get the names of the episodes in the file.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
114
115
116
117
def get_episode_names(self) -> Iterable[str]:
    """Get the names of the episodes in the file."""
    self._raise_if_not_initialized()
    return list(self._hdf5_data_group.keys())
get_num_episodes 
get_num_episodes() -> int
Source code in eden/utils/file_handler/hdf5_file_handler.py 
119
120
121
def get_num_episodes(self) -> int:
    self._raise_if_not_initialized()
    return self._demo_count
load_episode 
load_episode(episode_name: str, device: str = 'cpu') -> EpisodeData | None

Load episode data from the file.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def load_episode(self, episode_name: str, device: str = "cpu") -> EpisodeData | None:
    """Load episode data from the file."""
    self._raise_if_not_initialized()
    if episode_name not in self._hdf5_data_group:
        return None
    episode = EpisodeData()
    h5_episode_group = self._hdf5_data_group[episode_name]

    def load_dataset_helper(group):
        """Load a dataset that contains recursive dict objects."""
        data = {}
        for key in group:
            if isinstance(group[key], h5py.Group):
                data[key] = load_dataset_helper(group[key])
            else:
                # Converting group[key] to numpy array greatly improves the performance
                # when converting to torch tensor
                data[key] = torch.tensor(np.array(group[key]), device=device)
        return data

    episode.data = load_dataset_helper(h5_episode_group)

    for attr in _EPISODE_ATTRS:
        if attr in h5_episode_group.attrs:
            value = h5_episode_group.attrs[attr]
            if attr == "success":
                value = bool(value)
            elif attr in ("seed", "env_id"):
                value = int(value)
            setattr(episode, attr, value)

    return episode
write_episode 
write_episode(episode: EpisodeData)

Add an episode to the dataset.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def write_episode(self, episode: EpisodeData):
    """Add an episode to the dataset."""
    self._raise_if_not_initialized()
    if episode.is_empty():
        return

    episode_group_name = f"demo_{self._next_demo_id}"

    if episode_group_name in self._hdf5_data_group:
        raise ValueError(f"Episode group '{episode_group_name}' already exists in the dataset")

    h5_episode_group = self._hdf5_data_group.create_group(episode_group_name)

    for attr in _EPISODE_ATTRS:
        val = getattr(episode, attr)
        if val is not None:
            h5_episode_group.attrs[attr] = val

    def create_dataset_helper(group, key, value):
        """Create a dataset that contains recursive dict objects."""
        if isinstance(value, dict):
            key_group = group.create_group(key)
            for sub_key, sub_value in value.items():
                create_dataset_helper(key_group, sub_key, sub_value)
        else:
            data = to_numpy(value)
            group.create_dataset(key, data=data, compression="gzip")

    for key, value in episode.data.items():
        create_dataset_helper(h5_episode_group, key, value)

    self._next_demo_id += 1
    self._demo_count += 1
    self._hdf5_data_group.attrs["total"] = self._demo_count
flush 
flush()

Flush the episode data to disk.

Source code in eden/utils/file_handler/hdf5_file_handler.py 
198
199
200
201
def flush(self):
    """Flush the episode data to disk."""
    self._raise_if_not_initialized()
    self._hdf5_file_stream.flush()
close 
close()
Source code in eden/utils/file_handler/hdf5_file_handler.py 
203
204
205
206
207
208
209
def close(self):
    if self._hdf5_file_stream is not None:
        self._hdf5_file_stream.close()
        self._hdf5_file_stream = None
        self._hdf5_data_group = None
        self._next_demo_id = 0
        self._demo_count = 0

Functions:

NPZ dataset file handler.

Attributes

Classes

NPZFileHandler 

NPZFileHandler()

Bases: FileHandlerBase

NPZ dataset file handler for storing and loading episode data.

Since NPZ files are flat key-value stores of numpy arrays, we use a naming convention to represent the hierarchical episode structure:

dataset_file.npz
├── __meta__env_cfg              # JSON string stored as bytes array
├── __meta__total                # scalar: number of episodes
├── demo_0/actions               # episode data (slash-separated hierarchy)
├── demo_0/states/obs            # nested episode data
├── demo_0/__attrs__             # episode attrs as a JSON bytes array
├── demo_1/actions
└── ...
Source code in eden/utils/file_handler/npz_file_handler.py 
77
78
79
80
81
82
83
84
85
86
def __init__(self):
    self._file_path: str | None = None
    self._arrays: dict[str, np.ndarray] = {}
    # _next_demo_id is the next writable demo index; _demo_count is the actual
    # episode count. They diverge when a resumed dataset has gaps in numbering, and
    # keeping both means the next write never collides and get_num_episodes() never over-reports.
    self._next_demo_id: int = 0
    self._demo_count: int = 0
    self._writable: bool = False
    self._dirty: bool = False

Attributes

demo_count property 
demo_count: int

Methods:

resolve_path 
resolve_path(file_path: str) -> str
Source code in eden/utils/file_handler/npz_file_handler.py 
88
89
90
91
def resolve_path(self, file_path: str) -> str:
    if not file_path.endswith(".npz"):
        file_path += ".npz"
    return file_path
open 
open(file_path: str, mode: str = 'r', env_cfg: dict | None = None)
Source code in eden/utils/file_handler/npz_file_handler.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def open(self, file_path: str, mode: str = "r", env_cfg: dict | None = None):
    if self._file_path is not None:
        raise RuntimeError("NPZ dataset file handler is already in use")

    file_path = self.resolve_path(file_path)
    with np.load(file_path, allow_pickle=False) as data:
        self._arrays = {k: np.array(v) for k, v in data.items()}
    self._file_path = file_path
    self._writable = mode != "r"
    self._dirty = False

    if f"{_META_PREFIX}total" not in self._arrays:
        raise RuntimeError("Invalid dataset file: missing metadata")

    existing_indices = [
        int(m.group(1)) for m in (_DEMO_NAME_RE.fullmatch(name) for name in self._collect_episode_names()) if m
    ]
    self._next_demo_id = max(existing_indices) + 1 if existing_indices else 0
    self._demo_count = len(existing_indices)

    if env_cfg is not None and f"{_META_PREFIX}env_cfg" in self._arrays:
        try:
            stored_cfg = _decode_bytes_to_json(self._arrays[f"{_META_PREFIX}env_cfg"])
        except (ValueError, json.JSONDecodeError):
            stored_cfg = None
        if stored_cfg is not None and stored_cfg != env_cfg:
            import warnings

            warnings.warn(
                f"Resumed NPZ dataset '{file_path}' was recorded with a different env_cfg; "
                "mixing episodes from different configs may break downstream consumers.",
                stacklevel=2,
            )
create 
create(file_path: str, env_cfg: dict)
Source code in eden/utils/file_handler/npz_file_handler.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create(self, file_path: str, env_cfg: dict):
    if self._file_path is not None:
        raise RuntimeError("NPZ dataset file handler is already in use")
    file_path = self.resolve_path(file_path)
    os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)

    self._file_path = file_path
    self._writable = True
    self._arrays = {}
    self._dirty = True

    self._arrays[f"{_META_PREFIX}total"] = np.array(0)
    self._arrays[f"{_META_PREFIX}env_cfg"] = _encode_json_to_bytes(env_cfg)
    self._next_demo_id = 0
    self._demo_count = 0
get_episode_names 
get_episode_names() -> Iterable[str]
Source code in eden/utils/file_handler/npz_file_handler.py 
154
155
156
def get_episode_names(self) -> Iterable[str]:
    self._raise_if_not_initialized()
    return sorted(self._collect_episode_names())
get_num_episodes 
get_num_episodes() -> int
Source code in eden/utils/file_handler/npz_file_handler.py 
158
159
160
def get_num_episodes(self) -> int:
    self._raise_if_not_initialized()
    return self._demo_count
load_episode 
load_episode(episode_name: str, device: str = 'cpu') -> EpisodeData | None
Source code in eden/utils/file_handler/npz_file_handler.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def load_episode(self, episode_name: str, device: str = "cpu") -> EpisodeData | None:
    self._raise_if_not_initialized()

    prefix = f"{episode_name}/"
    attrs_key = f"{episode_name}{_ATTRS_SUFFIX}"

    # Collect all data arrays for this episode
    flat_data: dict[str, np.ndarray] = {}
    has_any = False
    for key, value in self._arrays.items():
        if not key.startswith(prefix):
            continue
        has_any = True
        # Skip the attrs key
        if key == attrs_key:
            continue
        # Strip episode name prefix
        sub_key = key[len(prefix) :]
        flat_data[sub_key] = value

    if not has_any:
        return None

    episode = EpisodeData()
    episode.data = _unflatten_dict(flat_data, device=device)

    # Load attributes
    if attrs_key in self._arrays:
        attrs = _decode_bytes_to_json(self._arrays[attrs_key])
        for attr in _EPISODE_ATTRS:
            if attr in attrs:
                value = attrs[attr]
                if attr == "success":
                    value = bool(value)
                elif attr in ("seed", "env_id"):
                    value = int(value)
                setattr(episode, attr, value)

    return episode
write_episode 
write_episode(episode: EpisodeData)
Source code in eden/utils/file_handler/npz_file_handler.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def write_episode(self, episode: EpisodeData):
    self._raise_if_not_initialized()
    if episode.is_empty():
        return

    episode_name = f"demo_{self._next_demo_id}"

    # Check for duplicates
    prefix = f"{episode_name}/"
    if any(k.startswith(prefix) for k in self._arrays):
        raise ValueError(f"Episode '{episode_name}' already exists in the dataset")

    # Flatten and store episode data
    flat = _flatten_dict(episode.data, episode_name)
    self._arrays.update(flat)

    # Store episode attributes
    attrs = {}
    for attr in _EPISODE_ATTRS:
        val = getattr(episode, attr)
        if val is not None:
            attrs[attr] = val
    if attrs:
        self._arrays[f"{episode_name}{_ATTRS_SUFFIX}"] = _encode_json_to_bytes(attrs)

    self._next_demo_id += 1
    self._demo_count += 1
    self._arrays[f"{_META_PREFIX}total"] = np.array(self._demo_count)
    self._dirty = True
flush 
flush()
Source code in eden/utils/file_handler/npz_file_handler.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
def flush(self):
    # NPZ has no incremental write format, so flush rewrites the full archive,
    # atomically (tmp file + rename) so a kill mid-write cannot leave a
    # partially-written dataset. Its cost is O(total dataset size): for large
    # datasets prefer close() over a flush() per episode.
    self._raise_if_not_initialized()
    if not self._writable or not self._dirty:
        return
    tmp_path = f"{self._file_path}.tmp.npz"
    with open(tmp_path, "wb") as f:
        np.savez_compressed(f, **self._arrays)
    os.replace(tmp_path, self._file_path)
    self._dirty = False
close 
close()
Source code in eden/utils/file_handler/npz_file_handler.py 
253
254
255
256
257
258
259
260
261
262
def close(self):
    if self._file_path is not None:
        if self._writable:
            self.flush()
        self._file_path = None
        self._arrays = {}
        self._next_demo_id = 0
        self._demo_count = 0
        self._writable = False
        self._dirty = False

Functions:

Configuration options for dataset file handlers.

Classes

FileHandlerOptions 

FileHandlerOptions(**data)

Bases: DynamicConfigurableOptions

Source code in eden/options/options.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def __init__(self, **data):
    # Seed type metadata as construction data so BaseModel validates it into
    # __pydantic_extra__ (no-op for extra="forbid" classes, which use
    # the serialization envelope instead).
    data.update(option_metadata(self))
    # NOTE: all of the option modifications/validations should be done in the __init__ method
    try:
        BaseModel.__init__(self, **data)
    except ValidationError as exc:
        # A misspelled field on an extra="forbid" class should teach the field set, not just
        # name the typo. Only rewritten when every error is extra_forbidden.
        enriched = _enrich_extra_forbidden(type(self), exc)
        if enriched is None:
            raise
        raise enriched from None

LeRobot conversion

Convert an Eden HDF5 recording into a LeRobot v3 dataset (eden lerobot convert).

Convert an Eden teleoperation HDF5 recording into a LeRobot v3 dataset.

Used as a library:

from eden.utils.lerobot import convert_hdf5_to_lerobot
convert_hdf5_to_lerobot([...])  # argparse-style argv

or via CLI: eden lerobot convert --hdf5 ... --output-dir ....

Attributes

LEROBOT_INSTALL module-attribute 

LEROBOT_INSTALL = 'pip install -U "lerobot[dataset]>=0.4"'

Classes

Functions:

convert 

convert(hdf5_path: Path, output_dir: Path, repo_id: str, robot_type: str, task: str, force: bool = False, image_writer_threads: int = 4) -> None

Convert an HDF5 recording into a LeRobot v3 dataset on disk.

Source code in eden/utils/lerobot/convert.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def convert(
    hdf5_path: Path,
    output_dir: Path,
    repo_id: str,
    robot_type: str,
    task: str,
    force: bool = False,
    image_writer_threads: int = 4,
) -> None:
    """Convert an HDF5 recording into a LeRobot v3 dataset on disk."""
    # Before --force deletes anything: a missing or outdated lerobot must not cost the user their old dataset.
    LeRobotDataset = _import_lerobot_dataset()

    if not hdf5_path.is_file():
        raise FileNotFoundError(f"Input HDF5 file not found: {hdf5_path}")

    if output_dir.exists():
        if not force:
            raise FileExistsError(f"Output directory already exists: {output_dir}")
        if output_dir.is_symlink() or not output_dir.is_dir():
            raise ValueError(f"Refusing to --force overwrite non-directory path: {output_dir}")
        if any(output_dir.iterdir()) and not _looks_like_lerobot_dataset(output_dir):
            raise ValueError(
                f"Refusing to --force overwrite {output_dir}: it is not empty and does not look "
                "like a LeRobot dataset (missing meta/info.json). Delete it manually if intended."
            )
        shutil.rmtree(output_dir)

    with h5py.File(hdf5_path, "r") as hdf5_file:
        demos = _list_demos(hdf5_file)
        features = _build_features(hdf5_file, demos)
        fps = _infer_fps(hdf5_file[f"data/{demos[0]}/timestamp"][:])

        dataset = LeRobotDataset.create(
            repo_id=repo_id,
            root=output_dir,
            fps=fps,
            robot_type=robot_type,
            features=features,
            use_videos=True,
            image_writer_threads=image_writer_threads,
        )

        for demo in demos:
            demo_group = hdf5_file[f"data/{demo}"]
            # Bulk-load each demo's arrays once instead of per-step h5 indexing.
            actions = np.asarray(demo_group["action"][:], dtype=np.float32)
            timestamps = _normalize_timestamps(demo_group["timestamp"][:])
            qpos = np.asarray(demo_group["state/qpos"][:], dtype=np.float32)
            dofs_vel = np.asarray(demo_group["state/dofs_vel"][:], dtype=np.float32)
            pos = np.asarray(demo_group["state/pos"][:], dtype=np.float32)
            quat = np.asarray(demo_group["state/quat"][:], dtype=np.float32)

            expected_lengths = {
                "action": actions.shape[0],
                "timestamp": timestamps.shape[0],
                "qpos": qpos.shape[0],
                "dofs_vel": dofs_vel.shape[0],
                "pos": pos.shape[0],
                "quat": quat.shape[0],
            }
            if len(set(expected_lengths.values())) != 1:
                raise ValueError(f"Length mismatch in {demo}: {expected_lengths}")

            states = np.concatenate([qpos, dofs_vel, pos, quat], axis=1)
            length = int(actions.shape[0])
            for t in range(length):
                dataset.add_frame(
                    {
                        "observation.state": states[t],
                        "action": actions[t],
                        "task": task,
                    }
                )

            dataset.save_episode()

        dataset.finalize()

convert_hdf5_to_lerobot 

convert_hdf5_to_lerobot(argv: list[str] | None = None)

CLI entry point: eden lerobot convert ....

Source code in eden/utils/lerobot/convert.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def convert_hdf5_to_lerobot(argv: list[str] | None = None):
    """CLI entry point: ``eden lerobot convert ...``."""
    from argparse import ArgumentParser

    description = (
        f"{colorize('Convert an Eden HDF5 recording into a LeRobot v3 dataset.', ANSI_BOLD, ANSI_CYAN)}\n"
        "\n"
        "Reads `state/qpos`, `state/dofs_vel`, `state/pos`, `state/quat`, `action`, and\n"
        "`timestamp` from each `demo_*` group under `/data` (the schema produced by\n"
        f"{colorize('StateRecorder', ANSI_GREEN)} + {colorize('ActionRecorder', ANSI_GREEN)}), "
        "then writes a LeRobotDataset rooted at\n"
        f"{colorize('--output-dir', ANSI_GREEN)}."
    )
    epilog = (
        f"{colorize('Examples:', ANSI_BOLD, ANSI_YELLOW)}\n"
        f"  {colorize('# Convert a teleop recording with the default robot_type:', ANSI_DIM)}\n"
        "  eden lerobot convert \\\n"
        f"      {colorize('--hdf5', ANSI_GREEN)} eden_teleoperation/teleoperation.hdf5 \\\n"
        f"      {colorize('--output-dir', ANSI_GREEN)} lerobot_output \\\n"
        f"      {colorize('--repo-id', ANSI_GREEN)} local/teleop \\\n"
        f"      {colorize('--force', ANSI_GREEN)}"
    )
    parser = ArgumentParser(
        prog="eden lerobot convert",
        description=description,
        epilog=epilog,
        formatter_class=ColorHelpFormatter,
    )
    parser.add_argument("--hdf5", type=Path, required=True, help="Path to the input HDF5 file.")
    parser.add_argument(
        "--output-dir",
        type=Path,
        required=True,
        help="Directory where the LeRobot dataset will be created.",
    )
    parser.add_argument(
        "--repo-id",
        type=str,
        default="local/eden_teleoperation",
        help="LeRobot dataset repo_id.",
    )
    parser.add_argument(
        "--robot-type",
        type=str,
        default="eden",
        help="Robot type recorded in LeRobot metadata.",
    )
    parser.add_argument(
        "--task",
        type=str,
        default="teleoperate the robot",
        help="Task string stored for every frame.",
    )
    parser.add_argument(
        "--image-writer-threads",
        type=int,
        default=4,
        help="Number of image writer threads used by LeRobot when encoding videos.",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="Overwrite --output-dir if it already exists (must be empty or a LeRobot dataset).",
    )
    args = parser.parse_args(argv)

    convert(
        hdf5_path=args.hdf5,
        output_dir=args.output_dir,
        repo_id=args.repo_id,
        robot_type=args.robot_type,
        task=args.task,
        force=args.force,
        image_writer_threads=args.image_writer_threads,
    )

Images

Image conversion and resizing helpers.

Eden camera terms (:func:eden.managers.terms.observations.camera.rgb_image) return (num_envs, H, W, C) uint8 tensors. Consumers (policy servers, recorders, ...) differ only in layout (HWC vs CHW) and in how they want the image resized, so those are the two axes exposed here:

  • :func:to_uint8_hwc / :func:to_chw_uint8 — layout, no resampling.
  • :func:resize_uint8 — plain resize (cv2 bilinear; Isaac-GR00T's SimplerEnv eval).
  • :func:resize_with_pad — aspect-preserving letterbox resize, a bit-exact port of openpi_client.image_tools.resize_with_pad (pinned by tests/utils/test_image.py) so OpenPI-family clients do not need openpi_client installed.

Functions:

to_uint8_hwc 

to_uint8_hwc(image: Tensor | ndarray) -> ndarray

(1, H, W, C) or (H, W, C) -> (H, W, 3) uint8 numpy.

A 4-channel (RGBA, as the madrona/Nyx renderers return) input has its alpha channel dropped. Any other channel count, dtype, or a batch axis larger than one raises: a float image or a batched render reaching a server is a wiring error, not something to rescale or slice quietly.

Source code in eden/utils/image.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def to_uint8_hwc(image: torch.Tensor | np.ndarray) -> np.ndarray:
    """``(1, H, W, C)`` or ``(H, W, C)`` -> ``(H, W, 3)`` ``uint8`` numpy.

    A 4-channel (RGBA, as the madrona/Nyx renderers return) input has its alpha channel
    dropped. Any other channel count, dtype, or a batch axis larger than one raises: a
    float image or a batched render reaching a server is a wiring error, not something
    to rescale or slice quietly.
    """
    arr = image.detach().cpu().numpy() if isinstance(image, torch.Tensor) else np.asarray(image)
    if arr.ndim == 4:
        if arr.shape[0] != 1:
            raise ValueError(f"Expected a single-env image with leading axis 1, got shape {arr.shape}.")
        arr = arr[0]
    if arr.ndim != 3:
        raise ValueError(f"Expected an HxWxC image (optionally with a leading batch axis), got shape {arr.shape}.")
    if arr.shape[-1] == 4:
        arr = arr[..., :3]
    if arr.shape[-1] != 3:
        raise ValueError(f"Expected 3 (or 4 with alpha) channels, got shape {arr.shape}.")
    if arr.dtype != np.uint8:
        raise TypeError(
            f"Expected a uint8 image, got dtype {arr.dtype}. Eden's rgb_image term returns uint8; "
            "convert explicitly at the source if a renderer produces floats."
        )
    return arr

to_chw_uint8 

to_chw_uint8(image: Tensor | ndarray) -> ndarray

(1, H, W, C) or (H, W, C) -> contiguous (3, H, W) uint8 numpy.

Source code in eden/utils/image.py 
48
49
50
def to_chw_uint8(image: torch.Tensor | np.ndarray) -> np.ndarray:
    """``(1, H, W, C)`` or ``(H, W, C)`` -> contiguous ``(3, H, W)`` ``uint8`` numpy."""
    return np.ascontiguousarray(to_uint8_hwc(image).transpose(2, 0, 1))

resize_uint8 

resize_uint8(image_np: ndarray, image_size: tuple[int, int]) -> ndarray

Resize an (H, W, 3) uint8 image to image_size=(h, w) with cv2 bilinear filtering.

No-op when the size already matches. Matches gr00t/eval/sim/SimplerEnv/simpler_env.py.

Source code in eden/utils/image.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def resize_uint8(image_np: np.ndarray, image_size: tuple[int, int]) -> np.ndarray:
    """Resize an ``(H, W, 3)`` ``uint8`` image to ``image_size=(h, w)`` with cv2 bilinear filtering.

    No-op when the size already matches. Matches ``gr00t/eval/sim/SimplerEnv/simpler_env.py``.
    """
    import cv2

    h, w = image_size
    if image_np.shape[:2] != (h, w):
        image_np = cv2.resize(image_np, (w, h))
    if image_np.dtype != np.uint8:
        image_np = image_np.astype(np.uint8)
    return image_np

resize_with_pad 

resize_with_pad(images: ndarray, height: int, width: int) -> ndarray

Letterbox-resize [..., H, W, C] images to (height, width) without distortion.

Port of openpi_client.image_tools.resize_with_pad (itself replicating tf.image.resize_with_pad): scale by the larger of the two ratios with PIL bilinear filtering, then paste centred on a zero canvas. Returns the input unchanged when the size already matches.

Source code in eden/utils/image.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def resize_with_pad(images: np.ndarray, height: int, width: int) -> np.ndarray:
    """Letterbox-resize ``[..., H, W, C]`` images to ``(height, width)`` without distortion.

    Port of ``openpi_client.image_tools.resize_with_pad`` (itself replicating
    ``tf.image.resize_with_pad``): scale by the larger of the two ratios with PIL bilinear
    filtering, then paste centred on a zero canvas. Returns the input unchanged when the
    size already matches.
    """
    from PIL import Image

    if images.shape[-3:-1] == (height, width):
        return images

    original_shape = images.shape
    flat = images.reshape(-1, *original_shape[-3:])
    resized = np.stack([_resize_with_pad_pil(Image.fromarray(im), height, width) for im in flat])
    return resized.reshape(*original_shape[:-3], *resized.shape[-3:])

convert_to_uint8 

convert_to_uint8(img: ndarray) -> ndarray

[0, 1] float image -> uint8; uint8 input is returned unchanged.

Port of openpi_client.image_tools.convert_to_uint8. Kept only for the OpenPI pipeline, where it is documented as the last preprocessing step; new code should hand :func:to_uint8_hwc a uint8 render instead.

Source code in eden/utils/image.py 
106
107
108
109
110
111
112
113
114
115
def convert_to_uint8(img: np.ndarray) -> np.ndarray:
    """``[0, 1]`` float image -> ``uint8``; ``uint8`` input is returned unchanged.

    Port of ``openpi_client.image_tools.convert_to_uint8``. Kept only for the OpenPI
    pipeline, where it is documented as the last preprocessing step; new code should
    hand :func:`to_uint8_hwc` a ``uint8`` render instead.
    """
    if np.issubdtype(img.dtype, np.floating):
        img = (255 * img).astype(np.uint8)
    return img

prepare_image_padded 

prepare_image_padded(image: Tensor | ndarray, image_size: tuple[int, int]) -> ndarray

Eden render -> (h, w, 3) uint8 via :func:resize_with_pad (the OpenPI client pipeline).

Source code in eden/utils/image.py 
118
119
120
121
122
def prepare_image_padded(image: torch.Tensor | np.ndarray, image_size: tuple[int, int]) -> np.ndarray:
    """Eden render -> ``(h, w, 3)`` ``uint8`` via :func:`resize_with_pad` (the OpenPI client pipeline)."""
    img = to_uint8_hwc(image)
    h, w = image_size
    return convert_to_uint8(resize_with_pad(img, h, w))

prepare_image_resized 

prepare_image_resized(image: Tensor | ndarray, image_size: tuple[int, int]) -> ndarray

Eden render -> (h, w, 3) uint8 via :func:resize_uint8 (the GR00T client pipeline).

Source code in eden/utils/image.py 
125
126
127
def prepare_image_resized(image: torch.Tensor | np.ndarray, image_size: tuple[int, int]) -> np.ndarray:
    """Eden render -> ``(h, w, 3)`` ``uint8`` via :func:`resize_uint8` (the GR00T client pipeline)."""
    return resize_uint8(to_uint8_hwc(image), image_size)

Body-model pickles

MANO, SMPL and SMPL-H .pkl files pickle chumpy arrays; these helpers read them without installing chumpy.

Read the body-model pickles that hold chumpy arrays (MANO, SMPL, SMPL-H) without chumpy.

Those releases pickle shapedirs as a chumpy.ch.Ch (SMPL) or a chumpy.reordering.Select over one (MANO, SMPL-H). chumpy itself is unmaintained, its PyPI release does not build on a current numpy, and a published wheel cannot depend on the git commit that does, so Eden reads the two classes itself: :class:ChumpyFreeUnpickler for Eden's own loader (:func:eden.datasets.utils.body_model.body_model_arrays), and :func:install_chumpy_pickle_modules for a third-party loader that calls a plain pickle.load (smplx, in the SOMA retargeter's worker process).

numpy-only on purpose: the SOMA worker imports it and must stay free of Genesis.

Classes

ChumpyFreeUnpickler

Bases: Unpickler

Unpickles a body-model .pkl with :data:_CHUMPY_CLASSES standing in for chumpy's, so the file reads without chumpy installed (its PyPI release does not build on a current numpy). Every other class resolves as pickle.load would resolve it.

Methods:

find_class 
find_class(module: str, name: str)
Source code in eden/utils/chumpy_pickle.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
def find_class(self, module: str, name: str):
    if module == "chumpy" or module.startswith("chumpy."):
        cls = _CHUMPY_CLASSES.get((module, name))
        if cls is None:
            known = ", ".join(f"{m}.{n}" for m, n in _CHUMPY_CLASSES)
            raise pickle.UnpicklingError(
                f"This body-model pickle holds a {module}.{name}, a chumpy class Eden does not read (it reads "
                f"{known}: every chumpy class in the MANO, SMPL and SMPL-H releases). Use the model's .npz "
                f"instead (put it next to the .pkl), or convert the .pkl to an .npz with the same keys. "
                f"Installing chumpy does not help: Eden never imports it."
            )
        return cls
    return super().find_class(module, name)

Functions:

install_chumpy_pickle_modules 

install_chumpy_pickle_modules() -> None

Make chumpy.ch.Ch / chumpy.reordering.Select resolve to Eden's stand-ins, process-wide.

For a third-party loader that reads a model .pkl with a plain pickle.load (smplx, under the SOMA retargeter), which a custom unpickler cannot reach. It registers chumpy, chumpy.ch and chumpy.reordering in sys.modules, shadowing an installed chumpy for the rest of the process, so only a process that exists to run such a loader calls it (the SOMA worker). The stand-ins read every MANO, SMPL and SMPL-H release bit-identically to chumpy. A chumpy the process has already imported is left in place; a sys.modules["chumpy"] = None (an import blocked on purpose) is not an import, and is replaced.

Source code in eden/utils/chumpy_pickle.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def install_chumpy_pickle_modules() -> None:
    """Make ``chumpy.ch.Ch`` / ``chumpy.reordering.Select`` resolve to Eden's stand-ins, process-wide.

    For a third-party loader that reads a model ``.pkl`` with a plain ``pickle.load`` (smplx, under the SOMA
    retargeter), which a custom unpickler cannot reach. It registers ``chumpy``, ``chumpy.ch`` and
    ``chumpy.reordering`` in ``sys.modules``, shadowing an installed chumpy for the rest of the process, so only a
    process that exists to run such a loader calls it (the SOMA worker). The stand-ins read every MANO, SMPL and
    SMPL-H release bit-identically to chumpy. A chumpy the process has already imported is left in place; a
    ``sys.modules["chumpy"] = None`` (an import blocked on purpose) is not an import, and is replaced.
    """
    if sys.modules.get("chumpy") is not None:
        return
    modules = {name: types.ModuleType(name) for name in ("chumpy", "chumpy.ch", "chumpy.reordering")}
    for (module, name), cls in _CHUMPY_CLASSES.items():
        setattr(modules[module], name, cls)
    modules["chumpy"].ch = modules["chumpy.ch"]
    modules["chumpy"].reordering = modules["chumpy.reordering"]
    sys.modules.update(modules)

Buffers

The buffers in eden/utils/buffers/ (CircularBuffer for observation history, ConcatBatchTensor, and the ResetStateBuffer that the ResetFromStateBuffer event restores from) are not rendered here: the directory has no __init__.py. Read their docstrings in the source.

See also