Datasets

Motion datasets load human or robot motion (motion capture, retargeted clips, hand-object recordings) into the BatchMotionSequence layout the motion manager plays back. Each loader is a DatasetBase subclass that fetches its files, parses them, caches the result and applies DatasetModifiers. The motion datasets guide covers downloading, licences and the runnable examples.

from eden.datasets import DATASET_REGISTRY, LAFAN1RetargetedG1Dataset
from eden.options.datasets import DatasetOptions

Datasets and modifiers register under a snake-case key derived from the class name.

Base class and registry

Base class and registry for motion datasets backed by HuggingFace repos.

Attributes

DATASET_REGISTRY module-attribute

DATASET_REGISTRY = Registry('DATASET')

Classes

DatasetBase

DatasetBase(options: DatasetOptions, device: device)

Bases: ConfigurableMixin[DatasetOptions], FilterMixin

Base class for motion datasets backed by a HuggingFace repo.

PARAMETER	DESCRIPTION
registry	HuggingFace namespace (user or organisation) that hosts the dataset repo. Subclasses should set this to enable automatic downloads, e.g. registry = "ember-lab-berkeley". TYPE: str | None
dataset	HuggingFace dataset repository name, e.g. "AMASS_Retargeted_for_G1". TYPE: str | None
raw_data_path	Path to the raw data directory. When set, it is used as-is and must already exist — a missing explicit path raises rather than downloading. When None and the class defines registry / dataset, the HF repo is downloaded to the HuggingFace global cache. TYPE: str | None
processed_data_path	Path to the processed data file used by this dataset. TYPE: str | None
target_framerate	The target framerate of the motion. This must match the simulation frequency. Defaults to 50.0. TYPE: float
drop_first_n_frames	Number of frames to drop from the start of each raw sequence during prepare(). Applied by subclasses when slicing loaded motions, through :meth:_frame_window. Defaults to 0. TYPE: int
drop_last_n_frames	Number of frames to drop from the end of each raw sequence during prepare(). Defaults to 0. TYPE: int
modifiers	Offline transforms applied after cache load or raw prepare; never baked into the cache. TYPE: tuple[DatasetModifierOptions, ...] | list[DatasetModifierOptions]
include_pattern	Regexes re.search-ed against each motion file's basename (see :class:DatasetOptions). TYPE: str | None
exclude_pattern	Regexes re.search-ed against each motion file's basename (see :class:DatasetOptions). TYPE: str | None
include_list_file	YAML/JSON lists of motion files to keep / drop (see :func:_load_motion_list_file). TYPE: str | None
exclude_list_file	YAML/JSON lists of motion files to keep / drop (see :func:_load_motion_list_file). TYPE: str | None
recalculate_velocities	Re-estimate velocities by finite differences of the (retimed) positions instead of carrying the source velocities through. Defaults to True. TYPE: bool
strict_cache_identity	Refuse a cache that predates one of _cache_identity_fields (its value is unknown); False accepts it with a warning. A differing value always raises. Defaults to True. TYPE: bool
body_model_params	Also load the parametric body-model pose and shape (MANO / SMPL family) of the sequences that have one, as body_pose_quat / body_shape / body_model. Off by default: nothing extra is read, stored or cached, and no simulation-side buffer ever holds them. Only datasets that declare support take it (_supports_body_model_params); a cache prepared without it is refused rather than loaded short. TYPE: bool
body_model_dir	Directory of the body-model files (e.g. MANO_RIGHT.pkl / MANO_LEFT.pkl), for datasets that need the model to read a release (expanding a PCA pose, locating the root joint). The files are licence-gated and never shipped. TYPE: str | None

Class Attributes

file_extension: str The extension of the motion files. motion_sequence_keys: tuple[str, ...] The distinct names of the actors present in the dataset. motion_dofs_name: dict[str, list[str]] Per actor (motion_sequence_keys entry), the names of the stored dofs_pos columns, in order. motion_links_name: dict[str, list[str]] Per actor, the names of the stored link rows, in order. _hf_allow_patterns: tuple[str, ...] | None Glob patterns limiting what the automatic download fetches from the HF repo. None fetches the whole repo. Set this when the repo holds splits this loader never reads. _hf_data_subdir: str | None Directory inside the downloaded (and, if applicable, unpacked) repo that holds this loader's motion files, as a relative /-separated path. None reads the repo root.

Notes

The automatic download runs three independent stages, each opt-in: _hf_allow_patterns decides what is fetched, _post_download_process unpacks archives in place and returns the root, and _hf_data_subdir narrows that root to the directory this loader reads. Only the download path goes through them — an explicit raw_data_path is used as given, so it must already point at the equivalent of the narrowed directory.

The two layout attributes are private class attributes, not options: a repo's layout belongs to the loader that reads it, so a retarget for another robot or split is a subclass declaring its own. Both are checked when the class is defined.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str | None = None

dataset class-attribute instance-attribute

dataset: str | None = None

raw_data_path class-attribute instance-attribute

raw_data_path: str | None = None

processed_data_path class-attribute instance-attribute

processed_data_path: str | None = None

target_framerate class-attribute instance-attribute

target_framerate: float = 50.0

drop_first_n_frames class-attribute instance-attribute

drop_first_n_frames: int = 0

drop_last_n_frames class-attribute instance-attribute

drop_last_n_frames: int = 0

modifiers class-attribute instance-attribute

modifiers: tuple[DatasetModifierOptions, ...] | list[DatasetModifierOptions] = ()

include_pattern class-attribute instance-attribute

include_pattern: str | None = None

exclude_pattern class-attribute instance-attribute

exclude_pattern: str | None = None

include_list_file class-attribute instance-attribute

include_list_file: str | None = None

exclude_list_file class-attribute instance-attribute

exclude_list_file: str | None = None

recalculate_velocities class-attribute instance-attribute

recalculate_velocities: bool = True

strict_cache_identity class-attribute instance-attribute

strict_cache_identity: bool = True

body_model_params class-attribute instance-attribute

body_model_params: bool = False

body_model_dir class-attribute instance-attribute

body_model_dir: str | None = None

file_extension class-attribute

file_extension: str = '.npz'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] | None = None

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] | None = None

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] | None = None

motion_sequences instance-attribute

motion_sequences: dict[str, BatchMotionSequence]

all_motion_files instance-attribute

all_motion_files: list[str]

device property

device: device

buffer_length property

buffer_length: Tensor

framerate property

framerate: Tensor

Methods:

prepare abstractmethod

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/base.py

@abstractmethod
def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    pass

get_motion_sequence

get_motion_sequence(key: str) -> BatchMotionSequence

Source code in eden/datasets/base.py

def get_motion_sequence(self, key: str) -> BatchMotionSequence:
    return self.motion_sequences[key]

select_trajectories

select_trajectories(trajectories_idx: Tensor | list[int] | slice) -> None

Select a subset of trajectories from the loaded dataset, discarding the rest.

PARAMETER	DESCRIPTION
trajectories_idx	The trajectory indices to keep. TYPE: Tensor | list[int] | slice

Source code in eden/datasets/base.py

def select_trajectories(self, trajectories_idx: torch.Tensor | list[int] | slice) -> None:
    """Select a subset of trajectories from the loaded dataset, discarding the rest.

    Parameters
    ----------
    trajectories_idx : torch.Tensor | list[int] | slice
        The trajectory indices to keep.
    """
    for key in self.motion_sequence_keys:
        self.motion_sequences[key] = self.motion_sequences[key].select(trajectories_idx)

    if isinstance(trajectories_idx, torch.Tensor):
        trajectories_idx = trajectories_idx.tolist()
    elif isinstance(trajectories_idx, slice):
        trajectories_idx = list(range(len(self.all_motion_files)))[trajectories_idx]

    self.metadata["all_motion_files"] = [self.metadata["all_motion_files"][i] for i in trajectories_idx]
    self.all_motion_files = [self.all_motion_files[i] for i in trajectories_idx]
    for key in self._PER_CLIP_METADATA_KEYS:
        value = self.metadata.get(key)
        if value is None:
            continue
        self.metadata[key] = (
            value[trajectories_idx] if isinstance(value, np.ndarray) else [value[i] for i in trajectories_idx]
        )

motion_labels

motion_labels(key: str) -> list[str]

The per-clip label list self.metadata[key], one entry per loaded trajectory.

PARAMETER	DESCRIPTION
key	A per-clip metadata key the dataset declares in _PER_CLIP_METADATA_KEYS (e.g. "objects"). TYPE: str

Source code in eden/datasets/base.py

def motion_labels(self, key: str) -> list[str]:
    """The per-clip label list ``self.metadata[key]``, one entry per loaded trajectory.

    Parameters
    ----------
    key : str
        A per-clip metadata key the dataset declares in ``_PER_CLIP_METADATA_KEYS`` (e.g. ``"objects"``).
    """
    if key not in self._PER_CLIP_METADATA_KEYS or key not in self.metadata:
        raise KeyError(
            f"{type(self).__name__} has no per-clip metadata {key!r}; its per-clip keys are "
            f"{list(self._PER_CLIP_METADATA_KEYS)} (metadata present: {sorted(self.metadata)})"
        )
    labels = [str(v) for v in self.metadata[key]]
    if len(labels) != len(self):
        raise ValueError(
            f"{type(self).__name__}.metadata[{key!r}] has {len(labels)} entries but the dataset holds {len(self)} "
            f"trajectories; per-clip metadata must stay aligned with the trajectories."
        )
    return labels

select_dofs

select_dofs(dofs_idx_dict: dict[str, slice | list[int] | Tensor]) -> None

Select a subset of dofs from the loaded dataset, discarding the rest.

motion_dofs_name is selected with the same indices, so a name keeps labelling its own column after a subset or a reorder.

PARAMETER	DESCRIPTION
dofs_idx_dict	The dofs indices to keep, in the order to keep them. TYPE: dict[str, slice | list[int] | Tensor]

Source code in eden/datasets/base.py

def select_dofs(self, dofs_idx_dict: dict[str, slice | list[int] | torch.Tensor]) -> None:
    """Select a subset of dofs from the loaded dataset, discarding the rest.

    ``motion_dofs_name`` is selected with the same indices, so a name keeps labelling its own
    column after a subset or a reorder.

    Parameters
    ----------
    dofs_idx_dict : dict[str, slice | list[int] | torch.Tensor]
        The dofs indices to keep, in the order to keep them.
    """
    for key, dofs_idx in dofs_idx_dict.items():
        self.motion_sequences[key] = self.motion_sequences[key].select_dofs(dofs_idx)
    self.motion_dofs_name = _select_names(self.motion_dofs_name, dofs_idx_dict)

select_links

select_links(links_idx_dict: dict[str, slice | list[int] | Tensor]) -> None

Select a subset of links from the loaded dataset, discarding the rest.

motion_links_name is selected with the same indices, so a name keeps labelling its own row after a subset or a reorder.

PARAMETER	DESCRIPTION
links_idx_dict	The links indices to keep, in the order to keep them. TYPE: dict[str, slice | list[int] | Tensor]

Source code in eden/datasets/base.py

def select_links(self, links_idx_dict: dict[str, slice | list[int] | torch.Tensor]) -> None:
    """Select a subset of links from the loaded dataset, discarding the rest.

    ``motion_links_name`` is selected with the same indices, so a name keeps labelling its own
    row after a subset or a reorder.

    Parameters
    ----------
    links_idx_dict : dict[str, slice | list[int] | torch.Tensor]
        The links indices to keep, in the order to keep them.
    """
    for key, links_idx in links_idx_dict.items():
        self.motion_sequences[key] = self.motion_sequences[key].select_links(links_idx)
    self.motion_links_name = _select_names(self.motion_links_name, links_idx_dict)

compute_dofs_pos_range

compute_dofs_pos_range(entity_name: str, *, margin: float = 0.15, margin_overrides: dict[str, float] | None = None) -> dict[str, tuple[float, float]]

Per-DOF (min - m, max + m) over every loaded demo frame.

Mirrors dexmachina's demo_data.get_joint_init_limits (the limits half). Output keys come from self.motion_dofs_name[entity_name]; values are plain Python floats so the dict can be passed to EntityOptions.soft_dofs_pos_limits without binding to a device at config time.

Configuration-time only: forces a GPU→CPU sync via .cpu().tolist(), so callers should resolve the dict once during config build (not per step).

PARAMETER	DESCRIPTION
entity_name	Key into self.motion_sequences. TYPE: str
margin	Default pad applied to both ends. Dexmachina's default is 0.15. TYPE: float DEFAULT: 0.15
margin_overrides	Substring → margin. If a DOF name contains a key, its value replaces margin for that DOF. When more than one substring matches a single DOF the last one wins, and a config-time warning is logged. Pass {"tx": 0.2, "ty": 0.2, "tz": 0.2, "roll": 0.5, "pitch": 0.5, "yaw": 0.5} to mirror dexmachina exactly. TYPE: dict[str, float] | None DEFAULT: None

RAISES	DESCRIPTION
KeyError if ``entity_name`` is not a registered motion sequence key.
ValueError if the entity has no ``dofs_pos`` data, zero DOFs, or
``motion_dofs_name`` desynchronized from ``dofs_pos`` (a name list
that does not name every column).

Source code in eden/datasets/base.py

def compute_dofs_pos_range(
    self,
    entity_name: str,
    *,
    margin: float = 0.15,
    margin_overrides: dict[str, float] | None = None,
) -> dict[str, tuple[float, float]]:
    """Per-DOF ``(min - m, max + m)`` over every loaded demo frame.

    Mirrors dexmachina's ``demo_data.get_joint_init_limits`` (the ``limits`` half).
    Output keys come from ``self.motion_dofs_name[entity_name]``; values are
    plain Python floats so the dict can be passed to
    ``EntityOptions.soft_dofs_pos_limits`` without binding to a device at
    config time.

    Configuration-time only: forces a GPU→CPU sync via ``.cpu().tolist()``,
    so callers should resolve the dict once during config build (not per
    step).

    Parameters
    ----------
    entity_name:
        Key into ``self.motion_sequences``.
    margin:
        Default pad applied to both ends. Dexmachina's default is 0.15.
    margin_overrides:
        Substring → margin. If a DOF name *contains* a key, its value
        replaces ``margin`` for that DOF. When more than one substring
        matches a single DOF the last one wins, and a config-time warning
        is logged. Pass
        ``{"tx": 0.2, "ty": 0.2, "tz": 0.2, "roll": 0.5, "pitch": 0.5, "yaw": 0.5}``
        to mirror dexmachina exactly.

    Raises
    ------
    KeyError if ``entity_name`` is not a registered motion sequence key.
    ValueError if the entity has no ``dofs_pos`` data, zero DOFs, or
    ``motion_dofs_name`` desynchronized from ``dofs_pos`` (a name list
    that does not name every column).
    """
    bms, names = self._validate_dofs_motion(entity_name)
    flat = bms.dofs_pos.concatenated_tensor  # (total_frames, num_dofs)
    lo = flat.amin(dim=0).cpu().tolist()
    hi = flat.amax(dim=0).cpu().tolist()
    margins = self._resolve_dof_margins(names, margin, margin_overrides)
    return {n: (lo_i - m, hi_i + m) for n, lo_i, hi_i, m in zip(names, lo, hi, margins, strict=True)}

compute_dofs_pos_at

compute_dofs_pos_at(entity_name: str, *, frame: int = 0, sequence_idx: int = 0) -> dict[str, float]

Per-DOF qpos at frame of demo sequence sequence_idx.

Companion to :meth:compute_dofs_pos_range for the per-clip default_dofs_pos override (dexmachina set_custom_init_qpos). Defaults to the first frame of the first loaded sequence — the dexmachina convention is one clip per task, so this is the common case. When the dataset holds multiple clips, pass sequence_idx to pick the one this task config targets; TaskMod configs select a clip explicitly.

Configuration-time only: forces a GPU→CPU sync via .cpu().tolist(), so callers should resolve the dict once during config build (not per step).

Returns {dof_name: q} as plain Python floats.

RAISES	DESCRIPTION
KeyError if ``entity_name`` is not a registered motion sequence key.
ValueError on any of: no ``dofs_pos`` data, ``motion_dofs_name``
desynchronized from ``dofs_pos`` (a name list that does not name every column),
or ``frame`` / ``sequence_idx`` negative or out of range. Dataset-level
desync is checked **before** ``frame`` / ``sequence_idx`` bounds so the
deeper config error surfaces first.

Source code in eden/datasets/base.py

def compute_dofs_pos_at(self, entity_name: str, *, frame: int = 0, sequence_idx: int = 0) -> dict[str, float]:
    """Per-DOF qpos at ``frame`` of demo sequence ``sequence_idx``.

    Companion to :meth:`compute_dofs_pos_range` for the per-clip
    ``default_dofs_pos`` override (dexmachina ``set_custom_init_qpos``).
    Defaults to the first frame of the first loaded sequence — the
    dexmachina convention is one clip per task, so this is the common
    case. When the dataset holds multiple clips, pass ``sequence_idx`` to
    pick the one this task config targets; ``TaskMod`` configs select a clip
    explicitly.

    Configuration-time only: forces a GPU→CPU sync via ``.cpu().tolist()``,
    so callers should resolve the dict once during config build (not per
    step).

    Returns ``{dof_name: q}`` as plain Python floats.

    Raises
    ------
    KeyError if ``entity_name`` is not a registered motion sequence key.
    ValueError on any of: no ``dofs_pos`` data, ``motion_dofs_name``
    desynchronized from ``dofs_pos`` (a name list that does not name every column),
    or ``frame`` / ``sequence_idx`` negative or out of range. Dataset-level
    desync is checked **before** ``frame`` / ``sequence_idx`` bounds so the
    deeper config error surfaces first.
    """
    if frame < 0:
        raise ValueError(f"frame must be >= 0, got {frame}")
    if sequence_idx < 0:
        raise ValueError(f"sequence_idx must be >= 0, got {sequence_idx}")
    bms, names = self._validate_dofs_motion(entity_name)
    num_seqs = int(bms.buffer_length.numel())
    if sequence_idx >= num_seqs:
        raise ValueError(
            f"sequence_idx {sequence_idx} out of range; entity '{entity_name}' has {num_seqs} sequence(s)."
        )
    seq_len = int(bms.buffer_length[sequence_idx].item())
    if frame >= seq_len:
        raise ValueError(
            f"frame {frame} is out of range for sequence {sequence_idx} of "
            f"entity '{entity_name}' (buffer_length={seq_len})."
        )
    # (num_dofs,) at the chosen (sequence, frame); mask is scalar True after the bounds check.
    qpos_t, _ = bms.dofs_pos.get_with_mask((sequence_idx, frame))
    qpos = qpos_t.cpu().tolist()
    return dict(zip(names, qpos, strict=True))

save_cached

save_cached(data_path: str | None = None)

Save the processed dataset to data_path in the current cache format (_CACHE_FORMAT_VERSION).

The on-disk layout is a single torch.save call over a plain dict with a _meta envelope (matching the project's config-serialisation convention in eden/utils/configs.py)::

{
    "_meta": {
        "_format_version": 4,
        "eden": {"version": ..., "commit": ...}, "genesis": {...},
        "target_framerate": ..., "motion_sequence_keys": [...],
        "motion_dofs_name": ..., "motion_links_name": ...,
        "cache_identity": {...}, "frame_window": ..., "cache_semantics": ...,
        "all_motion_files": [...], ...,  # plus the dataset's own metadata keys
    },
    "sequences": {key: state_dict, ...},
}

Each per-key state_dict is produced by BatchMotionSequence.to_state_dict() and contains only tensors and primitive Python types — no custom objects — so the file is torch.load(mmap=True)-loadable. Multiple DDP ranks on the same node share the OS page cache instead of each materialising a private copy of the motion library.

Source code in eden/datasets/base.py

def save_cached(self, data_path: str | None = None):
    """Save the processed dataset to ``data_path`` in the current cache format (``_CACHE_FORMAT_VERSION``).

    The on-disk layout is a single ``torch.save`` call over a plain dict
    with a ``_meta`` envelope (matching the project's config-serialisation
    convention in ``eden/utils/configs.py``)::

        {
            "_meta": {
                "_format_version": 4,
                "eden": {"version": ..., "commit": ...}, "genesis": {...},
                "target_framerate": ..., "motion_sequence_keys": [...],
                "motion_dofs_name": ..., "motion_links_name": ...,
                "cache_identity": {...}, "frame_window": ..., "cache_semantics": ...,
                "all_motion_files": [...], ...,  # plus the dataset's own metadata keys
            },
            "sequences": {key: state_dict, ...},
        }

    Each per-key ``state_dict`` is produced by
    ``BatchMotionSequence.to_state_dict()`` and contains only tensors and
    primitive Python types — no custom objects — so the file is
    ``torch.load(mmap=True)``-loadable. Multiple DDP ranks on the same
    node share the OS page cache instead of each materialising a private
    copy of the motion library.
    """
    data_path = data_path or self.processed_data_path
    if data_path is None:
        raise ValueError("No data path provided")

    data_dir = os.path.dirname(data_path)
    if data_dir:
        os.makedirs(data_dir, exist_ok=True)

    # Spread self.metadata first so subclass-supplied keys (e.g. Arctic's
    # "objects", OMOMO's "human_height") survive, then layer the canonical
    # envelope keys on top so a load→modify→save round-trip can't carry
    # over stale version markers.
    meta = {
        **self.metadata,
        "_format_version": _CACHE_FORMAT_VERSION,
        "eden": get_eden_provenance(),
        "genesis": get_package_provenance("genesis-world"),
        "target_framerate": self.target_framerate,
        "motion_sequence_keys": list(self.motion_sequence_keys),
        "motion_dofs_name": self.motion_dofs_name,
        "motion_links_name": self.motion_links_name,
        "cache_identity": cache_identity(self),
        "frame_window": _FRAME_WINDOW_SEMANTICS,
        "cache_semantics": type(self)._cache_semantics,
    }
    sequences_sd = {
        # Move tensors to CPU before saving so the file is portable across
        # different GPU layouts (DDP rank 0 may have written from cuda:0
        # but a future job loading the cache may run on a different GPU).
        key: {k: (v.cpu() if isinstance(v, torch.Tensor) else v) for k, v in seq.to_state_dict().items()}
        for key, seq in self.motion_sequences.items()
    }

    payload = {"_meta": meta, "sequences": sequences_sd}
    en.logger.info(f"Saving processed data to {data_path}")
    # Write-then-rename, never straight to data_path. ``load_cached`` mmaps the cache, so
    # ``sequences_sd`` can be backed by the very file being written: saving in place
    # truncates it out from under its own payload and leaves 0 bytes, which every later run
    # then fails to open with torch's "mmap can only be used with files saved with
    # _use_new_zipfile_serialization" — a message about the wrong thing entirely. Reachable
    # by just re-running ``eden preprocess`` once a cache exists. The rename is atomic
    # within the directory, so a crash or a second writer mid-save also cannot leave a
    # partial cache behind.
    tmp_path = f"{data_path}.{os.getpid()}.tmp"
    try:
        torch.save(payload, tmp_path)
        os.replace(tmp_path, data_path)
    finally:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)

load_cached

load_cached(data_path: str) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Load the processed dataset from a cache file in the current format (_CACHE_FORMAT_VERSION).

Uses torch.load(mmap=True) so tensors are page-cached at the OS level instead of materialised into per-process buffers — repeated loads (e.g. across DDP ranks on the same node) avoid re-reading the file from disk and avoid duplicating it in RAM.

Source code in eden/datasets/base.py

def load_cached(self, data_path: str) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    """Load the processed dataset from a cache file in the current format (``_CACHE_FORMAT_VERSION``).

    Uses ``torch.load(mmap=True)`` so tensors are page-cached at the OS
    level instead of materialised into per-process buffers — repeated
    loads (e.g. across DDP ranks on the same node) avoid re-reading the
    file from disk and avoid duplicating it in RAM.
    """
    if not os.path.exists(data_path):
        raise FileNotFoundError(f"Processed data file {data_path} not found")

    # weights_only=False is required because the _meta envelope carries
    # str/list/dict primitives alongside tensors. Caches are produced by
    # trusted Eden code on the machine that consumes them; weights_only=True
    # would require moving _meta to a sidecar.
    payload = torch.load(data_path, mmap=True, weights_only=False)

    meta = payload.get("_meta") if isinstance(payload, dict) else None
    if meta is None or meta.get("_format_version") != _CACHE_FORMAT_VERSION:
        raise RuntimeError(
            f"Cache at {data_path} is in an unsupported format "
            f"(expected v{_CACHE_FORMAT_VERSION}). Delete it to rebuild — "
            f"it is not an Eden cache, or an Eden with a different cache layout wrote it."
        )

    sequences_sd = payload["sequences"]

    if not isinstance(meta.get("eden"), dict):
        en.logger.warning(
            f"Cache at {data_path} records no Eden provenance (it predates the record), so it cannot be checked "
            "against this Eden. If it may be stale, delete it to rebuild."
        )
    for key, name, current in (
        ("eden", "Eden", get_eden_provenance()),
        ("genesis", "Genesis", get_package_provenance("genesis-world")),
    ):
        if change := describe_provenance_change(meta.get(key), current):
            en.logger.warning(
                f"{name} differs from the one that wrote {data_path} ({change}): the loaded data may be "
                f"incompatible with this {name}."
            )
    if meta["target_framerate"] != self.target_framerate:
        raise ValueError(f"Target framerate mismatch: {meta['target_framerate']} != {self.target_framerate}")
    if (semantics := type(self)._cache_semantics) is not None and meta.get("cache_semantics") != semantics:
        raise ValueError(
            f"Cache at {data_path} was prepared by an earlier {type(self).__name__} "
            f"(cache_semantics {meta.get('cache_semantics')!r}); this one prepares {semantics!r}, so the cached "
            f"motion means something else. Delete it to rebuild, or point processed_data_path at a different "
            f"file. strict_cache_identity=False does not relax this: no setting makes the old motion right."
        )
    # Settings that shaped the cached bytes. A cache written before the key existed carries
    # none and loads untouched (the Eden warnings above, a mismatch or no record, are the signal for an old
    # cache); one that carries the key is read strictly, because what this catches is
    # silent — a library prepared under other settings loads and trains perfectly well.
    if (cached_identity := meta.get("cache_identity")) is not None:
        identity = cache_identity(self)
        differing = {
            name: (cached_identity[name], configured)
            for name, configured in identity.items()
            if name in cached_identity and cached_identity[name] != configured
        }
        untracked = [name for name in identity if name not in cached_identity]
        # Which unknowns are fatal, resolved before anything is logged or raised — else the
        # "accepting it" warning below can land above an exception rejecting the cache
        # anyway, since a differing value is fatal whatever the flag says.
        strict_failure = untracked if self.strict_cache_identity else []
        if differing or strict_failure:
            reasons = [f"{name}: cached {was!r} != configured {now!r}" for name, (was, now) in differing.items()]
            if strict_failure:
                reasons.append(f"written before {', '.join(strict_failure)} was tracked, so its value is unknown")
            # Offered only when it would actually get the caller past this, i.e. every
            # reason is an unknown — naming it beside a differing value would read as "this
            # is how you proceed" for a conflict no flag relaxes.
            remedy = (
                " Or, if you know those settings have not changed since the cache was built, "
                "accept it with a warning by setting strict_cache_identity=False on the "
                "dataset options — from the CLI that is a nested --override, e.g. "
                "--override 'motion_options: {default: {dataset_options: "
                "{strict_cache_identity: false}}}' for a motion-tracking task (dotted keys "
                "are not supported)."
                if strict_failure and not differing
                else ""
            )
            raise ValueError(
                f"Cache at {data_path} does not match the current settings "
                f"({'; '.join(reasons)}). Delete it to rebuild, or point processed_data_path "
                f"at a different file — these settings change the prepared motion, and "
                f"reusing the cache would train on data the current config did not ask for."
                f"{remedy}"
            )
        # Reached only when the load goes ahead, so non-empty here implies the opt-out.
        if untracked:
            en.logger.warning(
                f"Cache at {data_path} was written before {', '.join(untracked)} was tracked, "
                f"so it cannot be checked against the current config "
                f"({', '.join(f'{name}={identity[name]!r}' for name in untracked)}). Accepting "
                f"it because strict_cache_identity=False — if any of those settings has "
                f"changed since the cache was built, this run trains on the older motion."
            )
    cached_window = meta.get("frame_window")
    trusted = cached_window is None and all(
        name in self._drops_cut_before_marker
        for name in ("drop_first_n_frames", "drop_last_n_frames")
        if getattr(self, name)
    )
    if self._trims_clips and cached_window != _FRAME_WINDOW_SEMANTICS and not trusted:
        raise ValueError(
            f"Cache at {data_path} was prepared with drop_first_n_frames={self.drop_first_n_frames} / "
            f"drop_last_n_frames={self.drop_last_n_frames} before every reader cut clips the same way (OMOMO and "
            f"Weave ignored both; AMASS, PHUMA and HHI wrapped a drop_last_n_frames longer than a clip around to "
            f"its start), so its clips may not be trimmed as configured. Delete it to rebuild, or point "
            f"processed_data_path at a different file."
        )
    # Saved as list (JSON-friendly), compared against tuple class attribute.
    if tuple(meta["motion_sequence_keys"]) != tuple(self.motion_sequence_keys):
        raise ValueError(
            f"Motion sequence keys mismatch: {meta['motion_sequence_keys']} != {self.motion_sequence_keys}"
        )

    sequences = {key: BatchMotionSequence.from_state_dict(sd) for key, sd in sequences_sd.items()}
    return sequences, meta

Functions:

cache_identity

cache_identity(dataset: DatasetBase) -> dict[str, Any]

The prepare-shaping settings a cache was built with, for the _meta envelope.

Public because a dataset that derives its own default processed_data_path has to key that path by the same settings :meth:DatasetBase.load_cached hard-checks against it. Reading both off this one function is what stops the two drifting: a field added to _cache_identity_fields then starts keying a new cache file in the same commit that starts rejecting the old one, rather than turning two valid configs into one contested path that each rejects for the other.

PARAMETER	DESCRIPTION
dataset	Dataset to read the values from. Only ever reads attributes that have class-level defaults, so an uninitialized instance is a valid argument. TYPE: DatasetBase

RETURNS	DESCRIPTION
dict[str, Any]	Field name to :func:~eden.utils.misc.to_jsonish value, in MRO order.

Source code in eden/datasets/base.py

def cache_identity(dataset: DatasetBase) -> dict[str, Any]:
    """The prepare-shaping settings a cache was built with, for the ``_meta`` envelope.

    Public because a dataset that *derives* its own default ``processed_data_path`` has to key
    that path by the same settings :meth:`DatasetBase.load_cached` hard-checks against it.
    Reading both off this one function is what stops the two drifting: a field added to
    ``_cache_identity_fields`` then starts keying a new cache file in the same commit that
    starts rejecting the old one, rather than turning two valid configs into one contested
    path that each rejects for the other.

    Parameters
    ----------
    dataset : DatasetBase
        Dataset to read the values from. Only ever reads attributes that have class-level
        defaults, so an uninitialized instance is a valid argument.

    Returns
    -------
    dict[str, Any]
        Field name to :func:`~eden.utils.misc.to_jsonish` value, in MRO order.
    """
    return {name: to_jsonish(getattr(dataset, name)) for name in _cache_identity_field_names(type(dataset))}

Options

Dataset and dataset-modifier configuration options.

Classes

DatasetModifierOptions

DatasetModifierOptions(**data)

Bases: DynamicConfigurableOptions

Options for dataset modifiers.

PARAMETER	DESCRIPTION
motion_sequence_keys	Keys in DatasetBase.motion_sequences this modifier applies to. When None the modifier is applied to every key.

Source code in eden/options/options.py

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

Attributes

motion_sequence_keys class-attribute instance-attribute

motion_sequence_keys: tuple[str, ...] | None = None

DatasetOptions

DatasetOptions(**data)

Bases: DynamicConfigurableOptions

Dataset options.

PARAMETER	DESCRIPTION
registry	HuggingFace namespace (user or organisation) that hosts the dataset repo. Subclasses should set this to enable automatic downloads, e.g. registry = "ember-lab-berkeley".
dataset	HuggingFace dataset repository name, e.g. "AMASS_Retargeted_for_G1".
raw_data_path	Path to the raw data directory. When provided it is used as-is and must already exist — a missing explicit path raises rather than downloading. When None and the dataset class defines registry / dataset, the dataset is downloaded to the HuggingFace global cache.
processed_data_path	Path to the processed (cached) data file.
target_framerate	The target framerate of the motion.
drop_first_n_frames	Frames dropped from the start of each raw clip during prepare(). Must be >= 0. A dataset that cannot trim its clips at read time refuses a non-zero value; trim with the TrimFrames modifier.
drop_last_n_frames	Frames dropped from the end of each raw clip during prepare(). Must be >= 0; see drop_first_n_frames.
modifiers	Ordered sequence of offline transforms applied to the loaded motion sequences after cache load or raw prepare. Modifiers are not baked into the processed cache — changing the config re-applies them on next load.
include_pattern	Regex applied to each motion file's basename via re.search. When set, only files matching are kept. None keeps every file. Applied before prepare() so unmatched files are never read or interpolated. Plain substrings work as regexes ("walk" keeps any file whose basename contains "walk").
exclude_pattern	Regex applied to each motion file's basename via re.search. Files matching are dropped after include_pattern filtering. None disables exclusion.
include_list_file	Path to a YAML or JSON file enumerating motion files to keep. The file may be (a) a flat list of paths/basenames, (b) a list of dicts each with a file: key (ProtoMotions amass_*_train.yaml shape), or (c) a dict wrapping either form under a motions: key. Extensions are stripped. An entry with a directory component matches only a file with the same trailing path (CMU/45/45_01_stageii.motion matches .../CMU/45/45_01_stageii.npz); a bare basename matches any file of that name. Applied alongside include_pattern.
exclude_list_file	Path to a YAML or JSON file enumerating motion files to drop. Same accepted shapes and matching rules as include_list_file. Applied alongside exclude_pattern.

Source code in eden/options/options.py

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

Attributes

registry class-attribute instance-attribute

registry: str | None = None

dataset class-attribute instance-attribute

dataset: str | None = None

raw_data_path class-attribute instance-attribute

raw_data_path: str | None = None

processed_data_path class-attribute instance-attribute

processed_data_path: str | None = None

target_framerate class-attribute instance-attribute

target_framerate: float = 50.0

drop_first_n_frames class-attribute instance-attribute

drop_first_n_frames: NonNegativeInt = 0

drop_last_n_frames class-attribute instance-attribute

drop_last_n_frames: NonNegativeInt = 0

modifiers class-attribute instance-attribute

modifiers: tuple[DatasetModifierOptions, ...] | list[DatasetModifierOptions] = ()

include_pattern class-attribute instance-attribute

include_pattern: str | None = None

exclude_pattern class-attribute instance-attribute

exclude_pattern: str | None = None

include_list_file class-attribute instance-attribute

include_list_file: str | None = None

exclude_list_file class-attribute instance-attribute

exclude_list_file: str | None = None

Humanoid motion

AMASSRetargetedG1Dataset

AMASSRetargetedG1Dataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the AMASS retargeted G1 dataset.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'ember-lab-berkeley'

dataset class-attribute instance-attribute

dataset: str = 'AMASS_Retargeted_for_G1'

file_extension class-attribute

file_extension: str = '.npz'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('g1',)

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'g1': G1_DOFS29_NAME}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'g1': G1_DOFS29_TRACKED_LINKS_NAME}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/amass_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    # No forward kinematics: the AMASS retarget already holds body positions and rotations (IsaacLab AMP
    # format). Velocities and base-frame transforms are computed by _pack_motion_sequence.
    return self._prepare_single_key(self._read_single_sequence, desc="AMASS: reading raw .npz")

LAFAN1RetargetedG1Dataset

LAFAN1RetargetedG1Dataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the LAFAN1 retargeted G1 dataset.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'lvhaidong'

dataset class-attribute instance-attribute

dataset: str = 'LAFAN1_Retargeting_Dataset'

file_extension class-attribute

file_extension: str = '.csv'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('g1',)

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'g1': G1_DOFS29_NAME}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'g1': G1_DOFS29_TRACKED_LINKS_NAME}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Prepare the LAFAN1 retargeting dataset; the CSVs carry base + DOFs only, so links come from FK.

Source code in eden/datasets/lafan1_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    """Prepare the LAFAN1 retargeting dataset; the CSVs carry base + DOFs only, so links come from FK."""
    return self._prepare_single_key(
        self._read_single_sequence,
        desc="LAFAN1: reading raw csv",
        robot=en.robots.G1Dofs29(dofs_name=self.motion_dofs_name["g1"]),
    )

SEEDRetargetedG1Dataset

SEEDRetargetedG1Dataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the SEED motion library retargeted to the Unitree G1.

https://huggingface.co/datasets/bones-studio/seed

SEED ships three splits (SOMA Uniform BVH, SOMA Proportional BVH, and the G1 MuJoCo-compatible CSV split). This loader only consumes the G1 CSV split, distributed inside g1.tar.gz (~23 GB).

Notes

The HuggingFace repo is gated; users must request access on HuggingFace and have a token configured (huggingface-cli login) before the automatic download will succeed.

CSV layout (one file per motion, header row + one row per frame at 120 fps): Frame, root_translate{X,Y,Z}, root_rotate{X,Y,Z}, <29 DOFs>

where translation is in centimetres and the DOF columns are joint angles in degrees in the same order as motion_dofs_name['g1'] below. The root rotation columns are XYZ Euler magnitudes in degrees, but the rotation is composed in intrinsic ZYX order (R = Rz · Ry · Rx) — the Maya export convention. Reading them as intrinsic XYZ produces a visually plausible standing pose at frame 0 but lifts one foot off the ground mid-clip, while ZYX keeps the feet planted throughout.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'bones-studio'

dataset class-attribute instance-attribute

dataset: str = 'seed'

file_extension class-attribute

file_extension: str = '.csv'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('g1',)

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'g1': G1_DOFS29_NAME}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'g1': G1_DOFS29_TRACKED_LINKS_NAME}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Prepare the SEED retargeted G1 dataset; the CSVs carry base + DOFs only, so links come from FK.

Source code in eden/datasets/seed_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    """Prepare the SEED retargeted G1 dataset; the CSVs carry base + DOFs only, so links come from FK."""
    return self._prepare_single_key(
        self._read_single_sequence,
        desc="SEED: reading raw csv",
        robot=en.robots.G1Dofs29(dofs_name=self.motion_dofs_name["g1"]),
    )

PHUMADataset

PHUMADataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the PHUMA dataset.

https://huggingface.co/datasets/DAVIAN-Robotics/PHUMA. Unset, raw_data_path downloads its data.zip into the Hugging Face cache and unzips it there on first use; set it to an already unpacked copy instead.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'DAVIAN-Robotics'

dataset class-attribute instance-attribute

dataset: str = 'PHUMA'

file_extension class-attribute

file_extension: str = '.npy'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('g1',)

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'g1': G1_DOFS29_NAME}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'g1': G1_DOFS29_TRACKED_LINKS_NAME}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Prepare the PHUMA dataset.

Reads the ~73k per-motion .npy files in parallel on CPU, packs them with a single torch.cat per field, then moves the batched sequence to self.device for the Genesis-based forward-velocity FK pass.

Source code in eden/datasets/phuma_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    """
    Prepare the PHUMA dataset.

    Reads the ~73k per-motion .npy files in parallel on CPU, packs them
    with a single torch.cat per field, then moves the batched sequence to
    ``self.device`` for the Genesis-based forward-velocity FK pass.
    """
    return self._prepare_single_key(
        self._read_single_sequence,
        desc="PHUMA: reading raw .npy",
        robot=en.robots.G1Dofs29(dofs_name=self.motion_dofs_name["g1"]),
    )

Human-object and human-human interaction

OMOMOHolosomaDataset

OMOMOHolosomaDataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the OMOMO data processed by Holosoma pipeline.

Each sequence is stored as an individual .pt file containing an InterMimic-format tensor of shape (T, 591). The tensor encodes:

• offset 0-2: root position (3)
• offset 3-6: root quaternion in wxyz (4)
• offset 162-317: 52 SMPL-H joint world positions (52 x 3)
• offset 318-324: object pose [x, y, z, qx, qy, qz, qw] (7)

Unlike the original OMOMO pickle format, no FK is required — joint positions are pre-computed. The 52 joints follow the :data:SMPLH_KEYPOINTS_NAME order.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'Kashu7100'

dataset class-attribute instance-attribute

dataset: str = 'eden_omomo_holosoma'

file_extension class-attribute

file_extension: str = '.pt'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('human', 'object')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'human': [], 'object': []}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'human': list(SMPLH_KEYPOINTS_NAME), 'object': []}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/omomo_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    en.logger.info(f"Loading {len(self.all_motion_files)} sequences from {self.raw_data_path}")

    # Parallel CPU read; bulk H2D moves once everything is packed.
    pairs = self._parallel_read_motion_files(
        self.all_motion_files,
        self._read_single_sequence,
        desc="OMOMO: reading raw .pt",
    )
    human_sequences: list[MotionSequence] = [h for h, _ in pairs]
    object_sequences: list[MotionSequence] = [o for _, o in pairs]

    # Object name parsing is cheap and order-sensitive, do it sequentially
    # against the (already-ordered) file list.
    object_names: list[str] = []
    for file_path in self.all_motion_files:
        # filename format: sub{id}_{object_name}_{clip_index}.pt
        seq_name = os.path.splitext(os.path.basename(file_path))[0]
        obj_name = "_".join(seq_name.split("_")[1:-1])
        assert obj_name in OMOMO_OBJECTS_REGISTRY, f"Object name '{obj_name}' not found in OMOMO_OBJECTS_REGISTRY"
        object_names.append(obj_name)

    motion_data = {
        "human": BatchMotionSequence.from_motion_sequences(
            human_sequences,
            num_links=len(self.motion_links_name["human"]),
        ).to(self.device),
        "object": BatchMotionSequence.from_motion_sequences(
            object_sequences,
        ).to(self.device),
    }

    metadata: dict[str, Any] = {
        "objects": object_names,
        "human_height": {
            "sub1": np.float32(1.6637051),
            "sub2": np.float32(1.8438001),
            "sub3": np.float32(1.9411048),
            "sub4": np.float32(1.7904495),
            "sub5": np.float32(1.8171062),
            "sub6": np.float32(1.7621962),
            "sub7": np.float32(1.8078182),
            "sub8": np.float32(1.8449197),
            "sub9": np.float32(1.753231),
            "sub10": np.float32(1.6509807),
            "sub11": np.float32(1.7666422),
            "sub12": np.float32(1.8030452),
            "sub13": np.float32(1.8110343),
            "sub14": np.float32(1.8263087),
            "sub15": np.float32(1.6846002),
            "sub16": np.float32(1.7229546),
            "sub17": np.float32(1.7366357),
        },
    }
    return motion_data, metadata

WeaveG1InspireHOIDataset

WeaveG1InspireHOIDataset(options, device: device)

Bases: DatasetBase

Weave G1-Inspire-HOI reference clips: a G1 + two Inspire RH56DFQ hands tracking an OMOMO object.

PARAMETER	DESCRIPTION
split	"train" (7,869 clips, 19.6 h) or "test" (1,605 clips, 3.7 h; five approach variants per clip). TYPE: str
objects	Load only these objects' files (a subset of :data:WEAVE_OBJECTS); None loads all nine. Also narrows the download. TYPE: list[str] | None

Source code in eden/datasets/weave_dataset.py

def __init__(self, options, device: torch.device):
    split = getattr(options, "split", None) or type(self).split
    if split not in ("train", "test"):
        raise ValueError(f"WeaveG1InspireHOIDataset.split must be 'train' or 'test', got {split!r}.")
    objects = getattr(options, "objects", None)
    unknown = sorted(set(objects or ()) - set(WEAVE_OBJECTS))
    if unknown:
        raise ValueError(
            f"WeaveG1InspireHOIDataset.objects has unknown object(s) {unknown}; Weave publishes {list(WEAVE_OBJECTS)}."
        )
    if objects is not None and not objects:
        raise ValueError("WeaveG1InspireHOIDataset.objects is empty; pass None to load all nine objects.")
    # The download narrows to this split's reference files of the selected objects, plus their geometry. Those
    # are options, so the patterns are set per instance — before the base __init__ downloads — rather than as
    # the class-level layout most loaders declare; the base download (and its per-file lock) does the fetching.
    names = tuple(objects) if objects is not None else WEAVE_OBJECTS
    self._hf_allow_patterns = (
        *(f"reference/{split}/{obj}_*.npz" for obj in names),
        *(f"objects/{obj}/*" for obj in names),
        "objects/geometry/*",
    )
    super().__init__(options, device)

Attributes

registry class-attribute instance-attribute

registry: str = 'appolyn'

dataset class-attribute instance-attribute

dataset: str = 'Weave'

split class-attribute instance-attribute

split: Literal['train', 'test'] = 'test'

objects class-attribute instance-attribute

objects: list[str] | None = None

recalculate_velocities class-attribute instance-attribute

recalculate_velocities: bool = False

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('robot', 'left_hand', 'right_hand', 'object')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'robot': list(WEAVE_JOINT_NAMES[:29]), 'left_hand': _hand_dofs_name('L'), 'right_hand': _hand_dofs_name('R'), 'object': []}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'robot': list(WEAVE_ROBOT_LINK_NAMES), 'left_hand': _hand_link_names('L'), 'right_hand': _hand_link_names('R'), 'object': []}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/weave_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    wanted = set(self.all_motion_files)  # after include/exclude filters
    sequences: dict[str, list[MotionSequence]] = {key: [] for key in self.motion_sequence_keys}
    objects: list[str] = []
    motion_names: list[str] = []
    kept_files: list[str] = []
    short: list[_ShortClipError] = []
    window = self._frame_window()
    for obj, npz in self._npz_by_object.items():
        en.logger.info(f"Weave: reading {npz}")
        with np.load(npz, allow_pickle=False) as raw:
            data = {k: raw[k] for k in raw.files}
        fps = float(np.asarray(data["fps"]).reshape(-1)[0])
        starts = np.concatenate([[0], np.cumsum(data["motion_lengths"])])
        if starts[-1] != data["joint_pos"].shape[0]:
            raise ValueError(
                f"{npz}: motion_lengths sum to {starts[-1]} frames but the arrays hold {data['joint_pos'].shape[0]}."
            )
        for i, name in enumerate(data["motion_names"].tolist()):
            pseudo = f"{npz}/{name}"
            if pseudo not in wanted:
                continue
            clip = {k: data[k][starts[i] : starts[i + 1]][window] for k in _PER_FRAME_KEYS}
            token = _READING_MOTION_FILE.set(pseudo)  # names the clip in a short-clip message
            try:
                split = self._split_clip(clip, fps)
            except _ShortClipError as e:
                short.append(e)
                continue
            finally:
                _READING_MOTION_FILE.reset(token)
            for key, seq in split.items():
                sequences[key].append(seq)
            objects.append(str(data["object_names"][i]))
            motion_names.append(str(name))
            kept_files.append(pseudo)
    self._skip_short_clips(short, len(kept_files) + len(short))
    # prepare() owns the clip order; _initialize stores all_motion_files as the per-trajectory metadata.
    self.all_motion_files = kept_files

    motion_data: dict[str, BatchMotionSequence] = {}
    for key in self.motion_sequence_keys:
        batch = BatchMotionSequence.from_motion_sequences(
            sequences[key],
            num_dofs=len(self.motion_dofs_name[key]),
            num_links=len(self.motion_links_name[key]),
        )
        batch.is_free_floating = key in ("robot", "object")
        batch.entity_kind = {"robot": "humanoid", "object": "rigid_object"}.get(key, "hand")
        motion_data[key] = batch.to(self.device)
    return motion_data, {"objects": objects, "motion_names": motion_names}

HHIAssistDataset

HHIAssistDataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the HHI-Assist dataset.

The HHI-Assist dataset contains human-human interaction motion capture data recorded with an Optitrack system at 120 Hz. Each interaction consists of a caregiver (CG) and a care recipient (CR), captured as paired BVH files.

References

• https://github.com/vita-epfl/HHI-Assist/
• https://huggingface.co/datasets/jose-barreiros-tri/hhi-assist

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'jose-barreiros-tri'

dataset class-attribute instance-attribute

dataset: str = 'hhi-assist'

unit_scale class-attribute instance-attribute

unit_scale: float = 1.0

file_extension class-attribute

file_extension: str = '.bvh'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('caregiver', 'recipient')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'caregiver': [], 'recipient': []}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'caregiver': HHI_JOINT_NAMES, 'recipient': HHI_JOINT_NAMES}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/hhi_assist_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    pairs = self._pair_files()
    if not pairs:
        raise ValueError(
            f"No valid caregiver/recipient BVH pairs found in {self.raw_data_path}. "
            "Expected paired files with (CG)/(CR) role annotations in filenames."
        )

    en.logger.info(f"Found {len(pairs)} caregiver-recipient pairs")

    # Update all_motion_files to match pair count (one entry per pair)
    self.all_motion_files = [cg for cg, _ in pairs]

    # Parallel paired CPU read; helper takes a single-arg fn so we adapt
    # via tuple. Each worker parses two BVHs (caregiver + recipient).
    results = self._parallel_read_motion_files(
        list(pairs),
        lambda pair: self._read_paired_sequence(*pair),
        desc="HHI-Assist: parsing BVH pairs",
    )
    cg_sequences: list[MotionSequence] = [cg for cg, _ in results]
    cr_sequences: list[MotionSequence] = [cr for _, cr in results]

    data = {
        "caregiver": BatchMotionSequence.from_motion_sequences(
            cg_sequences,
            num_links=len(self.motion_links_name["caregiver"]),
        ).to(self.device),
        "recipient": BatchMotionSequence.from_motion_sequences(
            cr_sequences,
            num_links=len(self.motion_links_name["recipient"]),
        ).to(self.device),
    }

    metadata = {
        "pairs": [(os.path.basename(cg), os.path.basename(cr)) for cg, cr in pairs],
    }
    return data, metadata

Hand-object manipulation

ArcticDataset

ArcticDataset(options: DatasetOptions, device: device)

Bases: DatasetBase

Dataset class for the Arctic dataset.

Arctic dataset contains sequences with a left hand, a right hand, and an object. Each of these entities is represented by a separate MotionSequence.

References

• https://huggingface.co/datasets/Kashu7100/eden_arctic

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

registry class-attribute instance-attribute

registry: str = 'Kashu7100'

dataset class-attribute instance-attribute

dataset: str = 'eden_arctic'

file_extension class-attribute

file_extension: str = '.npy'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('object', 'left_hand', 'right_hand')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'object': ['rotation'], 'left_hand': [], 'right_hand': []}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'object': ['bottom', 'top'], 'left_hand': list(MANO_KEYPOINTS_NAME), 'right_hand': list(MANO_KEYPOINTS_NAME)}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Prepare the Arctic dataset by reading every raw .npy file in all_motion_files.

RETURNS	DESCRIPTION
tuple[dict[str, BatchMotionSequence], dict[str, Any]]	The object, left-hand and right-hand sequences by motion_sequence_keys, and the per-clip objects metadata.

Source code in eden/datasets/arctic_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    """
    Prepare the Arctic dataset by reading every raw ``.npy`` file in ``all_motion_files``.

    Returns
    -------
    tuple[dict[str, BatchMotionSequence], dict[str, Any]]
        The object, left-hand and right-hand sequences by ``motion_sequence_keys``, and the per-clip
        ``objects`` metadata.
    """
    # Parallel CPU read; bulk H2D move once everything is packed.
    all_motion_sequences = self._parallel_read_motion_files(
        self.all_motion_files,
        self._read_single_sequence,
        desc="Arctic: reading raw .npy",
    )

    all_obj_sequences = BatchMotionSequence.from_motion_sequences(
        [obj_seq for obj_seq, _, _, _ in all_motion_sequences],
        num_links=len(self.motion_links_name["object"]),
    ).to(self.device)
    all_left_hand_sequences = BatchMotionSequence.from_motion_sequences(
        [left_hand_seq for _, left_hand_seq, _, _ in all_motion_sequences]
    ).to(self.device)
    all_right_hand_sequences = BatchMotionSequence.from_motion_sequences(
        [right_hand_seq for _, _, right_hand_seq, _ in all_motion_sequences]
    ).to(self.device)

    metadata = {
        "objects": [meta["object"] for _, _, _, meta in all_motion_sequences],
    }

    data = dict(
        zip(
            self.motion_sequence_keys,
            (all_obj_sequences, all_left_hand_sequences, all_right_hand_sequences),
            strict=True,
        )
    )

    return data, metadata

DexYCBDataset

DexYCBDataset(options: DatasetOptions, device: device)

Bases: DatasetBase

DexYCB single-hand YCB-grasping dataset.

Each sequence is one MANO hand (left or right) interacting with up to four YCB objects; the loader keeps only the grasped object indicated by meta.yml::ycb_grasp_ind. Frames where pose_m is all-zero (no MANO annotation, typically pre-grasp) are trimmed from the leading edge; clips whose invalid frames are not contiguously at the start are skipped with a warning so BatchMotionSequence always sees a contiguous range.

Expected on-disk layout (after extracting dex-ycb-20210415.tar.gz or the per-subject tarballs from https://dex-ycb.github.io/)::

<root>/<YYYYMMDD-subject-NN>/<YYYYMMDD_HHMMSS>/
    ├── meta.yml          # ycb_ids, ycb_grasp_ind, mano_sides, ...
    ├── pose.npz          # world-frame ground-truth poses
    └── <camera_serial>/  # color/depth/per-camera labels (unused here)

pose.npz carries:

• pose_y — shape (T, num_objects, 7), layout [qx, qy, qz, qw, tx, ty, tz]. Quaternion in xyzw, converted to Eden's wxyz convention here.
• pose_m — shape (T, 1, 51), layout [pose(48), trans(3)] in dex-ycb-toolkit's MANO parameterisation (manopth: the global wrist axis-angle, 45 finger PCA coefficients relative to the mean hand, and MANO's translation). pose[0:3] is loaded into base_quat_w. The loader picks up DexYCB's precomputed MANO joint positions from each clip's <seq>/<master_serial>/labels_*.npz::joint_3d and packs them into links_pos_w (21 keypoints in :data:eden.datasets.constants.MANO_KEYPOINTS_NAME order); base_pos_w is their wrist. (MANO's translation is not the wrist: the wrist sits a shape-dependent offset, ~9 cm, away from it.)

With body_model_params=True the hand also carries its MANO pose and shape (body_pose_quat / body_shape, body_model mano_<side>): the PCA expanded into the 15 finger rotations relative to the flat hand, and the subject's betas from calibration/mano_<id>/mano.yml. That needs the MANO model files (body_model_dir) for the PCA basis and the mean hand.

Raw pose.npz poses live in the master camera's frame (x-right, y-down, z-forward, the first entry in meta.yml::serials). calibration_dir (the extracted calibration/ directory) is required: the loader applies each clip's master-camera extrinsic to bring poses into the AprilTag world frame (z-up, origin on the table tag) so a Genesis viewer renders correctly. Preparing the dataset from raw data without calibration_dir raises FileNotFoundError.

Captures run at 30 fps.

No HuggingFace mirror is provided: the upstream is gated behind a Google Drive consent form, so users must download and extract manually, then pass the extracted root as raw_data_path.

References

• https://dex-ycb.github.io/
• https://github.com/NVlabs/dex-ycb-toolkit

PARAMETER	DESCRIPTION
calibration_dir	Path to DexYCB's calibration/ directory. Required: each clip's apriltag extrinsic is left-applied so poses land in the z-up AprilTag world frame. prepare() raises FileNotFoundError when this is unset or doesn't exist.
ground_initial_pose	Translates each clip uniformly along world-Z so the object's world AABB bottom touches z=0 at frame 0; orientation passes through unchanged.
apply_mesh_correction	Applies :data:DEX_YCB_OBJECT_CORRECTIONS as a mesh-local right-multiply on the object trajectory. Disable when rendering against a non-Eden YCB asset version.
models_dir	DexYCB's models/ directory, read by :meth:object_mesh when apply_mesh_correction is off (the poses then refer to the release's meshes). Defaults to the models/ next to calibration_dir.
retargeted_label	When set, also load each clip's retargeted_<label>.npz as a third motion key "robot" (see the comment on the attribute).

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

calibration_dir class-attribute instance-attribute

calibration_dir: str | None = None

ground_initial_pose class-attribute instance-attribute

ground_initial_pose: bool = True

apply_mesh_correction class-attribute instance-attribute

apply_mesh_correction: bool = True

models_dir class-attribute instance-attribute

models_dir: str | None = None

retargeted_label class-attribute instance-attribute

retargeted_label: str | None = None

file_extension class-attribute

file_extension: str = 'pose.npz'

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('hand', 'object')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {'hand': [], 'object': []}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'hand': list(MANO_KEYPOINTS_NAME), 'object': []}

Methods:

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/dex_ycb_dataset.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    if not self.all_motion_files:
        raise FileNotFoundError(
            f"No DexYCB sequences found under {self.raw_data_path}. "
            "Expected `<root>/<subject>/<sequence>/pose.npz` after extracting "
            "`dex-ycb-20210415.tar.gz` or the per-subject archives."
        )

    if self.calibration_dir is None or not os.path.isdir(self.calibration_dir):
        raise FileNotFoundError(
            f"DexYCB: `calibration_dir` is required for the apriltag-world "
            f"transform (got: {self.calibration_dir!r}). Pass "
            "`DexYCBDataset.configure(calibration_dir=...)` pointing at an "
            "extracted `calibration/` directory."
        )

    results = self._parallel_read_motion_files(
        self.all_motion_files,
        self._read_single_sequence,
        desc="DexYCB: reading raw pose.npz",
    )

    valid_files: list[str] = []
    per_clip: dict[str, list] = {
        k: [] for k in ("sequence_ids", "objects", "sides", "first_valid_frames", "master_serials")
    }
    hand_seqs: list[MotionSequence] = []
    object_seqs: list[MotionSequence] = []
    grounding_z_shifts: list[float] = []
    for file_path, result in zip(self.all_motion_files, results):
        if result is None:
            en.logger.warning(f"DexYCB: skipping {file_path} (no contiguous valid hand frames)")
            continue
        hand_seq, object_seq, info = result
        valid_files.append(file_path)
        # ``os.path.basename`` is always ``"pose.npz"``; use a relpath
        # to keep ids unique and human-readable.
        per_clip["sequence_ids"].append(os.path.relpath(os.path.dirname(file_path), self.raw_data_path))
        per_clip["objects"].append(info["object"])
        per_clip["sides"].append(info["side"])
        per_clip["first_valid_frames"].append(info["first_valid_frame"])
        per_clip["master_serials"].append(info["master_serial"])
        grounding_z_shifts.append(info["grounding_z_shift"])
        hand_seqs.append(hand_seq)
        object_seqs.append(object_seq)

    if not hand_seqs:
        raise RuntimeError(
            "All DexYCB sequences were filtered out. Expect at least one clip "
            "with a contiguous run of valid `pose_m` frames; inspect the "
            "`pose.npz::pose_m` arrays for unexpected zero gaps."
        )

    # ------------------------------------------------------------------
    # Optional second pass: read inline `retargeted_<label>.npz` files
    # next to each surviving pose.npz. Clips missing the retargeted file
    # are dropped here so the resulting hand/object/robot sequences stay
    # aligned across motion keys.
    # ------------------------------------------------------------------
    robot_seqs: list[MotionSequence] = []
    robot_dofs_name: list[str] = []
    if self.retargeted_label is not None:
        keep_idx: list[int] = []
        for i, file_path in enumerate(valid_files):
            robot_seq, this_dofs_name = self._read_retargeted_robot(file_path)
            if robot_seq is None:
                continue
            if not robot_dofs_name:
                robot_dofs_name = this_dofs_name
            elif this_dofs_name != robot_dofs_name:
                raise ValueError(
                    f"DexYCB: `retargeted_<{self.retargeted_label}>.npz` files have "
                    f"inconsistent `dofs_name` across clips. First clip declared "
                    f"{robot_dofs_name}; {file_path} declared {this_dofs_name}."
                )
            robot_seqs.append(robot_seq)
            keep_idx.append(i)

        if not robot_seqs:
            raise RuntimeError(
                f"DexYCB: `retargeted_label={self.retargeted_label!r}` but no clip "
                f"under {self.raw_data_path} has a matching "
                f"`retargeted_{self.retargeted_label}.npz`. Run "
                f"`examples/recipes/retargeting/dex_retargeter/dex_ycb_position_retargeting.py` first."
            )
        # Filter hand/object/per-clip metadata down to the clips that
        # actually had a retargeted file.
        if len(keep_idx) < len(valid_files):
            en.logger.warning(
                f"DexYCB: dropping {len(valid_files) - len(keep_idx)} clip(s) "
                f"missing `retargeted_{self.retargeted_label}.npz`; "
                f"{len(keep_idx)} retained."
            )
            hand_seqs = [hand_seqs[i] for i in keep_idx]
            object_seqs = [object_seqs[i] for i in keep_idx]
            grounding_z_shifts = [grounding_z_shifts[i] for i in keep_idx]
            for key in per_clip:
                per_clip[key] = [per_clip[key][i] for i in keep_idx]
            valid_files = [valid_files[i] for i in keep_idx]

    # Keep the base class's post-prepare ``metadata['all_motion_files']``
    # in sync with the surviving motion buffers.
    self.all_motion_files = valid_files

    data = {
        "hand": BatchMotionSequence.from_motion_sequences(hand_seqs).to(self.device),
        "object": BatchMotionSequence.from_motion_sequences(object_seqs).to(self.device),
    }
    if self.retargeted_label is not None:
        data["robot"] = BatchMotionSequence.from_motion_sequences(robot_seqs).to(self.device)
        # Shadow the ClassVars at the instance level so downstream
        # consumers (and ``select_trajectories``) see the third key.
        self.motion_sequence_keys = (*type(self).motion_sequence_keys, "robot")
        self.motion_dofs_name = {**type(self).motion_dofs_name, "robot": robot_dofs_name}

    metadata: dict[str, Any] = {
        **per_clip,
        # Already applied to ``base_pos_w``; surfaced so downstream
        # code re-loading positions from the raw source (e.g. MANO
        # keypoints from per-frame labels) can match the dataset frame.
        "grounding_z_shifts": np.asarray(grounding_z_shifts, dtype=np.float32),
    }
    if self.retargeted_label is not None:
        metadata["retargeted_label"] = self.retargeted_label
    return data, metadata

object_mesh

object_mesh(object_name: str, materials: bool = False)

The trimesh.Trimesh (metres) that clip object poses of object_name place: its geometry, loaded once per dataset.

With apply_mesh_correction the poses were moved onto Eden's YCB asset (ycb/models/<name>/textured.obj); without it they are the release's, on <models_dir>/<name>/textured_simple.obj. With materials, read again with its material and texture, and not cached: trimesh decodes the 4096^2 texture as it loads it (~50 MB per object), which only an export of the mesh needs.

Source code in eden/datasets/dex_ycb_dataset.py

def object_mesh(self, object_name: str, materials: bool = False):
    """The ``trimesh.Trimesh`` (metres) that clip object poses of ``object_name`` place: its geometry, loaded once
    per dataset.

    With ``apply_mesh_correction`` the poses were moved onto Eden's YCB asset (``ycb/models/<name>/textured.obj``);
    without it they are the release's, on ``<models_dir>/<name>/textured_simple.obj``. With ``materials``, read
    again with its material and texture, and not cached: trimesh decodes the 4096^2 texture as it loads it
    (~50 MB per object), which only an export of the mesh needs.
    """
    if materials:
        return self._load_object_mesh(object_name, materials=True)
    if object_name not in self._meshes:  # an import asks once per clip (its floor) for the ~20 YCB objects
        self._meshes[object_name] = self._load_object_mesh(object_name)
    return self._meshes[object_name]

load_apriltag_master_calibration

load_apriltag_master_calibration(extrinsics_id: str) -> tuple[ndarray | None, str]

Read <calibration_dir>/extrinsics_<id>/extrinsics.yml.

Returns (T_apriltag_master, master_serial). The apriltag entry in the file is T_master_apriltag (AprilTag pose in the master camera's frame); we invert it so left-applying brings master-cam-frame poses into the z-up AprilTag world. master_serial is the calibration's top-level master field, used by downstream code that reads per-camera files (e.g. MANO keypoints from labels_*.npz).

Source code in eden/datasets/dex_ycb_dataset.py

def load_apriltag_master_calibration(self, extrinsics_id: str) -> tuple[np.ndarray | None, str]:
    """Read ``<calibration_dir>/extrinsics_<id>/extrinsics.yml``.

    Returns ``(T_apriltag_master, master_serial)``. The ``apriltag``
    entry in the file is ``T_master_apriltag`` (AprilTag pose in the
    master camera's frame); we invert it so left-applying brings
    master-cam-frame poses into the z-up AprilTag world.
    ``master_serial`` is the calibration's top-level ``master``
    field, used by downstream code that reads per-camera files
    (e.g. MANO keypoints from ``labels_*.npz``).
    """
    if self.calibration_dir is None:
        return None, ""
    path = os.path.join(self.calibration_dir, f"extrinsics_{extrinsics_id}", "extrinsics.yml")
    if not os.path.exists(path):
        en.logger.warning(f"DexYCB: extrinsics file missing: {path}")
        return None, ""
    with open(path, "r") as f:
        ext = yaml.load(f, Loader=_DexYCBYamlLoader)
    block = ext.get("extrinsics", {})
    raw = block.get("apriltag")
    if raw is None:
        en.logger.warning(f"DexYCB: 'apriltag' entry missing in {path}; available: {sorted(block)}")
        return None, str(ext.get("master", ""))
    T_master_apriltag = np.asarray(raw, dtype=np.float32).reshape(3, 4)
    R = T_master_apriltag[:3, :3]
    t = T_master_apriltag[:3, 3]
    R_inv = R.T
    t_inv = -R_inv @ t
    return np.concatenate([R_inv, t_inv[:, None]], axis=1), str(ext.get("master", ""))

MANO hand-object releases

TACO, OakInk2 and HOT3D ship MANO parameters rather than keypoints, so their loaders need the licence-gated MANO model files (body_model_dir). They share HandObjectDatasetBase.

Shared base of the MANO hand-object releases (TACO, OakInk2, HOT3D): per-side hand and held-object sequences.

Classes

HandObjectDatasetBase

HandObjectDatasetBase(options: DatasetOptions, device: device)

Bases: DatasetBase

Base of the MANO hand-object releases: each clip is the hands of sides and the object each one holds.

Keys: <side>_hand (base = the MANO wrist, links_pos_w = the 21 :data:~eden.datasets.constants.MANO_KEYPOINTS_NAME keypoints) and object_<side> (the held object's pose) for each side in sides. A clip where one of those hands holds nothing is left out and named in metadata["skipped"], as is any clip its reader rejects. With body_model_params=True each hand also carries its MANO pose and shape. The releases ship MANO parameters, not keypoints, so the MANO model files (body_model_dir) are always needed.

Per-clip metadata: clip_names, recordings, objects_<side> (the held object id); the poses of every object of a recording come from :meth:recording_object_poses, read on demand from the release.

The file filters (include_pattern, the list files, ...) match each clip's <unit>/<clip name> path, on a fresh build as on a cache load: a clip is only known once its unit is read, so a build reads every unit first.

PARAMETER	DESCRIPTION
sides	both, right or left: the hands loaded, each of which must hold an object. TYPE: str

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

sides class-attribute instance-attribute

sides: Literal['both', 'right', 'left'] = 'both'

source_framerate class-attribute

source_framerate: float

has_recording_context class-attribute

has_recording_context: bool = False

motion_sequence_keys class-attribute

motion_sequence_keys: tuple[str, ...] = ('right_hand', 'object_right', 'left_hand', 'object_left')

motion_dofs_name class-attribute

motion_dofs_name: dict[str, list[str]] = {k: [] for k in motion_sequence_keys}

motion_links_name class-attribute

motion_links_name: dict[str, list[str]] = {'right_hand': list(MANO_KEYPOINTS_NAME), 'left_hand': list(MANO_KEYPOINTS_NAME), 'object_right': [], 'object_left': []}

Methods:

units

units(root: Path) -> list[str]

The release's units of work (recordings or sequences), as paths under root; raises naming the expected layout when root is not the release.

Source code in eden/datasets/hand_object_base.py

def units(self, root: Path) -> list[str]:
    """The release's units of work (recordings or sequences), as paths under ``root``; raises naming the expected
    layout when ``root`` is not the release.
    """
    raise NotImplementedError

read_unit

read_unit(unit: str) -> list[HandObjectClip | tuple[str, str]]

The clips of unit; a (name, reason) pair is a clip the reader rejected.

Source code in eden/datasets/hand_object_base.py

def read_unit(self, unit: str) -> list[HandObjectClip | tuple[str, str]]:
    """The clips of ``unit``; a ``(name, reason)`` pair is a clip the reader rejected."""
    raise NotImplementedError

object_mesh

object_mesh(obj_id: str, materials: bool = False)

The trimesh.Trimesh (metres) of object obj_id, in the frame its poses place: its geometry, cached.

With materials, read again with its materials and textures, and not cached: trimesh decodes the textures as it loads them (a 4096^2 texture holds ~50 MB), which only an export of the mesh needs.

Source code in eden/datasets/hand_object_base.py

def object_mesh(self, obj_id: str, materials: bool = False):
    """The ``trimesh.Trimesh`` (metres) of object ``obj_id``, in the frame its poses place: its geometry, cached.

    With ``materials``, read again with its materials and textures, and not cached: trimesh decodes the textures
    as it loads them (a 4096^2 texture holds ~50 MB), which only an export of the mesh needs.
    """
    if materials:
        return self._load_object_mesh(obj_id, materials=True)
    if obj_id not in self._meshes:
        self._meshes[obj_id] = self._load_object_mesh(obj_id)
    return self._meshes[obj_id]

recording_object_poses

recording_object_poses(recording: str) -> dict[str, ndarray]

Every object's (T, 4, 4) world pose over the whole recording (read from the release on demand).

Source code in eden/datasets/hand_object_base.py

def recording_object_poses(self, recording: str) -> dict[str, np.ndarray]:
    """Every object's (T, 4, 4) world pose over the whole recording (read from the release on demand)."""
    raise NotImplementedError(
        f"{type(self).__name__} clips are whole recordings; their own object poses are the recording's."
    )

prepare

prepare() -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]

Source code in eden/datasets/hand_object_base.py

def prepare(self) -> tuple[dict[str, BatchMotionSequence], dict[str, Any]]:
    if self.body_model_dir is None:
        raise ValueError(
            f"{type(self).__name__} ships MANO parameters, not keypoints, so it needs the MANO model files: set "
            f"body_model_dir to the directory holding MANO_RIGHT.pkl / MANO_LEFT.pkl (mano.is.tue.mpg.de)."
        )
    results = self._parallel_read_motion_files(
        self.all_motion_files, self.read_unit, desc=f"{type(self).__name__}: reading"
    )
    sides = self._sides
    skipped: list[tuple[str, str]] = []
    accepted: list[tuple[str, HandObjectClip]] = []
    for unit, clips in zip(self.all_motion_files, results):
        for clip in clips:
            if isinstance(clip, tuple):
                skipped.append(clip)
                continue
            missing = [s for s in sides if s not in clip.hands or s not in clip.objects]
            if missing:
                skipped.append((clip.name, f"no hand with an object ({'/'.join(missing)})"))
                continue
            lengths = {len(clip.hands[s]) for s in sides} | {len(clip.poses[clip.objects[s]]) for s in sides}
            if len(lengths) > 1:
                skipped.append((clip.name, f"frame counts differ ({sorted(lengths)})"))
                continue
            clip = self._drop_frames(clip, lengths.pop())
            if isinstance(clip, tuple):
                skipped.append(clip)
                continue
            accepted.append((os.path.join(unit, clip.name), clip))
    self._clips_known = True
    if accepted:
        _, keep = self._apply_file_filter([path for path, _ in accepted])
        if keep is not None:
            accepted = [accepted[i] for i in keep]

    sequences: dict[str, list] = {key: [] for key in self.motion_sequence_keys}
    meta: dict[str, list] = {k: [] for k in self._PER_CLIP_METADATA_KEYS}
    files: list[str] = []
    for path, clip in accepted:
        try:  # every side packed before any is kept, so a clip is kept whole or not at all
            packed = {
                side: (self._pack_hand(clip, side), self._pack_object(clip.poses[clip.objects[side]]))
                for side in sides
            }
        except _ShortClipError:
            skipped.append((clip.name, "too short after retiming to target_framerate"))
            continue
        for side, (hand_seq, object_seq) in packed.items():
            sequences[f"{side}_hand"].append(hand_seq)
            sequences[f"object_{side}"].append(object_seq)
            meta[f"objects_{side}"].append(clip.objects[side])
        for side in SIDES:
            if side not in sides:
                meta[f"objects_{side}"].append("")
        meta["clip_names"].append(clip.name)
        meta["recordings"].append(clip.recording)
        files.append(path)
    if skipped:
        reasons = count_reasons(reason for _, reason in skipped)
        en.logger.info(f"{type(self).__name__}: kept {len(files)} clips, skipped {len(skipped)} {reasons}")
    if not files:
        raise RuntimeError(
            f"{type(self).__name__}: every clip under {self.raw_data_path} was skipped "
            f"({len(skipped)}; first: {skipped[:3]}). sides={self.sides!r} needs each of those hands to hold an "
            f"object."
        )
    # prepare() owns the clip order; _initialize stores all_motion_files as the per-trajectory metadata.
    self.all_motion_files = files
    data = {}
    for key, seqs in sequences.items():
        batch = BatchMotionSequence.from_motion_sequences(
            seqs, num_dofs=0, num_links=len(self.motion_links_name[key])
        )
        batch.entity_kind = "hand" if key.endswith("_hand") else "rigid_object"
        data[key] = batch.to(self.device)
    return data, {**meta, "skipped": skipped}

near

near(obj_id: str, pts: ndarray, bound: float) -> ndarray

Distance from each (N, 3) object-frame point to the nearest mesh vertex where below bound, else inf; points outside the vertices' box padded by bound skip the KD-tree.

Source code in eden/datasets/hand_object_base.py

def near(self, obj_id: str, pts: np.ndarray, bound: float) -> np.ndarray:
    """Distance from each (N, 3) object-frame point to the nearest mesh vertex where below ``bound``, else inf;
    points outside the vertices' box padded by ``bound`` skip the KD-tree.
    """
    if obj_id not in self._vertex_trees:
        v = np.asarray(self.object_mesh(obj_id).vertices)
        self._vertex_trees[obj_id] = cKDTree(v), v.min(0), v.max(0)
    tree, lo, hi = self._vertex_trees[obj_id]
    d = np.full(len(pts), np.inf)
    box = ((pts >= lo - bound) & (pts <= hi + bound)).all(1)
    d[box] = tree.query(pts[box], distance_upper_bound=bound)[0]
    return d

HandObjectClip dataclass

HandObjectClip(name: str, recording: str, hands: dict[str, MANOHand], objects: dict[str, str], poses: dict[str, ndarray])

One clip a release's reader cut: hands, the object each holds, and object poses (object -> world).

PARAMETER	DESCRIPTION
name	Unique clip name. TYPE: str
recording	The recording it was cut from; clips of one recording share :meth:HandObjectDatasetBase.recording_object_poses. TYPE: str
hands	Hand side -> hand. TYPE: dict[str, MANOHand]
objects	Hand side -> the object id it holds (both sides may name one object). TYPE: dict[str, str]
poses	Object id -> (T, 4, 4) pose. TYPE: dict[str, ndarray]

Attributes

name instance-attribute

name: str

recording instance-attribute

recording: str

hands instance-attribute

hands: dict[str, MANOHand]

objects instance-attribute

objects: dict[str, str]

poses instance-attribute

poses: dict[str, ndarray]

Methods:

MANOHand dataclass

MANOHand(side: str, root_pos: ndarray, root_rotvec: ndarray, finger_pose: ndarray, betas: ndarray, keypoints: ndarray | None = None)

One hand of a clip in the dataset's z-up world, at the release's frame rate.

PARAMETER	DESCRIPTION
side	right or left: which MANO model it is. TYPE: str
root_pos	(T, 3) MANO root joint (the wrist). TYPE: ndarray
root_rotvec	(T, 3) global orientation, axis-angle. TYPE: ndarray
finger_pose	(T, 45) finger axis-angles relative to the flat hand. TYPE: ndarray
betas	(T, B) or (B,) shape coefficients. TYPE: ndarray
keypoints	(T, 21, 3) float32 :data:~eden.datasets.constants.MANO_KEYPOINTS_NAME keypoints, once a reader that ran the model for its own analysis stored them (:meth:keep_keypoints); kept by :meth:frames. The vertices are never kept: at 778 x 3 per frame they would hold a whole release in memory until it is packed. TYPE: ndarray | None DEFAULT: None

Attributes

side instance-attribute

side: str

root_pos instance-attribute

root_pos: ndarray

root_rotvec instance-attribute

root_rotvec: ndarray

finger_pose instance-attribute

finger_pose: ndarray

betas instance-attribute

betas: ndarray

keypoints class-attribute instance-attribute

keypoints: ndarray | None = None

Methods:

frames

frames(idx: ndarray | slice) -> MANOHand

The hand on frames idx (a per-frame betas is sliced, a constant one kept).

Source code in eden/datasets/hand_object_base.py

def frames(self, idx: np.ndarray | slice) -> MANOHand:
    """The hand on frames ``idx`` (a per-frame ``betas`` is sliced, a constant one kept)."""
    betas = self.betas[idx] if self.betas.ndim == 2 else self.betas
    keypoints = None if self.keypoints is None else self.keypoints[idx]
    return MANOHand(self.side, self.root_pos[idx], self.root_rotvec[idx], self.finger_pose[idx], betas, keypoints)

rotated

rotated(R: ndarray) -> MANOHand

The hand with the world rotated by R; finger rotations are local and stay, keypoints are dropped.

Source code in eden/datasets/hand_object_base.py

def rotated(self, R: np.ndarray) -> MANOHand:
    """The hand with the world rotated by ``R``; finger rotations are local and stay, keypoints are dropped."""
    rotvec = (Rotation.from_matrix(R) * Rotation.from_rotvec(self.root_rotvec)).as_rotvec()
    return MANOHand(self.side, self.root_pos @ R.T, rotvec, self.finger_pose, self.betas)

forward

forward(model_dir: str | None) -> tuple[ndarray, ndarray]

(T, 16, 3) joints and (T, 778, 3) vertices, float64.

Source code in eden/datasets/hand_object_base.py

def forward(self, model_dir: str | None) -> tuple[np.ndarray, np.ndarray]:
    """(T, 16, 3) joints and (T, 778, 3) vertices, float64."""
    joints, verts = body_model_forward(
        mano_model_name(self.side),
        model_dir,
        torch.from_numpy(self.root_pos),
        quat_from_axis_angle(torch.from_numpy(self.root_rotvec)),
        axis_angles_to_body_pose_quat(self.finger_pose),
        torch.from_numpy(np.broadcast_to(self.betas, (len(self), self.betas.shape[-1])).copy()),
    )
    return joints.double().numpy(), verts.double().numpy()

keep_keypoints

keep_keypoints(joints: ndarray, verts: ndarray) -> None

Store the keypoints of this hand's :meth:forward output, so packing does not run the model again.

Source code in eden/datasets/hand_object_base.py

def keep_keypoints(self, joints: np.ndarray, verts: np.ndarray) -> None:
    """Store the keypoints of this hand's :meth:`forward` output, so packing does not run the model again."""
    self.keypoints = mano_keypoints(torch.from_numpy(joints), torch.from_numpy(verts)).float().numpy()

keypoint_track

keypoint_track(model_dir: str | None) -> ndarray

(T, 21, 3) keypoints: the stored ones, or the model's.

Source code in eden/datasets/hand_object_base.py

def keypoint_track(self, model_dir: str | None) -> np.ndarray:
    """(T, 21, 3) keypoints: the stored ones, or the model's."""
    if self.keypoints is None:
        return mano_keypoints(*map(torch.from_numpy, self.forward(model_dir))).float().numpy()
    return self.keypoints

TacoDataset

TacoDataset(options: DatasetOptions, device: device)

Bases: HandObjectDatasetBase

TACO: one clip per recording, both hands and the tool and target objects.

Layout (github.com/Satorinal/TACO)::

<root>/Hand_Poses/<(action, tool, target)>/<seq>/{right,left}_hand.pkl, {right,left}_hand_shape.pkl
<root>/Object_Poses/<(action, tool, target)>/<seq>/{tool,target}_<id>.npy
<root>/object_models_released/<id>_cm.obj

Hands are MANO relative to the flat hand, with the wrist given; objects (4, 4) per frame; 30 Hz, z-up, turned 180 degrees about z (CHORD's frame). Each hand holds the object it stays nearest to over the clip (both hands the one object when there is only one). Clip names taco_<action>__<tool>__<target>_<seq>. Needs the MANO model files (body_model_dir); see :class:~eden.datasets.hand_object_base.HandObjectDatasetBase.

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

source_framerate class-attribute

source_framerate: float = 30.0

Methods:

units

units(root: Path) -> list[str]

Source code in eden/datasets/taco_dataset.py

def units(self, root: Path) -> list[str]:
    missing = [d for d in ("Hand_Poses", "Object_Poses", "object_models_released") if not (root / d).is_dir()]
    if missing:
        raise FileNotFoundError(
            f"{root} is not a TACO release: {missing} missing. Point raw_data_path at the directory holding "
            "Hand_Poses/, Object_Poses/ and object_models_released/ (github.com/Satorinal/TACO)."
        )
    return sorted(str(d) for d in (root / "Hand_Poses").glob("*/*") if d.is_dir())

read_unit

read_unit(unit: str) -> list[HandObjectClip | tuple[str, str]]

Source code in eden/datasets/taco_dataset.py

def read_unit(self, unit: str) -> list[HandObjectClip | tuple[str, str]]:
    hand_dir = Path(unit)
    triplet, seq = hand_dir.parent.name, hand_dir.name
    name = "taco_" + "__".join(w.replace(" ", "_") for w in triplet.strip("()").split(", ")) + f"_{seq}"
    pose_dir = Path(self.raw_data_path) / "Object_Poses" / triplet / seq
    poses = {}
    for kind in ("tool", "target"):
        files = sorted(pose_dir.glob(f"{kind}_*.npy"))
        if len(files) == 1:
            poses[files[0].stem.split("_", 1)[1]] = np.load(files[0]).astype(np.float64)
    hands = {}
    for side in ("right", "left"):
        if (hand_dir / f"{side}_hand.pkl").exists() and (hand_dir / f"{side}_hand_shape.pkl").exists():
            with open(hand_dir / f"{side}_hand.pkl", "rb") as f:
                frames = pickle.load(f)
            keys = sorted(frames)
            pose = np.stack([frames[k]["hand_pose"].numpy() for k in keys]).astype(np.float64)
            with open(hand_dir / f"{side}_hand_shape.pkl", "rb") as f:
                betas = pickle.load(f)["hand_shape"].numpy().astype(np.float64)
            wrist = np.stack([frames[k]["hand_trans"].numpy() for k in keys]).astype(np.float64)
            hands[side] = MANOHand(side, wrist, pose[:, :3], pose[:, 3:], betas).rotated(_TACO_TO_WORLD)
    if not poses:
        return [(name, "no tool or target object")]
    if len({len(p) for p in poses.values()} | {len(h) for h in hands.values()}) > 1:
        return [(name, "frame counts differ")]
    poses = rotate_poses(poses, _TACO_TO_WORLD)
    centres = {o: p[:, :3, :3] @ self.object_mesh(o).vertices.mean(0) + p[:, :3, 3] for o, p in poses.items()}
    objects = assign_nearest({s: h.root_pos for s, h in hands.items()}, centres)
    return [HandObjectClip(name, f"{triplet}/{seq}", hands, objects, poses)]

OakInk2Dataset

OakInk2Dataset(options: DatasetOptions, device: device)

Bases: HandObjectDatasetBase

OakInk2: each recording cut into the stages of its program, one clip per stage.

Layout::

<root>/anno_preview/<seq>.pkl               # raw_mano and obj_transf per mocap frame
<root>/program/program_info/<seq>.json      # the program's stages: frame spans, objects per hand
<root>/object_repair/align_ds/<id>/model.obj

120 Hz, y-up turned z-up. A stage is the frames from its first span's start to its last span's end (clip <seq>__stNN). Each hand holds the object its 16 MANO links touch most (a vertex within contact of the object's mesh, counted per link and frame), as CHORD assigns them, since the program can name the other hand's object; the program's own object when it touches none. A stage with several objects in one loaded hand (sides), or with a mocap frame missing from its span, is skipped, as is a recording whose mocap frame ids step by more than 1 (packed at 120 Hz, it would play too fast). :meth:recording_object_poses gives every object with a mesh over the whole recording. Needs the MANO model files (body_model_dir).

PARAMETER	DESCRIPTION
contact	Distance (m) under which a MANO vertex touches an object. TYPE: float

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

contact class-attribute instance-attribute

contact: float = 0.01

source_framerate class-attribute

source_framerate: float = 120.0

has_recording_context class-attribute

has_recording_context: bool = True

Methods:

units

units(root: Path) -> list[str]

Source code in eden/datasets/oakink2_dataset.py

def units(self, root: Path) -> list[str]:
    missing = [
        d for d in ("anno_preview", "program/program_info", "object_repair/align_ds") if not (root / d).is_dir()
    ]
    if missing:
        raise FileNotFoundError(
            f"{root} is not an OakInk2 release: {missing} missing. Point raw_data_path at the directory holding "
            "anno_preview/, program/ and object_repair/."
        )
    programs = root / "program" / "program_info"
    return sorted(str(p) for p in (root / "anno_preview").glob("*.pkl") if (programs / f"{p.stem}.json").exists())

read_unit

read_unit(unit: str) -> list[HandObjectClip | tuple[str, str]]

Source code in eden/datasets/oakink2_dataset.py

def read_unit(self, unit: str) -> list[HandObjectClip | tuple[str, str]]:
    path = Path(unit)
    seq = path.stem
    with open(path, "rb") as f:
        anno = pickle.load(f)
    with open(Path(self.raw_data_path) / "program" / "program_info" / f"{seq}.json") as f:
        program = json.load(f)
    frame_ids = list(anno["mocap_frame_id_list"])
    # The recording's mocap stride (its most common step). Packed at source_framerate, a recording that steps by
    # more than 1 would play stride x too fast, so it is skipped; a larger step inside a stage is a missing frame.
    steps = np.diff(frame_ids)
    stride = int(np.bincount(steps[steps > 0]).argmax()) if (steps > 0).any() else 1
    if stride != 1:
        return [(seq, f"mocap frame ids step by {stride}, not 1 ({self.source_framerate:g} Hz)")]
    context = self._context(anno)
    out: list[HandObjectClip | tuple[str, str]] = []
    for i, (key, stage) in enumerate(program.items()):
        name = f"{seq}__st{i:02d}"
        spans = [s for s in ast.literal_eval(key) if s is not None]
        lists = {"right": stage.get("obj_list_rh") or [], "left": stage.get("obj_list_lh") or []}
        if not spans:
            out.append((name, "no stage interval"))
            continue
        if any(len(lists[side]) > 1 for side in self._sides):
            out.append((name, "several objects in one hand"))
            continue
        start, end = min(s[0] for s in spans), max(s[1] for s in spans)
        idx = np.array([k for k, f in enumerate(frame_ids) if start <= f < end], dtype=int)
        if not len(idx):
            out.append((name, f"no mocap frame in the stage interval (frames {start}-{end})"))
            continue
        frames = [frame_ids[k] for k in idx]
        gaps = np.flatnonzero(np.diff(frames) > 1)
        if len(gaps):
            out.append((name, f"missing frames ({len(gaps)} gap(s), first after mocap frame {frames[gaps[0]]})"))
            continue
        hands = {side: self._hand(anno, frames, side) for side in self._sides}
        verts = {}
        for side, hand in hands.items():
            joints, verts[side] = hand.forward(self.body_model_dir)
            hand.keep_keypoints(joints, verts[side])
        objects = self._held(verts, {o: p[idx] for o, p in context.items()}, lists)
        # The program's object when no hand touches one with a mesh: it can lack a mesh or a pose track.
        tracked = {str(o) for o in anno["obj_transf"]}
        unusable = sorted(o for o in set(objects.values()) if o not in tracked or not self._mesh_path(o).exists())
        if unusable:
            what = "pose track" if unusable[0] not in tracked else "mesh"
            out.append((name, f"held object has no {what} ({unusable[0]})"))
            continue
        # From the annotation, as FlashDexRetarget reads them.
        poses = _object_poses(anno, frames, set(objects.values()))
        out.append(HandObjectClip(name, seq, hands, objects, poses))
    return out

recording_object_poses

recording_object_poses(recording: str) -> dict[str, ndarray]

Source code in eden/datasets/oakink2_dataset.py

def recording_object_poses(self, recording: str) -> dict[str, np.ndarray]:
    with open(Path(self.raw_data_path) / "anno_preview" / f"{recording}.pkl", "rb") as f:
        return self._context(pickle.load(f))

Hot3dDataset

Hot3dDataset(options: DatasetOptions, device: device)

Bases: HandObjectDatasetBase

HOT3D, read as CHORD's hot3d_loader.py and cut into hand-object segments as its segment_to_atomic.py.

Layout::

<root>/dataset/<recording>/{mano_hand_pose_trajectory.jsonl, dynamic_objects.csv,
                            headset_trajectory.csv, metadata.json}
<root>/_assets/assets/<uid>.glb

Recordings with hand and object ground truth; one timestamp per 10 ms, 30 Hz; the frames with both hands and every object (a segment across a missing frame is skipped). Quest3's y-up world is turned z-up, then yawed so the workspace (the mean of every object position and hand transl) lies along +y from the headset's first position, turned 90 degrees further (CHORD's frame). Hands are MANO with 15 PCA components relative to the mean hand. drop_first_n_frames / drop_last_n_frames trim a segment after its pad_frames are added, so they take the still padding first.

Segments (clip <recording>_segNNN, CHORD's numbering): a hand is active while a fingertip is within tip_contact of an object (gaps under gap_frames bridged); windows of activity split where the touched object changes (each frame's object: the one most of both hands' 16 MANO links are nearest to, within link_contact); segments under min_frames dropped, pad_frames still frames added at both ends. Mode A: one hand active, B: both on one object, C: on two objects; the segments of modes are kept. In mode B both hands hold the object; in A only the active hand holds one (so sides='both' skips the segment, its other hand holding nothing), and in C each hand holds its own. FlashDexRetarget hands every hand the segment's one object, which labels an idle hand (A) or the other hand's object (C) as held. :meth:recording_object_poses gives every object over the recording. Needs the MANO model files (body_model_dir).

PARAMETER	DESCRIPTION
tip_contact	Fingertip-object distance (m) under which a hand is active. TYPE: float
link_contact	Vertex-object distance (m) under which a MANO link touches an object. TYPE: float
gap_frames	Inactive gaps shorter than this are bridged. TYPE: int
min_frames	Shorter segments are dropped. TYPE: int
pad_frames	Still frames repeated at each end of a segment. TYPE: int
modes	Segment modes kept (A, B, C). TYPE: tuple[str, ...]

Source code in eden/datasets/base.py

def __init__(self, options: DatasetOptions, device: torch.device):
    super().__init__(options)
    if not self._applies_frame_window and self._trims_clips:
        raise ValueError(
            f"{type(self).__name__} does not trim its clips at read time, so it cannot apply "
            f"drop_first_n_frames={self.drop_first_n_frames} / drop_last_n_frames={self.drop_last_n_frames}. "
            f"Leave both at 0 and trim after loading with the TrimFrames modifier instead: "
            f"modifiers=[TrimFrames.configure(trim_first={self.drop_first_n_frames}, "
            f"trim_last={self.drop_last_n_frames})]."
        )

    self._device = device
    self._initialize()

Attributes

tip_contact class-attribute instance-attribute

tip_contact: float = 0.02

link_contact class-attribute instance-attribute

link_contact: float = 0.01

gap_frames class-attribute instance-attribute

gap_frames: int = 15

min_frames class-attribute instance-attribute

min_frames: int = 30

pad_frames class-attribute instance-attribute

pad_frames: int = 10

modes class-attribute instance-attribute

modes: tuple[Literal['A', 'B', 'C'], ...] = ('B',)

source_framerate class-attribute

source_framerate: float = 30.0

has_recording_context class-attribute

has_recording_context: bool = True

Methods:

units

units(root: Path) -> list[str]

Source code in eden/datasets/hot3d_dataset.py

def units(self, root: Path) -> list[str]:
    missing = [d for d in ("dataset", "_assets/assets") if not (root / d).is_dir()]
    if missing:
        raise FileNotFoundError(
            f"{root} is not a HOT3D release: {missing} missing. Point raw_data_path at the directory holding "
            "dataset/<recording>/ and _assets/assets/."
        )
    out = []
    for p in sorted((root / "dataset").iterdir()):
        if not (p / "mano_hand_pose_trajectory.jsonl").exists():
            continue
        with open(p / "metadata.json") as f:
            if json.load(f).get("have_hand_object_pose_gt"):
                out.append(str(p))
    return out

read_unit

read_unit(unit: str) -> list[HandObjectClip | tuple[str, str]]

Source code in eden/datasets/hot3d_dataset.py

def read_unit(self, unit: str) -> list[HandObjectClip | tuple[str, str]]:
    rec = Path(unit)
    recording = self._recording(rec)
    if isinstance(recording, str):
        return [(rec.name, recording)]
    hands, poses = recording.hands, recording.poses
    verts = {}
    for side, hand in hands.items():  # the vertices segment the recording, then go: only the keypoints stay
        joints, verts[side] = hand.forward(self.body_model_dir)
        hand.keep_keypoints(joints, verts[side])
    out: list[HandObjectClip | tuple[str, str]] = []
    for i, (a, b, mode, held) in enumerate(self._segments(verts, poses, recording.uids)):
        name = f"{rec.name}_seg{i:03d}"
        if mode not in self.modes:
            out.append((name, f"mode {mode}"))
            continue
        if not held:
            out.append((name, "no object in contact"))
            continue
        missing = recording.gaps[(recording.gaps >= a) & (recording.gaps < b - 1)]
        if len(missing):
            after = recording.stamps[missing[0]]
            out.append((name, f"missing frames ({len(missing)} gap(s), first after timestamp_ns {after})"))
            continue
        pad = self.pad_frames  # still frames at both ends
        idx = np.concatenate([np.full(pad, a), np.arange(a, b), np.full(pad, b - 1)]).astype(int)
        out.append(
            HandObjectClip(
                name,
                rec.name,
                {s: h.frames(idx) for s, h in hands.items()},
                held,
                {o: poses[o][idx] for o in set(held.values())},
            )
        )
    return out

recording_object_poses

recording_object_poses(recording: str) -> dict[str, ndarray]

Source code in eden/datasets/hot3d_dataset.py

def recording_object_poses(self, recording: str) -> dict[str, np.ndarray]:
    out = self._recording(Path(self.raw_data_path) / "dataset" / recording, hands=False)
    if isinstance(out, str):
        raise ValueError(f"HOT3D recording {recording}: {out}")
    return out.poses

Modifiers

Offline transforms applied to loaded sequences, in order. DatasetModifierOptions.motion_sequence_keys selects the sequences each one targets.

Base class and registry for dataset modifiers.

Attributes

DATASET_MODIFIER_REGISTRY module-attribute

DATASET_MODIFIER_REGISTRY = Registry('DATASET_MODIFIER')

Classes

DatasetModifier

DatasetModifier(options: T)

Bases: ConfigurableMixin[DatasetModifierOptions]

Base class for composable dataset modifiers.

Dataset modifiers are offline transforms applied to loaded motion sequences (after cache load or raw prepare) rather than per simulation step. They are useful for data augmentation and canonicalization — e.g. spatial offsets, rotations, time warps, or dropping fields that the downstream task does not need. Modifiers are composed via :class:Compose, similar to :class:~eden.managers.modifiers.actions.actuators.Compose::

modifiers = [
    Translate.configure(offset=(0.0, 0.0, 0.05)),
]

Subclasses override :meth:apply_to_sequence to transform a single :class:~eden.utils.motion_reference.motion_data.BatchMotionSequence. The default :meth:apply iterates over the configured keys (or every key when motion_sequence_keys is None) and delegates to :meth:apply_to_sequence.

Source code in eden/utils/common.py

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)
        )

Attributes

motion_sequence_keys class-attribute instance-attribute

motion_sequence_keys: tuple[str, ...] | None = None

Methods:

apply

apply(motion_sequences: dict[str, 'BatchMotionSequence']) -> dict[str, 'BatchMotionSequence']

Apply this modifier to every selected key and return the result.

motion_sequence_keys=None (the default) applies to every key; an explicit empty tuple applies to none and is a deliberate no-op — use that to temporarily disable a modifier without removing it.

Source code in eden/datasets/modifiers/base.py

def apply(
    self,
    motion_sequences: dict[str, "BatchMotionSequence"],
) -> dict[str, "BatchMotionSequence"]:
    """Apply this modifier to every selected key and return the result.

    ``motion_sequence_keys=None`` (the default) applies to every key;
    an explicit empty tuple applies to none and is a deliberate no-op —
    use that to temporarily disable a modifier without removing it.
    """
    out = dict(motion_sequences)
    for key in self._selected_keys(motion_sequences):
        out[key] = self.apply_to_sequence(out[key])
    return out

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Transform a single :class:BatchMotionSequence. Override in subclasses.

Source code in eden/datasets/modifiers/base.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    """Transform a single :class:`BatchMotionSequence`. Override in subclasses."""
    return motion_seq

Functions:

build_modifier

build_modifier(options: DatasetModifierOptions | dict) -> DatasetModifier

Build the modifier options names, by registry name.

options may be the options object or its dict form (a config loaded from YAML).

Source code in eden/datasets/modifiers/base.py

def build_modifier(options: DatasetModifierOptions | dict) -> DatasetModifier:
    """Build the modifier ``options`` names, by registry name.

    ``options`` may be the options object or its dict form (a config loaded from YAML).
    """
    if isinstance(options, dict):
        options = DatasetModifierOptions(**options)
    return DATASET_MODIFIER_REGISTRY.get(options.name)(options=options)

Built-in dataset modifiers.

Attributes

Classes

Compose

Compose(options: DatasetModifierOptions)

Bases: DatasetModifier

Chain multiple dataset modifiers in sequence.

PARAMETER	DESCRIPTION
modifiers	Ordered sequence of child modifier configurations. Each modifier's :meth:apply runs against the output of the previous one. TYPE: tuple[DatasetModifierOptions, ...] | list[DatasetModifierOptions]

Example

::

Compose.configure(
    modifiers=[
        Translate.configure(offset=(0.0, 0.0, 0.05)),
        Translate.configure(
            offset=(0.5, 0.0, 0.0),
            motion_sequence_keys=("obj",),
        ),
    ]
)

Source code in eden/datasets/modifiers/transforms.py

def __init__(self, options: DatasetModifierOptions):
    super().__init__(options=options)
    self._modifiers: list[DatasetModifier] = [build_modifier(mod_options) for mod_options in self.modifiers]

Attributes

modifiers class-attribute instance-attribute

modifiers: tuple[DatasetModifierOptions, ...] | list[DatasetModifierOptions] = ()

Methods:

apply

apply(motion_sequences)

Source code in eden/datasets/modifiers/transforms.py

def apply(self, motion_sequences):
    selected = {k: motion_sequences[k] for k in self._selected_keys(motion_sequences)}
    for mod in self._modifiers:
        selected = mod.apply(selected)
    merged = dict(motion_sequences)
    merged.update(selected)
    return merged

Translate

Translate(options: T)

Bases: DatasetModifier

Shift world-frame translations of every motion by a constant offset.

The offset is added to base_pos_w and, if populated, links_pos_w. Body-frame fields (links_pos_b) and all velocity fields are invariant under a pure translation and are left untouched.

PARAMETER	DESCRIPTION
offset	(x, y, z) offset in the dataset world frame. Typical use is a z-only offset to lift trajectories off the ground for a taller robot. TYPE: tuple[float, float, float]

Source code in eden/utils/common.py

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)
        )

Attributes

offset class-attribute instance-attribute

offset: tuple[float, float, float] = (0.0, 0.0, 0.0)

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/transforms.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if all(v == 0.0 for v in self.offset):
        return motion_seq
    if motion_seq.base_pos_w is None:
        return motion_seq

    device = motion_seq.base_pos_w.device
    dtype = motion_seq.base_pos_w.dtype
    offset = torch.tensor(self.offset, dtype=dtype, device=device)

    # ConcatBatchTensor stores frames in a flat (total_frames, 3) buffer;
    # a broadcast-add over the last dim is equivalent to adding the offset
    # to every frame of every motion.
    motion_seq.base_pos_w.concatenated_tensor.add_(offset)
    if motion_seq.links_pos_w is not None:
        motion_seq.links_pos_w.concatenated_tensor.add_(offset)
    return motion_seq

Rotate

Rotate(options: T)

Bases: DatasetModifier

Rotate world-frame fields about a given axis through the world origin.

Rotates every world-frame position, quaternion, linear velocity, and angular velocity by a single rigid rotation. Body-frame fields (links_*_b) are invariant under a world rotation and are left untouched.

Because the rotation is about the world origin, combining :class:Rotate with :class:Translate is order-sensitive — rotate first, then translate to apply the rotation about the current origin; translate first, then rotate to orbit the motion about a point.

PARAMETER	DESCRIPTION
axis	Unit rotation axis in the world frame. Automatically normalized. Default is the Z axis, i.e. a yaw rotation. TYPE: tuple[float, float, float]
angle_deg	Rotation angle in degrees (right-handed about axis). TYPE: float

Source code in eden/utils/common.py

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)
        )

Attributes

axis class-attribute instance-attribute

axis: tuple[float, float, float] = (0.0, 0.0, 1.0)

angle_deg class-attribute instance-attribute

angle_deg: float = 0.0

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/transforms.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if self.angle_deg == 0.0:
        return motion_seq

    # No world-frame field populated: nothing to rotate.
    ref_field = next(
        (
            getattr(motion_seq, name)
            for name in self._VEC_FIELDS + self._QUAT_FIELDS
            if getattr(motion_seq, name) is not None
        ),
        None,
    )
    if ref_field is None:
        return motion_seq

    device = ref_field.device
    dtype = ref_field.dtype

    axis = torch.tensor(self.axis, dtype=dtype, device=device)
    if axis.norm().item() < 1e-8:
        raise ValueError(f"Rotate axis must be non-zero, got {self.axis}")
    q = axis_angle_to_quat(torch.tensor(math.radians(self.angle_deg), dtype=dtype, device=device), axis)

    for name in self._VEC_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is None:
            continue
        t = cbt.concatenated_tensor  # (N, 3) or (N, L, 3)
        t.copy_(transform_by_quat(t, q))

    for name in self._QUAT_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is None:
            continue
        t = cbt.concatenated_tensor  # (N, 4) or (N, L, 4)
        # transform_quat_by_quat(v, u) returns u * v — pass u=q to left-multiply.
        t.copy_(transform_quat_by_quat(t, q))

    return motion_seq

MirrorLR

MirrorLR(options: T)

Bases: DatasetModifier

Mirror motion data across a world-frame plane and relabel symmetric parts.

Mirroring is an improper transformation: vectors, pseudovectors, and unit quaternions all follow different sign rules, which this modifier applies automatically. What it cannot know is how the robot's DOFs and links map to each other under reflection — that is robot-specific, and the user provides it as explicit integer swap tables.

For a mirror plane with normal on axis a (e.g. Y for left/right):

• Positions and linear velocities flip component a.
• Angular velocities (pseudovectors under improper transform) flip the other two components.
• Unit quaternions (w, x, y, z) flip every xyz component except the one on axis a; w is unchanged.

The sign pattern is applied to both world-frame and body-frame link fields. The body-frame rule follows from applying the same mirror consistently to the base frame; for perfectly symmetric robots this is exact, and for slightly asymmetric ones it is a close approximation — consumers requiring strict accuracy should recompute body-frame fields from world-frame fields downstream.

Ordering within the modifier: axis sign flips → DOF sign flips → DOF swaps → link swaps. User-supplied dofs_sign_flip and dofs_swap both reference the original DOF indices (before swapping).

PARAMETER	DESCRIPTION
axis	Mirror-plane normal index: 0 = X, 1 = Y (default, left-right), 2 = Z. TYPE: int
dofs_swap	Pairs of DOF indices to swap after mirroring. Example for a biped: ((hip_l_idx, hip_r_idx), (knee_l_idx, knee_r_idx), ...). TYPE: tuple[tuple[int, int], ...]
dofs_sign_flip	DOF indices whose stored value is negated under mirroring — typically joints that rotate about an axis orthogonal to the mirror plane (e.g. a hip-roll axial joint when mirroring left/right). TYPE: tuple[int, ...]
links_swap	Pairs of link indices to swap after mirroring. Applied to every links_* field along the link dimension. TYPE: tuple[tuple[int, int], ...]

Source code in eden/utils/common.py

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)
        )

Attributes

axis class-attribute instance-attribute

axis: int = 1

dofs_swap class-attribute instance-attribute

dofs_swap: tuple[tuple[int, int], ...] = ()

dofs_sign_flip class-attribute instance-attribute

dofs_sign_flip: tuple[int, ...] = ()

links_swap class-attribute instance-attribute

links_swap: tuple[tuple[int, int], ...] = ()

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/transforms.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if self.axis not in (0, 1, 2):
        raise ValueError(f"MirrorLR.axis must be 0, 1, or 2; got {self.axis}")

    # Pick a populated field to read device/dtype from; otherwise this is a no-op.
    ref = next(
        (
            getattr(motion_seq, name)
            for name in (*self._POS_LIKE_FIELDS, *self._ANG_VEL_FIELDS, *self._QUAT_FIELDS, *self._DOF_FIELDS)
            if getattr(motion_seq, name) is not None
        ),
        None,
    )
    if ref is None:
        return motion_seq
    device, dtype = ref.device, ref.dtype

    # Sign patterns — see docstring for derivation.
    pos_sign = torch.ones(3, dtype=dtype, device=device)
    pos_sign[self.axis] = -1.0
    ang_sign = -pos_sign  # flip the *other two* components
    quat_sign = torch.ones(4, dtype=dtype, device=device)
    quat_sign[1:] = ang_sign

    # 1. Axis sign flips (position-like, angular velocity, quaternion).
    for name in self._POS_LIKE_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is not None:
            cbt.concatenated_tensor.mul_(pos_sign)
    for name in self._ANG_VEL_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is not None:
            cbt.concatenated_tensor.mul_(ang_sign)
    for name in self._QUAT_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is not None:
            cbt.concatenated_tensor.mul_(quat_sign)

    # 2. Per-DOF sign flips (original-index convention).
    if self.dofs_sign_flip:
        flip_idx = torch.tensor(list(self.dofs_sign_flip), dtype=torch.long, device=device)
        for name in self._DOF_FIELDS:
            cbt = getattr(motion_seq, name)
            if cbt is None:
                continue
            cbt.concatenated_tensor[:, flip_idx] *= -1.0

    # 3. DOF swaps.
    if self.dofs_swap:
        for name in self._DOF_FIELDS:
            cbt = getattr(motion_seq, name)
            if cbt is None:
                continue
            t = cbt.concatenated_tensor
            perm = list(range(t.shape[1]))
            for a, b in self.dofs_swap:
                perm[a], perm[b] = perm[b], perm[a]
            idx = torch.tensor(perm, dtype=torch.long, device=t.device)
            t.copy_(t[:, idx])

    # 4. Link swaps.
    if self.links_swap:
        for name in self._LINK_FIELDS:
            cbt = getattr(motion_seq, name)
            if cbt is None:
                continue
            t = cbt.concatenated_tensor  # (N, L, *)
            if t.shape[1] == 0:
                continue
            perm = list(range(t.shape[1]))
            for a, b in self.links_swap:
                perm[a], perm[b] = perm[b], perm[a]
            idx = torch.tensor(perm, dtype=torch.long, device=t.device)
            t.copy_(t[:, idx])

    return motion_seq

Functions:

Temporal dataset modifiers.

Modifiers in this module change the number of frames per motion while keeping the dataset's target_framerate (and therefore the simulation dt) fixed. They rebuild the underlying :class:ConcatBatchTensor buffers and update motion_seq.buffer_length accordingly.

Attributes

Classes

SpeedScale

SpeedScale(options: T)

Bases: DatasetModifier

Scale playback speed by resampling frames at a fixed target_framerate.

A speed of 2.0 makes the motion play twice as fast — each sequence is resampled to half the number of frames — while keeping the framerate (and therefore the simulation dt) unchanged. Because covered distances are preserved but traversed in half the wall-clock time, every stored velocity is multiplied by speed.

Position-like fields are interpolated linearly; quaternion fields are interpolated with slerp. Body-frame fields are resampled alongside world-frame fields so consumers that read them stay consistent.

The per-sequence frame count is rounded to max(2, round(T / speed)).

PARAMETER	DESCRIPTION
speed	Playback-speed factor. > 1 fast-forwards, < 1 slows down. Must be positive. TYPE: float

Source code in eden/utils/common.py

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)
        )

Attributes

speed class-attribute instance-attribute

speed: float = 1.0

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/temporal.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if self.speed == 1.0:
        return motion_seq
    if self.speed <= 0:
        raise ValueError(f"SpeedScale.speed must be > 0, got {self.speed}")

    old_sizes = motion_seq.buffer_length.tolist()
    new_sizes = [max(2, round(s / self.speed)) for s in old_sizes]

    for name, rows in MOTION_FIELD_ROWS.items():
        cbt = getattr(motion_seq, name)
        if cbt is None:
            continue
        resampled = [_resample(cbt[i], new_sizes[i], rows) for i in range(len(cbt))]
        setattr(motion_seq, name, _rebuild_cbt(cbt, resampled))

    for name in _VELOCITY_FIELDS:
        cbt = getattr(motion_seq, name)
        if cbt is None:
            continue
        cbt.concatenated_tensor.mul_(self.speed)

    motion_seq.buffer_length = torch.tensor(
        new_sizes,
        device=motion_seq.buffer_length.device,
        dtype=motion_seq.buffer_length.dtype,
    )
    return motion_seq

TrimFrames

TrimFrames(options: T)

Bases: DatasetModifier

Drop leading and trailing frames from every sequence.

Equivalent to DatasetBase.drop_first_n_frames / drop_last_n_frames but available as a composable post-load modifier — useful for cropping warm-up or settling frames without re-preparing the cache.

Note

When the dataset has multiple motion_sequence_keys (e.g. robot + object in a manipulation dataset), apply the same trim to all keys to preserve the per-sequence length invariant across keys. Selecting a subset via motion_sequence_keys=(...) will desynchronize the lengths.

PARAMETER	DESCRIPTION
trim_first	Number of frames to drop from the start of each sequence. Must be >= 0. TYPE: int
trim_last	Number of frames to drop from the end of each sequence. Must be >= 0. TYPE: int

Source code in eden/utils/common.py

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)
        )

Attributes

trim_first class-attribute instance-attribute

trim_first: int = 0

trim_last class-attribute instance-attribute

trim_last: int = 0

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/temporal.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if self.trim_first == 0 and self.trim_last == 0:
        return motion_seq
    if self.trim_first < 0 or self.trim_last < 0:
        raise ValueError(
            f"TrimFrames requires non-negative trim counts, got "
            f"trim_first={self.trim_first}, trim_last={self.trim_last}"
        )

    old_sizes = motion_seq.buffer_length.tolist()
    new_sizes = [s - self.trim_first - self.trim_last for s in old_sizes]
    for s, new_s in zip(old_sizes, new_sizes, strict=True):
        if new_s < _MIN_CLIP_FRAMES:
            raise ValueError(
                f"TrimFrames(trim_first={self.trim_first}, trim_last={self.trim_last}) leaves {new_s} frame(s) "
                f"in a sequence of length {s}; every clip needs at least {_MIN_CLIP_FRAMES}, because a single "
                f"frame spans no time. Lower trim_first / trim_last."
            )

    # Every per-frame field, contact annotations included: a field left untrimmed keeps the old length and
    # desynchronises from the rest of its sequence.
    window = slice(self.trim_first, -self.trim_last or None)
    for name in BatchMotionSequence.attrs_with_frame_dim:
        cbt = getattr(motion_seq, name)
        if cbt is None:
            continue
        setattr(motion_seq, name, _rebuild_cbt(cbt, [cbt[i][window].clone() for i in range(len(cbt))]))

    motion_seq.buffer_length = torch.tensor(
        new_sizes,
        device=motion_seq.buffer_length.device,
        dtype=motion_seq.buffer_length.dtype,
    )
    return motion_seq

Noise-based dataset modifiers.

Attributes

Classes

Jitter

Jitter(options: T)

Bases: DatasetModifier

Add additive noise to selected motion fields.

Applies independent, per-element noise to every entry of each listed :class:BatchMotionSequence field. Randomness comes from the global torch RNG — seed it upstream (e.g. via torch.manual_seed in your script) for reproducibility.

Quaternion fields are not rejected, but jittering them with additive noise breaks the unit-norm invariant and should be used carefully — prefer :class:Rotate for orientation perturbation.

PARAMETER	DESCRIPTION
fields	:class:BatchMotionSequence attribute names to jitter, e.g. ("dofs_pos",) or ("base_pos_w", "links_pos_w"). TYPE: tuple[str, ...]
distribution	Noise distribution. TYPE: ('gaussian', 'uniform') DEFAULT: "gaussian"
std	Standard deviation for distribution="gaussian". Must be >= 0; 0 is a no-op. TYPE: float
low	Bounds for distribution="uniform". Require low <= high; low == high is a no-op. TYPE: float
high	Bounds for distribution="uniform". Require low <= high; low == high is a no-op. TYPE: float

Source code in eden/utils/common.py

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)
        )

Attributes

fields class-attribute instance-attribute

fields: tuple[str, ...] = ('dofs_pos',)

distribution class-attribute instance-attribute

distribution: Literal['gaussian', 'uniform'] = 'gaussian'

std class-attribute instance-attribute

std: float = 0.0

low class-attribute instance-attribute

low: float = 0.0

high class-attribute instance-attribute

high: float = 0.0

Methods:

apply_to_sequence

apply_to_sequence(motion_seq: 'BatchMotionSequence') -> 'BatchMotionSequence'

Source code in eden/datasets/modifiers/noise.py

def apply_to_sequence(self, motion_seq: "BatchMotionSequence") -> "BatchMotionSequence":
    if self.distribution == "gaussian":
        if self.std < 0:
            raise ValueError(f"Jitter.std must be >= 0, got {self.std}")
        if self.std == 0.0:
            return motion_seq
    elif self.distribution == "uniform":
        if self.low > self.high:
            raise ValueError(f"Jitter.low ({self.low}) must be <= high ({self.high})")
        if self.low == self.high:
            return motion_seq
    else:
        raise ValueError(f"Jitter.distribution must be 'gaussian' or 'uniform', got {self.distribution!r}")

    for name in self.fields:
        cbt = getattr(motion_seq, name, None)
        if cbt is None:
            raise ValueError(
                f"Jitter: field '{name}' is not populated on the motion sequence. "
                f"Jitterable fields must be present on the loaded dataset."
            )
        if not isinstance(cbt, ConcatBatchTensor):
            raise ValueError(
                f"Jitter: field '{name}' is a {type(cbt).__name__}, not a ConcatBatchTensor. "
                f"Jitter only supports frame-dim fields (dofs_pos, base_pos_w, links_pos_w, etc.); "
                f"batch-dim scalars like 'framerate' and 'buffer_length' cannot be jittered."
            )
        t = cbt.concatenated_tensor
        if self.distribution == "gaussian":
            noise = torch.randn(t.shape, dtype=t.dtype, device=t.device) * self.std
        else:
            noise = torch.empty(t.shape, dtype=t.dtype, device=t.device).uniform_(self.low, self.high)
        t.add_(noise)
    return motion_seq

Ground-the-first-frame dataset modifier.

For each motion sequence, set the robot to frame-0's qpos in a transient Genesis env, read the world-frame AABB's bottom-z, and (if below the configured margin) shift the entire clip uniformly along +Z so the first frame sits at-or-above the floor. The shift is a pure translation, so all velocity and body-frame fields stay invariant — same idiom as :class:~eden.datasets.modifiers.transforms.Translate, only the offset is auto-computed per clip from FK instead of being configured up front.

Use this when retargeting can leave clips penetrating the ground at clip start — e.g. AMASS clips retargeted to a robot whose feet sit lower than the source actor, or any pipeline where the source capture's floor plane isn't perfectly aligned with z=0.

It shares the transient-env plumbing (:func:eden.utils.genesis_runtime.transient_robot_env) with :class:~eden.datasets.modifiers.smooth_start_end.SmoothStartTransition.

Attributes

Classes

GroundFirstFrame

GroundFirstFrame(options: T)

Bases: DatasetModifier

Shift each clip along +Z so frame-0's world AABB sits at-or-above z=0.

Pipeline per clip:

1. Build the robot from robot_options in a transient kinematics env (one env per clip — frame-0 FK runs in a single batched pass).
2. set_qpos to (base_pos_w[0], base_quat_w[0], dofs_pos[0]).
3. Read robot.get_AABB()[:, 0, 2] per env → per-clip floor min-z.
4. dz = max(0, safety_margin_m - min_z). Clips already above the floor get dz=0 and pass through unchanged.
5. Add dz to every frame of base_pos_w (and links_pos_w when populated). Body-frame link fields and all velocity fields are invariant under pure translation and are left untouched.

PARAMETER	DESCRIPTION
robot_options	Eden robot options used to build the transient kinematics env. Must match the robot the dataset is retargeted to (same DOF order, same morph). Required. TYPE: EntityOptions
safety_margin_m	Extra clearance above z=0 to enforce. dz is computed as max(0, safety_margin_m - aabb_min_z), so a positive value nudges the foot off the floor; 0.0 (default) lets the foot sit exactly on the plane. TYPE: float
chunk_size	Maximum number of clips processed per Genesis env. None (default) allocates one env per clip in a single batched FK pass — fastest for LAFAN1-class libraries (~75 clips). For SEED-class libraries (100k+ clips) one-env-per-clip would OOM; set this to a few hundred / few thousand to bound the env count. The transient Genesis env is built once at chunk_size and re-used across chunks via repeated set_qpos — only the last chunk's qpos is padded with the previous chunk's last entry to fill the env count (cheap extra FK on a few unused envs). TYPE: int | None

Notes

Multi-entity datasets: this modifier shifts a single robot key in place. If the dataset carries paired entities (robot + object) that must move together, scope the modifier to the robot key with motion_sequence_keys=(robot_key,) and add a matching shift on the object key — translating the robot alone would desync them.

Source code in eden/utils/common.py

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)
        )

Attributes

robot_options class-attribute instance-attribute

robot_options: EntityOptions = None

safety_margin_m class-attribute instance-attribute

safety_margin_m: float = 0.0

chunk_size class-attribute instance-attribute

chunk_size: int | None = None

Methods:

apply

apply(motion_sequences)

Source code in eden/datasets/modifiers/ground.py

def apply(self, motion_sequences):
    if self.safety_margin_m < 0:
        raise ValueError(
            f"GroundFirstFrame: safety_margin_m must be >= 0 (clearance above the floor); "
            f"got {self.safety_margin_m}."
        )
    if self.chunk_size is not None and self.chunk_size < 1:
        raise ValueError(
            f"GroundFirstFrame: chunk_size must be >= 1 (or None for no chunking); got {self.chunk_size}."
        )

    out = dict(motion_sequences)
    for key in self._selected_keys(motion_sequences, single=True):
        out[key] = self._apply_to_key(out[key], key)
    return out

Functions:

Smooth start-transition dataset modifier.

Pre-pends every clip with a synthesized stand-still → IK-stepped → SLERP- merged segment so RSI from a default pose to clip-frame-0 is no longer a teleport. Ports OpenTrack's StartEndTransitionHandler to Eden's :class:DatasetModifier API. The actual IK math lives in :mod:eden.utils.motion_reference.bipedal_ik; this module is the glue that maps Eden's :class:BatchMotionSequence ↔ a Genesis RigidEntity and re-FKs link fields after the transition is inserted.

Why a modifier and not a per-step env hook

The IK solve is per-clip and sequential (no benefit from vectorising over envs), and it runs hundreds of forward-kinematics passes per clip. Doing it inside the training loop would thrash wall-time. As a modifier it runs once per dataset load, before training starts. Modifiers are not baked into the processed cache, so the solve is repeated on every load.

What this modifier mutates

The output BatchMotionSequence has per-sequence lengths extended by the synthesized prefix. Fields rebuilt from scratch: * base_pos_w, base_quat_w, dofs_pos * base_lin_vel_w, base_ang_vel_w, dofs_vel (FD from positions) * links_pos_w, links_quat_w (re-FK through Genesis) * links_lin_vel_w, links_ang_vel_w (FD from link poses) * links_pos_b, links_quat_b, links_lin_vel_b, links_ang_vel_b (recomputed from world-frame + the new base pose)

Attributes

Classes

SmoothStartTransition

SmoothStartTransition(options: T)

Bases: DatasetModifier

Pre-pend each clip with a synthesized stand-up → step → merge → clip segment.

PARAMETER	DESCRIPTION
ik_profile	Robot-agnostic IK spec — see :class:BipedalIKProfile. The G1 preset lives at eden.options.robots.unitree_g1.G1_IK_PROFILE and is the usual value here. Required at apply() time; defaults to None only so the modifier's configure() works without it. TYPE: BipedalIKProfile | None
motion_links_name_per_key	Map key → dataset link name order used for Genesis link-index lookup when re-FK-ing link fields. None skips link rebuild — useful for datasets that don't carry per-link data. TYPE: dict[str, list[str]] | None
default_pose_sec	Duration (s) of stand-still padding at the start. OpenTrack default: 1.0 s. TYPE: float
single_step_threshold_m	If the inter-foot distance differs between stance and clip frame 0 by more than this, two steps are synthesized; otherwise one. TYPE: float
transition_sec	SLERP-merge length between the IK endpoint and clip frame 0. TYPE: float
foot_height_m	Peak Z bump of the swing foot during the synthesized step. TYPE: float
com_xy_offset_m	Forward offset of the IK's CoM target relative to the foot midpoint. TYPE: float
step_sec	Duration of each synthesized step. TYPE: float
double_stance_frames	Frames of double-stance dwell between successive steps. TYPE: int
start_foot	Which foot lifts first. TYPE: Literal['left', 'right']
max_ik_iters	Per-frame IK iteration cap. TYPE: int

Source code in eden/utils/common.py

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)
        )

Attributes

ik_profile class-attribute instance-attribute

ik_profile: BipedalIKProfile | None = None

motion_links_name_per_key class-attribute instance-attribute

motion_links_name_per_key: dict[str, list[str]] | None = None

default_pose_sec class-attribute instance-attribute

default_pose_sec: float = 1.0

single_step_threshold_m class-attribute instance-attribute

single_step_threshold_m: float = 0.1

transition_sec class-attribute instance-attribute

transition_sec: float = 0.5

foot_height_m class-attribute instance-attribute

foot_height_m: float = 0.05

com_xy_offset_m class-attribute instance-attribute

com_xy_offset_m: float = 0.03

step_sec class-attribute instance-attribute

step_sec: float = 0.5

double_stance_frames class-attribute instance-attribute

double_stance_frames: int = 3

start_foot class-attribute instance-attribute

start_foot: str = 'right'

max_ik_iters class-attribute instance-attribute

max_ik_iters: int = 200

Methods:

apply

apply(motion_sequences)

Source code in eden/datasets/modifiers/smooth_start_end.py

def apply(self, motion_sequences):
    if self.ik_profile is None:
        raise ValueError(
            "SmoothStartTransition: ik_profile is required — pass a BipedalIKProfile "
            "(e.g. eden.options.robots.unitree_g1.G1_IK_PROFILE)."
        )
    profile = self.ik_profile

    # The modifier owns one bipedal robot — the one in
    # ``profile.robot_options``. Each key in ``motion_sequences`` is
    # treated as that robot's qpos sequence (laid out in the robot's
    # ``dofs_name`` order). Multi-entity datasets (robot + object) need
    # the user to scope the modifier explicitly via
    # ``motion_sequence_keys=(robot_key,)``; otherwise the link-name
    # lookup would silently misindex against the bipedal scene. The
    # ``single=True`` reject catches the "forgot to scope" case the
    # ``motion_sequence_keys`` filter alone can't.
    keys = self._selected_keys(motion_sequences, single=True)

    # Build a single Genesis env + robot, reuse across every sequence in
    # the key. Genesis env build is heavyweight (kernel compile, GPU
    # buffers); one build per ``apply()`` amortises it across the
    # library.
    with transient_robot_env(profile.robot_options) as (_env, robot):
        out = dict(motion_sequences)
        for key in keys:
            out[key] = self._apply_to_key(out[key], profile, robot, key)
        return out

Functions:

TCRS terrain-adaptation dataset modifier.

Runs the TCRS terrain-adaptive motion optimizer (:mod:eden.extensions.motion_optimization.tcrs) over every clip of a motion dataset: MPPI foot-swing planning against a terrain height field, support-aware pelvis height re-targeting, and whole-body Jacobian IK on a MuJoCo model of the robot. Flat-ground clips come out with footsteps and pelvis height that track the configured terrain.

The terrain can be described three ways (all reduce to one height field):

• TerrainOptions — Eden's procedural terrain (the generator grid is run once, off-scene),
• EntityOptions — a 3D terrain mesh file, rasterized by downward raycasts at mesh_resolution,
• str — path to a height-field .npz exported via :func:eden.utils.terrains.io.export_height_field.

The optimization costs minutes per clip and dataset modifiers rerun on every dataset load (the dataset cache is deliberately saved pre-modifier), so results are cached on disk keyed by a content hash of (clip data, terrain, parameters, robot MJCF) — the second load of an unchanged dataset is near-free. Set cache_dir=None to disable.

This modifier adapts clips 1→1 in place at a single terrain_offset placement. Upstream-style 1→N random-placement augmentation is out of scope here; see examples/recipes/motion_optimization/tcrs for batch variant generation.

Attributes

Classes

TCRSTerrainAdapt

TCRSTerrainAdapt(options: T)

Bases: DatasetModifier

Adapt flat-ground clips to a terrain with the TCRS optimizer.

PARAMETER	DESCRIPTION
terrain	The terrain to adapt to: procedural TerrainOptions, a mesh-file EntityOptions, or the path of an exported height-field .npz. Required. TYPE: TerrainOptions | EntityOptions | str
robot_options	Eden robot options whose file resolves to the robot's MJCF; the IK solves against this model. Must match the robot the dataset is retargeted to (e.g. en.robots.G1Dofs29()). Required. TYPE: EntityOptions
terrain_offset	SE(2) placement (dx, dy, dyaw) applied to every clip before adaptation: the clip is yaw-rotated by dyaw (radians) about its first-frame pelvis position, then translated by (dx, dy). Use it to position the motion on the terrain feature of interest. TYPE: tuple[float, float, float]
mesh_resolution	Rasterization cell size in meters when terrain is a mesh-file EntityOptions. Ignored for the other terrain kinds. TYPE: float
robot_spec	A :class:~eden.options.extensions.motion_optimization.TCRSRobotOptions (or its dict form / a partial override dict, e.g. {"dofs_name": [...]} when the dataset's DOF order differs). Unspecified fields keep the G1 defaults — dofs_name/links_name default to Eden's canonical G1 orderings, which is what Eden datasets produce. Unknown keys raise at construction. Partial ik_config dicts deep-merge onto the defaults (DEFAULT_IK_CONFIG): solver keys merge individually, link-weight entries merge per link name. TYPE: TCRSRobotOptions | dict | None
planner	A :class:~eden.options.extensions.motion_optimization.TCRSPlannerOptions (or its dict form / a partial override dict) for the pipeline knobs (MPPI sampling, stabilizer, lowpass cutoff, ...). Unknown keys raise at construction. IK solver weights live in the robot spec: override via robot_spec={"ik_config": {...}}. TYPE: TCRSPlannerOptions | dict | None
cache_dir	Directory for the content-addressed result cache. None disables caching — every dataset load then re-runs the full optimization. TYPE: str | None

Notes

links_pos_w/links_quat_w must be populated on the sequence (Eden mimic datasets fill them via FK). contact_points_w is left untouched and a warning is emitted when present: world-frame contact points recorded on flat ground are stale after terrain adaptation.

Source code in eden/utils/common.py

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)
        )

Attributes

terrain class-attribute instance-attribute

terrain: TerrainOptions | EntityOptions | str = None

robot_options class-attribute instance-attribute

robot_options: EntityOptions = None

terrain_offset class-attribute instance-attribute

terrain_offset: tuple[float, float, float] = (0.0, 0.0, 0.0)

mesh_resolution class-attribute instance-attribute

mesh_resolution: float = 0.02

robot_spec class-attribute instance-attribute

robot_spec: TCRSRobotOptions | dict | None = None

planner class-attribute instance-attribute

planner: TCRSPlannerOptions | dict | None = None

cache_dir class-attribute instance-attribute

cache_dir: str | None = '~/.cache/eden/tcrs'

Methods:

apply

apply(motion_sequences)

Source code in eden/datasets/modifiers/terrain_adapt.py

def apply(self, motion_sequences):
    if self.terrain is None:
        raise ValueError(
            "TCRSTerrainAdapt: 'terrain' is required — pass TerrainOptions, a mesh-file EntityOptions, "
            "or the path of an exported height-field .npz."
        )
    if self.robot_options is None:
        raise ValueError(
            "TCRSTerrainAdapt: 'robot_options' is required — pass the robot's EntityOptions "
            "(e.g. en.robots.G1Dofs29()) so the IK can load its MJCF."
        )
    offset = tuple(float(v) for v in self.terrain_offset)
    if len(offset) != 3:
        raise ValueError(f"TCRSTerrainAdapt: terrain_offset must be (dx, dy, dyaw); got {self.terrain_offset}.")

    keys = self._selected_keys(motion_sequences, single=True)

    robot_options = self._coerce_options(self.robot_options, EntityOptions, "robot_options")
    if not robot_options.file:
        raise ValueError(
            "TCRSTerrainAdapt: robot_options.file is empty — it must name the robot's MJCF "
            "(an empty pattern would trigger a full asset-registry download)."
        )
    mjcf_path = robot_options.fetch_asset()
    if not str(mjcf_path).endswith(".xml"):
        raise ValueError(
            f"TCRSTerrainAdapt: robot_options.file must resolve to a MuJoCo MJCF (.xml); got '{mjcf_path}'. "
            "The TCRS IK runs on a MuJoCo model of the robot."
        )

    terrain = self._build_terrain()
    terrain.print_info()
    adapter = self._build_adapter(mjcf_path)

    out = dict(motion_sequences)
    for key in keys:
        out[key] = self._apply_to_key(out[key], key, terrain, adapter, mjcf_path, offset)
    return out

Functions:

Body models

Parametric body models (MANO, SMPL, SMPL-H, SMPL-X) behind the opt-in body_model_params=True fields, run on a numpy linear-blend-skinning rig with no smplx dependency.

Parametric body models behind the opt-in body-model fields of a motion sequence.

A dataset loaded with body_model_params=True stores, per sequence that has one, the model's pose and shape: body_pose_quat (the local rotation of every joint below the root, wxyz, relative to the model's rest pose), body_shape (betas) and the per-clip body_model name; the root is the sequence's base, base_pos_w being the root joint's position. This module names the models (:data:BODY_MODELS), converts a release's own MANO parameterisation into that form, and runs the model back to joints and vertices.

The model files are licence-gated (MANO: mano.is.tue.mpg.de) and never shipped; they are read from a directory the caller passes (DatasetBase.body_model_dir). The models run on :class:SkinnedBodyModel, a numpy linear-blend- skinning rig that matches the smplx package to well under a micrometre (tests/datasets/utils) and also skins rigs outside the SMPL family (OTRetarget's SOMA). A model is read from its .npz when the directory has one, else its .pkl. The MANO, SMPL and SMPL-H .pkl files pickle a chumpy array or two (shapedirs); :func:body_model_arrays reads those without chumpy (:class:~eden.utils.chumpy_pickle.ChumpyFreeUnpickler), so no model file needs an extra.

Attributes

BodyModelFamily module-attribute

BodyModelFamily = Literal['mano', 'smpl', 'smplh', 'smplx']

BODY_MODELS module-attribute

BODY_MODELS: dict[str, BodyModelSpec] = {'mano_right': BodyModelSpec('mano', 'right', 15, 'MANO_RIGHT.pkl'), 'mano_left': BodyModelSpec('mano', 'left', 15, 'MANO_LEFT.pkl'), **{f'{family}_{gender}': BodyModelSpec(family, gender, joints, f'{family.upper()}_{gender.upper()}.pkl') for family, joints in (('smpl', 23), ('smplh', 51), ('smplx', 54)) for gender in ('neutral', 'male', 'female')}}

MANO_TIP_VERTICES module-attribute

MANO_TIP_VERTICES: tuple[int, ...] = (745, 317, 444, 556, 673)

CHORD_TIP_VERTICES module-attribute

CHORD_TIP_VERTICES: dict[str, int] = {'thumb': 766, 'index': 353, 'middle': 467, 'ring': 576, 'pinky': 695}

Classes

BodyModelSpec dataclass

BodyModelSpec(family: BodyModelFamily, variant: str, num_body_joints: int, file: str)

One parametric body model.

PARAMETER	DESCRIPTION
family	mano, smpl, smplh or smplx. TYPE: BodyModelFamily
variant	The hand side (MANO) or the gender (SMPL family). TYPE: str
num_body_joints	Joints below the root, i.e. body_pose_quat's second dimension. TYPE: int
file	The model file's name inside the body-model directory. TYPE: str

Attributes

family instance-attribute

family: BodyModelFamily

variant instance-attribute

variant: str

num_body_joints instance-attribute

num_body_joints: int

file instance-attribute

file: str

Methods:

SkinnedBodyModel

SkinnedBodyModel(rest_vertices: ndarray, faces: ndarray, skin_weights: ndarray, parents: ndarray, bind_rot: ndarray, bind_pos: ndarray, joint_names: list[str], posedirs: ndarray | None = None)

A linear-blend-skinning rig with a fixed identity.

PARAMETER	DESCRIPTION
rest_vertices	(V, 3) mesh in the bind pose (model frame). TYPE: ndarray
faces	(F, 3). TYPE: ndarray
skin_weights	(V, J), rows summing to one. TYPE: ndarray
parents	(J,) parent joint index, -1 for the root. TYPE: ndarray
bind_rot	(J, 3, 3) joint world orientations in the bind pose. TYPE: ndarray
bind_pos	(J, 3) joint world positions in the bind pose. TYPE: ndarray
joint_names	Joint names. TYPE: list[str]
posedirs	(V, 3, 9 * (J - 1)) pose-corrective blend shapes, driven by (R_local - I) of every non-root joint (SMPL convention). TYPE: ndarray DEFAULT: None

Source code in eden/datasets/utils/body_model.py

def __init__(
    self,
    rest_vertices: np.ndarray,
    faces: np.ndarray,
    skin_weights: np.ndarray,
    parents: np.ndarray,
    bind_rot: np.ndarray,
    bind_pos: np.ndarray,
    joint_names: list[str],
    posedirs: np.ndarray | None = None,
) -> None:
    self.rest_vertices = np.asarray(rest_vertices, dtype=np.float64)
    self.faces = np.asarray(faces, dtype=np.int64)
    self.skin_weights = np.asarray(skin_weights, dtype=np.float64)
    self.parents = np.asarray(parents, dtype=np.int64)
    self.bind_rot = np.asarray(bind_rot, dtype=np.float64)
    self.bind_pos = np.asarray(bind_pos, dtype=np.float64)
    self.joint_names = list(joint_names)
    self.posedirs = None if posedirs is None else np.asarray(posedirs, dtype=np.float64)
    n = len(self.parents)
    for j in range(n):
        if self.parents[j] >= j:
            raise ValueError(
                f"SkinnedBodyModel: joint {j} ({self.joint_names[j]!r}) has parent {self.parents[j]}, but parents "
                f"must precede their children so a single forward pass poses the chain."
            )
    # Bone offsets in the parent's bind frame.
    self._offsets = np.zeros((n, 3))
    for j in range(1, n):
        p = self.parents[j]
        self._offsets[j] = self.bind_rot[p].T @ (self.bind_pos[j] - self.bind_pos[p])

Attributes

rest_vertices instance-attribute

rest_vertices = np.asarray(rest_vertices, dtype=np.float64)

faces instance-attribute

faces = np.asarray(faces, dtype=np.int64)

skin_weights instance-attribute

skin_weights = np.asarray(skin_weights, dtype=np.float64)

parents instance-attribute

parents = np.asarray(parents, dtype=np.int64)

bind_rot instance-attribute

bind_rot = np.asarray(bind_rot, dtype=np.float64)

bind_pos instance-attribute

bind_pos = np.asarray(bind_pos, dtype=np.float64)

joint_names instance-attribute

joint_names = list(joint_names)

posedirs instance-attribute

posedirs = None if posedirs is None else np.asarray(posedirs, dtype=np.float64)

Methods:

forward

forward(local_rot: ndarray, root_pos: ndarray, chunk: int = 256) -> tuple[ndarray, ndarray, ndarray]

Pose the rig.

PARAMETER	DESCRIPTION
local_rot	(T, J, 3, 3) joint rotations relative to the bind pose, each in the joint's own bind frame: W_j = W_parent @ (B_parent^T B_j) @ local_j and W_root = B_root @ local_root. TYPE: ndarray
root_pos	(T, 3) root joint world position. TYPE: ndarray
chunk	Frames skinned per batch (bounds memory). TYPE: int DEFAULT: 256

RETURNS	DESCRIPTION
vertices	(T, V, 3). TYPE: ndarray
joint_pos	(T, J, 3). TYPE: ndarray
joint_rot	(T, J, 3, 3) joint world orientations. TYPE: ndarray

Source code in eden/datasets/utils/body_model.py

def forward(
    self, local_rot: np.ndarray, root_pos: np.ndarray, chunk: int = 256
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Pose the rig.

    Parameters
    ----------
    local_rot : np.ndarray
        ``(T, J, 3, 3)`` joint rotations relative to the bind pose, each in the joint's own bind
        frame: ``W_j = W_parent @ (B_parent^T B_j) @ local_j`` and ``W_root = B_root @ local_root``.
    root_pos : np.ndarray
        ``(T, 3)`` root joint world position.
    chunk : int
        Frames skinned per batch (bounds memory).

    Returns
    -------
    vertices : np.ndarray
        ``(T, V, 3)``.
    joint_pos : np.ndarray
        ``(T, J, 3)``.
    joint_rot : np.ndarray
        ``(T, J, 3, 3)`` joint world orientations.
    """
    local_rot = np.asarray(local_rot, dtype=np.float64)
    n_frames, n_joints = local_rot.shape[:2]
    if n_joints != len(self.parents):
        raise ValueError(
            f"SkinnedBodyModel.forward: local_rot has {n_joints} joints per frame; the rig has {len(self.parents)} "
            f"({self.joint_names[:4]}...), root included."
        )
    joint_rot = np.zeros((n_frames, n_joints, 3, 3))
    joint_pos = np.zeros((n_frames, n_joints, 3))
    for j in range(n_joints):
        p = self.parents[j]
        if p < 0:
            joint_rot[:, j] = self.bind_rot[j] @ local_rot[:, j]
            joint_pos[:, j] = root_pos
        else:
            joint_rot[:, j] = joint_rot[:, p] @ (self.bind_rot[p].T @ self.bind_rot[j]) @ local_rot[:, j]
            joint_pos[:, j] = joint_pos[:, p] + np.einsum("tab,b->ta", joint_rot[:, p], self._offsets[j])
    pose_feature = None
    if self.posedirs is not None:
        pose_feature = (local_rot[:, 1:] - np.eye(3)).reshape(n_frames, -1)
    vertices = self.skin(joint_rot, joint_pos, pose_feature, chunk)
    return vertices, joint_pos, joint_rot

skin

skin(joint_rot: ndarray, joint_pos: ndarray, pose_feature: ndarray | None = None, chunk: int = 256) -> ndarray

Skin the mesh to given joint world transforms.

PARAMETER	DESCRIPTION
joint_rot	(T, J, 3, 3) joint world orientations. TYPE: ndarray
joint_pos	(T, J, 3) joint world positions. TYPE: ndarray
pose_feature	(T, 9 * (J - 1)) pose-corrective drive (see posedirs); None skips the correctives. TYPE: ndarray DEFAULT: None
chunk	Frames skinned per batch. TYPE: int DEFAULT: 256

RETURNS	DESCRIPTION
ndarray	(T, V, 3) float32 vertices.

Source code in eden/datasets/utils/body_model.py

def skin(
    self, joint_rot: np.ndarray, joint_pos: np.ndarray, pose_feature: np.ndarray | None = None, chunk: int = 256
) -> np.ndarray:
    """Skin the mesh to given joint world transforms.

    Parameters
    ----------
    joint_rot : np.ndarray
        ``(T, J, 3, 3)`` joint world orientations.
    joint_pos : np.ndarray
        ``(T, J, 3)`` joint world positions.
    pose_feature : np.ndarray, optional
        ``(T, 9 * (J - 1))`` pose-corrective drive (see ``posedirs``); None skips the correctives.
    chunk : int
        Frames skinned per batch.

    Returns
    -------
    np.ndarray
        ``(T, V, 3)`` float32 vertices.
    """
    n_frames, n_verts = len(joint_rot), len(self.rest_vertices)
    vertices = np.zeros((n_frames, n_verts, 3), dtype=np.float32)
    # Skinning transform per joint: bind frame -> posed frame, packed (rotation 9, translation 3) so one matrix
    # product blends both: einsums over (T, V, J) without a contraction path ran 2.4x slower than these BLAS calls.
    skin_rot = joint_rot @ np.transpose(self.bind_rot, (0, 2, 1))[None]
    skin_trans = joint_pos - np.einsum("tjab,jb->tja", skin_rot, self.bind_pos)
    packed = np.concatenate([skin_rot.reshape(n_frames, -1, 9), skin_trans], axis=2)  # (T, J, 12)
    posedirs = None if self.posedirs is None else self.posedirs.reshape(n_verts * 3, -1)
    for s in range(0, n_frames, chunk):
        e = min(s + chunk, n_frames)
        rest = np.broadcast_to(self.rest_vertices, (e - s, n_verts, 3))
        if pose_feature is not None and posedirs is not None:
            rest = rest + (pose_feature[s:e] @ posedirs.T).reshape(e - s, n_verts, 3)
        blended = (self.skin_weights @ packed[s:e]).reshape(e - s, n_verts, 12)
        rot = blended[..., :9].reshape(e - s, n_verts, 3, 3)
        vertices[s:e] = np.einsum("tvab,tvb->tva", rot, rest, optimize=True) + blended[..., 9:]
    return vertices

Functions:

mano_model_name

mano_model_name(side: str) -> str

The :data:BODY_MODELS name of the MANO hand on side (right / left).

Source code in eden/datasets/utils/body_model.py

def mano_model_name(side: str) -> str:
    """The :data:`BODY_MODELS` name of the MANO hand on ``side`` (``right`` / ``left``)."""
    if side not in ("right", "left"):
        raise ValueError(f"A MANO hand is 'right' or 'left', got {side!r}.")
    return f"mano_{side}"

body_model_file

body_model_file(model_dir: str | Path | None, body_model: str) -> Path

The file of body_model in model_dir (the first of :func:model_file_names present); raises naming where to get it when either is missing.

Source code in eden/datasets/utils/body_model.py

def body_model_file(model_dir: str | Path | None, body_model: str) -> Path:
    """The file of ``body_model`` in ``model_dir`` (the first of :func:`model_file_names` present); raises naming
    where to get it when either is missing.
    """
    spec = _spec(body_model)
    if model_dir is None:
        raise ValueError(
            f"Reading {body_model} needs its model file ({spec.file}), but no body-model directory was given. Set "
            f"body_model_dir on the dataset options to the directory holding it."
        )
    names = model_file_names(body_model)
    for name in names:
        if (Path(model_dir) / name).exists():
            return Path(model_dir) / name
    where = (
        "register at mano.is.tue.mpg.de and download 'Models & Code' (mano_v1_2.zip); its models/ directory holds it"
        if spec.family == "mano"
        else f"download {spec.family.upper()} from its project page"
    )
    raise FileNotFoundError(
        f"{Path(model_dir) / spec.file} missing (also looked for {', '.join(n for n in names if n != spec.file)}). "
        f"The body-model files are licence-gated and never shipped: {where}."
    )

body_model_arrays

body_model_arrays(path: str | Path) -> dict[str, ndarray]

Every array of a body-model file (.npz or .pkl), as plain numpy in the file's own dtypes.

A .pkl that pickles chumpy objects (MANO, SMPL, SMPL-H) is read without chumpy (:class:~eden.utils.chumpy_pickle.ChumpyFreeUnpickler); its sparse J_regressor is densified. The last two files read stay cached (a MANO pair; an SMPL-X file holds ~170 MB), keyed by modification time and size so a replaced file is read again. The returned dict is shared: copy before writing into it.

Source code in eden/datasets/utils/body_model.py

def body_model_arrays(path: str | Path) -> dict[str, np.ndarray]:
    """Every array of a body-model file (``.npz`` or ``.pkl``), as plain numpy in the file's own dtypes.

    A ``.pkl`` that pickles chumpy objects (MANO, SMPL, SMPL-H) is read without chumpy
    (:class:`~eden.utils.chumpy_pickle.ChumpyFreeUnpickler`); its sparse ``J_regressor`` is densified. The last two
    files read stay cached (a MANO pair; an SMPL-X file holds ~170 MB), keyed by modification time and size so a
    replaced file is read again. The returned dict is shared: copy before writing into it.
    """
    stat = Path(path).stat()
    return _load_arrays(str(path), stat.st_mtime_ns, stat.st_size)

shape_components

shape_components(arrays: dict[str, ndarray], family: BodyModelFamily) -> int

How many of shapedirs' blend shapes are body shape (betas) rather than facial expression.

SMPL-X stacks the shape blend shapes, then the expression ones: 300 + 100 in v1.1, 10 + 10 in v1.0 (the smplx package's split). The other families' shapedirs are all shape.

Source code in eden/datasets/utils/body_model.py

def shape_components(arrays: dict[str, np.ndarray], family: BodyModelFamily) -> int:
    """How many of ``shapedirs``' blend shapes are body shape (``betas``) rather than facial expression.

    SMPL-X stacks the shape blend shapes, then the expression ones: 300 + 100 in v1.1, 10 + 10 in v1.0 (the
    ``smplx`` package's split). The other families' ``shapedirs`` are all shape.
    """
    n = arrays["shapedirs"].shape[-1]
    if family != "smplx":
        return n
    return 300 if n >= 400 else n // 2

skinned_body_model

skinned_body_model(path: str | Path, family: BodyModelFamily, betas: ndarray, joint_names: list[str] | None = None, v_template: ndarray | None = None, posedirs: ndarray | None = None) -> SkinnedBodyModel

The rig of one body-model file shaped by betas: SMPL-style bind pose (identity joint frames).

PARAMETER	DESCRIPTION
path	The model file (.npz or .pkl). TYPE: str | Path
family	Its family; decides how many of shapedirs are shape (:func:shape_components). TYPE: BodyModelFamily
betas	Shape coefficients, up to the model's shape components; coefficients past them must be zero. TYPE: ndarray
joint_names	Names for the joints (default joint_0 ...). TYPE: list[str] DEFAULT: None
v_template	A personal template mesh replacing the model's mean shape (GRAB ships each subject as one); betas still apply on top of it. TYPE: ndarray DEFAULT: None
posedirs	The file's posedirs already in float64, shared by every rig built from it: an SMPL-X file stores them in float32, and each rig converting its own copied 122 MB. TYPE: ndarray DEFAULT: None

RETURNS	DESCRIPTION
SkinnedBodyModel

Source code in eden/datasets/utils/body_model.py

def skinned_body_model(
    path: str | Path,
    family: BodyModelFamily,
    betas: np.ndarray,
    joint_names: list[str] | None = None,
    v_template: np.ndarray | None = None,
    posedirs: np.ndarray | None = None,
) -> SkinnedBodyModel:
    """The rig of one body-model file shaped by ``betas``: SMPL-style bind pose (identity joint frames).

    Parameters
    ----------
    path : str | Path
        The model file (``.npz`` or ``.pkl``).
    family : BodyModelFamily
        Its family; decides how many of ``shapedirs`` are shape (:func:`shape_components`).
    betas : np.ndarray
        Shape coefficients, up to the model's shape components; coefficients past them must be zero.
    joint_names : list[str], optional
        Names for the joints (default ``joint_0`` ...).
    v_template : np.ndarray, optional
        A personal template mesh replacing the model's mean shape (GRAB ships each subject as one); ``betas`` still
        apply on top of it.
    posedirs : np.ndarray, optional
        The file's ``posedirs`` already in float64, shared by every rig built from it: an SMPL-X file stores them in
        float32, and each rig converting its own copied 122 MB.

    Returns
    -------
    SkinnedBodyModel
    """
    data = body_model_arrays(str(path))
    betas = np.asarray(betas, dtype=np.float64).reshape(-1)
    n_model_shape = shape_components(data, family)
    if np.any(betas[n_model_shape:] != 0.0):
        raise ValueError(
            f"skinned_body_model: {len(betas)} betas, but {path} has {n_model_shape} shape components (shapedirs "
            f"{data['shapedirs'].shape}; any others are expression blend shapes). For SMPL-X use the v1.1 model files "
            f"(300 shape components); truncating the betas would change the body the motion was fitted to."
        )
    n_shape = min(len(betas), n_model_shape)
    template = data["v_template"] if v_template is None else np.asarray(v_template)
    if template.shape != data["v_template"].shape:
        raise ValueError(
            f"skinned_body_model: v_template has shape {template.shape}; {path} expects {data['v_template'].shape} "
            f"(one row per model vertex)."
        )
    shaped = template.astype(np.float64) + np.einsum(
        "vck,k->vc", data["shapedirs"][:, :, :n_shape].astype(np.float64), betas[:n_shape]
    )
    joints = data["J_regressor"].astype(np.float64) @ shaped
    parents = data["kintree_table"][0].astype(np.int64)
    parents[0] = -1
    return SkinnedBodyModel(
        rest_vertices=shaped,
        faces=data["f"].astype(np.int64),
        skin_weights=data["weights"],
        parents=parents,
        bind_rot=np.tile(np.eye(3), (len(parents), 1, 1)),
        bind_pos=joints,
        joint_names=list(joint_names) if joint_names is not None else [f"joint_{j}" for j in range(len(parents))],
        posedirs=data["posedirs"] if posedirs is None else posedirs,
    )

mano_pose_from_release

mano_pose_from_release(pose: ndarray, model_dir: str | Path | None, side: str, *, num_pca_comps: int | None, flat_hand_mean: bool) -> ndarray

A release's MANO finger pose -> the 45 axis-angles relative to the flat hand that body_pose_quat holds.

PARAMETER	DESCRIPTION
pose	(T, num_pca_comps) PCA coefficients, or (T, 45) axis-angles when num_pca_comps is None. TYPE: ndarray
model_dir	Directory holding MANO_<SIDE>.pkl; read only when the pose is PCA or relative to the mean hand. TYPE: str | Path | None
side	right or left. TYPE: str
num_pca_comps	The release's PCA size, or None for axis-angles. All 45 components are still PCA: smplx turns PCA off when asked for 45 and would read the coefficients as axis-angles, which is why this exists. TYPE: int | None
flat_hand_mean	Whether the release's pose is relative to the flat hand (MANO's flat_hand_mean=True) rather than the mean hand. TYPE: bool

RETURNS	DESCRIPTION
ndarray	(T, 45) axis-angles, relative to the flat hand.

Source code in eden/datasets/utils/body_model.py

def mano_pose_from_release(
    pose: np.ndarray, model_dir: str | Path | None, side: str, *, num_pca_comps: int | None, flat_hand_mean: bool
) -> np.ndarray:
    """A release's MANO finger pose -> the 45 axis-angles relative to the flat hand that ``body_pose_quat`` holds.

    Parameters
    ----------
    pose : np.ndarray
        (T, num_pca_comps) PCA coefficients, or (T, 45) axis-angles when ``num_pca_comps`` is None.
    model_dir : str | Path | None
        Directory holding ``MANO_<SIDE>.pkl``; read only when the pose is PCA or relative to the mean hand.
    side : str
        ``right`` or ``left``.
    num_pca_comps : int | None
        The release's PCA size, or None for axis-angles. All 45 components are still PCA: ``smplx`` turns PCA off
        when asked for 45 and would read the coefficients as axis-angles, which is why this exists.
    flat_hand_mean : bool
        Whether the release's pose is relative to the flat hand (MANO's ``flat_hand_mean=True``) rather than the
        mean hand.

    Returns
    -------
    np.ndarray
        (T, 45) axis-angles, relative to the flat hand.
    """
    pose = np.asarray(pose, dtype=np.float64)
    if num_pca_comps is None and flat_hand_mean:
        return pose
    arrays = _mano_arrays(str(body_model_file(model_dir, mano_model_name(side))))
    if num_pca_comps is not None:
        if pose.shape[-1] != num_pca_comps:
            raise ValueError(f"Expected {num_pca_comps} MANO PCA coefficients per frame, got {pose.shape[-1]}.")
        pose = pose @ arrays["hands_components"][:num_pca_comps]
    if not flat_hand_mean:
        pose = pose + arrays["hands_mean"]
    return pose

axis_angles_to_body_pose_quat

axis_angles_to_body_pose_quat(pose: ndarray) -> Tensor

(T, 3 * J) axis-angles -> (T, J, 4) wxyz, float32.

Source code in eden/datasets/utils/body_model.py

def axis_angles_to_body_pose_quat(pose: np.ndarray) -> torch.Tensor:
    """(T, 3 * J) axis-angles -> (T, J, 4) wxyz, float32."""
    pose = torch.as_tensor(np.asarray(pose), dtype=torch.float64)
    return quat_from_axis_angle(pose.reshape(len(pose), -1, 3)).to(torch.float32)

body_model_forward

body_model_forward(body_model: str, model_dir: str | Path | None, root_pos: Tensor, root_quat: Tensor, body_pose_quat: Tensor, body_shape: Tensor) -> tuple[Tensor, Tensor]

Joints and vertices of a body-model sequence, its root joint placed on root_pos.

PARAMETER	DESCRIPTION
body_model	A :data:BODY_MODELS name. TYPE: str
model_dir	Directory holding the model file. TYPE: str | Path | None
root_pos	(T, 3) root joint positions and (T, 4) wxyz global orientations: the sequence's base_pos_w / base_quat_w. TYPE: Tensor
root_quat	(T, 3) root joint positions and (T, 4) wxyz global orientations: the sequence's base_pos_w / base_quat_w. TYPE: Tensor
body_pose_quat	(T, J, 4) and (T, B): the sequence's body-model fields. The pose is relative to the rest pose (MANO: the flat hand), and the jaw, eyes and expression of SMPL-X are held at rest. TYPE: Tensor
body_shape	(T, J, 4) and (T, B): the sequence's body-model fields. The pose is relative to the rest pose (MANO: the flat hand), and the jaw, eyes and expression of SMPL-X are held at rest. TYPE: Tensor

RETURNS	DESCRIPTION
tuple[Tensor, Tensor]	(T, J + 1, 3) joints, root first, and (T, V, 3) vertices, float32 on root_pos's device.

Source code in eden/datasets/utils/body_model.py

def body_model_forward(
    body_model: str,
    model_dir: str | Path | None,
    root_pos: torch.Tensor,
    root_quat: torch.Tensor,
    body_pose_quat: torch.Tensor,
    body_shape: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
    """Joints and vertices of a body-model sequence, its root joint placed on ``root_pos``.

    Parameters
    ----------
    body_model : str
        A :data:`BODY_MODELS` name.
    model_dir : str | Path | None
        Directory holding the model file.
    root_pos, root_quat : torch.Tensor
        (T, 3) root joint positions and (T, 4) wxyz global orientations: the sequence's ``base_pos_w`` /
        ``base_quat_w``.
    body_pose_quat, body_shape : torch.Tensor
        (T, J, 4) and (T, B): the sequence's body-model fields. The pose is relative to the rest pose (MANO: the flat
        hand), and the jaw, eyes and expression of SMPL-X are held at rest.

    Returns
    -------
    tuple[torch.Tensor, torch.Tensor]
        (T, J + 1, 3) joints, root first, and (T, V, 3) vertices, float32 on ``root_pos``'s device.
    """
    spec = _spec(body_model)
    if body_pose_quat.shape[1] != spec.num_body_joints:
        raise ValueError(
            f"body_model_forward: {body_model} has {spec.num_body_joints} joints below the root, but body_pose_quat "
            f"has {body_pose_quat.shape[1]} per frame (shape {tuple(body_pose_quat.shape)})."
        )
    path = str(body_model_file(model_dir, body_model))
    device = root_pos.device
    quats = torch.cat([root_quat[:, None], body_pose_quat], dim=1).detach().to("cpu", torch.float64).numpy()
    local_rot = quat_to_R(quats)
    root = root_pos.detach().to("cpu", torch.float64).numpy()
    shapes = body_shape.detach().to("cpu", torch.float64).numpy()
    joints = np.zeros((len(root), spec.num_body_joints + 1, 3))
    verts = None
    # One rig per distinct shape: a sequence's betas are almost always constant, but the fields allow per-frame ones.
    # The float64 pose correctives are made once and shared by every rig.
    posedirs = body_model_arrays(path)["posedirs"].astype(np.float64)
    unique, which = np.unique(shapes, axis=0, return_inverse=True)
    for k, betas in enumerate(unique):
        rows = np.flatnonzero(which.reshape(-1) == k)
        rig = skinned_body_model(path, spec.family, betas, posedirs=posedirs)
        v, j, _ = rig.forward(local_rot[rows], root[rows])
        if verts is None:
            verts = np.zeros((len(root), v.shape[1], 3), dtype=np.float32)
        joints[rows], verts[rows] = j, v
    as_tensor = functools.partial(torch.as_tensor, dtype=torch.float32, device=device)
    return as_tensor(joints), as_tensor(verts)

mano_root_offset

mano_root_offset(model_dir: str | Path | None, side: str, betas: ndarray) -> ndarray

(…, 3) position of MANO's root joint at zero translation: the offset from a release's transl to the wrist.

The root joint is regressed from the shaped template alone, so the offset depends on betas and not on the pose: wrist = transl + mano_root_offset(betas).

Source code in eden/datasets/utils/body_model.py

def mano_root_offset(model_dir: str | Path | None, side: str, betas: np.ndarray) -> np.ndarray:
    """(…, 3) position of MANO's root joint at zero translation: the offset from a release's ``transl`` to the wrist.

    The root joint is regressed from the shaped template alone, so the offset depends on ``betas`` and not on the
    pose: ``wrist = transl + mano_root_offset(betas)``.
    """
    path = body_model_file(model_dir, mano_model_name(side))
    data = body_model_arrays(path)
    betas = np.asarray(betas, dtype=np.float64)
    if betas.shape[-1] > data["shapedirs"].shape[-1]:
        raise ValueError(
            f"mano_root_offset: {betas.shape[-1]} betas per hand, but {path} has {data['shapedirs'].shape[-1]} shape "
            f"components; pass the first {data['shapedirs'].shape[-1]}."
        )
    regressor = data["J_regressor"][0].astype(np.float64)  # the wrist's row
    base = regressor @ data["v_template"].astype(np.float64)
    dirs = np.einsum("v,vck->kc", regressor, data["shapedirs"][..., : betas.shape[-1]].astype(np.float64))
    return base + betas @ dirs

mano_vertex_links

mano_vertex_links(model_dir: str | Path | None, side: str) -> ndarray

(778,) the MANO joint (0-15) each vertex follows most: the argmax of its skinning weights.

Source code in eden/datasets/utils/body_model.py

def mano_vertex_links(model_dir: str | Path | None, side: str) -> np.ndarray:
    """(778,) the MANO joint (0-15) each vertex follows most: the argmax of its skinning weights."""
    return body_model_arrays(str(body_model_file(model_dir, mano_model_name(side))))["weights"].argmax(1)

mano_keypoints

mano_keypoints(joints: Tensor, vertices: Tensor) -> Tensor

(T, 21, 3) keypoints in :data:~eden.datasets.constants.MANO_KEYPOINTS_NAME order from MANO's (T, 16, 3) joints and (T, 778, 3) vertices: the 16 joints in MANO's order, then the :data:MANO_TIP_VERTICES tips.

Source code in eden/datasets/utils/body_model.py

def mano_keypoints(joints: torch.Tensor, vertices: torch.Tensor) -> torch.Tensor:
    """(T, 21, 3) keypoints in :data:`~eden.datasets.constants.MANO_KEYPOINTS_NAME` order from MANO's (T, 16, 3)
    joints and (T, 778, 3) vertices: the 16 joints in MANO's order, then the :data:`MANO_TIP_VERTICES` tips.
    """
    if joints.shape[-2] != 16:
        raise ValueError(f"Expected MANO's 16 joints, got {joints.shape[-2]}.")
    return torch.cat([joints, vertices[..., list(MANO_TIP_VERTICES), :]], dim=-2)

BVH parsing

BVH (Biovision Hierarchy) file parser with forward kinematics.

Classes

BvhJoint dataclass

BvhJoint(name: str, offset: ndarray, channels: list[str] = list(), children: list[BvhJoint] = list(), parent: BvhJoint | None = None)

Attributes

name instance-attribute

name: str

offset instance-attribute

offset: ndarray

channels class-attribute instance-attribute

channels: list[str] = field(default_factory=list)

children class-attribute instance-attribute

children: list[BvhJoint] = field(default_factory=list)

parent class-attribute instance-attribute

parent: BvhJoint | None = field(default=None, repr=False)

Methods:

BvhData dataclass

BvhData(root: BvhJoint, joints: list[BvhJoint], joint_names: list[str], framerate: float, num_frames: int, frame_data: ndarray)

Attributes

root instance-attribute

root: BvhJoint

joints instance-attribute

joints: list[BvhJoint]

joint_names instance-attribute

joint_names: list[str]

framerate instance-attribute

framerate: float

num_frames instance-attribute

num_frames: int

frame_data instance-attribute

frame_data: ndarray

Methods:

Functions:

parse_bvh

parse_bvh(filepath: str) -> BvhData

Parse a BVH file into structured data.

PARAMETER	DESCRIPTION
filepath	Path to the BVH file. TYPE: str

RETURNS	DESCRIPTION
BvhData	Parsed skeleton hierarchy and motion data.

Source code in eden/datasets/utils/bvh.py

def parse_bvh(filepath: str) -> BvhData:
    """
    Parse a BVH file into structured data.

    Parameters
    ----------
    filepath : str
        Path to the BVH file.

    Returns
    -------
    BvhData
        Parsed skeleton hierarchy and motion data.
    """
    with open(filepath, encoding="utf-8") as f:
        lines = [line.strip() for line in f.readlines()]

    joints: list[BvhJoint] = []
    stack: list[BvhJoint] = []
    i = 0

    # Skip to HIERARCHY
    while i < len(lines) and not lines[i].startswith("HIERARCHY"):
        i += 1
    i += 1

    while i < len(lines):
        line = lines[i]

        if line.startswith("ROOT") or line.startswith("JOINT"):
            name = line.split(None, 1)[1] if len(line.split(None, 1)) > 1 else ""
            # Strip the mocap take prefix (e.g. "AA_50_T2:Hips" -> "Hips")
            if ":" in name:
                name = name.split(":")[-1]

            joint = BvhJoint(name=name, offset=np.zeros(3))
            if stack:
                joint.parent = stack[-1]
                stack[-1].children.append(joint)
            joints.append(joint)

        elif line.startswith("End Site"):
            # Skip the entire End Site block
            i += 1
            depth = 0
            while i < len(lines):
                if lines[i].startswith("{"):
                    depth += 1
                elif lines[i].startswith("}"):
                    depth -= 1
                    if depth == 0:
                        break
                i += 1

        elif line == "{":
            stack.append(joints[-1])

        elif line.startswith("}"):
            if stack:
                stack.pop()

        elif line.startswith("OFFSET"):
            parts = line.split()
            joints[-1].offset = np.array([float(parts[1]), float(parts[2]), float(parts[3])])

        elif line.startswith("CHANNELS"):
            parts = line.split()
            n = int(parts[1])
            joints[-1].channels = parts[2 : 2 + n]

        elif line.startswith("MOTION"):
            i += 1
            break

        i += 1

    # Parse Frames and Frame Time
    num_frames = 0
    frame_time = 0.0
    while i < len(lines):
        line = lines[i]
        if line.startswith("Frames"):
            num_frames = int(line.split(":")[1].strip())
        elif line.startswith("Frame Time"):
            frame_time = float(line.split(":")[1].strip())
            i += 1
            break
        i += 1

    # Read motion data
    total_channels = sum(len(j.channels) for j in joints)
    frame_data = np.zeros((num_frames, total_channels))
    for f_idx in range(num_frames):
        if i + f_idx >= len(lines) or not lines[i + f_idx]:
            break
        values = lines[i + f_idx].split()
        frame_data[f_idx] = [float(v) for v in values[:total_channels]]

    framerate = 1.0 / frame_time if frame_time > 0 else 120.0

    return BvhData(
        root=joints[0],
        joints=joints,
        joint_names=[j.name for j in joints],
        framerate=framerate,
        num_frames=num_frames,
        frame_data=frame_data,
    )

forward_kinematics

forward_kinematics(bvh: BvhData) -> tuple[ndarray, ndarray]

Compute global joint positions and orientations via forward kinematics.

PARAMETER	DESCRIPTION
bvh	Parsed BVH data. TYPE: BvhData

RETURNS	DESCRIPTION
positions	Global positions of all joints. TYPE: (ndarray, shape(num_frames, num_joints, 3))
quaternions	Global orientations in wxyz format. TYPE: (ndarray, shape(num_frames, num_joints, 4))

Source code in eden/datasets/utils/bvh.py

def forward_kinematics(
    bvh: BvhData,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Compute global joint positions and orientations via forward kinematics.

    Parameters
    ----------
    bvh : BvhData
        Parsed BVH data.

    Returns
    -------
    positions : np.ndarray, shape (num_frames, num_joints, 3)
        Global positions of all joints.
    quaternions : np.ndarray, shape (num_frames, num_joints, 4)
        Global orientations in wxyz format.
    """
    positions, rot_matrices = forward_kinematics_matrices(bvh)
    flat_quat_xyzw = Rotation.from_matrix(rot_matrices.reshape(-1, 3, 3)).as_quat()
    quaternions = xyzw_to_wxyz(flat_quat_xyzw).reshape(*rot_matrices.shape[:2], 4)
    return positions, quaternions

forward_kinematics_matrices

forward_kinematics_matrices(bvh: BvhData) -> tuple[ndarray, ndarray]

Global joint positions and rotation matrices: :func:forward_kinematics without the quaternion conversion.

A joint's rotation channels compose in the order they are listed (Zrotation Yrotation Xrotation is Rz @ Ry @ Rx), and its position channels, on any joint, replace the matching axes of its offset: a SOMA-rig BVH gives every joint all six channels.

PARAMETER	DESCRIPTION
bvh	Parsed BVH data. TYPE: BvhData

RETURNS	DESCRIPTION
positions	Global positions of all joints. TYPE: (ndarray, shape(num_frames, num_joints, 3))
rot_matrices	Global orientations. TYPE: (ndarray, shape(num_frames, num_joints, 3, 3))

Source code in eden/datasets/utils/bvh.py

def forward_kinematics_matrices(bvh: BvhData) -> tuple[np.ndarray, np.ndarray]:
    """
    Global joint positions and rotation matrices: :func:`forward_kinematics` without the quaternion conversion.

    A joint's rotation channels compose in the order they are listed (``Zrotation Yrotation Xrotation`` is
    ``Rz @ Ry @ Rx``), and its position channels, on any joint, replace the matching axes of its offset: a
    SOMA-rig BVH gives every joint all six channels.

    Parameters
    ----------
    bvh : BvhData
        Parsed BVH data.

    Returns
    -------
    positions : np.ndarray, shape (num_frames, num_joints, 3)
        Global positions of all joints.
    rot_matrices : np.ndarray, shape (num_frames, num_joints, 3, 3)
        Global orientations.
    """
    num_frames = bvh.num_frames
    num_joints = len(bvh.joints)

    positions = np.zeros((num_frames, num_joints, 3))
    rot_matrices = np.zeros((num_frames, num_joints, 3, 3))
    joint_id_map = {id(j): i for i, j in enumerate(bvh.joints)}

    # Process each joint (parse order guarantees parent before child)
    ch_off = 0
    for j_idx, joint in enumerate(bvh.joints):
        local_pos = np.tile(np.asarray(joint.offset, dtype=np.float64), (num_frames, 1))
        rot_order = []
        rot_angles = []
        for ch_name in joint.channels:
            ch_lower = ch_name.lower()
            axis = ch_lower[0]
            if "position" in ch_lower:
                local_pos[:, "xyz".index(axis)] = bvh.frame_data[:, ch_off]
            elif "rotation" in ch_lower:
                rot_order.append(axis)
                rot_angles.append(bvh.frame_data[:, ch_off])
            ch_off += 1

        # Uppercase axes: scipy's intrinsic convention, i.e. the channels' matrices multiplied in listed order
        if rot_order:
            angles = np.stack(rot_angles, axis=-1)  # (num_frames, n_rot)
            local_rot = Rotation.from_euler("".join(rot_order).upper(), angles, degrees=True).as_matrix()
        else:
            local_rot = np.tile(np.eye(3), (num_frames, 1, 1))

        if joint.parent is None:
            positions[:, j_idx] = local_pos
            rot_matrices[:, j_idx] = local_rot
        else:
            p_idx = joint_id_map[id(joint.parent)]
            parent_rot = rot_matrices[:, p_idx]
            positions[:, j_idx] = positions[:, p_idx] + np.einsum("fij,fj->fi", parent_rot, local_pos)
            rot_matrices[:, j_idx] = np.einsum("fij,fjk->fik", parent_rot, local_rot)

    return positions, rot_matrices

Keypoint names

Shared constants for motion datasets.

Attributes

MANO_KEYPOINTS_NAME module-attribute

MANO_KEYPOINTS_NAME: tuple[str, ...] = ('wrist', 'index1', 'index2', 'index3', 'middle1', 'middle2', 'middle3', 'pinky1', 'pinky2', 'pinky3', 'ring1', 'ring2', 'ring3', 'thumb1', 'thumb2', 'thumb3', 'thumb_tip', 'index_tip', 'middle_tip', 'ring_tip', 'pinky_tip')

SMPLH_KEYPOINTS_NAME module-attribute

SMPLH_KEYPOINTS_NAME = ['Pelvis', 'L_Hip', 'L_Knee', 'L_Ankle', 'L_Toe', 'R_Hip', 'R_Knee', 'R_Ankle', 'R_Toe', 'Torso', 'Spine', 'Chest', 'Neck', 'Head', 'L_Thorax', 'L_Shoulder', 'L_Elbow', 'L_Wrist', 'L_Index1', 'L_Index2', 'L_Index3', 'L_Middle1', 'L_Middle2', 'L_Middle3', 'L_Pinky1', 'L_Pinky2', 'L_Pinky3', 'L_Ring1', 'L_Ring2', 'L_Ring3', 'L_Thumb1', 'L_Thumb2', 'L_Thumb3', 'R_Thorax', 'R_Shoulder', 'R_Elbow', 'R_Wrist', 'R_Index1', 'R_Index2', 'R_Index3', 'R_Middle1', 'R_Middle2', 'R_Middle3', 'R_Pinky1', 'R_Pinky2', 'R_Pinky3', 'R_Ring1', 'R_Ring2', 'R_Ring3', 'R_Thumb1', 'R_Thumb2', 'R_Thumb3']

SMPLX_KEYPOINTS_NAME module-attribute

SMPLX_KEYPOINTS_NAME = ['Pelvis', 'L_Hip', 'R_Hip', 'Spine1', 'L_Knee', 'R_Knee', 'Spine2', 'L_Ankle', 'R_Ankle', 'Spine3', 'L_Foot', 'R_Foot', 'Neck', 'L_Collar', 'R_Collar', 'Head', 'L_Shoulder', 'R_Shoulder', 'L_Elbow', 'R_Elbow', 'L_Wrist', 'R_Wrist']

MOCAP_KEYPOINTS_NAME module-attribute

MOCAP_KEYPOINTS_NAME = ['Hips', 'Spine', 'Spine1', 'Neck', 'Head', 'LeftShoulder', 'LeftArm', 'LeftForeArm', 'LeftHand', 'LeftHandThumb1', 'LeftHandThumb2', 'LeftHandThumb3', 'LeftHandIndex1', 'LeftHandIndex2', 'LeftHandIndex3', 'LeftHandMiddle1', 'LeftHandMiddle2', 'LeftHandMiddle3', 'LeftHandRing1', 'LeftHandRing2', 'LeftHandRing3', 'LeftHandPinky1', 'LeftHandPinky2', 'LeftHandPinky3', 'RightShoulder', 'RightArm', 'RightForeArm', 'RightHand', 'RightHandThumb1', 'RightHandThumb2', 'RightHandThumb3', 'RightHandIndex1', 'RightHandIndex2', 'RightHandIndex3', 'RightHandMiddle1', 'RightHandMiddle2', 'RightHandMiddle3', 'RightHandRing1', 'RightHandRing2', 'RightHandRing3', 'RightHandPinky1', 'RightHandPinky2', 'RightHandPinky3', 'LeftUpLeg', 'LeftLeg', 'LeftFoot', 'LeftToeBase', 'RightUpLeg', 'RightLeg', 'RightFoot', 'RightToeBase', 'LeftFootMod', 'RightFootMod']

LAFAN1_KEYPOINTS_NAME module-attribute

LAFAN1_KEYPOINTS_NAME = ['Hips', 'RightUpLeg', 'RightLeg', 'RightFoot', 'RightToeBase', 'LeftUpLeg', 'LeftLeg', 'LeftFoot', 'LeftToeBase', 'Spine', 'Spine1', 'Spine2', 'Neck', 'Head', 'RightShoulder', 'RightArm', 'RightForeArm', 'RightHand', 'LeftShoulder', 'LeftArm', 'LeftForeArm', 'LeftHand']

See also

• Motion datasets guide
• Motion reference
• Retargeting extensions