Skip to content

Modifiers

Modifiers post-process a term's input or output. Action modifiers sit between an action term and the actuator and model its imperfections (delay, friction, effort clipping, motor strength, backlash, deadband); chain several with Compose. On the observation side, noise models corrupt a term's output during training only, and the observation modifiers in a term's modifiers=[...] list (a moving average, a delay) run in every mode. The actions guide and observations guide show them in context.

from eden.managers.modifiers.actions.actuators import ActionDelay, Compose
from eden.managers.modifiers.observations.noise import UniformNoise

Base classes and registries

Base classes and registries for manager term modifiers.

Attributes

NOISE_MODEL_REGISTRY module-attribute 

NOISE_MODEL_REGISTRY = TermRegistry('NOISE_MODEL', NoiseModel, populated_by=('eden.managers.modifiers.observations',))

ACTION_MODIFIER_REGISTRY module-attribute 

ACTION_MODIFIER_REGISTRY = TermRegistry('ACTION_MODIFIER', ActionModifier, populated_by=('eden.managers.modifiers.actions',))

OBSERVATION_MODIFIER_REGISTRY module-attribute 

OBSERVATION_MODIFIER_REGISTRY = TermRegistry('OBSERVATION_MODIFIER', ObservationModifier, populated_by=('eden.managers.modifiers.observations',))

Classes

NoiseModel 

NoiseModel(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[NoiseOptions]

Base class for noise models applied to observation terms.

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

operation class-attribute instance-attribute 
operation: NoiseOperation = NoiseOperation.ADD

ActionModifier 

ActionModifier(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[ActionModifierOptions]

Base class for composable action modifiers.

Action modifiers intercept and transform processed actions (target positions/velocities) and/or control torques. They are designed to be composed together using :class:~eden.managers.modifiers.actions.actuators.Compose.

Subclasses override any of three hooks, which run at two different rates:

  • modify_processed_action: once per control step, from the controller's compute(). Its output is processed_action — the command a deployment sends.
  • modify_applied_target: once per physics step (sim_dt), from the controller's apply_actions(), on the command the actuator acts on (e.g. a delay).
  • modify_ctrl_torque: once per physics step, on the torque an explicit controller computed.
Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Methods:

build 
build(num_envs: int, device: str, entity: 'Entity' | None = None, dofs_idx_local: Tensor | None = None) -> None

Initialize internal buffers, called by the owning action term after the entity is resolved.

PARAMETER DESCRIPTION
num_envs

Number of parallel environments.

TYPE: int

device

Torch device string.

TYPE: str

entity

The entity being controlled. Needed by state-dependent modifiers.

TYPE: Entity DEFAULT: None

dofs_idx_local

Local DOF indices being controlled.

TYPE: Tensor DEFAULT: None

Source code in eden/managers/modifiers/base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def build(
    self,
    num_envs: int,
    device: str,
    entity: "Entity" | None = None,
    dofs_idx_local: torch.Tensor | None = None,
) -> None:
    """Initialize internal buffers, called by the owning action term after the entity is resolved.

    Parameters
    ----------
    num_envs : int
        Number of parallel environments.
    device : str
        Torch device string.
    entity : Entity, optional
        The entity being controlled. Needed by state-dependent modifiers.
    dofs_idx_local : torch.Tensor, optional
        Local DOF indices being controlled.
    """
    pass
compute 
compute() -> None

No-op by default. Override if the modifier needs per-step computation.

Source code in eden/managers/modifiers/base.py 
67
68
69
def compute(self) -> None:
    """No-op by default. Override if the modifier needs per-step computation."""
    pass
modify_processed_action 
modify_processed_action(processed_action: Tensor) -> Tensor

Transform target positions/velocities before PD control. Default is identity.

Runs once per control step, from the controller's compute(); what it returns becomes processed_action.

Source code in eden/managers/modifiers/base.py 
71
72
73
74
75
76
77
def modify_processed_action(self, processed_action: torch.Tensor) -> torch.Tensor:
    """Transform target positions/velocities before PD control. Default is identity.

    Runs once per control step, from the controller's ``compute()``; what it returns becomes
    ``processed_action``.
    """
    return processed_action
modify_applied_target 
modify_applied_target(target: Tensor) -> Tensor

Transform the command the actuator acts on during this physics step. Default is identity.

Runs once per physics step (sim_dt) — decimation times per control step — from the controller's apply_actions(), on the controller's whole command (processed_action, both halves of a position+velocity one), after every modify_processed_action and before the control law. A stateful override may count on exactly one call per physics step.

processed_action itself is left alone, and a deployment sends that and never calls apply_actions(), so what this hook does stays in simulation — the right place for a model of the robot's own latency, which the hardware already has.

Source code in eden/managers/modifiers/base.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
def modify_applied_target(self, target: torch.Tensor) -> torch.Tensor:
    """Transform the command the actuator acts on during this physics step. Default is identity.

    Runs once per physics step (``sim_dt``) — ``decimation`` times per control step — from the
    controller's ``apply_actions()``, on the controller's whole command (``processed_action``,
    both halves of a position+velocity one), after every ``modify_processed_action`` and before
    the control law. A stateful override may count on exactly one call per physics step.

    ``processed_action`` itself is left alone, and a deployment sends that and never calls
    ``apply_actions()``, so what this hook does stays in simulation — the right place for a
    model of the robot's own latency, which the hardware already has.
    """
    return target
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor

Transform computed torques after PD control. Default is identity.

pos_err (target - current joint position) is supplied by PD-position controllers for modifiers that need it (e.g. position deadbands); it is None for velocity-only controllers.

Source code in eden/managers/modifiers/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    """Transform computed torques after PD control. Default is identity.

    ``pos_err`` (target - current joint position) is supplied by PD-position
    controllers for modifiers that need it (e.g. position deadbands); it is
    ``None`` for velocity-only controllers.
    """
    return ctrl_torque
get 
get(cls: type) -> ActionModifier | None

Return self if it is an instance of cls, else None.

:class:~eden.managers.modifiers.actions.actuators.Compose overrides this to search its children. Callers that require the modifier to exist should check for None and raise their own actionable error.

Source code in eden/managers/modifiers/base.py 
107
108
109
110
111
112
113
114
def get(self, cls: type) -> ActionModifier | None:
    """Return self if it is an instance of ``cls``, else ``None``.

    :class:`~eden.managers.modifiers.actions.actuators.Compose` overrides this
    to search its children. Callers that require the modifier to exist should
    check for ``None`` and raise their own actionable error.
    """
    return self if isinstance(self, cls) else None

ObservationModifier 

ObservationModifier(env: EnvBase, options: _TermOptionsT)

Bases: ConfigurableTermBase[ObservationModifierOptions]

Base class for a transform the observation manager applies to one term's output, frame by frame.

Listed on an observation term as modifiers=[...]. Each frame of the term goes through its modifiers in list order, then noise, clip and scale, then the group's nan_policy, and last the history window. Unlike noise, a modifier runs in every mode (training, eval, deployment). Each term instance owns its modifier instances, so the same modifier on a policy and a critic copy of a term keeps two independent states.

compute(frame) is a read, the same rule as for an observation term. The manager calls it every time the observation is computed, which can be several times per control step: from EnvBase.reset(), at phase 19, at phase 12 for the final-observation snapshot, from get_observations(), and once at build to check the output. It returns the output for frame given the modifier's current state and changes nothing.

advance(frame) is the commit. A stateful modifier (a filter, a delay) folds frame into its state here. The manager calls it right after compute(frame), with the same frame, on the pass that also pushes the history windows (compute(update_history=True)). That pass runs once per control step (phase 19 of RLEnvBase.step, EnvBase.step, each deploy tick) and once per EnvBase.reset(). A partial env.reset(envs_idx) therefore advances every env, the same way it pushes every env's history window. So a modifier's state moves together with the history window, and a delay of k is k pushes. A read that comes after the push and before the next step (get_observations() between steps) applies the current frame once more, as the history window's read does. With backfill, right after a reset, that gives the same value; otherwise it runs one push ahead. A stateless modifier leaves advance as the no-op it is here.

Two rules keep reads and commits apart:

  • compute returns a tensor it does not write afterwards. The manager uses the output after calling advance, so a view of state that advance updates in place would change under it. Return a fresh tensor, or .clone() the state.
  • Neither method writes into frame, and advance copies what it keeps (self._last.copy_(frame), not self._last = frame). frame may be the raw output that another group's copy of the term also reads (the manager's cross-group cache), the term's own buffer, or a view of simulator state.

The other hooks: build(frame) allocates the state from a sample frame, before the env steps (allocate here, then write in place: see "Term state lifecycle" in docs/guides/terms.md). reset(envs_idx) restarts the rows of the envs whose episode restarts. It is called with the same indices as the term's history window: None for every env, otherwise an index or boolean-mask tensor to index with.

A modifier returns a frame of the shape, dtype and device it received. The manager checks this at build, because scale, history and concatenation are all sized from the term's raw output.

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Methods:

build 
build(frame: Tensor) -> None

Allocate per-env state sized like frame. No-op by default.

PARAMETER DESCRIPTION
frame

A sample of the frames this modifier will receive, (num_envs, ...): the term's output from the manager's build-time shape probe, passed through the modifiers listed before this one. Size state from it (torch.zeros_like(frame) gets shape, dtype and device right). Do not seed state from its values: they are read from the scene before any reset.

TYPE: Tensor

Source code in eden/managers/modifiers/base.py 
169
170
171
172
173
174
175
176
177
178
179
def build(self, frame: torch.Tensor) -> None:
    """Allocate per-env state sized like ``frame``. No-op by default.

    Parameters
    ----------
    frame : torch.Tensor
        A sample of the frames this modifier will receive, ``(num_envs, ...)``: the term's output
        from the manager's build-time shape probe, passed through the modifiers listed before this
        one. Size state from it (``torch.zeros_like(frame)`` gets shape, dtype and device right).
        Do not seed state from its values: they are read from the scene before any reset.
    """
reset 
reset(envs_idx: Tensor | None = None) -> None

Restart the state of the envs in envs_idx (None: every env). No-op by default.

Source code in eden/managers/modifiers/base.py 
181
182
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    """Restart the state of the envs in ``envs_idx`` (``None``: every env). No-op by default."""
compute 
compute(frame: Tensor) -> Tensor

Return this frame's output given the current state, without changing the state or frame.

Source code in eden/managers/modifiers/base.py 
184
185
186
def compute(self, frame: torch.Tensor) -> torch.Tensor:
    """Return this frame's output given the current state, without changing the state or ``frame``."""
    raise NotImplementedError(f"{type(self).__name__} must implement compute(frame).")
advance 
advance(frame: Tensor) -> None

Fold frame into the state: called once per control step, right after compute(frame).

No-op by default, which is right for a stateless modifier.

Source code in eden/managers/modifiers/base.py 
188
189
190
191
192
def advance(self, frame: torch.Tensor) -> None:
    """Fold ``frame`` into the state: called once per control step, right after ``compute(frame)``.

    No-op by default, which is right for a stateless modifier.
    """

Action modifiers

Composable action modifiers for action terms.

Action modifiers sit between action terms and the physics engine. They intercept and transform processed actions (target positions/velocities) and/or control torques via composable operations like clipping, delay, friction, and motor strength scaling.

Modifiers are designed to be composed together using :class:Compose, similar to torchvision.transforms.Compose::

modifiers = Compose.configure(
    modifiers=[
        ActionDelay.configure(delay_range_steps=(1, 3)),
        FrictionModel.configure(),
        EffortClip.configure(),
    ]
)

Each modifier implements up to three hooks:

  • modify_processed_action: transforms the target positions/velocities once per control step, before PD control (e.g., envelope clip, backlash).
  • modify_applied_target: transforms the command the actuator acts on, once per physics step (e.g., delay).
  • modify_ctrl_torque: transforms the computed torques after PD control (e.g., friction, effort clipping, motor strength scaling).

Modifiers that need motor spec parameters (friction, T-N curve limits) read them automatically from the entity's actuator spec (dofs_spec) at build time. Explicit overrides can still be passed to use custom values.

Attributes

CURVE_SPEED_DENOM_MIN module-attribute 

CURVE_SPEED_DENOM_MIN = 1e-06

MOTOR_STALL_VEL_THRESHOLD module-attribute 

MOTOR_STALL_VEL_THRESHOLD = 0.01

Classes

Compose 

Compose(env: EnvBase, options: ActionModifierOptions)

Bases: ActionModifier

Chains multiple action modifiers in sequence.

Each hook calls the children in list order, but the three hooks run at different points of a step (see :class:~eden.managers.modifiers.ActionModifier): every modify_processed_action once per control step in compute(), then every modify_applied_target once per physics step in apply_actions(), then every modify_ctrl_torque after the control law. A physics-step modifier such as :class:ActionDelay therefore acts after every control-step one such as :class:EnvelopeClip, wherever each is listed. Listing it first logs a warning, since the list would then read in a different order from the one that runs.

PARAMETER DESCRIPTION
modifiers

Ordered sequence of modifier configurations.

TYPE: tuple[ActionModifierOptions, ...] | list[ActionModifierOptions]

Example

::

Compose.configure(
    modifiers=[
        ActionDelay.configure(delay_range_steps=(1, 3)),
        FrictionModel.configure(),
        EffortClip.configure(),
    ]
)
Source code in eden/managers/modifiers/actions/actuators.py 
248
249
250
251
252
253
254
255
256
257
def __init__(self, env: EnvBase, options: ActionModifierOptions):
    super().__init__(env=env, options=options)
    self._modifiers: list[ActionModifier] = []
    for mod_options in self.modifiers:
        # Normalize plain dicts (e.g. from YAML/JSON config loading) to ActionModifierOptions.
        if isinstance(mod_options, dict):
            mod_options = ActionModifierOptions(**mod_options)
        mod = ACTION_MODIFIER_REGISTRY.get(mod_options.name)(env=env, options=mod_options)
        self._modifiers.append(mod)
    self._warn_if_listed_out_of_run_order()

Attributes

modifiers class-attribute instance-attribute 
modifiers: tuple[ActionModifierOptions, ...] | list[ActionModifierOptions] = ()

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
282
283
284
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    for mod in self._modifiers:
        mod.build(num_envs, device, entity=entity, dofs_idx_local=dofs_idx_local)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
286
287
288
def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> None:
    for mod in self._modifiers:
        mod.reset(envs_idx)
modify_processed_action 
modify_processed_action(processed_action: Tensor) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
290
291
292
293
def modify_processed_action(self, processed_action: torch.Tensor) -> torch.Tensor:
    for mod in self._modifiers:
        processed_action = mod.modify_processed_action(processed_action)
    return processed_action
modify_applied_target 
modify_applied_target(target: Tensor) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
295
296
297
298
def modify_applied_target(self, target: torch.Tensor) -> torch.Tensor:
    for mod in self._modifiers:
        target = mod.modify_applied_target(target)
    return target
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
300
301
302
303
304
305
306
307
308
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    for mod in self._modifiers:
        ctrl_torque = mod.modify_ctrl_torque(ctrl_torque, dofs_vel, pos_err=pos_err)
    return ctrl_torque
get 
get(cls: type) -> ActionModifier | None

Return the first child modifier that is an instance of cls, or None.

Recurses into nested Compose chains via each child's get.

Source code in eden/managers/modifiers/actions/actuators.py 
310
311
312
313
314
315
316
317
318
319
def get(self, cls: type) -> ActionModifier | None:
    """Return the first child modifier that is an instance of ``cls``, or ``None``.

    Recurses into nested ``Compose`` chains via each child's ``get``.
    """
    for mod in self._modifiers:
        found = mod.get(cls)
        if found is not None:
            return found
    return None

ActionDelay 

ActionDelay(env: EnvBase, options: ActionModifierOptions)

Bases: ActionModifier

Delays the command the actuator acts on by a random number of physics steps, drawn per episode.

The delay is applied at the actuator, once per physics step (sim_dt), not once per control step: every apply_actions() pushes the controller's command into a history and drives the actuator with the entry d physics steps old. So a latency shorter than a control step can be expressed — at sim_dt=0.005, decimation=4, d=2 keeps the first two physics steps of every control step on the previous command. This matches the physics-step unit of Isaac Lab's DelayedPDActuatorCfg and mjlab's actuator delay_* fields.

Each env draws d uniformly from the inclusive range when it resets and keeps it until its next reset. The range is given either in physics steps (delay_range_steps) or in seconds (delay_range_s), but not both; with neither, commands pass through. After a reset the history is backfilled with the first new command, so the actuator never acts on a command from before the reset.

processed_action stays undelayed. It is the command the policy issued, and what a deployment sends. The delay acts only inside apply_actions(), after every modify_processed_action modifier wherever it sits in a :class:Compose chain (which warns when it is listed before one). It models the simulated actuator's latency, and no deployment applies it: eden deploy never calls apply_actions(), so a real robot has only its own latency, and the MuJoCo sim-to-sim robot has none. A policy trained against a delay therefore runs there with less latency than it trained with. With both ends of the range equal, the lag was fixed, so it expects that lag on every step, and a MuJoCo sim-to-sim run does not reproduce it.

PARAMETER DESCRIPTION
delay_range_steps

The inclusive delay range (min, max) in physics steps — the unit of Isaac Lab's min_delay / max_delay.

TYPE: tuple[int, int] | None

delay_range_s

The inclusive delay range in seconds instead. Converted to physics steps at build, so each end must be a whole multiple of sim_dt.

TYPE: tuple[float, float] | None

Source code in eden/managers/modifiers/actions/actuators.py 
373
374
375
376
377
378
def __init__(self, env: EnvBase, options: ActionModifierOptions):
    super().__init__(env=env, options=options)
    self._min_steps = 0
    self._max_steps = 0
    self._history: CircularBuffer | None = None
    self._delay_steps: torch.Tensor | None = None

Attributes

delay_range_steps class-attribute instance-attribute 
delay_range_steps: tuple[int, int] | None = None
delay_range_s class-attribute instance-attribute 
delay_range_s: tuple[float, float] | None = None
resolved_range_steps property 
resolved_range_steps: tuple[int, int]

The inclusive delay range in physics steps, resolved at build from whichever field was set.

delay_steps property 
delay_steps: Tensor | None

Each env's current delay in physics steps, shape (num_envs,); None when the range is (0, 0).

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
433
434
435
436
437
438
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    self._min_steps, self._max_steps = self._resolve_delay_steps()
    if self._max_steps > 0:
        self._history = CircularBuffer(max_len=self._max_steps + 1, batch_size=num_envs, device=device)
        self._delay_steps = torch.full((num_envs,), self._min_steps, dtype=torch.long, device=device)
        self._redraw_delays(slice(None))
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
452
453
454
455
456
def reset(self, envs_idx: (slice | torch.Tensor) | None = None) -> None:
    if self._history is None:
        return
    self._history.reset(envs_idx)
    self._redraw_delays(slice(None) if envs_idx is None else envs_idx)
modify_applied_target 
modify_applied_target(target: Tensor) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
458
459
460
461
462
def modify_applied_target(self, target: torch.Tensor) -> torch.Tensor:
    if self._history is None:
        return target
    self._history.append(target)
    return self._history[self._delay_steps]

FrictionModel 

FrictionModel(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Applies tanh-based static + viscous dynamic friction to torques.

By default, friction parameters are read from the entity's actuator spec (dofs_spec). Explicit values override the entity defaults.

PARAMETER DESCRIPTION
static_friction

Static friction coefficient per DOF, or a scalar broadcast. None → read from entity spec.

TYPE: LaxFArrayType | None

dynamic_friction

Dynamic (viscous) friction coefficient. None → read from entity spec.

TYPE: LaxFArrayType | None

friction_activation_vel

Velocity at which static friction is fully activated [rad/s]. None → read from entity spec.

TYPE: LaxFArrayType | None

friction_offset

Constant friction offset torque per DOF (or scalar broadcast).

TYPE: LaxFArrayType

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

static_friction class-attribute instance-attribute 
static_friction: LaxFArrayType | None = None
dynamic_friction class-attribute instance-attribute 
dynamic_friction: LaxFArrayType | None = None
friction_activation_vel class-attribute instance-attribute 
friction_activation_vel: LaxFArrayType | None = None
friction_offset class-attribute instance-attribute 
friction_offset: LaxFArrayType = 0.0

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    self._static_friction = _get_param_tensor_from_entity(
        self.static_friction,
        spec_attr="static_friction",
        entity=entity,
        dofs_idx_local=dofs_idx_local,
        device=device,
    )
    self._dynamic_friction = _get_param_tensor_from_entity(
        self.dynamic_friction,
        spec_attr="dynamic_friction",
        entity=entity,
        dofs_idx_local=dofs_idx_local,
        device=device,
    )
    self._friction_activation_vel = _get_param_tensor_from_entity(
        self.friction_activation_vel,
        spec_attr="friction_activation_speed",
        entity=entity,
        dofs_idx_local=dofs_idx_local,
        device=device,
    )
    self._friction_offset_tensor = _explicit_lax_to_tensor(
        self.friction_offset,
        param_name="friction_offset",
        dofs_idx_local=dofs_idx_local,
        device=device,
    )
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    return _tanh_friction(
        ctrl_torque,
        dofs_vel,
        self._static_friction,
        self._dynamic_friction,
        self._friction_activation_vel,
        self._friction_offset_tensor,
    )

EffortClip 

EffortClip(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Clips torques to motor torque-speed (T-N) curve limits.

By default, T-N curve parameters are read from the entity's actuator spec (dofs_spec). Explicit values override the entity defaults.

PARAMETER DESCRIPTION
driving_torque_limit

Maximum torque when torque and velocity are in the same direction. None → read from entity spec.

TYPE: LaxFArrayType | None

braking_torque_limit

Maximum torque when torque and velocity are in opposite directions. None → read from entity spec.

TYPE: LaxFArrayType | None

full_torque_speed

Speed below which full torque is available [rad/s]. None → read from entity spec.

TYPE: LaxFArrayType | None

no_load_speed

Speed at which available torque drops to zero [rad/s]. None → read from entity spec.

TYPE: LaxFArrayType | None

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

driving_torque_limit class-attribute instance-attribute 
driving_torque_limit: LaxFArrayType | None = None
braking_torque_limit class-attribute instance-attribute 
braking_torque_limit: LaxFArrayType | None = None
full_torque_speed class-attribute instance-attribute 
full_torque_speed: LaxFArrayType | None = None
no_load_speed class-attribute instance-attribute 
no_load_speed: LaxFArrayType | None = None

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    def _resolve(param_value, spec_attr):
        return _get_param_tensor_from_entity(
            param_value,
            spec_attr=spec_attr,
            entity=entity,
            dofs_idx_local=dofs_idx_local,
            device=device,
        )

    self._driving_torque_limit = _resolve(self.driving_torque_limit, "driving_torque_limit")
    self._braking_torque_limit = _resolve(self.braking_torque_limit, "braking_torque_limit")
    self._full_torque_speed = _resolve(self.full_torque_speed, "full_torque_speed")
    self._no_load_speed = _resolve(self.no_load_speed, "no_load_speed")
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
579
580
581
582
583
584
585
586
587
588
589
590
591
592
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    return _clip_effort(
        ctrl_torque,
        dofs_vel,
        self._driving_torque_limit,
        self._braking_torque_limit,
        self._full_torque_speed,
        self._no_load_speed,
    )

MotorStrength 

MotorStrength(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Scales control torques by a motor strength multiplier.

PARAMETER DESCRIPTION
motor_strength

Per-DOF multiplier, or a scalar broadcast to every controlled DOF.

TYPE: LaxFArrayType

Note

This modifier provides a deterministic motor strength multiplier. To randomize it, use the :class:~eden.managers.terms.events.domain_rand.RandomizeMotorStrength event term instead.

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

motor_strength class-attribute instance-attribute 
motor_strength: LaxFArrayType = 1.0
motor_strength_realized property 
motor_strength_realized: Tensor

The realized per-env torque-scale buffer, shape (num_envs, n_dof).

Named *_realized rather than motor_strength to avoid shadowing the motor_strength configuration field (the nominal value, exposed as :attr:motor_strength_baseline).

motor_strength_baseline property 
motor_strength_baseline: Tensor

The configured nominal torque-scale row, shape (n_dof,) — the value reset restores.

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
612
613
614
615
616
617
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if dofs_idx_local is None:
        raise ValueError("MotorStrength requires dofs_idx_local to be passed in build().")
    n_dof = len(dofs_idx_local)
    self._motor_strength_row = _per_dof_float_row(self.motor_strength, n_dof, device, param_name="motor_strength")
    self._motor_strength = self._motor_strength_row.unsqueeze(0).repeat(num_envs, 1)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
619
620
621
622
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._motor_strength[envs_idx] = self._motor_strength_row
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
639
640
641
642
643
644
645
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    return ctrl_torque * self._motor_strength

TorqueOffset 

TorqueOffset(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Adds a constant offset to the control torques.

PARAMETER DESCRIPTION
torque_offset

Per-DOF offset, or a scalar broadcast to every controlled DOF.

TYPE: LaxFArrayType

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

torque_offset class-attribute instance-attribute 
torque_offset: LaxFArrayType = 0.0

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
660
661
662
663
664
665
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if dofs_idx_local is None:
        raise ValueError("TorqueOffset requires dofs_idx_local to be passed in build().")
    n_dof = len(dofs_idx_local)
    self._torque_offset_row = _per_dof_float_row(self.torque_offset, n_dof, device, param_name="torque_offset")
    self._torque_offset = self._torque_offset_row.unsqueeze(0).repeat(num_envs, 1)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
667
668
669
670
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._torque_offset[envs_idx] = self._torque_offset_row
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
672
673
674
675
676
677
678
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    return ctrl_torque + self._torque_offset

ConstantTorqueKick 

ConstantTorqueKick(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Adds a flat directional torque on top of computed control torque.

This acts like a minimum drive effort: when the joint has a nonzero position error, the modifier adds torque_kick in the direction of that error. It is useful for matching actuators that move sharply through small errors or have firmware/static-friction compensation that a pure proportional term cannot reproduce.

PARAMETER DESCRIPTION
torque_kick

Nonnegative torque magnitude per DOF, or a scalar broadcast.

TYPE: LaxFArrayType

activation_epsilon

Position-error magnitude (rad) below which no torque kick is added. This only applies when pos_err is available.

TYPE: LaxFArrayType

direction_source

Direction signal for the torque kick. "pos_err" pushes toward the target when a position error is supplied and falls back to "ctrl_torque" without one. "ctrl_torque" follows the sign of the already computed torque.

TYPE: str

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

torque_kick class-attribute instance-attribute 
torque_kick: LaxFArrayType = 0.0
activation_epsilon class-attribute instance-attribute 
activation_epsilon: LaxFArrayType = 0.0
direction_source class-attribute instance-attribute 
direction_source: str = 'pos_err'

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if dofs_idx_local is None:
        raise ValueError("ConstantTorqueKick requires dofs_idx_local to be passed in build().")
    if self.direction_source not in {"pos_err", "ctrl_torque"}:
        raise ValueError("ConstantTorqueKick direction_source must be 'pos_err' or 'ctrl_torque'.")
    n_dof = len(dofs_idx_local)
    torque_kick_row = _per_dof_float_row(self.torque_kick, n_dof, device, param_name="torque_kick")
    if (torque_kick_row < 0.0).any().item():
        raise ValueError("ConstantTorqueKick torque_kick must be nonnegative.")
    self._torque_kick_row = torque_kick_row
    self._activation_epsilon_row = _per_dof_float_row(
        self.activation_epsilon,
        n_dof,
        device,
        param_name="activation_epsilon",
    ).abs()
    self._torque_kick = self._torque_kick_row.unsqueeze(0).repeat(num_envs, 1)
    self._activation_epsilon = self._activation_epsilon_row.unsqueeze(0).repeat(num_envs, 1)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
728
729
730
731
732
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._torque_kick[envs_idx] = self._torque_kick_row
    self._activation_epsilon[envs_idx] = self._activation_epsilon_row
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
734
735
736
737
738
739
740
741
742
743
744
745
746
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    if self.direction_source == "pos_err" and pos_err is not None:
        direction_signal = pos_err
        active = pos_err.abs() > self._activation_epsilon
    else:
        direction_signal = ctrl_torque
        active = direction_signal != 0.0
    return ctrl_torque + torch.sign(direction_signal) * self._torque_kick * active.to(ctrl_torque.dtype)

Deadband 

Deadband(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Zeroes PD control torque inside a position-error deadband.

No torque is applied when the instantaneous position error |pos_err| is smaller than deadband_epsilon. Only applicable to PD-position controllers that pass pos_err into :meth:modify_ctrl_torque — velocity-only controllers do not and the modifier is a no-op there.

The threshold is stored as a per-env per-DOF tensor so it can be domain-randomized (see :class:~eden.managers.terms.events.domain_rand.RandomizeDeadbandEpsilon).

PARAMETER DESCRIPTION
deadband_epsilon

Position-error magnitude (rad) below which torque is zeroed. Per :class:~genesis.typing.LaxFArrayType, a scalar broadcasts; an iterable must match len(dofs_idx_local).

TYPE: LaxFArrayType

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

deadband_epsilon class-attribute instance-attribute 
deadband_epsilon: LaxFArrayType = 0.0

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
771
772
773
774
775
776
777
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if dofs_idx_local is None:
        raise ValueError("Deadband requires dofs_idx_local to be passed in build().")
    n_dof = len(dofs_idx_local)
    row = _per_dof_float_row(self.deadband_epsilon, n_dof, device, param_name="deadband_epsilon")
    self._deadband_epsilon_row = row
    self._deadband_epsilon = row.unsqueeze(0).repeat(num_envs, 1)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
779
780
781
782
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._deadband_epsilon[envs_idx] = self._deadband_epsilon_row
modify_ctrl_torque 
modify_ctrl_torque(ctrl_torque: Tensor, dofs_vel: Tensor, pos_err: Tensor | None = None) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
784
785
786
787
788
789
790
791
792
def modify_ctrl_torque(
    self,
    ctrl_torque: torch.Tensor,
    dofs_vel: torch.Tensor,
    pos_err: torch.Tensor | None = None,
) -> torch.Tensor:
    if pos_err is None:
        return ctrl_torque
    return ctrl_torque.masked_fill(pos_err.abs() < self._deadband_epsilon, 0.0)

GearBacklash 

GearBacklash(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Adds direction-dependent gear slop to position targets before PD control.

This models lost motion in the drivetrain, where the load settles on the side of the gear gap selected by the latest commanded motion. After a positive target move the effective motor-side target becomes target + backlash; after a negative move it becomes target - backlash. Small target changes below reversal_threshold do not change the selected side, which prevents command noise from flipping the gap state.

PARAMETER DESCRIPTION
backlash

One-sided position offset in radians. A scalar broadcasts to all controlled DOFs; an iterable must match len(dofs_idx_local).

TYPE: LaxFArrayType

reversal_threshold

Minimum target-position change in radians required to select a new backlash side.

TYPE: LaxFArrayType

takeup_rate

Fraction of the remaining gap state to take up on each action update. 1.0 switches sides instantly; values in [0, 1) create a smooth slosh transition over multiple control updates.

TYPE: LaxFArrayType

initial_side

Initial gap side on reset. 0.0 starts centered until the first command movement selects a side.

TYPE: LaxFArrayType

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

backlash class-attribute instance-attribute 
backlash: LaxFArrayType = 0.0
reversal_threshold class-attribute instance-attribute 
reversal_threshold: LaxFArrayType = 0.0
takeup_rate class-attribute instance-attribute 
takeup_rate: LaxFArrayType = 1.0
initial_side class-attribute instance-attribute 
initial_side: LaxFArrayType = 0.0

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if dofs_idx_local is None:
        raise ValueError("GearBacklash requires dofs_idx_local to be passed in build().")
    n_dof = len(dofs_idx_local)
    self._backlash_row = _per_dof_float_row(self.backlash, n_dof, device, param_name="backlash")
    if (self._backlash_row < 0.0).any().item():
        raise ValueError("GearBacklash backlash must be nonnegative.")
    self._reversal_threshold_row = _per_dof_float_row(
        self.reversal_threshold,
        n_dof,
        device,
        param_name="reversal_threshold",
    ).abs()
    self._takeup_rate_row = _per_dof_float_row(self.takeup_rate, n_dof, device, param_name="takeup_rate")
    if ((self._takeup_rate_row < 0.0) | (self._takeup_rate_row > 1.0)).any().item():
        raise ValueError("GearBacklash takeup_rate must be in [0, 1].")
    self._initial_side_row = _per_dof_float_row(self.initial_side, n_dof, device, param_name="initial_side").clamp(
        min=-1.0,
        max=1.0,
    )

    self._backlash = self._backlash_row.unsqueeze(0).repeat(num_envs, 1)
    self._reversal_threshold = self._reversal_threshold_row.unsqueeze(0).repeat(num_envs, 1)
    self._takeup_rate = self._takeup_rate_row.unsqueeze(0).repeat(num_envs, 1)
    self._initial_side = self._initial_side_row.unsqueeze(0).repeat(num_envs, 1)
    self._backlash_side = self._initial_side.clone()
    self._prev_target = torch.zeros(num_envs, n_dof, device=device)
    self._initialized = torch.zeros(num_envs, n_dof, dtype=torch.bool, device=device)
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
858
859
860
861
862
863
864
865
866
867
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._backlash[envs_idx] = self._backlash_row
    self._reversal_threshold[envs_idx] = self._reversal_threshold_row
    self._takeup_rate[envs_idx] = self._takeup_rate_row
    self._initial_side[envs_idx] = self._initial_side_row
    self._backlash_side[envs_idx] = self._initial_side[envs_idx]
    self._prev_target[envs_idx] = 0.0
    self._initialized[envs_idx] = False
modify_processed_action 
modify_processed_action(processed_action: Tensor) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
def modify_processed_action(self, processed_action: torch.Tensor) -> torch.Tensor:
    first_update = ~self._initialized
    delta = processed_action - self._prev_target
    moving_positive = delta > self._reversal_threshold
    moving_negative = delta < -self._reversal_threshold
    requested_side = torch.where(
        moving_positive,
        torch.ones_like(self._backlash_side),
        torch.where(moving_negative, -torch.ones_like(self._backlash_side), self._backlash_side),
    )
    requested_side = torch.where(first_update, self._initial_side, requested_side)
    self._backlash_side += (requested_side - self._backlash_side) * self._takeup_rate
    self._prev_target[:] = processed_action
    self._initialized[:] = True
    return processed_action + self._backlash_side * self._backlash

EnvelopeClip 

EnvelopeClip(env: EnvBase, options: _TermOptionsT)

Bases: ActionModifier

Clips target positions so PD torques stay within the motor torque-speed envelope.

Back-solves the motor torque-speed curve to find the range of target positions that yield feasible torques, then clips targets into that range. This ensures the PD controller never commands torques outside the motor's physical capability.

The torque-speed envelope is defined by four parameters::

    Torque, N·m
        ^
Y2──────|
        |──────────────Y1
        |              │\\
        |              │ \\
        |              │  \\
        |              |   \\
--------+--------------|------> velocity, rad/s
                      X1   X2

Given current joint state (q, dq) and PD gains (kp, kd), the modifier computes the maximum and minimum feasible torques from the T-N curve, then back-solves for the corresponding target position bounds::

target_low  = (tau_low  + kd * dq) / kp + q
target_high = (tau_high + kd * dq) / kp + q

By default, T-N curve parameters are read from the entity's actuator spec (dofs_spec). Explicit values override the entity defaults.

PARAMETER DESCRIPTION
driving_torque_limit

(Y1) Maximum torque when torque and velocity are in the same direction. None → read from entity spec.

TYPE: LaxFArrayType | None

braking_torque_limit

(Y2) Maximum torque when torque and velocity are in opposite directions. None → read from entity spec.

TYPE: LaxFArrayType | None

full_torque_speed

(X1) Speed below which full torque is available [rad/s]. None → read from entity spec.

TYPE: LaxFArrayType | None

no_load_speed

(X2) Speed at which available torque drops to zero [rad/s]. None → read from entity spec.

TYPE: LaxFArrayType | None

Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

driving_torque_limit class-attribute instance-attribute 
driving_torque_limit: LaxFArrayType | None = None
braking_torque_limit class-attribute instance-attribute 
braking_torque_limit: LaxFArrayType | None = None
full_torque_speed class-attribute instance-attribute 
full_torque_speed: LaxFArrayType | None = None
no_load_speed class-attribute instance-attribute 
no_load_speed: LaxFArrayType | None = None

Methods:

build 
build(num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
def build(self, num_envs: int, device: str, entity=None, dofs_idx_local=None) -> None:
    if entity is None:
        raise ValueError("EnvelopeClip requires entity to be passed in build().")
    self._entity = entity
    # The selector for the entity getters below: a `range` when the DOFs are contiguous, Genesis'
    # sync-free slice path — `modify_processed_action` reads the joint state every control step.
    self._dofs_idx_local = None if dofs_idx_local is None else as_genesis_index(dofs_idx_local)

    # Cache PD gains
    self._kp = entity.get_dofs_kp(dofs_idx_local=self._dofs_idx_local)
    self._kd = entity.get_dofs_kd(dofs_idx_local=self._dofs_idx_local)

    # T-N curve parameters — read from entity spec unless explicitly overridden
    def _resolve(param_value, spec_attr):
        return _get_param_tensor_from_entity(
            param_value,
            spec_attr=spec_attr,
            entity=entity,
            dofs_idx_local=dofs_idx_local,
            device=device,
        )

    self._motor_y1 = _resolve(self.driving_torque_limit, "driving_torque_limit")
    self._motor_y2 = _resolve(self.braking_torque_limit, "braking_torque_limit")
    self._motor_x1 = _resolve(self.full_torque_speed, "full_torque_speed")
    self._motor_x2 = _resolve(self.no_load_speed, "no_load_speed")
reset 
reset(envs_idx: slice | Tensor | None = None) -> None
Source code in eden/managers/modifiers/actions/actuators.py 
965
966
967
968
969
def reset(self, envs_idx: slice | torch.Tensor | None = None) -> None:
    if envs_idx is None:
        envs_idx = slice(None)
    self._kp[envs_idx] = self._entity.get_dofs_kp(dofs_idx_local=self._dofs_idx_local, envs_idx=envs_idx)
    self._kd[envs_idx] = self._entity.get_dofs_kd(dofs_idx_local=self._dofs_idx_local, envs_idx=envs_idx)
modify_processed_action 
modify_processed_action(processed_action: Tensor) -> Tensor
Source code in eden/managers/modifiers/actions/actuators.py 
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
def modify_processed_action(self, processed_action: torch.Tensor) -> torch.Tensor:
    joint_pos = self._entity.get_dofs_pos(dofs_idx_local=self._dofs_idx_local)
    joint_vel = self._entity.get_dofs_vel(dofs_idx_local=self._dofs_idx_local)

    abs_dq = joint_vel.abs()
    over = torch.clamp(abs_dq - self._motor_x1, min=0.0)
    denom = torch.clamp(self._motor_x2 - self._motor_x1, min=CURVE_SPEED_DENOM_MIN)

    # Positive torque limit (tau_high >= 0)
    base_pos = torch.where(
        abs_dq <= MOTOR_STALL_VEL_THRESHOLD,
        self._motor_y2,
        torch.where(joint_vel >= 0.0, self._motor_y1, self._motor_y2),
    )
    tau_high = torch.clamp(base_pos - (base_pos / denom) * over, min=0.0)

    # Negative torque limit (tau_low <= 0)
    base_neg = torch.where(
        abs_dq <= MOTOR_STALL_VEL_THRESHOLD,
        -self._motor_y2,
        torch.where(joint_vel >= 0.0, -self._motor_y2, -self._motor_y1),
    )
    tau_low = torch.clamp(base_neg + (-base_neg / denom) * over, max=0.0)

    # Back-solve: tau = kp * (target - q) - kd * dq
    kd_dq = self._kd * joint_vel
    target_low = (tau_low + kd_dq) / self._kp + joint_pos
    target_high = (tau_high + kd_dq) / self._kp + joint_pos

    return torch.clip(processed_action, target_low, target_high)

Functions:

Observation noise

Observation noise modifiers (constant, uniform, Gaussian).

Attributes

Classes

ConstantNoise 

ConstantNoise(env: EnvBase, options: NoiseOptions)

Bases: NoiseModel

Source code in eden/managers/modifiers/observations/noise.py 
22
23
24
def __init__(self, env: EnvBase, options: NoiseOptions):
    super().__init__(env=env, options=options)
    self.bias = ensure_tensor_on_device(self.bias, device=self.device)

Attributes

bias class-attribute instance-attribute 
bias: Tensor | float = ensure_tensor_on_device(self.bias, device=self.device)

Methods:

compute 
compute(data: Tensor) -> Tensor
Source code in eden/managers/modifiers/observations/noise.py 
26
27
28
29
30
31
32
33
def compute(self, data: torch.Tensor) -> torch.Tensor:
    match self.operation:
        case NoiseOperation.ADD:
            return data + self.bias
        case NoiseOperation.SCALE:
            return data * self.bias
        case _:
            return self.bias

UniformNoise 

UniformNoise(env: EnvBase, options: NoiseOptions)

Bases: NoiseModel

Source code in eden/managers/modifiers/observations/noise.py 
41
42
43
44
45
46
47
48
def __init__(self, env: EnvBase, options: NoiseOptions):
    super().__init__(env=env, options=options)
    if isinstance(self.n_min, (int, float)) and isinstance(self.n_max, (int, float)):
        if self.n_min >= self.n_max:
            raise ValueError(f"n_min ({self.n_min}) must be less than n_max ({self.n_max})")

    self.n_min = ensure_tensor_on_device(self.n_min, device=self.device)
    self.n_max = ensure_tensor_on_device(self.n_max, device=self.device)

Attributes

n_min class-attribute instance-attribute 
n_min: Tensor | float = ensure_tensor_on_device(self.n_min, device=self.device)
n_max class-attribute instance-attribute 
n_max: Tensor | float = ensure_tensor_on_device(self.n_max, device=self.device)

Methods:

compute 
compute(data: Tensor) -> Tensor
Source code in eden/managers/modifiers/observations/noise.py 
50
51
52
53
54
55
56
57
58
59
def compute(self, data: torch.Tensor) -> torch.Tensor:
    # Generate uniform noise in [0, 1) and scale to [n_min, n_max).
    noise = torch.rand_like(data) * (self.n_max - self.n_min) + self.n_min
    match self.operation:
        case NoiseOperation.ADD:
            return data + noise
        case NoiseOperation.SCALE:
            return data * noise
        case _:
            return noise

GaussianNoise 

GaussianNoise(env: EnvBase, options: NoiseOptions)

Bases: NoiseModel

Source code in eden/managers/modifiers/observations/noise.py 
67
68
69
70
71
72
73
def __init__(self, env: EnvBase, options: NoiseOptions):
    super().__init__(env=env, options=options)
    if isinstance(self.std, (int, float)) and self.std <= 0:
        raise ValueError(f"std ({self.std}) must be positive")

    self.mean = ensure_tensor_on_device(self.mean, device=self.device)
    self.std = ensure_tensor_on_device(self.std, device=self.device)

Attributes

mean class-attribute instance-attribute 
mean: Tensor | float = ensure_tensor_on_device(self.mean, device=self.device)
std class-attribute instance-attribute 
std: Tensor | float = ensure_tensor_on_device(self.std, device=self.device)

Methods:

compute 
compute(data: Tensor) -> Tensor
Source code in eden/managers/modifiers/observations/noise.py 
75
76
77
78
79
80
81
82
83
84
def compute(self, data: torch.Tensor) -> torch.Tensor:
    # Generate standard normal noise and scale.
    noise = self.mean + self.std * torch.randn_like(data)
    match self.operation:
        case NoiseOperation.ADD:
            return data + noise
        case NoiseOperation.SCALE:
            return data * noise
        case _:
            return noise

Functions:

ensure_tensor_on_device 

ensure_tensor_on_device(value: Tensor | float, device: device) -> Tensor | float

Ensure tensor is on the correct device, leave scalars unchanged.

Source code in eden/managers/modifiers/observations/noise.py 
11
12
13
14
15
def ensure_tensor_on_device(value: torch.Tensor | float, device: torch.device) -> torch.Tensor | float:
    """Ensure tensor is on the correct device, leave scalars unchanged."""
    if isinstance(value, torch.Tensor):
        return value.to(device=device)
    return value

Temporal observation modifiers

Stateful observation modifiers that act across control steps: an exponential moving average and a delay.

Attributes

Classes

ExponentialMovingAverage 

ExponentialMovingAverage(env: EnvBase, options: ObservationModifierOptions)

Bases: ObservationModifier

Exponential moving average, updated once per control step.

y(t) = alpha * x(t) + (1 - alpha) * y(t - 1).

alpha is the weight of the newest frame, as in pandas' ewm(alpha=...). Some filter libraries write the complement, so check that before porting a value. The term's frame must be floating-point — the weights are held in its dtype, so an integer one would truncate them to zero; :class:ObservationDelay is the modifier that carries any dtype.

A non-finite frame keeps its kind: an infinity entering the average stays an infinity, so a clip on the term still bounds it — a raycast's +inf for "no hit" reaches the policy as the clip's far end, not as a NaN that no clip bounds — and a NaN stays a NaN. Either one stays in the average until the env resets. The group's nan_policy checks the term's output after its modifiers, so it replaces what the policy receives but cannot clean the average itself. The one frame that still averages to NaN is a component that sees both +inf and -inf, their weighted sum being genuinely undefined.

PARAMETER DESCRIPTION
alpha

Weight of the newest frame, in (0, 1]. Required. alpha=1 passes each frame through unchanged; a smaller alpha smooths more.

TYPE: float

backfill

What the average starts from after a reset (default True). True: the first frame after the reset starts the average, so it is delivered unchanged, as if every earlier frame had equalled it. False: the average starts from zero, so the first frame delivers alpha * x.

TYPE: bool

Source code in eden/managers/modifiers/observations/temporal.py 
52
53
54
55
56
57
58
59
60
def __init__(self, env: EnvBase, options: ObservationModifierOptions):
    super().__init__(env=env, options=options)
    if self.alpha is None:
        raise ValueError("ExponentialMovingAverage: alpha is required (the weight of the newest frame, in (0, 1]).")
    if not 0.0 < self.alpha <= 1.0:
        raise ValueError(f"ExponentialMovingAverage: alpha must be in (0, 1], got {self.alpha}.")
    # The weight a row's first frame after a reset gets. With backfill it is 1, so that frame replaces
    # the zeroed average instead of being blended with it.
    self._first_frame_weight = 1.0 if self.backfill else self.alpha

Attributes

alpha class-attribute instance-attribute 
alpha: float | None = None
backfill class-attribute instance-attribute 
backfill: bool = True

Methods:

build 
build(frame: Tensor) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def build(self, frame: torch.Tensor) -> None:
    if not frame.is_floating_point():
        # The weights live in the frame's dtype, and ``torch.full(..., 0.2, dtype=torch.int64)`` is 0
        # without raising, so an integer frame would deliver zeros from its second frame on.
        raise ValueError(
            f"ExponentialMovingAverage: needs a floating-point frame, got {frame.dtype}. Average a float "
            f"term — a count, an index or a flag has no meaningful average — or drop the modifier. "
            f"ObservationDelay carries a frame of any dtype unchanged."
        )
    self._average = torch.zeros_like(frame)
    # One weight per env, broadcast over the feature dims: ``_first_frame_weight`` on a row whose
    # first frame since the reset is still to come, ``alpha`` once that frame has been advanced in.
    # ``_hold``, the weight of the stored average, is carried alongside rather than computed as
    # ``1 - _weight`` per read, so the pair is written together everywhere it changes.
    weight_shape = (frame.shape[0],) + (1,) * (frame.dim() - 1)
    self._weight = torch.full(weight_shape, self._first_frame_weight, dtype=frame.dtype, device=frame.device)
    self._hold = torch.full(weight_shape, 1.0 - self._first_frame_weight, dtype=frame.dtype, device=frame.device)
reset 
reset(envs_idx: Tensor | None = None) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    if is_env_mask(envs_idx):
        # Auto-reset passes a mask every step; ``masked_fill_`` applies it on the device, as
        # ``CircularBuffer.reset`` does.
        rows = envs_idx.view(self._weight.shape)
        self._average.masked_fill_(rows, 0.0)
        self._weight.masked_fill_(rows, self._first_frame_weight)
        self._hold.masked_fill_(rows, 1.0 - self._first_frame_weight)
    else:
        rows = slice(None) if envs_idx is None else envs_idx
        self._average[rows] = 0.0
        self._weight[rows] = self._first_frame_weight
        self._hold[rows] = 1.0 - self._first_frame_weight
compute 
compute(frame: Tensor) -> Tensor
Source code in eden/managers/modifiers/observations/temporal.py 
 94
 95
 96
 97
 98
 99
100
101
def compute(self, frame: torch.Tensor) -> torch.Tensor:
    # ``(1 - w) * y + w * x``, written out rather than as ``torch.lerp(y, x, w)``: lerp evaluates
    # ``y + w * (x - y)`` (``x - (x - y) * (1 - w)`` for ``w >= 0.5``), so an infinite frame meets
    # ``inf - inf`` and comes out NaN — which ``clip`` passes through, since ``clamp`` does, and
    # which then reaches the policy. This form turns an infinity into an infinity, so the term's
    # clip still bounds it. Its one indeterminate product, ``(1 - w) * y``, needs ``y`` to be zero
    # wherever ``1 - w`` is; :meth:`advance` is what keeps that true.
    return torch.addcmul(self._hold * self._average, self._weight, frame)
advance 
advance(frame: Tensor) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
103
104
105
106
107
108
109
110
def advance(self, frame: torch.Tensor) -> None:
    if self.alpha < 1.0:
        self._average.copy_(self.compute(frame))
    # ``alpha == 1`` is pass-through: ``_hold`` is zero everywhere and the average is never read
    # again, so it is left zeroed rather than written. Storing an infinite frame in it would make
    # the next ``(1 - w) * y`` a ``0 * inf``, and so a NaN, for a modifier that changes nothing.
    self._weight.fill_(self.alpha)
    self._hold.fill_(1.0 - self.alpha)

ObservationDelay 

ObservationDelay(env: EnvBase, options: ObservationModifierOptions)

Bases: ObservationModifier

Delivers the frame the term produced control_steps control steps earlier.

The delay is counted in pushes of the observation window (see :class:~eden.managers.modifiers.base.ObservationModifier), which happen once per control step. That is the same clock as history_length, so a delayed term with a history window holds the delayed frames.

PARAMETER DESCRIPTION
control_steps

How many control steps to delay by, >= 1. Required.

TYPE: int

backfill

What is delivered while fewer than control_steps frames have been seen since the reset (default True). True: the first frame after the reset. False: zeros. This setting is separate from the term's own backfill, which applies only to its history window.

TYPE: bool

Source code in eden/managers/modifiers/observations/temporal.py 
135
136
137
138
139
140
141
142
143
def __init__(self, env: EnvBase, options: ObservationModifierOptions):
    super().__init__(env=env, options=options)
    if self.control_steps is None:
        raise ValueError("ObservationDelay: control_steps is required (how many control steps to delay by, >= 1).")
    if isinstance(self.control_steps, bool) or not isinstance(self.control_steps, int) or self.control_steps < 1:
        raise ValueError(
            f"ObservationDelay: control_steps must be an int >= 1, got {self.control_steps!r}. "
            f"A delay of 0 is no delay; drop the modifier instead."
        )

Attributes

control_steps class-attribute instance-attribute 
control_steps: int | None = None
backfill class-attribute instance-attribute 
backfill: bool = True

Methods:

build 
build(frame: Tensor) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
145
146
147
148
149
150
def build(self, frame: torch.Tensor) -> None:
    # The window holds the current frame and the ``control_steps`` before it, so its oldest slot is the
    # delayed frame. ``CircularBuffer`` owns backfill, per-row reset and the read without an append.
    self._window = CircularBuffer(
        max_len=self.control_steps + 1, batch_size=frame.shape[0], device=frame.device, backfill=self.backfill
    )
reset 
reset(envs_idx: Tensor | None = None) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
152
153
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    self._window.reset(batch_ids=envs_idx)
compute 
compute(frame: Tensor) -> Tensor
Source code in eden/managers/modifiers/observations/temporal.py 
155
156
def compute(self, frame: torch.Tensor) -> torch.Tensor:
    return self._window.peek_oldest(frame)
advance 
advance(frame: Tensor) -> None
Source code in eden/managers/modifiers/observations/temporal.py 
158
159
def advance(self, frame: torch.Tensor) -> None:
    self._window.append(frame)

Functions:

See also