Core Utilities¶

The machinery the rest of Eden is built on: registries, the configurable mixin that binds an options class to its runtime object, parameter descriptions, multi-GPU helpers, logging and run identity, profiling, and small string and formatting helpers.

Package entry point¶

import eden isolates one GPU per DDP rank before CUDA is initialized, so import Eden before genesis and torch. It imports neither itself; eden.init() does, and every process that builds a scene calls it. The built-in term modules and the options namespaces (eden.rewards, eden.robots, eden.scenes, ...) load lazily on first use.

Eden package entry point: per-rank GPU isolation for DDP and runtime initialization.

Importing :mod:eden performs per-rank GPU isolation before CUDA is initialized, so Eden must be imported before genesis and torch (importing those first would pin every rank to the same device). :func:init then configures the torch backends, determinism, and Genesis logging.

import eden imports neither genesis nor torch: between them they would be ~2.8 s of the import, paid by eden --help, by eden list, and by every script that only reads a config. Both are imported by :func:init, which every process that builds a scene calls anyway. The isolation runs at the top of this module, so any import order that starts with import eden is correct by construction -- but a module that wants genesis must import it itself rather than relying on Eden having pulled it in.

Common entry points:

:func:init — set up the runtime (TF32, determinism, per-rank logging).
:func:get_log_dir (a.k.a. eden.log_dir) — this run's log directory.
eden.envs.base.RLEnvBase.from_config — build a vectorized environment from a config.
:data:eden.tasks.TASK_REGISTRY — look up built-in task configs by name.

Functions:¶

init ¶

init(backend=_DEFAULT_BACKEND, initialize_genesis: bool = True, debug: bool = False, logging_level: int | None = None, logger_verbose_time: bool = False, log_root_path: str | None = None, run_name: str = '', allow_tf32: bool = True, deterministic: bool = False, performance_mode: bool = False, precision: Literal['32', '64'] = '32', eps: float = 1e-08, *, logger_theme: Literal['auto', 'human', 'agent'] | None = None)

Initialize the Eden environment.

PARAMETER	DESCRIPTION
`backend`	The backend device to use for the simulation (`gs.cpu`, `gs.gpu`, ...). `None` -- the default -- means `gs.cpu`, resolved inside this call: spelling it in the signature would evaluate `gs.cpu` at import and drag `import genesis` back onto `import eden`. DEFAULT: `_DEFAULT_BACKEND`
`initialize_genesis`	Whether to initialize the Genesis here if not already initialized. TYPE: `bool` DEFAULT: `True`
`debug`	Whether to run in debug mode. TYPE: `bool` DEFAULT: `False`
`logging_level`	The logging level to use. TYPE: `int \| None` DEFAULT: `None`
`logger_verbose_time`	Whether to log the time in the logger. TYPE: `bool` DEFAULT: `False`
`log_root_path`	The root path to use for the logs. Defaults to `./logs`. The run directory under it is resolved here but only created on the first :func:`get_log_dir` / `eden.log_dir` access. TYPE: `str \| None` DEFAULT: `None`
`run_name`	Optional human-facing run name folded into the log-dir basename as `<timestamp>-<run_name>-<id>` (sanitized). Because rsl_rl derives the wandb run name from the log-dir basename, this names the run in the directory and in wandb; empty (the default) yields `<timestamp>-<id>`. TYPE: `str` DEFAULT: `''`
`allow_tf32`	Whether to use TF32 precision for faster computation on Ampere+ GPUs. TYPE: `bool` DEFAULT: `True`
`deterministic`	Whether to use deterministic algorithms for reproducibility. TYPE: `bool` DEFAULT: `False`
`performance_mode`	Whether to use performance mode for the simulation. TYPE: `bool` DEFAULT: `False`
`precision`	Floating-point precision passed to `gs.init`. Default '32'. Some solvers (e.g. the SAP coupler) require '64' for numerical stability. TYPE: `Literal['32', '64']` DEFAULT: `'32'`
`eps`	Global numerical epsilon passed to `gs.init`. Default 1e-8 (chosen for rigid-solver stability). Lower it (e.g. 1e-12) when a coupler uses tighter convergence tolerances than 1e-8 — the SAP coupler's default 1e-10 tolerances are corrupted by a coarser eps, which manifests as contacts exploding. TYPE: `float` DEFAULT: `1e-08`
`logger_theme`	Keyword-only. How Eden (and Genesis) render their console output. `'agent'` — the default — drops every ANSI escape and the greeting box, for output that will be read out of a file, a CI log or a model's context window, which is where most of it goes. `'human'` is the colored, box-greeted terminal rendering, and is now something you ask for: pass it here, or set `EDEN_LOG_THEME=human` to get it for a whole process tree (exported empty counts as unset). `'auto'` is the middle setting — `'human'` on a TTY and `'agent'` otherwise, honoring `NO_COLOR` / `FORCE_COLOR`. Passing this exports `EDEN_LOG_THEME`, because the theme has consumers this call never touches: `colorize`, the argparse help formatters, and every subprocess a run spawns. The request is exported rather than the resolution, so an explicit `'auto'` stays a per-process TTY decision downstream instead of an inherited one. This also sets Genesis' theme (`genesis.styles.style.theme`: `THEME.raw` for `'agent'`, `THEME.dark` for `'human'`) so the escapes are gone from the whole run's output rather than from Eden's share of it — including under `initialize_genesis=False`. Assign `genesis.styles.style.theme` after this call to choose a different Genesis theme. Nothing sets it before this call — genesis is off `import eden`'s path entirely — so a config `repr`, which Genesis colorizes through it, prints plain until then under either theme. Eden's own rendering does not read it at all. TYPE: `Literal['auto', 'human', 'agent'] \| None` DEFAULT: `None`

Source code in eden/__init__.py

def init(
    backend=_DEFAULT_BACKEND,
    initialize_genesis: bool = True,
    debug: bool = False,
    logging_level: int | None = None,
    logger_verbose_time: bool = False,
    log_root_path: str | None = None,
    run_name: str = "",
    allow_tf32: bool = True,
    deterministic: bool = False,
    performance_mode: bool = False,
    precision: Literal["32", "64"] = "32",
    eps: float = 1e-8,
    # Keyword-only, and last: inserted into the middle of this list it would silently re-bind
    # every caller that passes `log_root_path` and beyond positionally -- `eden.init(gs.gpu,
    # True, False, None, False, "./runs")` would hand "./runs" to `logger_theme` and raise on a
    # perfectly good log root. The `*` keeps the same trap from being set for whatever is added
    # next.
    *,
    logger_theme: Literal["auto", "human", "agent"] | None = None,
):
    """
    Initialize the Eden environment.

    Parameters
    ----------
    backend: genesis backend constant | None
        The backend device to use for the simulation (``gs.cpu``, ``gs.gpu``, ...). ``None`` --
        the default -- means ``gs.cpu``, resolved inside this call: spelling it in the signature
        would evaluate ``gs.cpu`` at import and drag ``import genesis`` back onto
        ``import eden``.
    initialize_genesis: bool
        Whether to initialize the Genesis here if not already initialized.
    debug: bool
        Whether to run in debug mode.
    logging_level: int | str | None
        The logging level to use.
    logger_verbose_time: bool
        Whether to log the time in the logger.
    log_root_path: str | None
        The root path to use for the logs.
        Defaults to ``./logs``. The run directory under it is *resolved* here but only
        created on the first :func:`get_log_dir` / ``eden.log_dir`` access.
    run_name: str
        Optional human-facing run name folded into the log-dir basename as
        ``<timestamp>-<run_name>-<id>`` (sanitized). Because rsl_rl derives the wandb run
        name from the log-dir basename, this names the run in the directory and in wandb;
        empty (the default) yields ``<timestamp>-<id>``.
    allow_tf32: bool
        Whether to use TF32 precision for faster computation on Ampere+ GPUs.
    deterministic: bool
        Whether to use deterministic algorithms for reproducibility.
    performance_mode: bool
        Whether to use performance mode for the simulation.
    precision: {'32', '64'}
        Floating-point precision passed to ``gs.init``. Default '32'.
        Some solvers (e.g. the SAP coupler) require '64' for numerical stability.
    eps: float
        Global numerical epsilon passed to ``gs.init``. Default 1e-8 (chosen for
        rigid-solver stability). Lower it (e.g. 1e-12) when a coupler uses tighter
        convergence tolerances than 1e-8 — the SAP coupler's default 1e-10 tolerances
        are corrupted by a coarser eps, which manifests as contacts exploding.
    logger_theme: {'auto', 'human', 'agent'} | None
        Keyword-only. How Eden (and Genesis) render their console output. ``'agent'`` — **the
        default** — drops every ANSI escape and the greeting box, for output that will be read
        out of a file, a CI log or a model's context window, which is where most of it goes.
        ``'human'`` is the colored, box-greeted terminal rendering, and is now something you ask
        for: pass it here, or set ``EDEN_LOG_THEME=human`` to get it for a whole process tree
        (exported empty counts as unset). ``'auto'`` is the middle setting — ``'human'`` on a
        TTY and ``'agent'`` otherwise, honoring ``NO_COLOR`` / ``FORCE_COLOR``.
        Passing this exports ``EDEN_LOG_THEME``, because the theme has consumers this call never
        touches: ``colorize``, the argparse help formatters, and every subprocess a run spawns.
        The *request* is exported rather than the resolution, so an explicit ``'auto'`` stays a
        per-process TTY decision downstream instead of an inherited one.
        This also sets Genesis' theme (``genesis.styles.style.theme``: ``THEME.raw`` for
        ``'agent'``, ``THEME.dark`` for ``'human'``) so the escapes are gone from the whole run's
        output rather than from Eden's share of it — including under
        ``initialize_genesis=False``. Assign ``genesis.styles.style.theme`` after this call to
        choose a different Genesis theme. Nothing sets it before this call — genesis is off
        ``import eden``'s path entirely — so a config ``repr``, which Genesis colorizes
        through it, prints plain until then under either theme. Eden's own rendering does not
        read it at all.
    """
    # Declared up front: the double-init warning below reads the module ``logger`` before the
    # rebind further down reassigns it, and Python forbids using a name ahead of its ``global``.
    global logger

    # Deferred so `import eden` costs neither -- this is the first point in a process
    # that genuinely needs them, and `_isolate_gpu_for_ddp()` ran at module import, long before.
    import genesis as gs
    import torch

    from eden.utils import distributed
    from eden.utils.torch import configure_torch_backends

    if backend is _DEFAULT_BACKEND:
        backend = gs.cpu

    configure_torch_backends(allow_tf32=allow_tf32, deterministic=deterministic)

    # Resolved before anything can log or print a colorized repr. Eden's theme drives Genesis'
    # palette too: "agent" means no escapes in the *run's* output, not just in Eden's share of it.
    if logger_theme is not None:
        # Exported, because the theme has consumers this call never touches: `colorize` and the
        # argparse help formatters read EDEN_LOG_THEME, as does every subprocess (a DDP worker,
        # a spawned viewer); otherwise `--help` and every colorized repr would stay plain. The
        # *request* goes in the variable, not the resolution, so an explicit "auto" stays a
        # per-process TTY decision downstream.
        os.environ[LOG_THEME_ENV_VAR] = logger_theme
    theme = resolve_theme(logger_theme)
    gs_theme = _GENESIS_THEMES[theme]

    # Suppress Genesis logging on non-main ranks to avoid noisy output.
    gs_log_level = logging_level
    if distributed.is_distributed() and not distributed.is_main_process():
        gs_log_level = "warning"

    if initialize_genesis:
        if not gs._initialized:
            gs.init(
                backend=backend,
                precision=precision,
                theme=gs_theme,
                logging_level=gs_log_level,
                # NOTE: this is for the numerical stability of the simulation
                eps=eps,
                # A placeholder, not the run's seed: gs.init() seeds numpy/torch, and a
                # constant here alone would make an unseeded training run come out *identical*
                # for every configured seed. `eden.utils.random.apply_run_seed` overrides it from runner_options.seed
                # before the env is built. Varies by *local* rank so a DDP run that never
                # reaches that call still differs between ranks on a node (two nodes' local
                # rank 0 share it — another reason not to treat this as the run's seed).
                seed=int(os.environ.get("LOCAL_RANK", 0)),
                performance_mode=performance_mode,
                use_deterministic_algorithms=deterministic,
            )
        else:
            # Loud, not a quiet info: re-initializing Eden/Genesis in one already-initialized
            # process is a common cause of the opaque CUDA "unknown error" that costs training
            # attempts. Genesis is not re-initialized (the guard above holds), but
            # the rest of init() below still re-runs — NCCL setup and, notably, the run's log
            # directory is re-resolved to a new timestamp, orphaning the earlier one. Warn so the
            # double-init is visible; the fix is to call eden.init() exactly once per process, or
            # pass initialize_genesis=False when a second init() is deliberate (e.g. reconfiguring
            # logging or the log directory).
            logger.warning(
                "eden.init() called again in a process where Genesis is already initialized. "
                "Genesis will not be re-initialized, but re-running init() in one process is a "
                "common source of an opaque CUDA 'unknown error' and it re-resolves the log "
                "directory. Call eden.init() exactly once per process (one process per run)."
            )
    # Also for the paths the gs.init() above did not take -- Genesis already initialized, or
    # initialize_genesis=False. Unconditional, so one rule holds everywhere: after eden.init()
    # Genesis' theme is the one Eden's asked for. A caller who wants Genesis' "light" assigns
    # `genesis.styles.style.theme` afterwards.
    from genesis.styles import THEME, style

    style.theme = THEME[gs_theme]

    # Up to https://github.com/Genesis-Embodied-AI/Genesis/pull/3431, `gs.init()` also set torch's default device
    # and dtype. Eden does not restore the device: torch implements it as a Python TorchFunctionMode that every torch
    # call is dispatched through, 1.7-2.0 ms per velocity-tracking control step and 4 ms on BeyondMimic, so a tensor
    # built without `device=` is a CPU tensor, as in plain torch. The dtype is a C-level setting with no per-call
    # cost, and it is what makes a dtype-less `torch.zeros(n)` match Genesis' own tensors under `precision="64"` --
    # where a float32 buffer would silently downcast an assignment and make an indexed write or a matmul against a
    # Genesis tensor raise -- so Eden sets it on every path that leaves Genesis initialized.
    if gs._initialized:
        torch.set_default_dtype(gs.tc_float)

    # ── Initialize NCCL right after Genesis, before scene building ──
    #   gs.init() → torch.cuda.set_device(0) → dist.init_process_group("nccl")
    # Doing it here (before any scene.build) ensures the CUDA context is
    # in a clean state for NCCL communicator creation. Device 0 is this rank's GPU:
    # `_isolate_gpu_for_ddp` narrowed CUDA_VISIBLE_DEVICES to it.
    if distributed.is_distributed():
        torch.cuda.set_device(0)
        distributed.setup("nccl")

    if logging_level is None:
        logging_level = _stdlib_logging.DEBUG if debug else _stdlib_logging.INFO
    logger = Logger(logging_level, logger_verbose_time, theme)
    # Apply Eden's level to the third-party stdlib "genesis" logger here (runtime setup), not in
    # Logger.__init__ — a bare `import eden` must not mutate third-party logging state. This also
    # covers a Genesis initialized before eden.init(). Genesis' logger stops propagating to the root
    # logger on its own since https://github.com/Genesis-Embodied-AI/Genesis/pull/3427.
    _stdlib_logging.getLogger("genesis").setLevel(logging_level)

    # Only show greeting and version info on the main process.
    if distributed.is_main_process():
        from eden.utils.misc import _display_greeting, format_provenance, get_eden_provenance, get_package_provenance

        # The box is three lines of terminal decoration, and off a TTY it cannot even be sized
        # (os.get_terminal_size raises, and the 80-column fallback is the piped case) — so the
        # agent theme skips it rather than rendering it blind. The version line below stays:
        # it is provenance, not decoration.
        if theme != "agent":
            _display_greeting(logger.INFO_length)

        eden_version = format_provenance(get_eden_provenance())
        genesis_version = format_provenance(get_package_provenance("genesis-world"))
        logger.info(f"🍎 Eden ver: ~~<{eden_version}>~~, 🌏 Genesis ver: ~~<{genesis_version}>~~.")

    if distributed.is_distributed():
        logger.info(
            f"🚀 DDP enabled: rank {distributed.get_global_rank()}/{distributed.get_world_size()}"
            f" (local_rank={distributed.get_local_rank()})"
        )

    # NOTE: resolve the log directory (leaf name ``<timestamp>-<run id>``; see
    # _new_run_dirname for why the random id, not the timestamp, carries uniqueness).
    # Under DDP, rank 0 generates the name and broadcasts it so every rank agrees on one
    # path. The directory is *not* created here; get_log_dir() creates it on first use.
    global _log_dir, _run_timestamp, _run_id
    base_log_dir = log_root_path or "./logs"
    if distributed.is_distributed():
        # _new_run_dirname() also sets _run_timestamp/_run_id (left-to-right eval), so we
        # broadcast all three together — rank 0's parts, not each rank re-deriving them.
        payload = (
            [_new_run_dirname(run_name), _run_timestamp, _run_id]
            if distributed.is_main_process()
            else [None, None, None]
        )
        torch.distributed.broadcast_object_list(payload, src=0)
        run_dirname, _run_timestamp, _run_id = payload
    else:
        run_dirname = _new_run_dirname(run_name)
    _log_dir = os.path.abspath(os.path.join(base_log_dir, run_dirname))
    logger.info(f"💽 Log directory: ~~<{_log_dir}>~~ (created on first use)")

get_log_dir ¶

get_log_dir() -> str

Return this run's log directory, creating it if it does not exist yet.

eden.log_dir is an alias for this call, so f"{en.log_dir}/video.mp4" and os.path.join(en.log_dir, "config.json") keep working unchanged.

Creation is deferred to the first access rather than done in :func:init so that a process which never logs anything — an evaluation, export, or viewer script, all of which must still call :func:init before they can build a scene — leaves no empty logs/<timestamp>/ directory behind. That matters beyond tidiness: the usual way to find a trained run is to take the newest directory under logs/, and an eagerly-created directory makes that resolve to the current process's own empty one.

RETURNS	DESCRIPTION
`str`	Absolute path to the log directory.

RAISES	DESCRIPTION
`RuntimeError`	If :func:`init` has not been called yet, so no log directory is resolved.

Source code in eden/__init__.py

def get_log_dir() -> str:
    """Return this run's log directory, creating it if it does not exist yet.

    ``eden.log_dir`` is an alias for this call, so ``f"{en.log_dir}/video.mp4"`` and
    ``os.path.join(en.log_dir, "config.json")`` keep working unchanged.

    Creation is deferred to the first access rather than done in :func:`init` so that a
    process which never logs anything — an evaluation, export, or viewer script, all of
    which must still call :func:`init` before they can build a scene — leaves no empty
    ``logs/<timestamp>/`` directory behind. That matters beyond tidiness: the usual way
    to find a trained run is to take the newest directory under ``logs/``, and an
    eagerly-created directory makes that resolve to the current process's own empty one.

    Returns
    -------
    str
        Absolute path to the log directory.

    Raises
    ------
    RuntimeError
        If :func:`init` has not been called yet, so no log directory is resolved.
    """
    if _log_dir is None:
        raise RuntimeError("Eden is not initialized: call `eden.init()` before using the log directory.")
    # Unconditional (not main-rank-only): with lazy creation any rank may be the first to
    # write, and makedirs(exist_ok=True) is safe to race.
    os.makedirs(_log_dir, exist_ok=True)
    return _log_dir

get_run_id ¶

get_run_id() -> str | None

Return this run's base-36 id (the uniqueness primitive of the log-dir basename).

None before :func:init has resolved a run. Provenance metadata only — a caller that needs a directory should use :func:get_log_dir / eden.log_dir.

Source code in eden/__init__.py

def get_run_id() -> str | None:
    """Return this run's base-36 id (the uniqueness primitive of the log-dir basename).

    ``None`` before :func:`init` has resolved a run. Provenance metadata only — a caller
    that needs a directory should use :func:`get_log_dir` / ``eden.log_dir``.
    """
    return _run_id

get_run_timestamp ¶

get_run_timestamp() -> str | None

Return this run's creation timestamp (the get_now() prefix of the basename).

None before :func:init has resolved a run.

Source code in eden/__init__.py

def get_run_timestamp() -> str | None:
    """Return this run's creation timestamp (the ``get_now()`` prefix of the basename).

    ``None`` before :func:`init` has resolved a run.
    """
    return _run_timestamp

Registries¶

A Registry maps a name to an object. Keys default to to_snake_case(obj.__name__). A TermRegistry (every term kind) takes no name=: a term's key is always its own snake-cased name, because that is what .configure() stamps into options.name and what the manager looks it up by. A lookup of a missing key raises KeyError with the closest match and the available keys.

Generic Registry plus term/variant registries for pluggable components.

A :class:Registry maps name -> object so user code can register custom terms, tasks, robots, etc. Keys default to to_snake_case(obj.__name__); pass name= to choose a different one. Two naming caveats: keys starting with _ emit a warning, and class names with a digit suffix (e.g. Foo1) do not gain an underscore before the digit.

Configurables (terms, file handlers, dataset modifiers, terrain generators — anything whose .configure() stamps options.name and is resolved back out of its registry by that name) are the exception: their key is to_snake_case(__name__), so :class:TermRegistry takes no name= at all, and :class:Registry refuses one that would alias a configurable. See :func:_key_is_derived_elsewhere.

Classes¶

Registry ¶

Registry(name: str, populated_by: Sequence[str] = ())

Bases: Iterable[tuple[str, Any]]

Registry providing a name -> object mapping to support third-party custom modules.

Keys default to to_snake_case(obj.__name__); pass name= to choose a different one. That override is for objects looked up only by their key — assets, robots, tasks. A :class:~eden.utils.common.ConfigurableMixin is not one of them: its key is also what configure() stamps into options.name, so name= on one is refused rather than honoured (see :func:_key_is_derived_elsewhere), and :class:TermRegistry — whose entries are all configurable — takes no name= at all.

To create a registry (e.g. a robot registry):

ROBOT_REGISTRY = Registry('ROBOT')

To register an object:

@ROBOT_REGISTRY.register()
class Go1():
    ...

Or:

ROBOT_REGISTRY.register(Go1)

PARAMETER	DESCRIPTION
`name`	The name of this registry, quoted in every message it raises. TYPE: `str`
`populated_by`	Modules whose import-time `@register` decorators fill this registry, imported on the first read (:meth:`_populate`) rather than by whoever imported the registry. A registry whose entries are defined in the module that declares it needs none. TYPE: `Sequence[str]` DEFAULT: `()`

Source code in eden/utils/registry.py

def __init__(self, name: str, populated_by: Sequence[str] = ()) -> None:
    self._name: str = name
    self._obj_map: dict[str, Any] = {}
    self._populated_by: tuple[str, ...] = tuple(populated_by)
    self._populated: bool = False
    #: Thread ids currently inside :meth:`_populate`. A set, not a flag, because the only
    #: thread that may skip the import is one already doing it — see :meth:`_populate`.
    self._populating: set[int] = set()
    #: The exception a failed :meth:`_populate` raised, replayed to every later reader
    #: rather than retried — see :meth:`_populate`.
    self._populate_error: BaseException | None = None
    #: That exception's traceback as first caught, kept separately because re-raising the
    #: exception rebinds its ``__traceback__`` to a longer chain each time — see
    #: :meth:`_populate`.
    self._populate_error_tb: TracebackType | None = None

Methods:¶

register ¶

register(obj: Any = None, name: str | None = None) -> Any

Register the given object, usable as either a decorator or a direct call.

Both forms return the registered object, so the direct call can be chained and the decorator leaves the decorated name bound to it. See the class docstring for usage.

Source code in eden/utils/registry.py

def register(self, obj: Any = None, name: str | None = None) -> Any:
    """Register the given object, usable as either a decorator or a direct call.

    Both forms return the registered object, so the direct call can be chained and the
    decorator leaves the decorated name bound to it. See the class docstring for usage.
    """
    if obj is None:
        # used as a decorator
        def deco(func_or_class: Any) -> Any:
            self._do_register(_derive_name(func_or_class, name, self._name), func_or_class)
            return func_or_class

        return deco

    # used as a function call
    self._do_register(_derive_name(obj, name, self._name), obj)
    return obj

get ¶

get(name: str) -> Any

Source code in eden/utils/registry.py

def get(self, name: str) -> Any:
    self._populate()
    ret = self._obj_map.get(name)
    if ret is None:
        available = ", ".join(sorted(self._obj_map.keys())) or "none"
        # The closest key first: with ~200 registered terms the roster alone makes the reader
        # (or an agent) diff two long lists to find the one character that differs.
        #
        # Guarded on ``str``, because ``name`` is annotated as one but is not checked to be:
        # ``ManagerTermOptions.name`` is ``str | None`` and ``ManagerBase._build_term`` passes
        # it straight through, so an options entry that never got a name reaches here as
        # ``None``. ``difflib.get_close_matches`` iterates it and raises ``TypeError:
        # 'NoneType' object is not iterable`` from inside the message — losing the readable
        # KeyError that names the missing key and lists the roster, which for a config
        # missing a ``name=`` is the whole diagnosis. Message-building code must not raise
        # (docs/guides/error-messages.md); a non-string key simply gets no suggestion, and
        # the ``'{name}'`` below renders it fine whatever it is.
        close = difflib.get_close_matches(name, self._obj_map, n=1) if isinstance(name, str) else []
        suggestion = f" Did you mean '{close[0]}'?" if close else ""
        raise KeyError(
            f"No object named '{name}' found in '{self._name}' registry.{suggestion}"
            f"{self._lookup_hint(name)} Available options: {available}"
        )
    return ret

keys ¶

keys()

Source code in eden/utils/registry.py

def keys(self):
    self._populate()
    return self._obj_map.keys()

TermRegistry ¶

TermRegistry(name: str, term_class: Type[ManagerTermBase], term_fn_wrapper_class: Type[ManagerTermFuncWrapperBase] | None = None, populated_by: Sequence[str] = ())

Bases: Registry

Registry for manager terms.

Every entry is keyed by its own to_snake_case(__name__) and :meth:register takes no name=, so two terms can only collide by being named the same thing.

PARAMETER	DESCRIPTION
`name`	The name of the registry. TYPE: `str`
`term_class`	The class of the manager term. TYPE: `Type[ManagerTermBase]`
`term_fn_wrapper_class`	The class of the manager term function wrapper. TYPE: `Type[ManagerTermFuncWrapperBase] \| None` DEFAULT: `None`
`populated_by`	Modules whose import-time registrations fill this registry — see :meth:`Registry._populate`. For a manager kind this is its `eden.managers.terms.` module, which `import eden` does not import. TYPE:* `Sequence[str]` DEFAULT: `()`

Source code in eden/utils/registry.py

def __init__(
    self,
    name: str,
    term_class: Type[ManagerTermBase],
    term_fn_wrapper_class: Type[ManagerTermFuncWrapperBase] | None = None,
    populated_by: Sequence[str] = (),
) -> None:
    super().__init__(name, populated_by)
    self._term_class = term_class
    self._term_fn_wrapper_class = term_fn_wrapper_class
    if term_fn_wrapper_class is not None:
        # Let ``configure(func=...)`` verify the function is registered here (see
        # ConfigurableFuncWrapperMixin._build_config_dict). Stamped on the wrapper *class*, so a
        # wrapper shared by two registries would make one registry's terms validate against the
        # other's — refuse it (each registry needs its own func-wrapper subclass). Re-stamping the
        # same registry (e.g. a re-import) is a harmless no-op.
        existing = getattr(term_fn_wrapper_class, "_term_registry", None)
        if existing is not None and existing is not self:
            raise ValueError(
                f"func-wrapper class {term_fn_wrapper_class.__name__} is already bound to registry "
                f"'{existing._name}'; a wrapper class must belong to a single registry (give "
                f"registry '{name}' its own {term_fn_wrapper_class.__name__} subclass)."
            )
        term_fn_wrapper_class._term_registry = self
    # Stamp this registry onto the kind's term-options class so ManagerTermOptions can
    # validate parameter names at construction time (see
    # ManagerTermOptions._validate_param_names). First stamp wins: a second registry built
    # over the same options class (e.g. a user registry whose terms subclass a built-in
    # term) leaves the original binding intact — names it can't resolve simply skip the
    # construction-time check and fall back to the build-time one. ``__dict__`` (not
    # getattr) so a subclassed options class can still be bound to its own registry.
    for term_or_wrapper in filter(None, (term_class, term_fn_wrapper_class)):
        options_class = getattr(term_or_wrapper, "_options_class_", None)
        if options_class is not None and options_class.__dict__.get("_term_registry_") is None:
            options_class._term_registry_ = self

Attributes¶

cli_kind `property` ¶

cli_kind: str

What eden terms --kind calls this registry: REWARD_TERM -> reward.

Derived rather than declared so a new registry cannot ship with the wrong one, and pinned against :data:eden.managers.catalog.KIND_NAMES by tests/utils/test_registry.py, beside tests/managers/test_catalog.py asserting the catalog covers every registry in the tree.

term_class `property` ¶

term_class: Type[ManagerTermBase]

The base class every class term in this registry subclasses.

func_wrapper_class `property` ¶

func_wrapper_class: Type[ManagerTermFuncWrapperBase] | None

The wrapper a function term of this kind is configured through, or None.

None means the kind takes class terms only — :meth:_register_term refuses a plain function there. Exposed (with :attr:term_class) so :mod:eden.managers.catalog can read the per-kind fields every term carries off the base, rather than reaching into _-private attributes to describe a registry it does not own.

Methods:¶

catalog_hint ¶

catalog_hint(term: str | None = None) -> str

One line pointing at the eden terms command that answers what the error left open.

The roster :meth:Registry.get prints answers "which terms are there?" but not "which one does what I need?" or "what does this one take?", and a reader who cannot answer those reaches for the 92 kB generated catalog or greps the installed source. Both wordings live here so the kind is derived once and the two cannot disagree. One line, no newline: the lookup caller raises KeyError, whose __str__ is repr(args[0]).

PARAMETER	DESCRIPTION
`term`	A registered term to point at, when there is one. `None` — the failed-lookup case — gets the search form with a `<text>` placeholder instead: the key that just failed is by definition not in the catalog, so grepping for it would print nothing and read as a dead end. A typo is covered by the `Did you mean` :meth:`Registry.get` adds. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	A sentence beginning with `Run` and no trailing space.

Source code in eden/utils/registry.py

def catalog_hint(self, term: str | None = None) -> str:
    """One line pointing at the ``eden terms`` command that answers what the error left open.

    The roster :meth:`Registry.get` prints answers "which terms are there?" but not "which one
    does what I need?" or "what does this one take?", and a reader who cannot answer those
    reaches for the 92 kB generated catalog or greps the installed source. Both
    wordings live here so the kind is derived once and the two cannot disagree. One line, no
    newline: the lookup caller raises ``KeyError``, whose ``__str__`` is ``repr(args[0])``.

    Parameters
    ----------
    term : str | None
        A *registered* term to point at, when there is one. ``None`` — the failed-lookup case
        — gets the search form with a ``<text>`` placeholder instead: the key that just failed
        is by definition not in the catalog, so grepping for it would print nothing and read
        as a dead end. A typo is covered by the ``Did you mean`` :meth:`Registry.get` adds.

    Returns
    -------
    str
        A sentence beginning with ``Run`` and no trailing space.
    """
    if term is None:
        return (
            f"Run `eden terms --kind {self.cli_kind} --grep <text>` to search these by name, parameter "
            f"and summary, or `eden terms --kind {self.cli_kind}` to list them all with their parameters."
        )
    return f"Run `eden terms --kind {self.cli_kind} --grep {term}` for its parameters, defaults and summary."

registered_class ¶

registered_class(key: str) -> type

The class the term registered under key is built as, which is where its defaults live.

A class term is stored as its class — the defaults-carrying subclass when register was given keyword defaults (:meth:_with_defaults). A function term is stored as partial(wrapper, func=...), and wrapper is the per-registration subclass carrying its keyword defaults, not the kind's shared *FuncWrapper: that subclass is the only place @METRIC_TERM_REGISTRY.register(direction=MetricDirection.LIB) on a function is recorded, so a default read off the shared wrapper is the kind's, not the term's.

PARAMETER	DESCRIPTION
`key`	A registered key. An unknown one raises the :meth:`Registry.get` `KeyError`. TYPE: `str`

RETURNS	DESCRIPTION
`type`	The registered class term, or the wrapper class of a function term.

Source code in eden/utils/registry.py

def registered_class(self, key: str) -> type:
    """The class the term registered under ``key`` is built as, which is where its defaults live.

    A class term is stored as its class — the defaults-carrying subclass when ``register`` was
    given keyword defaults (:meth:`_with_defaults`). A function term is stored as
    ``partial(wrapper, func=...)``, and ``wrapper`` is the per-registration subclass carrying
    *its* keyword defaults, not the kind's shared ``*FuncWrapper``: that subclass is the only
    place ``@METRIC_TERM_REGISTRY.register(direction=MetricDirection.LIB)`` on a function is
    recorded, so a default read off the shared wrapper is the kind's, not the term's.

    Parameters
    ----------
    key : str
        A registered key. An unknown one raises the :meth:`Registry.get` ``KeyError``.

    Returns
    -------
    type
        The registered class term, or the wrapper class of a function term.
    """
    registered = self.get(key)
    return registered.func if isinstance(registered, partial) else registered

register ¶

register(obj: Any = None, *, name: str | None = None, **term_defaults: Any) -> Any

Register a class term, a function term, or a func-wrapper subclass.

Takes no name: a term's key is always its own to_snake_case(__name__) — see :data:_WHY_THE_KEY_IS_THE_NAME, the cause this refusal shares with :func:_derive_name. name is declared only so it can be refused — **term_defaults would otherwise swallow it and quietly set a class attribute called name on the term instead.

Source code in eden/utils/registry.py

def register(self, obj: Any = None, *, name: str | None = None, **term_defaults: Any) -> Any:
    """Register a class term, a function term, or a func-wrapper subclass.

    Takes no ``name``: a term's key is always its own ``to_snake_case(__name__)`` — see
    :data:`_WHY_THE_KEY_IS_THE_NAME`, the cause this refusal shares with :func:`_derive_name`.
    ``name`` is declared only so it can be *refused* — ``**term_defaults`` would otherwise
    swallow it and quietly set a class attribute called ``name`` on the term instead.
    """
    if name is not None:
        raise ValueError(
            f"{self._name}_REGISTRY.register() takes no name=: a term's key is always its own "
            f"name, because {_WHY_THE_KEY_IS_THE_NAME}. Passing '{name}' cannot change that. "
            f"Drop it, and rename the term itself if you want a different key."
        )
    if obj is None:
        # used as a decorator
        def deco(func_or_class: Any) -> Any:
            key = _derive_name(func_or_class, None, self._name)
            return self._register_term(func_or_class, key, term_defaults)

        return deco

    # used as a function call
    return self._register_term(obj, _derive_name(obj, None, self._name), term_defaults)

VariantRegistry ¶

VariantRegistry(name: str, populated_by: Sequence[str] = ())

Bases: Registry

Registry for asset families with multiple mesh-file variants.

A single decorated class declares its source directory name and the list of available variant ids. The decorator stamps _base_name and _default_variant (the first id in variant_list) onto the class, merges family metadata into cls.metadata, and registers one partial(cls, variant=N) callable per variant under the key f"{base_name}_{N}". The decorated class itself is also returned so it stays importable for type-level use.

Asset families (objaverse, ...) follow this contract by declaring a base options class that defines _base_name/_default_variant as ClassVar and an __init__ (or model_post_init) that materializes file from those + the requested variant.

Calling register() with neither base_name nor variant_list falls through to :class:Registry's plain-name registration, so procedural primitives (boxes, spheres) can share the same registry as variant assets.

Source code in eden/utils/registry.py

def __init__(self, name: str, populated_by: Sequence[str] = ()) -> None:
    self._name: str = name
    self._obj_map: dict[str, Any] = {}
    self._populated_by: tuple[str, ...] = tuple(populated_by)
    self._populated: bool = False
    #: Thread ids currently inside :meth:`_populate`. A set, not a flag, because the only
    #: thread that may skip the import is one already doing it — see :meth:`_populate`.
    self._populating: set[int] = set()
    #: The exception a failed :meth:`_populate` raised, replayed to every later reader
    #: rather than retried — see :meth:`_populate`.
    self._populate_error: BaseException | None = None
    #: That exception's traceback as first caught, kept separately because re-raising the
    #: exception rebinds its ``__traceback__`` to a longer chain each time — see
    #: :meth:`_populate`.
    self._populate_error_tb: TracebackType | None = None

Methods:¶

register ¶

register(obj: Any = None, *, name: str | None = None, base_name: str = '', variant_list: Sequence[int] | None = None) -> Any

Source code in eden/utils/registry.py

def register(
    self,
    obj: Any = None,
    *,
    name: str | None = None,
    base_name: str = "",
    variant_list: Sequence[int] | None = None,
) -> Any:
    if base_name == "" and variant_list is None:
        return super().register(obj, name=name)

    if not variant_list:
        raise ValueError(f"variant_list is required when base_name is set (base_name={base_name!r})")

    def _register(options):
        options._base_name = base_name
        options._default_variant = variant_list[0]
        merged_metadata = dict(getattr(options, "metadata", None) or {})
        merged_metadata.update(
            {
                "base_name": base_name,
                "variant_list": list(variant_list),
                "num_variants": len(variant_list),
                "variants_name": [f"{base_name}_{variant}" for variant in variant_list],
            }
        )
        options.metadata = merged_metadata
        for variant in variant_list:
            self._do_register(f"{base_name}_{variant}", partial(options, variant=variant))
        return options

    if obj is None:
        return _register
    return _register(obj)

Functions:¶

module_is_initializing ¶

module_is_initializing(module: Any) -> bool

Whether module is still executing its own body — a partial import, not a finished one.

importlib.import_module returns immediately, with the half-built module, when the calling thread is already importing it; only a different thread blocks until the body is done. This is the flag the import system itself reads to tell those apart (importlib._bootstrap._find_and_load), and two callers need it for the same reason — to know whether what they just got back is complete enough to draw a conclusion from: :meth:Registry._populate before it memoizes _populated, and ManagerTermOptions._resolve_term_registry before it reports "no registry claims this class".

ModuleSpec._initializing is private, hence the defensive getattr chain — though not obscure: it has been there since the spec protocol landed in 3.4. Absent (or an object with no __spec__, such as a C extension loaded the old way) reads as finished.

PARAMETER	DESCRIPTION
`module`	The module :func:`importlib.import_module` returned. TYPE: `ModuleType`

RETURNS	DESCRIPTION
`bool`	`True` while the module's body is still running.

Source code in eden/utils/registry.py

def module_is_initializing(module: Any) -> bool:
    """Whether ``module`` is still executing its own body — a partial import, not a finished one.

    ``importlib.import_module`` returns immediately, with the half-built module, when the calling
    thread is already importing it; only a *different* thread blocks until the body is done. This
    is the flag the import system itself reads to tell those apart
    (``importlib._bootstrap._find_and_load``), and two callers need it for the same reason — to
    know whether what they just got back is complete enough to draw a conclusion from:
    :meth:`Registry._populate` before it memoizes ``_populated``, and
    ``ManagerTermOptions._resolve_term_registry`` before it reports "no registry claims this
    class".

    ``ModuleSpec._initializing`` is private, hence the defensive ``getattr`` chain — though not
    obscure: it has been there since the spec protocol landed in 3.4. Absent (or an object with
    no ``__spec__``, such as a C extension loaded the old way) reads as finished.

    Parameters
    ----------
    module : ModuleType
        The module :func:`importlib.import_module` returned.

    Returns
    -------
    bool
        ``True`` while the module's body is still running.
    """
    return bool(getattr(getattr(module, "__spec__", None), "_initializing", False))

Configurable objects¶

ConfigurableMixin[Options] binds a runtime class (a term, a deployment, a dataset, ...) to its options class: Cls.configure(**kwargs) returns validated options, and constructing the object from them mirrors the options onto it.

ConfigurableMixin: bind options classes to their runtime objects.

Attributes¶

T `module-attribute` ¶

T = TypeVar('T', bound='ConfigurableOptions')

Classes¶

ConfigurableMixin ¶

ConfigurableMixin(options: T)

Bases: Generic[T]

Bind a runtime object (term / entity) to its declarative Options class.

The binding is reflection-driven, so subclasses just declare class attributes and get a matching options class "for free":

Options class discovery — :meth:__init_subclass__ reads the ConfigurableMixin[T] type argument off __orig_bases__ and stores it as _options_class_.
Parameter discovery — :meth:get_parameter_names walks the MRO and collects every plain (non-function, non-underscore, non-property/classmethod/staticmethod) class attribute, minus _non_parameter_names; those are the configurable parameters.
Options ↔ attribute mirroring — :meth:configure packs the parameters into an _options_class_ instance, and __init__ mirrors them back onto the runtime object via :func:mirror_options_onto (deep-copying both the options' values and the class-level defaults, so an instance never aliases either).

_extra_option_params (below) are the one documented exception to the mirroring.

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¶

name `property` ¶

name: str

uid `property` ¶

uid

options `property` ¶

options: T

Methods:¶

get_parameter_names `classmethod` ¶

get_parameter_names() -> list[str]

Source code in eden/utils/common.py

@classmethod
def get_parameter_names(cls) -> list[str]:
    params = set()
    excluded: set[str] = set()
    # Only collect attributes from classes in the ConfigurableMixin
    # hierarchy, skipping unrelated bases like nn.Module.
    for base_cls in cls.__mro__:
        excluded |= base_cls.__dict__.get("_non_parameter_names", frozenset())
        if base_cls is ConfigurableMixin or not issubclass(base_cls, ConfigurableMixin):
            continue
        for name, value in base_cls.__dict__.items():
            if name.startswith("__") or name.startswith("_"):
                continue
            if isinstance(value, (property, cached_property, classmethod, staticmethod)):
                continue
            if isinstance(value, FunctionType):
                continue
            params.add(name)
    return list(params - excluded)

fingerprint_names `classmethod` ¶

fingerprint_names(exclude: Iterable[str] = ()) -> tuple[str, ...]

The parameters :meth:configured_fingerprint reads, name-sorted, leaving out exclude.

Resolving them walks the class's MRO (:meth:get_parameter_names). A caller that fingerprints one object on every step resolves them once and hands them to :meth:configured_fingerprint as names.

Source code in eden/utils/common.py

@classmethod
def fingerprint_names(cls, exclude: Iterable[str] = ()) -> tuple[str, ...]:
    """The parameters :meth:`configured_fingerprint` reads, name-sorted, leaving out ``exclude``.

    Resolving them walks the class's MRO (:meth:`get_parameter_names`). A caller that fingerprints one object
    on every step resolves them once and hands them to :meth:`configured_fingerprint` as ``names``.
    """
    excluded = frozenset(exclude)
    return tuple(name for name in sorted(cls.get_parameter_names()) if name not in excluded)

configured_fingerprint ¶

configured_fingerprint(exclude: Iterable[str] = (), names: tuple[str, ...] | None = None) -> tuple[tuple[str, Hashable], ...]

Every configured parameter's current value, by name, as :func:frozen_value makes it comparable.

Two objects with equal fingerprints are configured alike right now; one fingerprint taken twice differs exactly when a parameter changed in between (a curriculum moving it, a nested options model edited in place, a tensor rebound). Name-sorted, so the pairs line up between two calls. exclude leaves parameters out; names gives the parameters as :meth:fingerprint_names already resolved them, instead of exclude.

Source code in eden/utils/common.py

def configured_fingerprint(
    self, exclude: Iterable[str] = (), names: tuple[str, ...] | None = None
) -> tuple[tuple[str, Hashable], ...]:
    """Every configured parameter's current value, by name, as :func:`frozen_value` makes it comparable.

    Two objects with equal fingerprints are configured alike right now; one fingerprint taken twice
    differs exactly when a parameter changed in between (a curriculum moving it, a nested options model
    edited in place, a tensor rebound). Name-sorted, so the pairs line up between two calls. ``exclude``
    leaves parameters out; ``names`` gives the parameters as :meth:`fingerprint_names` already resolved them,
    instead of ``exclude``.
    """
    excluded = frozenset(exclude)  # once: a generator is truthy even when it yields nothing
    if names is None:
        names = self.fingerprint_names(excluded)
    elif excluded:
        raise ValueError(
            f"configured_fingerprint of {type(self).__name__} takes exclude or names, not both: names are "
            f"already resolved without the excluded ones (fingerprint_names(exclude={sorted(excluded)!r}))."
        )
    return tuple((name, frozen_value(getattr(self, name, None))) for name in names)

get_parameter_specs `classmethod` ¶

get_parameter_specs() -> tuple[ParamInfo, ...]

:meth:get_parameter_names with each parameter's declared type and default.

Name-sorted, because get_parameter_names collects into a set and its order is therefore arbitrary — while an error message quoting the accepted set, and the generated catalogue page, both have to be reproducible.

Nothing is ever reported required: a class parameter is a class attribute, so it always has a default (an annotation-only declaration is not a parameter at all — see :meth:get_parameter_names). The required/optional split exists only for function terms, in :meth:ConfigurableFuncWrapperMixin.get_function_parameter_specs.

Source code in eden/utils/common.py

@classmethod
def get_parameter_specs(cls) -> tuple[ParamInfo, ...]:
    """:meth:`get_parameter_names` with each parameter's declared type and default.

    Name-sorted, because ``get_parameter_names`` collects into a set and its order is
    therefore arbitrary — while an error message quoting the accepted set, and the generated
    catalogue page, both have to be reproducible.

    Nothing is ever reported ``required``: a class parameter *is* a class attribute, so it
    always has a default (an annotation-only declaration is not a parameter at all — see
    :meth:`get_parameter_names`). The required/optional split exists only for function terms,
    in :meth:`ConfigurableFuncWrapperMixin.get_function_parameter_specs`.
    """
    annotations: dict[str, Any] = {}
    # Reversed MRO so a subclass's annotation wins over the base's, and ``__dict__`` rather
    # than ``getattr`` because ``cls.__annotations__`` silently falls back to the nearest
    # annotated base — which would attribute a base's type to an unannotated override.
    for base_cls in reversed(cls.__mro__):
        annotations.update(base_cls.__dict__.get("__annotations__") or {})
    return tuple(
        ParamInfo(name, render_annotation(annotations.get(name)), render_default(getattr(cls, name)))
        for name in sorted(cls.get_parameter_names())
    )

get_configure_specs `classmethod` ¶

get_configure_specs() -> tuple[ParamInfo, ...]

Every parameter :meth:configure accepts — what an "invalid parameter" error must quote.

:meth:get_parameter_specs plus the options-only :attr:_extra_option_params, whose type and default are read off the options class because they have no class attribute to read one from (that is exactly what makes them options-only).

Source code in eden/utils/common.py

@classmethod
def get_configure_specs(cls) -> tuple[ParamInfo, ...]:
    """Every parameter :meth:`configure` accepts — what an "invalid parameter" error must quote.

    :meth:`get_parameter_specs` plus the options-only :attr:`_extra_option_params`, whose type
    and default are read off the options class because they have no class attribute to read one
    from (that is exactly what makes them options-only).
    """
    fields = getattr(cls._options_class_, "model_fields", None) or {}
    extras = []
    for name in sorted(cls._extra_option_params):
        field = fields.get(name)
        if field is None:
            extras.append(ParamInfo(name))
            continue
        required = field.is_required()
        extras.append(
            ParamInfo(
                name,
                render_annotation(field.annotation),
                # ``get_default`` (not ``field.default``) so a ``default_factory`` field renders
                # its value, not the ``PydanticUndefined`` sentinel.
                "" if required else render_default(field.get_default(call_default_factory=True)),
                required=required,
            )
        )
    return cls.get_parameter_specs() + tuple(extras)

configure `classmethod` ¶

configure(**kwargs) -> T

Configure the class with the given parameters.

PARAMETER	DESCRIPTION
`**kwargs`	The parameters to configure the class with. DEFAULT: `{}`

RETURNS	DESCRIPTION
`T`	The options instance for the class.

Source code in eden/utils/common.py

@classmethod
def configure(cls, **kwargs) -> T:
    """
    Configure the class with the given parameters.

    Parameters
    ----------
    **kwargs: dict
        The parameters to configure the class with.

    Returns
    -------
    T
        The options instance for the class.
    """
    config = cls._build_config_dict(**kwargs)
    if cls._options_class_ is None:
        raise TypeError(
            f"{cls.__name__} must parameterize ConfigurableMixin[T] with a concrete Options type "
            f"or set _options_class_ explicitly."
        )
    # ``name`` is framework-injected so managers can resolve terms by registry key
    # (``ManagerBase._build_term`` reads ``term_option.name``). Manager-term options always
    # subclass the extra="allow" ``*TermOptions`` bases and carry it as an extra. Every other
    # configure target (Camera, Sensor, Entity, extension options) never reads it, so for an
    # extra="forbid" options class that doesn't declare a ``name`` field we drop it instead of
    # tripping the forbid validation.
    options_class = cls._options_class_
    if options_class.model_config.get("extra") == "forbid" and "name" not in options_class.model_fields:
        config.pop("name", None)
    return options_class(**config)

ConfigurableFuncWrapperMixin ¶

ConfigurableFuncWrapperMixin(options: T)

Bases: ConfigurableMixin[T]

Source code in eden/utils/common.py

def __init__(self, options: T) -> None:
    super().__init__(options=options)
    # Build-time backstop for ``params`` KEYS: construction and assignment validate them
    # (ManagerTermOptions), but IN-PLACE mutation of the raw params dict
    # (``options.params["typo"] = ...``) is invisible to ``__setattr__``. Catching it here
    # names the term at env build instead of a bare TypeError at the first ``compute()``.
    # ``_func`` is set by ManagerTermFuncWrapperBase.__init__, which every concrete wrapper
    # calls first; a missing func defers to the compute-time error.
    func = getattr(self, "_func", None)
    if func is None:
        return
    invalid_keys = invalid_function_param_keys(func, self.params)
    if invalid_keys:
        raise ValueError(
            invalid_params_message(options.name, invalid_keys, self.get_function_parameter_specs(func))
        )

Attributes¶

params `class-attribute` `instance-attribute` ¶

params: dict[str, Any] = {}

name `property` ¶

name: str

Methods:¶

get_function_parameter_specs `classmethod` ¶

get_function_parameter_specs(func: FunctionType) -> tuple[ParamInfo, ...]

The params={...} entries an author supplies for func, typed and defaulted.

The listable counterpart of :meth:get_required_function_parameter_names, which returns only the subset a rejection may be based on. This is wider on purpose — see :func:~eden.utils.params.function_param_specs — because describing a term means naming every key it accepts, not only the ones configure() will refuse to go without.

Source code in eden/utils/common.py

@classmethod
def get_function_parameter_specs(cls, func: FunctionType) -> tuple[ParamInfo, ...]:
    """The ``params={...}`` entries an author supplies for ``func``, typed and defaulted.

    The listable counterpart of :meth:`get_required_function_parameter_names`, which returns
    only the subset a *rejection* may be based on. This is wider on purpose — see
    :func:`~eden.utils.params.function_param_specs` — because describing a term means naming
    every key it accepts, not only the ones ``configure()`` will refuse to go without.
    """
    return function_param_specs(func, exclude=cls._manager_injected_params)

get_required_function_parameter_names `classmethod` ¶

get_required_function_parameter_names(func: FunctionType) -> set[str]

Parameters of func without a default that the author must supply via params.

Keyword-only and positional-or-keyword params both count (compute calls func(env=..., **params), so either is supplied by keyword), matching what :meth:get_function_parameter_specs advertises as required. _manager_injected_params (env, envs_idx) are excluded.

Source code in eden/utils/common.py

@classmethod
def get_required_function_parameter_names(cls, func: FunctionType) -> set[str]:
    """Parameters of ``func`` without a default that the author must supply via ``params``.

    Keyword-only and positional-or-keyword params both count (``compute`` calls
    ``func(env=..., **params)``, so either is supplied by keyword), matching what
    :meth:`get_function_parameter_specs` advertises as required. ``_manager_injected_params``
    (``env``, ``envs_idx``) are excluded.
    """
    return {
        name
        for name, param in inspect.signature(func).parameters.items()
        if param.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
        and param.default is inspect.Parameter.empty
    } - cls._manager_injected_params

configure `classmethod` ¶

configure(func: FunctionType, **kwargs) -> T

Configure the class with the given parameters.

PARAMETER	DESCRIPTION
`func`	The function to configure the class with. TYPE: `FunctionType`
`**kwargs`	The parameters to configure the class with. DEFAULT: `{}`

RETURNS	DESCRIPTION
`T`	The options instance for the class.

Source code in eden/utils/common.py

@classmethod
def configure(cls, func: FunctionType, **kwargs) -> T:
    """
    Configure the class with the given parameters.

    Parameters
    ----------
    func: FunctionType
        The function to configure the class with.
    **kwargs: dict
        The parameters to configure the class with.

    Returns
    -------
    T
        The options instance for the class.
    """
    config = cls._build_config_dict(func=func, **kwargs)
    if cls._options_class_ is None:
        raise TypeError(
            f"{cls.__name__} must parameterize ConfigurableMixin[T] with a concrete Options type "
            f"or set _options_class_ explicitly."
        )
    return cls._options_class_(**config)

Functions:¶

frozen_value ¶

frozen_value(value: Any, depth: int = 0) -> Hashable

value as a hashable constant that compares equal exactly when the value is the same configuration.

For keying or fingerprinting what a term was configured with (:meth:ConfigurableMixin.configured_fingerprint). A scalar, string or enum is itself. A tensor is where it lives (its storage address, shape, strides and dtype), not its values: two terms reading one tensor are configured alike, and a tensor written in place is the same configuration while one rebound to another tensor is not. A numpy array is its values (a host array is read as a constant, so one edited in place is another configuration). A list, tuple, set or dict is its items, and an options model or a dataclass is its fields, so a value edited in place inside one is seen; past _FROZEN_DEPTH levels, and for any other object, a value is its identity.

Source code in eden/utils/common.py

def frozen_value(value: Any, depth: int = 0) -> Hashable:
    """``value`` as a hashable constant that compares equal exactly when the value is the same configuration.

    For keying or fingerprinting what a term was configured with (:meth:`ConfigurableMixin.configured_fingerprint`).
    A scalar, string or enum is itself. A tensor is where it lives (its storage address, shape, strides and dtype),
    not its values: two terms reading one tensor are configured alike, and a tensor written in place is the same
    configuration while one rebound to another tensor is not. A numpy array is its values (a host array is read
    as a constant, so one edited in place is another configuration). A list, tuple, set or dict is its items,
    and an options model or a dataclass is its fields, so a value edited in place inside one is seen; past
    ``_FROZEN_DEPTH`` levels, and for any other object, a value is its identity.
    """
    if value is None or isinstance(value, (bool, int, float, str, Enum)):
        return value
    if isinstance(value, torch.Tensor):
        return ("tensor", value.data_ptr(), tuple(value.shape), value.stride(), value.dtype)
    if isinstance(value, np.ndarray):
        return ("ndarray", value.shape, value.dtype.str, value.tobytes())
    if isinstance(value, np.generic):
        return value.item()
    if depth >= _FROZEN_DEPTH:
        return ("object", id(value))
    if isinstance(value, (list, tuple)):
        return tuple(frozen_value(item, depth + 1) for item in value)
    if isinstance(value, (set, frozenset)):
        return frozenset(frozen_value(item, depth + 1) for item in value)
    if isinstance(value, dict):
        return tuple((key, frozen_value(item, depth + 1)) for key, item in value.items())
    if isinstance(value, BaseModel):
        items = [(name, getattr(value, name)) for name in type(value).model_fields]
        items += list((value.model_extra or {}).items())
        return (type(value), tuple((name, frozen_value(item, depth + 1)) for name, item in items))
    if is_dataclass(value) and not isinstance(value, type):
        return (
            type(value),
            tuple((f.name, frozen_value(getattr(value, f.name), depth + 1)) for f in dataclass_fields(value)),
        )
    return ("object", id(value))

mirror_options_onto ¶

mirror_options_onto(obj: ConfigurableMixin, options: ConfigurableOptions) -> None

Mirror every configurable parameter of obj from options onto obj itself.

A parameter the options do not carry falls back to obj's own class attribute. Both sources are deep-copied, so a runtime object never shares a mutable value with

its class — otherwise every instance is handed the same list/dict, and mutating one instance leaks into all the others and permanently corrupts the class default for the rest of the process; or
its options — otherwise mutating the runtime attribute in place rewrites the config EnvBase.config serializes, and two objects built from one options instance (a config reused across envs) silently share state.

:class:Entity, :class:Sensor and :class:Camera call this directly rather than delegating __init__ to :class:ConfigurableMixin: their options also carry fields that are consumed instead of mirrored (morph construction, the nested Genesis sensor/camera options), which the mixin's unused-parameter check would reject.

Two consequences of copying the options' side. A mirrored attribute and obj._options.<name> are independent from here on, so code that mutates one must not expect the other to follow — read a mirrored field off obj, not off the options. And a built Genesis :class:Surface (the non-SurfaceOptions half of SurfaceLike) is aliased, not copied — see :func:_copy_mirrored_value for why.

Source code in eden/utils/common.py

def mirror_options_onto(obj: ConfigurableMixin, options: ConfigurableOptions) -> None:
    """Mirror every configurable parameter of ``obj`` from ``options`` onto ``obj`` itself.

    A parameter the options do not carry falls back to ``obj``'s own class attribute. Both
    sources are deep-copied, so a runtime object never shares a mutable value with

    * its **class** — otherwise every instance is handed the *same* list/dict, and mutating
      one instance leaks into all the others *and* permanently corrupts the class default
      for the rest of the process; or
    * its **options** — otherwise mutating the runtime attribute in place rewrites the
      config ``EnvBase.config`` serializes, and two objects built from one options instance
      (a config reused across envs) silently share state.

    :class:`Entity`, :class:`Sensor` and :class:`Camera` call this directly rather than
    delegating ``__init__`` to :class:`ConfigurableMixin`: their options also carry fields
    that are *consumed* instead of mirrored (morph construction, the nested Genesis
    sensor/camera options), which the mixin's unused-parameter check would reject.

    Two consequences of copying the options' side. A mirrored attribute and
    ``obj._options.<name>`` are independent from here on, so code that mutates one must not
    expect the other to follow — read a mirrored field off ``obj``, not off the options. And a
    built Genesis :class:`Surface` (the non-``SurfaceOptions`` half of ``SurfaceLike``) is
    *aliased*, not copied — see :func:`_copy_mirrored_value` for why.
    """
    # Field names only. ``keys()`` is a cached, value-free enumeration; ``model_dump()`` would
    # recursively serialize the entire options subtree (material, surface, every default_dofs_*
    # table) once per call, and emit serializer warnings for arbitrary-typed fields.
    option_fields = set(options.keys())
    # ``ManagerTermOptions.params`` defaults to ``None``, meaning "not a function term".
    # Mirroring that ``None`` would shadow the wrapper's ``{}`` class default and every function
    # term would die at its first ``compute()`` with "argument after ** must be a mapping", so an
    # empty ``params`` falls through to the class default. Same predicate the unused-parameter
    # check below uses, deliberately.
    if not getattr(options, "params", None):
        option_fields.discard("params")
    for name in obj.get_parameter_names():
        source = options if name in option_fields else obj
        setattr(obj, name, _copy_mirrored_value(getattr(source, name)))

Parameter descriptions¶

How a configurable's parameters are described, shared by eden terms, the term catalog and the configure() errors so all three word a parameter identically.

How a configurable's parameters are described — and the wording that describes them.

A term's accepted parameter set has to be visible outside its signature: an author who guessed a parameter name wrong learns nothing from an error that names only the offending key. This module is the single place that renders a parameter (name, type, default) and the single place that words the "that parameter is not accepted" errors, so:

the three checks that run at construction (ManagerTermOptions._validate_param_names), at configure() and at build (ConfigurableFuncWrapperMixin) cannot drift in wording, and
eden terms and :mod:eden.managers.catalog describe a term exactly the way its errors do.

No eden imports (only :mod:inspect, :mod:typing and pydantic's TypeAdapter), so both :mod:eden.options.managers.base and :mod:eden.utils.common can reach it — which is also why invalid_function_param_keys lives here rather than with the manager options (see its docstring).

Classes¶

ParamInfo `dataclass` ¶

ParamInfo(name: str, annotation: str = '', default: str = '', required: bool = False)

One configurable parameter, as an author has to spell it: name, type, default.

Every field is already-rendered text rather than a live object. The only consumers are human- and agent-readable listings, and the term modules use from __future__ import annotations, so a signature's annotations arrive as source text anyway.

Attributes¶

name `instance-attribute` ¶

name: str

annotation `class-attribute` `instance-attribute` ¶

annotation: str = ''

default `class-attribute` `instance-attribute` ¶

default: str = ''

required `class-attribute` `instance-attribute` ¶

required: bool = False

Methods:¶

Functions:¶

render_annotation ¶

render_annotation(annotation: Any) -> str

annotation as source text, or "" when it is absent.

An annotation of literal None also renders as "": it is indistinguishable from "unannotated" for a class attribute (dict.get returns None for both) and no term declares one.

Source code in eden/utils/params.py

def render_annotation(annotation: Any) -> str:
    """``annotation`` as source text, or ``""`` when it is absent.

    An annotation of literal ``None`` also renders as ``""``: it is indistinguishable from
    "unannotated" for a class attribute (``dict.get`` returns ``None`` for both) and no term
    declares one.
    """
    if annotation is inspect.Parameter.empty or annotation is None:
        return ""
    return annotation if isinstance(annotation, str) else _format_type(annotation)

render_default ¶

render_default(value: Any) -> str

value as a one-line literal an author could paste back into a config.

Enum members by name — repr gives <ReferenceSource.DEFAULT: 'default'>, which is not something you can write — and any line breaks collapsed to single spaces, so a multi-line repr cannot split the markdown table row or terminal line it is rendered into.

Source code in eden/utils/params.py

def render_default(value: Any) -> str:
    """``value`` as a one-line literal an author could paste back into a config.

    Enum members by name — ``repr`` gives ``<ReferenceSource.DEFAULT: 'default'>``, which is not
    something you can write — and any line breaks collapsed to single spaces, so a multi-line repr
    cannot split the markdown table row or terminal line it is rendered into.
    """
    if isinstance(value, Enum):
        return f"{type(value).__name__}.{value.name}"
    # Collapse only line breaks, not every space run (which would corrupt a string default's content).
    return repr(value).replace("\r\n", " ").replace("\r", " ").replace("\n", " ")

function_param_specs ¶

function_param_specs(func: FunctionType, *, exclude: Collection[str] = ()) -> tuple[ParamInfo, ...]

The params={...} entries an author supplies for function term func, in signature order.

Every named parameter except those in exclude (the manager-injected env / envs_idx) is listed, because every one of them is a legal params key: compute calls func(env=..., **params) and invalid_function_param_keys validates against the whole signature. That deliberately includes positional-or-keyword params — Eden's built-ins declare author-supplied params keyword-only, but nothing stops an out-of-tree term from doing otherwise, and omitting one from the listing would describe the term wrongly.

*args / **kwargs are dropped: neither is a parameter an author can name. A **kwargs term accepts any key (see invalid_function_param_keys), which a name list cannot express.

Source code in eden/utils/params.py

def function_param_specs(func: FunctionType, *, exclude: Collection[str] = ()) -> tuple[ParamInfo, ...]:
    """The ``params={...}`` entries an author supplies for function term ``func``, in signature order.

    Every named parameter except those in ``exclude`` (the manager-injected ``env`` / ``envs_idx``)
    is listed, because every one of them is a legal ``params`` key: ``compute`` calls
    ``func(env=..., **params)`` and ``invalid_function_param_keys`` validates against the whole
    signature. That deliberately includes positional-or-keyword params — Eden's built-ins declare
    author-supplied params keyword-only, but nothing stops an out-of-tree term from doing otherwise,
    and omitting one from the listing would describe the term wrongly.

    ``*args`` / ``**kwargs`` are dropped: neither is a parameter an author can name. A ``**kwargs``
    term accepts *any* key (see ``invalid_function_param_keys``), which a name list cannot express.
    """
    return tuple(
        ParamInfo(
            name,
            render_annotation(param.annotation),
            "" if param.default is inspect.Parameter.empty else render_default(param.default),
            required=param.default is inspect.Parameter.empty,
        )
        for name, param in inspect.signature(func).parameters.items()
        if param.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
        and name not in exclude
    )

render_params ¶

render_params(specs: Iterable[ParamInfo]) -> str

specs as one comma-separated line, required first — or "(none)".

Required first because that is the part an author is missing when this text appears in an error; the rest is reference. Stable within each half (input order is signature order).

Source code in eden/utils/params.py

def render_params(specs: Iterable[ParamInfo]) -> str:
    """``specs`` as one comma-separated line, required first — or ``"(none)"``.

    Required first because that is the part an author is missing when this text appears in an
    error; the rest is reference. Stable within each half (input order is signature order).
    """
    ordered = sorted(specs, key=lambda spec: not spec.required)
    return ", ".join(str(spec) for spec in ordered) or "(none)"

invalid_function_param_keys ¶

invalid_function_param_keys(func, params) -> set[str]

Return the params keys that are not parameters of func.

Empty when every key is a parameter of func — or when func accepts **kwargs, in which case any key is legal. Shared by the construction-time check (:meth:~eden.options.managers.base.ManagerTermOptions._validate_param_names) and the build-time backstop (ConfigurableFuncWrapperMixin.__init__) so the two can't drift. Callers that need the accepted set for a message build it from :func:function_param_specs, which also carries the types and defaults a bare name list cannot.

Here rather than in :mod:eden.options.managers.base because :mod:eden.utils.common is its other caller and eden.options.entities reaches back into eden.utils (via eden.utils.terrains) — so a utils module that imports eden.options at module scope is a cycle as soon as it is the first of the three to be imported. This module's contract is exactly that: no eden imports, reachable from either side.

PARAMETER	DESCRIPTION
`func`	The function term whose signature defines the legal keys. TYPE: `Callable`
`params`	The `params` dict to check; `None` and `{}` are both legal and yield no keys. TYPE: `Mapping \| None`

RETURNS	DESCRIPTION
`set[str]`	The offending keys, empty when there are none.

Source code in eden/utils/params.py

def invalid_function_param_keys(func, params) -> set[str]:
    """Return the ``params`` keys that are not parameters of ``func``.

    Empty when every key is a parameter of ``func`` — or when ``func`` accepts ``**kwargs``,
    in which case any key is legal. Shared by the construction-time check
    (:meth:`~eden.options.managers.base.ManagerTermOptions._validate_param_names`) and the
    build-time backstop (``ConfigurableFuncWrapperMixin.__init__``) so the two can't drift.
    Callers that need the accepted set for a message build it from :func:`function_param_specs`,
    which also carries the types and defaults a bare name list cannot.

    Here rather than in :mod:`eden.options.managers.base` because :mod:`eden.utils.common` is its
    other caller and ``eden.options.entities`` reaches back into ``eden.utils`` (via
    ``eden.utils.terrains``) — so a utils module that imports ``eden.options`` at module scope is a
    cycle as soon as it is the first of the three to be imported.
    This module's contract is exactly that: no eden imports, reachable from either side.

    Parameters
    ----------
    func : Callable
        The function term whose signature defines the legal keys.
    params : Mapping | None
        The ``params`` dict to check; ``None`` and ``{}`` are both legal and yield no keys.

    Returns
    -------
    set[str]
        The offending keys, empty when there are none.
    """
    signature_params = inspect.signature(func).parameters
    if any(p.kind is inspect.Parameter.VAR_KEYWORD for p in signature_params.values()):
        return set()
    return set(params or {}) - set(signature_params)

invalid_params_message ¶

invalid_params_message(term_name: str, invalid_keys: Collection[str], accepted: Iterable[ParamInfo]) -> str

The one wording for "these params keys are not parameters of this term".

Shared by the construction-time, configure()-time and build-time checks so an author who hits it twice reads the same sentence, and so the accepted set is never omitted by one of them.

Source code in eden/utils/params.py

def invalid_params_message(term_name: str, invalid_keys: Collection[str], accepted: Iterable[ParamInfo]) -> str:
    """The one wording for "these ``params`` keys are not parameters of this term".

    Shared by the construction-time, ``configure()``-time and build-time checks so an author who
    hits it twice reads the same sentence, and so the accepted set is never omitted by one of them.
    """
    return (
        f"The following params keys are invalid: {sorted(invalid_keys)} for function term "
        f"'{term_name}', accepted params: {render_params(accepted)}"
    )

unknown_parameter_names ¶

unknown_parameter_names(extras: Iterable[str], configurable: type, *, tolerated: Collection[str] = ()) -> set[str]

The names in extras (an options object's extra fields) that configurable takes no parameter for.

The one check behind every options class that carries a configurable's parameters as extras (term options, camera overlay options): a parameter is a class parameter (get_parameter_names) or an options-only one (_extra_option_params). The options' own metadata (_option_*) always passes, and tolerated names what else a caller's options carry that is not a parameter (term options' class_name).

Source code in eden/utils/params.py

def unknown_parameter_names(extras: Iterable[str], configurable: type, *, tolerated: Collection[str] = ()) -> set[str]:
    """The names in ``extras`` (an options object's extra fields) that ``configurable`` takes no parameter for.

    The one check behind every options class that carries a configurable's parameters as extras (term options,
    camera overlay options): a parameter is a class parameter (``get_parameter_names``) or an options-only one
    (``_extra_option_params``). The options' own metadata (``_option_*``) always passes, and ``tolerated`` names what
    else a caller's options carry that is not a parameter (term options' ``class_name``).
    """
    allowed = set(configurable.get_parameter_names()) | set(getattr(configurable, "_extra_option_params", ()))
    return {key for key in extras if not key.startswith("_option_")} - allowed - set(tolerated)

invalid_parameters_message ¶

invalid_parameters_message(subject: str, invalid_keys: Collection[str], accepted: Iterable[ParamInfo], retired: Mapping[str, str]) -> str

The one wording for "these are not parameters of subject" — a class term or configurable.

The counterpart of :func:invalid_params_message for top-level parameters, shared by the same three checks. A key in retired (a class's _retired_params) also gets that class's migration note: a retired name is still invalid, but a config saved before it was retired should fail with directions, not only with the list of what is accepted now.

Source code in eden/utils/params.py

def invalid_parameters_message(
    subject: str,
    invalid_keys: Collection[str],
    accepted: Iterable[ParamInfo],
    retired: Mapping[str, str],
) -> str:
    """The one wording for "these are not parameters of ``subject``" — a class term or configurable.

    The counterpart of :func:`invalid_params_message` for top-level parameters, shared by the same
    three checks. A key in ``retired`` (a class's ``_retired_params``) also gets that class's
    migration note: a retired name is still invalid, but a config saved before it was retired should
    fail with directions, not only with the list of what is accepted now.
    """
    message = (
        f"The following parameters are invalid: {sorted(invalid_keys)} for {subject}, "
        f"allowed parameters: {render_params(accepted)}"
    )
    retired_keys_by_note: dict[str, list[str]] = {}
    for key in sorted(invalid_keys):
        if key in retired:
            retired_keys_by_note.setdefault(retired[key], []).append(key)
    notes = [f"Retired {keys}: {note}" for note, keys in retired_keys_by_note.items()]
    return ". ".join([message, *notes])

validated_parameter_values ¶

validated_parameter_values(configurable: type, values: Mapping[str, Any], *, subject: str) -> dict[str, Any]

values with each of configurable's parameters validated against its declared type and converted the way pydantic converts a declared options field: a list a JSON file read back becomes the declared tuple, "0.5" becomes 0.5.

For options that carry a configurable's parameters as extras, which pydantic leaves exactly as written. A key that is not a parameter passes through untouched; :func:unknown_parameter_names is what refuses those.

RAISES	DESCRIPTION
`ValueError`	For a value the declared type does not admit (`None` or a bool for a number, a string for a list), naming `subject`, the parameter, its type and its default.
`TypeError`	For a parameter whose annotation cannot be resolved (a type imported only for type checking) or validated.

Source code in eden/utils/params.py

def validated_parameter_values(configurable: type, values: Mapping[str, Any], *, subject: str) -> dict[str, Any]:
    """``values`` with each of ``configurable``'s parameters validated against its declared type and converted the way
    pydantic converts a declared options field: a list a JSON file read back becomes the declared tuple, ``"0.5"``
    becomes ``0.5``.

    For options that carry a configurable's parameters as extras, which pydantic leaves exactly as written. A key that
    is not a parameter passes through untouched; :func:`unknown_parameter_names` is what refuses those.

    Raises
    ------
    ValueError
        For a value the declared type does not admit (``None`` or a bool for a number, a string for a list), naming
        ``subject``, the parameter, its type and its default.
    TypeError
        For a parameter whose annotation cannot be resolved (a type imported only for type checking) or validated.
    """
    params = set(configurable.get_parameter_names())
    validated = dict(values)
    for key, value in values.items():
        if key not in params:
            continue
        validator = _parameter_validator(configurable, key)
        if validator is None:
            continue
        annotation, adapter = validator
        try:
            # Lax, as an options field converts (``"0.5"`` to 0.5, a list to a tuple), except that a bool counts as a
            # number only where the type admits a bool: ``shadow_strength: yes`` in a YAML file is a mistake, not 1.0.
            validated[key] = adapter.validate_python(value, strict=True if isinstance(value, bool) else None)
        except ValidationError as e:
            raise ValueError(
                f"{subject} needs {key} as {render_annotation(annotation)}; got {value!r} ({type(value).__name__}): "
                f"{e.errors()[0]['msg']}. E.g. {key}={render_default(getattr(configurable, key))} (the default)."
            ) from None
    return validated

Multi-GPU (DDP)¶

Rank and world-size queries and process-group setup. Also reachable as eden.distributed. See the multi-GPU guide.

Distributed Data Parallel (DDP) utilities for Eden.

This module provides a simple, universal interface for multi-GPU training with PyTorch DDP. Eden uses the NCCL backend for collective communication.

import eden restricts each process's CUDA_VISIBLE_DEVICES to one physical GPU so that Genesis and PyTorch both see cuda:0. The NCCL process group is initialized early in eden.init() — right after gs.init() and before any scene building — so that NCCL gets a clean CUDA context.

Usage with torchrun::

torchrun --standalone --nproc_per_node=2 my_script.py

The module reads LOCAL_RANK, RANK, and WORLD_SIZE environment variables set by torchrun (or any compatible launcher).

Functions:¶

get_local_rank ¶

get_local_rank() -> int

Return the local rank of the current process (GPU index on this node).

Source code in eden/utils/distributed.py

def get_local_rank() -> int:
    """Return the local rank of the current process (GPU index on this node)."""
    return int(os.environ.get("LOCAL_RANK", 0))

get_global_rank ¶

get_global_rank() -> int

Return the global rank of the current process.

Source code in eden/utils/distributed.py

def get_global_rank() -> int:
    """Return the global rank of the current process."""
    return int(os.environ.get("RANK", 0))

get_world_size ¶

get_world_size() -> int

Return the total number of processes.

Source code in eden/utils/distributed.py

def get_world_size() -> int:
    """Return the total number of processes."""
    return int(os.environ.get("WORLD_SIZE", 1))

is_distributed ¶

is_distributed() -> bool

Return True if running in a multi-process (DDP) context.

Source code in eden/utils/distributed.py

def is_distributed() -> bool:
    """Return ``True`` if running in a multi-process (DDP) context."""
    return get_world_size() > 1

is_main_process ¶

is_main_process() -> bool

Return True if this is rank 0 (the main process).

Source code in eden/utils/distributed.py

def is_main_process() -> bool:
    """Return ``True`` if this is rank 0 (the main process)."""
    return get_global_rank() == 0

setup ¶

setup(backend: str = 'nccl') -> None

Initialize the torch.distributed process group.

This uses init_method="env://", which expects MASTER_ADDR, MASTER_PORT, RANK, and WORLD_SIZE to be set by the launcher (e.g. torchrun).

PARAMETER	DESCRIPTION
`backend`	Communication backend (default `"nccl"`). TYPE: `str` DEFAULT: `'nccl'`

Source code in eden/utils/distributed.py

def setup(backend: str = "nccl") -> None:
    """Initialize the ``torch.distributed`` process group.

    This uses ``init_method="env://"``, which expects ``MASTER_ADDR``,
    ``MASTER_PORT``, ``RANK``, and ``WORLD_SIZE`` to be set by the launcher
    (e.g. ``torchrun``).

    Parameters
    ----------
    backend : str
        Communication backend (default ``"nccl"``).
    """
    if dist.is_initialized():
        return
    if not is_distributed():
        return
    dist.init_process_group(backend=backend, init_method="env://")

cleanup ¶

cleanup() -> None

Destroy the process group and synchronize all ranks.

Source code in eden/utils/distributed.py

def cleanup() -> None:
    """Destroy the process group and synchronize all ranks."""
    if not dist.is_initialized():
        return
    dist.barrier()
    dist.destroy_process_group()

barrier ¶

barrier() -> None

Synchronize all processes. No-op if not distributed.

Source code in eden/utils/distributed.py

def barrier() -> None:
    """Synchronize all processes. No-op if not distributed."""
    if dist.is_initialized():
        dist.barrier()

main_only ¶

main_only(fn)

Make a decorated function execute only on the main process.

Source code in eden/utils/distributed.py

def main_only(fn):
    """Make a decorated function execute only on the main process."""

    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        if is_main_process():
            return fn(*args, **kwargs)
        return None

    return wrapper

Run identity and provenance¶

The per-run reproducibility manifest (run.json).

Eden's canonical run identity is the log-directory basename <timestamp>-<run id> (resolved in :mod:eden.__init__). This module writes a sibling run.json recording everything needed to reproduce or trace a run: the applied seed, the eden and Genesis versions, the eden working-tree git state, the RL library that wrote the checkpoints, the full command line, and the task.

It is deliberately separate from the per-config config.json envelope (_meta), which is re-stamped on every save_as_json — including deploy_config.json — and describes the task, not the run instance. run.json is written exactly once, at launch, into the (lazily created) log directory.

Attributes¶

RUN_MANIFEST_FILENAME `module-attribute` ¶

RUN_MANIFEST_FILENAME = 'run.json'

RUN_MANIFEST_SCHEMA_VERSION `module-attribute` ¶

RUN_MANIFEST_SCHEMA_VERSION = 3

RL_LIBRARY_DISTRIBUTIONS `module-attribute` ¶

RL_LIBRARY_DISTRIBUTIONS = {'rsl_rl': 'rsl-rl-lib', 'rl_games': 'rl-games'}

Functions:¶

get_rl_library ¶

get_rl_library(backend: str | None) -> dict[str, str | None] | None

The RL library record for backend, or None when there is nothing to record.

PARAMETER	DESCRIPTION
`backend`	Eden's runner-backend key — a key of :data:`RL_LIBRARY_DISTRIBUTIONS` (`"rsl_rl"` / `"rl_games"`), or `None` when the caller doesn't know. TYPE: `str \| None`

RETURNS	DESCRIPTION
`dict \| None`	`{"backend": ..., "distribution": ..., "version": ...}`, with `version` `None` if the distribution isn't installed under that name. `None` for an unknown or missing `backend` — provenance is best-effort and never fabricated.

Source code in eden/utils/run_manifest.py

def get_rl_library(backend: str | None) -> dict[str, str | None] | None:
    """The RL library record for ``backend``, or ``None`` when there is nothing to record.

    Parameters
    ----------
    backend : str | None
        Eden's runner-backend key — a key of :data:`RL_LIBRARY_DISTRIBUTIONS`
        (``"rsl_rl"`` / ``"rl_games"``), or ``None`` when the caller doesn't know.

    Returns
    -------
    dict | None
        ``{"backend": ..., "distribution": ..., "version": ...}``, with ``version``
        ``None`` if the distribution isn't installed under that name. ``None`` for an
        unknown or missing ``backend`` — provenance is best-effort and never fabricated.
    """
    distribution = RL_LIBRARY_DISTRIBUTIONS.get(backend or "")
    if distribution is None:
        return None
    # The backend can be known (the config named it) while the distribution is absent under
    # that name — a source checkout on PYTHONPATH, say: the version is None, the backend recorded anyway.
    return {"backend": backend, "distribution": distribution, "version": installed_version(distribution)}

read_run_manifest ¶

read_run_manifest(log_dir: str) -> dict[str, Any] | None

Read <log_dir>/run.json, or None if it is missing or unreadable.

Best-effort by contract: callers enrich diagnostics with this, so a run predating the manifest must degrade to "unrecorded", never to an error.

PARAMETER	DESCRIPTION
`log_dir`	A run's log directory (the one holding `run.json` and the checkpoints). TYPE: `str`

RETURNS	DESCRIPTION
`dict \| None`	The parsed manifest, or `None`.

Source code in eden/utils/run_manifest.py

def read_run_manifest(log_dir: str) -> dict[str, Any] | None:
    """Read ``<log_dir>/run.json``, or ``None`` if it is missing or unreadable.

    Best-effort by contract: callers enrich diagnostics with this, so a run predating the
    manifest must degrade to "unrecorded", never to an error.

    Parameters
    ----------
    log_dir : str
        A run's log directory (the one holding ``run.json`` and the checkpoints).

    Returns
    -------
    dict | None
        The parsed manifest, or ``None``.
    """
    path = os.path.join(log_dir, RUN_MANIFEST_FILENAME)
    try:
        with open(path) as f:
            manifest = json.load(f)
    except (OSError, ValueError):  # missing/unreadable, or truncated JSON
        return None
    return manifest if isinstance(manifest, dict) else None

build_run_manifest ¶

build_run_manifest(*, log_dir: str, run_id: str | None, timestamp: str | None, task: str, seed: int | None, world_size: int, run_name: str, argv: list[str], resumed_from: str | None = None, rl_backend: str | None = None) -> dict[str, Any]

Assemble the run-manifest dict, resolving the eden/Genesis/RL-library versions.

Split from :func:write_run_manifest so the field set can be asserted without touching the filesystem. Not pure: resolving the versions shells out to git and reads installed-package metadata, which is why tests stub those three lookups.

PARAMETER	DESCRIPTION
`log_dir`	The run's log directory (its basename is the canonical run identity). TYPE: `str`
`run_id`	This run's base-36 id, from :func:`eden.get_run_id`. Taken as an argument rather than parsed out of `log_dir` so the manifest never has to guess how the basename is composed (a later run_name decoration would break basename parsing). TYPE: `str \| None`
`timestamp`	This run's creation timestamp, from :func:`eden.get_run_timestamp`. TYPE: `str \| None`
`task`	The registered task name being trained. TYPE: `str`
`seed`	The base seed applied to the run's RNGs (`None` when the backend exposes no seed field). Under DDP this is rank 0's seed; rank `r` uses `seed + r` (see `world_size`), so the whole run reproduces from this base plus the topology. TYPE: `int \| None`
`world_size`	Number of DDP ranks (1 for single-GPU). Makes `seed` unambiguous: the per-rank seeds are `seed` through `seed + world_size - 1`. TYPE: `int`
`run_name`	The human-facing run name (`""` when unset). TYPE: `str`
`argv`	The full process command line (typically `sys.argv`). TYPE: `list[str]`
`resumed_from`	Basename of the parent run this one resumed from (`--resume`), or `None` for a fresh run. Records the fork lineage a resume creates (a resume trains in a new run directory rather than continuing the parent in place). TYPE: `str \| None` DEFAULT: `None`
`rl_backend`	Eden's runner-backend key (`"rsl_rl"` / `"rl_games"`) — the library that will write this run's checkpoints. Its installed version is resolved and recorded under `rl_library` (see :func:`get_rl_library`); `None` records `rl_library: null`. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`dict`	The JSON-serializable manifest.

Source code in eden/utils/run_manifest.py

def build_run_manifest(
    *,
    log_dir: str,
    run_id: str | None,
    timestamp: str | None,
    task: str,
    seed: int | None,
    world_size: int,
    run_name: str,
    argv: list[str],
    resumed_from: str | None = None,
    rl_backend: str | None = None,
) -> dict[str, Any]:
    """Assemble the run-manifest dict, resolving the eden/Genesis/RL-library versions.

    Split from :func:`write_run_manifest` so the field set can be asserted without
    touching the filesystem. Not pure: resolving the versions shells out to git and reads
    installed-package metadata, which is why tests stub those three lookups.

    Parameters
    ----------
    log_dir : str
        The run's log directory (its basename is the canonical run identity).
    run_id : str | None
        This run's base-36 id, from :func:`eden.get_run_id`. Taken as an argument rather
        than parsed out of ``log_dir`` so the manifest never has to guess how the basename
        is composed (a later run_name decoration would break basename parsing).
    timestamp : str | None
        This run's creation timestamp, from :func:`eden.get_run_timestamp`.
    task : str
        The registered task name being trained.
    seed : int | None
        The base seed applied to the run's RNGs (``None`` when the backend exposes no seed
        field). Under DDP this is rank 0's seed; rank ``r`` uses ``seed + r`` (see
        ``world_size``), so the whole run reproduces from this base plus the topology.
    world_size : int
        Number of DDP ranks (1 for single-GPU). Makes ``seed`` unambiguous: the per-rank
        seeds are ``seed`` through ``seed + world_size - 1``.
    run_name : str
        The human-facing run name (``""`` when unset).
    argv : list[str]
        The full process command line (typically ``sys.argv``).
    resumed_from : str | None
        Basename of the parent run this one resumed from (``--resume``), or ``None`` for a
        fresh run. Records the fork lineage a resume creates (a resume trains in a new run
        directory rather than continuing the parent in place).
    rl_backend : str | None
        Eden's runner-backend key (``"rsl_rl"`` / ``"rl_games"``) — the library that will
        write this run's checkpoints. Its installed version is resolved and recorded under
        ``rl_library`` (see :func:`get_rl_library`); ``None`` records ``rl_library: null``.

    Returns
    -------
    dict
        The JSON-serializable manifest.
    """
    return {
        "schema_version": RUN_MANIFEST_SCHEMA_VERSION,
        "run_id": run_id,
        "run_name": run_name,
        "timestamp": timestamp,
        "log_dir": os.path.abspath(log_dir),
        "task": task,
        "seed": seed,
        "world_size": world_size,
        "resumed_from": resumed_from,
        "argv": list(argv),
        # The release version, the commit (a clone's HEAD, else null) and whether that tree
        # was dirty. Genesis is a dependency with no working tree of its own, so it has no
        # dirty flag; its commit is the one a git+ or editable install records.
        "eden": get_eden_run_provenance(),
        "genesis": get_package_provenance("genesis-world"),
        # The RL library owns the checkpoint layout, so replaying a `.pt` needs a
        # compatible version of *it*, not just of eden/genesis.
        "rl_library": get_rl_library(rl_backend),
    }

write_run_manifest ¶

write_run_manifest(log_dir: str, manifest: dict[str, Any]) -> str

Write manifest to <log_dir>/run.json and return the path.

Overwrites any existing manifest (a run writes it exactly once, at launch). The directory is assumed to exist: the caller resolves log_dir through eden.log_dir, which creates the directory on first access.

PARAMETER	DESCRIPTION
`log_dir`	The run's (already-created) log directory. TYPE: `str`
`manifest`	The manifest to write, from :func:`build_run_manifest`. TYPE: `dict`

RETURNS	DESCRIPTION
`str`	The path to the written `run.json`.

Source code in eden/utils/run_manifest.py

def write_run_manifest(log_dir: str, manifest: dict[str, Any]) -> str:
    """Write ``manifest`` to ``<log_dir>/run.json`` and return the path.

    Overwrites any existing manifest (a run writes it exactly once, at launch). The
    directory is assumed to exist: the caller resolves ``log_dir`` through
    ``eden.log_dir``, which creates the directory on first access.

    Parameters
    ----------
    log_dir : str
        The run's (already-created) log directory.
    manifest : dict
        The manifest to write, from :func:`build_run_manifest`.

    Returns
    -------
    str
        The path to the written ``run.json``.
    """
    path = os.path.join(log_dir, RUN_MANIFEST_FILENAME)
    with open(path, "w") as f:
        json.dump(manifest, f, indent=2)
    return path

The name pip install and importlib.metadata know Eden by (not its import name), and how to install an extra.

Eden imports as eden and is distributed as eden-world: the PyPI name eden belongs to an unrelated Kivy project, so pip install eden installs the wrong package. [project].name in pyproject.toml is the source of truth; tests/test_pyproject.py pins :data:DISTRIBUTION_NAME to it, and checks that every pip install "<name>[<extra>]" written in eden/, docs/ and examples/ names this distribution and an extra it declares.

A message that tells the reader how to install an extra says it with :func:install_hint, which names both commands: whoever reads it may have Eden from PyPI or a clone. An extra in :data:UV_ONLY_EXTRAS gets the checkout's command alone, with the reason pip cannot install it.

This module is the one home for what is particular about installing an extra: each such fact is a table here, which :func:pip_install_extra / :func:uv_sync_extra read and tests/test_pyproject.py ties to pyproject.toml. A call site never writes its own install command. It imports nothing beyond the standard library, so import eden can read :func:get_eden_version without loading torch.

Attributes¶

DISTRIBUTION_NAME `module-attribute` ¶

DISTRIBUTION_NAME = 'eden-world'

UNPUBLISHED_REQUIREMENTS `module-attribute` ¶

UNPUBLISHED_REQUIREMENTS: dict[str, tuple[str, ...]] = {'metaquest': ('televuer @ git+https://github.com/Kashu7100/televuer@a1d8d6a0784fb289ddcde9aa108b92c7cf52f41b',), 'flashsac': ('rsl-rl-flashsac @ git+https://github.com/Holiday-Robot/rsl_rl_flashsac@9822b1b7e0b5cce45e7f79d6b3e2c169fa042faa',)}

REINSTALLED_PACKAGES `module-attribute` ¶

REINSTALLED_PACKAGES: dict[str, str] = {'onnx': 'onnxruntime', 'onnx-gpu': 'onnxruntime-gpu'}

UV_ONLY_EXTRAS `module-attribute` ¶

UV_ONLY_EXTRAS: dict[str, str] = {'soma-retargeter': "eden-soma-retargeter 0.4.0 pins trimesh==4.11.3 while genesis-world needs trimesh>=5, and only the checkout's uv override reconciles them"}

Functions:¶

get_eden_version ¶

get_eden_version() -> str

The installed version of Eden's distribution (eden-world, e.g. 0.1.0), or "" when it is not installed under that name (a source tree on PYTHONPATH, say). eden.__version__ reads it.

The same in a checkout as in a wheel, so it suits a stamp that should change only with a release: a checkout's commit (:func:eden.utils.misc.get_eden_commit) changes with every commit.

Source code in eden/utils/distribution.py

def get_eden_version() -> str:
    """The installed version of Eden's distribution (``eden-world``, e.g. ``0.1.0``), or ``""`` when it is not
    installed under that name (a source tree on ``PYTHONPATH``, say).
    ``eden.__version__`` reads it.

    The same in a checkout as in a wheel, so it suits a stamp that should change only with a release: a checkout's
    commit (:func:`eden.utils.misc.get_eden_commit`) changes with every commit.
    """
    return installed_version(DISTRIBUTION_NAME) or ""

installed_version ¶

installed_version(name: str) -> str | None

The installed version of distribution name (its pip install name), or None when it is not installed under that name. Never raises, so a message or a provenance record can call it.

Source code in eden/utils/distribution.py

def installed_version(name: str) -> str | None:
    """The installed version of distribution ``name`` (its ``pip install`` name), or ``None`` when it is not
    installed under that name. Never raises, so a message or a provenance record can call it.
    """
    try:
        return metadata.version(name)
    except metadata.PackageNotFoundError:
        return None

direct_url_record ¶

direct_url_record(dist: Distribution) -> dict

dist's PEP 610 direct_url.json (how a git+, local-directory or editable install was made), or {} when it has none: an install from an index records none, and a corrupt record says nothing either.

Source code in eden/utils/distribution.py

def direct_url_record(dist: metadata.Distribution) -> dict:
    """``dist``'s PEP 610 ``direct_url.json`` (how a ``git+``, local-directory or editable install was made), or ``{}``
    when it has none: an install from an index records none, and a corrupt record says nothing either.
    """
    try:
        record = json.loads(dist.read_text("direct_url.json") or "{}")
    except ValueError:
        return {}
    return record if isinstance(record, dict) else {}

editable_checkout ¶

editable_checkout(dist: Distribution) -> Path | None

The checkout an editable install of dist points at, as its direct_url.json records it, or None when dist is not an editable install of a local directory. The directory may since have moved.

Source code in eden/utils/distribution.py

def editable_checkout(dist: metadata.Distribution) -> Path | None:
    """The checkout an editable install of ``dist`` points at, as its ``direct_url.json`` records it, or ``None`` when
    ``dist`` is not an editable install of a local directory. The directory may since have moved.
    """
    record = direct_url_record(dist)
    url = urlparse(str(record.get("url", "")))
    if not (record.get("dir_info", {}).get("editable") and url.scheme == "file"):
        return None
    return Path(url2pathname(url.path))

pip_install_extra ¶

pip_install_extra(extra: str) -> str

The shell command that installs Eden with extra, e.g. pip install "eden-world[vis]".

For an extra with :data:UNPUBLISHED_REQUIREMENTS the command installs those too, so it is complete as printed. For one in :data:REINSTALLED_PACKAGES whose wheel is already installed, it then reinstalls that wheel alone at the installed version: pip install "eden-world[onnx]" && pip install --force-reinstall --no-deps "onnxruntime==1.27.0" (--no-deps, or pip would reinstall torch with it). So the command reads this environment. Each requirement is double-quoted: single quotes reach pip verbatim from Windows cmd, double quotes work in every shell.

Source code in eden/utils/distribution.py

def pip_install_extra(extra: str) -> str:
    """The shell command that installs Eden with ``extra``, e.g. ``pip install "eden-world[vis]"``.

    For an extra with :data:`UNPUBLISHED_REQUIREMENTS` the command installs those too, so it is complete as printed.
    For one in :data:`REINSTALLED_PACKAGES` whose wheel is already installed, it then reinstalls that wheel alone at
    the installed version: ``pip install "eden-world[onnx]" && pip install --force-reinstall --no-deps
    "onnxruntime==1.27.0"`` (``--no-deps``, or pip would reinstall torch with it). So the command reads this
    environment. Each requirement is double-quoted: single quotes reach pip verbatim from Windows ``cmd``, double
    quotes work in every shell.
    """
    command = " ".join(
        [
            "pip install",
            f'"{DISTRIBUTION_NAME}[{extra}]"',
            *(f'"{req}"' for req in UNPUBLISHED_REQUIREMENTS.get(extra, ())),
        ]
    )
    package = REINSTALLED_PACKAGES.get(extra)
    installed = installed_version(package) if package else None
    if installed:
        command += f' && pip install --force-reinstall --no-deps "{package}=={installed}"'
    return command

uv_sync_extra ¶

uv_sync_extra(extra: str) -> str

The command that adds extra to a checkout's environment, e.g. uv sync --inexact --extra vis.

--inexact because uv sync is exact by default: it uninstalls every extra and group the command does not name, so a reader synced with --all-extras --all-groups who pasted a sync of --extra vis alone from an error message would lose everything else. tests/test_pyproject.py holds the docs and examples to the same rule.

For an extra with :data:UNPUBLISHED_REQUIREMENTS it also syncs the extra's <extra>-git dependency group, which holds them: uv sync --inexact --extra flashsac --group flashsac-git. For one in :data:REINSTALLED_PACKAGES it reinstalls that wheel: uv sync --inexact --extra onnx --reinstall-package onnxruntime.

Source code in eden/utils/distribution.py

def uv_sync_extra(extra: str) -> str:
    """The command that adds ``extra`` to a checkout's environment, e.g. ``uv sync --inexact --extra vis``.

    ``--inexact`` because ``uv sync`` is exact by default: it uninstalls every extra and group the command does not
    name, so a reader synced with ``--all-extras --all-groups`` who pasted a sync of ``--extra vis`` alone from an
    error message would lose everything else. ``tests/test_pyproject.py`` holds the docs and examples to the same rule.

    For an extra with :data:`UNPUBLISHED_REQUIREMENTS` it also syncs the extra's ``<extra>-git`` dependency group,
    which holds them: ``uv sync --inexact --extra flashsac --group flashsac-git``. For one in
    :data:`REINSTALLED_PACKAGES` it reinstalls that wheel: ``uv sync --inexact --extra onnx --reinstall-package
    onnxruntime``.
    """
    group = [f"--group {extra}-git"] if extra in UNPUBLISHED_REQUIREMENTS else []
    reinstall = [f"--reinstall-package {REINSTALLED_PACKAGES[extra]}"] if extra in REINSTALLED_PACKAGES else []
    return " ".join(["uv sync --inexact", f"--extra {extra}", *group, *reinstall])

install_hint ¶

install_hint(extra: str) -> str

How to install extra, for an error message or a help text, in both the ways a reader may need: with Eden from PyPI, and in a clone. For vis it reads::

`pip install "eden-world[vis]"`, or `uv sync --inexact --extra vis` in a checkout

Both commands are complete as printed (see :func:pip_install_extra and :func:uv_sync_extra). An extra in :data:UV_ONLY_EXTRAS gets the checkout's command alone, and why pip cannot install it.

Source code in eden/utils/distribution.py

def install_hint(extra: str) -> str:
    """How to install ``extra``, for an error message or a help text, in both the ways a reader may need: with Eden
    from PyPI, and in a clone. For ``vis`` it reads::

        `pip install "eden-world[vis]"`, or `uv sync --inexact --extra vis` in a checkout

    Both commands are complete as printed (see :func:`pip_install_extra` and :func:`uv_sync_extra`). An extra in
    :data:`UV_ONLY_EXTRAS` gets the checkout's command alone, and why pip cannot install it.
    """
    if extra in UV_ONLY_EXTRAS:
        return f"`{uv_sync_extra(extra)}` in a checkout of Eden (pip cannot install it: {UV_ONLY_EXTRAS[extra]})"
    return f"`{pip_install_extra(extra)}`, or `{uv_sync_extra(extra)}` in a checkout"

Smoke runs¶

Lightweight build-and-step smoke harness for any registered task.

build_and_step("<task>") (or eden smoke --task <name>) constructs a task config by name, builds its Genesis env, and steps it a few times with no action — a one-call end-to-end sanity check for refactors and cheap coverage. Task variants take the same route as the CLI's mod flags: build_and_step("move_forward", modifier_kwargs={"robot": "go1"}) is eden smoke --task move_forward --robot go1.

It encapsulates the steps that are otherwise easy to get wrong:

initialize Genesis first (done here if the caller hasn't),
pick MimicRLEnvBase vs RLEnvBase from the config type,
build() then reset() before stepping,
step(None) advances the sim with no action tensor.

Classes¶

Functions:¶

is_smoke ¶

is_smoke() -> bool

True in "smoke" mode — the few-step, headless run used for CI/sanity checks.

Controlled by EDEN_SMOKE (the canonical knob, set to 1 by eden examples smoke and the recipe test harness); PYTEST_VERSION (set by pytest) is a fallback so a bare test run is treated as smoke. Examples branch their step horizon on it: horizon = 5 if en.is_smoke() else 1000.

EDEN_SMOKE is parsed as a boolean, not by presence: EDEN_SMOKE=0 / false / no is a real off-switch. An explicitly-set value is authoritative — it wins over PYTEST_VERSION, so a test can opt back out of smoke mode with EDEN_SMOKE=0. An unset or empty value falls through to the PYTEST_VERSION presence check. (Falsey vocabulary matches eden.extensions.policies.onnx._env_bool.)

Source code in eden/utils/smoke.py

def is_smoke() -> bool:
    """True in "smoke" mode — the few-step, headless run used for CI/sanity checks.

    Controlled by ``EDEN_SMOKE`` (the canonical knob, set to ``1`` by ``eden examples smoke``
    and the recipe test harness); ``PYTEST_VERSION`` (set by pytest) is a fallback so a bare
    test run is treated as smoke. Examples branch their step horizon on it:
    ``horizon = 5 if en.is_smoke() else 1000``.

    ``EDEN_SMOKE`` is parsed as a boolean, not by presence: ``EDEN_SMOKE=0`` / ``false`` / ``no``
    is a real off-switch. An explicitly-set value is authoritative — it wins over
    ``PYTEST_VERSION``, so a test can opt back out of smoke mode with ``EDEN_SMOKE=0``. An unset
    or empty value falls through to the ``PYTEST_VERSION`` presence check. (Falsey vocabulary
    matches ``eden.extensions.policies.onnx._env_bool``.)
    """
    eden_smoke = os.environ.get("EDEN_SMOKE", "").strip().lower()
    if eden_smoke:
        return eden_smoke not in ("0", "false", "no")
    return bool(os.environ.get("PYTEST_VERSION"))

build_and_step ¶

build_and_step(task: Union[str, EdenConfig], *, modifier_kwargs: dict[str, Any] | None = None, num_envs: int = 2, num_steps: int = 2, eval_mode: bool = False, show_viewer: bool = False, backend=None) -> RLEnvBase

Build a task's env and step it num_steps times with no action.

One-shot: each call builds a fresh scene and does not destroy it, so it is meant for a single sanity check (or a session-scoped test), not a tight loop over many tasks in one process.

PARAMETER	DESCRIPTION
`task`	A registered task name, or an already-constructed config object. A name goes through `TASK_REGISTRY.build` so the task's TaskMods run — a mod-bound task is not buildable without them. TYPE: `str \| EdenConfig`
`modifier_kwargs`	TaskMod overrides forwarded to `TASK_REGISTRY.build(task, *modifier_kwargs)` — the Python side of `eden smoke --task <name> --<flag> <value>`, e.g. `{"robot": "go1"}`. Unknown flags raise the registry's `ValueError`. Only a task name* can take them: a config object is already built, mods included, so passing both raises `TypeError` rather than silently ignoring the kwargs. TYPE: `dict[str, Any]` DEFAULT: `None`
`num_envs`	Number of parallel envs to build. TYPE: `int` DEFAULT: `2`
`num_steps`	Number of no-action `step(None)` calls to run after the initial reset. TYPE: `int` DEFAULT: `2`
`eval_mode`	Build in eval mode (training/randomization off). Defaults to False so the full training path is exercised. TYPE: `bool` DEFAULT: `False`
`show_viewer`	Open the Genesis viewer. TYPE: `bool` DEFAULT: `False`
`backend`	Passed to `en.init` only when Genesis isn't initialized yet. `None` leaves `en.init`'s default; pass `gs.gpu` / `gs.cpu` to choose. TYPE: `genesis backend` DEFAULT: `None`

RETURNS	DESCRIPTION
`RLEnvBase`	The built, stepped env — returned so callers can inspect it. Any error in the build/step path propagates; surfacing it is the point.

Source code in eden/utils/smoke.py

def build_and_step(
    task: "Union[str, EdenConfig]",
    *,
    modifier_kwargs: dict[str, Any] | None = None,
    num_envs: int = 2,
    num_steps: int = 2,
    eval_mode: bool = False,
    show_viewer: bool = False,
    backend=None,
) -> "RLEnvBase":
    """Build a task's env and step it ``num_steps`` times with no action.

    One-shot: each call builds a fresh scene and does not destroy it, so it is
    meant for a single sanity check (or a session-scoped test), not a tight
    loop over many tasks in one process.

    Parameters
    ----------
    task : str | EdenConfig
        A registered task name, or an already-constructed config object. A name
        goes through ``TASK_REGISTRY.build`` so the task's TaskMods run — a
        mod-bound task is not buildable without them.
    modifier_kwargs : dict[str, Any], optional
        TaskMod overrides forwarded to ``TASK_REGISTRY.build(task, **modifier_kwargs)``
        — the Python side of ``eden smoke --task <name> --<flag> <value>``, e.g.
        ``{"robot": "go1"}``. Unknown flags raise the registry's ``ValueError``. Only
        a task *name* can take them: a config object is already built, mods included,
        so passing both raises ``TypeError`` rather than silently ignoring the kwargs.
    num_envs : int
        Number of parallel envs to build.
    num_steps : int
        Number of no-action ``step(None)`` calls to run after the initial reset.
    eval_mode : bool
        Build in eval mode (training/randomization off). Defaults to False so the
        full training path is exercised.
    show_viewer : bool
        Open the Genesis viewer.
    backend : genesis backend, optional
        Passed to ``en.init`` only when Genesis isn't initialized yet. ``None``
        leaves ``en.init``'s default; pass ``gs.gpu`` / ``gs.cpu`` to choose.

    Returns
    -------
    RLEnvBase
        The built, stepped env — returned so callers can inspect it. Any error
        in the build/step path propagates; surfacing it is the point.
    """
    import eden as en
    import genesis as gs

    from eden.envs.base import env_class_for_config
    from eden.tasks import TASK_REGISTRY
    from eden.utils.configs import EdenConfig

    if isinstance(task, EdenConfig) and modifier_kwargs:
        raise TypeError(
            f"build_and_step() got modifier_kwargs {sorted(modifier_kwargs)} together with an already-built "
            f"{type(task).__name__}. TaskMods run while a task is built from its name, so either pass the task "
            "name here or apply them yourself with TASK_REGISTRY.build(name, **modifier_kwargs) and pass the result."
        )

    if not getattr(gs, "_initialized", False):
        en.init(**({"backend": backend} if backend is not None else {}))

    cfg = task if isinstance(task, EdenConfig) else TASK_REGISTRY.build(task, **(modifier_kwargs or {}))
    cfg.env_options.num_envs = num_envs

    env = env_class_for_config(cfg).from_config(cfg, show_viewer=show_viewer, eval_mode=eval_mode)
    env.build()
    env.reset()
    for _ in range(num_steps):
        env.step(None)
    return env

Profiling¶

Per-term cost profilers for the step loop.

On-demand, one-shot per-term cost profiler for a single task.

Why this exists

A term that is correct but slow is invisible. Nothing in a normal run tells the author that the observation term they just wrote allocates a torch.tensor(...) constant every step and now dominates rollout collection — the only symptom is a training run that feels slow. Measured on the Go1 backflip benchmark, two runs of the same task on the same GPU differed 35x in throughput purely because of per-call allocations inside a task's own terms (see docs/guides/terms.md).

This profiler makes that cost investigable on demand. eden smoke --task <name> --profile builds the task, steps it a bounded window, attributes wall time to each term the per-step managers run, and writes a ranked, machine-readable term_cost.json — so an author (or a coding agent that just wrote the term) can see which term is the bottleneck and what fraction of the step it costs.

Relationship to :class:~eden.utils.profiling.step_profiler.RLStepProfiler

That profiler attaches to a caller-driven loop (training / rollout) and also rolls up per-manager overhead; this one owns a short self-driven window and emits a compact ranking with no manager rollup. They share the term-discovery and timing mechanism in :mod:eden.utils.profiling.core, so both stay faithful to the same manager internals and the same both-sides CUDA sync.

Purely additive, zero overhead when off

Nothing here runs unless a caller asks for it. The profiler wraps term.compute from the outside for the measurement window only, times it with a CUDA sync on both sides so the number reflects the term's own GPU work, then restores every original method. A run without --profile creates no profiler object and wraps no method anywhere in the framework.

Classes¶

TermCost `dataclass` ¶

TermCost(manager: str, term: str, us_per_step: float, calls_per_step: float, pct_step: float = 0.0)

One term's measured cost, normalized per env step.

Attributes¶

manager `instance-attribute` ¶

manager: str

term `instance-attribute` ¶

term: str

us_per_step `instance-attribute` ¶

us_per_step: float

calls_per_step `instance-attribute` ¶

calls_per_step: float

pct_step `class-attribute` `instance-attribute` ¶

pct_step: float = 0.0

Methods:¶

ProfileReport `dataclass` ¶

ProfileReport(num_steps: int, step_us: float, sync: bool, rows: list[TermCost] = list())

Ranked per-term cost for a profiled window. Agent- and human-readable.

Attributes¶

num_steps `instance-attribute` ¶

num_steps: int

step_us `instance-attribute` ¶

step_us: float

sync `instance-attribute` ¶

sync: bool

rows `class-attribute` `instance-attribute` ¶

rows: list[TermCost] = field(default_factory=list)

HINT `class-attribute` `instance-attribute` ¶

HINT = 'Hot-path rules: resolve indices and allocate constant tensors in build(), not compute(); avoid .item()/.cpu()/bool(tensor). See docs/guides/terms.md.'

NOTE `class-attribute` `instance-attribute` ¶

NOTE = 'With sync=true, us/step and step_us are wall time with CUDA synced on both sides, so they run higher than an unsynced run; the pct_step shares are the faithful signal.'

Methods:¶

as_dict ¶

as_dict() -> dict

Machine-readable summary (JSON-serializable), terms ranked costliest-first.

Source code in eden/utils/profiling/term_profiler.py

def as_dict(self) -> dict:
    """Machine-readable summary (JSON-serializable), terms ranked costliest-first."""
    return {
        "num_steps": self.num_steps,
        "step_us": round(self.step_us, 1),
        "sync": self.sync,
        "terms": [
            {
                "manager": r.manager,
                "term": r.term,
                "us_per_step": round(r.us_per_step, 1),
                "pct_step": round(r.pct_step, 1),
                "calls_per_step": round(r.calls_per_step, 2),
            }
            for r in self.rows
        ],
        "hint": self.HINT,
        "note": self.NOTE,
    }

to_json ¶

to_json(path: str) -> None

Write :meth:as_dict to path (creating parent dirs).

Source code in eden/utils/profiling/term_profiler.py

def to_json(self, path: str) -> None:
    """Write :meth:`as_dict` to ``path`` (creating parent dirs)."""
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w") as fh:
        json.dump(self.as_dict(), fh, indent=2)

render ¶

render(top: int | None = 15) -> str

Human-readable table, costliest term first (top rows, then a rollup).

Source code in eden/utils/profiling/term_profiler.py

def render(self, top: int | None = 15) -> str:
    """Human-readable table, costliest term first (``top`` rows, then a rollup)."""
    term_total = sum(r.us_per_step for r in self.rows)
    lines = [
        f"Per-term compute cost — {self.num_steps} steps, "
        f"{self.step_us:.0f} us/env-step ({term_total:.0f} us/step in terms).",
        f"  {'us/step':>10}  {'%step':>6}  {'calls':>5}  term",
    ]
    shown = self.rows if top is None else self.rows[:top]
    for r in shown:
        lines.append(
            f"  {r.us_per_step:>10.0f}  {r.pct_step:>5.1f}%  {r.calls_per_step:>5.1f}  {r.manager}:{r.term}"
        )
    if top is not None and len(self.rows) > top:
        rest = sum(r.us_per_step for r in self.rows[top:])
        lines.append(f"  {rest:>10.0f}  {'':>6}  {'':>5}  ({len(self.rows) - top} more terms)")
    lines.append("  " + self.HINT)
    if self.sync:
        lines.append("  " + self.NOTE)
    return "\n".join(lines)

Functions:¶

profile_terms ¶

profile_terms(env: 'EnvBase', *, warmup: int = 10, measure: int = 50, step: Callable[[], object] | None = None, sync: bool | None = None) -> ProfileReport

Attribute wall time to each per-step term over a bounded window.

Runs warmup steps uninstrumented (kernel compilation and lazy allocation land there), then wraps every term's compute, runs measure steps, and restores the originals. The env is left stepped, not reset — the caller owns it.

Event terms are excluded (they fire on reset / interval, not per step) and so are action terms (their dominant cost is in apply_actions, not the timed compute). The reported step_us is the whole env.step() measured under the same both-sides sync, so it runs high in absolute terms but the per-term pct_step shares are faithful (see :class:~eden.utils.profiling.step_profiler.RLStepProfiler for the same caveat).

PARAMETER	DESCRIPTION
`env`	A built, reset env. TYPE: `EnvBase`
`warmup`	Uninstrumented steps before measuring. TYPE: `int` DEFAULT: `10`
`measure`	Instrumented steps; per-term cost is normalized by this (a term computed more than once per step accumulates all its calls into that step's cost). TYPE: `int` DEFAULT: `50`
`step`	Zero-arg step to call each iteration. Defaults to `lambda: env.step(None)`. TYPE: `callable` DEFAULT: `None`
`sync`	Synchronize CUDA around each timed call. Defaults to whether CUDA is present. TYPE: `bool` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProfileReport`	Ranked per-term cost plus the measured env-step time.

Source code in eden/utils/profiling/term_profiler.py

def profile_terms(
    env: "EnvBase",
    *,
    warmup: int = 10,
    measure: int = 50,
    step: Callable[[], object] | None = None,
    sync: bool | None = None,
) -> ProfileReport:
    """Attribute wall time to each per-step term over a bounded window.

    Runs ``warmup`` steps uninstrumented (kernel compilation and lazy allocation
    land there), then wraps every term's ``compute``, runs ``measure`` steps, and
    restores the originals. The env is left stepped, not reset — the caller owns it.

    Event terms are excluded (they fire on reset / interval, not per step) and so
    are action terms (their dominant cost is in ``apply_actions``, not the timed
    ``compute``). The reported ``step_us`` is the whole ``env.step()`` measured
    under the same both-sides sync, so it runs high in absolute terms but the
    per-term ``pct_step`` shares are faithful (see
    :class:`~eden.utils.profiling.step_profiler.RLStepProfiler` for the same caveat).

    Parameters
    ----------
    env : EnvBase
        A built, reset env.
    warmup : int
        Uninstrumented steps before measuring.
    measure : int
        Instrumented steps; per-term cost is normalized by this (a term computed
        more than once per step accumulates all its calls into that step's cost).
    step : callable, optional
        Zero-arg step to call each iteration. Defaults to ``lambda: env.step(None)``.
    sync : bool, optional
        Synchronize CUDA around each timed call. Defaults to whether CUDA is present.

    Returns
    -------
    ProfileReport
        Ranked per-term cost plus the measured env-step time.
    """
    synchronize = resolve_sync(sync)
    step = step or (lambda: env.step(None))

    for _ in range(warmup):
        step()

    timeable, skipped = discover_timeable(env, include_event=False, include_action=False)
    warn_untimed("profile_terms", skipped)

    records: dict[tuple[str, str], _Timing] = {}
    patches = []  # list[core.Patch]
    for manager, label, term in timeable:
        rec = records.setdefault((manager, label), _Timing())
        patches.append(time_method(term, "compute", rec.add, synchronize))

    step_time = _Timing()
    try:
        for _ in range(measure):
            time_call(step, step_time.add, synchronize)
    finally:
        for patch in patches:
            patch.restore()

    steps = max(measure, 1)
    step_us = step_time.total_s / steps * 1e6
    rows = [
        TermCost(
            manager=manager,
            term=label,
            us_per_step=rec.total_s / steps * 1e6,
            calls_per_step=rec.calls / steps,
        )
        for (manager, label), rec in records.items()
    ]
    rows.sort(key=lambda r: r.us_per_step, reverse=True)
    for r in rows:
        r.pct_step = (r.us_per_step / step_us * 100.0) if step_us > 0 else 0.0
    return ProfileReport(num_steps=measure, step_us=step_us, sync=synchronize is not None, rows=rows)

run_smoke_profile ¶

run_smoke_profile(task: 'str | EdenConfig', *, modifier_kwargs: dict[str, Any] | None = None, num_envs: int = 2, warmup: int = 10, measure: int = 50, eval_mode: bool = False, out_dir: str | None = None) -> tuple[ProfileReport, str]

Build a task and profile its per-term cost. The smoke --profile entry point.

PARAMETER	DESCRIPTION
`task`	Registered task name or a constructed config. TYPE: `str \| EdenConfig`
`modifier_kwargs`	TaskMod overrides for a task name, forwarded to :func:`eden.utils.smoke.build_and_step` (and from there to `TASK_REGISTRY.build`); rejected alongside a constructed config. TYPE: `dict[str, Any]` DEFAULT: `None`
`num_envs`	Parallel envs to build. TYPE: `int` DEFAULT: `2`
`warmup`	Uninstrumented warmup steps. TYPE: `int` DEFAULT: `10`
`measure`	Instrumented steps. TYPE: `int` DEFAULT: `50`
`eval_mode`	Build in eval mode (training/randomization off). TYPE: `bool` DEFAULT: `False`
`out_dir`	Directory for `term_cost.json`. Defaults to `eden.get_log_dir()`. TYPE: `str` DEFAULT: `None`

RETURNS	DESCRIPTION
`tuple of (ProfileReport, str)`	The report and the path the JSON was written to.

Source code in eden/utils/profiling/term_profiler.py

def run_smoke_profile(
    task: "str | EdenConfig",
    *,
    modifier_kwargs: dict[str, Any] | None = None,
    num_envs: int = 2,
    warmup: int = 10,
    measure: int = 50,
    eval_mode: bool = False,
    out_dir: str | None = None,
) -> tuple[ProfileReport, str]:
    """Build a task and profile its per-term cost. The smoke ``--profile`` entry point.

    Parameters
    ----------
    task : str | EdenConfig
        Registered task name or a constructed config.
    modifier_kwargs : dict[str, Any], optional
        TaskMod overrides for a task *name*, forwarded to
        :func:`eden.utils.smoke.build_and_step` (and from there to
        ``TASK_REGISTRY.build``); rejected alongside a constructed config.
    num_envs : int
        Parallel envs to build.
    warmup : int
        Uninstrumented warmup steps.
    measure : int
        Instrumented steps.
    eval_mode : bool
        Build in eval mode (training/randomization off).
    out_dir : str, optional
        Directory for ``term_cost.json``. Defaults to ``eden.get_log_dir()``.

    Returns
    -------
    tuple of (ProfileReport, str)
        The report and the path the JSON was written to.
    """
    import eden as en

    from eden.utils.smoke import build_and_step

    env = build_and_step(task, modifier_kwargs=modifier_kwargs, num_envs=num_envs, num_steps=0, eval_mode=eval_mode)
    report = profile_terms(env, warmup=warmup, measure=measure)
    path = os.path.join(out_dir or en.get_log_dir(), "term_cost.json")
    report.to_json(path)
    return report, path

Live-loop per-term profiler for the RL step loop.

Instruments every manager term's compute() — plus each manager's top-level compute(), reset() and post_reset() and the other hooks the step loop calls on it (the action manager's apply_actions(), the motion manager's fill_data(), ...) — so the wall-clock cost of observations / rewards / terminations / commands / events / curriculum / metrics / actions / motion can be attributed term by term while a real loop drives the env. Use it to surface the framework-level overhead Eden spends dispatching and composing terms on top of the raw Genesis physics step.

The headline numbers are:

per-term mean/total wall time and share of the step,
per-manager time split into terms (its term calls) and overhead (the manager's own methods around them) — no split for a manager whose terms are not timed (the motion manager's run inside fill_data, not through compute),
the step-loop work no manager owns: the Genesis scene.step physics tick (physics) and the env's own reset work around the managers' resets (reset: scene.reset, solver bookkeeping, buffer zeroing, index handling), and
the unattributed remainder of the step — the Python in env.step between instrumented calls.

The per-term table and :meth:RLStepProfiler.as_dict's total_ms_per_step are inclusive: a call's wall time counts every instrumented call nested inside it. The rollup instead charges each call only its own time (its wall time minus that of the instrumented calls nested inside it), and only inside :meth:RLStepProfiler.step windows, so its rows partition the measured step exactly. A RESET event that fills the motion window is charged its own work under event and the fill under motion; reset excludes everything instrumented inside it. With sync=True the syncs around a nested call land in its caller's own time. A call that raises is still recorded, with its wall time up to the raise (no closing sync).

For the one-shot "profile the term I just wrote" workflow, see :func:eden.utils.profiling.term_profiler.profile_terms. Both share the term-discovery and timing mechanism in :mod:eden.utils.profiling.core.

GPU note

Manager terms launch CUDA kernels asynchronously, so naive host-side timing measures launch cost, not execution. With sync=True (default) the profiler calls :func:torch.cuda.synchronize around every measured call, yielding true wall time per term at the cost of serializing kernels that would otherwise overlap — absolute numbers run high but the relative breakdown is faithful. With sync=False you instead measure pure Python dispatch cost (useful for spotting per-term CPU overhead). The two views answer different questions.

Classes¶

TermTiming `dataclass` ¶

TermTiming(manager: str, label: str, kind: str, calls: int = 0, total_s: float = 0.0, self_s: float = 0.0, sumsq_s: float = 0.0, max_s: float = 0.0)

Running timing accumulator for a single instrumented callable.

Attributes¶

manager `instance-attribute` ¶

manager: str

label `instance-attribute` ¶

label: str

kind `instance-attribute` ¶

kind: str

calls `class-attribute` `instance-attribute` ¶

calls: int = 0

total_s `class-attribute` `instance-attribute` ¶

total_s: float = 0.0

self_s `class-attribute` `instance-attribute` ¶

self_s: float = 0.0

sumsq_s `class-attribute` `instance-attribute` ¶

sumsq_s: float = 0.0

max_s `class-attribute` `instance-attribute` ¶

max_s: float = 0.0

mean_s `property` ¶

mean_s: float

std_s `property` ¶

std_s: float

Methods:¶

add ¶

add(dt: float, self_dt: float | None = None) -> None

Source code in eden/utils/profiling/step_profiler.py

def add(self, dt: float, self_dt: float | None = None) -> None:
    self.calls += 1
    self.total_s += dt
    self.self_s += dt if self_dt is None else self_dt
    self.sumsq_s += dt * dt
    if dt > self.max_s:
        self.max_s = dt

RLStepProfiler ¶

RLStepProfiler(env: EnvBase, *, sync: bool = True)

Attribute RL-step wall time to individual manager terms.

Typical usage::

profiler = RLStepProfiler(env)
profiler.attach()
for _ in range(warmup):
    env.step(sample_action())
profiler.reset()            # discard warmup, keep instrumentation
for _ in range(steps):
    with profiler.step():   # times the whole env.step()
        env.step(sample_action())
print(profiler.report())
profiler.detach()

Or as a context manager that attaches/detaches automatically::

with RLStepProfiler(env) as profiler:
    ...

Source code in eden/utils/profiling/step_profiler.py

def __init__(self, env: EnvBase, *, sync: bool = True):
    self._env = env
    # Only sync when there's actually a CUDA device to wait on.
    self._synchronize = resolve_sync(sync)
    self._stats: dict[tuple[str, str, str], TermTiming] = {}
    self._patches: list[Patch] = []
    # One entry per instrumented call in progress: the wall time of the instrumented calls
    # nested inside it so far, subtracted from its own to give its self time.
    self._nested_s: list[float] = []
    self._step = TermTiming(manager="", label="step", kind="step")
    # Whether a ``step()`` window is open. Own time accrues only inside one, so the rollup
    # partitions the measured step; instrumented calls outside (an ``env.reset()`` between
    # timed steps) still reach the per-term table, and their wall time is ``_outside_s``.
    self._in_step = False
    self._outside_s = 0.0
    self._attached = False

Attributes¶

num_steps `property` ¶

num_steps: int

Methods:¶

attach ¶

attach() -> RLStepProfiler

Source code in eden/utils/profiling/step_profiler.py

def attach(self) -> RLStepProfiler:
    if self._attached:
        return self
    hooks = self._step_loop_hooks()
    # discover_timeable dedups a term shared across groups so it is wrapped once, not
    # double-counted, and drops any term that can't carry the timing override.
    timeable, skipped = discover_timeable(self._env, include_event=True)
    warn_untimed("RLStepProfiler", skipped)
    # A run of RESET terms that fires as one ``compute_batch`` never calls its terms' own ``compute``:
    # time the batch as one row, ``reset.<a>+<b>+...``, in their place.
    event_manager = getattr(self._env, "event_manager", None)
    batches = list(event_manager.iter_reset_batches()) if event_manager is not None else []
    batched = {id(term) for batch in batches for term in batch.terms}
    for mgr_name, label, term in timeable:
        if id(term) not in batched:
            self._wrap((mgr_name, label, "term"), term, "compute")
    for batch in batches:
        self._wrap(("event", f"reset.{'+'.join(batch.names)}", "term"), batch, "compute")
    for key, owner, method in hooks:
        self._wrap(key, owner, method)
    self._attached = True
    return self

detach ¶

detach() -> None

Source code in eden/utils/profiling/step_profiler.py

def detach(self) -> None:
    for patch in reversed(self._patches):
        patch.restore()
    self._patches.clear()
    self._attached = False

reset ¶

reset() -> None

Zero all accumulators (e.g. to discard warmup) without detaching.

Source code in eden/utils/profiling/step_profiler.py

def reset(self) -> None:
    """Zero all accumulators (e.g. to discard warmup) without detaching."""
    for rec in self._stats.values():
        rec.calls = 0
        rec.total_s = 0.0
        rec.self_s = 0.0
        rec.sumsq_s = 0.0
        rec.max_s = 0.0
    self._step = TermTiming(manager="", label="step", kind="step")
    self._outside_s = 0.0

step ¶

step() -> _StepTimer

Context manager timing one whole env.step() call.

Source code in eden/utils/profiling/step_profiler.py

def step(self) -> _StepTimer:
    """Context manager timing one whole ``env.step()`` call."""
    return self._StepTimer(self)

as_dict ¶

as_dict() -> dict[str, Any]

Machine-readable summary (JSON-serializable).

terms, managers (by manager) and loop (by step-loop row: physics, reset) hold one record per instrumented callable: every call, inclusive of the calls nested inside it (total_ms_per_step, pct_step), plus its own time inside step windows (self_ms_per_step). rollup is the report's rollup table: own time per row, summing to step_ms. outside_step_ms is the wall time of instrumented calls made outside :meth:step, which only the records count.

Source code in eden/utils/profiling/step_profiler.py

def as_dict(self) -> dict[str, Any]:
    """Machine-readable summary (JSON-serializable).

    ``terms``, ``managers`` (by manager) and ``loop`` (by step-loop row: ``physics``,
    ``reset``) hold one record per instrumented callable: every call, inclusive of the calls
    nested inside it (``total_ms_per_step``, ``pct_step``), plus its own time inside step
    windows (``self_ms_per_step``). ``rollup`` is the report's rollup table: own time per
    row, summing to ``step_ms``. ``outside_step_ms`` is the wall time of instrumented calls
    made outside :meth:`step`, which only the records count.
    """
    n = max(self.num_steps, 1)
    step_total = self._step.total_s

    def pack(r: TermTiming) -> dict[str, Any]:
        return {
            "manager": r.manager,
            "label": r.label,
            "kind": r.kind,
            "calls_per_step": r.calls / n,
            "mean_ms": r.mean_s * 1e3,
            "total_ms_per_step": r.total_s / n * 1e3,
            "self_ms_per_step": r.self_s / n * 1e3,
            "std_ms": r.std_s * 1e3,
            "max_ms": r.max_s * 1e3,
            "pct_step": (r.total_s / step_total * 100.0) if step_total > 0 else 0.0,
        }

    def pack_by_owner(kind: str) -> dict[str, dict[str, dict[str, Any]]]:
        return {
            owner: {label: pack(r) for label, r in entries.items()}
            for owner, entries in self._records_by_owner(kind).items()
        }

    terms = sorted(self._term_records(), key=lambda r: r.total_s, reverse=True)
    manager_rows, other_rows = self._rollup()
    return {
        "num_steps": self.num_steps,
        "sync": self._sync,
        "step_ms": self._step.mean_s * 1e3,
        "steps_per_s": (n / step_total) if step_total > 0 else 0.0,
        "terms": [pack(r) for r in terms],
        "managers": pack_by_owner("manager"),
        "loop": pack_by_owner("loop"),
        "rollup": {
            row: {
                "ms_per_step": own_s / n * 1e3,
                "terms_ms_per_step": None if t is None else t / n * 1e3,
                "overhead_ms_per_step": None if o is None else o / n * 1e3,
                "pct_step": (own_s / step_total * 100.0) if step_total > 0 else 0.0,
            }
            for row, own_s, t, o in [*manager_rows, *other_rows]
        },
        "outside_step_ms": self._outside_s * 1e3,
    }

report ¶

report(*, top: int | None = None, sort_by: str = 'total') -> str

Build human-readable breakdown tables.

PARAMETER	DESCRIPTION
`top`	If given, only show the `top` slowest terms. TYPE: `int` DEFAULT: `None`
`sort_by`	Term ordering, either `"total"` (default) or `"mean"`. TYPE: `str` DEFAULT: `'total'`

RETURNS	DESCRIPTION
`str`	The formatted breakdown tables.

Source code in eden/utils/profiling/step_profiler.py

def report(self, *, top: int | None = None, sort_by: str = "total") -> str:
    """Build human-readable breakdown tables.

    Parameters
    ----------
    top : int, optional
        If given, only show the ``top`` slowest terms.
    sort_by : str, optional
        Term ordering, either ``"total"`` (default) or ``"mean"``.

    Returns
    -------
    str
        The formatted breakdown tables.
    """
    n = max(self.num_steps, 1)
    step_total = self._step.total_s
    step_ms = self._step.mean_s * 1e3

    key = (lambda r: r.mean_s) if sort_by == "mean" else (lambda r: r.total_s)
    terms = sorted(self._term_records(), key=key, reverse=True)
    if top is not None:
        terms = terms[:top]

    def pct(total_s: float) -> str:
        return f"{(total_s / step_total * 100.0):.1f}" if step_total > 0 else "-"

    term_tbl = PrettyTable()
    # Calls outside a step window still count here, over the timed steps: say so on the table itself.
    outside = ", incl. calls outside step()" if self._outside_s > 0.0 else ""
    term_tbl.title = f"Per-Term Timing  ({self.num_steps} steps, sync={self._sync}{outside})"
    term_tbl.field_names = ["Manager", "Term", "Calls/step", "Mean (ms)", "Total/step (ms)", "% step"]
    term_tbl.align["Term"] = "l"
    term_tbl.align["Manager"] = "l"
    for r in terms:
        term_tbl.add_row(
            [
                r.manager,
                r.label,
                f"{r.calls / n:.2f}",
                f"{r.mean_s * 1e3:.4f}",
                f"{r.total_s / n * 1e3:.4f}",
                pct(r.total_s),
            ]
        )

    # Per-manager rollup, by own time so the rows sum to the step: each manager's terms and its
    # own methods around them (dispatch/bookkeeping), then the step-loop rows no manager owns,
    # and whatever no instrumented call covers.
    mgr_tbl = PrettyTable()
    mgr_tbl.title = "Per-Manager Rollup  (own time; rows sum to the step)"
    mgr_tbl.field_names = ["Manager", "Own/step (ms)", "Terms (ms)", "Overhead (ms)", "% step"]
    mgr_tbl.align["Manager"] = "l"

    def add_rollup_row(row: str, own_s: float, terms_s: float | None, overhead_s: float | None) -> None:
        split = ["-", "-"] if terms_s is None else [f"{terms_s / n * 1e3:.4f}", f"{overhead_s / n * 1e3:.4f}"]
        mgr_tbl.add_row([row, f"{own_s / n * 1e3:.4f}", *split, pct(own_s)])

    manager_rows, other_rows = self._rollup()
    for rollup_row in manager_rows:
        add_rollup_row(*rollup_row)
    mgr_tbl.add_row(["─" * 7, "─" * 13, "─" * 10, "─" * 13, "─" * 6])
    for rollup_row in other_rows:
        add_rollup_row(*rollup_row)

    steps_per_s = (n / step_total) if step_total > 0 else 0.0
    header = (
        f"RL step profile — {self.num_steps} steps, {self._env.num_envs} envs, "
        f"sync={self._sync}\n"
        f"  step: {step_ms:.3f} ms/step  ({steps_per_s:,.0f} steps/s, "
        f"{steps_per_s * self._env.num_envs:,.0f} env-steps/s)\n"
    )
    if self._outside_s > 0.0:
        header += (
            f"  {self._outside_s * 1e3:.3f} ms of instrumented calls ran outside profiler.step() "
            "(e.g. an env.reset() between timed steps): counted in the per-term table, not the rollup\n"
        )
    return header + "\n" + mgr_tbl.get_string() + "\n\n" + term_tbl.get_string() + "\n"

save_json ¶

save_json(path: str) -> None

Source code in eden/utils/profiling/step_profiler.py

def save_json(self, path: str) -> None:
    with open(path, "w") as f:
        json.dump(self.as_dict(), f, indent=2)

Functions:¶

Shared instrumentation core for the per-term profilers.

Both front-ends attribute wall time to individual manager terms by the same mechanism: find every term the step loop runs, wrap its compute() with a CUDA-synced timer, and restore it afterwards. That mechanism lives here so there is a single source of truth for

the coupling to manager term storage (:func:iter_terms), and
the reversible, sync-both-sides timing wrapper (:func:time_call times one call, :class:Patch installs and removes a wrapper, :func:time_method joins the two). The one-shot profiler uses :func:time_method; the live-loop one builds its own wrapper from :func:time_call and :class:Patch, to subtract the calls nested inside a call from its own time.

The two front-ends differ only in policy — what they measure and how they report it — not in this mechanism:

:class:eden.utils.profiling.step_profiler.RLStepProfiler attaches to a caller-driven loop (training / rollout), also times manager-level methods, and produces a full per-manager rollup. Use it to profile terms in an actual loop.
:func:eden.utils.profiling.term_profiler.profile_terms owns a short, self-driven window and produces a compact, agent-readable ranking. Use it to profile a term you just wrote, in isolation.

Attributes¶

FLAT_TERM_MANAGERS `module-attribute` ¶

FLAT_TERM_MANAGERS = ('action', 'reward', 'termination', 'command', 'curriculum', 'metric')

Classes¶

Patch ¶

Patch(owner: Any, name: str, wrapper: Callable)

A reversible replacement of one instance method with a wrapper.

Records whether the method was a per-instance attribute or an inherited class method, so :meth:restore puts back exactly what was there — deleting the instance shadow in the common class-method case rather than leaving a bound method attribute behind.

Source code in eden/utils/profiling/core.py

def __init__(self, owner: Any, name: str, wrapper: Callable) -> None:
    self._owner = owner
    self._name = name
    self._had_own = name in owner.__dict__
    self._original = owner.__dict__.get(name)  # None => inherited class method
    setattr(owner, name, wrapper)

Methods:¶

restore ¶

restore() -> None

Undo the patch, restoring the original bound/class method.

Source code in eden/utils/profiling/core.py

def restore(self) -> None:
    """Undo the patch, restoring the original bound/class method."""
    if self._had_own:
        setattr(self._owner, self._name, self._original)
    elif self._name in self._owner.__dict__:
        delattr(self._owner, self._name)

Functions:¶

resolve_sync ¶

resolve_sync(sync: bool | None) -> Callable[[], None] | None

Return the CUDA-synchronize callable to use around timed calls, or None.

PARAMETER	DESCRIPTION
`sync`	`True` to synchronize, `False` to skip, `None` to auto-enable only when a CUDA device is present. In every case a synchronize callable is returned only if CUDA is actually available (there is nothing to wait on otherwise). TYPE: `bool or None`

RETURNS	DESCRIPTION
`callable or None`	`torch.cuda.synchronize` when syncing, else `None`.

Source code in eden/utils/profiling/core.py

def resolve_sync(sync: bool | None) -> Callable[[], None] | None:
    """Return the CUDA-synchronize callable to use around timed calls, or ``None``.

    Parameters
    ----------
    sync : bool or None
        ``True`` to synchronize, ``False`` to skip, ``None`` to auto-enable only
        when a CUDA device is present. In every case a synchronize callable is
        returned only if CUDA is actually available (there is nothing to wait on
        otherwise).

    Returns
    -------
    callable or None
        ``torch.cuda.synchronize`` when syncing, else ``None``.
    """
    import torch

    if sync is None:
        sync = torch.cuda.is_available()
    return torch.cuda.synchronize if (sync and torch.cuda.is_available()) else None

iter_terms ¶

iter_terms(env: 'EnvBase', *, include_event: bool = True, include_action: bool = True) -> Iterator[tuple[str, str, Any]]

Yield (manager, label, term) for every term the step loop runs.

Composes the public term iterators of the three manager shapes: the observation manager's iter_terms() (per group), the event manager's iter_terms() (per mode), and the flat terms dict every :class:~eden.managers.base.FlatManagerBase exposes. Managers absent on this env (e.g. a task with no command manager) are skipped.

PARAMETER	DESCRIPTION
`env`	A built environment. TYPE: `EnvBase`
`include_event`	Whether to include event-manager terms. They fire on reset / interval through mode dispatch rather than once per step, so a per-step cost view (the one-shot profiler) excludes them; the live-loop profiler includes them. TYPE: `bool` DEFAULT: `True`
`include_action`	Whether to include action-manager terms. An action term's dominant cost is in `apply_actions` (per decimation substep), not the `compute` this module times, so a compute-only view (the one-shot profiler) excludes it to avoid reporting a misleading ~0; the live-loop profiler, which also times `apply_actions`, includes it. TYPE: `bool` DEFAULT: `True`

YIELDS	DESCRIPTION
`tuple of (str, str, object)`	The owning manager's name, a human-readable label (`"group.term"` for observations, `"mode.term"` for events, the bare term name otherwise), and the term instance.

Source code in eden/utils/profiling/core.py

def iter_terms(
    env: "EnvBase", *, include_event: bool = True, include_action: bool = True
) -> Iterator[tuple[str, str, Any]]:
    """Yield ``(manager, label, term)`` for every term the step loop runs.

    Composes the public term iterators of the three manager shapes: the
    observation manager's ``iter_terms()`` (per group), the event manager's
    ``iter_terms()`` (per mode), and the flat ``terms`` dict every
    :class:`~eden.managers.base.FlatManagerBase` exposes. Managers absent on this
    env (e.g. a task with no command manager) are skipped.

    Parameters
    ----------
    env : EnvBase
        A built environment.
    include_event : bool
        Whether to include event-manager terms. They fire on reset / interval
        through mode dispatch rather than once per step, so a per-step cost view
        (the one-shot profiler) excludes them; the live-loop profiler includes
        them.
    include_action : bool
        Whether to include action-manager terms. An action term's dominant cost
        is in ``apply_actions`` (per decimation substep), not the ``compute`` this
        module times, so a compute-only view (the one-shot profiler) excludes it
        to avoid reporting a misleading ~0; the live-loop profiler, which also
        times ``apply_actions``, includes it.

    Yields
    ------
    tuple of (str, str, object)
        The owning manager's name, a human-readable label (``"group.term"`` for
        observations, ``"mode.term"`` for events, the bare term name otherwise),
        and the term instance.
    """
    om = getattr(env, "observation_manager", None)
    if om is not None:
        for group, name, term in om.iter_terms():
            yield "observation", f"{group}.{name}", term
    if include_event:
        ev = getattr(env, "event_manager", None)
        if ev is not None:
            for mode, name, term in ev.iter_terms():
                yield "event", f"{mode}.{name}", term
    for mgr_name in FLAT_TERM_MANAGERS:
        if mgr_name == "action" and not include_action:
            continue
        mgr = getattr(env, f"{mgr_name}_manager", None)
        if mgr is None:
            continue
        for name, term in mgr.terms.items():
            yield mgr_name, name, term

can_time ¶

can_time(owner: Any) -> bool

Whether owner can carry the per-instance compute override the timers install.

:func:time_method shadows the method with an instance attribute, so the owner needs a writable __dict__. A term defining __slots__ without one cannot be wrapped — the setattr would raise. No built-in term is slotted, but an author could write one, so :func:discover_timeable skips such a term instead of crashing the profiler. Manager instances always qualify.

Source code in eden/utils/profiling/core.py

def can_time(owner: Any) -> bool:
    """Whether ``owner`` can carry the per-instance ``compute`` override the timers install.

    :func:`time_method` shadows the method with an instance attribute, so the owner
    needs a writable ``__dict__``. A term defining ``__slots__`` without one cannot
    be wrapped — the ``setattr`` would raise. No built-in term is slotted, but an
    author could write one, so :func:`discover_timeable` skips such a term instead
    of crashing the profiler. Manager instances always qualify.
    """
    return hasattr(owner, "__dict__")

discover_timeable ¶

discover_timeable(env: 'EnvBase', *, include_event: bool = True, include_action: bool = True) -> tuple[list[tuple[str, str, Any]], list[str]]

Split :func:iter_terms into the terms the profilers can time and the labels they can't.

Dedups by term identity (a term shared across groups is timed once, not double-counted) and drops any term whose compute is missing or which cannot be wrapped (:func:can_time). Both front-ends discover through this, so the "which terms are timeable" policy has a single home.

PARAMETER	DESCRIPTION
`env`	A built environment. TYPE: `EnvBase`
`include_event`	Forwarded to :func:`iter_terms`. TYPE: `bool` DEFAULT: `True`
`include_action`	Forwarded to :func:`iter_terms`. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`tuple of (list, list)`	`(timeable, skipped_labels)` — the `(manager, label, term)` triples to wrap, and the labels dropped as un-timeable. Pass the latter to :func:`warn_untimed` so the omission is surfaced rather than silent.

Source code in eden/utils/profiling/core.py

def discover_timeable(
    env: "EnvBase", *, include_event: bool = True, include_action: bool = True
) -> tuple[list[tuple[str, str, Any]], list[str]]:
    """Split :func:`iter_terms` into the terms the profilers can time and the labels they can't.

    Dedups by term identity (a term shared across groups is timed once, not
    double-counted) and drops any term whose ``compute`` is missing or which cannot
    be wrapped (:func:`can_time`). Both front-ends discover through this, so the
    "which terms are timeable" policy has a single home.

    Parameters
    ----------
    env : EnvBase
        A built environment.
    include_event, include_action : bool
        Forwarded to :func:`iter_terms`.

    Returns
    -------
    tuple of (list, list)
        ``(timeable, skipped_labels)`` — the ``(manager, label, term)`` triples to
        wrap, and the labels dropped as un-timeable. Pass the latter to
        :func:`warn_untimed` so the omission is surfaced rather than silent.
    """
    timeable: list[tuple[str, str, Any]] = []
    skipped: list[str] = []
    seen: set[int] = set()
    for manager, label, term in iter_terms(env, include_event=include_event, include_action=include_action):
        if id(term) in seen:
            continue
        seen.add(id(term))
        if not callable(getattr(term, "compute", None)):
            continue
        if not can_time(term):
            skipped.append(label)
            continue
        timeable.append((manager, label, term))
    return timeable, skipped

warn_untimed ¶

warn_untimed(context: str, labels: list[str]) -> None

Warn that labels were skipped as un-timeable; a no-op when empty.

Keeps the message identical across both front-ends. Imports eden lazily so importing this module never triggers the package init.

Source code in eden/utils/profiling/core.py

def warn_untimed(context: str, labels: list[str]) -> None:
    """Warn that ``labels`` were skipped as un-timeable; a no-op when empty.

    Keeps the message identical across both front-ends. Imports ``eden`` lazily so
    importing this module never triggers the package init.
    """
    if not labels:
        return
    import eden as en

    en.logger.warning(
        f"{context}: skipped {len(labels)} term(s) that define __slots__ (no __dict__) and "
        f"cannot be timed — omitted from the profile: {', '.join(labels)}"
    )

time_call ¶

time_call(fn: Callable[[], Any], on_dt: Callable[[float], None], synchronize: Callable[[], None] | None) -> Any

Run fn once, reporting its wall time to on_dt.

Syncs on both sides when synchronize is given: the pre-sync drains kernels queued by earlier work so they are not charged here, the post-sync waits out this call's own kernels. Returns fn's result.

A call that raises is still reported, with its wall time up to the raise, and the exception propagates. It skips the post-sync: after a CUDA error that sync would raise again, over the original exception.

Source code in eden/utils/profiling/core.py

def time_call(fn: Callable[[], Any], on_dt: Callable[[float], None], synchronize: Callable[[], None] | None) -> Any:
    """Run ``fn`` once, reporting its wall time to ``on_dt``.

    Syncs on *both* sides when ``synchronize`` is given: the pre-sync drains
    kernels queued by earlier work so they are not charged here, the post-sync
    waits out this call's own kernels. Returns ``fn``'s result.

    A call that raises is still reported, with its wall time up to the raise, and
    the exception propagates. It skips the post-sync: after a CUDA error that sync
    would raise again, over the original exception.
    """
    if synchronize is not None:
        synchronize()
    t0 = perf_counter()
    try:
        out = fn()
    except BaseException:
        on_dt(perf_counter() - t0)
        raise
    if synchronize is not None:
        synchronize()
    on_dt(perf_counter() - t0)
    return out

time_method ¶

time_method(owner: Any, name: str, on_dt: Callable[[float], None], synchronize: Callable[[], None] | None) -> Patch

Wrap owner.name so each call's wall time is reported to on_dt.

Returns a :class:Patch; call :meth:Patch.restore to remove the wrapper. Timing is delegated to :func:time_call, so both levels of instrumentation share one clock and one sync policy.

Source code in eden/utils/profiling/core.py

def time_method(owner: Any, name: str, on_dt: Callable[[float], None], synchronize: Callable[[], None] | None) -> Patch:
    """Wrap ``owner.name`` so each call's wall time is reported to ``on_dt``.

    Returns a :class:`Patch`; call :meth:`Patch.restore` to remove the wrapper.
    Timing is delegated to :func:`time_call`, so both levels of instrumentation
    share one clock and one sync policy.
    """
    orig = getattr(owner, name)

    def wrapper(*args, **kwargs):
        return time_call(lambda: orig(*args, **kwargs), on_dt, synchronize)

    return Patch(owner, name, wrapper)

Strings, formatting and warnings¶

term_subject(term) renders the "<Class> on entity '<name>'" subject an entity-bound term opens its errors with (see Error Messages).

String helpers: glob name-matching, and rendering name lists and term subjects for a log line.

Functions:¶

is_glob_pattern ¶

is_glob_pattern(pattern: str) -> bool

Best-effort heuristic to determine whether the provided pattern looks like a glob.

Source code in eden/utils/string.py

def is_glob_pattern(pattern: str) -> bool:
    """Best-effort heuristic to determine whether the provided pattern looks like a glob."""
    if not isinstance(pattern, str):
        return False
    has_glob_wildcards = bool(_GLOB_WILDCARD_PATTERN.search(pattern))
    if not has_glob_wildcards:
        return False
    if _REGEX_INDICATOR_PATTERN.search(pattern):
        return False
    if _DOT_STAR_PATTERN.search(pattern):
        return False
    return True

name_matches ¶

name_matches(pattern: str, name: str, use_glob: bool | None = None) -> bool

Whether pattern, read as a pattern, matches name.

A glob (per :func:is_glob_pattern, unless use_glob forces the mode) is matched with :func:fnmatch.fnmatchcase; anything else is a regular expression that must match the whole name. This is the pattern half of :func:key_selects only: it sees one name, so it cannot take a key that is exactly one of the names literally. To ask what a config key selects, use :func:key_selects.

PARAMETER	DESCRIPTION
`pattern`	The glob or regular expression. TYPE: `str`
`name`	The candidate name. TYPE: `str`
`use_glob`	`True`/`False` forces glob/regex matching; `None` infers it from `pattern`. TYPE: `bool \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	`True` when `pattern` matches all of `name`.

Source code in eden/utils/string.py

def name_matches(pattern: str, name: str, use_glob: bool | None = None) -> bool:
    """Whether ``pattern``, read as a pattern, matches ``name``.

    A glob (per :func:`is_glob_pattern`, unless ``use_glob`` forces the mode) is matched with
    :func:`fnmatch.fnmatchcase`; anything else is a regular expression that must match the whole
    name. This is the pattern half of :func:`key_selects` only: it sees one name, so it cannot take a
    key that is exactly one of the names literally. To ask what a config key selects, use
    :func:`key_selects`.

    Parameters
    ----------
    pattern: str
        The glob or regular expression.
    name: str
        The candidate name.
    use_glob: bool | None
        ``True``/``False`` forces glob/regex matching; ``None`` infers it from ``pattern``.

    Returns
    -------
    bool
        ``True`` when ``pattern`` matches all of ``name``.
    """
    glob = is_glob_pattern(pattern) if use_glob is None else use_glob
    if glob:
        return fnmatch.fnmatchcase(name, pattern)
    return re.fullmatch(pattern, name) is not None

key_selects ¶

key_selects(key: str, name: str, names: Collection[str], use_glob: bool | None = None) -> bool

Whether the config key key selects name, one of names.

A key that is exactly one of names selects that name alone; any other key is a glob or a regex (:func:name_matches). A name can carry a pattern metacharacter (pin.1, arm(2), an exporter's link.001): read as a regex, pin.1 would also select pin_1, and arm(2) would select nothing, not even itself. :func:resolve_matching_names and :func:resolve_matching_names_values select with this test, so a caller matching keys one name at a time agrees with them.

PARAMETER	DESCRIPTION
`key`	The name, glob or regex. TYPE: `str`
`name`	The candidate name. TYPE: `str`
`names`	Every name the key could select, which decides whether `key` is one of them. Pass a set when asking about many names. TYPE: `Collection[str]`
`use_glob`	`True`/`False` forces glob/regex matching of a pattern key; `None` infers it. TYPE: `bool \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	`True` when `key` selects `name`.

Source code in eden/utils/string.py

def key_selects(key: str, name: str, names: Collection[str], use_glob: bool | None = None) -> bool:
    """Whether the config key ``key`` selects ``name``, one of ``names``.

    A key that is exactly one of ``names`` selects that name alone; any other key is a glob or a regex
    (:func:`name_matches`). A name can carry a pattern metacharacter (``pin.1``, ``arm(2)``, an
    exporter's ``link.001``): read as a regex, ``pin.1`` would also select ``pin_1``, and ``arm(2)``
    would select nothing, not even itself. :func:`resolve_matching_names` and
    :func:`resolve_matching_names_values` select with this test, so a caller matching keys one name at
    a time agrees with them.

    Parameters
    ----------
    key: str
        The name, glob or regex.
    name: str
        The candidate name.
    names: Collection[str]
        Every name the key could select, which decides whether ``key`` is one of them. Pass a set when
        asking about many names.
    use_glob: bool | None
        ``True``/``False`` forces glob/regex matching of a pattern key; ``None`` infers it.

    Returns
    -------
    bool
        ``True`` when ``key`` selects ``name``.
    """
    if key in names:
        return key == name
    return name_matches(key, name, use_glob=use_glob)

first_few ¶

first_few(names: Iterable[str], limit: int = 5) -> str

Render names for a log line, truncating a long list to its first few plus a count.

For messages that name the joints/DOFs/links behind a diagnostic: a 20-DOF hand or a 29-DOF humanoid would otherwise wrap the console, and past the first few entries the list stops being read anyway. The count of what is hidden is kept, so the reader can tell "a couple of joints" from "the whole robot" without the full list.

PARAMETER	DESCRIPTION
`names`	The names to render, in the order they should appear. Any iterable: callers pass whatever the entity handed them, which is as often `dict.keys()` as a list. TYPE: `Iterable[str]`
`limit`	How many names to show before truncating. TYPE: `int` DEFAULT: `5`

RETURNS	DESCRIPTION
`str`	`"['a', 'b']"` when the list fits, `"['a', 'b'] (+3 more)"` when it does not.

Source code in eden/utils/string.py

def first_few(names: Iterable[str], limit: int = 5) -> str:
    """Render ``names`` for a log line, truncating a long list to its first few plus a count.

    For messages that name the joints/DOFs/links behind a diagnostic: a 20-DOF hand or a
    29-DOF humanoid would otherwise wrap the console, and past the first few entries the
    list stops being read anyway. The count of what is hidden is kept, so the reader can
    tell "a couple of joints" from "the whole robot" without the full list.

    Parameters
    ----------
    names:
        The names to render, in the order they should appear. Any iterable: callers pass
        whatever the entity handed them, which is as often ``dict.keys()`` as a list.
    limit:
        How many names to show before truncating.

    Returns
    -------
    str
        ``"['a', 'b']"`` when the list fits, ``"['a', 'b'] (+3 more)"`` when it does not.
    """
    names = list(names)
    if len(names) <= limit:
        return f"{names}"
    return f"{names[:limit]} (+{len(names) - limit} more)"

term_subject ¶

term_subject(term: Any, entity_name: str | None = None) -> str

Render the subject an entity-bound term opens its errors with: "<Class> on entity '<name>'".

The same class can be configured twice against different entities (one reward per arm, one randomization per object), so the class alone does not say which of them failed. Class and entity only: ManagerBase._build_term already adds the config key as a note, and a term that restated it would print it twice.

PARAMETER	DESCRIPTION
`term`	A term that declares `entity_name`. A term with no entity (`time_out`, most commands) names its class alone and does not call this. TYPE: `Any`
`entity_name`	The entity, for a term that names it in a field of its own (a writing metric's `tool_name`, a whiteboard metric's `marker_name`); `term.entity_name` by default. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	`"FeetAirTime on entity 'robot'"`.

Source code in eden/utils/string.py

def term_subject(term: Any, entity_name: str | None = None) -> str:
    """Render the subject an entity-bound term opens its errors with: ``"<Class> on entity '<name>'"``.

    The same class can be configured twice against different entities (one reward per arm,
    one randomization per object), so the class alone does not say which of them failed.
    Class and entity only: ``ManagerBase._build_term`` already adds the config key as a note,
    and a term that restated it would print it twice.

    Parameters
    ----------
    term:
        A term that declares ``entity_name``. A term with no entity (``time_out``, most
        commands) names its class alone and does not call this.
    entity_name:
        The entity, for a term that names it in a field of its own (a writing metric's
        ``tool_name``, a whiteboard metric's ``marker_name``); ``term.entity_name`` by default.

    Returns
    -------
    str
        ``"FeetAirTime on entity 'robot'"``.
    """
    return f"{type(term).__name__} on entity '{term.entity_name if entity_name is None else entity_name}'"

resolve_matching_names ¶

resolve_matching_names(keys: str | Sequence[str], list_of_strings: Sequence[str], preserve_order: bool = False, use_glob: bool | None = None, *, literal_names: Collection[str] | None = None) -> tuple[list[int], list[str]]

Match a list of query patterns against a list of strings and return the matched indices and names.

When a list of query regular expressions or glob patterns is provided, the function checks each target string against each query and returns the indices of the matched strings and the matched strings.

If the :attr:preserve_order is True, the ordering of the matched indices and names follows the order of the query patterns: all matches for the first pattern come first, then the second, and so on.

If the :attr:preserve_order is False, the ordering of the matched indices and names is the same as the order of the provided list of strings.

For example, consider the list of strings is ['a', 'b', 'c', 'd', 'e'] and the patterns are ['a|c', 'b']. If :attr:preserve_order is False, then the function will return the indices of the matched strings and the strings as: ([0, 1, 2], ['a', 'b', 'c']). When :attr:preserve_order is True, it will return them as: ([0, 2, 1], ['a', 'c', 'b']).

A key that is exactly one of the strings selects that string alone, even when it contains a pattern character: pin.1 does not also select pin_1, and arm(2) selects arm(2) (:func:key_selects).

Note: The function does not sort the indices. It returns the indices in the order they are found.

PARAMETER	DESCRIPTION
`keys`	A regular expression, glob pattern, or a list of either to match the strings in the list. TYPE: `str \| Sequence[str]`
`list_of_strings`	A list of strings to match. TYPE: `Sequence[str]`
`preserve_order`	Whether to preserve the order of the query keys in the returned values. Defaults to False. TYPE: `bool` DEFAULT: `False`
`use_glob`	Optional override for the matching strategy. When True, treat all keys as Unix shell-style glob patterns; when False, treat them as regular expressions. When None, automatically infer the mode per key using :func:`is_glob_pattern`. Defaults to None. TYPE: `bool \| None` DEFAULT: `None`
`literal_names`	The names a key is taken literally against, when `list_of_strings` is a scope cut from a larger set (an entity's controlled joints out of all its joints). A key that is exactly one of them selects that name or, outside the scope, nothing; it never falls back to a pattern that selects a neighbour (`pin.1` passive, `pin_1` in scope). Defaults to `list_of_strings`. TYPE: `Collection[str] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`tuple[list[int], list[str]]`	A tuple of lists containing the matched indices and names.

RAISES	DESCRIPTION
`ValueError`	When multiple matches are found for a string in the list.
`ValueError`	When not all query patterns are matched.

Source code in eden/utils/string.py

def resolve_matching_names(
    keys: str | Sequence[str],
    list_of_strings: Sequence[str],
    preserve_order: bool = False,
    use_glob: bool | None = None,
    *,
    literal_names: Collection[str] | None = None,
) -> tuple[list[int], list[str]]:
    """Match a list of query patterns against a list of strings and return the matched indices and names.

    When a list of query regular expressions or glob patterns is provided, the function checks each target string
    against each query and returns the indices of the matched strings and the matched strings.

    If the :attr:`preserve_order` is True, the ordering of the matched indices and names follows the order of
    the query patterns: all matches for the first pattern come first, then the second, and so on.

    If the :attr:`preserve_order` is False, the ordering of the matched indices and names is the same as the
    order of the provided list of strings.

    For example, consider the list of strings is ['a', 'b', 'c', 'd', 'e'] and the patterns are ['a|c', 'b'].
    If :attr:`preserve_order` is False, then the function will return the indices of the matched strings and the
    strings as: ([0, 1, 2], ['a', 'b', 'c']). When :attr:`preserve_order` is True, it will return them as:
    ([0, 2, 1], ['a', 'c', 'b']).

    A key that is exactly one of the strings selects that string alone, even when it contains a pattern
    character: ``pin.1`` does not also select ``pin_1``, and ``arm(2)`` selects ``arm(2)`` (:func:`key_selects`).

    Note:
        The function does not sort the indices. It returns the indices in the order they are found.

    Parameters
    ----------
    keys: str | Sequence[str]
        A regular expression, glob pattern, or a list of either to match the strings in the list.
    list_of_strings: Sequence[str]
        A list of strings to match.
    preserve_order: bool
        Whether to preserve the order of the query keys in the returned values. Defaults to False.
    use_glob: bool | None
        Optional override for the matching strategy. When True, treat all keys as Unix shell-style glob
        patterns; when False, treat them as regular expressions. When None, automatically infer the mode per key
        using :func:`is_glob_pattern`. Defaults to None.
    literal_names: Collection[str] | None
        The names a key is taken literally against, when ``list_of_strings`` is a scope cut from a larger set
        (an entity's controlled joints out of all its joints). A key that is exactly one of them selects that
        name or, outside the scope, nothing; it never falls back to a pattern that selects a neighbour
        (``pin.1`` passive, ``pin_1`` in scope). Defaults to ``list_of_strings``.

    Returns
    -------
    tuple[list[int], list[str]]
        A tuple of lists containing the matched indices and names.

    Raises
    ------
    ValueError
        When multiple matches are found for a string in the list.
    ValueError
        When not all query patterns are matched.
    """
    # resolve name keys
    if isinstance(keys, str):
        keys = [keys]
    # find matching patterns
    index_list = []
    names_list = []
    key_idx_list = []
    # book-keeping to check that we always have a one-to-one mapping
    # i.e. each target string should match only one pattern
    target_strings_match_found = [None for _ in range(len(list_of_strings))]
    keys_match_found = [[] for _ in range(len(keys))]
    names = set(list_of_strings if literal_names is None else literal_names)
    # loop over all target strings
    for target_index, potential_match_string in enumerate(list_of_strings):
        for key_index, re_key in enumerate(keys):
            if key_selects(re_key, potential_match_string, names, use_glob=use_glob):
                # check if match already found
                if target_strings_match_found[target_index]:
                    raise ValueError(
                        f"Multiple matches for '{potential_match_string}':"
                        f" '{target_strings_match_found[target_index]}' and '{re_key}'!"
                    )
                # add to list
                target_strings_match_found[target_index] = re_key
                index_list.append(target_index)
                names_list.append(potential_match_string)
                key_idx_list.append(key_index)
                # add for pattern key
                keys_match_found[key_index].append(potential_match_string)
    # reorder keys if they should be returned in order of the query keys
    if preserve_order:
        reordered_index_list = [None] * len(index_list)
        global_index = 0
        for key_index in range(len(keys)):
            for key_idx_position, key_idx_entry in enumerate(key_idx_list):
                if key_idx_entry == key_index:
                    reordered_index_list[key_idx_position] = global_index
                    global_index += 1
        # reorder index and names list
        index_list_reorder = [None] * len(index_list)
        names_list_reorder = [None] * len(index_list)
        for idx, reorder_idx in enumerate(reordered_index_list):
            index_list_reorder[reorder_idx] = index_list[idx]
            names_list_reorder[reorder_idx] = names_list[idx]
        # update
        index_list = index_list_reorder
        names_list = names_list_reorder
    # check that all query patterns are matched
    if not all(keys_match_found):
        # make this print nicely aligned for debugging
        msg = "\n"
        for key, value in zip(keys, keys_match_found):
            msg += f"\t{key}: {value}\n"
        msg += f"Available strings: {list_of_strings}\n"
        # raise error
        raise ValueError(
            f"Not all query patterns are matched! Please check that the provided patterns are correct: {msg}"
        )
    # return
    return index_list, names_list

columns_by_name ¶

columns_by_name(available_names: Sequence[Any], wanted_names: Sequence[str], *, what: str = 'names') -> list[int]

Column indices of wanted_names within available_names (exact match, query order).

Decodes a motion file's columns by the names it stores beside them — a motion .npz's body_names / dof_names, a BVH's joints — or by a dataset's documented column order, so the mapping is independent of the writer's column order and of Genesis's internal link/DOF ordering (which changed breadth-first -> depth-first in genesis 1.2.0). available_names entries are coerced to str (npz string arrays load as numpy str_).

Unlike :func:resolve_matching_names, this is a plain exact-name lookup (no glob/regex), so names containing pattern metacharacters are matched literally.

PARAMETER	DESCRIPTION
`available_names`	The column labels, in column order (e.g. the npz `body_names`). TYPE: `Sequence[Any]`
`wanted_names`	The names to locate, in the desired output order. TYPE: `Sequence[str]`
`what`	What `available_names` are, for the error message only (e.g. `"motion npz 'body_names'"`). TYPE: `str` DEFAULT: `'names'`

RETURNS	DESCRIPTION
`list[int]`	One column index per entry in `wanted_names`, in that order.

RAISES	DESCRIPTION
`ValueError`	If any of `wanted_names` is absent from `available_names`.

Source code in eden/utils/string.py

def columns_by_name(available_names: Sequence[Any], wanted_names: Sequence[str], *, what: str = "names") -> list[int]:
    """Column indices of ``wanted_names`` within ``available_names`` (exact match, query order).

    Decodes a motion file's columns by the names it stores beside them — a motion ``.npz``'s ``body_names`` /
    ``dof_names``, a BVH's joints — or by a dataset's documented column order, so the mapping is independent of the
    writer's column order and of Genesis's internal link/DOF ordering (which changed breadth-first -> depth-first in
    genesis 1.2.0). ``available_names`` entries are coerced to ``str`` (npz string arrays load as numpy ``str_``).

    Unlike :func:`resolve_matching_names`, this is a plain exact-name lookup (no glob/regex), so
    names containing pattern metacharacters are matched literally.

    Parameters
    ----------
    available_names:
        The column labels, in column order (e.g. the npz ``body_names``).
    wanted_names:
        The names to locate, in the desired output order.
    what:
        What ``available_names`` are, for the error message only (e.g. ``"motion npz 'body_names'"``).

    Returns
    -------
    list[int]
        One column index per entry in ``wanted_names``, in that order.

    Raises
    ------
    ValueError
        If any of ``wanted_names`` is absent from ``available_names``.
    """
    available = [str(n) for n in available_names]
    position = {name: i for i, name in enumerate(available)}
    missing = [n for n in wanted_names if n not in position]
    if missing:
        raise ValueError(f"{what} is missing {missing}; available: {available}")
    return [position[n] for n in wanted_names]

resolve_matching_names_values ¶

resolve_matching_names_values(data: dict[str, Any], list_of_strings: Sequence[str], preserve_order: bool = False, use_glob: bool | None = None, *, literal_names: Collection[str] | None = None) -> tuple[list[int], list[str], list[Any]]

Match query patterns against a list of strings, returning matched indices, names, and values.

If the :attr:preserve_order is True, the ordering of the matched indices, names, and values follows the order of the query patterns (the dictionary keys): all matches for the first key come first, then the second, and so on. The returned values are NOT positionally aligned with the target list — scatter them through the returned indices instead.

If the :attr:preserve_order is False, the ordering of the matched indices and names is the same as the order of the provided list of strings.

As in :func:resolve_matching_names, a key that is exactly one of the strings selects that string alone.

For example, consider the dictionary is {"a|d|e": 1, "b|c": 2}, the list of strings is ['a', 'b', 'c', 'd', 'e']. If :attr:preserve_order is False, then the function will return the indices of the matched strings, the matched strings, and the values as: ([0, 1, 2, 3, 4], ['a', 'b', 'c', 'd', 'e'], [1, 2, 2, 1, 1]). When :attr:preserve_order is True, it will return them as: ([0, 3, 4, 1, 2], ['a', 'd', 'e', 'b', 'c'], [1, 1, 1, 2, 2]).

PARAMETER	DESCRIPTION
`data`	A dictionary mapping query patterns (regular expressions or globs) to values to match against the list. TYPE: `dict[str, Any]`
`list_of_strings`	A list of strings to match. TYPE: `Sequence[str]`
`preserve_order`	Whether to preserve the order of the query keys in the returned values. Defaults to False. TYPE: `bool` DEFAULT: `False`
`use_glob`	Optional override for the matching strategy. When True, treat all dictionary keys as glob patterns; when False, treat them as regular expressions. When None, automatically infer the mode per key using :func:`is_glob_pattern`. Defaults to None. TYPE: `bool \| None` DEFAULT: `None`
`literal_names`	The names a key is taken literally against, as in :func:`resolve_matching_names`. Defaults to `list_of_strings`. TYPE: `Collection[str] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`tuple[list[int], list[str], list[Any]]`	A tuple of lists containing the matched indices, names, and values.

RAISES	DESCRIPTION
`TypeError`	When the input argument :attr:`data` is not a dictionary.
`ValueError`	When multiple matches are found for a string in the dictionary.
`ValueError`	When not all query patterns in the data keys are matched.

Source code in eden/utils/string.py

def resolve_matching_names_values(
    data: dict[str, Any],
    list_of_strings: Sequence[str],
    preserve_order: bool = False,
    use_glob: bool | None = None,
    *,
    literal_names: Collection[str] | None = None,
) -> tuple[list[int], list[str], list[Any]]:
    """Match query patterns against a list of strings, returning matched indices, names, and values.

    If the :attr:`preserve_order` is True, the ordering of the matched indices, names, and values follows the
    order of the query patterns (the dictionary keys): all matches for the first key come first, then the
    second, and so on. The returned values are NOT positionally aligned with the target list — scatter them
    through the returned indices instead.

    If the :attr:`preserve_order` is False, the ordering of the matched indices and names is the same as the
    order of the provided list of strings.

    As in :func:`resolve_matching_names`, a key that is exactly one of the strings selects that string alone.

    For example, consider the dictionary is {"a|d|e": 1, "b|c": 2}, the list of strings is
    ['a', 'b', 'c', 'd', 'e']. If :attr:`preserve_order` is False, then the function will return the indices of the
    matched strings, the matched strings, and the values as: ([0, 1, 2, 3, 4], ['a', 'b', 'c', 'd', 'e'],
    [1, 2, 2, 1, 1]). When :attr:`preserve_order` is True, it will return them as: ([0, 3, 4, 1, 2],
    ['a', 'd', 'e', 'b', 'c'], [1, 1, 1, 2, 2]).

    Parameters
    ----------
    data: dict[str, Any]
        A dictionary mapping query patterns (regular expressions or globs) to values to match against the list.
    list_of_strings: Sequence[str]
        A list of strings to match.
    preserve_order: bool
        Whether to preserve the order of the query keys in the returned values. Defaults to False.
    use_glob: bool | None
        Optional override for the matching strategy. When True, treat all dictionary keys as glob patterns;
        when False, treat them as regular expressions. When None, automatically infer the mode per key using
        :func:`is_glob_pattern`. Defaults to None.
    literal_names: Collection[str] | None
        The names a key is taken literally against, as in :func:`resolve_matching_names`. Defaults to
        ``list_of_strings``.

    Returns
    -------
    tuple[list[int], list[str], list[Any]]
        A tuple of lists containing the matched indices, names, and values.

    Raises
    ------
    TypeError
        When the input argument :attr:`data` is not a dictionary.
    ValueError
        When multiple matches are found for a string in the dictionary.
    ValueError
        When not all query patterns in the data keys are matched.
    """
    # check valid input
    if not isinstance(data, dict):
        raise TypeError(f"Input argument `data` should be a dictionary. Received: {data}")
    # find matching patterns
    index_list = []
    names_list = []
    values_list = []
    key_idx_list = []
    # book-keeping to check that we always have a one-to-one mapping
    # i.e. each target string should match only one pattern
    target_strings_match_found = [None for _ in range(len(list_of_strings))]
    keys_match_found = [[] for _ in range(len(data))]
    names = set(list_of_strings if literal_names is None else literal_names)
    # loop over all target strings
    for target_index, potential_match_string in enumerate(list_of_strings):
        for key_index, (re_key, value) in enumerate(data.items()):
            if key_selects(re_key, potential_match_string, names, use_glob=use_glob):
                # check if match already found
                if target_strings_match_found[target_index]:
                    raise ValueError(
                        f"Multiple matches for '{potential_match_string}':"
                        f" '{target_strings_match_found[target_index]}' and '{re_key}'!"
                    )
                # add to list
                target_strings_match_found[target_index] = re_key
                index_list.append(target_index)
                names_list.append(potential_match_string)
                values_list.append(value)
                key_idx_list.append(key_index)
                # add for pattern key
                keys_match_found[key_index].append(potential_match_string)
    # reorder keys if they should be returned in order of the query keys
    if preserve_order:
        reordered_index_list = [None] * len(index_list)
        global_index = 0
        for key_index in range(len(data)):
            for key_idx_position, key_idx_entry in enumerate(key_idx_list):
                if key_idx_entry == key_index:
                    reordered_index_list[key_idx_position] = global_index
                    global_index += 1
        # reorder index and names list
        index_list_reorder = [None] * len(index_list)
        names_list_reorder = [None] * len(index_list)
        values_list_reorder = [None] * len(index_list)
        for idx, reorder_idx in enumerate(reordered_index_list):
            index_list_reorder[reorder_idx] = index_list[idx]
            names_list_reorder[reorder_idx] = names_list[idx]
            values_list_reorder[reorder_idx] = values_list[idx]
        # update
        index_list = index_list_reorder
        names_list = names_list_reorder
        values_list = values_list_reorder
    # check that all query patterns are matched
    if not all(keys_match_found):
        # make this print nicely aligned for debugging
        msg = "\n"
        for key, value in zip(data.keys(), keys_match_found):
            msg += f"\t{key}: {value}\n"
        msg += f"Available strings: {list_of_strings}\n"
        # raise error
        raise ValueError(
            f"Not all query patterns are matched! Please check that the provided patterns are correct: {msg}"
        )
    # return
    return index_list, names_list, values_list

Miscellaneous helpers (env-index sanitization, entity resolution, hashing, snake_case, key loading, paths).

Attributes¶

ASSETS_DIR_ENV_VAR `module-attribute` ¶

ASSETS_DIR_ENV_VAR = 'EDEN_ASSETS_DIR'

Classes¶

DofRuns ¶

Bases: tuple

A non-contiguous DOF selection as its contiguous runs: ((dofs_range, columns_slice), ...).

What :func:as_genesis_control_index hands a RigidEntity.control_dofs_* setter for a selection with gaps (a parallel gripper's two driver joints, [0, 3]): the setter writes each run through its own range — the run's DOFs from the value's columns_slice — instead of once through an index tensor, which Genesis' control_dofs_* turns into a GPU->CPU sync twice per call. Only the control setters accept it; hand a getter the index tensor or :func:as_genesis_index.

DofPermutation ¶

Bases: tuple

A DOF selection that permutes a contiguous block, as (dofs_range, order).

What :func:as_genesis_control_index hands a RigidEntity.control_dofs_* setter when idx names every DOF of dofs_range exactly once in another order (a quadruped's joints in leg-major order over a joint-major asset): the setter writes the whole block through the range from value.index_select(-1, order) — order[j] is the value column of the block's j-th DOF — instead of through an index tensor, which Genesis' control_dofs_* syncs on twice per call. Same DOFs, same values. Only the control setters accept it.

Attributes¶

dofs `property` ¶

dofs: range

order `property` ¶

order: Tensor

Functions:¶

resolve_named_entity ¶

resolve_named_entity(entities: dict, name: str, *, hint: str | None = None)

Look up name in an env's entities, naming what is there when it is missing.

Shared by EnvBase.get_entity, ManagerTermBase.resolve_entity and PolicyBase.require_entity: the name is always one someone typed, so the useful failure lists the names that exist, plus hint for where the caller declares its own. Takes the mapping, not the env, so a bare-entities stub resolves too.

Source code in eden/utils/misc.py

def resolve_named_entity(entities: dict, name: str, *, hint: str | None = None):
    """Look up ``name`` in an env's ``entities``, naming what is there when it is missing.

    Shared by ``EnvBase.get_entity``, ``ManagerTermBase.resolve_entity`` and
    ``PolicyBase.require_entity``: the name is always one someone typed, so the useful failure
    lists the names that exist, plus ``hint`` for where the caller declares its own. Takes the
    mapping, not the env, so a bare-``entities`` stub resolves too.
    """
    try:
        return entities[name]
    except KeyError:
        message = f"No scene entity named '{name}'; this env has {sorted(entities)}."
        raise KeyError(f"{message} {hint}" if hint else message) from None

contiguous_range ¶

contiguous_range(idx) -> range | None

idx as the equivalent range if it counts up one at a time from its first entry, else None.

The form to hand Genesis for a fixed DOF / q selection. Genesis turns a range into a slice, while an index tensor takes its advanced-indexing path: about 5x the host time per call (32 vs 6 us for get_dofs_position on a G1), a GPU->CPU sync in every control_dofs_* call, and no bool-mask fast path in the setters (RigidEntity.setters_take_every_row). Reads idx on the host, so resolve it once, in build(), and keep the tensor for anything that does tensor math.

None for a selection that counts from the end: Genesis turns range(-2, 0) into slice(-2, 0), which selects nothing, where the index tensor [-2, -1] it would replace selects the last two.

Source code in eden/utils/misc.py

def contiguous_range(idx) -> range | None:
    """``idx`` as the equivalent ``range`` if it counts up one at a time from its first entry, else None.

    The form to hand Genesis for a fixed DOF / q selection. Genesis turns a ``range`` into a slice,
    while an index tensor takes its advanced-indexing path: about 5x the host time per call (32 vs
    6 us for ``get_dofs_position`` on a G1), a GPU->CPU sync in every ``control_dofs_*`` call, and no
    bool-mask fast path in the setters (``RigidEntity.setters_take_every_row``). Reads ``idx`` on the
    host, so resolve it once, in ``build()``, and keep the tensor for anything that does tensor math.

    None for a selection that counts from the end: Genesis turns ``range(-2, 0)`` into ``slice(-2, 0)``,
    which selects nothing, where the index tensor ``[-2, -1]`` it would replace selects the last two.
    """
    host = _on_host(idx)
    if host.ndim != 1 or len(host) == 0:
        return None
    start = int(host[0])
    if start < 0:
        return None
    if torch.equal(host, torch.arange(start, start + len(host), dtype=host.dtype, device="cpu")):
        return range(start, start + len(host))
    return None

as_genesis_index ¶

as_genesis_index(idx: Tensor) -> range | Tensor

idx as its :func:contiguous_range when it has one, else idx itself — the selector to hand Genesis.

Same order, same entries, so the two forms select identical DOFs; only the Genesis code path differs. Resolve it once, in build() (it reads idx on the host).

Source code in eden/utils/misc.py

def as_genesis_index(idx: torch.Tensor) -> range | torch.Tensor:
    """``idx`` as its :func:`contiguous_range` when it has one, else ``idx`` itself — the selector to hand Genesis.

    Same order, same entries, so the two forms select identical DOFs; only the Genesis code path differs.
    Resolve it once, in ``build()`` (it reads ``idx`` on the host).
    """
    as_range = contiguous_range(idx)
    return idx if as_range is None else as_range

as_genesis_control_index ¶

as_genesis_control_index(idx: Tensor, device: device | str | None = None) -> range | DofPermutation | DofRuns | Tensor

idx as the selector to hand a RigidEntity.control_dofs_* setter.

A range when idx is contiguous (as :func:as_genesis_index); a :class:DofPermutation when it reorders a contiguous block; a :class:DofRuns when it splits into at most a few contiguous runs; else idx itself. Every form writes the same DOFs from the same value columns. device is where the permutation's order lives — the values' device; idx's by default. Resolve it once, in build() (it reads idx on the host).

Source code in eden/utils/misc.py

def as_genesis_control_index(
    idx: torch.Tensor, device: torch.device | str | None = None
) -> range | DofPermutation | DofRuns | torch.Tensor:
    """``idx`` as the selector to hand a ``RigidEntity.control_dofs_*`` setter.

    A ``range`` when ``idx`` is contiguous (as :func:`as_genesis_index`); a :class:`DofPermutation`
    when it reorders a contiguous block; a :class:`DofRuns` when it splits into at most a few
    contiguous runs; else ``idx`` itself. Every form writes the same DOFs from the same value columns.
    ``device`` is where the permutation's ``order`` lives — the values' device; ``idx``'s by default.
    Resolve it once, in ``build()`` (it reads ``idx`` on the host).
    """
    as_range = contiguous_range(idx)
    if as_range is not None:
        return as_range
    host = _on_host(idx).reshape(-1).tolist()
    if not host:
        return idx
    lo = min(host)
    if lo < 0:
        # Counted from the end: a run or block `range` would select something else (`contiguous_range`).
        return idx
    if sorted(host) == list(range(lo, lo + len(host))):
        order = sorted(range(len(host)), key=host.__getitem__)
        if device is None:
            device = idx.device if isinstance(idx, torch.Tensor) else "cpu"
        return DofPermutation((range(lo, lo + len(host)), torch.tensor(order, dtype=torch.long, device=device)))
    runs, start = [], 0
    for i in range(1, len(host) + 1):
        if i == len(host) or host[i] != host[i - 1] + 1:
            runs.append((range(host[start], host[i - 1] + 1), slice(start, i)))
            start = i
    return DofRuns(runs) if len(runs) <= _MAX_CONTROL_RUNS else idx

is_env_mask ¶

is_env_mask(envs_idx) -> bool

Whether envs_idx is a bool env mask (rather than a slice, None or int ids).

Source code in eden/utils/misc.py

def is_env_mask(envs_idx) -> bool:
    """Whether ``envs_idx`` is a bool env mask (rather than a slice, ``None`` or int ids)."""
    return isinstance(envs_idx, torch.Tensor) and envs_idx.dtype == torch.bool

sum_over_envs ¶

sum_over_envs(values: Tensor, envs_idx: slice | Tensor) -> Tensor

values[envs_idx].sum(dim=0), without a GPU->CPU sync when envs_idx is a bool mask.

Gathering with a bool mask (values[mask]) has to size its output, which is a sync; the auto-reset in RLEnvBase.step hands every manager's reset() exactly such a mask (reset_buf), so a per-term gather for a log value costs one sync per term per step. A mask is reduced in place instead. A bool values sums to the count of True entries.

Source code in eden/utils/misc.py

def sum_over_envs(values: torch.Tensor, envs_idx: slice | torch.Tensor) -> torch.Tensor:
    """``values[envs_idx].sum(dim=0)``, without a GPU->CPU sync when ``envs_idx`` is a bool mask.

    Gathering with a bool mask (``values[mask]``) has to size its output, which is a sync; the
    auto-reset in ``RLEnvBase.step`` hands every manager's ``reset()`` exactly such a mask
    (``reset_buf``), so a per-term gather for a log value costs one sync per term per step.
    A mask is reduced in place instead. A bool ``values`` sums to the count of ``True`` entries.
    """
    if is_env_mask(envs_idx):
        return torch.where(envs_idx.view(-1, *([1] * (values.ndim - 1))), values, 0).sum(dim=0)
    return _selected_rows(values, envs_idx).sum(dim=0)

mean_over_envs ¶

mean_over_envs(values: Tensor, envs_idx: slice | Tensor) -> Tensor

values[envs_idx].mean(dim=0) — zero when nothing is selected — without a sync for a bool mask.

See :func:sum_over_envs.

Source code in eden/utils/misc.py

def mean_over_envs(values: torch.Tensor, envs_idx: slice | torch.Tensor) -> torch.Tensor:
    """``values[envs_idx].mean(dim=0)`` — zero when nothing is selected — without a sync for a bool mask.

    See :func:`sum_over_envs`.
    """
    if is_env_mask(envs_idx):
        return sum_over_envs(values, envs_idx) / envs_idx.sum().clamp(min=1)
    selected = _selected_rows(values, envs_idx)
    return selected.sum(dim=0) / max(selected.shape[0], 1)

assign_over_envs ¶

assign_over_envs(target: Tensor, envs_idx: slice | Tensor, source: Tensor) -> None

target[envs_idx] = source[envs_idx] for a full-batch source, without a sync for a bool mask.

target[mask] = source[mask] gathers first, which is a GPU->CPU sync (see :func:sum_over_envs); a mask is applied with torch.where into target instead.

Source code in eden/utils/misc.py

def assign_over_envs(target: torch.Tensor, envs_idx: slice | torch.Tensor, source: torch.Tensor) -> None:
    """``target[envs_idx] = source[envs_idx]`` for a full-batch ``source``, without a sync for a bool mask.

    ``target[mask] = source[mask]`` gathers first, which is a GPU->CPU sync (see :func:`sum_over_envs`);
    a mask is applied with ``torch.where`` into ``target`` instead.
    """
    if is_env_mask(envs_idx):
        torch.where(envs_idx.view(-1, *([1] * (target.ndim - 1))), source, target, out=target)
    else:
        target[envs_idx] = source[envs_idx]

env_indices ¶

env_indices(envs_idx: slice | Tensor) -> slice | Tensor

A bool env mask as the index tensor of its True entries; a slice or an index tensor unchanged.

For code that must gather the selected rows (values[ids]): every values[mask] gather is a GPU->CPU sync, and the one nonzero() here pays for all of them. Keep the mask for scalar fills (buf[mask] = 0.0) and torch.where: a scalar written through an index tensor is the one form that syncs the other way round. Keep it for Genesis' state setters too where they take one on the device (RigidEntity.setters_take_every_row, with a range selection): handed ids, a setter writes its error flag through them, a sync per call.

Inside :func:sharing_env_indices the mask it was opened on costs nothing here: its ids were taken once, for the whole reset.

Source code in eden/utils/misc.py

def env_indices(envs_idx: slice | torch.Tensor) -> slice | torch.Tensor:
    """A bool env mask as the index tensor of its ``True`` entries; a slice or an index tensor unchanged.

    For code that must *gather* the selected rows (``values[ids]``): every ``values[mask]`` gather is
    a GPU->CPU sync, and the one ``nonzero()`` here pays for all of them. Keep the mask for
    scalar fills (``buf[mask] = 0.0``) and ``torch.where``: a scalar written through an *index*
    tensor is the one form that syncs the other way round. Keep it for Genesis' state setters too
    where they take one on the device (``RigidEntity.setters_take_every_row``, with a ``range``
    selection): handed ids, a setter writes its error flag through them, a sync per call.

    Inside :func:`sharing_env_indices` the mask it was opened on costs nothing here: its ids were
    taken once, for the whole reset.
    """
    if not is_env_mask(envs_idx):
        return envs_idx
    shared = _shared_ids(envs_idx)
    return envs_idx.nonzero()[:, 0] if shared is None else shared

sharing_env_indices ¶

sharing_env_indices(envs_idx) -> Iterator[None]

Take a bool envs_idx mask's ids once, for every :func:env_indices call on it inside.

The auto-reset hands the same reset_buf to the scene reset, every RESET event and every manager. Each consumer that gathers selected rows would otherwise take its own nonzero() — one GPU->CPU sync apiece, 3-9 a step on the shipped tasks. Bracketing the reset in this takes the one sync up front and hands its ids to all of them, along with the selected count (:func:sanitize_envs_idx's return_n_envs) that gates the reset.

The ids are matched to the mask by identity, so a mask derived from it (mask & active) takes its own nonzero(). The mask must not be written while the bracket is open, or the ids would describe the old selection; the env opens it on a reset_buf that nothing writes during a reset. Anything but a bool mask (None, a slice, an index tensor) passes through untouched.

Source code in eden/utils/misc.py

@contextmanager
def sharing_env_indices(envs_idx) -> Iterator[None]:
    """Take a bool ``envs_idx`` mask's ids once, for every :func:`env_indices` call on it inside.

    The auto-reset hands the same ``reset_buf`` to the scene reset, every RESET event and every
    manager. Each consumer that gathers selected rows would otherwise take its own ``nonzero()`` —
    one GPU->CPU sync apiece, 3-9 a step on the shipped tasks. Bracketing the reset in this
    takes the one sync up front and hands its ids to all of them, along with the selected count
    (:func:`sanitize_envs_idx`'s ``return_n_envs``) that gates the reset.

    The ids are matched to the mask by identity, so a mask derived from it (``mask & active``)
    takes its own ``nonzero()``. The mask must not be written while the bracket is open,
    or the ids would describe the old selection; the env opens it on a ``reset_buf`` that nothing
    writes during a reset. Anything but a bool mask (``None``, a slice, an index tensor) passes
    through untouched.
    """
    if not is_env_mask(envs_idx):
        yield
        return
    token = _SHARED_ENV_INDICES.set((envs_idx, envs_idx.nonzero()[:, 0]))
    try:
        yield
    finally:
        _SHARED_ENV_INDICES.reset(token)

sanitize_envs_idx ¶

sanitize_envs_idx(envs_idx: int | range | slice | tuple[int, ...] | list[int] | Tensor | ndarray | None, num_envs: int, *, prefer_slice: bool = True, name: str = 'envs_idx', return_n_envs: bool = False) -> tuple[slice | Tensor, int] | slice | Tensor

Normalize environment indices with an optional slice fast-path.

PARAMETER	DESCRIPTION
`envs_idx`	Environment selector. A mask must be a bool `torch.Tensor` or `np.ndarray` of shape `(num_envs,)`; a bare bool or a list/tuple of bools raises `IndexError`. TYPE: `int \| range \| slice \| tuple[int, ...] \| list[int] \| Tensor \| ndarray \| None`
`num_envs`	Total number of environments. TYPE: `int`
`prefer_slice`	If True, preserve slices for faster backend indexing when possible. TYPE: `bool` DEFAULT: `True`
`name`	Name used in validation errors. TYPE: `str` DEFAULT: `'envs_idx'`
`return_n_envs`	If True, also return the number of environments selected by the index. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`envs_idx`	Normalized environment indices. TYPE: `slice \| Tensor`
`n_envs`	Number of environments selected by the index. Only returned if return_n_envs is True. TYPE: `(int, optional)`

Source code in eden/utils/misc.py

def sanitize_envs_idx(
    envs_idx: int | range | slice | tuple[int, ...] | list[int] | torch.Tensor | np.ndarray | None,
    num_envs: int,
    *,
    prefer_slice: bool = True,
    name: str = "envs_idx",
    return_n_envs: bool = False,
) -> tuple[slice | torch.Tensor, int] | slice | torch.Tensor:
    """Normalize environment indices with an optional slice fast-path.

    Parameters
    ----------
    envs_idx: int | range | slice | tuple[int, ...] | list[int] | torch.Tensor | np.ndarray | None
        Environment selector. A mask must be a bool ``torch.Tensor`` or ``np.ndarray`` of shape
        ``(num_envs,)``; a bare bool or a list/tuple of bools raises ``IndexError``.
    num_envs: int
        Total number of environments.
    prefer_slice: bool
        If True, preserve slices for faster backend indexing when possible.
    name: str
        Name used in validation errors.
    return_n_envs: bool
        If True, also return the number of environments selected by the index.

    Returns
    -------
    envs_idx: slice | torch.Tensor
        Normalized environment indices.
    n_envs: int, optional
        Number of environments selected by the index. Only returned if return_n_envs is True.
    """
    # Deferred, not module-scope: importing any ``genesis.*`` submodule runs
    # ``genesis/__init__.py``, so this one name would cost every importer of this module the full
    # ``import genesis``. The per-call price of importing it here is a ``sys.modules`` lookup:
    # measured at 59 ns against this function's own 19.8 us, i.e. 0.3%, which is why it is not
    # worth caching the name in a module global. By the time anything sanitizes an env index
    # Genesis is long since imported.
    from genesis.utils.misc import sanitize_index as gs_sanitize_index

    if num_envs < 0:
        raise ValueError(f"num_envs must be non-negative, got {num_envs}")

    if is_env_mask(envs_idx):
        if envs_idx.numel() != num_envs:
            raise IndexError(f"{name} boolean mask has length {envs_idx.numel()}, expected num_envs={num_envs}.")
        out = envs_idx
    elif envs_idx is None:
        out = slice(None) if prefer_slice else _device_arange(0, num_envs, 1)
    elif isinstance(envs_idx, slice):
        start, stop, step = envs_idx.indices(num_envs)
        out = slice(start, stop, step) if prefer_slice else _device_arange(start, stop, step)
    else:
        # Genesis's sanitize_index disabled its out-of-range check (perf), so
        # validate here before delegating the int-tensor conversion.
        _check_envs_idx_in_range(envs_idx, num_envs, name)
        out = gs_sanitize_index(envs_idx, -1, num_envs, 0, name)

    if return_n_envs:
        return out, _selected_env_count(out, num_envs)
    return out

invert_segmentation_idx_dict ¶

invert_segmentation_idx_dict(segmentation_idx_dict: dict) -> dict[int, list[int]]

Invert Genesis' segmentation_idx_dict into entity_idx -> [segmentation ids].

Genesis maps segmentation id -> scene element, where the element is a bare entity index at segmentation_level="entity" and an (entity_idx, link_idx, ...) tuple at "link" / "geom". Callers want the reverse, and at link/geom level an entity owns several ids — one per link/geom. Keeping only one (what a plain d[entity_idx] = seg_idx loop does, since the last write wins) silently reduces the entity to a single link: a masked point cloud loses most of the object, and a background mask stops covering most of the background.

Source code in eden/utils/misc.py

def invert_segmentation_idx_dict(segmentation_idx_dict: dict) -> dict[int, list[int]]:
    """Invert Genesis' ``segmentation_idx_dict`` into ``entity_idx -> [segmentation ids]``.

    Genesis maps *segmentation id -> scene element*, where the element is a bare entity index at
    ``segmentation_level="entity"`` and an ``(entity_idx, link_idx, ...)`` tuple at ``"link"`` /
    ``"geom"``. Callers want the reverse, and at link/geom level an entity owns **several** ids —
    one per link/geom. Keeping only one (what a plain ``d[entity_idx] = seg_idx`` loop does, since
    the last write wins) silently reduces the entity to a single link: a masked point cloud loses
    most of the object, and a background mask stops covering most of the background.
    """
    entity_seg_ids: dict[int, list[int]] = {}
    for seg_idx, scene_elem in segmentation_idx_dict.items():
        entity_idx = scene_elem[0] if isinstance(scene_elem, (list, tuple)) else scene_elem
        entity_seg_ids.setdefault(entity_idx, []).append(seg_idx)
    return entity_seg_ids

verify_file_hash ¶

verify_file_hash(path: str, expected_hash: str) -> bool

Verify file integrity using SHA256.

Source code in eden/utils/misc.py

def verify_file_hash(path: str, expected_hash: str) -> bool:
    """Verify file integrity using SHA256."""
    if not os.path.exists(path):
        return False

    sha256_hash = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            sha256_hash.update(chunk)

    return sha256_hash.hexdigest() == expected_hash

string_to_sha256_hex ¶

string_to_sha256_hex(text: str) -> str

Convert a string to the hex SHA-256 hash of its UTF-8 encoding.

Source code in eden/utils/misc.py

def string_to_sha256_hex(text: str) -> str:
    """Convert a string to the hex SHA-256 hash of its UTF-8 encoding."""
    encoded_string = text.encode("utf-8")
    hasher = hashlib.sha256()
    hasher.update(encoded_string)
    hex_digest = hasher.hexdigest()
    return hex_digest

to_snake_case ¶

to_snake_case(camel_case_string)

Convert a CamelCase string to snake_case using regular expressions.

Source code in eden/utils/misc.py

def to_snake_case(camel_case_string):
    """Convert a CamelCase string to snake_case using regular expressions."""
    if not camel_case_string:
        return ""

    # Add an underscore before each capitalized word (an uppercase letter followed by lowercase).
    s1 = re.sub("([^_])([A-Z][a-z]+)", r"\1_\2", camel_case_string)
    # Handle cases like "HTTP" by adding an underscore between consecutive uppercase letters.
    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()

get_package_provenance ¶

get_package_provenance(package_name: str) -> dict[str, str | None]

Which package_name is installed, for provenance: {"version": ..., "commit": ...}, each None when unknown (both are, when the package is not installed).

version is the distribution's (1.4.0). commit is the git commit it was installed from: the one a git+ install recorded (PEP 610's direct_url.json), or the HEAD of the checkout an editable install points at; an install from an index has none. Two fields, not one holding either, so whoever reads a saved record never has to tell a commit from a version by its shape. :func:get_eden_provenance is the same record for Eden.

Source code in eden/utils/misc.py

def get_package_provenance(package_name: str) -> dict[str, str | None]:
    """Which ``package_name`` is installed, for provenance: ``{"version": ..., "commit": ...}``, each ``None`` when
    unknown (both are, when the package is not installed).

    ``version`` is the distribution's (``1.4.0``). ``commit`` is the git commit it was installed from: the one a
    ``git+`` install recorded (PEP 610's ``direct_url.json``), or the ``HEAD`` of the checkout an editable install
    points at; an install from an index has none. Two fields, not one holding either, so whoever reads a saved record
    never has to tell a commit from a version by its shape. :func:`get_eden_provenance` is the same record for Eden.
    """
    try:
        dist = metadata.distribution(package_name)
    except metadata.PackageNotFoundError:
        return {"version": None, "commit": None}
    return {"version": dist.version or None, "commit": _installed_commit(dist)}

get_eden_commit ¶

get_eden_commit() -> str | None

The HEAD of the git checkout eden is imported from, or None outside one.

The commit is for developers working in a clone. Every install, a pip install "eden-world @ git+..." included, is identified by its release version alone (:func:get_eden_version), on purpose.

Found by path, the way :func:get_eden_working_tree_dirty finds the dirty flag, so the two always describe the same tree. Not by distribution name through importlib.metadata, which takes the first match on sys.path: under python -m that is the repo root, where an in-tree setuptools build leaves an eden_world.egg-info with no direct_url.json, so the commit would read back as the bare version.

Source code in eden/utils/misc.py

def get_eden_commit() -> str | None:
    """The ``HEAD`` of the git checkout ``eden`` is imported from, or ``None`` outside one.

    The commit is for developers working in a clone. Every install, a ``pip install "eden-world @ git+..."`` included,
    is identified by its release version alone (:func:`get_eden_version`), on purpose.

    Found by path, the way :func:`get_eden_working_tree_dirty` finds the dirty flag, so the two always describe the
    same tree. Not by distribution name through ``importlib.metadata``, which takes the first match on ``sys.path``:
    under ``python -m`` that is the repo root, where an in-tree setuptools build leaves an ``eden_world.egg-info``
    with no ``direct_url.json``, so the commit would read back as the bare version.
    """
    checkout = _eden_checkout()
    return checkout[1] if checkout is not None else None

get_eden_provenance ¶

get_eden_provenance() -> dict[str, str | None]

Which Eden this is, for provenance: {"version": "0.1.0", "commit": <HEAD in a clone, else None>}, the record :func:get_package_provenance gives any other package. A run's own record adds dirty (:func:get_eden_run_provenance).

Source code in eden/utils/misc.py

def get_eden_provenance() -> dict[str, str | None]:
    """Which Eden this is, for provenance: ``{"version": "0.1.0", "commit": <HEAD in a clone, else None>}``, the
    record :func:`get_package_provenance` gives any other package. A run's own record adds ``dirty``
    (:func:`get_eden_run_provenance`).
    """
    return {"version": get_eden_version() or None, "commit": get_eden_commit()}

get_eden_run_provenance ¶

get_eden_run_provenance() -> dict[str, str | bool | None]

:func:get_eden_provenance plus dirty (:func:get_eden_working_tree_dirty), for a run's own record (run.json, an export's policy_spec.json). One checkout lookup answers both, so the commit and the dirty flag always describe the same tree.

Source code in eden/utils/misc.py

def get_eden_run_provenance() -> dict[str, str | bool | None]:
    """:func:`get_eden_provenance` plus ``dirty`` (:func:`get_eden_working_tree_dirty`), for a run's own record
    (``run.json``, an export's ``policy_spec.json``). One checkout lookup answers both, so the commit and the dirty
    flag always describe the same tree.
    """
    checkout = _eden_checkout()
    return {
        "version": get_eden_version() or None,
        "commit": checkout[1] if checkout is not None else None,
        "dirty": _working_tree_dirty(checkout[0]) if checkout is not None else None,
    }

format_provenance ¶

format_provenance(provenance: dict) -> str

A provenance record for a log line: 0.1.0 (c246ebd), the version and the commit cut to 7 characters, either alone when the other is unknown, or unknown.

Source code in eden/utils/misc.py

def format_provenance(provenance: dict) -> str:
    """A provenance record for a log line: ``0.1.0 (c246ebd)``, the version and the commit cut to 7 characters, either
    alone when the other is unknown, or ``unknown``.
    """
    version, commit = provenance.get("version"), provenance.get("commit")
    if version and commit:
        return f"{version} ({commit[:7]})"
    return version or (commit[:7] if commit else "unknown")

describe_provenance_change ¶

describe_provenance_change(saved: object, current: dict) -> str | None

How current differs from the saved record, as saved 0.1.0 (c246ebd), running 0.1.0 (3f2a1b2), or None when the two match or cannot be compared.

Commits are compared when both records have one, else versions: a file saved in a clone and read from a wheel of the same release matches. A saved that is not a record (a file written before these fields existed) cannot be compared.

Source code in eden/utils/misc.py

def describe_provenance_change(saved: object, current: dict) -> str | None:
    """How ``current`` differs from the ``saved`` record, as ``saved 0.1.0 (c246ebd), running 0.1.0 (3f2a1b2)``, or
    ``None`` when the two match or cannot be compared.

    Commits are compared when both records have one, else versions: a file saved in a clone and read from a wheel of
    the same release matches. A ``saved`` that is not a record (a file written before these fields existed) cannot be
    compared.
    """
    if not isinstance(saved, dict):
        return None
    for key in ("commit", "version"):
        if saved.get(key) and current.get(key):
            if saved[key] == current[key]:
                return None
            return f"saved {format_provenance(saved)}, running {format_provenance(current)}"
    return None

get_now ¶

get_now()

Source code in eden/utils/misc.py

def get_now():
    return datetime.datetime.now().strftime("%y%b%dT%H%M%S")

sanitize_run_name ¶

sanitize_run_name(run_name: str) -> str

Return run_name made safe as a path segment / wandb name (or "" if empty).

Collapses runs of path-separator, URL-reserved, and whitespace characters to a single -, trims leading/trailing separators, and caps the length at 64. Used both for the log-dir basename segment and for runner_options.run_name so the two always agree.

PARAMETER	DESCRIPTION
`run_name`	The raw, user-supplied run name. TYPE: `str`

RETURNS	DESCRIPTION
`str`	The sanitized run name, empty when the input is empty or all-unsafe.

Source code in eden/utils/misc.py

def sanitize_run_name(run_name: str) -> str:
    """Return ``run_name`` made safe as a path segment / wandb name (or ``""`` if empty).

    Collapses runs of path-separator, URL-reserved, and whitespace characters to a single
    ``-``, trims leading/trailing separators, and caps the length at 64. Used both for the
    log-dir basename segment and for ``runner_options.run_name`` so the two always agree.

    Parameters
    ----------
    run_name : str
        The raw, user-supplied run name.

    Returns
    -------
    str
        The sanitized run name, empty when the input is empty or all-unsafe.
    """
    return _RUN_NAME_UNSAFE.sub("-", run_name.strip()).strip("-")[:64]

create_symlink ¶

create_symlink(src, dst)

Source code in eden/utils/misc.py

def create_symlink(src, dst):
    if not os.path.exists(src):
        raise FileNotFoundError(f"[*] {src} not found")
    if os.path.islink(dst):
        os.remove(dst)
    os.symlink(src, dst)

get_eden_working_tree_dirty ¶

get_eden_working_tree_dirty() -> bool | None

Return whether the eden source tree has uncommitted changes, for run provenance.

Complements the commit reported by :func:get_eden_commit: that sha is a lie for reproducibility if the tree it names had local edits, and a commit alone can never convey that. Consumed by the run manifest (:mod:eden.utils.run_manifest).

Best-effort: a non-editable (wheel) install, an unavailable git, or any subprocess failure yields None (unknown) rather than raising — writing a run manifest must never abort training.

"Dirty" means the tracked source differs from HEAD (staged or unstaged edits). Untracked files are excluded (--untracked-files=no): a run always drops new files into the tree (its own logs, __pycache__, scratch), so counting those would flip the flag to True on essentially every run and drain it of meaning — the question for reproducibility is whether the committed code was modified.

For Eden vendored inside a larger repository, the flag is that repository's: an edit anywhere in its tracked files counts, because the commit beside it (:func:get_eden_commit) is that repository's HEAD and names all of them.

RETURNS	DESCRIPTION
`bool \| None`	`True` if tracked files differ from `HEAD`, `False` if clean, and `None` when it could not be determined (no git checkout / git unavailable).

Source code in eden/utils/misc.py

def get_eden_working_tree_dirty() -> bool | None:
    """Return whether the eden source tree has uncommitted changes, for run provenance.

    Complements the commit reported by :func:`get_eden_commit`: that sha is a
    lie for reproducibility if the tree it names had local edits, and a commit alone can
    never convey that. Consumed by the run manifest (:mod:`eden.utils.run_manifest`).

    Best-effort: a non-editable (wheel) install, an unavailable ``git``, or any
    subprocess failure yields ``None`` (unknown) rather than raising — writing a run
    manifest must never abort training.

    "Dirty" means the *tracked* source differs from ``HEAD`` (staged or unstaged edits).
    Untracked files are excluded (``--untracked-files=no``): a run always drops new files
    into the tree (its own logs, ``__pycache__``, scratch), so counting those would flip
    the flag to ``True`` on essentially every run and drain it of meaning — the question
    for reproducibility is whether the committed code was modified.

    For Eden vendored inside a larger repository, the flag is that repository's: an edit anywhere in its tracked files
    counts, because the commit beside it (:func:`get_eden_commit`) is that repository's ``HEAD`` and names all of them.

    Returns
    -------
    bool | None
        ``True`` if tracked files differ from ``HEAD``, ``False`` if clean, and ``None``
        when it could not be determined (no git checkout / git unavailable).
    """
    checkout = _eden_checkout()
    return _working_tree_dirty(checkout[0]) if checkout is not None else None

set_api_keys_from_files ¶

set_api_keys_from_files()

Load API keys from eden/generators/keys/ into environment variables.

Reads all *.key files from the eden/generators/keys/ directory, extracts the API key from each file, and sets them as environment variables.

The environment variable name is the filename without its extension, upper-cased: a file named 'my_api.key' sets an environment variable named 'MY_API'.

Source code in eden/utils/misc.py

def set_api_keys_from_files():
    """Load API keys from ``eden/generators/keys/`` into environment variables.

    Reads all ``*.key`` files from the ``eden/generators/keys/`` directory,
    extracts the API key from each file, and sets them as environment variables.

    The environment variable name is the filename without its extension, upper-cased:
    a file named 'my_api.key' sets an environment variable named 'MY_API'.
    """
    keys_directory = os.path.join(os.path.dirname(os.path.dirname(__file__)), "generators", "keys")
    key_files_pattern = os.path.join(keys_directory, "*.key")

    if not os.path.isdir(keys_directory):
        # No keys directory is the normal state for most installs.
        en.logger.debug(f"No API keys directory at {keys_directory}; nothing to load.")
        return

    key_files = glob.glob(key_files_pattern)

    if not key_files:
        en.logger.debug(f"No API key files found in {keys_directory}")
        return

    for key_file_path in key_files:
        try:
            # Derive the API name from the filename
            # e.g., /path/to/eden/generators/keys/some_api_name.key -> some_api_name
            base_name = os.path.basename(key_file_path)
            api_name_parts = base_name.split(".")[:-1]  # Remove .key extension
            api_name = "_".join(api_name_parts)

            if not api_name:
                en.logger.warning(f"Could not derive API name from file: {key_file_path}")
                continue

            # The environment variable name (e.g., SOME_API_NAME)
            env_var_name = f"{api_name.upper()}"

            with open(key_file_path, "r") as f:
                api_key = f.read().strip()

            if api_key:
                os.environ[env_var_name] = api_key
                en.logger.info(f"Set environment variable: {env_var_name}")
            else:
                en.logger.warning(f"API key file is empty: {key_file_path}")

        except Exception as e:
            en.logger.error(f"Error processing file {key_file_path}: {e}")

list_host_ips ¶

list_host_ips()

Best-effort list of this host's LAN IPv4 addresses, most-likely-reachable first.

Used by the teleop bridges to print the address an XR headset should dial.

Source code in eden/utils/misc.py

def list_host_ips():
    """Best-effort list of this host's LAN IPv4 addresses, most-likely-reachable first.

    Used by the teleop bridges to print the address an XR headset should dial.
    """
    ips = []

    # Primary outbound interface (the one the headset most likely shares).
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.connect(("8.8.8.8", 80))
        ips.append(sock.getsockname()[0])
        sock.close()
    except Exception:
        pass

    # Everything else resolvable from the hostname.
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            ip = info[4][0]
            if not ip.startswith("127.") and ip not in ips:
                ips.append(ip)
    except Exception:
        pass

    return ips

get_src_dir ¶

get_src_dir()

Source code in eden/utils/misc.py

def get_src_dir():
    return os.path.dirname(os.path.dirname(__file__))

get_assets_dir ¶

get_assets_dir(*parts: str) -> str

Root that Eden downloads Hugging Face assets into, joined with parts.

Robots, objects, scenes, policy checkpoints and benchmark-suite assets are fetched file by file into this directory. It is $EDEN_ASSETS_DIR when that is set and non-empty (~ expanded), else get_cache_dir("assets"): $XDG_CACHE_HOME/eden/assets, by default ~/.cache/eden/assets, which is per-user, writable without root, and shared by every venv and reinstall.

A relative $EDEN_ASSETS_DIR raises ValueError. Read on every call, it would name a different directory each time the process (or a worker a launcher starts elsewhere) changes its working directory.

Read on every call, like :func:get_cache_dir. Nothing caches it: an unset local_dir names a subdirectory of it that is resolved at load time (:func:resolve_local_dir), so a saved config's local_dir stays unset.

Empty parts are skipped, so get_assets_dir("") (an asset family's unset assets_subdir) is the root itself rather than the root with a trailing separator.

This function neither creates the directory nor checks that it is writable. The first download does both, and raises naming the path and EDEN_ASSETS_DIR when it cannot write there.

Source code in eden/utils/misc.py

def get_assets_dir(*parts: str) -> str:
    """Root that Eden downloads Hugging Face assets into, joined with ``parts``.

    Robots, objects, scenes, policy checkpoints and benchmark-suite assets are fetched file by file
    into this directory. It is ``$EDEN_ASSETS_DIR`` when that is set and non-empty (``~`` expanded), else
    ``get_cache_dir("assets")``: ``$XDG_CACHE_HOME/eden/assets``, by default ``~/.cache/eden/assets``, which
    is per-user, writable without root, and shared by every venv and reinstall.

    A relative ``$EDEN_ASSETS_DIR`` raises ``ValueError``. Read on every call, it would name a different
    directory each time the process (or a worker a launcher starts elsewhere) changes its working directory.

    Read on every call, like :func:`get_cache_dir`. Nothing caches it: an unset ``local_dir`` names a
    subdirectory of it that is resolved at load time (:func:`resolve_local_dir`), so a saved config's
    ``local_dir`` stays unset.

    Empty ``parts`` are skipped, so ``get_assets_dir("")`` (an asset family's unset ``assets_subdir``)
    is the root itself rather than the root with a trailing separator.

    This function neither creates the directory nor checks that it is writable. The first download
    does both, and raises naming the path and ``EDEN_ASSETS_DIR`` when it cannot write there.
    """
    override = os.environ.get(ASSETS_DIR_ENV_VAR)
    if not override:
        return get_cache_dir("assets", *(part for part in parts if part))
    root = os.path.expanduser(override)
    if not os.path.isabs(root):
        raise ValueError(
            f"{ASSETS_DIR_ENV_VAR}={override!r} is a relative path, so it would move with the working directory "
            f"of whichever process reads it. Export an absolute one: export {ASSETS_DIR_ENV_VAR}={os.path.abspath(root)}"
        )
    return os.path.join(root, *(part for part in parts if part))

resolve_local_dir ¶

resolve_local_dir(local_dir: str | None, assets_subdir: str = '') -> str

Where an asset owner caches its files: local_dir when set, else assets_subdir under the asset root.

The one statement of the rule. EntityOptions, PCDOptions and PolicyBase each expose it as a resolve_local_dir() method over their own fields; :func:eden.utils.assets.get_asset_path applies it with no subdirectory. Resolved on each call, so EDEN_ASSETS_DIR moves every unset local_dir.

Source code in eden/utils/misc.py

def resolve_local_dir(local_dir: str | None, assets_subdir: str = "") -> str:
    """Where an asset owner caches its files: ``local_dir`` when set, else ``assets_subdir`` under the asset root.

    The one statement of the rule. ``EntityOptions``, ``PCDOptions`` and ``PolicyBase`` each expose it as a
    ``resolve_local_dir()`` method over their own fields; :func:`eden.utils.assets.get_asset_path` applies it with
    no subdirectory. Resolved on each call, so ``EDEN_ASSETS_DIR`` moves every unset ``local_dir``.
    """
    return local_dir or get_assets_dir(assets_subdir)

get_cache_dir ¶

get_cache_dir(*parts: str) -> str

Per-user Eden cache directory, joined with parts.

Everything Eden writes rather than ships goes here: writable without root (unlike the installed package), and not shared between users on a multi-tenant host (unlike /tmp).

Source code in eden/utils/misc.py

def get_cache_dir(*parts: str) -> str:
    """Per-user Eden cache directory, joined with ``parts``.

    Everything Eden *writes* rather than ships goes here: writable without root (unlike the
    installed package), and not shared between users on a multi-tenant host (unlike ``/tmp``).
    """
    root = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache")
    return os.path.join(root, "eden", *parts)

get_gs_assets_dir ¶

get_gs_assets_dir()

Source code in eden/utils/misc.py

def get_gs_assets_dir():
    # Resolve the actual Genesis package root (e.g. .../site-packages/genesis)
    # instead of matching arbitrary nested paths that merely contain "genesis".
    try:
        genesis_module = __import__("genesis")
        genesis_dir = os.path.dirname(os.path.abspath(genesis_module.__file__))
        assets_dir = os.path.join(genesis_dir, "assets")
        if os.path.isdir(assets_dir):
            return assets_dir
    except Exception:
        pass

    # Fallback: recover package root from sys.path entries that may include
    # nested paths like ".../genesis/ext/.../bin".
    for path in sys.path:
        normalized_path = os.path.abspath(path)
        marker = f"{os.sep}genesis{os.sep}"
        if marker in normalized_path:
            genesis_root = normalized_path.split(marker, 1)[0] + marker.rstrip(os.sep)
        elif os.path.basename(normalized_path) == "genesis":
            genesis_root = normalized_path
        else:
            continue

        assets_dir = os.path.join(genesis_root, "assets")
        if os.path.isdir(assets_dir):
            return assets_dir

    raise FileNotFoundError(
        "Could not locate the Genesis 'assets' directory. "
        "Tried resolving it via the 'genesis' package location and entries in sys.path."
    )

to_jsonish ¶

to_jsonish(value: Any) -> Any

Normalise a value so it compares equal to itself across a torch.save round-trip.

Tuples come back as lists, and an array-like compares elementwise — comparing one directly raises Boolean value of Tensor is ambiguous. tolist() covers torch tensors, numpy arrays and numpy scalars alike.

Source code in eden/utils/misc.py

def to_jsonish(value: Any) -> Any:
    """Normalise a value so it compares equal to itself across a ``torch.save`` round-trip.

    Tuples come back as lists, and an array-like compares elementwise — comparing one directly
    raises ``Boolean value of Tensor is ambiguous``. ``tolist()`` covers torch tensors, numpy
    arrays and numpy scalars alike.
    """
    if hasattr(value, "tolist"):
        return to_jsonish(value.tolist())
    if isinstance(value, (tuple, list)):
        return [to_jsonish(v) for v in value]
    if isinstance(value, dict):
        return {k: to_jsonish(v) for k, v in value.items()}
    return value

Terminal styling and formatting helpers (ANSI colors, argparse help formatter).

Whether anything here emits an escape is decided by Eden's log theme — the same eden.init(logger_theme=...) / EDEN_LOG_THEME / eden --log-theme setting that governs the logger — so one choice covers the whole of a command's output. The default theme is agent, which emits none.

The environment's own reading — NO_COLOR / FORCE_COLOR / isatty — is not the switch: on its own it would color eden train --task X --help on a pty for a run that asked, through the theme, for no escapes. It is what the auto theme means; :func:color_enabled is that reading, and :func:eden._logging.resolve_theme calls it.

Attributes¶

ANSI_RESET `module-attribute` ¶

ANSI_RESET = '\x1b[0m'

ANSI_BOLD `module-attribute` ¶

ANSI_BOLD = '\x1b[1m'

ANSI_DIM `module-attribute` ¶

ANSI_DIM = '\x1b[2m'

ANSI_ITALIC `module-attribute` ¶

ANSI_ITALIC = '\x1b[3m'

ANSI_UNDERLINE `module-attribute` ¶

ANSI_UNDERLINE = '\x1b[4m'

ANSI_PREV_LINE `module-attribute` ¶

ANSI_PREV_LINE = '\x1b[1F'

ANSI_BLACK `module-attribute` ¶

ANSI_BLACK = '\x1b[30m'

ANSI_RED `module-attribute` ¶

ANSI_RED = '\x1b[31m'

ANSI_GREEN `module-attribute` ¶

ANSI_GREEN = '\x1b[32m'

ANSI_YELLOW `module-attribute` ¶

ANSI_YELLOW = '\x1b[33m'

ANSI_BLUE `module-attribute` ¶

ANSI_BLUE = '\x1b[34m'

ANSI_MAGENTA `module-attribute` ¶

ANSI_MAGENTA = '\x1b[35m'

ANSI_CYAN `module-attribute` ¶

ANSI_CYAN = '\x1b[36m'

ANSI_WHITE `module-attribute` ¶

ANSI_WHITE = '\x1b[37m'

Classes¶

ColorHelpFormatter ¶

Argparse help formatter that ANSI-colorizes section headings and --flag tokens.

Suppresses the auto-generated usage: block in full --help output (where it duplicates the options: section) but keeps it on parse errors so parser.error(...) still produces a useful diagnostic. Wraps :class:argparse.RawDescriptionHelpFormatter so descriptions and epilogs preserve newlines.

Subclassing is deferred to __new__ so argparse is not imported at module load.

Functions:¶

color_enabled ¶

color_enabled() -> bool

Return whether the environment wants color — the auto theme's reading.

Honors the de-facto standards:

NO_COLOR (any value) disables color.
FORCE_COLOR (any value) enables color regardless of TTY.
Otherwise, color is enabled only when sys.stdout is a TTY.

This is one input to the decision, not the decision: what actually gets emitted is :func:color_output_enabled, which asks the log theme. A caller reaching for this one wants "would a person be watching?", which is a different question from "does this run print escapes?" — and the answer to the second is no by default, on a TTY included.

Source code in eden/utils/format.py

def color_enabled() -> bool:
    """Return whether the *environment* wants color — the ``auto`` theme's reading.

    Honors the de-facto standards:

    - ``NO_COLOR`` (any value) disables color.
    - ``FORCE_COLOR`` (any value) enables color regardless of TTY.
    - Otherwise, color is enabled only when ``sys.stdout`` is a TTY.

    This is one *input* to the decision, not the decision: what actually gets emitted is
    :func:`color_output_enabled`, which asks the log theme. A caller reaching for this one wants
    "would a person be watching?", which is a different question from "does this run print
    escapes?" — and the answer to the second is no by default, on a TTY included.
    """
    if os.environ.get("NO_COLOR"):
        return False
    if os.environ.get("FORCE_COLOR"):
        return True
    return sys.stdout.isatty()

color_output_enabled ¶

color_output_enabled() -> bool

Return whether this process emits ANSI escapes, per Eden's resolved log theme.

human yes, agent no, auto per :func:color_enabled. One switch for the whole of a command's output, so the logger and everything here (--help included) agree. The switch is EDEN_LOG_THEME, which both eden.init(logger_theme=...) and eden --log-theme export, so a theme asked for either way governs this and not just the logger.

RETURNS	DESCRIPTION
`bool`	Whether to emit escapes.

Source code in eden/utils/format.py

def color_output_enabled() -> bool:
    """Return whether this process emits ANSI escapes, per Eden's resolved log theme.

    ``human`` yes, ``agent`` no, ``auto`` per :func:`color_enabled`. One switch for the whole of a
    command's output, so the logger and everything here (``--help`` included) agree. The switch is
    ``EDEN_LOG_THEME``, which both ``eden.init(logger_theme=...)`` and ``eden --log-theme``
    export, so a theme asked for either way governs this and not just the logger.

    Returns
    -------
    bool
        Whether to emit escapes.
    """
    # Deferred: ``eden._logging`` imports this module at module scope for :func:`color_enabled`,
    # so the edge has to run one way only.
    from eden._logging import DEFAULT_LOGGER_THEME, resolve_theme

    try:
        theme = resolve_theme()
    except ValueError:
        # Not a swallowed misconfiguration: ``import eden`` resolves the theme unconditionally
        # and raises there, so a value this bad has already taken the process down -- the only
        # way to reach it here is an ``os.environ`` mutation *after* import. And this function is
        # reached from ``ColorHelpFormatter``, i.e. from ``parser.error(...)``, where raising
        # would replace the user's actual argparse error with a secondary one about a variable
        # they changed mid-process. Message-building code does not raise.
        theme = DEFAULT_LOGGER_THEME
    return theme == "human"

format_table ¶

format_table(title: str, field_names: list[str], rows: Iterable[Iterable[Any]], *, align: dict[str, str] | None = None, max_width: dict[str, int] | None = None) -> str

Render a titled table in the resolved log theme.

human gets a boxed :class:prettytable.PrettyTable. agent gets the same cells with none of the box: a title: col | col header, then one cell | cell line per row. The borders and the column padding are about half of a boxed table's tokens and carry nothing a reader of a transcript uses. Under agent a leading Index / No. column is dropped, since it restates the row order, and a multi-line cell is joined with ", ", since a row has to be one line to be read as one.

PARAMETER	DESCRIPTION
`title`	The table's title. TYPE: `str`
`field_names`	The column headers. TYPE: `list[str]`
`rows`	One iterable of cells per row, in `field_names` order. TYPE: `Iterable[Iterable[Any]]`
`align`	PrettyTable alignment (`"l"`, `"c"`, `"r"`) per column name; `human` only. TYPE: `dict[str, str] \| None` DEFAULT: `None`
`max_width`	PrettyTable wrap width per column name; `human` only. TYPE: `dict[str, int] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	The rendered table, without a trailing newline.

Source code in eden/utils/format.py

def format_table(
    title: str,
    field_names: list[str],
    rows: Iterable[Iterable[Any]],
    *,
    align: dict[str, str] | None = None,
    max_width: dict[str, int] | None = None,
) -> str:
    """Render a titled table in the resolved log theme.

    ``human`` gets a boxed :class:`prettytable.PrettyTable`. ``agent`` gets the same cells with
    none of the box: a ``title: col | col`` header, then one ``  cell | cell`` line per row. The
    borders and the column padding are about half of a boxed table's tokens and carry nothing a
    reader of a transcript uses. Under ``agent`` a leading ``Index`` / ``No.`` column is dropped,
    since it restates the row order, and a multi-line cell is joined with ``", "``, since a row
    has to be one line to be read as one.

    Parameters
    ----------
    title : str
        The table's title.
    field_names : list[str]
        The column headers.
    rows : Iterable[Iterable[Any]]
        One iterable of cells per row, in ``field_names`` order.
    align : dict[str, str] | None
        PrettyTable alignment (``"l"``, ``"c"``, ``"r"``) per column name; ``human`` only.
    max_width : dict[str, int] | None
        PrettyTable wrap width per column name; ``human`` only.

    Returns
    -------
    str
        The rendered table, without a trailing newline.
    """
    if color_output_enabled():
        from prettytable import PrettyTable

        table = PrettyTable()
        table.title = title
        table.field_names = field_names
        for col, a in (align or {}).items():
            table.align[col] = a
        for col, width in (max_width or {}).items():
            table.max_width[col] = width
        for row in rows:
            table.add_row(list(row))
        return table.get_string()

    skip = 1 if field_names and field_names[0] in _INDEX_COLUMNS else 0
    lines = [f"{title}: {' | '.join(field_names[skip:])}"]
    for row in rows:
        cells = [str(cell).replace("\n", ", ") for cell in list(row)[skip:]]
        lines.append(f"  {' | '.join(cells)}")
    return "\n".join(lines)

colorize ¶

colorize(text: str, *codes: str) -> str

Wrap text with the given ANSI escape codes and a reset suffix.

Returns text unchanged when :func:color_output_enabled is False, so it's safe to use unconditionally. Multiple codes are concatenated (e.g. ANSI_BOLD + ANSI_CYAN).

Source code in eden/utils/format.py

def colorize(text: str, *codes: str) -> str:
    """Wrap ``text`` with the given ANSI escape ``codes`` and a reset suffix.

    Returns ``text`` unchanged when :func:`color_output_enabled` is False, so it's safe to
    use unconditionally. Multiple codes are concatenated (e.g. ``ANSI_BOLD + ANSI_CYAN``).
    """
    if not codes or not color_output_enabled():
        return text
    return f"{''.join(codes)}{text}{ANSI_RESET}"

Warning helpers: emitting a warning once where the same one would otherwise repeat.

Functions:¶

warn_once `cached` ¶

warn_once(message: str) -> None

Warn once per distinct text — for a diagnostic a loop would otherwise repeat verbatim.

Put the varying part (class name, version, path) in the message and the dedup follows it. The cache is bounded because it keys on the whole string; cache_clear() resets it.

Source code in eden/utils/warning.py

@functools.lru_cache(maxsize=128)
def warn_once(message: str) -> None:
    """Warn once per distinct text — for a diagnostic a loop would otherwise repeat verbatim.

    Put the varying part (class name, version, path) in the message and the dedup follows it.
    The cache is bounded because it keys on the whole string; ``cache_clear()`` resets it.
    """
    en.logger.warning(message)

A fixed-shape float np.ndarray pydantic field type, e.g. FloatArray[3, 3] for a camera intrinsic.

Eden's own replacement for numpydantic.NDArray[Shape["3, 3"], float], which imports cv2 (through numpydantic's video interface) and so would pull it into every task config for two camera-matrix annotations. Only numpy and pydantic here, so :mod:eden.options.camera can import it without an eden.utils <-> eden.options cycle.

A saved matrix loads back as an np.ndarray because EdenConfig.load_from_file builds every ConfigurableOptions with model_validate, which runs :class:FloatArrayShape on the nested list. Declare a FloatArray field on a ConfigurableOptions subclass only: a genesis-native Options loads through model_construct, so nothing would validate the field.

Classes¶

FloatArrayShape `dataclass` ¶

FloatArrayShape(shape: tuple[int, ...])

Annotated metadata that validates a value into a float np.ndarray of shape.

Accepts an np.ndarray or a nested sequence of real numbers. A float array is kept as-is (same object, same float dtype); anything else is converted with np.asarray and, if it holds integers, cast to float64. Serializes to nested lists in JSON mode and stays an np.ndarray in Python mode.

PARAMETER	DESCRIPTION
`shape`	The exact shape a value must have. TYPE: `tuple[int, ...]`

Attributes¶

shape `instance-attribute` ¶

shape: tuple[int, ...]

Methods:¶

FloatArray ¶

FloatArray[3, 3] is Annotated[np.ndarray, FloatArrayShape((3, 3))].

Examples:

>>> class Opts(pydantic.BaseModel):
...     intrinsic: FloatArray[3, 3] | None = None

Core Utilities¶

Package entry point¶

Functions:¶

init ¶

get_log_dir ¶

get_run_id ¶

get_run_timestamp ¶

Registries¶

Classes¶

Registry ¶

Methods:¶

register ¶

get ¶

keys ¶

TermRegistry ¶

Attributes¶

cli_kind property ¶

term_class property ¶

func_wrapper_class property ¶

Methods:¶

catalog_hint ¶

registered_class ¶

register ¶

VariantRegistry ¶

Methods:¶

register ¶

Functions:¶

module_is_initializing ¶

Configurable objects¶

Attributes¶

T module-attribute ¶

Classes¶

ConfigurableMixin ¶

Attributes¶

name property ¶

uid property ¶

options property ¶

Methods:¶

get_parameter_names classmethod ¶

fingerprint_names classmethod ¶

configured_fingerprint ¶

get_parameter_specs classmethod ¶

get_configure_specs classmethod ¶

configure classmethod ¶

ConfigurableFuncWrapperMixin ¶

Attributes¶

params class-attribute instance-attribute ¶

name property ¶

Methods:¶

get_function_parameter_specs classmethod ¶

get_required_function_parameter_names classmethod ¶

configure classmethod ¶

Functions:¶

frozen_value ¶

mirror_options_onto ¶

Parameter descriptions¶

Classes¶

ParamInfo dataclass ¶

Attributes¶

name instance-attribute ¶

annotation class-attribute instance-attribute ¶

default class-attribute instance-attribute ¶

required class-attribute instance-attribute ¶

Methods:¶

Functions:¶

render_annotation ¶

render_default ¶

function_param_specs ¶

render_params ¶

invalid_function_param_keys ¶

invalid_params_message ¶

unknown_parameter_names ¶

invalid_parameters_message ¶

validated_parameter_values ¶

Multi-GPU (DDP)¶

Functions:¶

get_local_rank ¶

get_global_rank ¶

get_world_size ¶

is_distributed ¶

cli_kind `property` ¶

term_class `property` ¶

func_wrapper_class `property` ¶

T `module-attribute` ¶

name `property` ¶

uid `property` ¶

options `property` ¶

get_parameter_names `classmethod` ¶

fingerprint_names `classmethod` ¶

get_parameter_specs `classmethod` ¶

get_configure_specs `classmethod` ¶

configure `classmethod` ¶

params `class-attribute` `instance-attribute` ¶

name `property` ¶

get_function_parameter_specs `classmethod` ¶

get_required_function_parameter_names `classmethod` ¶

configure `classmethod` ¶

ParamInfo `dataclass` ¶

name `instance-attribute` ¶

annotation `class-attribute` `instance-attribute` ¶

default `class-attribute` `instance-attribute` ¶

required `class-attribute` `instance-attribute` ¶

RUN_MANIFEST_FILENAME `module-attribute` ¶

RUN_MANIFEST_SCHEMA_VERSION `module-attribute` ¶

RL_LIBRARY_DISTRIBUTIONS `module-attribute` ¶

DISTRIBUTION_NAME `module-attribute` ¶

UNPUBLISHED_REQUIREMENTS `module-attribute` ¶

REINSTALLED_PACKAGES `module-attribute` ¶

UV_ONLY_EXTRAS `module-attribute` ¶

TermCost `dataclass` ¶

manager `instance-attribute` ¶

term `instance-attribute` ¶

us_per_step `instance-attribute` ¶

calls_per_step `instance-attribute` ¶

pct_step `class-attribute` `instance-attribute` ¶

ProfileReport `dataclass` ¶

num_steps `instance-attribute` ¶

step_us `instance-attribute` ¶

sync `instance-attribute` ¶

rows `class-attribute` `instance-attribute` ¶

HINT `class-attribute` `instance-attribute` ¶

NOTE `class-attribute` `instance-attribute` ¶

TermTiming `dataclass` ¶

manager `instance-attribute` ¶

label `instance-attribute` ¶

kind `instance-attribute` ¶

calls `class-attribute` `instance-attribute` ¶

total_s `class-attribute` `instance-attribute` ¶

self_s `class-attribute` `instance-attribute` ¶

sumsq_s `class-attribute` `instance-attribute` ¶

max_s `class-attribute` `instance-attribute` ¶

mean_s `property` ¶