Event Terms

Event terms change the simulation outside the policy's control: they reset state, place objects and randomize dynamics. Each runs in one mode from EventMode (STARTUP, RESET or INTERVAL, see constants). The events guide explains modes, ordering and what each randomizer needs from the scene; the term catalog lists every parameter.

from eden import events
from eden.constants import EventMode

The base classes EventTerm and EventTermFuncWrapper are on the managers page.

Resets

Reset event terms for base-state and DOF initialization.

Attributes

Classes

Functions:

reset_base_state_uniform

reset_base_state_uniform(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, pos_x_range: Vec2FType = (-0.5, 0.5), pos_y_range: Vec2FType = (-0.5, 0.5), pos_z_range: Vec2FType = (0.0, 0.0), roll_range: Vec2FType = (0.0, 0.0), pitch_range: Vec2FType = (0.0, 0.0), yaw_range: Vec2FType = (-3.14, 3.14))

Reset the base to its default pose, offset and rotated at random.

The offsets are added to the entity's default root position, and the sampled roll / pitch / yaw (world x-y-z, R_rpy = Rz @ Ry @ Rx) rotate the default orientation — R_rpy @ R_default — so an object authored lying on its side stays on its side. Both writes use the authored frame (relative=True), the frame default_root_pos is specified in: the asset's declared root origin lands on the target, not the centre of mass Genesis re-origins a free body onto. A heterogeneous entity's per-variant default poses are honored (Entity.get_default_root_pose). Similar to mjlab's reset_root_state_uniform, which composes its rotation in the body frame instead (R_default @ R_rpy); the two agree for an identity default orientation or a pure yaw about an upright default.

Rigid-only: relies on set_pos/set_quat, which raise on :class:ParticleEntity (MPM/SPH). For deformable entities, bake pose into the morph at creation time.

Source code in eden/managers/terms/events/resets.py

@EVENT_TERM_REGISTRY.register()
def reset_base_state_uniform(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    pos_x_range: Vec2FType = (-0.5, 0.5),
    pos_y_range: Vec2FType = (-0.5, 0.5),
    pos_z_range: Vec2FType = (0.0, 0.0),
    roll_range: Vec2FType = (0.0, 0.0),
    pitch_range: Vec2FType = (0.0, 0.0),
    yaw_range: Vec2FType = (-3.14, 3.14),
):
    """Reset the base to its default pose, offset and rotated at random.

    The offsets are added to the entity's default root position, and the sampled roll / pitch / yaw (world
    x-y-z, ``R_rpy = Rz @ Ry @ Rx``) rotate the default orientation — ``R_rpy @ R_default`` — so an object
    authored lying on its side stays on its side. Both writes use the authored frame (``relative=True``),
    the frame ``default_root_pos`` is specified in: the asset's declared root origin lands on the target,
    not the centre of mass Genesis re-origins a free body onto. A heterogeneous entity's per-variant default
    poses are honored (``Entity.get_default_root_pose``). Similar to mjlab's ``reset_root_state_uniform``,
    which composes its rotation in the body frame instead (``R_default @ R_rpy``); the two agree for an
    identity default orientation or a pure yaw about an upright default.

    Rigid-only: relies on ``set_pos``/``set_quat``, which raise on
    :class:`ParticleEntity` (MPM/SPH). For deformable entities, bake pose into
    the morph at creation time.
    """
    from eden.entities.rigid import RigidEntity

    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)

    entity = env.get_entity(entity_name)
    if not isinstance(entity, RigidEntity):
        raise TypeError(
            f"reset_base_state_uniform requires a RigidEntity for '{entity_name}', got {type(entity).__name__}."
        )

    default_pos, default_quat = entity.get_default_root_pose()  # fresh (num_envs, 3) / (num_envs, 4)

    ranges = (pos_x_range, pos_y_range, pos_z_range, roll_range, pitch_range, yaw_range)
    sampled = sample_uniform_ranges(ranges, env.num_envs, env.device)
    new_pos = default_pos + sampled[:, :3]
    new_quat = transform_quat_by_quat(default_quat, xyz_to_quat(sampled[:, 3:], rpy=True))

    if entity.root_qpos_is_authored_pose:
        # One write of the 7 root coordinates, which are the authored pose here. Velocities are zeroed as
        # Genesis' `set_pos` / `set_quat` zero them: a simulated entity's, not a kinematic one's.
        root_qpos = _rows_to_write(torch.cat([new_pos, new_quat], dim=-1), envs_idx)
        entity.set_qpos(root_qpos, qs_idx_local=_ROOT_QS, envs_idx=envs_idx, zero_velocity=not entity.is_kinematic)
        return

    # ``relative=True`` takes Genesis' general path, which wants the selected rows under every
    # ``envs_idx`` form (slice, index tensor or bool mask). Orientation first, as the build write does:
    # the solver position of a re-origined body depends on its orientation. One `nonzero()` for the
    # auto-reset's mask covers both gathers and both setters (`env_indices`).
    envs_idx = env_indices(envs_idx)
    entity.set_quat(new_quat[envs_idx], envs_idx=envs_idx, relative=True)
    entity.set_pos(new_pos[envs_idx], envs_idx=envs_idx, relative=True)

reset_dofs_by_offset

reset_dofs_by_offset(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, dofs_pos_range: Vec2FType = (-0.1, 0.1), dofs_vel_range: Vec2FType | None = None)

Reset DOFs to default + U(offset), clamped to soft limits.

The offset is drawn per env and DOF. Velocities are zeroed (the base's too), then set to U(dofs_vel_range) when that is given. Rigid-only: requires a :class:RigidEntity (DOFs are not defined on :class:ParticleEntity).

Source code in eden/managers/terms/events/resets.py

@EVENT_TERM_REGISTRY.register()
def reset_dofs_by_offset(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    dofs_pos_range: Vec2FType = (-0.1, 0.1),
    dofs_vel_range: Vec2FType | None = None,
):
    """Reset DOFs to ``default + U(offset)``, clamped to soft limits.

    The offset is drawn per env and DOF. Velocities are zeroed (the base's too), then set to
    ``U(dofs_vel_range)`` when that is given.
    Rigid-only: requires a :class:`RigidEntity` (DOFs are not defined on
    :class:`ParticleEntity`).
    """
    from eden.entities.rigid import RigidEntity

    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)

    entity = env.get_entity(entity_name)
    if not isinstance(entity, RigidEntity):
        raise TypeError(
            f"reset_dofs_by_offset requires a RigidEntity for '{entity_name}', got {type(entity).__name__}."
        )
    dofs_pos = entity.default_dofs_pos.clone() + sample_uniform(
        dofs_pos_range[0],
        dofs_pos_range[1],
        (env.num_envs, entity.num_dofs),
        device=env.device,
    )
    dofs_pos = dofs_pos.clamp(entity.soft_dofs_pos_limits[:, :, 0], entity.soft_dofs_pos_limits[:, :, 1])
    dofs_pos = _rows_to_write(dofs_pos, envs_idx)
    if dofs_vel_range is None:
        # `set_dofs_pos` zeroes every dof's velocity (the base's too) and runs forward kinematics once.
        entity.set_dofs_pos(dofs_pos, envs_idx=envs_idx)
        return

    # Velocities first, without forward kinematics: the position write below runs the one pass over these envs,
    # and that pass refreshes link velocities from the dof velocities written here; writing velocities after
    # the positions would run a second pass. Final state: every dof's velocity zero (the base's too, as
    # `set_dofs_pos`' default zeroing would leave it), then the sampled joint velocities.
    dofs_vel = _rows_to_write(
        sample_uniform(dofs_vel_range[0], dofs_vel_range[1], (env.num_envs, entity.num_dofs), device=env.device),
        envs_idx,
    )
    entity.zero_all_dofs_velocity(envs_idx=envs_idx, skip_forward=True)
    entity.set_dofs_vel(dofs_vel, envs_idx=envs_idx, skip_forward=True)
    entity.set_dofs_pos(dofs_pos, envs_idx=envs_idx, zero_velocity=False)

reset_dofs_by_scale

reset_dofs_by_scale(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, dofs_pos_scale_range: Vec2FType = (0.95, 1.05))

Reset DOFs to default * U(scale) per env/DOF; velocities to 0.

Multiplicative variant of :func:reset_dofs_by_offset (legged_gym-style init used by e.g. extreme-parkour). Rigid-only: requires a :class:RigidEntity.

Source code in eden/managers/terms/events/resets.py

@EVENT_TERM_REGISTRY.register()
def reset_dofs_by_scale(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    dofs_pos_scale_range: Vec2FType = (0.95, 1.05),
):
    """Reset DOFs to ``default * U(scale)`` per env/DOF; velocities to 0.

    Multiplicative variant of :func:`reset_dofs_by_offset` (legged_gym-style init used by
    e.g. extreme-parkour). Rigid-only: requires a :class:`RigidEntity`.
    """
    from eden.entities.rigid import RigidEntity

    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)

    entity = env.get_entity(entity_name)
    if not isinstance(entity, RigidEntity):
        raise TypeError(f"reset_dofs_by_scale requires a RigidEntity for '{entity_name}', got {type(entity).__name__}.")
    dofs_pos = entity.default_dofs_pos * sample_uniform(
        dofs_pos_scale_range[0],
        dofs_pos_scale_range[1],
        (env.num_envs, entity.num_dofs),
        device=env.device,
    )
    dofs_pos = dofs_pos.clamp(entity.soft_dofs_pos_limits[:, :, 0], entity.soft_dofs_pos_limits[:, :, 1])
    # `set_dofs_pos` zeroes every dof's velocity, the base's too.
    entity.set_dofs_pos(_rows_to_write(dofs_pos, envs_idx), envs_idx=envs_idx)

reset_dofs_by_scale_and_offset

reset_dofs_by_scale_and_offset(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, dofs_scale_range: Vec2FType = (0.8, 1.2), dofs_offset_range: Vec2FType = (-0.1, 0.1))

Reset DOFs to default * U(scale) + U(offset); velocities to 0.

Sampled per env and DOF, then clamped to the soft limits. OpenHomie's _reset_dofs: :func:reset_dofs_by_scale and :func:reset_dofs_by_offset composed. Rigid-only: requires a :class:RigidEntity.

Source code in eden/managers/terms/events/resets.py

@EVENT_TERM_REGISTRY.register()
def reset_dofs_by_scale_and_offset(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    dofs_scale_range: Vec2FType = (0.8, 1.2),
    dofs_offset_range: Vec2FType = (-0.1, 0.1),
):
    """Reset DOFs to ``default * U(scale) + U(offset)``; velocities to 0.

    Sampled per env and DOF, then clamped to the soft limits. OpenHomie's ``_reset_dofs``:
    :func:`reset_dofs_by_scale` and :func:`reset_dofs_by_offset` composed.
    Rigid-only: requires a :class:`RigidEntity`.
    """
    from eden.entities.rigid import RigidEntity

    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)

    entity = env.get_entity(entity_name)
    if not isinstance(entity, RigidEntity):
        raise TypeError(
            f"reset_dofs_by_scale_and_offset requires a RigidEntity for '{entity_name}', got {type(entity).__name__}."
        )
    shape = (env.num_envs, entity.num_dofs)
    scale = sample_uniform(dofs_scale_range[0], dofs_scale_range[1], shape, device=env.device)
    offset = sample_uniform(dofs_offset_range[0], dofs_offset_range[1], shape, device=env.device)
    limits = entity.soft_dofs_pos_limits
    dofs_pos = (entity.default_dofs_pos * scale + offset).clamp(limits[..., 0], limits[..., 1])
    # `set_dofs_pos` zeroes every dof's velocity, the base's too.
    entity.set_dofs_pos(_rows_to_write(dofs_pos, envs_idx), envs_idx=envs_idx)

Domain randomization

Pushes, external wrenches, DOF position and offset randomization, actuator randomization (motor strength, gains, backlash, deadband, torque noise) and mass, CoM and friction randomization.

Domain-randomization event terms (pushes, external wrenches, DOF randomization).

Attributes

Classes

PushByVelocity

PushByVelocity(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Periodically add a random velocity to a free base.

velocity_range maps x / y / z (world-frame linear, m/s) and roll / pitch / yaw (base-frame angular, rad/s) to (low, high); an axis left out is not perturbed.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.entity: Entity | None = None
    self.base_dofs_index: torch.Tensor | None = None

    range_list = [self.velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]
    self.ranges = torch.tensor(range_list, device=self._env.device)

Attributes

mode class-attribute instance-attribute

mode: EventMode = _INTERVAL

interval_range_s class-attribute instance-attribute

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

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

base_joint_name class-attribute instance-attribute

base_joint_name: str | None = None

velocity_range class-attribute instance-attribute

velocity_range: dict[str, Vec2FType] = {}

entity instance-attribute

entity: Entity | None = None

base_dofs_index instance-attribute

base_dofs_index: Tensor | None = None

ranges instance-attribute

ranges = torch.tensor(range_list, device=self._env.device)

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    self.entity = self.resolve_entity(self.entity_name)
    base_dofs_index = self.entity.find_free_base_dofs_idx_local(self.base_joint_name)
    self.base_dofs_index = torch.as_tensor(base_dofs_index, dtype=gs.tc_int, device=self.device).contiguous()
    self._genesis_base_dofs_idx = as_genesis_index(self.base_dofs_index)
    self._write_every_row = self.entity.setters_take_every_row and isinstance(self._genesis_base_dofs_idx, range)

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    rows = _push_rows(envs_idx, self._write_every_row)
    lin_vel_w = self.entity.get_vel(envs_idx=rows)
    ang_vel_w = self.entity.get_ang(envs_idx=rows)
    quat_w = self.entity.get_quat(envs_idx=rows)

    rand_vel = sample_uniform(
        self.ranges[:, 0],
        self.ranges[:, 1],
        (self._env.num_envs, 6),
        device=self._env.device,
    )[rows]
    lin_vel_w += rand_vel[:, :3]
    # Genesis stores a free joint's rotational DOF velocity in the base BODY frame (MuJoCo
    # qvel convention). Convert the current world ang-vel to body, add the (body-frame)
    # perturbation, and write the BODY-frame result straight into DOFs 3-5 — do NOT rotate
    # back to world first. Linear DOFs are world-frame and left as-is.
    ang_vel_b = inv_transform_by_quat(ang_vel_w, quat_w)
    ang_vel_b += rand_vel[:, 3:]
    self.entity.set_dofs_vel(
        torch.cat([lin_vel_w, ang_vel_b], dim=1),
        dofs_idx_local=self._genesis_base_dofs_idx,
        envs_idx=envs_idx,
    )

PushSetPlanarVelocity

PushSetPlanarVelocity(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Periodically SET the base planar velocity to a random value.

Unlike :class:PushByVelocity (which ADDS a perturbation), this overwrites vx/vy with U(velocity_xy_range) while keeping vz and the angular velocity unchanged — the legged_gym/extreme-parkour push semantics. Uses the same free-joint DOF-write convention: linear DOFs are world-frame, rotational DOF velocity is body-frame.

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 = _INTERVAL

interval_range_s class-attribute instance-attribute

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

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

base_joint_name class-attribute instance-attribute

base_joint_name: str | None = None

velocity_xy_range class-attribute instance-attribute

velocity_xy_range: Vec2FType = (-0.5, 0.5)

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    entity: RigidEntity = self.resolve_entity(self.entity_name)
    base_dofs_idx = entity.find_free_base_dofs_idx_local(self.base_joint_name)
    self._base_dofs_idx = as_genesis_index(
        torch.as_tensor(base_dofs_idx, dtype=gs.tc_int, device=self.device).contiguous()
    )
    self._write_every_row = entity.setters_take_every_row and isinstance(self._base_dofs_idx, range)

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    entity: RigidEntity = self.resolve_entity(self.entity_name)
    rows = _push_rows(envs_idx, self._write_every_row)
    lin_vel_w = entity.get_vel(envs_idx=rows)
    ang_vel_w = entity.get_ang(envs_idx=rows)
    quat_w = entity.get_quat(envs_idx=rows)

    rand_xy = sample_uniform(
        self.velocity_xy_range[0], self.velocity_xy_range[1], (self._env.num_envs, 2), device=self.device
    )[rows]
    lin_vel_w[:, :2] = rand_xy  # SET, not add
    # Rotational free-joint DOF velocity is stored in the base BODY frame (MuJoCo qvel
    # convention) — convert before writing back unchanged.
    ang_vel_b = inv_transform_by_quat(ang_vel_w, quat_w)
    entity.set_dofs_vel(
        torch.cat([lin_vel_w, ang_vel_b], dim=1),
        dofs_idx_local=self._base_dofs_idx,
        envs_idx=envs_idx,
    )

ApplyExternalForce

ApplyExternalForce(env: EnvBase, options: EventTermOptions)

Bases: _ApplyExternalWrenchBase

Apply a randomized external force on specified links.

force_x_range / force_y_range / force_z_range are the per-axis force ranges (N), sampled uniformly per env and link. See :class:_ApplyExternalWrenchBase for the shared parameters and the cross-term independence caveat.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.links_idx: torch.Tensor | None = None
    self.entity: RigidEntity | None = None
    self._ref_frame: gs.link_ref_frame | None = None

Attributes

force_x_range class-attribute instance-attribute

force_x_range: Vec2FType = (0.0, 0.0)

force_y_range class-attribute instance-attribute

force_y_range: Vec2FType = (0.0, 0.0)

force_z_range class-attribute instance-attribute

force_z_range: Vec2FType = (0.0, 0.0)

ApplyExternalTorque

ApplyExternalTorque(env: EnvBase, options: EventTermOptions)

Bases: _ApplyExternalWrenchBase

Apply a randomized external torque on specified links.

torque_x_range / torque_y_range / torque_z_range are the per-axis torque ranges (N·m), sampled uniformly per env and link. See :class:_ApplyExternalWrenchBase for the shared parameters and the cross-term independence caveat.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.links_idx: torch.Tensor | None = None
    self.entity: RigidEntity | None = None
    self._ref_frame: gs.link_ref_frame | None = None

Attributes

torque_x_range class-attribute instance-attribute

torque_x_range: Vec2FType = (0.0, 0.0)

torque_y_range class-attribute instance-attribute

torque_y_range: Vec2FType = (0.0, 0.0)

torque_z_range class-attribute instance-attribute

torque_z_range: Vec2FType = (0.0, 0.0)

SetRandomDofsPos

SetRandomDofsPos(env: EnvBase, options: _TermOptionsT)

Bases: _RandomizeDofsPosBase

Set the joint positions of an entity's DOFs to a random position (default mode: reset).

Defaults to ratio mode, sampling within the first 20% of each DOF's limit range. See :class:_RandomizeDofsPosBase for the full parameter description.

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

dofs_pos_range class-attribute instance-attribute

dofs_pos_range: Vec2FType = (0.0, 0.2)

apply_as_ratio class-attribute instance-attribute

apply_as_ratio: bool = True

RandomizeDofsPosOffset

RandomizeDofsPosOffset(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Per-env joint-position offset on joint position controllers (encoder bias).

Each selected controller draws its own offsets, one per env and controlled DOF, and takes them through :meth:~eden.managers.terms.actions.joint_actions._JointPDControllerBase.set_dofs_pos_offset. The draw adds to the controller's position target on top of its configured offset and its reference_source reference, and replaces neither, so it composes with a static offset and, in either order, with :class:RandomizeStartupDofsPosBias.

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	The joint position controller(s) to randomize: one term name, a list of them, or None (the default) for every joint position controller on entity_name — see :func:~eden.managers.terms.actions.joint_actions.is_joint_position_controller. None keeps a config that splits the entity's joints across several controllers (a per-limb scale or delay) from having to list, and so depend on, each term's name. A named term must be a joint position controller on entity_name, and None raises if the entity has none.
dofs_pos_range	Per-DOF offset range (radians) sampled uniformly per env and DOF.

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

entity_name class-attribute instance-attribute

entity_name: str = ''

action_term_name class-attribute instance-attribute

action_term_name: str | list[str] | None = None

dofs_pos_range class-attribute instance-attribute

dofs_pos_range: Vec2FType = (-0.01745, 0.01745)

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    # Touched only to fail fast on a bad name; compute() writes to the controllers, not the entity.
    self.resolve_entity(self.entity_name)
    self._targets: list[_JointPDControllerBase] = self._resolve_terms()

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    for term in self._targets:
        shape = (self._env.num_envs, len(term.dofs_name))
        offset = sample_uniform(*self.dofs_pos_range, shape, device=self._env.device)[envs_idx]
        term.set_dofs_pos_offset(offset, envs_idx)

RandomizeStartupDofsPos

RandomizeStartupDofsPos(env: EnvBase, options: _TermOptionsT)

Bases: _RandomizeDofsPosBase

Randomize the absolute joint positions of an entity's DOFs (default mode: reset).

Writes joint positions directly via set_dofs_pos — useful for spawning each episode with a randomized initial pose. Despite the name, defaults to mode=_RESET so it fires on every episode reset; override to _STARTUP for a one-time draw. Defaults to absolute mode with a small ±1° range. See :class:_RandomizeDofsPosBase for the full parameter description.

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

dofs_pos_range class-attribute instance-attribute

dofs_pos_range: Vec2FType = (-0.01745, 0.01745)

apply_as_ratio class-attribute instance-attribute

apply_as_ratio: bool = False

RandomizeStartupDofsPosBias

RandomizeStartupDofsPosBias(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Bake a per-env joint-zero bias into entity.default_dofs_pos at startup.

Simulates motor calibration drift: each env gets a small fixed offset on every DOF of the entity, sampled once at startup and held constant for the lifetime of the env. The bias persists across resets, unlike the per-reset jitter from :class:RandomizeDofsPosOffset. The two compose: a calibration drift baked in at startup + small per-episode jitter on top.

When action_terms_name lists one or more PD controllers, each controller's _offset cache (folded once at build time as default_offset + configured_offset, see _JointPDControllerBase._build_scale_and_offset) is updated in lockstep by adding the same per-env bias delta on the DOFs that controller owns. This avoids the "stale cache" trap where mutating entity.default_dofs_pos would diverge from a controller's pre-folded static offset.

Build-time contract. build() performs strict validation so that any stale-cache footgun surfaces loudly at startup rather than as silent drift at runtime:

1. action_terms_name must contain no duplicates.
2. Each listed term must be a :class:_JointPDControllerBase instance, target the same entity, and have reference_source == ReferenceSource.DEFAULT.

3. Every same-entity action term that caches default_dofs_pos at build must be either listed (and updated in lockstep) or rejected with a clear error. Specifically:

4. _JointPDControllerBase with reference_source=DEFAULT not in the listed set: error (its _offset would silently go stale).

5. :class:NullJointAction caches a copy of default_dofs_pos at build with no refresh path; if it is present on the same entity, build() errors out. OperationalSpaceController refreshes its declared nullspace reference at reset and therefore does not need this rejection.

PARAMETER	DESCRIPTION
entity_name	The entity to bias.
action_terms_name	Names of PD controller action terms whose _offset caches should be kept in sync with the bias. Each must be a :class:_JointPDControllerBase on entity_name with reference_source=DEFAULT. Must list every such controller on the entity — see the build-time contract above. Empty list is fine if and only if no DEFAULT-source PD controllers exist on this entity.
dofs_pos_range	Per-DOF bias range, sampled uniformly. Default ±1 degree.

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 = _STARTUP

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_terms_name class-attribute instance-attribute

action_terms_name: list[str] = []

dofs_pos_range class-attribute instance-attribute

dofs_pos_range: Vec2FType = (-0.01745, 0.01745)

entity instance-attribute

entity: RigidEntity

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    # Strict three-phase validation so any stale-cache footgun surfaces loudly at
    # startup rather than as silent drift at runtime (see the class docstring).
    self.entity = self.resolve_entity(self.entity_name)
    self._reject_duplicate_action_terms()
    self._targets = self._resolve_listed_controllers()
    self._reject_unhandled_cache_holders()
    self._applied = False

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    # Idempotence: startup terms are invoked once today, but make the contract explicit.
    if self._applied:
        return
    self._applied = True

    if envs_idx is None:
        envs_idx = slice(None)

    if self.entity.default_dofs_pos.shape[0] == 1 and self._env.num_envs != 1:
        self.entity.default_dofs_pos = self.entity.default_dofs_pos.repeat(self._env.num_envs, 1)
    elif self.entity.default_dofs_pos.shape[0] != self._env.num_envs:
        raise ValueError(
            f"{term_subject(self)} expected default_dofs_pos batch size 1 or {self._env.num_envs}, "
            f"got {self.entity.default_dofs_pos.shape[0]}"
        )

    bias_full = sample_uniform(
        self.dofs_pos_range[0],
        self.dofs_pos_range[1],
        (self._env.num_envs, self.entity.default_dofs_pos.shape[1]),
        device=self._env.device,
    )[envs_idx]

    # 1. Bake the bias into the entity's default DOF positions (additive).
    self.entity.default_dofs_pos[envs_idx] = self.entity.default_dofs_pos[envs_idx] + bias_full

    # 2. Keep each listed PD controller's pre-folded `_offset` cache in lockstep,
    #    on the subset of DOFs that controller owns.
    for term, columns in self._targets:
        term._offset[envs_idx] = term._offset[envs_idx] + bias_full[:, columns]

RandomizeMotorStrength

RandomizeMotorStrength(env, options)

Bases: _ResetModeActuatorDR

Randomize a :class:MotorStrength action modifier's scalar (default factor 0.9 to 1.1).

Requires

The action term must have a MotorStrength modifier (either directly or inside a Compose chain).

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	The name of the PD controller action term hosting the MotorStrength modifier.
motor_strength_range	Multiplicative range applied to the modifier's strength, sampled uniformly per env (one scalar per env).

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env, options) -> None:
    super().__init__(env=env, options=options)
    mode = getattr(options, "mode", _RESET)
    throttle = getattr(options, "min_step_count_between_reset", 0)
    if mode != _RESET or throttle != 0:
        raise ValueError(
            f"{type(self).__name__} randomizes in reset() and supports only mode='reset' with "
            f"min_step_count_between_reset=0 (got mode={mode!r}, min_step_count_between_reset="
            f"{throttle}). Its compute() is a no-op, so other modes/throttles would be "
            "silently inert."
        )

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_term_name class-attribute instance-attribute

action_term_name: str = ''

motor_strength_range class-attribute instance-attribute

motor_strength_range: Vec2FType = (0.9, 1.1)

entity instance-attribute

entity: RigidEntity

term instance-attribute

term: ExplicitPDController | VelocityFeedforwardPDController

modifier instance-attribute

modifier: MotorStrength

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import MotorStrength

    self.entity = self.resolve_entity(self.entity_name)
    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, MotorStrength, type(self).__name__, self.action_term_name)

RandomizeGearBacklash

RandomizeGearBacklash(env, options)

Bases: _ResetModeActuatorDR

Randomize the one-sided target offset of a :class:GearBacklash action modifier.

Requires

The action term must have a GearBacklash modifier (either directly or inside a Compose chain).

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	The name of the PD controller action term hosting the GearBacklash modifier.
backlash_range	When apply_as_ratio is false, the range of one-sided backlash offsets (rad) sampled uniformly per env and DOF. When apply_as_ratio is true, the range of multipliers applied to the modifier's configured per-DOF baseline (see apply_as_ratio).
apply_as_ratio	If false (default), backlash_range sets absolute backlash offsets in radians. If true, backlash_range is interpreted as (low, high) multipliers (uniform per env and DOF): the backlash becomes m * backlash_i, where backlash_i is that DOF's configured nominal backlash from :class:~eden.managers.modifiers.actions.actuators.GearBacklash at build time.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env, options) -> None:
    super().__init__(env=env, options=options)
    mode = getattr(options, "mode", _RESET)
    throttle = getattr(options, "min_step_count_between_reset", 0)
    if mode != _RESET or throttle != 0:
        raise ValueError(
            f"{type(self).__name__} randomizes in reset() and supports only mode='reset' with "
            f"min_step_count_between_reset=0 (got mode={mode!r}, min_step_count_between_reset="
            f"{throttle}). Its compute() is a no-op, so other modes/throttles would be "
            "silently inert."
        )

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_term_name class-attribute instance-attribute

action_term_name: str = ''

backlash_range class-attribute instance-attribute

backlash_range: Vec2FType = (0.0, 0.01)

apply_as_ratio class-attribute instance-attribute

apply_as_ratio: bool = False

entity instance-attribute

entity: RigidEntity

term instance-attribute

term: _JointPDControllerBase

modifier instance-attribute

modifier: GearBacklash

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import GearBacklash

    self.entity = self.resolve_entity(self.entity_name)
    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, GearBacklash, type(self).__name__, self.action_term_name)

RandomizeConstantTorqueKick

RandomizeConstantTorqueKick(env, options)

Bases: _ResetModeActuatorDR

Randomize the torque magnitude of a :class:ConstantTorqueKick action modifier.

Requires

The action term must have a ConstantTorqueKick modifier (either directly or inside a Compose chain).

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	The name of the PD controller action term hosting the ConstantTorqueKick modifier.
torque_kick_range	When apply_as_ratio is false, the range of torque-kick magnitudes (N·m) sampled uniformly per env and DOF. When apply_as_ratio is true, the range of multipliers applied to the modifier's configured per-DOF baseline (see apply_as_ratio).
apply_as_ratio	If false (default), torque_kick_range sets absolute torque-kick magnitudes. If true, torque_kick_range is interpreted as (low, high) multipliers (uniform per env and DOF): the kick becomes m * torque_kick_i, where torque_kick_i is that DOF's configured nominal magnitude from :class:~eden.managers.modifiers.actions.actuators.ConstantTorqueKick at build time.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env, options) -> None:
    super().__init__(env=env, options=options)
    mode = getattr(options, "mode", _RESET)
    throttle = getattr(options, "min_step_count_between_reset", 0)
    if mode != _RESET or throttle != 0:
        raise ValueError(
            f"{type(self).__name__} randomizes in reset() and supports only mode='reset' with "
            f"min_step_count_between_reset=0 (got mode={mode!r}, min_step_count_between_reset="
            f"{throttle}). Its compute() is a no-op, so other modes/throttles would be "
            "silently inert."
        )

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_term_name class-attribute instance-attribute

action_term_name: str = ''

torque_kick_range class-attribute instance-attribute

torque_kick_range: Vec2FType = (0.0, 0.01)

apply_as_ratio class-attribute instance-attribute

apply_as_ratio: bool = False

entity instance-attribute

entity: RigidEntity

term instance-attribute

term: _JointPDControllerBase

modifier instance-attribute

modifier: ConstantTorqueKick

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import ConstantTorqueKick

    self.entity = self.resolve_entity(self.entity_name)
    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, ConstantTorqueKick, type(self).__name__, self.action_term_name)

RandomizeKpKdGains

RandomizeKpKdGains(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Scale an entity's joint PD gains by a random factor.

With action_term_name set, the draw goes into that controller's own gain caches, over the controller's DOFs in its dofs_name order: each cache becomes the entity's current gains times the drawn factor. Writing them only in :meth:compute is not enough: the controller's reset() re-reads its caches from the entity, and in both reset and startup mode that re-read lands after the draw (the ordering :class:_ResetModeActuatorDR describes), leaving the randomization silently inert. So the term keeps each env's factors and applies them again in :meth:reset, which runs after action_manager.reset(); a draw in any mode then holds until that env's next draw. Keeping factors rather than gains is what lets it compose with anything else that changes the entity's gains (an entity-path RandomizeKpKdGains, a gain curriculum): each reset scales whatever the entity holds then, and an env not yet drawn keeps the entity's gains unchanged (factor 1).

Requires

env_options.batch_dofs_info=True to support per-env PD gain randomization.

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	If set, the name of an ExplicitPDController or VelocityFeedforwardPDController on entity_name, whose _kp / _kd caches receive the gains; the entity is left untouched. If None, the gains are written directly to the entity via set_dofs_kp / set_dofs_kd (the path Genesis' built-in PD, ImplicitPDController, reads).
kp_range	Multiplicative range, sampled uniformly per env and DOF. On the entity path it multiplies entity.default_dofs_kp; on the controller path (action_term_name set) it multiplies the entity's current kp, which is default_dofs_kp unless something else changed it.
kd_range	As kp_range, for kd (entity.default_dofs_kd on the entity path).

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

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_term_name class-attribute instance-attribute

action_term_name: str | None = None

kp_range class-attribute instance-attribute

kp_range: Vec2FType = (0.9, 1.1)

kd_range class-attribute instance-attribute

kd_range: Vec2FType = (0.9, 1.1)

entity instance-attribute

entity: RigidEntity

term instance-attribute

term: ExplicitPDController | VelocityFeedforwardPDController | None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    self.entity = self.resolve_entity(self.entity_name)
    self.term = None
    if self.action_term_name is None:
        return

    from eden.managers.terms.actions.joint_actions import ExplicitPDController, VelocityFeedforwardPDController

    term = self._env.action_manager.get_term(self.action_term_name)
    if not isinstance(term, (ExplicitPDController, VelocityFeedforwardPDController)):
        raise ValueError(
            f"{term_subject(self)} (action_term_name={self.action_term_name!r}) writes the joint gain caches of an "
            f"ExplicitPDController or VelocityFeedforwardPDController, but action term "
            f"{self.action_term_name!r} is of type {type(term).__name__}, which has none. For Genesis' built-in PD "
            "(ImplicitPDController) set action_term_name=None: the gains then go to the entity "
            "(set_dofs_kp / set_dofs_kd), which is what that controller reads."
        )
    if term.entity_name != self.entity_name:
        raise ValueError(
            f"{term_subject(self)} (action_term_name={self.action_term_name!r}): "
            f"action term {self.action_term_name!r} controls entity {term.entity_name!r}, so its gains would be "
            f"scaled from the wrong entity's default_dofs_kp / default_dofs_kd. Set "
            f"entity_name={term.entity_name!r}."
        )
    self.term = term
    # Each env's factors in the controller's layout, re-applied to the entity's gains by `reset()`.
    # Identity until the first draw.
    self._kp_factor = torch.ones_like(term._kp)
    self._kd_factor = torch.ones_like(term._kd)

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    if self.term is not None:
        self._draw_into_term(envs_idx)
        return
    n_dofs = len(self.entity.dofs_name)

    kp_factor = sample_uniform(
        self.kp_range[0],
        self.kp_range[1],
        (self._env.num_envs, n_dofs),
        device=self._env.device,
    )[envs_idx]
    kd_factor = sample_uniform(
        self.kd_range[0],
        self.kd_range[1],
        (self._env.num_envs, n_dofs),
        device=self._env.device,
    )[envs_idx]

    # NOTE: default_dofs_kp/kd may be (1, n_dofs) even with batch_dofs_info=True
    base_kp = self.entity.default_dofs_kp.expand(self._env.num_envs, -1)[envs_idx]
    base_kd = self.entity.default_dofs_kd.expand(self._env.num_envs, -1)[envs_idx]

    self.entity.set_dofs_kp(
        base_kp * kp_factor,
        dofs_idx_local=self.entity.dofs_idx_local,
        envs_idx=envs_idx,
    )
    self.entity.set_dofs_kd(
        base_kd * kd_factor,
        dofs_idx_local=self.entity.dofs_idx_local,
        envs_idx=envs_idx,
    )

reset

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

Source code in eden/managers/terms/events/domain_rand.py

def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    # Runs after `action_manager.reset()` re-read the controller's caches from the entity (which this term
    # leaves untouched): scale them by this env's factors again. See the class docstring.
    if self.term is None:
        return
    if envs_idx is None:
        envs_idx = slice(None)
    self._scale_term_gains(envs_idx)

RandomizeDeadbandEpsilon

RandomizeDeadbandEpsilon(env, options)

Bases: _ResetModeActuatorDR

Randomize the position-error threshold of a :class:Deadband action modifier.

Requires

The action term must have a Deadband modifier (either directly or inside a Compose chain).

PARAMETER	DESCRIPTION
entity_name	The name of the entity to control.
action_term_name	The name of the PD controller action term hosting the Deadband modifier.
deadband_epsilon_range	When apply_as_ratio is false, the range of the deadband (position error, rad) sampled uniformly per env and DOF. When apply_as_ratio is true, the range of multipliers applied to the modifier's configured per-DOF baseline (see apply_as_ratio).
apply_as_ratio	If false (default), deadband_epsilon_range sets absolute deadband values in radians. If true, deadband_epsilon_range is interpreted as (low, high) multipliers (uniform per env and DOF): the deadband becomes m * epsilon_i, where epsilon_i is that DOF's configured nominal deadband from :class:~eden.managers.modifiers.actions.actuators.Deadband at build time.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env, options) -> None:
    super().__init__(env=env, options=options)
    mode = getattr(options, "mode", _RESET)
    throttle = getattr(options, "min_step_count_between_reset", 0)
    if mode != _RESET or throttle != 0:
        raise ValueError(
            f"{type(self).__name__} randomizes in reset() and supports only mode='reset' with "
            f"min_step_count_between_reset=0 (got mode={mode!r}, min_step_count_between_reset="
            f"{throttle}). Its compute() is a no-op, so other modes/throttles would be "
            "silently inert."
        )

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

action_term_name class-attribute instance-attribute

action_term_name: str = ''

deadband_epsilon_range class-attribute instance-attribute

deadband_epsilon_range: Vec2FType = (0.0, 0.01)

apply_as_ratio class-attribute instance-attribute

apply_as_ratio: bool = False

entity instance-attribute

entity: RigidEntity

term instance-attribute

term: _JointPDControllerBase

modifier instance-attribute

modifier: Deadband

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import Deadband

    self.entity = self.resolve_entity(self.entity_name)
    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, Deadband, type(self).__name__, self.action_term_name)

RandomizeFrictionRatio

RandomizeFrictionRatio(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Randomize sliding friction on links_name to absolute values, not a scale.

Every compute() first pins the whole entity's base friction to 1.0 (set_friction(1.0), entity-wide — this discards any authored default_geoms_friction, including on links outside links_name, which are left at a flat 1.0) and then writes per-link ratios on links_name. friction_range is therefore an absolute coefficient range, regardless of what the asset authored; a source config expressed as "scale x authored value" must be converted to the resulting absolute range. Genesis multiplies the same ratio into the torsional and rolling coefficients where those axes are enabled, so this term randomizes spin resistance too.

Genesis resolves a contact pair's friction as max(a, b), so the randomized coefficient only takes effect against surfaces whose friction is below the sampled value — author the ground plane low (e.g. PlaneOptions(default_geoms_friction={"*": 0.01})) so the robot side always wins. Overlapping link sets (a whole-robot pass plus a tighter feet range) are ordered by the events' priority field (ascending). mode defaults to reset, but friction DR is usually a per-env-lifetime property — every in-tree task uses mode="startup". The full porting recipe lives in docs/guides/entities.md ("Porting a friction-randomization setup").

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.ls_idx_local: torch.Tensor | None = None
    self.entity: Entity | None = None

Attributes

mode class-attribute instance-attribute

mode: EventMode = _RESET

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

links_name class-attribute instance-attribute

links_name: list[str] = []

friction_range class-attribute instance-attribute

friction_range: Vec2FType = (0.3, 1.2)

ls_idx_local instance-attribute

ls_idx_local: Tensor | None = None

entity instance-attribute

entity: Entity | None = None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    entity = self.resolve_entity(self.entity_name)
    self.links_name, ls_idx_local = entity.find_named_links_idx_local(self.links_name)
    assert len(ls_idx_local) > 0, f"No links found for {self.links_name} in {self.__class__.__name__}"
    self.ls_idx_local = torch.as_tensor(ls_idx_local, dtype=gs.tc_int, device=self.device).contiguous()
    self.entity = entity

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    # NOTE: set the friction of the terrain and objects small so that the robot geometry can decide the friction
    # NOTE: the friction between the object A and B is defined by the maximum friction of the two objects
    # NOTE: friction_a = geoms_info.friction[i_ga] * geoms_state.friction_ratio[i_ga, i_b]
    # NOTE: friction_b = geoms_info.friction[i_gb] * geoms_state.friction_ratio[i_gb, i_a]
    # NOTE: contact_data.friction[i_c, i_b] = ti.max(ti.max(friction_a, friction_b), 1e-2)
    # NOTE: geoms_info.friction defaults to 1.0 and has to be in range [1e-2, 5.0] for simulation stability.

    friction_ratio = sample_uniform(
        self.friction_range[0],
        self.friction_range[1],
        (
            self._env.num_envs,
            len(self.links_name),
        ),
        device=self._env.device,
    )[envs_idx]

    self.entity.set_friction(1.0)
    self.entity.set_friction_ratio(
        friction_ratio,
        ls_idx_local=self.ls_idx_local,
        envs_idx=envs_idx,
    )

RandomizeComShift

RandomizeComShift(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Offset the named links' centres of mass, per env.

Samples a per-env, per-link xyz offset (meters, link frame) from the three axis ranges and writes nominal_COM + shift. Requires EnvOptions(batch_links_info=True) — Genesis stores link inertial info once for the whole batch otherwise, and rejects the per-env write. Rejects a link fixed to the world at build, whose COM the solver never reads.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.ls_idx_local: torch.Tensor | None = None
    self.entity: Entity | None = None

Attributes

mode class-attribute instance-attribute

mode: EventMode = _RESET

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

links_name class-attribute instance-attribute

links_name: list[str] = []

com_shift_x_range class-attribute instance-attribute

com_shift_x_range: Vec2FType = (-0.01, 0.01)

com_shift_y_range class-attribute instance-attribute

com_shift_y_range: Vec2FType = (-0.01, 0.01)

com_shift_z_range class-attribute instance-attribute

com_shift_z_range: Vec2FType = (-0.01, 0.01)

ls_idx_local instance-attribute

ls_idx_local: Tensor | None = None

entity instance-attribute

entity: Entity | None = None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    resolved_names, ls_idx_local = self.resolve_entity(self.entity_name).find_named_links_idx_local(self.links_name)
    assert len(ls_idx_local) > 0, f"No links found for {self.links_name} in {self.__class__.__name__}"
    self.ls_idx_local = torch.as_tensor(ls_idx_local, dtype=gs.tc_int, device=self.device).contiguous()
    self.entity = self.resolve_entity(self.entity_name)
    _reject_world_fixed_links(self, self.entity, self.links_name, resolved_names, ls_idx_local, "COM")
    _refuse_anchored_inertials(self, "COM")

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    com_shift = torch.cat(
        [
            self._sample_com_shift_axis(self.com_shift_x_range),
            self._sample_com_shift_axis(self.com_shift_y_range),
            self._sample_com_shift_axis(self.com_shift_z_range),
        ],
        dim=-1,
    )[envs_idx]
    self.entity.set_COM_shift(com_shift, ls_idx_local=self.ls_idx_local, envs_idx=envs_idx)

RandomizeMassShift

RandomizeMassShift(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Add a per-env offset to the named links' masses.

Samples a per-env, per-link shift (kg) from mass_shift_range and writes nominal_mass + shift. Requires EnvOptions(batch_links_info=True), and rejects a range whose lower bound could drive a selected link's mass non-positive — the write itself is unchecked (Genesis skips the device-side check for speed), so it would otherwise silently produce unphysical dynamics. It also rejects a link fixed to the world, whose mass the solver never reads.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.ls_idx_local: torch.Tensor | None = None
    self.entity: Entity | None = None

Attributes

mode class-attribute instance-attribute

mode: EventMode = _RESET

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

links_name class-attribute instance-attribute

links_name: list[str] = []

mass_shift_range class-attribute instance-attribute

mass_shift_range: Vec2FType = (-1.0, 3.0)

ls_idx_local instance-attribute

ls_idx_local: Tensor | None = None

entity instance-attribute

entity: Entity | None = None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    resolved_names, ls_idx_local = self.resolve_entity(self.entity_name).find_named_links_idx_local(self.links_name)
    assert len(ls_idx_local) > 0, f"No links found for {self.links_name} in {self.__class__.__name__}"
    self.ls_idx_local = torch.as_tensor(ls_idx_local, dtype=gs.tc_int, device=self.device).contiguous()
    self.entity = self.resolve_entity(self.entity_name)
    _reject_world_fixed_links(self, self.entity, self.links_name, resolved_names, ls_idx_local, "mass")
    _refuse_anchored_inertials(self, "mass")
    # The worst-case sampled mass is nominal + range lower bound. A one-time device sync
    # at build is the earliest well-defined point to reject a range that could go
    # non-positive mid-training.
    nominal = self.entity.get_links_inertial_mass(self.ls_idx_local)  # (num_envs, n_selected)
    lightest = nominal.min(dim=0).values
    if float(lightest.min()) + self.mass_shift_range[0] <= 0.0:
        offender = int(lightest.argmin())
        raise ValueError(
            f"{term_subject(self)}: mass_shift_range[0]={self.mass_shift_range[0]} can drive link "
            f"'{resolved_names[offender]}' ({float(lightest[offender]):.3g} kg) non-positive. Narrow the "
            f"range or drop the link from links_name."
        )

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    mass_shift = sample_uniform(
        self.mass_shift_range[0],
        self.mass_shift_range[1],
        (
            self._env.num_envs,
            len(self.ls_idx_local),
        ),
        device=self._env.device,
    )[envs_idx]
    self.entity.set_mass_shift(mass_shift, ls_idx_local=self.ls_idx_local, envs_idx=envs_idx)

RandomizeLinkMassScale

RandomizeLinkMassScale(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Multiplicative per-link mass randomization.

Scales each specified link's mass by a random factor sampled from [mass_scale_range[0], mass_scale_range[1]]. Internally uses set_mass_shift with shift = mass * (scale - 1). Requires EnvOptions(batch_links_info=True), and rejects a non-positive lower scale bound — the write itself is unchecked, so a scale of zero or below would silently produce a massless or negative-mass link.

PARAMETER	DESCRIPTION
entity_name	The entity to randomize.
links_name	Link name patterns to randomize. ["*"] selects every link, so it suits only an entity with no link fixed to the world. On a fixed-base arm, a mounted hand or a fixed=True prop, name the moving links: a fixed link is rejected at build, since its mass is never read, and the error gives a links_name that selects exactly the moving ones.
mass_scale_range	Multiplicative scale range, e.g. (0.9, 1.2) for 90-120%.

Source code in eden/managers/terms/events/domain_rand.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    self.ls_idx_local: torch.Tensor | None = None
    self.entity: RigidEntity | None = None
    self._base_mass: torch.Tensor | None = None

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

links_name class-attribute instance-attribute

links_name: list[str] = []

mass_scale_range class-attribute instance-attribute

mass_scale_range: Vec2FType = (0.9, 1.1)

ls_idx_local instance-attribute

ls_idx_local: Tensor | None = None

entity instance-attribute

entity: RigidEntity | None = None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    if self.mass_scale_range[0] <= 0.0:
        raise ValueError(
            f"{term_subject(self)}: mass_scale_range[0]={self.mass_scale_range[0]} must be strictly "
            f"positive — a scale of zero or below writes a massless or negative-mass link."
        )
    resolved_names, ls_idx_local = self.resolve_entity(self.entity_name).find_named_links_idx_local(self.links_name)
    assert len(ls_idx_local) > 0, f"No links found for {self.links_name} in {self.__class__.__name__}"
    self.ls_idx_local = torch.as_tensor(ls_idx_local, dtype=gs.tc_int, device=self.device).contiguous()
    self.entity = self.resolve_entity(self.entity_name)
    _reject_world_fixed_links(self, self.entity, self.links_name, resolved_names, ls_idx_local, "mass")
    _refuse_anchored_inertials(self, "mass")
    # Cache the unrandomized per-link mass at build time — the nominal this term's scale
    # factors multiply. ``get_links_inertial_mass`` is the per-link quantity being scaled;
    # ``get_mass()`` is the entity's total (not per-link at all).
    self._base_mass = self.entity.get_links_inertial_mass(self.ls_idx_local)  # (num_envs, n_selected)

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    # Sample at full env count and index — mirrors the rest of this module
    # and avoids shape mismatches when ``envs_idx`` is a slice subset.
    scale = sample_uniform(
        self.mass_scale_range[0],
        self.mass_scale_range[1],
        (self._env.num_envs, len(self.ls_idx_local)),
        device=self._env.device,
    )[envs_idx]
    # shift = base_mass * (scale - 1)
    mass_shift = self._base_mass[envs_idx] * (scale - 1.0)
    self.entity.set_mass_shift(mass_shift, ls_idx_local=self.ls_idx_local, envs_idx=envs_idx)

RandomizeTorqueNoise

RandomizeTorqueNoise(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Residual Force Injection (RFI) — adds random torque noise to actuated joints.

At each interval fire, samples [-rfi_scale, rfi_scale] per DOF and writes it into the TorqueOffset modifier's _torque_offset buffer, which the action term reads when applying control torques. Simulates actuator noise and unmodeled dynamics.

Must be used with mode = "interval" and an interval_range_s matching the control timestep; the event manager rejects an interval term left at interval_range_s=None.

Requires

The action term must have a TorqueOffset modifier in its chain.

PARAMETER	DESCRIPTION
action_term_name	Name of the action term whose TorqueOffset modifier should receive noise.
rfi_scale	Maximum magnitude of the random torque noise (N·m).

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 = _INTERVAL

interval_range_s class-attribute instance-attribute

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

action_term_name class-attribute instance-attribute

action_term_name: str = ''

rfi_scale class-attribute instance-attribute

rfi_scale: float = 0.5

term instance-attribute

term: ExplicitPDController | VelocityFeedforwardPDController

modifier instance-attribute

modifier: TorqueOffset

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import TorqueOffset

    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, TorqueOffset, type(self).__name__, self.action_term_name)

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    # Always sample for the full env count and index — matches the rest of
    # this module's pattern and avoids subtle shape mismatches when ``envs_idx``
    # is a slice subset (`n_sample == num_envs` but only a slice writes back).
    noise = sample_uniform(
        -self.rfi_scale,
        self.rfi_scale,
        (self._env.num_envs, self.term._n_dofs),
        device=self._env.device,
    )
    self.modifier._torque_offset[envs_idx] = noise[envs_idx]

ERFI50

ERFI50(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

ERFI-50 noise: 50% Random Actuation Offset (RAO) + 50% Random Force Injection (RFI).

For each environment, a random value r in [0, 1] is drawn at reset:

• r < erfi_ratio (RAO): a fixed scalar offset r * 3 * upper_bound is broadcast to all DOFs and held constant for the entire episode.
• r >= erfi_ratio (RFI): independent per-DOF torque noise drawn uniformly from [lower_bound, upper_bound].

Requires

The action term must have a TorqueOffset modifier.

PARAMETER	DESCRIPTION
interval_range_s	The interval range in seconds. Expected to be set to the control timestep.
action_term_name	The name of the action term to apply the noise to.
lower_bound	The lower bound of the torque noise (N·m).
upper_bound	The upper bound of the torque noise (N·m).
erfi_ratio	The fraction of envs that use RAO instead of RFI.

Reference

• Learning and Deploying Robust Locomotion Policies with Minimal Dynamics Randomization (https://arxiv.org/pdf/2209.12878)

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 = _INTERVAL

interval_range_s class-attribute instance-attribute

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

action_term_name class-attribute instance-attribute

action_term_name: str = ''

lower_bound class-attribute instance-attribute

lower_bound: float = -7.0 / 9.0

upper_bound class-attribute instance-attribute

upper_bound: float = 7.0 / 9.0

erfi_ratio class-attribute instance-attribute

erfi_ratio: float = 0.5

term instance-attribute

term: ExplicitPDController | VelocityFeedforwardPDController

modifier instance-attribute

modifier: TorqueOffset

Methods:

build

build() -> None

Source code in eden/managers/terms/events/domain_rand.py

def build(self) -> None:
    from eden.managers.modifiers.actions.actuators import TorqueOffset

    self.term = self._env.action_manager.get_term(self.action_term_name)
    self.modifier = _require_modifier(self.term, TorqueOffset, type(self).__name__, self.action_term_name)
    self._erfi_random = sample_uniform(0.0, 1.0, (self._env.num_envs,), device=self._env.device)
    self._is_rao = (self._erfi_random < self.erfi_ratio).unsqueeze(1)
    self._rao_noise = (self._erfi_random * 3.0 * self.upper_bound).unsqueeze(1).repeat(1, self.term._n_dofs)

reset

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

Source code in eden/managers/terms/events/domain_rand.py

def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    # Full-batch draws written into the selected rows (`assign_over_envs`): the auto-reset hands a
    # bool mask, and gathering or writing through it as an index is a GPU->CPU sync per op.
    # Resample per-env RAO / RFI selector.
    erfi_random = sample_uniform(0.0, 1.0, (self._env.num_envs,), device=self._env.device)
    assign_over_envs(self._erfi_random, envs_idx, erfi_random)
    assign_over_envs(self._is_rao, envs_idx, (erfi_random < self.erfi_ratio).unsqueeze(1))

    # RAO: fixed scalar noise broadcast to all DOFs.
    rao_noise = (erfi_random * 3.0 * self.upper_bound).unsqueeze(1).expand(-1, self.term._n_dofs)
    assign_over_envs(self._rao_noise, envs_idx, rao_noise)

    # Apply initial offset so first post-reset step is consistent.
    rfi_noise = sample_uniform(
        self.lower_bound, self.upper_bound, (self._env.num_envs, self.term._n_dofs), device=self._env.device
    )
    assign_over_envs(self.modifier._torque_offset, envs_idx, torch.where(self._is_rao, self._rao_noise, rfi_noise))

compute

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

Source code in eden/managers/terms/events/domain_rand.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)

    # RFI: independent per-DOF noise, drawn for every env and written into the due ones: the interval
    # tick hands a bool mask, which as an index is a GPU->CPU sync per op.
    rfi_noise = sample_uniform(
        self.lower_bound,
        self.upper_bound,
        (self._env.num_envs, self.term._n_dofs),
        device=self._env.device,
    )
    assign_over_envs(self.modifier._torque_offset, envs_idx, torch.where(self._is_rao, self._rao_noise, rfi_noise))

Functions:

Placement

Place objects on a support entity's surfaces, with constant, range, uniform or Gaussian samplers, or from a fixed list of layouts that place several entities together without overlap (PlaceFromLayouts).

Object-placement event terms and support-surface samplers.

Attributes

Classes

Gaussian2D

Gaussian2D(mu_x: float, mu_y: float, var_x: float, var_y: float)

Source code in eden/managers/terms/events/placement.py

def __init__(self, mu_x: float, mu_y: float, var_x: float, var_y: float):
    self.mu_x = float(mu_x)
    self.mu_y = float(mu_y)
    # Pre-calculate the normalization constant: 1 / (2 * pi * sigma_x * sigma_y)
    sigma_x = math.sqrt(var_x)
    sigma_y = math.sqrt(var_y)
    self.norm_const = 1.0 / (2 * math.pi * sigma_x * sigma_y)

    # Pre-calculate the inverse denominator for the exponent: 1 / (2 * var)
    # We multiply by this later instead of dividing for speed
    self.inv_var_x_2 = 0.5 / var_x
    self.inv_var_y_2 = 0.5 / var_y

Attributes

mu_x instance-attribute

mu_x = float(mu_x)

mu_y instance-attribute

mu_y = float(mu_y)

norm_const instance-attribute

norm_const = 1.0 / (2 * math.pi * sigma_x * sigma_y)

inv_var_x_2 instance-attribute

inv_var_x_2 = 0.5 / var_x

inv_var_y_2 instance-attribute

inv_var_y_2 = 0.5 / var_y

Methods:

get_probs

get_probs(points)

Compute the Gaussian weights for the given points.

PARAMETER	DESCRIPTION
points	The points to compute the weights for, of shape (N, 2)

RETURNS	DESCRIPTION
Tensor	The weights for the points, of shape (N,)

Source code in eden/managers/terms/events/placement.py

@torch.inference_mode()
def get_probs(self, points):
    """Compute the Gaussian weights for the given points.

    Parameters
    ----------
    points: torch.Tensor
        The points to compute the weights for, of shape (N, 2)

    Returns
    -------
    torch.Tensor
        The weights for the points, of shape (N,)
    """
    x, y = points.unbind(dim=-1)

    return _fused_gaussian_kernel(
        x,
        y,
        self.mu_x,
        self.mu_y,
        self.inv_var_x_2,
        self.inv_var_y_2,
        self.norm_const,
    )

PlaceGaussian

PlaceGaussian(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Event term that places an entity on a support surface via Gaussian-weighted sampling.

PARAMETER	DESCRIPTION
entity_name	The name of the entity to place
support_entity_name	The name of the support entity
mu_x	The mean of the x-coordinate in world frame
mu_y	The mean of the y-coordinate in world frame
var_x	The variance of the x-coordinate (m^2)
var_y	The variance of the y-coordinate (m^2)

Source code in eden/managers/terms/events/placement.py

def __init__(
    self,
    env: EnvBase,
    options: EventTermOptions,
):
    super().__init__(env=env, options=options)
    self.entity: Entity | None = None
    self.support_entity: Entity | None = None
    self.sampler = None

Attributes

entity_name class-attribute instance-attribute

entity_name: str = ''

support_entity_name class-attribute instance-attribute

support_entity_name: str = ''

mu_x class-attribute instance-attribute

mu_x: float = 0.0

mu_y class-attribute instance-attribute

mu_y: float = 0.0

var_x class-attribute instance-attribute

var_x: float = 0.1

var_y class-attribute instance-attribute

var_y: float = 0.1

entity instance-attribute

entity: Entity | None = None

support_entity instance-attribute

support_entity: Entity | None = None

sampler instance-attribute

sampler = None

Methods:

build

build() -> None

Source code in eden/managers/terms/events/placement.py

def build(self) -> None:
    self.entity = self.resolve_entity(self.entity_name)
    self.support_entity = self.resolve_entity(self.support_entity_name)
    self.sampler = partial(
        gaussian_sampler,
        gaussian_weights=Gaussian2D(
            mu_x=self.mu_x,
            mu_y=self.mu_y,
            var_x=self.var_x,
            var_y=self.var_y,
        ).get_probs,
    )

compute

compute(envs_idx: slice | Tensor | None)

Source code in eden/managers/terms/events/placement.py

def compute(self, envs_idx: slice | torch.Tensor | None):
    envs_idx = sanitize_envs_idx(envs_idx, self._env.num_envs)

    # NOTE: reset the orientation to the default root quat
    self.entity.set_quat(self.entity.get_default_root_pose()[1][envs_idx], envs_idx=envs_idx)
    self.entity.place_on_to(self.support_entity, sampler=self.sampler, envs_idx=envs_idx)

PlaceFromLayouts

PlaceFromLayouts(env: EnvBase, options: EventTermOptions)

Bases: EventTerm

Each reset env draws one of a fixed list of layouts and places every entity the layout names.

A layout maps entity names to (x, y, yaw_deg): the world-frame xy on the support surface (z comes off the surface) and a yaw in degrees on top of the entity's default root orientation. Drawing all the entities' poses as one layout is what keeps them apart: two entities placed by independent ranges over the same area can start overlapping, and physics then throws them apart on the first step.

PARAMETER	DESCRIPTION
layouts	The layouts, each naming the same entities. An env's layout is drawn uniformly at every reset.
support_entity_name	The entity whose surface the entities stand on.

Source code in eden/managers/terms/events/placement.py

def __init__(self, env: EnvBase, options: EventTermOptions):
    super().__init__(env=env, options=options)
    if not self.layouts:
        raise ValueError(
            "PlaceFromLayouts needs at least one layout: pass layouts=[{'<entity>': (x, y, yaw_deg), ...}, ...]."
        )
    self.entity_names = list(self.layouts[0])
    for i, layout in enumerate(self.layouts):
        if sorted(layout) != sorted(self.entity_names):
            raise ValueError(
                f"PlaceFromLayouts layout {i} names {sorted(layout)}, but layout 0 names "
                f"{sorted(self.entity_names)}; every layout must place the same entities."
            )
    poses = torch.tensor(
        [[layout[name] for name in self.entity_names] for layout in self.layouts],
        dtype=gs.tc_float,
        device=env.device,
    )  # (layouts, entities, 3)
    self._xy = poses[..., :2]
    rpy = torch.zeros_like(poses)
    rpy[..., 2] = torch.deg2rad(poses[..., 2])
    self._yaw_quat = xyz_to_quat(rpy.reshape(-1, 3), rpy=True).reshape(*poses.shape[:2], 4)

Attributes

mode class-attribute instance-attribute

mode: EventMode = EventMode.RESET

layouts class-attribute instance-attribute

layouts: list[dict[str, tuple[float, float, float]]] = []

support_entity_name class-attribute instance-attribute

support_entity_name: str = 'plane'

entity_names instance-attribute

entity_names = list(self.layouts[0])

Methods:

build

build() -> None

Source code in eden/managers/terms/events/placement.py

def build(self) -> None:
    # Touch every entity once so a misspelled name raises at env construction, not at the first reset.
    for name in (*self.entity_names, self.support_entity_name):
        self.resolve_entity(name)

compute

compute(envs_idx: slice | Tensor | None)

Source code in eden/managers/terms/events/placement.py

def compute(self, envs_idx: slice | torch.Tensor | None):
    envs_idx, n_envs = sanitize_envs_idx(envs_idx, self._env.num_envs, return_n_envs=True)
    if n_envs == 0:
        return
    layout = torch.randint(len(self.layouts), (n_envs,), device=self._env.device)
    support = self.resolve_entity(self.support_entity_name)
    for k, name in enumerate(self.entity_names):
        entity = self.resolve_entity(name)
        quat = transform_quat_by_quat(entity.get_default_root_pose()[1][envs_idx], self._yaw_quat[layout, k])
        entity.set_quat(quat, envs_idx=envs_idx)
        entity.place_on_to(
            support, sampler=partial(constant_xy_sampler, pos_xy=self._xy[layout, k]), envs_idx=envs_idx
        )

Functions:

constant_xy_sampler

constant_xy_sampler(support_data: list[ExtendedSupportData], asset_AABB: Tensor, batch_size: int = 1, *, pos_xy: tuple[float, float], support_idx: int | None = None, frame: Literal['world', 'local'] = 'world', **kwargs)

Sample a constant xy placement position on the support surface.

PARAMETER	DESCRIPTION
support_data	the support data of the asset to place TYPE: list[ExtendedSupportData]
asset_AABB	the AABB of the asset to place TYPE: Tensor
batch_size	the number of placement positions to sample (one per environment). TYPE: int DEFAULT: 1
pos_xy	the xy position of the object to place, interpreted in frame (see below). TYPE: tuple[float, float]
support_idx	which of the support entity's surfaces to place onto. None (the default) is the first that can hold the object. An integer indexes all of the support's surfaces, link by link and each link's largest first, like a Python list; out of range raises IndexError, and a surface that cannot hold the object raises ValueError. Both errors list the surfaces with their heights, centres and areas. TYPE: int | None DEFAULT: None
frame	coordinate frame for pos_xy. "world" (default) treats it as world-frame xy and reads z off the support plane at that xy. "local" treats it in the support facet's local frame (origin at an arbitrary mesh corner, axes possibly rotated) and maps it through d.transform. TYPE: Literal['world', 'local'] DEFAULT: 'world'
**kwargs	additional keyword arguments, ignored by this sampler. DEFAULT: {}
NOTE
NOTE	It is a user's responsibility to ensure the specified pos_xy is free from collision.

Source code in eden/managers/terms/events/placement.py

def constant_xy_sampler(
    support_data: list[ExtendedSupportData],
    asset_AABB: torch.Tensor,
    batch_size: int = 1,
    *,
    pos_xy: tuple[float, float],
    support_idx: int | None = None,
    frame: Literal["world", "local"] = "world",
    **kwargs,
):
    """Sample a constant xy placement position on the support surface.

    Parameters
    ----------
    support_data: list[ExtendedSupportData]
        the support data of the asset to place
    asset_AABB: torch.Tensor
        the AABB of the asset to place
    batch_size: int
        the number of placement positions to sample (one per environment).
    pos_xy: tuple[float, float]
        the xy position of the object to place, interpreted in ``frame`` (see below).
    support_idx: int | None
        which of the support entity's surfaces to place onto. ``None`` (the default) is the first that can hold
        the object. An integer indexes all of the support's surfaces, link by link and each link's largest
        first, like a Python list; out of range raises ``IndexError``, and a surface that cannot hold the
        object raises ``ValueError``. Both errors list the surfaces with their heights, centres and areas.
    frame: Literal["world", "local"]
        coordinate frame for ``pos_xy``. ``"world"`` (default) treats it as world-frame
        xy and reads z off the support plane at that xy. ``"local"`` treats it in the
        support facet's local frame (origin at an arbitrary mesh corner, axes possibly
        rotated) and maps it through ``d.transform``.
    **kwargs
        additional keyword arguments, ignored by this sampler.

    NOTE: This sampler aligns z-level to the support surface but *no guarantee* on the given pos_xy is on the surface.
    NOTE: This sampler does *not* check collisions.
          It is a user's responsibility to ensure the specified pos_xy is free from collision.
    """
    del asset_AABB  # unused
    d = _support_surface(support_data, support_idx)
    res_dir = d.normal

    if frame == "local":
        vertices_3d_on_plane = torch.zeros((batch_size, 1, 3), device=gs.device, dtype=gs.tc_float)  # (B, N, 3)
        if isinstance(pos_xy, tuple):
            vertices_3d_on_plane[:, :, 0] = pos_xy[0]
            vertices_3d_on_plane[:, :, 1] = pos_xy[1]
        else:
            vertices_3d_on_plane[:, :, :2] = pos_xy
        return transform_by_T(vertices_3d_on_plane, d.transform).squeeze(1), res_dir, None

    if frame != "world":
        raise ValueError(f"Unknown frame {frame!r}; expected 'world' or 'local'.")
    # NOTE: pos_xy is in *world* coordinates; the support plane only sets the z level
    # (the facet-local frame's origin is an arbitrary mesh corner, not the world origin).
    res_pos = torch.zeros((batch_size, 3), device=gs.device, dtype=gs.tc_float)  # (B, 3)
    if isinstance(pos_xy, tuple):
        res_pos[:, 0] = pos_xy[0]
        res_pos[:, 1] = pos_xy[1]
    else:
        res_pos[:, :2] = pos_xy
    res_pos[:, 2] = _z_on_support(res_pos, d)
    return res_pos, res_dir, None

range_xy_sampler

range_xy_sampler(support_data: list[ExtendedSupportData], asset_AABB: Tensor, batch_size: int = 1, *, range_x: tuple[float, float] | Tensor, range_y: tuple[float, float] | Tensor, support_idx: int | None = None, frame: Literal['world', 'local'] = 'world', **kwargs)

Sample an xy placement position uniformly from a rectangular range on the support surface.

PARAMETER	DESCRIPTION
support_data	the support data of the asset to place TYPE: list[ExtendedSupportData]
asset_AABB	the AABB of the asset to place TYPE: Tensor
batch_size	the number of placement positions to sample (one per environment). TYPE: int DEFAULT: 1
range_x	the range of the x-coordinate of the object to place, interpreted in frame. TYPE: tuple[float, float] | Tensor
range_y	the range of the y-coordinate of the object to place, interpreted in frame. TYPE: tuple[float, float] | Tensor
support_idx	which of the support entity's surfaces to place onto. None (the default) is the first that can hold the object. An integer indexes all of the support's surfaces, link by link and each link's largest first, like a Python list; out of range raises IndexError, and a surface that cannot hold the object raises ValueError. Both errors list the surfaces with their heights, centres and areas. TYPE: int | None DEFAULT: None
frame	coordinate frame for the ranges. "world" (default) samples world-frame xy and reads z off the support plane; "local" samples in the support facet's local frame and maps through d.transform. TYPE: Literal['world', 'local'] DEFAULT: 'world'
**kwargs	additional keyword arguments, ignored by this sampler. DEFAULT: {}
NOTE
NOTE	It is a user's responsibility to ensure the specified pos_xy is free from collision.

Source code in eden/managers/terms/events/placement.py

def range_xy_sampler(
    support_data: list[ExtendedSupportData],
    asset_AABB: torch.Tensor,
    batch_size: int = 1,
    *,
    range_x: tuple[float, float] | torch.Tensor,
    range_y: tuple[float, float] | torch.Tensor,
    support_idx: int | None = None,
    frame: Literal["world", "local"] = "world",
    **kwargs,
):
    """Sample an xy placement position uniformly from a rectangular range on the support surface.

    Parameters
    ----------
    support_data: list[ExtendedSupportData]
        the support data of the asset to place
    asset_AABB: torch.Tensor
        the AABB of the asset to place
    batch_size: int
        the number of placement positions to sample (one per environment).
    range_x: tuple[float, float] | torch.Tensor
        the range of the x-coordinate of the object to place, interpreted in ``frame``.
    range_y: tuple[float, float] | torch.Tensor
        the range of the y-coordinate of the object to place, interpreted in ``frame``.
    support_idx: int | None
        which of the support entity's surfaces to place onto. ``None`` (the default) is the first that can hold
        the object. An integer indexes all of the support's surfaces, link by link and each link's largest
        first, like a Python list; out of range raises ``IndexError``, and a surface that cannot hold the
        object raises ``ValueError``. Both errors list the surfaces with their heights, centres and areas.
    frame: Literal["world", "local"]
        coordinate frame for the ranges. ``"world"`` (default) samples world-frame xy
        and reads z off the support plane; ``"local"`` samples in the support facet's
        local frame and maps through ``d.transform``.
    **kwargs
        additional keyword arguments, ignored by this sampler.

    NOTE: This sampler aligns z-level to the support surface but *no guarantee* on the given pos_xy is on the surface.
    NOTE: This sampler does *not* check collisions.
          It is a user's responsibility to ensure the specified pos_xy is free from collision.
    """
    del asset_AABB  # unused
    d = _support_surface(support_data, support_idx)
    res_dir = d.normal

    if frame == "local":
        lower = torch.zeros(batch_size, 3, device=gs.device, dtype=gs.tc_float)
        upper = torch.zeros(batch_size, 3, device=gs.device, dtype=gs.tc_float)
        lower[:, 0], lower[:, 1] = range_x[0], range_y[0]
        upper[:, 0], upper[:, 1] = range_x[1], range_y[1]
        vertices_3d_on_plane = sample_uniform(lower, upper, (batch_size, 3), device=gs.device)  # (B, 3)
        return transform_by_T(vertices_3d_on_plane, d.transform).squeeze(1), res_dir, None

    if frame != "world":
        raise ValueError(f"Unknown frame {frame!r}; expected 'world' or 'local'.")
    # NOTE: range_x/range_y are in *world* coordinates; the support plane only sets the z level.
    lower = torch.zeros(batch_size, 2, device=gs.device, dtype=gs.tc_float)
    upper = torch.zeros(batch_size, 2, device=gs.device, dtype=gs.tc_float)
    lower[:, 0], lower[:, 1] = range_x[0], range_y[0]
    upper[:, 0], upper[:, 1] = range_x[1], range_y[1]
    res_pos = torch.zeros((batch_size, 3), device=gs.device, dtype=gs.tc_float)  # (B, 3)
    res_pos[:, :2] = sample_uniform(lower, upper, (batch_size, 2), device=gs.device)
    res_pos[:, 2] = _z_on_support(res_pos, d)
    return res_pos, res_dir, None

uniform_sampler

uniform_sampler(support_data: list[ExtendedSupportData], asset_AABB: Tensor, holes: Tensor | None = None, **kwargs)

Sample a placement position uniformly from the pre-sampled support points.

PARAMETER	DESCRIPTION
support_data	the support data of the asset to place TYPE: list[ExtendedSupportData]
asset_AABB	the AABB of the asset to place (B, 2, 3) TYPE: Tensor
holes	the AABBs of existing objects concatenated in dim=1, e.g., (B, L, 2, 3) for L holes TYPE: Tensor | None DEFAULT: None
**kwargs	additional keyword arguments, ignored by this sampler. DEFAULT: {}

Source code in eden/managers/terms/events/placement.py

def uniform_sampler(
    support_data: list[ExtendedSupportData],
    asset_AABB: torch.Tensor,
    holes: torch.Tensor | None = None,
    **kwargs,
):
    """Sample a placement position uniformly from the pre-sampled support points.

    Parameters
    ----------
    support_data: list[ExtendedSupportData]
        the support data of the asset to place
    asset_AABB: torch.Tensor
        the AABB of the asset to place (B, 2, 3)
    holes: torch.Tensor | None
        the AABBs of existing objects concatenated in dim=1, e.g., (B, L, 2, 3) for L holes
    **kwargs
        additional keyword arguments, ignored by this sampler.
    """
    offset = (asset_AABB[:, 1] - asset_AABB[:, 0]) / 2  # (B, 3)
    # consider only supports that are valid for current gravity
    support_data = [d for d in support_data if d.valid_mask.any()]

    res_pos = []
    res_dir = []
    res_valid_mask = []

    for d in support_data:
        if d.sample_points is not None:
            vertices_3d_on_plane = torch.zeros((1, d.num_sample_points, 3), device=gs.device, dtype=gs.tc_float)
            vertices_3d_on_plane[0, :, :2] = d.sample_points
            res_pos.append(transform_by_T(vertices_3d_on_plane, d.transform))  # B, N, 3
            res_dir.append(d.normal.unsqueeze(1).expand(-1, d.num_sample_points, 3))
            res_valid_mask.append(d.valid_mask.unsqueeze(1).expand(-1, d.num_sample_points))  # B, N

    if len(res_pos) > 0:
        res_pos = torch.cat(res_pos, dim=1)  # (B, sum(N), 3)
        res_dir = torch.cat(res_dir, dim=1)  # (B, sum(N), 3)
        res_valid_mask = torch.cat(res_valid_mask, dim=1)  # (B, sum(N),)

        # NOTE: filter out points inside holes
        if holes is not None:
            assert holes.ndim == 4, f"Unexpected holes shape {holes.shape}"
            if holes.shape[1] > 0:
                min_occ = holes[..., 0, :2] - offset[:, None, :2]
                max_occ = holes[..., 1, :2] + offset[:, None, :2]
                is_inside_tmp = torch.all(
                    (min_occ[:, None] <= res_pos[..., None, :2]) & (res_pos[..., None, :2] < max_occ[:, None]),
                    dim=-1,
                ).any(dim=-1)  # (B, sum(N))
                res_valid_mask = res_valid_mask & torch.logical_not(is_inside_tmp)

        valid_mask = res_valid_mask.any(dim=1)  # (B, )
        # multinomial rejects all-zero rows; make them uniform so sampling never
        # raises — the returned valid_mask flags those envs as failures downstream.
        probs = res_valid_mask.float()
        probs[probs.sum(dim=1) == 0] = 1.0
        random_true_indices = torch.multinomial(probs, 1).flatten()
        res_pos = res_pos[torch.arange(res_pos.shape[0], device=res_pos.device), random_true_indices]  # (B', 3)
        res_dir = res_dir[torch.arange(res_dir.shape[0], device=res_dir.device), random_true_indices]  # (B', 3)
        return res_pos, res_dir, valid_mask
    else:
        raise ValueError("No pre-sampled points found for the support data")

gaussian_sampler

gaussian_sampler(support_data: list[ExtendedSupportData], asset_AABB: Tensor, holes: Tensor | None = None, gaussian_weights: Callable | None = None, **kwargs)

Sample a placement position from the support points weighted by a Gaussian.

PARAMETER	DESCRIPTION
support_data	the support data of the asset to place TYPE: list[ExtendedSupportData]
asset_AABB	the AABB of the asset to place TYPE: Tensor
holes	the AABBs of existing objects concatenated in dim=1, e.g., (B, L, 2, 3) for L holes TYPE: Tensor | None DEFAULT: None
gaussian_weights	callable mapping candidate points to per-point sampling weights. If None, a default is used. TYPE: Callable | None DEFAULT: None
**kwargs	additional keyword arguments, ignored by this sampler. DEFAULT: {}

Source code in eden/managers/terms/events/placement.py

def gaussian_sampler(
    support_data: list[ExtendedSupportData],
    asset_AABB: torch.Tensor,
    holes: torch.Tensor | None = None,
    gaussian_weights: Callable | None = None,
    **kwargs,
):
    """Sample a placement position from the support points weighted by a Gaussian.

    Parameters
    ----------
    support_data: list[ExtendedSupportData]
        the support data of the asset to place
    asset_AABB: torch.Tensor
        the AABB of the asset to place
    holes: torch.Tensor | None
        the AABBs of existing objects concatenated in dim=1, e.g., (B, L, 2, 3) for L holes
    gaussian_weights: Callable | None
        callable mapping candidate points to per-point sampling weights. If None, a default is used.
    **kwargs
        additional keyword arguments, ignored by this sampler.
    """
    offset = (asset_AABB[:, 1] - asset_AABB[:, 0]) / 2  # (B, 3)
    # consider only supports that are valid for current gravity
    support_data = [d for d in support_data if d.valid_mask.any()]

    res_pos = []
    res_dir = []
    res_valid_mask = []
    res_weights = []

    for d in support_data:
        if d.sample_points is not None:
            vertices_3d_on_plane = torch.zeros((1, d.num_sample_points, 3), device=gs.device, dtype=gs.tc_float)
            vertices_3d_on_plane[0, :, :2] = d.sample_points
            pos = transform_by_T(vertices_3d_on_plane, d.transform)
            res_pos.append(pos)  # B, N, 3
            res_dir.append(d.normal.unsqueeze(1).expand(-1, d.num_sample_points, 3))

            res_weights.append(gaussian_weights(pos[..., :2]))  # B, N
            res_valid_mask.append(d.valid_mask.unsqueeze(1).expand(-1, d.num_sample_points))  # B, N

    if len(res_pos) > 0:
        res_pos = torch.cat(res_pos, dim=1)  # (B, sum(N), 3)
        res_dir = torch.cat(res_dir, dim=1)  # (B, sum(N), 3)
        res_valid_mask = torch.cat(res_valid_mask, dim=1)  # (B, sum(N),)
        res_weights = torch.cat(res_weights, dim=1)  # (B, sum(N),)

        # NOTE: filter out points inside holes
        if holes is not None:
            assert holes.ndim == 4, f"Unexpected holes shape {holes.shape}"
            if holes.shape[1] > 0:
                min_occ = holes[..., 0, :2] - offset[:, None, :2]
                max_occ = holes[..., 1, :2] + offset[:, None, :2]
                is_inside_tmp = torch.all(
                    (min_occ[:, None] <= res_pos[..., None, :2]) & (res_pos[..., None, :2] < max_occ[:, None]),
                    dim=-1,
                ).any(dim=-1)  # (B, sum(N))
                res_valid_mask = res_valid_mask & torch.logical_not(is_inside_tmp)

        valid_mask = res_valid_mask.any(dim=1)  # (B, )
        # multinomial rejects all-zero rows, which happen two ways: (1) the row has valid points
        # but their Gaussian weights underflowed to 0 — fall back to a uniform draw over just those
        # valid points (never invalid ones, e.g. inside a hole); (2) the row has no valid point at
        # all — give it all-ones so sampling doesn't raise (valid_mask flags it as a failure).
        probs = res_weights * res_valid_mask.float()
        zero_rows = probs.sum(dim=1) == 0
        probs[zero_rows] = res_valid_mask.float()[zero_rows]
        probs[~valid_mask] = 1.0
        random_true_indices = torch.multinomial(probs, 1).flatten()
        res_pos = res_pos[torch.arange(res_pos.shape[0], device=res_pos.device), random_true_indices]  # (B', 3)
        res_dir = res_dir[torch.arange(res_dir.shape[0], device=res_dir.device), random_true_indices]  # (B', 3)
        return res_pos, res_dir, valid_mask
    else:
        raise ValueError("No pre-sampled points found for the support data")

place_constant_xy

place_constant_xy(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, support_entity_name: str, pos_xy: tuple[float, float], frame: Literal['world', 'local'] = 'world', support_idx: int | None = None)

Source code in eden/managers/terms/events/placement.py

@EVENT_TERM_REGISTRY.register()
def place_constant_xy(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    support_entity_name: str,
    pos_xy: tuple[float, float],
    frame: Literal["world", "local"] = "world",
    support_idx: int | None = None,
):
    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)
    entity: Entity = env.get_entity(entity_name)
    support_entity: Entity = env.get_entity(support_entity_name)
    # NOTE: reset the orientation to the default root quat
    entity.set_quat(entity.get_default_root_pose()[1][envs_idx], envs_idx=envs_idx)
    sampler = partial(constant_xy_sampler, pos_xy=pos_xy, frame=frame, support_idx=support_idx)
    entity.place_on_to(support_entity, sampler=sampler, envs_idx=envs_idx)

place_range_xy

place_range_xy(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, support_entity_name: str, range_x: tuple[float, float], range_y: tuple[float, float], range_roll: tuple[float, float] = (0, 0), range_pitch: tuple[float, float] = (0, 0), range_yaw: tuple[float, float] = (0, 0), frame: Literal['world', 'local'] = 'world', support_idx: int | None = None)

Source code in eden/managers/terms/events/placement.py

@EVENT_TERM_REGISTRY.register()
def place_range_xy(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    support_entity_name: str,
    range_x: tuple[float, float],
    range_y: tuple[float, float],
    range_roll: tuple[float, float] = (0, 0),
    range_pitch: tuple[float, float] = (0, 0),
    range_yaw: tuple[float, float] = (0, 0),
    frame: Literal["world", "local"] = "world",
    support_idx: int | None = None,
):
    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)

    entity: Entity = env.get_entity(entity_name)
    support_entity: Entity = env.get_entity(support_entity_name)

    lower = torch.tensor(
        [range_roll[0], range_pitch[0], range_yaw[0]],
        device=env.device,
        dtype=torch.float32,
    )
    upper = torch.tensor(
        [range_roll[1], range_pitch[1], range_yaw[1]],
        device=env.device,
        dtype=torch.float32,
    )
    sampled = sample_uniform(lower, upper, (env.num_envs, 3), device=env.device)[envs_idx]
    quat = transform_quat_by_quat(entity.get_default_root_pose()[1][envs_idx], xyz_to_quat(sampled, rpy=True))
    entity.set_quat(quat, envs_idx=envs_idx)
    sampler = partial(range_xy_sampler, range_x=range_x, range_y=range_y, frame=frame, support_idx=support_idx)
    entity.place_on_to(support_entity, sampler=sampler, envs_idx=envs_idx)

place_uniform

place_uniform(env: EnvBase, envs_idx: slice | Tensor | None, *, entity_name: str, support_entity_name: str, samples_per_area: int = 100)

Source code in eden/managers/terms/events/placement.py

@EVENT_TERM_REGISTRY.register()
def place_uniform(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    entity_name: str,
    support_entity_name: str,
    samples_per_area: int = 100,
):
    envs_idx = sanitize_envs_idx(envs_idx, env.num_envs)
    entity: Entity = env.get_entity(entity_name)
    support_entity: Entity = env.get_entity(support_entity_name)
    # NOTE: reset the orientation to the default root quat
    entity.set_quat(entity.get_default_root_pose()[1][envs_idx], envs_idx=envs_idx)
    sampler = partial(uniform_sampler, samples_per_area=samples_per_area)
    entity.place_on_to(support_entity, sampler=sampler, envs_idx=envs_idx)

Restoring saved states

RESET event restoring each resetting env to a row of a :class:~eden.utils.buffers.reset_states.ResetStateBuffer.

Attributes

Classes

ResetFromStateBuffer

ResetFromStateBuffer(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Restore each resetting env to one stored row of a reset-state buffer file.

The buffer (:class:~eden.utils.buffers.reset_states.ResetStateBuffer, written by save()) names the entities it covers. Each reset writes those entities' full state, meaning the root pose and velocity in the authored frame and every joint by name, through one staged write; a fixed base built with batch_fixed_verts=True (a robot mount or fixture placed per row) gets its root pose too. Entities it does not name keep what the scene reset and the other RESET events leave. Which row each env gets is the sampler's call: uniform by default, or the curriculum term named by sampler (e.g. Success Guided Sampling), bound to this buffer's size at build.

RESET events run in ascending priority number, so give it a larger priority number than any RESET event that writes the same entities; one that runs after it overwrites the restored state. Task data stored with the rows (a goal) is read with :meth:current_column.

PARAMETER	DESCRIPTION
path	The buffer file (ResetStateBuffer.save). Loaded and checked against the built entities in build().
sampler	Config key of a curriculum term subclassing ResetConfigSampler, or None for uniform sampling.
category_weights	Draw a row's category by these weights (keyed by buffer.category_names; a category left out is never drawn), then the row uniformly within it: :class:CategoryWeightedSampler. Needs a labelled buffer (ResetStateBuffer.from_families) and no sampler.

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

path class-attribute instance-attribute

path: str | None = None

sampler class-attribute instance-attribute

sampler: str | None = None

category_weights class-attribute instance-attribute

category_weights: dict[str, float] | None = None

current_config property

current_config: Tensor

(num_envs,) row index each env was last reset to. Read it; don't write it.

Methods:

build

build() -> None

Source code in eden/managers/terms/events/state_buffer.py

def build(self) -> None:
    if not self.path:
        raise ValueError(
            "ResetFromStateBuffer needs path=: the reset-state buffer file to restore from. "
            "Fix: ResetFromStateBuffer.configure(path='<buffer>.pt'); write one with ResetStateBuffer.save()."
        )
    # In the solver's float dtype, so the reset math never mixes precisions (``precision='64'``).
    self.buffer = ResetStateBuffer.load(self.path, device=self.device).to(self.device, dtype=gs.tc_float)
    self._plans = [self._plan(name, states) for name, states in self.buffer.entities.items()]
    self._dry_run_stage()
    consumer = f"ResetFromStateBuffer(path='{self.path}')"
    self._sampler: ResetConfigSampler = self._resolve_sampler(consumer)
    self._sampler.bind(self.buffer.num_states, consumer=consumer)
    self._buffer_generation = self.buffer.generation
    #: Each env's current row. Our own copy: the sampler's buffer is live and only valid at the call.
    self._current = torch.zeros(self.num_envs, dtype=torch.long, device=self.device)

compute

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

Source code in eden/managers/terms/events/state_buffer.py

def compute(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    envs_idx = sanitize_envs_idx(envs_idx, self.num_envs, prefer_slice=False)
    if self.buffer.generation != self._buffer_generation:
        # write_rows may have relabelled rows: regroup, or the weighted sampler draws by the old families.
        if isinstance(self._sampler, CategoryWeightedSampler):
            self._sampler.relabel(self.buffer.category)
        self._buffer_generation = self.buffer.generation
    assignment = self._sampler.sample(envs_idx)
    if is_env_mask(envs_idx):
        torch.where(envs_idx, assignment, self._current, out=self._current)
        rows = slice(None)  # every env's row; the mask picks which are written, on the device
    else:
        self._current[envs_idx] = assignment[envs_idx]
        rows = envs_idx
    # Fixed roots first: the staged write's forward pass then places their links from the new base pose.
    self._write_fixed_roots(self._current, envs_idx)
    write = StagedStateWrite()
    self._stage(write, self._current, rows)
    write.apply(envs_idx)

current_column

current_column(name: str) -> Tensor

Task column name at each env's current row, (num_envs, ...); a goal stored with the states.

Source code in eden/managers/terms/events/state_buffer.py

def current_column(self, name: str) -> torch.Tensor:
    """Task column ``name`` at each env's current row, ``(num_envs, ...)``; a goal stored with the states."""
    if name not in self.buffer.columns:
        raise KeyError(
            f"ResetFromStateBuffer(path='{self.path}') has no task column '{name}'. "
            f"Columns: {', '.join(self.buffer.columns) or '(none)'}."
        )
    return self.buffer.columns[name].index_select(0, self._current)

Functions:

Contact solver parameters

Constraint-solver parameter events (contact stiffness tuning).

Attributes

Classes

Functions:

set_contact_sol_params

set_contact_sol_params(env: EnvBase, envs_idx: slice | Tensor | None, *, sol_params: tuple[float, ...], entity_links: dict[str, list[str] | None]) -> None

Set constraint-solver (solref/solimp) parameters on selected entity geoms.

A STARTUP event: applies sol_params — (timeconst, dampratio, dmin, dmax, width, mid, power), see the MuJoCo "Solver parameters" docs — to the collision geoms of the named entities, leaving every other geom (and all joint/limit constraints) at defaults. entity_links maps entity name to the link names whose geoms to modify (None = all of the entity's geoms).

Note genesis mixes a contact pair's parameters as the mean of its two geoms' values, so a stiffened geom mildly stiffens every pair it participates in (e.g. an object also contacts the table), not only the pair of interest.

envs_idx must be None: genesis stores geom sol_params unbatched, so a per-env selection would be silently ignored (the write always applies to all envs) — rejected loudly here to keep the term out of per-env randomization.

Source code in eden/managers/terms/events/solver.py

@EVENT_TERM_REGISTRY.register()
def set_contact_sol_params(
    env: EnvBase,
    envs_idx: slice | torch.Tensor | None,
    *,
    sol_params: tuple[float, ...],
    entity_links: dict[str, list[str] | None],
) -> None:
    """Set constraint-solver (solref/solimp) parameters on selected entity geoms.

    A ``STARTUP`` event: applies ``sol_params`` — ``(timeconst, dampratio, dmin,
    dmax, width, mid, power)``, see the MuJoCo "Solver parameters" docs — to the
    collision geoms of the named entities, leaving every other geom (and all
    joint/limit constraints) at defaults. ``entity_links`` maps entity name to
    the link names whose geoms to modify (``None`` = all of the entity's geoms).

    Note genesis mixes a contact pair's parameters as the mean of its two geoms'
    values, so a stiffened geom mildly stiffens *every* pair it participates in
    (e.g. an object also contacts the table), not only the pair of interest.

    ``envs_idx`` must be ``None``: genesis stores geom sol_params unbatched, so a
    per-env selection would be silently ignored (the write always applies to all
    envs) — rejected loudly here to keep the term out of per-env randomization.
    """
    if len(sol_params) != 7:
        raise ValueError(
            "sol_params must have 7 elements (timeconst, dampratio, dmin, dmax, width, mid, power); "
            f"got {len(sol_params)}"
        )
    if not (envs_idx is None or (isinstance(envs_idx, slice) and envs_idx == slice(None))):
        raise ValueError(
            "set_contact_sol_params only supports envs_idx=None: geom sol_params are unbatched in genesis, "
            "so a per-env selection would silently apply to ALL envs."
        )
    geoms_idx: list[int] = []
    for entity_name, links in entity_links.items():
        geoms = env.get_entity(entity_name).geoms
        if links is None:
            matched = [g.idx for g in geoms]
        else:
            link_names = {g.link.name for g in geoms}
            unmatched = [name for name in links if name not in link_names]
            if unmatched:
                raise ValueError(
                    f"entity '{entity_name}' has no collision geoms on links {unmatched}; "
                    f"available links with geoms: {sorted(link_names)}"
                )
            matched = [g.idx for g in geoms if g.link.name in links]
        if not matched:
            raise ValueError(f"entity '{entity_name}' has no collision geoms to apply sol_params to")
        geoms_idx += matched
    env.set_sol_params(sol_params, geoms_idx=geoms_idx)

Motion-reference initialization

RESET events that start each episode from a state of the motion reference (reference state initialization).

Motion-reference-driven event terms (RESET-mode initial-state seeding).

Attributes

Classes

ReferenceStateInitializationBase

ReferenceStateInitializationBase(env: EnvBase, options: _TermOptionsT)

Bases: EventTerm

Shared base for the motion-reference RSI (reference-state-initialization) events.

Two concrete flavours seed initial state from the motion reference at RESET: :class:ReferenceStateInitialization (the robot — a floating or immobile base) and :class:ObjectReferenceStateInitialization (a free rigid object). This base exists so the motion-end resync path can find every flavour by isinstance (not by matching one exact class name) and ask each to seed itself onto a resampled reference via :meth:teleport — so the branch knowledge lives in the term, in one place, instead of being re-derived in :class:~eden.managers.motion.resync.MotionEndResync.

A flavour does not write the simulator itself: it stages its entity's writes (:meth:_stage_reset, :meth:_stage_teleport) into a shared :class:~eden.entities.rigid.StagedStateWrite, and :meth:compute_batch / :meth:teleport_batch apply every term's writes at once. The event manager fires adjacent RSI terms through one :meth:compute_batch, so a reset seeding N entities gathers the reference once and pays one Genesis write per kind and one forward-kinematics pass, not N of each.

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

priority class-attribute instance-attribute

priority: int = 100

max_depenetration_velocity class-attribute instance-attribute

max_depenetration_velocity: float | None = None

depenetration_window_s class-attribute instance-attribute

depenetration_window_s: float = 0.1

Methods:

compute_batch classmethod

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

Seed every term's entity from the init reference state, in one write (see the class docstring).

The terms draw their perturbation noise in order, as they would one by one, and write disjoint state, so the result is what firing them one after another leaves.

Source code in eden/managers/terms/events/motion.py

@classmethod
def compute_batch(
    cls, terms: Sequence[ReferenceStateInitializationBase], envs_idx: slice | torch.Tensor | None
) -> None:
    """Seed every term's entity from the init reference state, in one write (see the class docstring).

    The terms draw their perturbation noise in order, as they would one by one, and write disjoint
    state, so the result is what firing them one after another leaves.
    """
    env = terms[0]._env
    if envs_idx is None or isinstance(envs_idx, slice):
        envs_idx = sanitize_envs_idx(envs_idx, env.num_envs, prefer_slice=False)
    # The reference is gathered for the selected rows, through int ids: a sync per op through a mask and
    # none through ids. The auto-reset's mask has its ids taken once for the whole reset, so this
    # costs nothing. The write itself takes the mask, which Genesis applies on the device.
    envs_ids = env_indices(envs_idx)
    init_states = env.motion_manager.get_reset_reference_state(envs_ids, [term.entity_name for term in terms])
    write = StagedStateWrite()
    for term in terms:
        term._stage_reset(write, envs_ids, init_states[term.entity_name])
    write.apply(envs_idx)
    for term in terms:
        term._arm_depenetration(envs_ids)

compute

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

Source code in eden/managers/terms/events/motion.py

def compute(self, envs_idx: slice | torch.Tensor | None = None):
    type(self).compute_batch([self], envs_idx)

teleport_batch classmethod

teleport_batch(terms: Sequence[ReferenceStateInitializationBase], wrap_idx: Tensor, refs: Mapping[str, MotionReferenceData]) -> None

:meth:teleport every term's entity onto refs[entity_name], in one write.

wrap_idx is int env ids, or a (num_envs,) bool mask: then every env's frame 0 is staged and the mask picks which are written, which Genesis does on the device — no nonzero(), so a mask that selects no env costs no GPU->CPU sync either (the motion-end resync calls this every step).

Source code in eden/managers/terms/events/motion.py

@classmethod
def teleport_batch(
    cls,
    terms: Sequence[ReferenceStateInitializationBase],
    wrap_idx: torch.Tensor,
    refs: Mapping[str, MotionReferenceData],
) -> None:
    """:meth:`teleport` every term's entity onto ``refs[entity_name]``, in one write.

    ``wrap_idx`` is int env ids, or a ``(num_envs,)`` bool mask: then every env's frame 0 is staged
    and the mask picks which are written, which Genesis does on the device — no ``nonzero()``, so a
    mask that selects no env costs no GPU->CPU sync either (the motion-end resync calls this every step).
    """
    rows = slice(None) if is_env_mask(wrap_idx) else wrap_idx
    write = StagedStateWrite()
    for term in terms:
        term._stage_teleport(write, rows, refs[term.entity_name])
    write.apply(wrap_idx)
    for term in terms:
        term._arm_depenetration(wrap_idx)

teleport

teleport(wrap_idx: Tensor, ref: MotionReferenceData) -> None

Snap wrap_idx envs onto ref verbatim (no reset perturbation).

The motion-end-resync counterpart of :meth:compute, staged and written the same way, but it reads the already-filled ref (this entity's MotionReferenceData) at frame 0 instead of recomputing the init reference state, and adds no noise (a wrap is mid-episode, so the entity snaps exactly onto the demo). wrap_idx is an int tensor of the env ids that wrapped, or the (num_envs,) bool mask of them.

Source code in eden/managers/terms/events/motion.py

def teleport(self, wrap_idx: torch.Tensor, ref: MotionReferenceData) -> None:
    """Snap ``wrap_idx`` envs onto ``ref`` verbatim (no reset perturbation).

    The motion-end-resync counterpart of :meth:`compute`, staged and written the same way,
    but it reads the already-filled ``ref`` (this entity's ``MotionReferenceData``) at
    frame 0 instead of recomputing the init reference state, and adds no noise (a wrap is
    mid-episode, so the entity snaps exactly onto the demo). ``wrap_idx`` is an int tensor
    of the env ids that wrapped, or the ``(num_envs,)`` bool mask of them.
    """
    type(self).teleport_batch([self], wrap_idx, {self.entity_name: ref})

ReferenceStateInitialization

ReferenceStateInitialization(env: EnvBase, options: _TermOptionsT)

Bases: ReferenceStateInitializationBase

On reset, initialise robot state from the motion reference (RSI).

Reads env.motion_manager.get_reset_reference_state(envs_idx)[entity_name] and writes it onto the entity's qpos/qvel, with optional per-DoF and base-pose perturbations. overall_noise_scale multiplies all configured perturbation ranges (pose, velocity, dofs_position) so a curriculum can anneal init noise from a single knob.

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

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

base_joint_name class-attribute instance-attribute

base_joint_name: str | None = None

pose_range class-attribute instance-attribute

pose_range: dict[str, tuple[float, float]] = {}

velocity_range class-attribute instance-attribute

velocity_range: dict[str, tuple[float, float]] = {}

dofs_position_range class-attribute instance-attribute

dofs_position_range: tuple[float, float] = (-0.1, 0.1)

overall_noise_scale class-attribute instance-attribute

overall_noise_scale: float = 1.0

Methods:

build

build()

Source code in eden/managers/terms/events/motion.py

def build(self):
    self.robot = self.resolve_entity(self.entity_name)
    # ``has_immobile_base`` covers both a fixed-base entity and one welded onto another
    # (a composite's dexterous hand): Genesis ``attach`` removes the child's root joint,
    # so there is no base to seed there either.
    if self.robot.has_immobile_base:
        if self.max_depenetration_velocity is not None:
            raise ValueError(
                f"{term_subject(self)} sets max_depenetration_velocity, but "
                f"the entity has no base to cap (it is fixed or welded to another entity). Drop the option here; "
                f"set it on the RSI term of the entity carrying the free base."
            )
        # No floating base joint to seed — only the actuated DOFs are
        # written (see the immobile-base branch in ``_stage_reset``). The write
        # goes through name-resolved Genesis indices in the MOTION dof
        # order, so no qpos-declaration-order assumption is needed.
        from eden.utils.motion_reference.lookup import get_robot_dofs_idx, untracked_entity_error

        motion_dofs_name = self._env.motion_manager.get_target_dofs_name(self.entity_name)
        if motion_dofs_name is None:
            raise untracked_entity_error(self._env, self.entity_name)
        if not motion_dofs_name:
            raise ValueError(
                f"{term_subject(self)} (immobile base) requires the motion reference to expose DOFs for it."
            )
        self._motion_dofs_idx = get_robot_dofs_idx(self._env, self.entity_name, motion_dofs_name)
        # Host copies for the staged write, which reads its indices in Python every reset.
        self._motion_dofs_host = self._motion_dofs_idx.tolist()
        # A ball joint's three DOFs make one quaternion, so the reference covers all of them or none.
        StagedStateWrite.check_dofs_pos(self.robot, self._motion_dofs_host)
        self._entity_dofs_host = range(self.robot.n_dofs)
        # See ``_soft_limits_cols``: gather the rows matching the motion DOF order so the
        # clip in ``_dofs_pos_with_noise`` can't silently apply another joint's range.
        self._soft_limits_cols = self.robot.find_named_dofs(motion_dofs_name).columns
    else:
        from eden.utils.motion_reference.lookup import assert_motion_field_present

        # The base is seeded from the reference's base pose, so an untracked entity, or a sequence
        # without one, is a config error: raise it here rather than at the first reset.
        assert_motion_field_present(self._env, self.entity_name, "base_pos_w")
        assert_motion_field_present(self._env, self.entity_name, "base_quat_w")
        base_dofs_index = self.robot.find_free_base_dofs_idx_local(self.base_joint_name)
        # Host lists for the staged write, which reads its indices in Python every reset: the free base's
        # six DOFs then the actuated ones, and the qs ``set_qpos`` writes by default (dofs_name order).
        self._base_and_dofs_host = [*base_dofs_index, *self.robot.dofs_idx_local.tolist()]
        self._qs_host = self.robot.qs_idx_local.tolist()
        self._build_depenetration(self.robot, base_dofs_index)

    range_list = [self.pose_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]
    self.pose_ranges = torch.tensor(range_list, device=self.device)
    range_list = [self.velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]
    self.velocity_ranges = torch.tensor(range_list, device=self.device)

ObjectReferenceStateInitialization

ObjectReferenceStateInitialization(env: EnvBase, options: _TermOptionsT)

Bases: ReferenceStateInitializationBase

On reset, set a free rigid object from the motion reference.

The object counterpart of :class:ReferenceStateInitialization for manipulation mimic tasks: writes the reference base pose and base velocities of a DOF-less free-floating entity (e.g. the manipulated object of a hand-object clip). Optional uniform XY perturbation matches ConTrack's object domain randomization.

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

entity_name class-attribute instance-attribute

entity_name: str = 'object'

xy_range class-attribute instance-attribute

xy_range: float = 0.0

Methods:

build

build()

Source code in eden/managers/terms/events/motion.py

def build(self):
    from eden.utils.motion_reference.lookup import assert_motion_field_present

    self.obj = self.resolve_entity(self.entity_name)
    if self.obj.has_immobile_base:
        reason = "is welded to another entity" if self.obj.is_attached else "is fixed-base"
        raise ValueError(
            f"{term_subject(self)} writes a base pose, so it requires a "
            f"free-floating entity; it {reason} and has no base DOFs."
        )
    assert_motion_field_present(self._env, self.entity_name, "base_pos_w")
    assert_motion_field_present(self._env, self.entity_name, "base_quat_w")
    # A free rigid body's 6 base DOFs and 7 base qs are its only ones; index them directly.
    self._base_dofs_host = range(6)
    self._base_qs_host = range(7)
    self._build_depenetration(self.obj, self._base_dofs_host)

Functions:

See also

• Events guide
• Event manager options