Managers¶

A manager owns one aspect of the step loop (actions, observations, rewards, ...) and holds the terms that implement it. Each section below documents one manager module: the manager class, its term base class, and, for the kinds that accept plain functions, the function wrapper. For how to write a term, read the terms guide; for the order managers run in, read the environments guide.

from eden.managers import REWARD_TERM_REGISTRY, RewardTerm
from eden.terms import DoneTerm, ObsTerm, RewardTerm  # short aliases of the function wrappers

Some term kinds declare their shared fields (a reward's weight, an event's mode, ...) on a private mixin. Those blocks list the fields with inherited_members so they render on the term classes.

Base classes¶

ManagerTermBase is the base of every term, ConfigurableTermBase of every class term, and ConfigurableTermFuncWrapper of every function wrapper. ManagerBase builds terms from options through build_term, which adds a note naming the config entry to any exception a term's build() raises.

Base classes and shared logic for managers and their terms.

A manager owns an ordered set of terms resolved from its options through a :class:~eden.utils.registry.Registry. Terms come in two flavors:

function-based (stateless): a plain function registered with @REGISTRY.register() and wrapped in a FuncWrapper.
class-based (stateful): subclass the kind's term base (a :class:ConfigurableTermBase) and implement compute(), optionally build() and reset().

Concrete managers (action, observation, reward, termination, ...) live in sibling modules and are imported lazily inside the env's _load_managers() to break the manager <-> env circular import.

Attributes¶

T `module-attribute` ¶

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

Classes¶

ManagerTermBase ¶

ManagerTermBase(env: EnvBase)

Bases: RBC

Base class for all manager terms.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase):
    self._env = env
    self._check_env_options_requirements()

Attributes¶

num_envs `property` ¶

num_envs: int

The number of parallel environments.

device `property` ¶

device: device

The torch device the term's tensors live on.

Methods:¶

resolve_entity ¶

resolve_entity(name: str) -> Entity

Resolve a named entity from the env — the declared way for a term to reach an entity.

Resolve through this in compute() rather than caching the handle in build(): the deployment hot-swaps env.entities[name] (sim entity → RobotStateEntity) so observations read real robot state, and a cached handle would keep pointing at the stale sim entity. Cache only derived data (resolved link/DOF indices stay valid across the swap), not the entity itself. (DeploymentBase repoints any straggler cached handles by object identity as a safety net; resolving lazily means a term never needs it.)

An unknown name raises here, listing the names the scene does have.

Source code in eden/managers/base.py

def resolve_entity(self, name: str) -> Entity:
    """Resolve a named entity from the env — the declared way for a term to reach an entity.

    Resolve through this in ``compute()`` rather than caching the handle in ``build()``:
    the deployment hot-swaps ``env.entities[name]`` (sim entity → ``RobotStateEntity``) so
    observations read real robot state, and a cached handle would keep pointing at the stale
    sim entity. Cache only *derived* data (resolved link/DOF indices stay valid across the
    swap), not the entity itself. (``DeploymentBase`` repoints any straggler cached handles
    by object identity as a safety net; resolving lazily means a term never needs it.)

    An unknown name raises here, listing the names the scene does have.
    """
    return resolve_named_entity(self._env.entities, name)

reset ¶

reset(envs_idx: Tensor | None = None) -> Any

Reset the manager term.

Source code in eden/managers/base.py

def reset(self, envs_idx: torch.Tensor | None = None) -> Any:
    """Reset the manager term."""
    del envs_idx  # Unused.

post_reset ¶

post_reset(envs_idx: slice | Tensor) -> None

Hook invoked after ALL managers have reset envs_idx.

Runs after every manager's reset(), the reset-mode events, and the zeroing of episode_length_buf — the guaranteed point to rebuild per-step caches from the final post-reset scene state, without depending on the order managers reset in. (A term's own reset() runs mid-sequence: state it derives from other entities or terms may still change before observations are computed.) No-op by default.

Managers dispatch source caches before their consumers (for example, command terms before metric terms). Runs under torch.inference_mode: mutate build()-allocated buffers in place; never rebind attributes to tensors created inside this hook.

PARAMETER DESCRIPTION

envs_idx

The envs that were reset — never None, but index with it, never measure it: the auto-reset path in RLEnvBase.step passes the full-length boolean reset_buf, while EnvBase.reset passes a slice or an index tensor. So buf[envs_idx] = ... and src[envs_idx] are always right, and len(envs_idx) / envs_idx.numel() silently mean num_envs rather than the reset count on the step path.

TYPE: slice | Tensor

Source code in eden/managers/base.py

def post_reset(self, envs_idx: slice | torch.Tensor) -> None:
    """Hook invoked after ALL managers have reset ``envs_idx``.

    Runs after every manager's ``reset()``, the reset-mode events, and the zeroing of
    ``episode_length_buf`` — the guaranteed point to rebuild per-step caches from the
    *final* post-reset scene state, without depending on the order managers reset in.
    (A term's own ``reset()`` runs mid-sequence: state it derives from other entities
    or terms may still change before observations are computed.) No-op by default.

    Managers dispatch source caches before their consumers (for example, command
    terms before metric terms). Runs under ``torch.inference_mode``: mutate
    ``build()``-allocated buffers in place; never rebind attributes to tensors
    created inside this hook.

    Parameters
    ----------
    envs_idx : slice | torch.Tensor
        The envs that were reset — never ``None``, but *index with it, never measure
        it*: the auto-reset path in ``RLEnvBase.step`` passes the full-length boolean
        ``reset_buf``, while ``EnvBase.reset`` passes a slice or an index tensor. So
        ``buf[envs_idx] = ...`` and ``src[envs_idx]`` are always right, and
        ``len(envs_idx)`` / ``envs_idx.numel()`` silently mean ``num_envs`` rather
        than the reset count on the step path.
    """
    del envs_idx  # Unused.

compute `abstractmethod` ¶

compute(*args, **kwargs) -> Any

Step the manager term.

Source code in eden/managers/base.py

@abstractmethod
def compute(self, *args, **kwargs) -> Any:
    """Step the manager term."""
    raise NotImplementedError

build ¶

build() -> None

Deferred build hook for terms that resolve entities after env setup.

Source code in eden/managers/base.py

def build(self) -> None:
    """Deferred build hook for terms that resolve entities after env setup."""
    return None

ManagerTermFuncWrapperBase ¶

ManagerTermFuncWrapperBase(func: FunctionType, env: EnvBase)

Bases: ManagerTermBase

Runtime contract for manager terms defined as a function: hold func and call it.

Like :class:ManagerTermBase, this carries no options machinery. Concrete wrappers pair it with ConfigurableFuncWrapperMixin[KindOptions], which contributes configure() and the params attribute that :meth:compute forwards to the wrapped function.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase):
    self._func = func
    ManagerTermBase.__init__(self, env=env)

Methods:¶

compute ¶

compute(*args, **kwargs) -> Any

Invoke the wrapped function with the environment and configured params.

Source code in eden/managers/base.py

def compute(self, *args, **kwargs) -> Any:
    """Invoke the wrapped function with the environment and configured params."""
    return self._func(env=self._env, **self.params, **kwargs)

ConfigurableTermBase ¶

ConfigurableTermBase(env: EnvBase, options: _TermOptionsT)

Bases: ManagerTermBase, ConfigurableMixin[_TermOptionsT]

A class term bound to its options: the base every kind's class-term base derives from.

Parameterize it with the kind's options class (ConfigurableTermBase[RewardTermOptions]); ConfigurableMixin reads the type argument through this intermediate generic base. The configurable fields every term of a kind shares go on a ConfigurableMixin subclass the kind's class base and its :class:ConfigurableTermFuncWrapper both inherit, so they are declared once; such a mixin must not define __init__.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Methods:¶

ConfigurableTermFuncWrapper ¶

ConfigurableTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ManagerTermFuncWrapperBase, ConfigurableFuncWrapperMixin[_TermOptionsT]

The function-term counterpart of :class:ConfigurableTermBase: each kind's *FuncWrapper base.

_func is set before ConfigurableFuncWrapperMixin.__init__ runs, because that is where the params keys are checked against the wrapped function's signature.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

Methods:¶

ManagerBase ¶

ManagerBase(env: EnvBase, options: T)

Bases: RBC, Generic[T]

Base class for all managers.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: T):
    self._env = env
    self._options: T = options
    self._prepare_terms()

Attributes¶

device_synced_fields `class-attribute` ¶

device_synced_fields: frozenset[str] = frozenset()

num_envs `property` ¶

num_envs: int

The number of parallel environments.

device `property` ¶

device: str

The torch device the manager's tensors live on.

Methods:¶

compute_once ¶

compute_once(key: Hashable, producer: Callable[[], _R]) -> _R

:meth:EnvBase.compute_once <eden.envs.base.EnvBase.compute_once> of this manager's env.

The memo belongs to the env's pass, not to a manager: a key a term of one manager produced is served to a term of another within one pass.

Source code in eden/managers/base.py

def compute_once(self, key: Hashable, producer: Callable[[], _R]) -> _R:
    """:meth:`EnvBase.compute_once <eden.envs.base.EnvBase.compute_once>` of this manager's env.

    The memo belongs to the env's pass, not to a manager: a key a term of one manager produced is served
    to a term of another within one pass.
    """
    return self._env.compute_once(key, producer)

compute_pass ¶

compute_pass() -> AbstractContextManager[None]

:meth:EnvBase.compute_pass <eden.envs.base.EnvBase.compute_pass> of this manager's env.

A manager wraps its term loop in this; inside a pass the env already opened, it joins that one.

Source code in eden/managers/base.py

def compute_pass(self) -> AbstractContextManager[None]:
    """:meth:`EnvBase.compute_pass <eden.envs.base.EnvBase.compute_pass>` of this manager's env.

    A manager wraps its term loop in this; inside a pass the env already opened, it joins that one.
    """
    return self._env.compute_pass()

sync_device_fields ¶

sync_device_fields() -> None

Bring the device copies of :attr:device_synced_fields up to date with the terms' values (a no-op here).

Source code in eden/managers/base.py

def sync_device_fields(self) -> None:
    """Bring the device copies of :attr:`device_synced_fields` up to date with the terms' values (a no-op here)."""

reset ¶

reset(envs_idx: Tensor | None = None) -> dict[str, Any]

Reset the manager and return logging info for the current step.

Source code in eden/managers/base.py

def reset(self, envs_idx: torch.Tensor | None = None) -> dict[str, Any]:
    """Reset the manager and return logging info for the current step."""
    del envs_idx  # Unused.
    return {}

compute `abstractmethod` ¶

compute(*args, **kwargs) -> Any

Compute the manager.

Source code in eden/managers/base.py

@abstractmethod
def compute(self, *args, **kwargs) -> Any:
    """Compute the manager."""
    raise NotImplementedError

iter_post_reset_terms `abstractmethod` ¶

iter_post_reset_terms() -> Iterable[ManagerTermBase]

Yield every term this manager owns, for the :meth:post_reset fan-out.

Abstract rather than defaulted so a new manager cannot silently drop the hook: :meth:post_reset is the only generic thing the env can call on an arbitrary :class:ManagerBase, and a no-op default would make a missed term look like a term that simply had nothing to rebuild. May yield the same term more than once (a term shared across observation groups or event modes); the fan-out dedupes.

Source code in eden/managers/base.py

@abstractmethod
def iter_post_reset_terms(self) -> Iterable[ManagerTermBase]:
    """Yield every term this manager owns, for the :meth:`post_reset` fan-out.

    Abstract rather than defaulted so a new manager cannot silently drop the hook:
    :meth:`post_reset` is the only generic thing the env can call on an arbitrary
    :class:`ManagerBase`, and a no-op default would make a missed term look like a
    term that simply had nothing to rebuild. May yield the same term more than once
    (a term shared across observation groups or event modes); the fan-out dedupes.
    """
    raise NotImplementedError

post_reset ¶

post_reset(envs_idx: slice | Tensor | None = None) -> None

Fan :meth:ManagerTermBase.post_reset out once per unique term.

The env calls this from its public entry points (:meth:~eden.envs.base.EnvBase.reset, :meth:~eden.envs.base.RLEnvBase.step) right after _reset_idx returns — not from inside it, so a subclass's extra reset work finishes first (the mimic env refills the motion buffer after super()._reset_idx). By then every manager's reset(), the reset-mode events and the episode_length_buf zeroing have all run, so terms can rebuild per-step caches from the final post-reset state. The env owns that call-time contract, like CommandManager.post_physics_refresh.

Decorated @torch.inference_mode() so terms whose buffers were created under an inference-moded manager path (metric and reward terms) can mutate them here; the flip side is the in-place-only contract in :meth:ManagerTermBase.post_reset.

PARAMETER	DESCRIPTION
`envs_idx`	The envs that were reset; `None` means all of them. Sanitized before it reaches the terms. TYPE: `slice \| Tensor` DEFAULT: `None`

Source code in eden/managers/base.py

@torch.inference_mode()
def post_reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    """Fan :meth:`ManagerTermBase.post_reset` out once per unique term.

    The env calls this from its public entry points (:meth:`~eden.envs.base.EnvBase.reset`,
    :meth:`~eden.envs.base.RLEnvBase.step`) right *after* ``_reset_idx`` returns — not from
    inside it, so a subclass's extra reset work finishes first (the mimic env refills the
    motion buffer after ``super()._reset_idx``). By then every manager's ``reset()``, the
    reset-mode events and the ``episode_length_buf`` zeroing have all run, so terms can
    rebuild per-step caches from the final post-reset state. The env owns that call-time
    contract, like ``CommandManager.post_physics_refresh``.

    Decorated ``@torch.inference_mode()`` so terms whose buffers were created under an
    inference-moded manager path (metric and reward terms) can mutate them here; the flip
    side is the in-place-only contract in :meth:`ManagerTermBase.post_reset`.

    Parameters
    ----------
    envs_idx : slice | torch.Tensor, optional
        The envs that were reset; ``None`` means all of them. Sanitized before it reaches
        the terms.
    """
    envs_idx = sanitize_envs_idx(envs_idx, self._env.num_envs)
    seen: set[int] = set()
    for term in self.iter_post_reset_terms():
        term_id = id(term)
        if term_id in seen:
            continue
        seen.add(term_id)
        term.post_reset(envs_idx)

FlatManagerBase ¶

FlatManagerBase(env: EnvBase, options: T)

Bases: ManagerBase[T]

Base for managers whose terms form a single flat, ordered set.

Owns the flat-term API (terms, active_terms, get_term, get_term_options) backed by _terms / _term_names, which :meth:_prepare_terms populates. The hierarchical managers (observations by group, events by mode) inherit :class:ManagerBase directly and expose their own named views instead.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: T):
    self._term_names: list[str] = list()
    self._terms: dict[str, ManagerTermBase] = dict()
    super().__init__(env, options)
    #: The terms a manager's ``reset()`` dispatches to: the class terms. A function term's ``reset()``
    #: is the inherited no-op, and the auto-reset calls ``reset()`` every step. Fixed at construction
    #: (``CommandManager.swap_term`` replaces terms after it, and resets every term itself).
    self._reset_terms: list[ManagerTermBase] = [
        term for term in self._terms.values() if not isinstance(term, ManagerTermFuncWrapperBase)
    ]

Attributes¶

terms `property` ¶

terms: dict[str, ManagerTermBase]

Mapping from term name to the built :class:ManagerTermBase instance.

active_terms `property` ¶

active_terms: list[str]

The names of the active terms, in registration order (the live internal list; do not mutate).

Methods:¶

get_term ¶

get_term(name: str) -> ManagerTermBase

Return the active term registered under name.

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active term. The message lists the active ones, in the wording :meth:`eden.utils.registry.Registry.get` uses for the same miss: a caller reaching here with the wrong name (a term renamed in the config, a reward term naming an action term that was never added) has no other way to see what is there.

Source code in eden/managers/base.py

def get_term(self, name: str) -> ManagerTermBase:
    """Return the active term registered under ``name``.

    Raises
    ------
    KeyError
        If ``name`` is not an active term. The message lists the active ones, in the wording
        :meth:`eden.utils.registry.Registry.get` uses for the same miss: a caller reaching here
        with the wrong name (a term renamed in the config, a reward term naming an action term
        that was never added) has no other way to see what *is* there.
    """
    if name not in self._terms:
        available = ", ".join(self._term_names) or "none"
        raise KeyError(f"No active term named '{name}' in {type(self).__name__}. Available options: {available}")
    return self._terms[name]

get_term_options ¶

get_term_options(term_name: str) -> Options

Return the options object for the active term term_name.

RAISES	DESCRIPTION
`ValueError`	If `term_name` is not an active term.

Source code in eden/managers/base.py

def get_term_options(self, term_name: str) -> Options:
    """Return the options object for the active term ``term_name``.

    Raises
    ------
    ValueError
        If ``term_name`` is not an active term.
    """
    if term_name not in self._term_names:
        raise ValueError(f"Term '{term_name}' not found in active terms.")
    return self._terms[term_name].options

iter_post_reset_terms ¶

iter_post_reset_terms() -> Iterable[ManagerTermBase]

The flat term set, in registration order.

Source code in eden/managers/base.py

def iter_post_reset_terms(self) -> Iterable[ManagerTermBase]:
    """The flat term set, in registration order."""
    return self._terms.values()

Functions:¶

build_term ¶

build_term(registry: Registry, env: EnvBase, options: Options, *, label: str) -> ManagerTermBase

Construct the term options names from registry and build() it.

The one place a term is built from its options — by every manager (:meth:ManagerBase._build_term, the observation manager's groups) and by the composite terms that build children of their own.

PARAMETER	DESCRIPTION
`registry`	The term registry `options.name` is a key of. TYPE: `Registry`
`env`	The env the term is built for. TYPE: `EnvBase`
`options`	The term's options; `options.name` is its registry key. TYPE: `Options`
`label`	Names the config entry being built, starting with the class that holds it (`"RewardManager term 'lift_milk'"`, `"MetricTermAND.terms[0]"`): the term itself only knows its key, and a task can hold a dozen entries of one term (`metric_reward` ×12 in robosuite PickPlace), so the message alone may not say which to fix. It opens the error for a nameless term and is added as a note to any exception constructing or building the term raises, whose type and message are left unchanged. TYPE: `str`

RAISES	DESCRIPTION
`ValueError`	If `options.name` is `None` — a legal placeholder the registry would answer with `KeyError: None`, naming neither the entry nor the fix.

Source code in eden/managers/base.py

def build_term(registry: Registry, env: EnvBase, options: Options, *, label: str) -> ManagerTermBase:
    """Construct the term ``options`` names from ``registry`` and ``build()`` it.

    The one place a term is built from its options — by every manager (:meth:`ManagerBase._build_term`,
    the observation manager's groups) and by the composite terms that build children of their own.

    Parameters
    ----------
    registry : Registry
        The term registry ``options.name`` is a key of.
    env : EnvBase
        The env the term is built for.
    options : Options
        The term's options; ``options.name`` is its registry key.
    label : str
        Names the config entry being built, starting with the class that holds it
        (``"RewardManager term 'lift_milk'"``, ``"MetricTermAND.terms[0]"``): the term itself only knows its
        key, and a task can hold a dozen entries of one term (``metric_reward`` ×12 in robosuite
        PickPlace), so the message alone may not say which to fix. It opens the error for a nameless term
        and is added as a note to any exception constructing or building the term raises, whose type and
        message are left unchanged.

    Raises
    ------
    ValueError
        If ``options.name`` is ``None`` — a legal placeholder the registry would answer with
        ``KeyError: None``, naming neither the entry nor the fix.
    """
    if options.name is None:
        raise ValueError(
            f"{label} names no registered term (name=None), so there is nothing to build. "
            f"Write it as <TermClass>.configure(...), which sets name, or give it name='<registry key>' — "
            f"`eden terms --grep <text>` lists them."
        )
    try:
        term = registry.get(options.name)(env=env, options=options)
        term.build()
    except Exception as e:
        e.add_note(f"while building {label} (name='{options.name}')")
        raise
    return term

ActionManager¶

Action manager and ActionTerm base: map policy actions to actuator commands.

The manager slices the policy action vector across its active :class:ActionTerm instances and applies them every control step (re-applied across decimation physics sub-steps). ActionTerm.dofs_order maps a DOF name to its position in the term's dofs_name; mind the DOF-ordering caveat in :mod:eden.entities.rigid when wiring actions to joints.

Attributes¶

ACTION_TERM_REGISTRY `module-attribute` ¶

ACTION_TERM_REGISTRY = TermRegistry('ACTION_TERM', ActionTerm, populated_by=('eden.managers.terms.actions',))

Classes¶

ActionTerm ¶

ActionTerm(env: EnvBase, options: ActionTermOptions)

Bases: ConfigurableTermBase[ActionTermOptions]

Base class for action terms.

Source code in eden/managers/action_manager.py

def __init__(self, env: EnvBase, options: ActionTermOptions):
    super().__init__(env=env, options=options)

    self._entity = None
    self.dofs_idx_local = None
    # The memo behind ``genesis_dofs_idx``: its value, and the ``dofs_idx_local`` it was derived from.
    self._genesis_dofs_idx_source: torch.Tensor | None = None
    self._genesis_dofs_idx: range | torch.Tensor | None = None
    self._n_dofs = 0
    self._raw_action = None
    self._prev_action = None
    self._processed_action = None
    # Set once the first-step command-shape check has run (see ``verify_command_shape``).
    self._command_shape_checked = False

Attributes¶

entity_name `class-attribute` `instance-attribute` ¶

entity_name: str = ''

dofs_name `class-attribute` `instance-attribute` ¶

dofs_name: str | list[str] = []

dofs_idx_local `instance-attribute` ¶

dofs_idx_local = None

entity `property` ¶

entity: Entity

genesis_dofs_idx `property` ¶

genesis_dofs_idx: range | Tensor

dofs_idx_local in the form to hand the entity's DOF accessors: a range when contiguous.

Selects exactly the DOFs dofs_idx_local does, in the same order, but a range takes Genesis' slice path — sync-free and ~4x cheaper on the host than an index tensor, which matters in apply_actions(), called once per physics substep (see :func:eden.utils.misc.contiguous_range). Derived from whichever tensor dofs_idx_local currently holds, so a subclass that rebinds it in its own build() never reads a stale one.

action `property` ¶

action: Tensor

The raw action this term received from the policy this step, shape (num_envs, action_dim).

prev_action `property` ¶

prev_action: Tensor

The raw action this term received on the previous step, shape (num_envs, action_dim).

processed_action `property` ¶

processed_action: Tensor

The command this term issued this step — what a deployment sends to the robot.

For the joint PD controllers this is the position target the controller tracks — raw_action * scale + offset + reference_offset, after any configured :class:~eden.managers.modifiers.ActionModifier's control-step modify_processed_action. Other terms command other quantities (a velocity, an effort, a binary weld flag), so both the units and the trailing width are term-specific; only the leading num_envs dimension is guaranteed.

A physics-step modifier (modify_applied_target — e.g. :class:~eden.managers.modifiers.actions.actuators.ActionDelay) is not folded in: under a delay the simulated actuator tracks an earlier processed_action, exactly as the real robot tracks the command it received a latency ago.

Read this rather than recomputing the target at the call site — a recomputation silently drops whatever control-step modifiers are configured, so the logged target diverges from the one the controller issued.

sim_surrogate `property` ¶

sim_surrogate: bool

Whether this term only stands in, in simulation, for something the real robot does itself.

True means the physical effect follows from another term's command (or from physics) on hardware, so a deployment has nothing to send for this term: :class:~eden.managers.terms.actions.welding_actions.ParallelJawWelding welds the object to the gripper because sim contact friction is unreliable, while on the robot the gripper joints closing — commanded by a separate joint term — hold it.

False (the default) means the term commands something in its own right, and a deployment that cannot carry that command must say so rather than drop it: :class:~eden.managers.terms.actions.welding_actions.SuctionCupWelding is the vacuum on/off bit, which no other term supplies. See :func:eden.extensions.deployment.base.build_term_action_maps.

dofs_order `property` ¶

dofs_order: dict[str, int]

Get the mapping from dofs name to its local index.

action_dim `property` ¶

action_dim: int

num_dofs `property` ¶

num_dofs: int

Methods:¶

build ¶

build() -> None

Source code in eden/managers/action_manager.py

def build(self) -> None:
    self._entity = self.resolve_entity(self.entity_name)
    self.dofs_name, dofs_idx_local = self.entity.find_named_dofs_idx_local(
        self.dofs_name, name_scope=self.entity.dofs_name, preserve_order=True
    )
    self._reject_multi_dof_joints()
    self.dofs_idx_local = torch.as_tensor(dofs_idx_local, dtype=gs.tc_int, device=self.device).contiguous()
    self._n_dofs = len(self.dofs_idx_local)
    self._raw_action = torch.zeros(self.num_envs, self.action_dim, device=self.device)
    self._prev_action = torch.zeros(self.num_envs, self.action_dim, device=self.device)
    self._processed_action = torch.zeros(self.num_envs, self.num_dofs, device=self.device)

verify_command_shape ¶

verify_command_shape(command: Tensor) -> None

Fail loud — naming this term — the first time a control command's width != the DOF count.

A wrong-width joint command otherwise reaches the raw Genesis call and surfaces as an opaque GenesisException: Invalid input shape: (N, W)... required size D naming neither the term, the entity, nor the DOF set. Controllers call this from apply_actions with the exact tensor they hand to entity.control_dofs_* — the only layer that knows both the term identity and the true (post-slice) command width.

Checked once per term, not every step: the command width is structural (fixed by the term's dofs_name and controller), so a first-step check catches a misconfiguration without any steady-state cost. No-op for a shape-less command (a Python scalar Genesis handles itself).

Source code in eden/managers/action_manager.py

def verify_command_shape(self, command: torch.Tensor) -> None:
    """Fail loud — naming this term — the first time a control command's width != the DOF count.

    A wrong-width joint command otherwise reaches the raw Genesis call and surfaces as an opaque
    ``GenesisException: Invalid input shape: (N, W)... required size D`` naming neither the term,
    the entity, nor the DOF set. Controllers call this from ``apply_actions``
    with the exact tensor they hand to ``entity.control_dofs_*`` — the only layer that knows both
    the term identity and the true (post-slice) command width.

    Checked once per term, not every step: the command width is structural (fixed by the term's
    ``dofs_name`` and controller), so a first-step check catches a misconfiguration without any
    steady-state cost. No-op for a shape-less command (a Python scalar Genesis handles itself).
    """
    if self._command_shape_checked:
        return
    self._command_shape_checked = True
    shape = getattr(command, "shape", None)
    if shape is None or len(shape) == 0:
        return
    width = shape[-1]
    if width != self._n_dofs:
        raise ValueError(
            f"Action term {term_subject(self)} produced a command of "
            f"width {width}, but it controls {self._n_dofs} DOFs ({self.dofs_name}). The command's last "
            f"dimension must equal the number of controlled DOFs — check the term's `dofs_name` and its "
            f"controller output size."
        )

apply_actions `abstractmethod` ¶

apply_actions() -> None

Source code in eden/managers/action_manager.py

@abstractmethod
def apply_actions(self) -> None:
    raise NotImplementedError

ActionManager ¶

ActionManager(env: EnvBase, options: ActionManagerOptions)

Bases: FlatManagerBase[ActionManagerOptions]

Action manager for processing actions sent to the environment.

Source code in eden/managers/action_manager.py

def __init__(self, env: EnvBase, options: ActionManagerOptions):
    super().__init__(env=env, options=options)

    self._reject_no_op_terms()
    self._reject_overlapping_dofs()

    # Create action slices after terms are prepared
    self._term_slice = {}
    idx = 0
    for term_name, term in self._terms.items():
        self._term_slice[term_name] = slice(idx, idx + term.action_dim)
        idx += term.action_dim

    # Create buffers to store actions.
    self._action = torch.zeros((self.num_envs, self.total_action_dim), device=self.device)
    self._prev_action = torch.zeros_like(self._action)

    # Cached once: the dict-action width check runs only in smoke mode (a process-level flag
    # set before construction), so real training pays nothing — not even the env lookup — per step.
    self._smoke = is_smoke()

Attributes¶

total_action_dim `cached` `property` ¶

total_action_dim: int

action_term_dim `cached` `property` ¶

action_term_dim: list[int]

The order corresponds to the .active_terms property.

total_dofs_dim `cached` `property` ¶

total_dofs_dim: int

dofs_term_dim `cached` `property` ¶

dofs_term_dim: list[int]

The order corresponds to the .active_terms property.

action `property` ¶

action: Tensor

The raw action vector received this step, shape (num_envs, total_action_dim).

prev_action `property` ¶

prev_action: Tensor

The raw action vector received on the previous step, shape (num_envs, total_action_dim).

processed_actions `property` ¶

processed_actions: dict[str, Tensor]

Each term's :attr:ActionTerm.processed_action, keyed by term name.

A mapping rather than one concatenated tensor (unlike :attr:action): terms command different quantities with different widths and dtypes, so there is no meaningful flat layout to concatenate them into. Hence the plural name — it keeps the type visible at the call site, so indexing it like the flat :attr:action fails with a clear AttributeError on the singular spelling rather than a puzzling KeyError from a tensor-style slice on a dict.

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/action_manager.py

def summary(self) -> str:
    return self._format_summary_table(
        title=f"Active Action Terms (shape: {self.total_action_dim})",
        field_names=["Index", "Name", "Dimension", "Slice"],
        rows=(
            [index, name, term.action_dim, self._term_slice[name]]
            for index, (name, term) in enumerate(self._terms.items())
        ),
        align={"Name": "l", "Dimension": "r", "Slice": "r"},
    )

clear_action_history ¶

clear_action_history(envs_idx: slice | Tensor | None = None) -> None

Zero the raw actions :meth:reset zeroes, the manager's and each term's, current and previous.

For a caller that restarts what a policy observes without resetting the env: a deployment's START or policy switch, where the robot keeps running. A last_action observation then reads zeros, as it does after an env reset. :meth:reset itself does more, which such a caller must not: it resets every term's modifiers and an ACCUMULATE target, state a reset-mode DR event rewrites right after it in an env reset and that nothing rewrites on a robot.

Source code in eden/managers/action_manager.py

def clear_action_history(self, envs_idx: (slice | torch.Tensor) | None = None) -> None:
    """Zero the raw actions :meth:`reset` zeroes, the manager's and each term's, current and previous.

    For a caller that restarts what a policy observes without resetting the env: a deployment's
    START or policy switch, where the robot keeps running. A ``last_action`` observation then reads
    zeros, as it does after an env reset. :meth:`reset` itself does more, which such a caller must
    not: it resets every term's modifiers and an ``ACCUMULATE`` target, state a reset-mode DR event
    rewrites right after it in an env reset and that nothing rewrites on a robot.
    """
    if envs_idx is None:
        envs_idx = slice(None)
    self._prev_action[envs_idx] = 0.0
    self._action[envs_idx] = 0.0
    for term in self._terms.values():
        if term._raw_action is not None:
            term._raw_action[envs_idx] = 0.0
            term._prev_action[envs_idx] = 0.0

reset ¶

reset(envs_idx: slice | Tensor | None = None) -> dict[str, float]

Source code in eden/managers/action_manager.py

def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> dict[str, float]:
    if envs_idx is None:
        envs_idx = slice(None)

    # Reset action history.
    # NOTE: use assignment here (it works for both slice and tensor index), not .zero_()
    self._prev_action[envs_idx] = 0.0
    self._action[envs_idx] = 0.0

    # Reset action terms.
    for term in self._terms.values():
        term.reset(envs_idx=envs_idx)
    return {}

compute ¶

compute(action: Tensor | dict[str, Tensor]) -> None

Source code in eden/managers/action_manager.py

def compute(self, action: torch.Tensor | dict[str, torch.Tensor]) -> None:
    if isinstance(action, dict):
        if self._smoke:
            # Smoke-only: name the offending term before term.compute turns a width mismatch
            # into a bare torch broadcast error (or an unknown key into a bare KeyError).
            self._validate_dict_action(action)
        for term_name, term_action in action.items():
            # The tensor form below copies into the device buffer, so a host-built action works there; the
            # dict form hands each term its tensor as given. Moved here (a no-op for one already on the
            # device) so both forms accept `torch.tensor([...])` — a CPU tensor, since Eden sets no torch
            # default device.
            term_action = term_action.to(self.device)
            term = self._terms[term_name]
            term.compute(term_action)
            if term._accepts_missing_command_:
                term_action = torch.where(torch.isnan(term_action), term.action, term_action)
            self._prev_action[:, self._term_slice[term_name]] = self._action[:, self._term_slice[term_name]]
            self._action[:, self._term_slice[term_name]] = term_action
    else:
        if self.total_action_dim != action.shape[1]:
            # Name the per-term layout so the caller can see which term owns which columns —
            # the mismatch is otherwise reported only as two totals.
            layout = ", ".join(f"{name}={term.action_dim}" for name, term in self._terms.items())
            raise ValueError(
                f"Invalid action shape, expected: {self.total_action_dim}, received: {action.shape[1]}. "
                f"Expected per-term layout: [{layout}]."
            )
        self._prev_action[:] = self._action
        self._action[:] = action
        for term_name, term in self._terms.items():
            term_actions = self._action[:, self._term_slice[term_name]]
            term.compute(term_actions)
            if term._accepts_missing_command_:
                self._action[:, self._term_slice[term_name]] = torch.where(
                    torch.isnan(term_actions), term.action, term_actions
                )

apply_actions ¶

apply_actions() -> None

Source code in eden/managers/action_manager.py

def apply_actions(self) -> None:
    for term in self._terms.values():
        term.apply_actions()

Functions:¶

ObservationManager¶

Observation manager for computing observations.

Attributes¶

ObservationTermLike `module-attribute` ¶

ObservationTermLike: TypeAlias = ObservationTerm | ObservationTermFuncWrapper

Union of a class-based observation term and its function-wrapper form.

OBSERVATION_TERM_REGISTRY `module-attribute` ¶

OBSERVATION_TERM_REGISTRY = TermRegistry('OBSERVATION_TERM', ObservationTerm, ObservationTermFuncWrapper, populated_by=('eden.managers.terms.observations',))

Classes¶

ObservationTerm ¶

ObservationTerm(env: EnvBase, options: ObservationTermOptions)

Bases: ConfigurableTermBase[ObservationTermOptions], _ObservationTermFields

Base class for observation terms.

compute() is a read, not a tick. The manager calls it more often than once per control step and never promises which call is "the" step: once per term at build to learn the output shape (before any reset()), from EnvBase.reset(), at phase 19 of RLEnvBase.step, again at phase 12 for the final-observation snapshot when record_final_observations is on, from RslRlVecEnvWrapper.get_observations() outside the step, and from the deploy pre-flight probe. The cross-group raw cache goes the other way and may call compute() on only one of several identically configured instances. A term that advances state inside compute() — an integral, a counter, a latch — therefore double-counts under any of those, silently: the failure is a wrong number, not an exception.

Per-step state lives on the term and moves in the lifecycle hooks instead: build() allocates the buffer, :meth:post_physics_refresh advances it exactly once per control step on the post-physics state (phase 7 — the same hook, epoch and reasoning as a command term's measurement state at 5, and after the metrics at 6 so a term may integrate one), reset(envs_idx) zeroes the rows of the envs that restart, post_reset(envs_idx) rebuilds what the post-reset scene determines, and compute() returns the buffer. :class:~eden.managers.terms.observations.locomotion.VelocityTrackingErrorIntegral is the shipped worked example. Function terms have none of these hooks — see :class:ObservationTermFuncWrapper.

The post-compute fields (modifiers/noise/clip/scale/history) are options-only (see ObservationTermOptions.POST_COMPUTE_FIELDS): configure() accepts them and stores them on the options, but they are NOT mirrored as term attributes — the manager applies them from outside so a compute() can't accidentally read (and be wrongly deduped on) a modifier value.

When each modifier is read. clip and history_length are read off self._options on every step, so a post-build write to them takes effect immediately. noise and scale are resolved once at build — into self._noise and self._resolved_scale — because both need setup the hot path must not repeat (a noise model instance; a length check against the term's feature dim plus a device tensor). Writing term._options.scale after env.build() therefore has no effect: it is not the value the manager multiplies by, and nothing detects the divergence for you. Read what is actually applied off :attr:effective_scale; to retune it at runtime, call :meth:~eden.managers.observation_manager.ObservationManager.set_term_scale, which re-runs the feature-dim check. Modifiers stay the manager's business in both directions — that is what keeps them off this class.

Source code in eden/managers/observation_manager.py

def __init__(self, env: EnvBase, options: ObservationTermOptions):
    super().__init__(env=env, options=options)
    self._init_post_compute_fields(env)

Attributes¶

effective_scale `property` ¶

effective_scale: float | Tensor | None

The scale actually applied to this term's output, as resolved at build.

Not the configured scale option: a vector is resolved to a device tensor, and anything that multiplies out to identity — the scalar 1.0 or an all-ones vector — resolves to None, meaning the manager skips the multiply entirely. None before the term is built into a group. Mirrors :attr:~eden.managers.terms.actions.joint_actions.JointAction.effective_scale.

Methods:¶

post_physics_refresh ¶

post_physics_refresh() -> None

Advance per-step state on the post-physics scene (phase 7 — the grading epoch).

No-op by default. The one call per control step that a stateful term may count on: it runs after the physics sub-steps, after command_manager.post_physics_refresh (5) and metric_manager.compute (6) — so a term may integrate a refreshed command cache or a shared metric and see this step's value — and before termination/reward; and, unlike compute(), exactly once, whatever else reads the observation this step. Accumulate here (add_ into a build()-time buffer, in place), and return the buffer from compute().

Running above the auto-reset (phase 13) is what gives a stateful observation the semantics one expects of it: the final-observation snapshot at phase 12 carries this step's increment, reset() then zeroes the envs that restart, and the fresh episode's first observation at phase 19 reads zero rather than one post-reset increment. EnvBase.reset() and the build-time shape probe run no phase 7 at all, so a term reads as zero there too.

Invoked by :meth:ObservationManager.post_physics_refresh from RLEnvBase.step and EnvBase.step, and, on hardware, once per control tick from DeploymentBase.state_to_observation after the robot state is synced (a command term's post_physics_refresh does not run there; this one does, because the policy's observations are the hardware path). Never from compute(), and never from a reset.

Source code in eden/managers/observation_manager.py

def post_physics_refresh(self) -> None:
    """Advance per-step state on the post-physics scene (phase 7 — the *grading* epoch).

    No-op by default. The one call per control step that a stateful term may count on: it runs
    after the physics sub-steps, after ``command_manager.post_physics_refresh`` (5) and
    ``metric_manager.compute`` (6) — so a term may integrate a refreshed command cache or a shared
    metric and see this step's value — and before termination/reward; and, unlike ``compute()``,
    exactly once, whatever else reads the observation this step. Accumulate here (``add_`` into a
    ``build()``-time buffer, in place), and return the buffer from ``compute()``.

    Running *above* the auto-reset (phase 13) is what gives a stateful observation the semantics
    one expects of it: the final-observation snapshot at phase 12 carries this step's increment,
    ``reset()`` then zeroes the envs that restart, and the fresh episode's first observation at
    phase 19 reads zero rather than one post-reset increment. ``EnvBase.reset()`` and the build-time
    shape probe run no phase 7 at all, so a term reads as zero there too.

    Invoked by :meth:`ObservationManager.post_physics_refresh` from ``RLEnvBase.step`` and
    ``EnvBase.step``, and, on hardware, once per control tick from
    ``DeploymentBase.state_to_observation`` after the robot state is synced (a command term's
    ``post_physics_refresh`` does **not** run there; this one does, because the policy's
    observations are the hardware path). Never from ``compute()``, and never from a reset.
    """

ObservationTermFuncWrapper ¶

ObservationTermFuncWrapper(func: FunctionType, env: EnvBase, options: ObservationTermOptions)

Bases: ConfigurableTermFuncWrapper[ObservationTermOptions], _ObservationTermFields

Base class for observation terms defined as function. See :class:ObservationTerm for the modifier fields (options-only, not mirrored as attributes), for when each one is read, and for :attr:effective_scale.

Stateless by construction. A function term is a pure forward of env state: the manager never dispatches reset(), post_reset() or post_physics_refresh() to it (the inherited reset and post_reset are no-ops, and there is no refresh hook to call), so state kept in a closure or a module global is never zeroed at episode reset and advances on every one of the several compute() calls a step may make. A term that needs per-episode or per-step state is a class term.

Source code in eden/managers/observation_manager.py

def __init__(self, func: FunctionType, env: EnvBase, options: ObservationTermOptions):
    super().__init__(func=func, env=env, options=options)
    self._init_post_compute_fields(env)

Attributes¶

effective_scale `property` ¶

effective_scale: float | Tensor | None

The scale actually applied to this term's output, as resolved at build.

Not the configured scale option: a vector is resolved to a device tensor, and anything that multiplies out to identity — the scalar 1.0 or an all-ones vector — resolves to None, meaning the manager skips the multiply entirely. None before the term is built into a group. Mirrors :attr:~eden.managers.terms.actions.joint_actions.JointAction.effective_scale.

Methods:¶

ObservationManager ¶

ObservationManager(env: EnvBase, options: ObservationManagerOptions)

Bases: ManagerBase[ObservationManagerOptions]

Source code in eden/managers/observation_manager.py

def __init__(self, env: EnvBase, options: ObservationManagerOptions):
    super().__init__(env=env, options=options)

    self._group_obs_dim: dict[str, tuple[int, ...] | list[tuple[int, ...]]] = dict()
    self._obs_buffer: dict[str, torch.Tensor | dict[str, torch.Tensor]] | None = None

    for group_name, group_term_dims in self._group_obs_term_dim.items():
        if self._group_obs_options[group_name].concatenate_terms:
            try:
                term_dims = torch.stack(
                    [torch.tensor(dims, device="cpu") for dims in group_term_dims],
                    dim=0,
                )
                if len(term_dims.shape) > 1:
                    dim = self._group_obs_options[group_name].concatenate_dim
                    dim_sum = torch.sum(term_dims[:, dim], dim=0)
                    term_dims[0, dim] = dim_sum
                    term_dims = term_dims[0]
                else:
                    term_dims = torch.sum(term_dims, dim=0)
                self._group_obs_dim[group_name] = tuple(term_dims.tolist())
            except RuntimeError:
                raise RuntimeError(f"Unable to concatenate observation terms in group `{group_name}`.") from None
        else:
            self._group_obs_dim[group_name] = group_term_dims

    # Cross-group raw-output cache. Many tasks reuse the same observation term in multiple groups (e.g. an
    # identical `policy` and `critic` group that differ only in noise/corruption); the raw `term.compute()` is then
    # computed once per group even though it's identical. Terms whose raw output is provably identical (same
    # registered term + same compute params, ignoring post-compute noise/clip/scale/history) compute once per step
    # and the result is reused. Safe because compute() is contractually a pure read of env state (see
    # ``ObservationTerm``): a stateful term advances in ``post_physics_refresh``, which reaches every instance,
    # so two identically configured instances hold identical state and either one's compute() stands for both.
    self._raw_cache: dict = {}
    self._build_dup_signatures()

    # ``(group, term)`` pairs that ``nan_policy="warn"`` has already warned about. Each pair is warned
    # about once and after that only sanitized, with no host read: a warning every step would flood a
    # training log.
    self._non_finite_warned: set[tuple[str, str]] = set()

Attributes¶

group_term_names `property` ¶

group_term_names: dict[str, list[str]]

The active observation term names keyed by group, in registration order (a copy, safe to mutate).

active_terms `property` ¶

active_terms: list[str]

The flat, de-duplicated union of all active term names across groups.

group_obs_dim `property` ¶

group_obs_dim: dict[str, tuple[int, ...] | list[tuple[int, ...]]]

group_obs_term_dim `property` ¶

group_obs_term_dim: dict[str, list[tuple[int, ...]]]

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/observation_manager.py

def summary(self) -> str:
    msg = f"<ObservationManager> contains {len(self._group_obs_term_names)} groups.\n"
    for group_name, group_dim in self._group_obs_dim.items():
        title = f"Active Observation Terms in Group: '{group_name}'"
        if self._group_obs_options[group_name].concatenate_terms:
            title += f" (shape: {group_dim})"
        rows = []
        obs_terms = zip(
            self._group_obs_term_names[group_name],
            self._group_obs_term_dim[group_name],
            self._group_obs_terms[group_name],
            strict=False,
        )
        for index, (name, dims, term) in enumerate(obs_terms):
            if term._options.history_length > 0 and term._options.flatten_history_dim:
                # Flattened history: show (9,) ← 3×(3,)
                original_size = int(np.prod(dims)) // term._options.history_length
                original_shape = (original_size,) if len(dims) == 1 else dims[1:]
                shape_str = f"{dims}  ← {term._options.history_length}×{original_shape}"
            else:
                shape_str = str(tuple(dims))
            rows.append([index, name, shape_str])
        msg += format_table(title, ["Index", "Name", "Shape"], rows, align={"Name": "l"})
        msg += "\n"
    return msg

iter_terms ¶

iter_terms() -> Iterator[tuple[str, str, ObservationTermLike]]

Yield (group, name, term) for every active observation term, in group order.

Public accessor so callers (e.g. the per-term profiler) can scan term objects grouped, without reaching into the private per-group storage; mirrors :meth:~eden.managers.reward_manager.RewardManager.iter_terms.

Source code in eden/managers/observation_manager.py

def iter_terms(self) -> Iterator[tuple[str, str, ObservationTermLike]]:
    """Yield ``(group, name, term)`` for every active observation term, in group order.

    Public accessor so callers (e.g. the per-term profiler) can scan term objects
    grouped, without reaching into the private per-group storage; mirrors
    :meth:`~eden.managers.reward_manager.RewardManager.iter_terms`.
    """
    for group_name in self._group_obs_term_names:
        for name, term in zip(
            self._group_obs_term_names[group_name], self._group_obs_terms[group_name], strict=True
        ):
            yield group_name, name, term

get_term ¶

get_term(name: str) -> ObservationTermLike

Source code in eden/managers/observation_manager.py

def get_term(self, name: str) -> ObservationTermLike:
    # term_keys() (not keys()) so declared config fields aren't iterated as obs groups.
    for group_name in self._options.term_keys():
        for term_name, term in zip(
            self._group_obs_term_names[group_name],
            self._group_obs_terms[group_name],
            strict=True,
        ):
            if term_name == name:
                return term
    # Available options listed for the same reason ``Registry.get`` and
    # ``FlatManagerBase.get_term`` list theirs: the caller has the wrong name and no other way
    # to see the right one. Terms are grouped here, so the groups are named too.
    available = ", ".join(
        f"{group}: {', '.join(names) or 'none'}" for group, names in self.group_term_names.items()
    )
    raise ValueError(f"Term '{name}' not found in observation manager. Available options: {available or 'none'}")

set_term_scale ¶

set_term_scale(group_name: str, term_name: str, scale: float | list[float]) -> None

Retune one term's post-compute scale after the env is built.

scale is resolved once during env construction (see :func:_resolve_term_scale), so assigning term._options.scale afterwards changes what the options report but not what the manager multiplies by. This is the supported way to change it: the value goes through exactly the check the build-time resolution ran — feature dim, length, identity short-circuit — and both halves are written together, so a term can never report a scale it is not applying. A rejected value leaves both untouched, because the resolution runs first.

Addressed by (group_name, term_name) rather than a term handle because the same term name in two groups is two independent instances with their own scale, which :meth:get_term cannot distinguish.

Call this once, at setup. Scale is applied per frame before the frame enters a history buffer, so retuning a term whose window already holds frames would leave the older ones at the previous scale and the newer ones at the new one — one feature carrying two scales for history_length steps, which nothing in the config can show. That is refused. The window counts as empty before the first compute() and again after a full reset(), so the usual setup orders — retune before the first step, or after env.reset(), which is where the training wrappers leave you — are all allowed.

A term with no history holds no cross-step state in the observation pipeline, so it stays retunable at any point as far as this manager is concerned. That is not a licence to retune mid-training: the RL side holds state derived from the old scale (an observation normalizer's running statistics, most obviously) that nothing here can see or invalidate.

PARAMETER	DESCRIPTION
`group_name`	The observation group holding the term. TYPE: `str`
`term_name`	The term to retune, as named in the group. TYPE: `str`
`scale`	The new scale. A list is matched against the term's raw feature dim, as at build. TYPE: `float \| list[float]`

Source code in eden/managers/observation_manager.py

def set_term_scale(self, group_name: str, term_name: str, scale: float | list[float]) -> None:
    """Retune one term's post-compute ``scale`` after the env is built.

    ``scale`` is resolved once during env construction (see :func:`_resolve_term_scale`), so assigning
    ``term._options.scale`` afterwards changes what the options report but not what the manager
    multiplies by. This is the supported way to change it: the value goes through exactly the check the
    build-time resolution ran — feature dim, length, identity short-circuit — and both halves are written
    together, so a term can never report a scale it is not applying. A rejected value leaves both
    untouched, because the resolution runs first.

    Addressed by ``(group_name, term_name)`` rather than a term handle because the same term name in two
    groups is two independent instances with their own scale, which :meth:`get_term` cannot distinguish.

    **Call this once, at setup.** Scale is applied per frame *before* the frame enters a history buffer,
    so retuning a term whose window already holds frames would leave the older ones at the previous scale
    and the newer ones at the new one — one feature carrying two scales for ``history_length`` steps,
    which nothing in the config can show. That is refused. The window counts as empty before the first
    ``compute()`` and again after a full ``reset()``, so the usual setup orders — retune before the first
    step, or after ``env.reset()``, which is where the training wrappers leave you — are all allowed.

    A term with no history holds no cross-step state in the *observation* pipeline, so it stays retunable
    at any point as far as this manager is concerned. That is not a licence to retune mid-training: the
    RL side holds state derived from the old scale (an observation normalizer's running statistics, most
    obviously) that nothing here can see or invalidate.

    Parameters
    ----------
    group_name: str
        The observation group holding the term.
    term_name: str
        The term to retune, as named in the group.
    scale: float or list[float]
        The new scale. A list is matched against the term's raw feature dim, as at build.
    """
    if group_name not in self._group_obs_term_names:
        available = ", ".join(self._group_obs_term_names) or "none"
        raise ValueError(f"Observation group '{group_name}' not found. Available options: {available}")
    names = self._group_obs_term_names[group_name]
    if term_name not in names:
        available = ", ".join(names) or "none"
        raise ValueError(
            f"Term '{term_name}' not found in observation group '{group_name}'. Available options: {available}"
        )
    self._assert_history_not_started(group_name, term_name)
    index = names.index(term_name)
    term = self._group_obs_terms[group_name][index]
    resolved = _resolve_term_scale(
        term_name, group_name, scale, self._group_obs_term_raw_dim[group_name][index], self._env.device
    )
    term._resolved_scale = resolved
    # Assignment on an options model is not validated (``validate_assignment`` is off), so normalize to
    # the declared ``float | list[float]`` rather than letting a tuple survive into the options.
    term._options.scale = float(scale) if isinstance(scale, (int, float)) else [float(v) for v in scale]

reset ¶

reset(envs_idx: Tensor | slice | None = None) -> dict[str, float]

Source code in eden/managers/observation_manager.py

def reset(self, envs_idx: torch.Tensor | slice | None = None) -> dict[str, float]:
    history_batch_ids = (
        None if envs_idx is None else sanitize_envs_idx(envs_idx, self._env.num_envs, prefer_slice=False)
    )

    for group_name in self._group_obs_term_names:
        for _, term in self._group_obs_class_terms.get(group_name, ()):
            term.reset(envs_idx=envs_idx)

        for term in self._group_obs_terms[group_name]:
            if term._noise is not None:
                term._noise.reset(envs_idx=envs_idx)
            # Modifier state is per-step state like the history window, so it restarts with it, on
            # the same indices.
            for modifier in term._modifiers:
                modifier.reset(envs_idx=history_batch_ids)

        history = self._group_obs_term_history_buffer[group_name]
        if isinstance(history, dict):
            for buffer in history.values():
                buffer.reset(batch_ids=history_batch_ids)
        else:
            # Group-level history: one CircularBuffer for the concatenated group.
            history.reset(batch_ids=history_batch_ids)
    return {}

iter_post_reset_terms ¶

iter_post_reset_terms() -> Iterator[ObservationTermLike]

Every observation term, in group order (a term shared across groups repeats).

Source code in eden/managers/observation_manager.py

def iter_post_reset_terms(self) -> Iterator[ObservationTermLike]:
    """Every observation term, in group order (a term shared across groups repeats)."""
    return (term for _, _, term in self.iter_terms())

iter_post_physics_terms ¶

iter_post_physics_terms() -> Iterator[tuple[str, str, ObservationTerm]]

Yield (group, name, term) for every term :meth:post_physics_refresh ticks, in tick order.

The class terms (a term shared across groups repeats, as it is ticked once per group): what a CUDA graph of the graded phase (EnvOptions.cuda_graph_options) watches beside the graded managers' terms.

Source code in eden/managers/observation_manager.py

def iter_post_physics_terms(self) -> Iterator[tuple[str, str, ObservationTerm]]:
    """Yield ``(group, name, term)`` for every term :meth:`post_physics_refresh` ticks, in tick order.

    The class terms (a term shared across groups repeats, as it is ticked once per group): what a CUDA graph
    of the graded phase (``EnvOptions.cuda_graph_options``) watches beside the graded managers' terms.
    """
    for group_name, terms in self._group_obs_class_terms.items():
        for name, term in terms:
            yield group_name, name, term

post_physics_refresh ¶

post_physics_refresh() -> None

Advance every class term's per-step state on the post-physics scene — phase 7, once per control step.

The once-per-step tick that compute() is not (see :class:ObservationTerm). Called by the env's step after command_manager.post_physics_refresh() (5) and metric_manager.compute() (6), before the graded layer and above the auto-reset; and once per control tick from the deploy loop. This method owns that call-time contract; terms override the no-op :meth:ObservationTerm.post_physics_refresh.

Reaches class terms only: a function term is stateless by construction (:class:ObservationTermFuncWrapper). Every instance is ticked, including the duplicates the cross-group raw cache computes only once — that is what keeps their state identical and the cache sound.

Source code in eden/managers/observation_manager.py

def post_physics_refresh(self) -> None:
    """Advance every class term's per-step state on the post-physics scene — phase 7, once per control step.

    The once-per-step tick that ``compute()`` is not (see :class:`ObservationTerm`). Called by the
    env's step after ``command_manager.post_physics_refresh()`` (5) and ``metric_manager.compute()``
    (6), before the graded layer and above the auto-reset; and once per control tick from the deploy
    loop. This method owns that call-time contract; terms override the no-op
    :meth:`ObservationTerm.post_physics_refresh`.

    Reaches class terms only: a function term is stateless by construction
    (:class:`ObservationTermFuncWrapper`). Every instance is ticked, including the duplicates the
    cross-group raw cache computes only once — that is what keeps their state identical and the
    cache sound.
    """
    for terms in self._group_obs_class_terms.values():
        for _, term in terms:
            term.post_physics_refresh()

compute ¶

compute(update_history: bool = False) -> dict[str, Tensor | dict[str, Tensor]]

Source code in eden/managers/observation_manager.py

def compute(self, update_history: bool = False) -> dict[str, torch.Tensor | dict[str, torch.Tensor]]:
    # Reset the per-step cross-group cache so reused raw outputs reflect the current state. Cleared (not just
    # overwritten) to avoid pinning tensors.
    self._raw_cache.clear()
    obs_buffer: dict[str, torch.Tensor | dict[str, torch.Tensor]] = dict()
    # One pass for `compute_once` (ManagerBase): terms share an intermediate within it, across groups too.
    with self.compute_pass():
        for group_name in self._group_obs_term_names:
            obs_buffer[group_name] = self.compute_group(group_name, update_history)
    self._obs_buffer = obs_buffer
    return obs_buffer

compute_group ¶

compute_group(group_name: str, update_history: bool = False) -> Tensor | dict[str, Tensor]

Source code in eden/managers/observation_manager.py

def compute_group(self, group_name: str, update_history: bool = False) -> torch.Tensor | dict[str, torch.Tensor]:
    # The frames ``nan_policy`` "warn" / "error" still has to check, read back together once the
    # group's terms are computed: one host read per group rather than one per term.
    unchecked: list[tuple[str, torch.Tensor]] = []

    # Fast path: pre-allocated concatenation buffer (no history, 1-D obs, concatenate_terms).
    concat_buf = self._group_concat_buffer.get(group_name)
    if concat_buf is not None:
        slices = self._group_concat_slices[group_name]
        for i, (term_name, term) in enumerate(
            zip(self._group_obs_term_names[group_name], self._group_obs_terms[group_name], strict=True)
        ):
            start, end = slices[i]
            concat_buf[:, start:end].copy_(self._term_frame(group_name, term_name, term, update_history, unchecked))
        self._check_non_finite(group_name, unchecked)
        return concat_buf.clone()

    # General path
    group_obs: dict[str, torch.Tensor] = OrderedDict()

    for term_name, term in zip(
        self._group_obs_term_names[group_name],
        self._group_obs_terms[group_name],
        strict=True,
    ):
        obs = self._term_frame(group_name, term_name, term, update_history, unchecked)
        if self._group_obs_options[group_name].history_length is not None:
            group_obs[term_name] = obs
        elif term._options.history_length > 0:
            circular_buffer = self._group_obs_term_history_buffer[group_name][term_name]
            if update_history or not circular_buffer.is_initialized:
                circular_buffer.append(obs)
                buf = circular_buffer.buffer
            else:
                # Snapshot path: produce the buffer that *would* result from
                # appending ``obs`` now, without mutating the underlying state.
                # The canonical once-per-step advance happens elsewhere
                # (post-reset compute in ``RLEnvBase.step``); doing it here
                # too would skip a frame for non-done envs, while skipping
                # it entirely would drop the freshly-computed post-physics
                # frame from the snapshot.
                buf = circular_buffer.peek_buffer(obs)

            if term._options.flatten_history_dim:
                group_obs[term_name] = buf.reshape(self._env.num_envs, -1)
            else:
                group_obs[term_name] = buf
        else:
            group_obs[term_name] = obs

    self._check_non_finite(group_name, unchecked)

    if self._group_obs_options[group_name].history_length is not None:
        group_obs = torch.cat(
            list(group_obs.values()),
            dim=self._group_obs_options[group_name].concatenate_dim,
        )
        circular_buffer = self._group_obs_term_history_buffer[group_name]
        if update_history or not circular_buffer.is_initialized:
            circular_buffer.append(group_obs)
            buf = circular_buffer.buffer
        else:
            # Snapshot path: see term-level branch above.
            buf = circular_buffer.peek_buffer(group_obs)

        if self._group_obs_options[group_name].flatten_history_dim:
            return buf.reshape(self._env.num_envs, -1)
        else:
            return buf

    if self._group_obs_options[group_name].concatenate_terms:
        return torch.cat(
            list(group_obs.values()),
            dim=self._group_obs_options[group_name].concatenate_dim,
        )
    return group_obs

Functions:¶

RewardManager¶

Reward manager for computing reward signals.

Attributes¶

DEFAULT_REWARD_GROUP `module-attribute` ¶

DEFAULT_REWARD_GROUP = 'default'

RewardTermLike `module-attribute` ¶

RewardTermLike: TypeAlias = RewardTerm | RewardTermFuncWrapper

Union of a class-based reward term and its function-wrapper form.

REWARD_TERM_REGISTRY `module-attribute` ¶

REWARD_TERM_REGISTRY = TermRegistry('REWARD_TERM', RewardTerm, RewardTermFuncWrapper, populated_by=('eden.managers.terms.rewards',))

Classes¶

RewardTerm ¶

RewardTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[RewardTermOptions], _RewardTermFields

Base class for reward terms.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes¶

range_s `class-attribute` `instance-attribute` ¶

range_s: list[tuple[float, float]] | None = None

weight `class-attribute` `instance-attribute` ¶

weight: float = 1.0

tags `class-attribute` `instance-attribute` ¶

tags: list[str] = []

is_temporal_dependent `property` ¶

is_temporal_dependent: bool

Whether the term is active only within range_s.

RewardTermFuncWrapper ¶

RewardTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermFuncWrapper[RewardTermOptions], _RewardTermFields

Base class for reward terms defined as a function.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

Attributes¶

range_s `class-attribute` `instance-attribute` ¶

range_s: list[tuple[float, float]] | None = None

weight `class-attribute` `instance-attribute` ¶

weight: float = 1.0

tags `class-attribute` `instance-attribute` ¶

tags: list[str] = []

is_temporal_dependent `property` ¶

is_temporal_dependent: bool

Whether the term is active only within range_s.

RewardManager ¶

RewardManager(env: RLEnvBase, options: RewardManagerOptions)

Bases: FlatManagerBase[RewardManagerOptions]

Computes the scalar step reward as the weighted sum of all active terms.

Terms are declared flat (:class:RewardTermOptions) or inside named groups (:class:RewardGroupOptions); flat terms implicitly belong to :data:DEFAULT_REWARD_GROUP. Grouping never changes the scalar reward the environment returns — :meth:compute always sums over all groups — it adds a per-group readout (:meth:get_group_reward, :meth:get_group_episode_sum) for multi-channel consumers (CMDP-style task/style/penalty critics, per-group logging). Groups partition the active terms, so summing :meth:get_group_reward over :attr:group_term_names reconstructs the value :meth:compute returned for the same step (up to floating-point summation order).

Unlike observation groups (which may share a term across groups), reward groups are a strict partition of one flat term set — duplicate term names across groups raise at construction. That is why this stays a :class:FlatManagerBase: the terms, buffers, and the flat API (terms, active_terms, get_term, iter_terms, get_episode_sum) are unchanged; groups are a labeling on top.

Source code in eden/managers/reward_manager.py

def __init__(self, env: RLEnvBase, options: RewardManagerOptions):
    super().__init__(env=env, options=options)
    num_terms = len(self._term_names)

    # Whether reset() emits the per-term reward log keys. None auto-selects:
    # off when explicit groups give a group-sum readout, on otherwise (a flat config has
    # nothing to summarize to). _has_explicit_groups is set by _prepare_terms above.
    log_pref = self._options.log_per_term_rewards
    self._log_per_term = (not self._has_explicit_groups) if log_pref is None else bool(log_pref)
    if not self._log_per_term and log_pref is None:
        # Declaring the first reward group silently flips this default, so every
        # ``Episode_Reward/<term>`` key a flat config was logging disappears. Say so once at
        # build rather than leaving it to be noticed as a gap in wandb.
        en.logger.info(
            f"RewardManager: per-term reward log keys (Episode_Reward/<term>) are off because this config "
            f"declares reward groups; logging Episode_Reward_Group/<group> for "
            f"{list(self._group_term_names)} instead. Set `log_per_term_rewards=True` to keep both."
        )

    # 2D episode sums for vectorized accumulation.
    # ``_episode_sums`` is the weighted+masked accumulation; ``_episode_raw_sums``
    # mirrors the raw (unweighted) term outputs for diagnostic logging.
    self._episode_sums = torch.zeros((self.num_envs, num_terms), dtype=torch.float, device=self.device)
    self._episode_raw_sums = torch.zeros((self.num_envs, num_terms), dtype=torch.float, device=self.device)
    self._reward_buf = torch.zeros(self.num_envs, dtype=torch.float, device=self.device)
    self._step_reward = torch.zeros((self.num_envs, num_terms), dtype=torch.float, device=self.device)
    # Unweighted (but temporally masked) term outputs of the last step: compute() fills it and
    # multiplies it into _step_reward; get_term_value() reads it.
    self._step_raw_reward = torch.zeros((self.num_envs, num_terms), dtype=torch.float, device=self.device)

    # Per-group column indices into the flat term buffers.
    self._group_col_idx: dict[str, torch.Tensor] = {
        group: torch.tensor([self._term_name_to_idx[name] for name in names], dtype=torch.long, device=self.device)
        for group, names in self._group_term_names.items()
    }
    # dt of the last compute() call; 0.0 until then so the group readout is
    # zeros before the first step, matching the zero-filled buffers.
    self._last_dt = 0.0
    # Constant zero scalar: sum-start for the per-group reset logs so empty groups
    # stay tensor-typed without a per-reset allocation. Read-only by contract —
    # it is returned as-is as the log value of an empty group.
    self._log_zero = torch.zeros((), dtype=torch.float, device=self.device)

    # Pre-compute weights vector for vectorized multiply. ``_weights_host`` is the same list on
    # the host: compute() compares against it and uploads only on a change (a curriculum moving
    # a weight), since writing a Python float into a CUDA tensor is a GPU->CPU sync.
    self._weights_host = [term.weight for term in self._terms.values()]
    self._weights = torch.tensor(self._weights_host, dtype=torch.float, device=self.device)

    # Pre-allocate temporal mask (1.0 for non-temporal terms, updated per-step for temporal).
    # ``range_s`` is in seconds; convert to integer step bounds once here so the per-step
    # compare uses ``episode_length_buf`` directly (rounding keeps both ends inclusive).
    self._temporal_mask = torch.ones((self.num_envs, num_terms), dtype=torch.float, device=self.device)
    self._temporal_term_info: list[tuple[int, list[tuple[int, int]]]] = []
    for term_idx, term in enumerate(self._terms.values()):
        if term.is_temporal_dependent:
            step_range = [(round(lo / self._env.dt), round(hi / self._env.dt)) for lo, hi in term.range_s]
            self._temporal_term_info.append((term_idx, step_range))

Attributes¶

device_synced_fields `class-attribute` ¶

device_synced_fields: frozenset[str] = frozenset({'weight'})

group_term_names `property` ¶

group_term_names: dict[str, list[str]]

The active term names keyed by group, in declaration order (a copy, safe to mutate).

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/reward_manager.py

def summary(self) -> str:
    if self._has_explicit_groups:
        group_of = {name: group for group, names in self._group_term_names.items() for name in names}
        return self._format_summary_table(
            title="Active Reward Terms",
            field_names=["Index", "Name", "Group", "Weight"],
            rows=(
                [index, name, group_of[name], term.weight] for index, (name, term) in enumerate(self._terms.items())
            ),
            align={"Name": "l", "Group": "l", "Weight": "r"},
        )
    return self._format_summary_table(
        title="Active Reward Terms",
        field_names=["Index", "Name", "Weight"],
        rows=([index, name, term.weight] for index, (name, term) in enumerate(self._terms.items())),
        align={"Name": "l", "Weight": "r"},
    )

iter_terms ¶

iter_terms()

Yield (name, term) pairs for every active reward term, flat across all groups.

Public accessor for sibling managers (e.g. curricula) that need to scan terms without reaching into the private _terms dict.

Source code in eden/managers/reward_manager.py

def iter_terms(self):
    """Yield ``(name, term)`` pairs for every active reward term, flat across all groups.

    Public accessor for sibling managers (e.g. curricula) that need to scan terms
    without reaching into the private ``_terms`` dict.
    """
    return self._terms.items()

get_episode_sum ¶

get_episode_sum(name: str, raw: bool = True) -> Tensor

Return the per-env running reward sum for term name as a (num_envs,) view.

raw=True (default) returns the unweighted, mask-applied accumulation; raw=False returns the weighted post-curriculum sum. Both buffers are reset to zero per env in :meth:reset. Raises KeyError (with the active term list) if name is not registered. See :meth:iter_terms for full-buffer scanning.

Source code in eden/managers/reward_manager.py

def get_episode_sum(self, name: str, raw: bool = True) -> torch.Tensor:
    """Return the per-env running reward sum for term ``name`` as a ``(num_envs,)`` view.

    ``raw=True`` (default) returns the unweighted, mask-applied accumulation;
    ``raw=False`` returns the weighted post-curriculum sum. Both buffers are
    reset to zero per env in :meth:`reset`. Raises ``KeyError`` (with the
    active term list) if ``name`` is not registered. See :meth:`iter_terms`
    for full-buffer scanning.
    """
    buf = self._episode_raw_sums if raw else self._episode_sums
    return buf[:, self._term_col(name)]

get_term_value ¶

get_term_value(name: str) -> Tensor

Return term name's output at the last computed step as a (num_envs,) view.

The value is the term's compute() output, temporally masked and zero for an env Genesis flagged as corrupt, with no weight and no dt — what an off-policy learner that weights its own reward channels stores per transition (e.g. FB-CPR's auxiliary rewards). The weighted, dt-scaled share of the scalar reward is :meth:get_group_reward's. Like MetricManager.get_value, the view is the manager's buffer: :meth:compute overwrites it in place, so treat it as read-only and clone it to keep a step. It reads zeros before the first :meth:compute and still reads the pre-reset step after an auto-reset. Raises KeyError (with the active term list) if name is not registered.

Source code in eden/managers/reward_manager.py

def get_term_value(self, name: str) -> torch.Tensor:
    """Return term ``name``'s output at the last computed step as a ``(num_envs,)`` view.

    The value is the term's ``compute()`` output, temporally masked and zero for an env Genesis flagged as
    corrupt, with no weight and no ``dt`` — what an off-policy learner that weights its own reward channels
    stores per transition (e.g. FB-CPR's auxiliary rewards). The weighted, ``dt``-scaled share of the scalar
    reward is :meth:`get_group_reward`'s. Like ``MetricManager.get_value``, the view is the manager's
    buffer: :meth:`compute` overwrites it in place, so treat it as read-only and clone it to keep a step. It
    reads zeros before the first :meth:`compute` and still reads the pre-reset step after an auto-reset.
    Raises ``KeyError`` (with the active term list) if ``name`` is not registered.
    """
    return self._step_raw_reward[:, self._term_col(name)]

get_group_reward ¶

get_group_reward(group: str) -> Tensor

Return the reward of group group for the last computed step as a (num_envs,) tensor.

The value is weighted, temporally masked, and dt-scaled — the group's contribution to the scalar reward :meth:compute returned, so summing this over all groups reconstructs that scalar (up to floating-point summation order). Valid until the next :meth:compute call (it reads the per-term step buffer that call refills); before the first compute it returns zeros, and after :meth:reset it still reads the pre-reset step — consistent with the scalar :meth:compute returned for that step (auto-reset runs after reward). Raises KeyError (with the active group list) for an unknown group.

Source code in eden/managers/reward_manager.py

def get_group_reward(self, group: str) -> torch.Tensor:
    """Return the reward of group ``group`` for the last computed step as a ``(num_envs,)`` tensor.

    The value is weighted, temporally masked, and dt-scaled — the group's
    contribution to the scalar reward :meth:`compute` returned, so summing
    this over all groups reconstructs that scalar (up to floating-point
    summation order). Valid until the next :meth:`compute` call (it reads
    the per-term step buffer that call refills); before the first compute
    it returns zeros, and after :meth:`reset` it still reads the pre-reset
    step — consistent with the scalar :meth:`compute` returned for that
    step (auto-reset runs after reward). Raises ``KeyError`` (with the
    active group list) for an unknown group.
    """
    return self._step_reward.index_select(1, self._group_cols(group)).sum(dim=1) * self._last_dt

get_group_episode_sum ¶

get_group_episode_sum(group: str, *, raw: bool) -> Tensor

Return the per-env running reward sum of group group as a (num_envs,) tensor.

raw=False sums the weighted post-curriculum accumulation — the quantity whose per-group parts add up to the episode's scalar reward (e.g. the CMDP task-return signal); raw=True sums the unweighted, mask-applied accumulation for diagnostics. raw is deliberately keyword-only with no default: the per-term :meth:get_episode_sum defaults to the raw view, so a silent (opposite) default here would flip semantics on consumers porting per-term code. Raises KeyError (with the active group list) for an unknown group.

Source code in eden/managers/reward_manager.py

def get_group_episode_sum(self, group: str, *, raw: bool) -> torch.Tensor:
    """Return the per-env running reward sum of group ``group`` as a ``(num_envs,)`` tensor.

    ``raw=False`` sums the weighted post-curriculum accumulation — the
    quantity whose per-group parts add up to the episode's scalar reward
    (e.g. the CMDP task-return signal); ``raw=True`` sums the unweighted,
    mask-applied accumulation for diagnostics. ``raw`` is deliberately
    keyword-only with no default: the per-term :meth:`get_episode_sum`
    defaults to the raw view, so a silent (opposite) default here would
    flip semantics on consumers porting per-term code. Raises ``KeyError``
    (with the active group list) for an unknown group.
    """
    buf = self._episode_raw_sums if raw else self._episode_sums
    return buf.index_select(1, self._group_cols(group)).sum(dim=1)

is_timestep_in_range `staticmethod` ¶

is_timestep_in_range(timestep: Tensor, ranges: list[tuple[float, float]]) -> Tensor

Check whether a batch of timesteps falls within any of the given intervals.

PARAMETER	DESCRIPTION
`timestep`	`(batch_size,)` tensor of timesteps. TYPE: `Tensor`
`ranges`	List of `(lower, upper)` tuples defining valid intervals, in the same units as `timestep` (step counts, not seconds). TYPE: `list of tuple of float`

RETURNS	DESCRIPTION
`Tensor`	`(batch_size,)` boolean tensor indicating membership.

Source code in eden/managers/reward_manager.py

@staticmethod
def is_timestep_in_range(timestep: torch.Tensor, ranges: list[tuple[float, float]]) -> torch.Tensor:
    """Check whether a batch of timesteps falls within any of the given intervals.

    Parameters
    ----------
    timestep : torch.Tensor
        ``(batch_size,)`` tensor of timesteps.
    ranges : list of tuple of float
        List of ``(lower, upper)`` tuples defining valid intervals, in the same
        units as ``timestep`` (step counts, not seconds).

    Returns
    -------
    torch.Tensor
        ``(batch_size,)`` boolean tensor indicating membership.
    """
    is_valid = torch.zeros_like(timestep, dtype=torch.bool, device=timestep.device)
    for lower, upper in ranges:
        is_valid |= (timestep >= lower) & (timestep <= upper)
    return is_valid

reset ¶

reset(envs_idx: Tensor | None = None) -> dict[str, Tensor]

Source code in eden/managers/reward_manager.py

def reset(self, envs_idx: torch.Tensor | None = None) -> dict[str, torch.Tensor]:
    if envs_idx is None:
        envs_idx = slice(None)
    extras = {}
    # Per-term episode averages are computed for every term (the group readout below is
    # derived from them), but only surfaced as log keys when _log_per_term is set —
    # otherwise a term-heavy config's per-iteration log is dominated by them.
    # One reduction over every term's column rather than a gather per term: on the auto-reset
    # path ``envs_idx`` is the bool ``reset_buf``, and each gather with it is a GPU->CPU sync.
    max_episode_length_s = self._env.max_episode_length_s
    episode_means = mean_over_envs(self._episode_sums, envs_idx) / max_episode_length_s
    per_term: dict[str, torch.Tensor] = dict(zip(self._term_names, episode_means.unbind()))
    if self._log_per_term:
        raw_episode_means = mean_over_envs(self._episode_raw_sums, envs_idx) / max_episode_length_s
        for key, raw_mean in zip(self._term_names, raw_episode_means.unbind()):
            extras["Episode_Reward/" + key] = per_term[key]
            extras["Episode_Reward_Raw/" + key] = raw_mean
    # Always cleared, even when not logged — the accumulators must reset every episode.
    self._episode_sums[envs_idx] = 0.0
    self._episode_raw_sums[envs_idx] = 0.0
    # Per-group episode averages, derived from the per-term scalars above: groups
    # partition the terms, so a group's average is exactly the sum of its members' —
    # no extra buffer indexing on the auto-reset hot path, and one normalization
    # convention instead of two. Only when groups are declared — ungrouped envs
    # log no group keys. Group sums are emitted regardless of
    # _log_per_term (they are the terse default readout, not per-term detail).
    if self._has_explicit_groups:
        for group, names in self._group_term_names.items():
            extras["Episode_Reward_Group/" + group] = sum((per_term[name] for name in names), self._log_zero)
    for terms in self._reset_terms:
        terms.reset(envs_idx=envs_idx)
    return extras

compute ¶

compute(dt: float) -> Tensor

Source code in eden/managers/reward_manager.py

@torch.inference_mode()
def compute(self, dt: float) -> torch.Tensor:
    # _step_raw_reward is filled and masked in-place through phases 1-2a; phase 2c writes its
    # weighted copy into _step_reward. After compute() returns, _step_raw_reward holds
    # compute() * mask and _step_reward holds compute() * mask * weight.
    self._last_dt = dt

    # Phase 1: Fill raw term outputs and sync weights (curriculum may modify them).
    # _step_raw_reward[:, i] = term_i.compute()
    # One pass for `compute_once` (ManagerBase): terms share an intermediate within it.
    with self.compute_pass():
        for term_idx, term in enumerate(self._terms.values()):
            self._step_raw_reward[:, term_idx] = term.compute()
    # An env Genesis flagged as corrupt this step is reset in the same step, but its terms just read that state
    # (a NaN contact force, say): the transition carries no signal, and a NaN would end the run in the learner's
    # NaN check and poison the episode sums. ``masked_fill_``, not a multiply: NaN * 0 is NaN.
    self._step_raw_reward.masked_fill_(self._env.get_solver_error_envs_mask()[:, None], 0.0)
    self.sync_device_fields()

    # Phase 2a: Apply temporal mask in-place. Raw episode sums respect this
    # mask too — accumulating before masking would log term outputs from
    # timesteps where range_s makes the term inactive.
    if self._temporal_term_info:
        for idx, step_range in self._temporal_term_info:
            self._temporal_mask[:, idx] = self.is_timestep_in_range(
                self._env.episode_length_buf, step_range
            ).float()
        # _step_raw_reward *= mask  →  now holds compute() * mask
        self._step_raw_reward.mul_(self._temporal_mask)

    # Phase 2b: Accumulate raw episode sums (compute() * mask, no weights).
    self._episode_raw_sums.add_(self._step_raw_reward, alpha=dt)

    # Phase 2c: Apply per-term weights into the weighted buffer; the raw one is kept for
    # get_term_value(). An out-of-place multiply, so keeping it costs no extra copy.
    # _step_reward = _step_raw_reward * weights  →  compute() * mask * weight
    torch.mul(self._step_raw_reward, self._weights, out=self._step_reward)

    # Phase 3: Reduce to scalar reward and accumulate episode sums.
    torch.sum(self._step_reward, dim=1, out=self._reward_buf)
    self._reward_buf.mul_(dt)
    self._episode_sums.add_(self._step_reward, alpha=dt)
    return self._reward_buf

sync_device_fields ¶

sync_device_fields() -> None

Copy the terms' current weight onto the device, when one changed since the last call.

A curriculum may move a term's weight between steps. compute() calls this; the CUDA graph of the graded phase (EnvOptions.cuda_graph_options) calls it before each replay instead, since a host-side comparison is exactly what a replay skips.

Source code in eden/managers/reward_manager.py

def sync_device_fields(self) -> None:
    """Copy the terms' current ``weight`` onto the device, when one changed since the last call.

    A curriculum may move a term's weight between steps. ``compute()`` calls this; the CUDA graph of the
    graded phase (``EnvOptions.cuda_graph_options``) calls it before each replay instead, since a host-side
    comparison is exactly what a replay skips.
    """
    weights = [term.weight for term in self._terms.values()]
    if weights != self._weights_host:
        self._weights.copy_(torch.tensor(weights, dtype=torch.float))
        self._weights_host = weights

Functions:¶

TerminationManager¶

Termination manager for computing done signals.

Attributes¶

TerminationTermLike `module-attribute` ¶

TerminationTermLike: TypeAlias = TerminationTerm | TerminationTermFuncWrapper

Union of a class-based termination term and its function-wrapper form.

TERMINATION_TERM_REGISTRY `module-attribute` ¶

TERMINATION_TERM_REGISTRY = TermRegistry('TERMINATION_TERM', TerminationTerm, TerminationTermFuncWrapper, populated_by=('eden.managers.terms.terminations', 'eden.managers.terms.metrics'))

SOLVER_ERROR_KEY `module-attribute` ¶

SOLVER_ERROR_KEY = 'solver_error'

Classes¶

TerminationTerm ¶

TerminationTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[TerminationTermOptions], _TerminationTermFields

Base class for termination terms.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes¶

time_out `class-attribute` `instance-attribute` ¶

time_out: bool = False

TerminationTermFuncWrapper ¶

TerminationTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermFuncWrapper[TerminationTermOptions], _TerminationTermFields

Base class for termination terms defined as a function.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

Attributes¶

time_out `class-attribute` `instance-attribute` ¶

time_out: bool = False

TerminationManager ¶

TerminationManager(env: EnvBase, options: TerminationManagerOptions)

Bases: FlatManagerBase[TerminationManagerOptions]

Source code in eden/managers/termination_manager.py

def __init__(self, env: EnvBase, options: TerminationManagerOptions):
    super().__init__(env=env, options=options)

    self._term_dones: dict[str, torch.Tensor] = {}
    for term_name in self._term_names:
        self._term_dones[term_name] = torch.zeros(self.num_envs, device=self.device, dtype=torch.bool)
    self._truncated_buf = torch.zeros(self.num_envs, device=self.device, dtype=torch.bool)
    self._terminated_buf = torch.zeros_like(self._truncated_buf)
    self._solver_error_buf = torch.zeros_like(self._truncated_buf)

Attributes¶

dones `property` ¶

dones: Tensor

timeouts `property` ¶

timeouts: Tensor

terminated `property` ¶

terminated: Tensor

solver_errors `property` ¶

solver_errors: Tensor

Per-env mask of the envs Genesis flagged on the last :meth:compute (see EnvBase.get_solver_error_envs_mask, which also says how a misconfiguration of the scene still raises).

They are reset as truncated (:attr:timeouts), never terminated, and logged only as Episode_Termination/solver_error: the simulator failed, not the task, so the learner bootstraps the step instead of reading it as a terminal outcome, and no term's verdict on the corrupt state counts.

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/termination_manager.py

def summary(self) -> str:
    return self._format_summary_table(
        title="Active Termination Terms",
        field_names=["Index", "Name", "Time Out"],
        rows=([index, name, term.time_out] for index, (name, term) in enumerate(self._terms.items())),
        align={"Name": "l"},
    )

reset ¶

reset(envs_idx: slice | Tensor | None = None) -> dict[str, Tensor]

Source code in eden/managers/termination_manager.py

def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> dict[str, torch.Tensor]:
    if envs_idx is None:
        envs_idx = slice(None)
    extras = {}
    # Kept on the device: a count read back with ``.item()`` is a GPU->CPU sync per term on every
    # auto-reset step, and the loggers take a 0-dim tensor as readily as an int.
    for key, dones in self._term_dones.items():
        extras["Episode_Termination/" + key] = sum_over_envs(dones, envs_idx)
    extras["Episode_Termination/" + SOLVER_ERROR_KEY] = sum_over_envs(self._solver_error_buf, envs_idx)
    for term in self._reset_terms:
        term.reset(envs_idx=envs_idx)
    return extras

compute ¶

compute() -> Tensor

Source code in eden/managers/termination_manager.py

def compute(self) -> torch.Tensor:
    self._truncated_buf[:] = False
    self._terminated_buf[:] = False
    # NOTE: reset corrupted state (call set_qpos to reset), per
    # https://github.com/Genesis-Embodied-AI/Genesis/issues/2239; truncated, not terminated, and without
    # any term's verdict (see `solver_errors`).
    self._solver_error_buf[:] = self._env.get_solver_error_envs_mask()
    healthy = ~self._solver_error_buf
    # One pass for `compute_once` (ManagerBase): terms share an intermediate within it.
    with self.compute_pass():
        for name, term in self._terms.items():
            dones = torch.logical_and(term.compute(), healthy, out=self._term_dones[name])  # written in place
            if term.time_out:
                self._truncated_buf |= dones
            else:
                self._terminated_buf |= dones
    self._truncated_buf |= self._solver_error_buf
    return self._truncated_buf | self._terminated_buf

get_term_dones ¶

get_term_dones(name: str) -> Tensor

Return the per-env done mask most recently computed for term name.

The term object itself is available via the inherited :meth:get_term.

Source code in eden/managers/termination_manager.py

def get_term_dones(self, name: str) -> torch.Tensor:
    """Return the per-env done mask most recently computed for term ``name``.

    The term object itself is available via the inherited :meth:`get_term`.
    """
    return self._term_dones[name]

Functions:¶

CommandManager¶

Command manager for generating and updating commands.

Attributes¶

COMMAND_TERM_REGISTRY `module-attribute` ¶

COMMAND_TERM_REGISTRY = TermRegistry('COMMAND_TERM', CommandTerm, populated_by=('eden.managers.terms.commands',))

Classes¶

CommandTerm ¶

CommandTerm(env: EnvBase, options: CommandTermOptions)

Bases: ConfigurableTermBase[CommandTermOptions]

Base class for command terms.

PARAMETER	DESCRIPTION
`resampling_time_range`	The range of time before commands are changed [s].
`debug_vis`	Whether to visualize debug information. Defaults to False.

Source code in eden/managers/command_manager.py

def __init__(self, env: EnvBase, options: CommandTermOptions):
    super().__init__(env=env, options=options)

    self.stats = dict()
    self.time_left = torch.zeros(self.num_envs, device=self.device)
    self.command_counter = torch.zeros(self.num_envs, device=self.device, dtype=torch.long)

Attributes¶

resampling_time_range `class-attribute` `instance-attribute` ¶

resampling_time_range: tuple[float, float] = None

debug_vis `class-attribute` `instance-attribute` ¶

debug_vis: bool = False

stats `instance-attribute` ¶

stats = dict()

time_left `instance-attribute` ¶

time_left = torch.zeros(self.num_envs, device=self.device)

command_counter `instance-attribute` ¶

command_counter = torch.zeros(self.num_envs, device=self.device, dtype=torch.long)

command `abstractmethod` `property` ¶

command

Methods:¶

reset ¶

reset(envs_idx: slice | Tensor | None = None) -> dict[str, Tensor]

Source code in eden/managers/command_manager.py

def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> dict[str, torch.Tensor]:
    envs_idx = sanitize_envs_idx(envs_idx, self.num_envs)
    extras = {}
    for stat_name, stat_value in self.stats.items():
        # A 0-dim tensor reduced on the device: gathering through the auto-reset's mask and `.item()`-ing
        # the mean were two GPU->CPU syncs per stat per step. The scalar fill stays sync-free.
        extras[stat_name] = mean_over_envs(stat_value, envs_idx)
        stat_value[envs_idx] = 0.0
    self.command_counter[envs_idx] = 0
    self._resample(envs_idx)
    return extras

compute ¶

compute(dt: float) -> None

Tick the resampling clock, then resample and publish (phase 17 — the preparation epoch).

In :meth:RLEnvBase.step this runs after the auto-reset and below the reward, so what it produces is the command the next action is chosen from and phase 19's observations carry — not the command this step was graded against. :meth:_update_command says what belongs here and what does not.

On the deployment path it is one of two command hooks that run: DeploymentBase.step drives :meth:advance_reference first — so a time-advancing reference keeps moving on hardware — then this method and :meth:draw_vis. :meth:post_physics_refresh does not run there at all. A deployed term must produce whatever the policy's observations read from here or from the advance — see docs/guides/terms.md.

Source code in eden/managers/command_manager.py

def compute(self, dt: float) -> None:
    """Tick the resampling clock, then resample and publish (phase 17 — the *preparation* epoch).

    In :meth:`RLEnvBase.step` this runs after the auto-reset and below the reward, so what it
    produces is the command the next action is chosen from and phase 19's observations carry —
    not the command this step was graded against. :meth:`_update_command` says what belongs
    here and what does not.

    On the deployment path it is one of **two** command hooks that run: ``DeploymentBase.step``
    drives :meth:`advance_reference` first — so a time-advancing reference keeps moving on
    hardware — then this method and :meth:`draw_vis`. :meth:`post_physics_refresh` does not run
    there at all. A deployed term must produce whatever the policy's observations read from here
    or from the advance — see ``docs/guides/terms.md``.
    """
    self.time_left -= dt
    resample_envs_mask = self.time_left <= 0.0
    # NOTE: no `.any()` gate here. `_resample` already counts the mask — its one GPU→CPU
    # sync — and returns early on an empty one, so a gate would only pay that sync twice.
    self._resample(resample_envs_mask)
    self._update_command()

draw_vis ¶

draw_vis() -> None

Source code in eden/managers/command_manager.py

def draw_vis(self) -> None:
    pass

post_physics_refresh ¶

post_physics_refresh() -> None

Refresh state derived from the post-physics robot pose (phase 5 — the grading epoch).

No-op by default; override to refresh caches. This runs after the physics sub-steps and before metric/termination/reward, which makes it the hook for everything that grades this step: pose and contact caches, progress counters, hold latches, tracking errors. A value written here is read by phase 9's reward in the same step — no lag, and nothing to snapshot across the auto-reset. Writing it in :meth:_update_command instead is the bug class docs/guides/terms.md documents under "Which hook: grade the step, or prepare the next one".

Invoked by :meth:CommandManager.post_physics_refresh, which owns the call-time contract — which means it is not invoked on the deployment path: DeploymentBase.step drives :meth:advance_reference and :meth:compute, and nothing between them, because on hardware there are no physics sub-steps to bracket and no reward to grade. State a deployed term's observations read has to be produced from :meth:compute too; an idempotent cache refresh can simply live in both, the way :class:~eden.managers.terms.commands.mimic.MotionCommand refreshes its robot-state cache.

Source code in eden/managers/command_manager.py

def post_physics_refresh(self) -> None:
    """Refresh state derived from the post-physics robot pose (phase 5 — the *grading* epoch).

    No-op by default; override to refresh caches. This runs after the physics sub-steps and
    before metric/termination/reward, which makes it the hook for **everything that grades
    this step**: pose and contact caches, progress counters, hold latches, tracking errors.
    A value written here is read by phase 9's reward in the same step — no lag, and nothing
    to snapshot across the auto-reset. Writing it in :meth:`_update_command` instead is the
    bug class ``docs/guides/terms.md`` documents under "Which hook: grade the step, or
    prepare the next one".

    Invoked by :meth:`CommandManager.post_physics_refresh`, which owns the call-time contract —
    which means it is **not** invoked on the deployment path: ``DeploymentBase.step`` drives
    :meth:`advance_reference` and :meth:`compute`, and nothing between them, because on
    hardware there are no physics sub-steps to bracket and no reward to grade. State a deployed
    term's observations read has to be produced from :meth:`compute` too; an idempotent cache
    refresh can simply live in both, the way
    :class:`~eden.managers.terms.commands.mimic.MotionCommand` refreshes its robot-state cache.
    """

pre_physics_advance ¶

pre_physics_advance() -> None

Advance the command BEFORE the physics sub-steps.

No-op by default. Overridden by terms that track a time-advancing reference and must advance it before physics so the post-physics robot is graded against the temporally-aligned frame ("advance-then-grade"; e.g. :class:~eden.managers.terms.commands.mimic.MotionCommand). Invoked once per control step by :meth:CommandManager.pre_physics_advance in :meth:RLEnvBase.step, before the decimation loop.

The first of the term's three per-step hooks; the other two are :meth:post_physics_refresh (grade the step that just happened) and :meth:_update_command (publish the command the next step is chosen against). Unlike :meth:post_physics_refresh, this one also runs on deployment through :meth:advance_reference by default. A term whose simulation advance resets the robot must override that deployment hook with a clock-only advance.

Source code in eden/managers/command_manager.py

def pre_physics_advance(self) -> None:
    """Advance the command BEFORE the physics sub-steps.

    No-op by default. Overridden by terms that track a time-advancing reference
    and must advance it *before* physics so the post-physics robot is graded
    against the temporally-aligned frame ("advance-then-grade"; e.g.
    :class:`~eden.managers.terms.commands.mimic.MotionCommand`). Invoked once
    per control step by :meth:`CommandManager.pre_physics_advance` in
    :meth:`RLEnvBase.step`, before the decimation loop.

    The first of the term's three per-step hooks; the other two are
    :meth:`post_physics_refresh` (grade the step that just happened) and
    :meth:`_update_command` (publish the command the next step is chosen against). Unlike
    :meth:`post_physics_refresh`, this one also runs on deployment through
    :meth:`advance_reference` by default. A term whose simulation advance resets the robot
    must override that deployment hook with a clock-only advance.
    """

advance_reference ¶

advance_reference() -> None

Advance a deployed reference without resetting the robot.

Preserve the existing advance hook by default. Terms that resample or teleport at clip boundaries in :meth:pre_physics_advance must provide a clock-only version here.

Source code in eden/managers/command_manager.py

def advance_reference(self) -> None:
    """Advance a deployed reference without resetting the robot.

    Preserve the existing advance hook by default. Terms that resample or teleport at
    clip boundaries in :meth:`pre_physics_advance` must provide a clock-only version here.
    """
    self.pre_physics_advance()

rewind_reference ¶

rewind_reference() -> None

Restart a time-advancing reference at its first frame, for every env (no-op by default).

The reset-side twin of :meth:advance_reference, and it exists for the deployment path alone: DeploymentBase.reset_motion_reference drives it at deploy reset, at the operator's START and on every policy switch, so a clip hands over from its beginning the way the manager path already does through MotionManager.rewind_to_start. Without it a term that advances a clock resumes wherever the clock happened to stop — mid-clip, at a pose the robot standing in its INIT stance is nowhere near.

Rewind the clock only. This runs on hardware, where there is no robot to place and no episode to randomize, so it must not reach for the reset-state initialization a wrap in :meth:pre_physics_advance performs in sim (MotionCommand._resample_command teleports the robot and perturbs its pose). Clip selection is not resampled either: a deploy replays the same clip deterministically.

Source code in eden/managers/command_manager.py

def rewind_reference(self) -> None:
    """Restart a time-advancing reference at its first frame, for every env (no-op by default).

    The reset-side twin of :meth:`advance_reference`, and it exists for the deployment
    path alone: ``DeploymentBase.reset_motion_reference`` drives it at deploy reset, at the
    operator's START and on every policy switch, so a clip hands over from its beginning the
    way the manager path already does through ``MotionManager.rewind_to_start``. Without it
    a term that advances a clock resumes wherever the clock happened to stop — mid-clip, at a
    pose the robot standing in its INIT stance is nowhere near.

    Rewind the **clock only**. This runs on hardware, where there is no robot to place and no
    episode to randomize, so it must not reach for the reset-state initialization a wrap in
    :meth:`pre_physics_advance` performs in sim (``MotionCommand._resample_command`` teleports
    the robot and perturbs its pose). Clip *selection* is not resampled either: a deploy
    replays the same clip deterministically.
    """

CommandManager ¶

CommandManager(env: EnvBase, options: CommandManagerOptions)

Bases: FlatManagerBase[CommandManagerOptions]

Source code in eden/managers/command_manager.py

def __init__(self, env: EnvBase, options: CommandManagerOptions):
    super().__init__(env=env, options=options)

    self._commands = dict()

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/command_manager.py

def summary(self) -> str:
    return self._format_summary_table(
        title="Active Command Terms",
        field_names=["Index", "Name", "Type"],
        rows=([index, name, term.__class__.__name__] for index, (name, term) in enumerate(self._terms.items())),
        align={"Name": "l"},
    )

draw_vis ¶

draw_vis() -> None

Source code in eden/managers/command_manager.py

def draw_vis(self) -> None:
    for term in self._terms.values():
        if term.debug_vis:
            term.draw_vis()

post_physics_refresh ¶

post_physics_refresh() -> None

Refresh every term's robot-state cache on the just-computed post-physics state.

Called once per control step in :meth:RLEnvBase.step after the physics sub-steps but BEFORE metric/termination/reward, so terms that cache robot state (e.g. the motion-imitation command) grade the CURRENT step's state (S_N) rather than the previous step's (S_{N-1}). This method owns that ordering contract; individual terms override the no-op :meth:CommandTerm.post_physics_refresh.

Being above the graded layer is also what makes this the right hook for a term's own per-step grading state, rather than :meth:CommandTerm._update_command below the reward — see that method, and docs/guides/terms.md.

Source code in eden/managers/command_manager.py

def post_physics_refresh(self) -> None:
    """Refresh every term's robot-state cache on the just-computed post-physics state.

    Called once per control step in :meth:`RLEnvBase.step` after the physics sub-steps
    but BEFORE metric/termination/reward, so terms that cache robot state (e.g. the
    motion-imitation command) grade the CURRENT step's state (S_N) rather than the
    previous step's (S_{N-1}). This method owns that ordering contract; individual terms
    override the no-op :meth:`CommandTerm.post_physics_refresh`.

    Being above the graded layer is also what makes this the right hook for a term's own
    per-step *grading* state, rather than :meth:`CommandTerm._update_command` below the
    reward — see that method, and ``docs/guides/terms.md``.
    """
    for term in self._terms.values():
        term.post_physics_refresh()

pre_physics_advance ¶

pre_physics_advance() -> None

Advance every term's command BEFORE the physics sub-steps (no-op by default).

Called once per control step in :meth:RLEnvBase.step, before the decimation loop, so terms tracking a time-advancing reference (e.g. the motion-imitation command) advance it ahead of physics — see :meth:CommandTerm.pre_physics_advance.

Source code in eden/managers/command_manager.py

def pre_physics_advance(self) -> None:
    """Advance every term's command BEFORE the physics sub-steps (no-op by default).

    Called once per control step in :meth:`RLEnvBase.step`, before the decimation
    loop, so terms tracking a time-advancing reference (e.g. the motion-imitation
    command) advance it ahead of physics — see :meth:`CommandTerm.pre_physics_advance`.
    """
    for term in self._terms.values():
        term.pre_physics_advance()

rewind_reference ¶

rewind_reference() -> None

Restart every term's time-advancing reference at its first frame (no-op by default).

Driven only by DeploymentBase.reset_motion_reference — see :meth:CommandTerm.rewind_reference for what a term may and may not do there.

Source code in eden/managers/command_manager.py

def rewind_reference(self) -> None:
    """Restart every term's time-advancing reference at its first frame (no-op by default).

    Driven only by ``DeploymentBase.reset_motion_reference`` — see
    :meth:`CommandTerm.rewind_reference` for what a term may and may not do there.
    """
    for term in self._terms.values():
        term.rewind_reference()

advance_reference ¶

advance_reference() -> None

Advance each deployed reference; simulation still uses pre_physics_advance.

Source code in eden/managers/command_manager.py

def advance_reference(self) -> None:
    """Advance each deployed reference; simulation still uses pre_physics_advance."""
    for term in self._terms.values():
        term.advance_reference()

reset ¶

reset(envs_idx: slice | Tensor | None = None) -> dict[str, Tensor]

Source code in eden/managers/command_manager.py

def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> dict[str, torch.Tensor]:
    envs_idx = sanitize_envs_idx(envs_idx, self._env.num_envs)
    extras = {}
    for name, term in self._terms.items():
        stats = term.reset(envs_idx=envs_idx)
        for stat_name, stat_value in stats.items():
            extras[f"Commands/{name}/{stat_name}"] = stat_value
    return extras

compute ¶

compute(dt: float) -> None

Resample and publish every term's command (phase 17 of :meth:RLEnvBase.step).

Called after the auto-reset and below the reward, so the command it leaves behind is the one phase 19's observations carry and the next action is chosen against — deliberately not the one this step was graded on, which is what keeps obs -> action -> reward aligned. See :meth:CommandTerm._update_command.

Source code in eden/managers/command_manager.py

def compute(self, dt: float) -> None:
    """Resample and publish every term's command (phase 17 of :meth:`RLEnvBase.step`).

    Called after the auto-reset and below the reward, so the command it leaves behind is the
    one phase 19's observations carry and the next action is chosen against — deliberately not
    the one this step was graded on, which is what keeps ``obs -> action -> reward`` aligned.
    See :meth:`CommandTerm._update_command`.
    """
    for term in self._terms.values():
        term.compute(dt)

get_command ¶

get_command(name: str) -> Tensor

Return the command for the specified command term.

Source code in eden/managers/command_manager.py

def get_command(self, name: str) -> torch.Tensor:
    """Return the command for the specified command term."""
    return self.get_term(name).command

swap_term ¶

swap_term(name: str, options: CommandTermOptions) -> CommandTerm

Replace the built term name with one configured by options; return the old one.

The post-build counterpart of what the config does before build: a deployment substitutes a task's trained command generator for one an operator drives (:class:~eden.managers.terms.commands.velocity.OperatorVelocityCommand) or one that simply holds a value, the way it substitutes the simulated entity for a :class:~eden.extensions.deployment.utils.state.RobotStateEntity. The caller owns the returned term and can pass its options back here to undo the swap.

The replacement is built and build()-ed exactly as _prepare_terms would, and its command must have the same shape as the one it replaces: the observation buffers and the trained policy's input width were fixed by the old term, so a different width would reach the policy as a silently misaligned observation vector.

Nothing here repoints terms that cached the replaced term in their own build() (the pattern self._command = env.command_manager.get_term(...)). A caller that needs those to follow the swap must repoint them — see :meth:~eden.extensions.deployment.base.DeploymentBase._repoint_command_term_refs.

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active term (the message lists the active ones).
`ValueError`	If the replacement's command shape differs from the replaced term's.

Source code in eden/managers/command_manager.py

def swap_term(self, name: str, options: CommandTermOptions) -> CommandTerm:
    """Replace the built term ``name`` with one configured by ``options``; return the old one.

    The post-build counterpart of what the config does before build: a deployment
    substitutes a task's trained command generator for one an operator drives
    (:class:`~eden.managers.terms.commands.velocity.OperatorVelocityCommand`) or one that
    simply holds a value, the way it substitutes the simulated entity for a
    :class:`~eden.extensions.deployment.utils.state.RobotStateEntity`. The caller owns the
    returned term and can pass its options back here to undo the swap.

    The replacement is built and ``build()``-ed exactly as ``_prepare_terms`` would, and its
    command must have the **same shape** as the one it replaces: the observation buffers and
    the trained policy's input width were fixed by the old term, so a different width would
    reach the policy as a silently misaligned observation vector.

    Nothing here repoints terms that cached the *replaced* term in their own ``build()``
    (the pattern ``self._command = env.command_manager.get_term(...)``). A caller that needs
    those to follow the swap must repoint them — see
    :meth:`~eden.extensions.deployment.base.DeploymentBase._repoint_command_term_refs`.

    Raises
    ------
    KeyError
        If ``name`` is not an active term (the message lists the active ones).
    ValueError
        If the replacement's command shape differs from the replaced term's.
    """
    old_term = self.get_term(name)
    new_term = self._build_term(name, term_option=options)
    # ``build()`` has already run, so the replacement may own something by now — an
    # OperatorVelocityCommand holds a *process-global* keyboard grab. Release it if we
    # end up rejecting it, or a config error leaves that grab behind with nothing
    # holding a reference to close it.
    try:
        if new_term.command.shape != old_term.command.shape:
            raise ValueError(
                f"Cannot swap command term '{name}': {type(new_term).__name__} commands "
                f"{tuple(new_term.command.shape)} but {type(old_term).__name__} commands "
                f"{tuple(old_term.command.shape)}. The observation width the policy was trained "
                "on comes from the replaced term, so the two must agree."
            )
    except BaseException:
        close = getattr(new_term, "close", None)
        if close is not None:
            close()
        raise
    self._terms[name] = new_term
    return old_term

restore_term ¶

restore_term(name: str, term: CommandTerm) -> None

Put back a term :meth:swap_term returned, undoing that swap.

Deliberately narrow — it takes an already-built term rather than options, because the only correct argument is the exact object swap_term displaced. Rebuilding the original from its options instead would drop whatever state it had accumulated and leave any reference the caller repointed pointing at a third object.

Source code in eden/managers/command_manager.py

def restore_term(self, name: str, term: CommandTerm) -> None:
    """Put back a term :meth:`swap_term` returned, undoing that swap.

    Deliberately narrow — it takes an already-built term rather than options, because the
    only correct argument is the exact object ``swap_term`` displaced. Rebuilding the
    original from its options instead would drop whatever state it had accumulated and
    leave any reference the caller repointed pointing at a third object.
    """
    if name not in self._terms:
        available = ", ".join(self._term_names) or "none"
        raise KeyError(f"No active term named '{name}' in {type(self).__name__}. Available options: {available}")
    self._terms[name] = term

Functions:¶

EventManager¶

Event manager for applying randomization and other event-driven changes.

Attributes¶

EVENT_TERM_REGISTRY `module-attribute` ¶

EVENT_TERM_REGISTRY = TermRegistry('EVENT_TERM', EventTerm, EventTermFuncWrapper, populated_by=('eden.managers.terms.events',))

Classes¶

EventTerm ¶

EventTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[EventTermOptions], _EventTermFields

Base class for event terms.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes¶

mode `class-attribute` `instance-attribute` ¶

mode: EventMode = _RESET

interval_range_s `class-attribute` `instance-attribute` ¶

interval_range_s: tuple[float, float] | None = None

is_global_time `class-attribute` `instance-attribute` ¶

is_global_time: bool = False

min_step_count_between_reset `class-attribute` `instance-attribute` ¶

min_step_count_between_reset: int = 0

priority `class-attribute` `instance-attribute` ¶

priority: int = 0

Methods:¶

compute_batch `classmethod` ¶

compute_batch(terms: Sequence[EventTerm], envs_idx: slice | Tensor | None) -> None

Fire terms, RESET terms of this kind adjacent in priority order, on the same envs_idx.

The default fires them one by one. A kind whose terms each write simulator state overrides it to write once for all of them (the reference-state-initialization terms: one Genesis write and one forward-kinematics pass per reset, not one per entity). The event manager hands a run of adjacent RESET terms to one call when each term's class takes both compute and compute_batch from the same class (a subclass that overrides compute alone runs alone) and none sets min_step_count_between_reset (so every term fires on exactly envs_idx). A throttle set on a batched term after the build splits its batch back into single terms.

Source code in eden/managers/event_manager.py

@classmethod
def compute_batch(cls, terms: Sequence[EventTerm], envs_idx: slice | torch.Tensor | None) -> None:
    """Fire ``terms``, RESET terms of this kind adjacent in priority order, on the same ``envs_idx``.

    The default fires them one by one. A kind whose terms each write simulator state overrides it
    to write once for all of them (the reference-state-initialization terms: one Genesis write and
    one forward-kinematics pass per reset, not one per entity). The event manager hands a run
    of adjacent RESET terms to one call when each term's class takes both ``compute`` and
    ``compute_batch`` from the same class (a subclass that overrides ``compute`` alone runs alone)
    and none sets ``min_step_count_between_reset`` (so every term fires on exactly ``envs_idx``). A
    throttle set on a batched term after the build splits its batch back into single terms.
    """
    for term in terms:
        term.compute(envs_idx=envs_idx)

EventTermFuncWrapper ¶

EventTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermFuncWrapper[EventTermOptions], _EventTermFields

Base class for event terms defined as function.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

Attributes¶

mode `class-attribute` `instance-attribute` ¶

mode: EventMode = _RESET

interval_range_s `class-attribute` `instance-attribute` ¶

interval_range_s: tuple[float, float] | None = None

is_global_time `class-attribute` `instance-attribute` ¶

is_global_time: bool = False

min_step_count_between_reset `class-attribute` `instance-attribute` ¶

min_step_count_between_reset: int = 0

priority `class-attribute` `instance-attribute` ¶

priority: int = 0

ResetTermBatch `dataclass` ¶

ResetTermBatch(owner: type[EventTerm], names: list[str], states: list[_ResetTermState])

RESET terms adjacent in priority order that one :meth:EventTerm.compute_batch fires together.

Every batchable term belongs to one, alone or with its neighbours. The event manager fires it through :meth:compute, never through the terms' own compute, so a per-term timer has to wrap this instead (RLStepProfiler does; see :meth:EventManager.iter_reset_batches).

Attributes¶

owner `instance-attribute` ¶

owner: type[EventTerm]

names `instance-attribute` ¶

names: list[str]

states `instance-attribute` ¶

states: list[_ResetTermState]

terms `property` ¶

terms: list[EventTerm]

Methods:¶

compute ¶

compute(envs_idx: slice | Tensor) -> None

Source code in eden/managers/event_manager.py

def compute(self, envs_idx: slice | torch.Tensor) -> None:
    self.owner.compute_batch(self.terms, envs_idx)

EventManager ¶

EventManager(env: EnvBase, options: T)

Bases: ManagerBase[EventManagerOptions]

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: T):
    self._env = env
    self._options: T = options
    self._prepare_terms()

Attributes¶

mode_term_names `property` ¶

mode_term_names: dict[EventMode, list[str]]

The active event term names keyed by mode, in priority order.

active_terms `property` ¶

active_terms: list[str]

The flat, de-duplicated union of all active term names across modes.

available_modes `property` ¶

available_modes: list[EventMode]

domain_randomization_fields `property` ¶

domain_randomization_fields: list[str]

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/event_manager.py

def summary(self) -> str:
    msg = f"<EventManager> contains {len(self._mode_term_names)} active terms.\n"
    for mode in self._mode_term_names.keys():
        terms = enumerate(zip(self._mode_term_names[mode], self._mode_terms[mode], strict=True))
        if mode == _INTERVAL:
            field_names = ["Index", "Name", "Priority", "Interval time range (s)"]
            rows = ([index, name, term.priority, term.interval_range_s] for index, (name, term) in terms)
        else:
            field_names = ["Index", "Name", "Priority"]
            rows = ([index, name, term.priority] for index, (name, term) in terms)
        msg += format_table(f"Active Event Terms in Mode: '{mode}'", field_names, rows, align={"Name": "l"})
        msg += "\n"
    if self._domain_randomization_fields:
        msg += format_table(
            "Domain Randomization Fields",
            ["Index", "Field Name"],
            enumerate(self._domain_randomization_fields),
            align={"Field Name": "l"},
        )
        msg += "\n"
    return msg

iter_terms ¶

iter_terms() -> Iterator[tuple[EventMode, str, ManagerTermBase]]

Yield (mode, name, term) for every active event term, in mode/priority order.

Public accessor mirroring :meth:~eden.managers.reward_manager.RewardManager.iter_terms so callers can scan term objects without touching the private per-mode storage.

Source code in eden/managers/event_manager.py

def iter_terms(self) -> Iterator[tuple[EventMode, str, ManagerTermBase]]:
    """Yield ``(mode, name, term)`` for every active event term, in mode/priority order.

    Public accessor mirroring :meth:`~eden.managers.reward_manager.RewardManager.iter_terms`
    so callers can scan term objects without touching the private per-mode storage.
    """
    for mode in self._mode_term_names:
        for name, term in zip(self._mode_term_names[mode], self._mode_terms[mode], strict=True):
            yield mode, name, term

get_term ¶

get_term(name: str) -> ManagerTermBase

Return the active event term registered under config key name, whatever its mode.

For a term another one reads from, e.g. ResetFromStateBuffer.current_column serving a goal stored with the reset states.

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active term. The message lists the active ones, worded as the flat managers' `get_term` words it.

Source code in eden/managers/event_manager.py

def get_term(self, name: str) -> ManagerTermBase:
    """Return the active event term registered under config key ``name``, whatever its mode.

    For a term another one reads from, e.g. ``ResetFromStateBuffer.current_column`` serving a goal stored
    with the reset states.

    Raises
    ------
    KeyError
        If ``name`` is not an active term. The message lists the active ones, worded as the flat
        managers' ``get_term`` words it.
    """
    for _, term_name, term in self.iter_terms():
        if term_name == name:
            return term
    available = ", ".join(self.active_terms) or "none"
    raise KeyError(f"No active term named '{name}' in {type(self).__name__}. Available options: {available}")

iter_reset_batches ¶

iter_reset_batches() -> Iterator[ResetTermBatch]

Yield every batch of RESET terms (see :class:ResetTermBatch), in priority order.

Their terms' own compute is never called on a reset, so time the batch instead.

Source code in eden/managers/event_manager.py

def iter_reset_batches(self) -> Iterator[ResetTermBatch]:
    """Yield every batch of RESET terms (see :class:`ResetTermBatch`), in priority order.

    Their terms' own ``compute`` is never called on a reset, so time the batch instead.
    """
    for run in self._reset_runs:
        if isinstance(run, ResetTermBatch):
            yield run

reset ¶

reset(envs_idx: slice | Tensor | None = None)

Source code in eden/managers/event_manager.py

def reset(self, envs_idx: (slice | torch.Tensor) | None = None):
    for mode_terms in self._mode_reset_terms.values():
        for term in mode_terms:
            term.reset(envs_idx=envs_idx)
    envs_idx = sanitize_envs_idx(envs_idx, self._env.num_envs)
    for state in self._interval_states:
        if not isinstance(state.term, EventTerm):
            continue
        if not state.term.is_global_time:
            assert state.term.interval_range_s is not None
            lower, upper = state.term.interval_range_s
            sampled_interval = torch.rand(self._env.num_envs, device=self.device) * (upper - lower) + lower
            assign_over_envs(state.time_left, envs_idx, sampled_interval)
    return {}

iter_post_reset_terms ¶

iter_post_reset_terms() -> Iterator[ManagerTermBase]

Every event term, in mode/priority order (a term shared across modes repeats).

Source code in eden/managers/event_manager.py

def iter_post_reset_terms(self) -> Iterator[ManagerTermBase]:
    """Every event term, in mode/priority order (a term shared across modes repeats)."""
    return (term for _, _, term in self.iter_terms())

compute ¶

compute(mode: EventMode, envs_idx: slice | Tensor | None = None, dt: float | None = None, global_env_step_count: int | None = None, n_envs: int | None = None)

Source code in eden/managers/event_manager.py

def compute(
    self,
    mode: EventMode,
    envs_idx: (slice | torch.Tensor) | None = None,
    dt: float | None = None,
    global_env_step_count: int | None = None,
    n_envs: int | None = None,
):
    # ``n_envs``: how many envs ``envs_idx`` selects, when the caller has already counted them (the
    # auto-reset has — it is the gate in ``EnvBase._reset_scene_and_events``). Counting a bool mask
    # is a GPU->CPU sync, so a known count is not paid for twice.
    # Dispatch per mode rather than re-checking ``mode`` for every term. The
    # per-term bodies live in ``_step_{interval,reset}_term`` so each mode's
    # trigger logic reads independently.
    if mode == _INTERVAL:
        if dt is None:
            raise ValueError(f"Event mode '{mode}' requires the time-step of the environment.")
        if envs_idx is not None:
            raise ValueError(
                f"Event mode '{mode}' does not require environment indices. This is an undefined behavior"
                " as the environment indices are computed based on the time left for each environment."
            )
        for state in self._interval_states:
            self._step_interval_term(state, dt)
    elif mode == _RESET:
        if global_env_step_count is None:
            raise ValueError(f"Event mode '{mode}' requires the total number of environment steps to be provided.")
        if n_envs is None:
            envs_idx, n_envs = sanitize_envs_idx(envs_idx, self._env.num_envs, return_n_envs=True)
        else:
            envs_idx = sanitize_envs_idx(envs_idx, self._env.num_envs)
        if n_envs == 0:
            return
        for run in self._reset_runs:
            if isinstance(run, ResetTermBatch):
                self._step_reset_batch(run, envs_idx, global_env_step_count)
            else:
                self._step_reset_term(run, envs_idx, global_env_step_count)
    elif mode == _STARTUP:
        for term in self._mode_terms[_STARTUP]:
            term.compute(envs_idx=envs_idx)
    else:
        raise ValueError(f"Invalid event mode: {mode}")

Functions:¶

CurriculumManager¶

CurriculumManager.get_state(name) returns what another curriculum term published this step and raises, naming the fix, whenever there is no such value. See the curricula guide.

Curriculum manager for updating environment quantities subject to a training curriculum.

Attributes¶

CURRICULUM_TERM_REGISTRY `module-attribute` ¶

CURRICULUM_TERM_REGISTRY = TermRegistry('CURRICULUM_TERM', CurriculumTerm, CurriculumTermFuncWrapper, populated_by=('eden.managers.terms.curricula',))

Classes¶

CurriculumTerm ¶

CurriculumTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[CurriculumTermOptions]

Base class for curriculum terms.

compute() returns the term's published state: a dict of scalars, logged as Curriculum/<term_name>/<key> and served to other terms by :meth:CurriculumManager.get_state. Publish it on every step. A term that returns a dict on some steps and None on others makes get_state raise on exactly the steps it skipped — the contract is "never None", so there is no quiet no-value for a consumer to branch on. A term with genuinely nothing to share simply returns None always, and its state is then not readable at all, which get_state says in as many words.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

CurriculumTermFuncWrapper ¶

CurriculumTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermFuncWrapper[CurriculumTermOptions]

Base class for curriculum terms defined as a function.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

CurriculumManager ¶

CurriculumManager(env: EnvBase, options: CurriculumManagerOptions)

Bases: FlatManagerBase[CurriculumManagerOptions]

Curriculum manager for managing the curriculum of the environment.

Source code in eden/managers/curriculum_manager.py

def __init__(self, env: EnvBase, options: CurriculumManagerOptions):
    super().__init__(env=env, options=options)

    # Keyed only once a term has actually computed, so "never computed" is a missing key
    # rather than a ``None`` value indistinguishable from "computed and published nothing".
    # :meth:`get_state` turns each of those into its own raise.
    self._curriculum_state: dict[str, Any] = dict()
    #: Name of the term currently inside :meth:`compute`, or ``None`` outside the loop. Lets
    #: :meth:`get_state` tell "declared after its consumer" from "read before the run's first
    #: compute()" exactly, rather than guessing from an empty cache -- the cache is *also*
    #: empty for the first term in the loop on the first step, which is the common ordering bug.
    self._computing: str | None = None
    #: ``common_step_counter`` at each term's last publish, written beside the value it dates.
    #: Per-term rather than one stamp for the loop so a term whose ``compute()`` *raised* keeps
    #: the date of the step it last succeeded on, instead of being credited with the step it
    #: died on -- the loop's bookkeeping must not lie to a caller that caught the exception.
    self._published_at: dict[str, int] = dict()

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/curriculum_manager.py

def summary(self) -> str:
    return self._format_summary_table(
        title="Active Curriculum Terms",
        field_names=["Index", "Name"],
        rows=([index, name] for index, name in enumerate(self._term_names)),
        align={"Name": "l"},
    )

reset ¶

reset(envs_idx: Tensor | slice | None = None) -> dict[str, float | Tensor]

Publish the cached per-term state, then feed the finished episodes to the terms.

The term half is frozen at eval, exactly like :meth:compute — a curriculum term's reset() is how it accumulates the statistics compute() would adapt from (an episode- length EMA, a success rate), so running one without the other would have the internal state drift away from the published one for the whole rollout, on evidence nothing is allowed to act on.

The publish half reports whatever the last :meth:compute cached — and an eval env has no last compute(): env.training is fixed at construction, so an eval rollout logs no Curriculum/* key at all, from its first reset to its last. That is deliberate: the level an eval was graded at is the one its config pins, which the saved config already records; publishing it from here would mean running compute() at build, and the two terms whose compute() has side effects — StageRewardWeightCurriculum and VelocityCommandCurriculum — would then overwrite the configured reward weights and command ranges with their step-0 stage.

Source code in eden/managers/curriculum_manager.py

def reset(self, envs_idx: torch.Tensor | slice | None = None) -> dict[str, float | torch.Tensor]:
    """Publish the cached per-term state, then feed the finished episodes to the terms.

    The term half is frozen at eval, exactly like :meth:`compute` — a curriculum term's
    ``reset()`` is how it accumulates the statistics ``compute()`` would adapt from (an episode-
    length EMA, a success rate), so running one without the other would have the internal state
    drift away from the published one for the whole rollout, on evidence nothing is allowed to
    act on.

    The publish half reports whatever the last :meth:`compute` cached — and **an eval env has
    no last compute()**: ``env.training`` is fixed at construction, so an eval rollout logs no
    ``Curriculum/*`` key at all, from its first reset to its last. That is deliberate: the
    level an eval was graded at is the one its config pins, which the
    saved config already records; publishing it from here would mean running ``compute()`` at
    build, and the two terms whose ``compute()`` has side effects —
    ``StageRewardWeightCurriculum`` and ``VelocityCommandCurriculum`` — would then *overwrite*
    the configured reward weights and command ranges with their step-0 stage.
    """
    extras = {}
    for term_name, term_state in self._curriculum_state.items():
        if term_state is not None:
            if isinstance(term_state, dict):
                for key, value in term_state.items():
                    extras[f"Curriculum/{term_name}/{key}"] = _published(value)
            else:
                extras[f"Curriculum/{term_name}"] = _published(term_state)
    if not self._env.training:
        return extras
    for term in self._reset_terms:
        term.reset(envs_idx=envs_idx)
    return extras

compute ¶

compute() -> None

Adapt every term and cache what it published — a no-op at eval.

Curricula adapt task difficulty from training statistics, so an eval rollout freezes them: otherwise its difficulty is non-stationary, e.g. an EMA-gated goal curriculum at num_envs=1 collapses its window to a single episode, so one failed episode retreats the level mid-rollout and every later episode is graded on an easier task than the one being reported. Terms still apply their configured initial level at build, so eval entry points pin difficulty through the term's own config.

The freeze is all-or-nothing and lives here rather than at the call site so it cannot go asymmetric: :meth:reset is gated the same way. Nothing is cached either, so :meth:get_state raises for every term for the whole rollout rather than handing a consumer a None that reads like "not ready yet".

Source code in eden/managers/curriculum_manager.py

def compute(self) -> None:
    """Adapt every term and cache what it published — a no-op at eval.

    Curricula adapt task difficulty from training statistics, so an eval rollout freezes them:
    otherwise its difficulty is non-stationary, e.g. an EMA-gated goal curriculum at
    ``num_envs=1`` collapses its window to a single episode, so one failed episode retreats the
    level mid-rollout and every later episode is graded on an easier task than the one being
    reported. Terms still apply their configured initial level at build, so eval entry points
    pin difficulty through the term's own config.

    The freeze is all-or-nothing and lives here rather than at the call site so it cannot go
    asymmetric: :meth:`reset` is gated the same way. Nothing is cached either, so
    :meth:`get_state` raises for every term for the whole rollout rather than handing a
    consumer a ``None`` that reads like "not ready yet".
    """
    if not self._env.training:
        return
    try:
        for name, term in self._terms.items():
            # Recorded so a term that reads a not-yet-computed sibling gets an error naming
            # *which config entry* to move -- the traceback carries the consumer's class, but a
            # task can hold several entries of one class (``penalty_torques`` and
            # ``penalty_smoothness`` in docs/guides/curricula.md), and the entry key is what you edit.
            self._computing = name
            self._curriculum_state[name] = term.compute()
            self._published_at[name] = self._env.common_step_counter
    finally:
        self._computing = None

get_state ¶

get_state(term_name: str) -> dict

Return the most recent compute() output cached for the named term.

Never returns None. Every way there can be no state is a raise, because the one thing a consumer must not do is read the absence as "not ready yet" and fall back to a default: that fallback then stands for the whole run, with nothing logged and nothing raised. When you want a level that is correct in both modes, read it off the term itself — curriculum_manager.get_term(name) — which is what ObjectVelocityDisturbance does; get_state is only for consuming what another term published this step, and it enforces that rather than only asking for it: a cached value from an earlier step raises too. Without that the same read is treated two ways -- the run's first step raises, and every step after silently serves the previous step's level, which is the intermittency this contract exists to remove.

Freshness is dated against env.common_step_counter, which advances once per step with the physics (:meth:RLEnvBase.step), so a read from phases 5-20 is placed exactly. Phases 1-3 run before that advance and so still date as the previous step; there is no earlier per-step marker to key off, and what they would read is in any case the newest publish that can exist, since this step's :meth:compute has not run.

RAISES DESCRIPTION

KeyError

If no curriculum term is named term_name, so a typo cannot pass for "not computed yet" and leave a consumer that treats that as "not ready" on its default for the whole run.

RuntimeError

If the named term has no state this step. The five causes get five messages, each naming the one edit that fixes it: the env is at eval, so nothing ever computes; the term is declared after the consumer now reading it; the read happens before the run's first compute() at all; the term's compute() returns None; or the read runs earlier in the step than phase 16, so the cache still holds the previous step's publish.

Source code in eden/managers/curriculum_manager.py

def get_state(self, term_name: str) -> dict:
    """Return the most recent ``compute()`` output cached for the named term.

    **Never returns ``None``.** Every way there can be no state is a raise, because the one
    thing a consumer must not do is read the absence as "not ready yet" and fall back to a
    default: that fallback then stands for the whole run, with nothing logged and nothing
    raised. When you want a level that is correct in *both* modes, read it off the term
    itself — ``curriculum_manager.get_term(name)`` — which is what ``ObjectVelocityDisturbance``
    does; ``get_state`` is only for consuming what another term *published* this step, and it
    enforces that rather than only asking for it: a cached value from an *earlier* step raises
    too. Without that the same read is treated two ways -- the run's first step raises, and
    every step after silently serves the previous step's level, which is the intermittency this
    contract exists to remove.

    Freshness is dated against ``env.common_step_counter``, which advances once per step with
    the physics (:meth:`RLEnvBase.step`), so a read from phases 5-20 is placed exactly. Phases
    1-3 run *before* that advance and so still date as the previous step; there is no earlier
    per-step marker to key off, and what they would read is in any case the newest publish that
    can exist, since this step's :meth:`compute` has not run.

    Raises
    ------
    KeyError
        If no curriculum term is named ``term_name``, so a typo cannot pass for "not computed
        yet" and leave a consumer that treats that as "not ready" on its default for the whole run.
    RuntimeError
        If the named term has no state this step. The five causes get five messages, each
        naming the one edit that fixes it: the env is at eval, so nothing ever computes; the
        term is declared after the consumer now reading it; the read happens before the run's
        first ``compute()`` at all; the term's ``compute()`` returns ``None``; or the read runs
        earlier in the step than phase 16, so the cache still holds the previous step's
        publish.
    """
    if term_name not in self._terms:
        # One line on purpose: KeyError.__str__ is repr(args[0]), so a newline here would
        # reach the reader as a literal "\n" in an escaped blob.
        raise KeyError(
            f"No curriculum term named '{term_name}'. Declared curriculum terms: "
            f"{', '.join(self._terms) or '(none)'}. Fix: use one of those keys, or add "
            f"'{term_name}' to curriculum_options."
        )
    if term_name in self._curriculum_state:
        state = self._curriculum_state[term_name]
        if state is None:
            raise RuntimeError(
                f"Curriculum term '{term_name}' ({self._term_label(term_name)}) computed but published "
                f"nothing: its compute() returned None.\n"
                f"Fix: return a dict of scalars from {self._term_label(term_name)}.compute() — it is both "
                f"what get_state() serves and what is logged as Curriculum/{term_name}/<key>.\n"
                "A term that publishes on some steps and not others raises here intermittently; publish "
                "every step, or do not let anything read this term's state."
            )
        published_at = self._published_at[term_name]
        step = self._env.common_step_counter
        if published_at != step:
            raise RuntimeError(
                f"Curriculum term '{term_name}' last published at step {published_at}, and it is now step "
                f"{step}: this cached value is one control step stale.\n"
                "reading from: a phase earlier in the step than CurriculumManager.compute(), which is "
                "phase 16 (docs/guides/environments.md) — a curriculum term's reset(), a metric, a "
                "termination and a reward all run before it\n"
                f"Fix: read the level off the term itself — curriculum_manager.get_term('{term_name}'), a "
                f"{self._term_label(term_name)} — which is current at every phase; or move this read into "
                "a hook that runs after phase 16, where get_state() serves the current step's publish.\n"
                "Do NOT call CurriculumManager.compute() yourself to refresh it: terms adapt as a side "
                "effect of computing — PenaltyCurriculum steps its scale by `degree` toward a threshold "
                "bound on each call — so a second call inside one step advances the curriculum twice."
            )
        return state
    if not self._env.training:
        raise RuntimeError(
            f"Curriculum term '{term_name}' has no state: this env was built with eval_mode=True, and "
            "CurriculumManager.compute() is a no-op at eval, so no term computes and nothing is ever "
            "cached — for the whole rollout, not just this step.\n"
            f"Fix: read the level off the term itself — curriculum_manager.get_term('{term_name}'), a "
            f"{self._term_label(term_name)} — which holds the configured level in training and at eval "
            "alike; that is what ObjectVelocityDisturbance._adaptive_scale does.\n"
            "Do NOT catch this and fall back to a default: the default would stand for every step of the "
            "eval, grading it at a difficulty no training step used, with nothing logged and nothing "
            "raised. Closing that silent fallback is why this raises at all."
        )
    if self._computing is not None:
        raise RuntimeError(
            f"Curriculum term '{term_name}' has not computed yet, so it has no state to read.\n"
            f"curriculum_options order: {', '.join(self._term_names)}\n"
            f"'{self._computing}' is reading '{term_name}', but '{term_name}' is declared after it\n"
            f"Fix: move '{term_name}' above '{self._computing}' in curriculum_options — "
            "CurriculumManager computes terms in declaration order, so a producer declared after its "
            "consumer would feed it a value one control step stale."
        )
    raise RuntimeError(
        f"Curriculum term '{term_name}' has no state yet: this read happens before the run's first "
        "CurriculumManager.compute(), which is step phase 16 (docs/guides/environments.md).\n"
        "reading from: outside the curriculum compute loop — build(), the first reset(), or task setup\n"
        f"Fix: read the term's own level off curriculum_manager.get_term('{term_name}'), a "
        f"{self._term_label(term_name)}, which is valid from build() onward. get_state() only ever "
        "serves what a term has already published."
    )

MetricManager¶

Metric manager and MetricTerm base for task success / evaluation signals.

Attributes¶

MetricTermLike `module-attribute` ¶

MetricTermLike: TypeAlias = MetricTerm | MetricTermFuncWrapper

Union of a class-based metric term and its function-wrapper form.

METRIC_TERM_REGISTRY `module-attribute` ¶

METRIC_TERM_REGISTRY = TermRegistry('METRIC_TERM', MetricTerm, MetricTermFuncWrapper, populated_by=('eden.managers.terms.metrics',))

Classes¶

SuccessCheckerMixin ¶

SuccessCheckerMixin(options: T)

Bases: ConfigurableMixin

The configurable fields every metric term has, class or function, and the success check they drive.

PARAMETER	DESCRIPTION
`success_threshold`	The threshold for success. The threshold is inclusive. Defaults to 1.0.
`direction`	The metric direction. Defaults to `MetricDirection.HIB` (`"hib"`).
`metric_mode`	The metric success mode. Defaults to `MetricMode.INTERVAL` (`"interval"`).
`is_cumulative`	Whether the metric is cumulative. Defaults to False.
`contributes_to_success`	Whether this term participates in task-success accounting (`success_buf`, `Metrics/Success`, `Metrics/Progress`, and the per-episode success label the recorder stamps). Defaults to True. Set False for a term that exists only to share a computed quantity with other terms (via `metric_manager.get_value` — see `metric_reward` / `metric_observation`) and has no notion of succeeding: such a term would otherwise pin `Metrics/Success` to its own trivially-true (or trivially-false) verdict for the whole task. `is_success` is never called on a non-contributing term, so its `success_threshold` / `direction` are unused.

Notes

These five settings are fixed once the metric manager is built: it reads them into its batched reset() then, so a later write to one (from a curriculum, say) would not take effect. Set them through configure.

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¶

success_threshold `class-attribute` `instance-attribute` ¶

success_threshold: float = 1.0

direction `class-attribute` `instance-attribute` ¶

direction: MetricDirection = MetricDirection.HIB

metric_mode `class-attribute` `instance-attribute` ¶

metric_mode: MetricMode = MetricMode.INTERVAL

is_cumulative `class-attribute` `instance-attribute` ¶

is_cumulative: bool = False

contributes_to_success `class-attribute` `instance-attribute` ¶

contributes_to_success: bool = True

Methods:¶

is_success ¶

is_success(value: Tensor, envs_idx: Tensor | None = None) -> Tensor

Check whether a metric is successful (default implementation).

This method can be overridden by subclasses to provide a custom implementation.

PARAMETER	DESCRIPTION
`value`	The value of the metric. TYPE: `Tensor`
`envs_idx`	The indices of the environments to evaluate. If None, all environments are used. Defaults to None. TYPE: `Tensor \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	A boolean tensor indicating whether the metric is successful.

Source code in eden/managers/metric_manager.py

def is_success(self, value: torch.Tensor, envs_idx: torch.Tensor | None = None) -> torch.Tensor:
    """Check whether a metric is successful (default implementation).

    This method can be overridden by subclasses to provide a custom implementation.

    Parameters
    ----------
    value: torch.Tensor
        The value of the metric.
    envs_idx: torch.Tensor, optional
        The indices of the environments to evaluate. If None, all environments are used. Defaults to None.

    Returns
    -------
    torch.Tensor
        A boolean tensor indicating whether the metric is successful.
    """
    if envs_idx is None:
        envs_idx = slice(None)
    if self.is_cumulative:
        value_norm = value[envs_idx] / torch.clamp_min(self._env.episode_length_buf[envs_idx], 1)
    else:
        value_norm = value[envs_idx]
    if self.direction == MetricDirection.HIB:
        return value_norm >= self.success_threshold
    elif self.direction == MetricDirection.LIB:
        return value_norm <= self.success_threshold
    else:
        raise ValueError(f"Invalid metric direction: {self.direction}")

MetricTerm ¶

MetricTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[MetricTermOptions], SuccessCheckerMixin

Base class for metric terms. The fields every metric term has are on :class:SuccessCheckerMixin.

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes¶

success_threshold `class-attribute` `instance-attribute` ¶

success_threshold: float = 1.0

direction `class-attribute` `instance-attribute` ¶

direction: MetricDirection = MetricDirection.HIB

metric_mode `class-attribute` `instance-attribute` ¶

metric_mode: MetricMode = MetricMode.INTERVAL

is_cumulative `class-attribute` `instance-attribute` ¶

is_cumulative: bool = False

contributes_to_success `class-attribute` `instance-attribute` ¶

contributes_to_success: bool = True

MetricTermFuncWrapper ¶

MetricTermFuncWrapper(func: FunctionType, env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermFuncWrapper[MetricTermOptions], SuccessCheckerMixin

Base class for metric terms defined as function. See :class:SuccessCheckerMixin for the shared fields.

Source code in eden/managers/base.py

def __init__(self, func: FunctionType, env: EnvBase, options: _TermOptionsT):
    ManagerTermFuncWrapperBase.__init__(self, func=func, env=env)
    ConfigurableFuncWrapperMixin.__init__(self, options=options)

Attributes¶

success_threshold `class-attribute` `instance-attribute` ¶

success_threshold: float = 1.0

direction `class-attribute` `instance-attribute` ¶

direction: MetricDirection = MetricDirection.HIB

metric_mode `class-attribute` `instance-attribute` ¶

metric_mode: MetricMode = MetricMode.INTERVAL

is_cumulative `class-attribute` `instance-attribute` ¶

is_cumulative: bool = False

contributes_to_success `class-attribute` `instance-attribute` ¶

contributes_to_success: bool = True

MetricManager ¶

MetricManager(env: EnvBase, options: MetricManagerOptions)

Bases: FlatManagerBase[MetricManagerOptions]

Manager for computing metrics.

Every term's value lives in one (num_envs, num_terms) buffer (get_value returns its column), so reset(), which the auto-reset runs every step, handles all terms at once. The terms' success_threshold, direction, metric_mode, is_cumulative and contributes_to_success are read at construction and fixed from then on: configure them, never write them at runtime.

Source code in eden/managers/metric_manager.py

def __init__(self, env: EnvBase, options: MetricManagerOptions):
    super().__init__(env=env, options=options)
    # One (num_envs, num_terms) buffer, each term's `_step_metric` entry a column view of it (stable
    # identity, so `get_value` callers may still hold on to theirs). `reset()`, which the auto-reset
    # runs every step, then normalizes, logs, checks and clears every term at once rather than
    # issuing a dozen small ops per term.
    self._values = torch.zeros(self.num_envs, len(self._term_names), dtype=torch.float, device=self.device)
    self._step_metric = {name: self._values[:, i] for i, name in enumerate(self._term_names)}
    terms = [self._terms[name] for name in self._term_names]
    self._cumulative = torch.tensor([term.is_cumulative for term in terms], dtype=torch.bool, device=self.device)
    # The RESET-mode success checks. Terms keeping the default `is_success` are evaluated together
    # from their columns; one that overrides it keeps its own call.
    reset_success = [
        (i, term)
        for i, term in enumerate(terms)
        if term.metric_mode == MetricMode.RESET and term.contributes_to_success
    ]
    batched = [(i, term) for i, term in reset_success if type(term).is_success is SuccessCheckerMixin.is_success]
    self._reset_success_cols = torch.tensor([i for i, _ in batched], dtype=torch.long, device=self.device)
    self._reset_success_threshold = torch.tensor(
        [term.success_threshold for _, term in batched], dtype=torch.float, device=self.device
    )
    self._reset_success_hib = torch.tensor(
        [term.direction == MetricDirection.HIB for _, term in batched], dtype=torch.bool, device=self.device
    )
    batched_cols = {i for i, _ in batched}
    self._reset_success_custom = [(self._term_names[i], term) for i, term in reset_success if i not in batched_cols]

    self._success_buf = torch.zeros(self.num_envs, dtype=torch.bool, device=self.device)
    self._progress_buf = torch.zeros(self.num_envs, dtype=torch.int, device=self.device)
    # False for an env Genesis flagged as corrupt on the last `compute()` (see there).
    self._healthy_buf = torch.ones(self.num_envs, dtype=torch.bool, device=self.device)
    # Each env's last finished episode, as `reset()` logged it: the per-env rows its "Metrics/*" means average.
    self._last_episode_keys = [f"Metrics/{name}" for name in self._term_names]
    if self._num_success_terms > 0:
        self._last_episode_keys += ["Metrics/Success", "Metrics/Progress"]
    self._last_episode = torch.zeros(
        self.num_envs, len(self._last_episode_keys), dtype=torch.float, device=self.device
    )
    self._last_episode_healthy = torch.zeros(self.num_envs, dtype=torch.bool, device=self.device)

Attributes¶

success_buf `property` ¶

success_buf: Tensor

num_step_success_terms `property` ¶

num_step_success_terms: int

How many INTERVAL metrics contribute to :attr:success_buf — with none, it never turns True.

progress_buf `property` ¶

progress_buf: Tensor

Methods:¶

last_episode ¶

last_episode() -> tuple[dict[str, Tensor], Tensor]

Each env's last finished episode: its Metrics/<name> values (and Metrics/Success / Metrics/Progress when a term contributes to success), and whether Genesis left the env healthy.

reset() logs the mean of these rows over the envs that reset (the healthy ones); an evaluation that has to pick which episodes it counts (each env's first k, say) reads the rows instead. An env that has not finished an episode yet reads zeros, unhealthy.

RETURNS	DESCRIPTION
`tuple[dict[str, Tensor], Tensor]`	`(num_envs,)` values per key, and the `(num_envs,)` bool healthy mask. The manager's own buffers: read them before the next reset.

Source code in eden/managers/metric_manager.py

def last_episode(self) -> tuple[dict[str, torch.Tensor], torch.Tensor]:
    """Each env's last finished episode: its ``Metrics/<name>`` values (and ``Metrics/Success`` /
    ``Metrics/Progress`` when a term contributes to success), and whether Genesis left the env healthy.

    ``reset()`` logs the mean of these rows over the envs that reset (the healthy ones); an evaluation that has to
    pick *which* episodes it counts (each env's first ``k``, say) reads the rows instead. An env that has not
    finished an episode yet reads zeros, unhealthy.

    Returns
    -------
    tuple[dict[str, torch.Tensor], torch.Tensor]
        ``(num_envs,)`` values per key, and the ``(num_envs,)`` bool healthy mask. The manager's own buffers:
        read them before the next reset.
    """
    values = {key: self._last_episode[:, i] for i, key in enumerate(self._last_episode_keys)}
    return values, self._last_episode_healthy

get_value ¶

get_value(name: str) -> Tensor

Return the latest cached per-env value buffer for metric name.

The returned tensor is the manager-owned buffer updated in place every :meth:compute (stable identity), so callers — e.g. the metric_reward / metric_observation pull terms — may grab the reference once and reuse it. For a cumulative metric this is the running episode sum; otherwise the current-step value. metric_manager runs before the reward and observation managers in env.step(), so the value reflects the current step. Treat it as read-only — mutating it corrupts the metric.

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active metric term.

Source code in eden/managers/metric_manager.py

def get_value(self, name: str) -> torch.Tensor:
    """Return the latest cached per-env value buffer for metric ``name``.

    The returned tensor is the manager-owned buffer updated in place every
    :meth:`compute` (stable identity), so callers — e.g. the ``metric_reward`` /
    ``metric_observation`` pull terms — may grab the reference once and reuse it.
    For a cumulative metric this is the running episode sum; otherwise the
    current-step value. ``metric_manager`` runs before the reward and observation
    managers in ``env.step()``, so the value reflects the current step. Treat it
    as read-only — mutating it corrupts the metric.

    Raises
    ------
    KeyError
        If ``name`` is not an active metric term.
    """
    if name not in self._step_metric:
        raise KeyError(f"Metric '{name}' not found. Active metrics: {list(self._step_metric)}")
    return self._step_metric[name]

is_cumulative ¶

is_cumulative(name: str) -> bool

Whether metric name accumulates over the episode.

When True, :meth:get_value returns the running episode sum (reset per-env on episode reset); when False, the current-step value. Pull terms use this to reject a cumulative metric where a per-step value is expected (see metric_reward).

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active metric term.

Source code in eden/managers/metric_manager.py

def is_cumulative(self, name: str) -> bool:
    """Whether metric ``name`` accumulates over the episode.

    When True, :meth:`get_value` returns the running episode sum (reset per-env on
    episode reset); when False, the current-step value. Pull terms use this to reject
    a cumulative metric where a per-step value is expected (see ``metric_reward``).

    Raises
    ------
    KeyError
        If ``name`` is not an active metric term.
    """
    if name not in self._terms:
        raise KeyError(f"Metric '{name}' not found. Active metrics: {list(self._terms)}")
    return self._terms[name].is_cumulative

is_current_step_success ¶

is_current_step_success(name: str) -> Tensor

Per-env success mask for an INTERVAL metric evaluated on the current step's value.

Thin public accessor over the metric's :meth:is_success check applied to the current-step value buffer — used by the on_metric_success termination so it need not reach into the manager's private term/value dicts.

RAISES	DESCRIPTION
`KeyError`	If `name` is not an active metric term.
`ValueError`	If `name` is not an INTERVAL-mode metric (current-step success is undefined for cumulative/RESET-mode metrics).

Source code in eden/managers/metric_manager.py

def is_current_step_success(self, name: str) -> torch.Tensor:
    """Per-env success mask for an INTERVAL metric evaluated on the current step's value.

    Thin public accessor over the metric's :meth:`is_success` check applied to the
    current-step value buffer — used by the ``on_metric_success`` termination so it
    need not reach into the manager's private term/value dicts.

    Raises
    ------
    KeyError
        If ``name`` is not an active metric term.
    ValueError
        If ``name`` is not an INTERVAL-mode metric (current-step success is undefined
        for cumulative/RESET-mode metrics).
    """
    if name not in self._terms:
        raise KeyError(f"Metric '{name}' not found. Active metrics: {list(self._terms)}")
    term = self._terms[name]
    if term.metric_mode != MetricMode.INTERVAL:
        raise ValueError(f"is_current_step_success requires an INTERVAL metric; '{name}' is {term.metric_mode}.")
    if not term.contributes_to_success:
        raise ValueError(
            f"metric '{name}' is declared contributes_to_success=False — it is a shared-quantity term "
            f"with no success notion, so its current-step success is undefined. Gate on a metric that "
            f"does define success."
        )
    return term.is_success(self._step_metric[name]) & self._healthy_buf

summary ¶

summary() -> str

Source code in eden/managers/metric_manager.py

def summary(self) -> str:
    return self._format_summary_table(
        title="Metrics Terms",
        field_names=[
            "Index",
            "Name",
            "Success Threshold",
            "Direction",
            "Success Mode",
            "Cumulative",
            "Counts For Success",
        ],
        rows=(
            [
                index,
                name,
                term.success_threshold,
                term.direction,
                term.metric_mode,
                term.is_cumulative,
                term.contributes_to_success,
            ]
            for index, (name, term) in enumerate(self._terms.items())
        ),
        align={"Name": "l"},
    )

reset ¶

reset(envs_idx: Tensor | None = None) -> dict[str, Tensor]

Source code in eden/managers/metric_manager.py

@torch.inference_mode()
def reset(self, envs_idx: torch.Tensor | None = None) -> dict[str, torch.Tensor]:
    if envs_idx is None:
        envs_idx = slice(None)
    extras = {}
    # Every read below is over all envs and reduced / applied through `envs_idx`, never a gather of
    # the selected rows: on the auto-reset path `envs_idx` is the bool `reset_buf`, and each gather
    # with it is a GPU->CPU sync — three per term per step.
    # A cumulative metric logs, and is judged on, its per-step mean over the episode.
    num_healthy = sum_over_envs(self._healthy_buf, envs_idx).clamp(min=1)
    episode_length = torch.clamp_min(self._env.episode_length_buf, 1).unsqueeze(1)
    normalized = torch.where(self._cumulative, self._values / episode_length, self._values)
    for name, mean in zip(self._term_names, self._mean_over_healthy(normalized, envs_idx, num_healthy).unbind()):
        extras[f"Metrics/{name}"] = mean
    # NOTE: check if the episode maintained the success status on the reset — the default check
    # (`SuccessCheckerMixin.is_success`) for every term keeping it at once, then any override.
    if self._reset_success_cols.numel():
        judged = normalized.index_select(1, self._reset_success_cols)
        threshold = self._reset_success_threshold
        succeeded = torch.where(self._reset_success_hib, judged >= threshold, judged <= threshold)
        count = succeeded.sum(dim=1, dtype=self._progress_buf.dtype)
        assign_over_envs(self._progress_buf, envs_idx, self._progress_buf + count)
    for name, term in self._reset_success_custom:
        is_success = term.is_success(self._step_metric[name]).to(self._progress_buf.dtype)
        assign_over_envs(self._progress_buf, envs_idx, self._progress_buf + is_success)
    self._values[envs_idx] = 0.0

    # NOTE: compute the success and progress metrics based on the maintenance and attainment terms.
    # Only success-contributing terms count, and with none of them there is no task success to
    # report: omit the keys rather than log a 0/0 that reads as a real 0% success rate.
    if self._num_success_terms > 0:
        success_val = (self._progress_buf >= self._num_success_terms).float()
        progress_val = self._progress_buf / self._num_success_terms
        extras["Metrics/Success"] = self._mean_over_healthy(success_val, envs_idx, num_healthy)
        extras["Metrics/Progress"] = self._mean_over_healthy(progress_val, envs_idx, num_healthy)
        normalized = torch.cat([normalized, success_val[:, None], progress_val[:, None]], dim=1)
    assign_over_envs(self._last_episode, envs_idx, normalized)
    assign_over_envs(self._last_episode_healthy, envs_idx, self._healthy_buf)
    self._progress_buf[envs_idx] = 0
    self._success_buf[envs_idx] = 0.0
    for terms in self._reset_terms:
        terms.reset(envs_idx=envs_idx)
    return extras

compute ¶

compute() -> dict[str, Tensor]

Source code in eden/managers/metric_manager.py

@torch.inference_mode()
def compute(self) -> dict[str, torch.Tensor]:
    if len(self._terms) == 0:
        return self._step_metric
    self._progress_buf.zero_()
    self._success_buf.zero_()
    # An env Genesis flagged as corrupt read garbage this step, and is reset in it. Its value is 0 so
    # no NaN reaches a buffer or a cumulative sum, but 0 is the *best* value of a lower-is-better metric: it
    # is kept out of every success check here, in `is_current_step_success` and in `reset()`'s logs.
    torch.logical_not(self._env.get_solver_error_envs_mask(), out=self._healthy_buf)
    # One pass for `compute_once` (ManagerBase): terms share an intermediate within it.
    with self.compute_pass():
        for name, term in self._terms.items():
            value = torch.where(self._healthy_buf, term.compute().float(), 0.0)
            if term.is_cumulative:
                self._step_metric[name] += value
            else:
                self._step_metric[name][:] = value

            if term.metric_mode == MetricMode.INTERVAL and term.contributes_to_success:
                is_success = term.is_success(self._step_metric[name]) & self._healthy_buf
                self._progress_buf += is_success.long()
    if self._num_interval_terms > 0:
        self._success_buf[:] = self._progress_buf >= self._num_interval_terms
    return self._step_metric

Functions:¶

RecorderManager¶

Recorder manager and RecorderTerm base for episode data capture and export.

Attributes¶

RECORDER_TERM_REGISTRY `module-attribute` ¶

RECORDER_TERM_REGISTRY = TermRegistry(name='RECORDER_TERM', term_class=RecorderTerm, populated_by=('eden.managers.terms.recorders',))

Classes¶

RecorderTerm ¶

RecorderTerm(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[RecorderTermOptions]

Source code in eden/managers/base.py

def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Methods:¶

compute ¶

compute(*args, **kwargs)

RecorderTerm does not use compute(); recording is driven by record_* hooks.

Source code in eden/managers/recorder_manager.py

def compute(self, *args, **kwargs):
    """RecorderTerm does not use compute(); recording is driven by record_* hooks."""
    raise NotImplementedError("RecorderTerm uses record_* methods instead of compute().")

record_pre_reset ¶

record_pre_reset(envs_idx: Tensor | None = None) -> dict[str, Tensor | dict]

Record data at the beginning of env.reset() before reset is effective.

Source code in eden/managers/recorder_manager.py

def record_pre_reset(self, envs_idx: torch.Tensor | None = None) -> dict[str, torch.Tensor | dict]:
    """Record data at the beginning of env.reset() before reset is effective."""
    return {}

record_post_reset ¶

record_post_reset(envs_idx: Tensor | None = None) -> dict[str, Tensor | dict]

Record data at the end of env.reset().

Source code in eden/managers/recorder_manager.py

def record_post_reset(self, envs_idx: torch.Tensor | None = None) -> dict[str, torch.Tensor | dict]:
    """Record data at the end of env.reset()."""
    return {}

record_pre_step ¶

record_pre_step() -> dict[str, Tensor | dict]

Record data at the beginning of env.step() before all managers are processed.

Source code in eden/managers/recorder_manager.py

def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    """Record data at the beginning of env.step() before all managers are processed."""
    return {}

record_post_step ¶

record_post_step() -> dict[str, Tensor | dict]

Record data at the end of env.step() after rewards/terminations are computed.

Source code in eden/managers/recorder_manager.py

def record_post_step(self) -> dict[str, torch.Tensor | dict]:
    """Record data at the end of env.step() after rewards/terminations are computed."""
    return {}

close ¶

close(file_path: str)

Finalize and "clean up" the recorder term.

Source code in eden/managers/recorder_manager.py

def close(self, file_path: str):
    """Finalize and "clean up" the recorder term."""
    pass

RecorderManager ¶

RecorderManager(env: EnvBase, options: RecorderManagerOptions)

Bases: FlatManagerBase[RecorderManagerOptions]

Lifecycle orchestrator for transition-first recording.

Public methods expose recording/export control to users. Internal _on_* methods are called from env lifecycle events and translate those events into recorder-term collection, step closure, episode rotation, and export.

Source code in eden/managers/recorder_manager.py

def __init__(self, env: EnvBase, options: RecorderManagerOptions):
    super().__init__(env=env, options=options)

    self._recording = False
    self._all_env_ids = list(range(env.num_envs))

    self._episode_store = _EpisodeStore(env.num_envs)
    self._exporter = _EpisodeExporter(
        options,
        self._env.config._to_dict(),
    )

Attributes¶

is_recording `property` ¶

is_recording: bool

exported_successful_episode_count `property` ¶

exported_successful_episode_count: int

exported_failed_episode_count `property` ¶

exported_failed_episode_count: int

demo_count `property` ¶

demo_count: int

dataset_file_handler `property` ¶

dataset_file_handler

failed_episode_dataset_file_handler `property` ¶

failed_episode_dataset_file_handler

Methods:¶

summary ¶

summary() -> str

Source code in eden/managers/recorder_manager.py

def summary(self) -> str:
    if not self.active_terms:
        return ""

    msg = f"<RecorderManager> contains {len(self._term_names)} active terms.\n"
    msg += format_table(
        "Active Recorder Terms", ["Index", "Name"], enumerate(self._term_names), align={"Name": "l"}
    )
    msg += "\n"
    return msg

start_recording ¶

start_recording(reset_scene: bool = False) -> None

Source code in eden/managers/recorder_manager.py

def start_recording(self, reset_scene: bool = False) -> None:
    if not self.active_terms:
        en.logger.warning("RecorderManager.start_recording() ignored because there are no active recorder terms.")
        return
    if self._recording:
        en.logger.warning("RecorderManager.start_recording() ignored because recording is already active.")
        return

    self._exporter.prepare_for_recording()
    self._recording = True
    if reset_scene:
        self._env.reset()
    else:
        self._collect_post_reset_data(envs_idx=None)
    en.logger.info(f"Started recording. Current saved demos: {self.demo_count}.")

end_recording ¶

end_recording(*, export_unfinished: bool = True) -> None

Stop recording.

PARAMETER	DESCRIPTION
`export_unfinished`	Export the episodes still in progress, cut short where they stand (the default — a teleoperator stopping means "keep what I did"). Pass False to drop them instead, when only complete episodes belong in the dataset — a batch generator stopping once it has enough. TYPE: `bool` DEFAULT: `True`

Source code in eden/managers/recorder_manager.py

def end_recording(self, *, export_unfinished: bool = True) -> None:
    """Stop recording.

    Parameters
    ----------
    export_unfinished : bool
        Export the episodes still in progress, cut short where they stand (the default — a
        teleoperator stopping means "keep what I did"). Pass False to drop them instead, when only
        complete episodes belong in the dataset — a batch generator stopping once it has enough.
    """
    if not self.active_terms:
        return

    demos_before = self.demo_count
    self._recording = False
    if export_unfinished:
        self._export_closed_episodes(self._all_env_ids)
    self._start_new_episodes(self._all_env_ids)
    saved_demos = self.demo_count - demos_before
    en.logger.info(f"Stopped recording. Saved {saved_demos} demo(s). Total saved demos: {self.demo_count}.")

get_exported_successful_episode_count ¶

get_exported_successful_episode_count(env_idx: int) -> int

Source code in eden/managers/recorder_manager.py

def get_exported_successful_episode_count(self, env_idx: int) -> int:
    return self._exporter.get_exported_successful_episode_count(env_idx)

get_exported_failed_episode_count ¶

get_exported_failed_episode_count(env_idx: int) -> int

Source code in eden/managers/recorder_manager.py

def get_exported_failed_episode_count(self, env_idx: int) -> int:
    return self._exporter.get_exported_failed_episode_count(env_idx)

get_episode ¶

get_episode(env_idx: int) -> EpisodeData

Source code in eden/managers/recorder_manager.py

def get_episode(self, env_idx: int) -> EpisodeData:
    return self._episode_store.get_episode(env_idx)

load_episode ¶

load_episode(demo_name, device=None)

Source code in eden/managers/recorder_manager.py

def load_episode(self, demo_name, device=None):
    if not self.active_terms:
        raise RuntimeError("Cannot load episodes: no active recorder terms.")
    if device is None:
        device = self._env.device
    return self._exporter.load_episode(demo_name, device=device)

set_episode_success ¶

set_episode_success(envs_idx: Tensor | None, success_values: Tensor)

Source code in eden/managers/recorder_manager.py

def set_episode_success(self, envs_idx: torch.Tensor | None, success_values: torch.Tensor):
    if not self.active_terms:
        return
    env_ids = self._get_env_ids(envs_idx)
    self._episode_store.set_episode_success(env_ids, success_values)

close ¶

close()

Source code in eden/managers/recorder_manager.py

def close(self):
    if not self.active_terms:
        return
    if self._recording:
        self.end_recording()
    # Terms must close before the exporter so they can still interact with open file handles.
    file_path = os.path.join(self._options.dataset_export_dir_path, self._options.dataset_filename)
    for term in self._terms.values():
        term.close(file_path)
    self._exporter.close()

reset ¶

reset(envs_idx: Tensor | slice | None = None) -> dict[str, Any]

Fan the term lifecycle reset() out to every recorder term.

Episode bookkeeping does not live here: the env brackets a reset with :meth:_on_reset_started / :meth:_on_reset_finished, which close and open episodes around the whole _reset_idx sequence. This is the per-term hook every other manager's reset() already fans out, so a recorder term gets the same two-phase contract as any other term — reset() here, then post_reset() from :meth:~eden.managers.base.ManagerBase.post_reset. Returns no log extras.

Source code in eden/managers/recorder_manager.py

def reset(self, envs_idx: torch.Tensor | slice | None = None) -> dict[str, Any]:
    """Fan the term lifecycle ``reset()`` out to every recorder term.

    Episode bookkeeping does not live here: the env brackets a reset with
    :meth:`_on_reset_started` / :meth:`_on_reset_finished`, which close and open episodes
    around the whole ``_reset_idx`` sequence. This is the per-term hook every other
    manager's ``reset()`` already fans out, so a recorder term gets the same two-phase
    contract as any other term — ``reset()`` here, then ``post_reset()`` from
    :meth:`~eden.managers.base.ManagerBase.post_reset`. Returns no log extras.
    """
    for term in self._terms.values():
        term.reset(envs_idx=envs_idx)
    return {}

compute ¶

compute(*args, **kwargs)

Source code in eden/managers/recorder_manager.py

def compute(self, *args, **kwargs):
    raise NotImplementedError

Functions:¶

MotionManager¶

The motion manager of MimicRLEnvBase, with MotionTerm and MOTION_TERM_REGISTRY, is documented on the motion reference page together with the collaborators MotionTerm delegates to.

Reset-configuration sampling¶

Which reset configuration each env starts its next episode from, for tasks that reset from a bank of states.

Which reset configuration each env starts its next episode from.

A reset configuration is one row of a fixed, indexed set of episode starts the task can reset to: a stored simulator state (robot and object poses), a start state paired with a goal, a scene variant such as an object instance or a terrain patch. What a row means is the consumer's business; the sampler sees only indices. Two roles meet at reset time and this module is the seam between them:

the consumer -- a RESET event term that owns the set, knows its size, and turns an index into a reset (writes the state, places the robot, sets the goal);
the sampler -- whatever decides which index each resetting env gets. Uniform by default; an adaptive sampler (Success Guided Sampling, a curriculum term) when the task opts in.

They meet through :class:ResetConfigSampler. The consumer resolves its sampler in build() (:func:resolve_reset_config_sampler), tells it the set size once (:meth:ResetConfigSampler.bind), and asks for indices from its RESET compute() (:meth:ResetConfigSampler.sample).

Why a RESET event drives the sampler rather than CurriculumManager.reset: the scene and the RESET events run before any manager reset (EnvBase._reset_idx), so by the time a curriculum term's own reset() runs, the env has already been placed.

The assignment is returned as the full (num_envs,) buffer, not as the selected rows: a reset selector may be a bool mask, and compacting one to its selected rows is a GPU->CPU sync on every step that resets anything. Consumers index the buffer with the same envs_idx they were handed.

Classes¶

ResetConfigSampler ¶

Bases: ABC

Chooses a reset-configuration index for every env being reset.

Subclassed by :class:UniformResetConfigSampler and by adaptive samplers, which are curriculum terms (class SuccessGuidedSampling(CurriculumTerm, ResetConfigSampler)). A consumer calls :meth:bind exactly once from its build(), then :meth:sample from every RESET compute().

The base owns everything every sampler shares: bind validation, the (num_envs,) assignment buffer, which envs' next reset is a first spawn (:attr:first_spawn), and turning a reset selector into a mask. A subclass implements :meth:_draw -- one proposed configuration per env, of which only the selected rows are kept -- and may allocate per-configuration state in :meth:_on_bind. It provides num_envs and device, which a manager term already has.

The base also owns the checkpoint: :meth:state_dict / :meth:load_state_dict record and check the size of the configuration set, and a subclass with state of its own adds it in :meth:_sampler_state / :meth:_load_sampler_state. Every load ends with every env's next reset a first spawn, whatever the subclass: the episodes running at the load started from the pre-load state, and grading them into the loaded one would mix the two. Both methods are final: a subclass that overrides either one, or whose MRO resolves either one to another class first (a default added to a curriculum-term base, say), raises TypeError when the class is defined, instead of silently skipping the size check, the first-spawn reset or the sampler's own state.

The bound state lives in class-level defaults rather than in an __init__: a curriculum term's __init__ chain calls its bases explicitly and never reaches this one.

Attributes¶

num_envs `abstractmethod` `property` ¶

num_envs: int

The number of parallel environments.

device `abstractmethod` `property` ¶

device: device | str

The torch device the assignment lives on.

num_configs `property` ¶

num_configs: int

Size of the bound configuration set.

assignment `property` ¶

assignment: Tensor

(num_envs,) long: the configuration each env's current episode started from.

first_spawn `property` ¶

first_spawn: Tensor

(num_envs,) bool: envs whose next reset finishes no episode this sampler should learn from.

True for every env until its first :meth:sample, and again for every env after a :meth:load_state_dict. Where it is False, :attr:assignment is the configuration the env's running episode started from, drawn by this sampler's current state.

Methods:¶

bind ¶

bind(num_configs: int, consumer: str) -> None

Fix the size of the configuration set this sampler draws from, then run :meth:_on_bind.

consumer labels the binding term in error messages, e.g. ResetFromStateBuffer(path='…'). A sampler bound twice raises: two consumers sharing one sampler would draw indices into two different sets while the sampler's statistics conflate them.

Source code in eden/managers/resets/sampling.py

def bind(self, num_configs: int, consumer: str) -> None:
    """Fix the size of the configuration set this sampler draws from, then run :meth:`_on_bind`.

    ``consumer`` labels the binding term in error messages, e.g. ``ResetFromStateBuffer(path='…')``.
    A sampler bound twice raises: two consumers sharing one sampler would draw indices into two
    different sets while the sampler's statistics conflate them.
    """
    label = type(self).__name__
    if self._consumer is not None:
        raise RuntimeError(
            f"{label} is already bound to reset consumer '{self._consumer}'; '{consumer}' tried to bind it "
            "too. One sampler serves one configuration set -- give each consumer its own sampler term."
        )
    if not isinstance(num_configs, int) or isinstance(num_configs, bool) or num_configs < 1:
        raise ValueError(
            f"{label}.bind(): '{consumer}' passed num_configs={num_configs!r}; expected a positive int "
            "(the number of rows in its reset-configuration set)."
        )
    self._num_configs = num_configs
    self._consumer = consumer
    # Normal tensors, not inference tensors: sample() runs both inside and outside torch.inference_mode.
    with torch.inference_mode(False):
        self._assignment = torch.zeros(self.num_envs, dtype=torch.long, device=self.device)
        self._first_spawn = torch.ones(self.num_envs, dtype=torch.bool, device=self.device)
        self._selected = torch.zeros(self.num_envs, dtype=torch.bool, device=self.device)
    self._on_bind(num_configs)

sample ¶

sample(envs_idx: slice | Tensor) -> Tensor

Assign a configuration to each env in envs_idx and return the assignment buffer.

envs_idx is a sanitized selector (eden.utils.misc.sanitize_envs_idx): a slice, a long index tensor, or a (num_envs,) bool mask. Returns the sampler's (num_envs,) long buffer of current assignments, with only the envs_idx rows changed by this call. The buffer is live -- the consumer must index it now, not keep it for later. The selected envs leave :attr:first_spawn after the draw, so :meth:_draw still sees which are first spawns.

Source code in eden/managers/resets/sampling.py

def sample(self, envs_idx: slice | torch.Tensor) -> torch.Tensor:
    """Assign a configuration to each env in ``envs_idx`` and return the assignment buffer.

    ``envs_idx`` is a sanitized selector (``eden.utils.misc.sanitize_envs_idx``): a slice, a
    long index tensor, or a ``(num_envs,)`` bool mask. Returns the sampler's ``(num_envs,)``
    long buffer of current assignments, with only the ``envs_idx`` rows changed by this call.
    The buffer is live -- the consumer must index it now, not keep it for later. The selected
    envs leave :attr:`first_spawn` after the draw, so :meth:`_draw` still sees which are first spawns.
    """
    self._require_bound("sample()")
    selected = self._selection_mask(envs_idx)
    # Proposed for every env and blended in under the mask, so no selector costs a GPU->CPU sync.
    torch.where(selected, self._draw(selected), self._assignment, out=self._assignment)
    self._first_spawn.logical_and_(selected.logical_not())
    return self._assignment

state_dict ¶

state_dict() -> dict

The sampler's checkpoint: the size of the set it was bound to, plus :meth:_sampler_state.

Source code in eden/managers/resets/sampling.py

def state_dict(self) -> dict:
    """The sampler's checkpoint: the size of the set it was bound to, plus :meth:`_sampler_state`."""
    self._require_bound("state_dict()")
    return {"num_configs": self._num_configs, **self._sampler_state()}

load_state_dict ¶

load_state_dict(state: dict) -> None

Restore a :meth:state_dict, then make every env's next reset a first spawn.

Raises if the checkpoint was saved for a configuration set of another size: its per-configuration state would index configurations it was never recorded on. The episodes running now started from the pre-load state, so none of them is graded into the loaded one.

Source code in eden/managers/resets/sampling.py

def load_state_dict(self, state: dict) -> None:
    """Restore a :meth:`state_dict`, then make every env's next reset a first spawn.

    Raises if the checkpoint was saved for a configuration set of another size: its per-configuration state
    would index configurations it was never recorded on. The episodes running now started from the pre-load
    state, so none of them is graded into the loaded one.
    """
    self._require_bound("load_state_dict()")
    saved = state.get("num_configs")
    if saved != self._num_configs:
        raise ValueError(
            f"{type(self).__name__}: the checkpoint was saved for a set of {saved} reset configurations, but "
            f"{self._consumer} bound this one to {self._num_configs}. Fix: resume with the task the checkpoint "
            "was trained on. Do NOT drop this term's checkpoint state to get past it: its statistics would be "
            "read against configurations they were never recorded on."
        )
    self._load_sampler_state(state)
    self._first_spawn.fill_(True)

UniformResetConfigSampler ¶

UniformResetConfigSampler(num_envs: int, device: device | str)

Bases: ResetConfigSampler

Draws every reset configuration with equal probability -- the default when a task opts out.

Not a manager term: a consumer with no sampler configured constructs one of these itself, so the uniform path has no config entry, no logging and no checkpoint state. Draws from the global torch RNG, like every other reset-time randomization, so seeding the run seeds it.

Source code in eden/managers/resets/sampling.py

def __init__(self, num_envs: int, device: torch.device | str) -> None:
    self._num_envs = num_envs
    self._device = device

Attributes¶

num_envs `property` ¶

num_envs: int

device `property` ¶

device: device | str

Methods:¶

CategoryWeightedSampler ¶

CategoryWeightedSampler(num_envs: int, device: device | str, category: Tensor, weights: Tensor)

Bases: ResetConfigSampler

Draws a category by weight, then a configuration uniformly within it (OmniReset's reset-family mix).

category labels each configuration ((num_configs,) long, values index weights); a category with weight 0 is never drawn, and its configurations with it. Unlike uniform sampling over the whole set, a category's share does not grow with its size. Like :class:UniformResetConfigSampler, a consumer constructs it itself and it draws from the global torch RNG. Sync-free: the configurations are sorted by category once, at bind, and again only on :meth:relabel.

Source code in eden/managers/resets/sampling.py

def __init__(
    self, num_envs: int, device: torch.device | str, category: torch.Tensor, weights: torch.Tensor
) -> None:
    self._num_envs = num_envs
    self._device = device
    self._weights = weights.to(device=device, dtype=torch.float32)
    if (
        self._weights.ndim != 1
        or not torch.isfinite(self._weights).all()
        or (self._weights < 0).any()
        or not (self._weights > 0).any()
    ):
        raise ValueError(
            f"CategoryWeightedSampler weights must be finite and non-negative with at least one positive, got "
            f"{self._weights.tolist()}."
        )
    self._label(category)

Attributes¶

num_envs `property` ¶

num_envs: int

device `property` ¶

device: device | str

Methods:¶

relabel ¶

relabel(category: Tensor) -> None

Take new labels for the same configurations, e.g. after ResetStateBuffer.write_rows changed some rows'.

Re-checks the labels and re-sorts the configurations by them: a few device-to-host reads, so call it when the labels change, not on every reset.

Source code in eden/managers/resets/sampling.py

def relabel(self, category: torch.Tensor) -> None:
    """Take new labels for the same configurations, e.g. after ``ResetStateBuffer.write_rows`` changed some rows'.

    Re-checks the labels and re-sorts the configurations by them: a few device-to-host reads, so call it when the
    labels change, not on every reset.
    """
    self._label(category)
    if self._num_configs is not None:
        self._on_bind(self._num_configs)

Functions:¶

resolve_reset_config_sampler ¶

resolve_reset_config_sampler(env: EnvBase, sampler_name: str | None, consumer: str) -> ResetConfigSampler

The sampler a reset consumer named, or a fresh uniform one when it named none.

sampler_name is a curriculum term's config key. Resolved from the consumer's build(): the curriculum manager is constructed before the event manager, so the term already exists, and a wrong name raises at env construction rather than on the first reset.

Source code in eden/managers/resets/sampling.py

def resolve_reset_config_sampler(env: EnvBase, sampler_name: str | None, consumer: str) -> ResetConfigSampler:
    """The sampler a reset consumer named, or a fresh uniform one when it named none.

    ``sampler_name`` is a curriculum term's config key. Resolved from the consumer's ``build()``:
    the curriculum manager is constructed before the event manager, so the term already exists, and
    a wrong name raises at env construction rather than on the first reset.
    """
    if sampler_name is None:
        return UniformResetConfigSampler(env.num_envs, env.device)
    manager = getattr(env, "curriculum_manager", None)  # absent on a plain EnvBase
    if manager is None:
        raise ValueError(
            f"Reset consumer '{consumer}' names sampler='{sampler_name}', but this env has no curriculum "
            f"manager. Fix: add '{sampler_name}' to curriculum_options, or drop sampler= to sample "
            "configurations uniformly."
        )
    term = manager.get_term(sampler_name)  # KeyError listing the declared terms on a typo
    if not isinstance(term, ResetConfigSampler):
        raise TypeError(
            f"Reset consumer '{consumer}' names sampler='{sampler_name}', but that curriculum term is a "
            f"{type(term).__name__}, which is not a ResetConfigSampler subclass. "
            "Fix: point sampler= at an adaptive reset sampler such as SuccessGuidedSampling."
        )
    return term

Term catalog¶

The machinery behind eden terms and the generated term catalog.

A catalog of every registered term: the parameters it accepts, their types, and their defaults.

Without it, a term's accepted parameter set exists in exactly one place — the function signature or class fields in the source — so the only way to learn that dofs_vel_limits takes entity_name and max_vel is to open eden/managers/terms/rewards/common.py. Measured on autonomous coding agents without a catalog, a capable one paid for that with a median 29-62 reads of Eden's installed source per task against 2-3 documentation reads; a weaker one skipped the reading and guessed until its wall clock ran out.

Nothing here is new information: the registries already enumerate the terms and get_parameter_specs / get_function_parameter_specs already extract the parameters. This module emits them, for three consumers:

eden terms (:mod:eden.utils.cli.terms) — in the terminal, --json for tooling.
two generated pages, both written by python -m eden.managers.catalog (:func:main), checked in, and verified current by tests/resources/test_term_catalog.py:
eden/resources/term_catalog.md (:func:render_index) ships inside the package, because an agent working against an installed Eden has eden/ but not docs/. It lists every term by name, form, defining module and summary, and no parameters: an agent reads a page it finds whole and keeps it in context for the rest of its session, so this page is sized to be read whole. A term's parameters are one eden terms --grep <name> away.
docs/api/terms/catalog.md (:func:render_markdown) is the full catalog, parameters included, for the docs site. It is what eden terms --markdown prints.
the same ParamInfo rendering the configure() errors use (:mod:eden.utils.params), so the catalog and the error that fires when you ignore it describe a term identically.

:data:KINDS is a curated in-tree table rather than a discovered one — the set of term registries is closed and each entry carries wiring (the import alias, the wrapper class) that no registry knows about. tests/managers/test_catalog.py asserts it covers every :class:~eden.utils.registry.TermRegistry in the tree, so adding one and forgetting the catalog fails a test instead of quietly shipping a term nobody can find.

Attributes¶

KINDS `module-attribute` ¶

KINDS: tuple[TermKind, ...] = (TermKind('action', 'eden.managers', 'ACTION_TERM_REGISTRY', 'actions'), TermKind('observation', 'eden.managers', 'OBSERVATION_TERM_REGISTRY', 'observations', 'ObsTerm'), TermKind('reward', 'eden.managers', 'REWARD_TERM_REGISTRY', 'rewards', 'RewardTerm'), TermKind('termination', 'eden.managers', 'TERMINATION_TERM_REGISTRY', 'terminations', 'DoneTerm'), TermKind('command', 'eden.managers', 'COMMAND_TERM_REGISTRY', 'commands'), TermKind('event', 'eden.managers', 'EVENT_TERM_REGISTRY', 'events', 'EventTerm'), TermKind('curriculum', 'eden.managers', 'CURRICULUM_TERM_REGISTRY', 'curricula', 'CurrTerm'), TermKind('metric', 'eden.managers', 'METRIC_TERM_REGISTRY', 'metrics', 'MetricTerm'), TermKind('recorder', 'eden.managers', 'RECORDER_TERM_REGISTRY', 'recorders'), TermKind('motion', 'eden.managers.motion_manager', 'MOTION_TERM_REGISTRY'), TermKind('noise_model', 'eden.managers.modifiers.base', 'NOISE_MODEL_REGISTRY'), TermKind('observation_modifier', 'eden.managers.modifiers.base', 'OBSERVATION_MODIFIER_REGISTRY'), TermKind('action_modifier', 'eden.managers.modifiers.base', 'ACTION_MODIFIER_REGISTRY'))

Classes¶

TermKind `dataclass` ¶

TermKind(name: str, module: str, registry_attr: str, terms_module: str | None = None, wrapper: str | None = None)

One term registry, plus how a config reaches the terms inside it.

module/registry_attr are resolved lazily (:meth:registry) so importing this module costs only what the caller already imported — eden.managers does not pull the motion manager or the modifiers, and listing rewards should not either.

Attributes¶

name `instance-attribute` ¶

name: str

module `instance-attribute` ¶

module: str

registry_attr `instance-attribute` ¶

registry_attr: str

terms_module `class-attribute` `instance-attribute` ¶

terms_module: str | None = None

wrapper `class-attribute` `instance-attribute` ¶

wrapper: str | None = None

Methods:¶

registry ¶

registry() -> TermRegistry

This kind's registry, which fills itself on the first read.

Which modules have to be imported for a registry to hold its terms is the registry's own business (populated_by, see :meth:~eden.utils.registry.Registry._populate), so it is not declared here a second time.

Source code in eden/managers/catalog.py

def registry(self) -> TermRegistry:
    """This kind's registry, which fills itself on the first read.

    Which modules have to be imported for a registry to hold its terms is the registry's own
    business (``populated_by``, see :meth:`~eden.utils.registry.Registry._populate`), so it is
    not declared here a second time.
    """
    return getattr(importlib.import_module(self.module), self.registry_attr)

TermEntry `dataclass` ¶

TermEntry(name: str, kind: str, form: str, qualname: str, module: str, source: str, summary: str, params: tuple[ParamInfo, ...], requires_env_options: tuple[str, ...] = ())

One registered term, described the way an author has to write it.

Attributes¶

name `instance-attribute` ¶

name: str

kind `instance-attribute` ¶

kind: str

form `instance-attribute` ¶

form: str

qualname `instance-attribute` ¶

qualname: str

module `instance-attribute` ¶

module: str

source `instance-attribute` ¶

source: str

summary `instance-attribute` ¶

summary: str

params `instance-attribute` ¶

params: tuple[ParamInfo, ...]

requires_env_options `class-attribute` `instance-attribute` ¶

requires_env_options: tuple[str, ...] = ()

haystack `property` ¶

haystack: str

Everything eden terms --grep searches, as one lowercase line.

Name, kind, class/function name, defining module, summary and parameter names and types — because "which term takes max_vel?" and "is there an observation for base height?" are the two questions that send a reader to the full catalog, and only one of them is answerable from term names alone.

Methods:¶

to_dict ¶

to_dict() -> dict[str, Any]

A JSON-serializable form, with each param expanded — what eden terms --json emits.

Source code in eden/managers/catalog.py

def to_dict(self) -> dict[str, Any]:
    """A JSON-serializable form, with each param expanded — what ``eden terms --json`` emits."""
    return {
        "name": self.name,
        "kind": self.kind,
        "form": self.form,
        "qualname": self.qualname,
        "module": self.module,
        "source": self.source,
        "summary": self.summary,
        "params": [
            {
                "name": param.name,
                "type": param.annotation,
                "default": param.default,
                "required": param.required,
            }
            for param in self.params
        ],
        "required": [param.name for param in self.params if param.required],
        "requires_env_options": list(self.requires_env_options),
    }

Functions:¶

builtin_catalog ¶

builtin_catalog(kind: str | None = None) -> tuple[TermEntry, ...]

:func:catalog restricted to the terms Eden itself ships — what the generated page holds.

A term registry is a global, open table: an out-of-tree term registers into the same one, and eden terms listing it is the point. The checked-in page must not depend on which of those happen to be imported in the process that generated it, or it would read as stale as often as it is right (the test suite alone registers a dozen throwaway terms).

"Shipped" is the eden.managers namespace — the term modules each registry names in its populated_by. eden.tasks.* / eden.extensions.* terms register when their task module is imported and would otherwise leak in by import order — they match a plain startswith("eden.") but not the eden.managers prefix tested here.

Source code in eden/managers/catalog.py

def builtin_catalog(kind: str | None = None) -> tuple[TermEntry, ...]:
    """:func:`catalog` restricted to the terms Eden itself ships — what the generated page holds.

    A term registry is a global, open table: an out-of-tree term registers into the same one, and
    ``eden terms`` listing it is the point. The *checked-in* page must not depend on which of those
    happen to be imported in the process that generated it, or it would read as stale as often as it
    is right (the test suite alone registers a dozen throwaway terms).

    "Shipped" is the ``eden.managers`` namespace — the term modules each registry names in its
    ``populated_by``. ``eden.tasks.*`` / ``eden.extensions.*`` terms register when their task
    module is imported and would otherwise leak in by import order — they match a plain
    ``startswith("eden.")`` but not the ``eden.managers`` prefix tested here.
    """
    return tuple(
        entry for entry in catalog(kind) if entry.module == "eden.managers" or entry.module.startswith("eden.managers.")
    )

render_markdown ¶

render_markdown(entries: Sequence[TermEntry], grep: str | None = None) -> str

entries as the full catalog page, parameters included: docs/api/terms/catalog.md.

What eden terms --markdown prints. The page that ships in the wheel is :func:render_index.

One row per term, so a grep for a parameter name lands on a line that also carries the term name, its form and its summary — the shape that answers "which term takes max_vel?" in one hit. Rows are grouped by defining module, which doubles as the source pointer and keeps the Source column (and its width) off every row.

PARAMETER	DESCRIPTION
`entries`	The entries to render, in catalog order (grouped by kind). TYPE: `Sequence[TermEntry]`
`grep`	The `--grep` pattern `entries` was filtered by, when it was. The intro otherwise claims to be the complete list, which a filtered page is not. TYPE: `str \| None` DEFAULT: `None`

Source code in eden/managers/catalog.py

def render_markdown(entries: Sequence[TermEntry], grep: str | None = None) -> str:
    """``entries`` as the full catalog page, parameters included: ``docs/api/terms/catalog.md``.

    What ``eden terms --markdown`` prints. The page that ships in the wheel is :func:`render_index`.

    One row per term, so a grep for a parameter name lands on a line that also carries the term
    name, its form and its summary — the shape that answers "which term takes ``max_vel``?" in one
    hit. Rows are grouped by defining module, which doubles as the source pointer and keeps the
    ``Source`` column (and its width) off every row.

    Parameters
    ----------
    entries : Sequence[TermEntry]
        The entries to render, in catalog order (grouped by kind).
    grep : str | None
        The ``--grep`` pattern ``entries`` was filtered by, when it was. The intro otherwise
        claims to be the complete list, which a filtered page is not.
    """
    # Derive the "complete list" claim from the entries rather than a flag, so the header can never
    # claim more than it shows: a single-kind view (``eden terms --kind reward --markdown``) says so,
    # while the full catalog — which spans every kind — keeps its original wording byte-for-byte
    # (the checked-in ``catalog.md`` parity test pins it).
    kinds_present = {entry.kind for entry in entries}
    only_kind = next(iter(kinds_present)) if len(kinds_present) == 1 else None
    if grep is not None:
        scope = f"{only_kind} term" if only_kind else "term"
        intro = (
            f"Every {scope} registered in Eden whose name, parameters or summary match `{grep}` "
            f"({len(entries)} of them) — a filtered view, not the whole catalog (`eden terms --markdown` "
            "emits every kind). The guide's [Terms & Registries](../../guides/terms.md) page explains how "
            "to write and wire one up."
        )
    elif only_kind is None:
        intro = (
            f"Every term registered in Eden ({len(entries)} of them) with the parameters it accepts, generated "
            "from the registries themselves. This is the complete list; the guide's "
            "[Terms & Registries](../../guides/terms.md) page explains how to write and wire one up."
        )
    else:
        intro = (
            f"Every {only_kind} term registered in Eden ({len(entries)} of them) with the parameters it accepts, "
            f"generated from the registries themselves — the {only_kind} section only, not the whole catalog "
            "(`eden terms --markdown` emits every kind). The guide's "
            "[Terms & Registries](../../guides/terms.md) page explains how to write and wire one up."
        )
    lines = [
        "<!-- Generated by `python -m eden.managers.catalog`. Do not edit by hand. -->",
        "",
        "# Term Catalog",
        "",
        intro,
        "",
        "- **Bold** parameters are required — omit one and `configure()` raises before the env is built.",
        (
            "- A term's parameters are the keys of its `params={...}` dict (function terms) or its "
            "`configure()` fields (class terms). The fields *every* term of a kind carries are listed "
            "once, under that kind."
        ),
        (
            "- A **Requires** column appears in the tables that have one: an `EnvOptions` flag the term "
            "needs set, which no parameter of the term can supply. The env build raises without it."
        ),
        (
            "- `eden terms` prints the same catalog in a terminal, `eden terms --kind reward` one section "
            "of it, `eden terms --grep height` only the terms whose name, parameters or summary match, and "
            "`eden terms --json` emits it as JSON."
        ),
        "",
    ]
    for kind, kind_entries in _by_kind(entries):
        lines += [f"## {kind.name} terms", "", f"{len(kind_entries)} registered.", ""]
        for recipe in _configure_recipes(kind, kind_entries):
            lines.append(f"- {recipe}")
        shared = kind_shared_specs(kind)
        if shared:
            lines.append(f"- Every {kind.name} term also accepts: {_markdown_params(shared)}")
        lines.append("")
        for module, module_entries in _by_module(kind_entries):
            # The ``Requires`` column only appears in the tables that have one. A fifth column of
            # "—" on all ~200 rows would cost every reader width to tell nearly all of them
            # nothing; the header is per-module already, so a module that has preconditions can
            # carry the column without the rest of the catalog paying for it.
            has_requires = any(entry.requires_env_options for entry in module_entries)
            header = "| Term | Form | Parameters |" + (" Requires |" if has_requires else "") + " Summary |"
            lines += [
                f"### `{module}`",
                "",
                header,
                "|---|---|---|---|---|" if has_requires else "|---|---|---|---|",
            ]
            for entry in module_entries:
                form = "function" if entry.form == "function" else f"class `{entry.qualname}`"
                summary = _markdown_cell(entry.summary) or "—"
                requires = f" {_markdown_requires(entry)} |" if has_requires else ""
                lines.append(f"| `{entry.name}` | {form} | {_markdown_params(entry.params)} |{requires} {summary} |")
            lines.append("")
    return "\n".join(lines).rstrip() + "\n"

render_index ¶

render_index(entries: Sequence[TermEntry]) -> str

entries as the shipped eden/resources/term_catalog.md index: every term, no parameters.

What :func:render_markdown carries minus its Parameters column, which was most of that page: grouped by kind and then by defining module, one line per term with its registry key, the class to import for a class term, and its summary; each kind keeps its configure recipe and the names of the fields all its terms share. Module headings are fully qualified, because a class term's recipe says to import it from the module its section names. An agent that finds a catalog in the installed package reads it whole and carries it for the rest of its session, so this page answers "is there a term for X, and where is it defined?" and leaves "what does it take?" to eden terms --grep <name>.

PARAMETER	DESCRIPTION
`entries`	The entries to render, in catalog order (grouped by kind). TYPE: `Sequence[TermEntry]`

Source code in eden/managers/catalog.py

def render_index(entries: Sequence[TermEntry]) -> str:
    """``entries`` as the shipped ``eden/resources/term_catalog.md`` index: every term, no parameters.

    What :func:`render_markdown` carries minus its ``Parameters`` column, which was most of that page:
    grouped by kind and then by defining module, one line per term with its registry key, the class
    to import for a class term, and its summary; each kind keeps its configure recipe and the names of
    the fields all its terms share. Module headings are fully qualified, because a class term's recipe
    says to import it from the module its section names. An agent that finds a catalog in the installed
    package reads it whole and carries it for the rest of its session, so this page answers "is there a
    term for X, and where is it defined?" and leaves "what does it take?" to ``eden terms --grep <name>``.

    Parameters
    ----------
    entries : Sequence[TermEntry]
        The entries to render, in catalog order (grouped by kind).
    """
    lines = [
        "<!-- Generated by `python -m eden.managers.catalog`. Do not edit by hand. -->",
        "",
        "# Term Catalog",
        "",
        (
            f"Every term registered in Eden ({len(entries)} of them): registry key, class name for a class "
            "term, and summary, under the module that defines it. Parameters are not listed — "
            "`eden terms --grep <name>` (or `python -m eden terms --grep <name>`) prints a term's "
            "parameters, types and defaults, and marks the required ones."
        ),
        "",
    ]
    for kind, kind_entries in _by_kind(entries):
        lines += [f"## {kind.name} terms ({len(kind_entries)})", ""]
        for recipe in _configure_recipes(kind, kind_entries):
            lines.append(f"- {recipe}")
        shared = kind_shared_specs(kind)
        if shared:
            # Names only, like the rest of the page: the types and defaults are parameters too.
            lines.append(f"- Every {kind.name} term also accepts: {', '.join(f'`{p.name}`' for p in shared)}")
        lines.append("")
        for module, module_entries in _by_module(kind_entries):
            lines.append(f"`{module}`:")
            for entry in module_entries:
                cls = f" (`{entry.qualname}`)" if entry.form != "function" else ""
                # Kept inline: a precondition no parameter can supply, so leaving it out would let a
                # config written from this page fail at build.
                # A bullet list, not a table: no cell escaping, so a summary's `|x|` reads as written.
                requires = f" Requires `{_render_requires(entry)}`." if entry.requires_env_options else ""
                lines.append(f"- `{entry.name}`{cls} — {entry.summary or '—'}{requires}")
            lines.append("")
    return "\n".join(lines).rstrip() + "\n"

main ¶

main(argv: Sequence[str] | None = None) -> int

Regenerate both checked-in catalog pages from the registries (--check: report, don't write).

Run it from a repo checkout after adding, renaming or re-parameterizing a term; tests/resources/test_term_catalog.py fails while either page is stale::

uv run python -m eden.managers.catalog          # write both pages
uv run python -m eden.managers.catalog --check  # exit 1 if either would change

Source code in eden/managers/catalog.py

def main(argv: Sequence[str] | None = None) -> int:
    """Regenerate both checked-in catalog pages from the registries (``--check``: report, don't write).

    Run it from a repo checkout after adding, renaming or re-parameterizing a term;
    ``tests/resources/test_term_catalog.py`` fails while either page is stale::

        uv run python -m eden.managers.catalog          # write both pages
        uv run python -m eden.managers.catalog --check  # exit 1 if either would change
    """
    import argparse

    parser = argparse.ArgumentParser(prog="python -m eden.managers.catalog", description=main.__doc__.splitlines()[0])
    parser.add_argument("--check", action="store_true", help="report staleness instead of writing")
    args = parser.parse_args(argv)

    # REPO_ROOT is `parents[2]` of this file, which in an installed eden is the site-packages parent: the
    # write below would `mkdir -p` a `docs/` there and succeed silently.
    if not ((REPO_ROOT / "pyproject.toml").is_file() and (REPO_ROOT / "docs").is_dir()):
        print(
            f"python -m eden.managers.catalog regenerates pages checked into an Eden repo checkout, but the eden "
            f"it imported ({Path(__file__).resolve()}) is not in one: '{REPO_ROOT}' lacks pyproject.toml or "
            "docs/. Run it from a clone (`git clone https://github.com/embodied-ai-nexus/Eden && cd Eden && "
            "uv run python -m eden.managers.catalog`). An installed eden needs no regeneration: it ships a current "
            "eden/resources/term_catalog.md, and `eden terms --markdown` prints the full catalog.",
            file=sys.stderr,
        )
        return 2

    entries = builtin_catalog()
    pages = ((CATALOG_PAGE, render_index(entries)), (DOCS_CATALOG_PAGE, render_markdown(entries)))
    stale = [(page, rendered) for page, rendered in pages if not page.exists() or page.read_text() != rendered]

    if args.check:
        for page, _ in stale:
            print(f"{page.relative_to(REPO_ROOT)} is stale — run: uv run python -m eden.managers.catalog")
        if not stale:
            print(f"both catalog pages are up to date ({len(entries)} terms)")
        return 1 if stale else 0

    for page, rendered in stale:
        page.parent.mkdir(parents=True, exist_ok=True)
        page.write_text(rendered)
        print(f"wrote {page.relative_to(REPO_ROOT)} ({len(rendered.encode()):,} bytes)")
    if not stale:
        print(f"both catalog pages already up to date ({len(entries)} terms)")
    return 0

Registries¶

Each term kind has a registry. A term registers under to_snake_case(<ClassOrFunctionName>); register() takes no name=.

Registry	Import
`ACTION_TERM_REGISTRY`	`from eden.managers import ACTION_TERM_REGISTRY`
`OBSERVATION_TERM_REGISTRY`	`from eden.managers import OBSERVATION_TERM_REGISTRY`
`REWARD_TERM_REGISTRY`	`from eden.managers import REWARD_TERM_REGISTRY`
`TERMINATION_TERM_REGISTRY`	`from eden.managers import TERMINATION_TERM_REGISTRY`
`COMMAND_TERM_REGISTRY`	`from eden.managers import COMMAND_TERM_REGISTRY`
`EVENT_TERM_REGISTRY`	`from eden.managers import EVENT_TERM_REGISTRY`
`CURRICULUM_TERM_REGISTRY`	`from eden.managers import CURRICULUM_TERM_REGISTRY`
`METRIC_TERM_REGISTRY`	`from eden.managers import METRIC_TERM_REGISTRY`
`RECORDER_TERM_REGISTRY`	`from eden.managers import RECORDER_TERM_REGISTRY`
`MOTION_TERM_REGISTRY`	`from eden.managers.motion_manager import MOTION_TERM_REGISTRY`

The registry classes themselves are documented under core utilities.

Managers¶

Base classes¶

Attributes¶

T module-attribute ¶

Classes¶

ManagerTermBase ¶

Attributes¶

num_envs property ¶

device property ¶

Methods:¶

resolve_entity ¶

reset ¶

post_reset ¶

compute abstractmethod ¶

build ¶

ManagerTermFuncWrapperBase ¶

Methods:¶

compute ¶

ConfigurableTermBase ¶

Methods:¶

ConfigurableTermFuncWrapper ¶

Methods:¶

ManagerBase ¶

Attributes¶

device_synced_fields class-attribute ¶

num_envs property ¶

device property ¶

Methods:¶

compute_once ¶

compute_pass ¶

sync_device_fields ¶

reset ¶

compute abstractmethod ¶

iter_post_reset_terms abstractmethod ¶

post_reset ¶

FlatManagerBase ¶

Attributes¶

terms property ¶

active_terms property ¶

Methods:¶

get_term ¶

get_term_options ¶

iter_post_reset_terms ¶

Functions:¶

build_term ¶

ActionManager¶

Attributes¶

ACTION_TERM_REGISTRY module-attribute ¶

Classes¶

ActionTerm ¶

Attributes¶

entity_name class-attribute instance-attribute ¶

dofs_name class-attribute instance-attribute ¶

dofs_idx_local instance-attribute ¶

entity property ¶

genesis_dofs_idx property ¶

action property ¶

prev_action property ¶

processed_action property ¶

sim_surrogate property ¶

dofs_order property ¶

action_dim property ¶

num_dofs property ¶

Methods:¶

build ¶

verify_command_shape ¶

apply_actions abstractmethod ¶

ActionManager ¶

Attributes¶

total_action_dim cached property ¶

action_term_dim cached property ¶

total_dofs_dim cached property ¶

dofs_term_dim cached property ¶

action property ¶

prev_action property ¶

processed_actions property ¶

Methods:¶

summary ¶

clear_action_history ¶

reset ¶

T `module-attribute` ¶

num_envs `property` ¶

device `property` ¶

compute `abstractmethod` ¶

device_synced_fields `class-attribute` ¶

num_envs `property` ¶

device `property` ¶

compute `abstractmethod` ¶

iter_post_reset_terms `abstractmethod` ¶

terms `property` ¶

active_terms `property` ¶

ACTION_TERM_REGISTRY `module-attribute` ¶

entity_name `class-attribute` `instance-attribute` ¶

dofs_name `class-attribute` `instance-attribute` ¶

dofs_idx_local `instance-attribute` ¶

entity `property` ¶

genesis_dofs_idx `property` ¶

action `property` ¶

prev_action `property` ¶

processed_action `property` ¶

sim_surrogate `property` ¶

dofs_order `property` ¶

action_dim `property` ¶

num_dofs `property` ¶

apply_actions `abstractmethod` ¶

total_action_dim `cached` `property` ¶

action_term_dim `cached` `property` ¶

total_dofs_dim `cached` `property` ¶

dofs_term_dim `cached` `property` ¶

action `property` ¶

prev_action `property` ¶

processed_actions `property` ¶

ObservationTermLike `module-attribute` ¶

OBSERVATION_TERM_REGISTRY `module-attribute` ¶

effective_scale `property` ¶

effective_scale `property` ¶

group_term_names `property` ¶

active_terms `property` ¶

group_obs_dim `property` ¶

group_obs_term_dim `property` ¶

DEFAULT_REWARD_GROUP `module-attribute` ¶

RewardTermLike `module-attribute` ¶

REWARD_TERM_REGISTRY `module-attribute` ¶

range_s `class-attribute` `instance-attribute` ¶

weight `class-attribute` `instance-attribute` ¶

tags `class-attribute` `instance-attribute` ¶

is_temporal_dependent `property` ¶

range_s `class-attribute` `instance-attribute` ¶

weight `class-attribute` `instance-attribute` ¶

tags `class-attribute` `instance-attribute` ¶

is_temporal_dependent `property` ¶

device_synced_fields `class-attribute` ¶

group_term_names `property` ¶

is_timestep_in_range `staticmethod` ¶