Deployment

A deployment backend runs a trained task policy on a real robot, or against a MuJoCo model of it for sim-to-sim validation. It builds the task's own env for its observation and action terms, reads the robot's state, and sends the action terms' commands over the robot's SDK at the control rate. The deployment guide covers eden deploy, every supported robot, safety and timing.

from eden.extensions.deployment.base import DEPLOYMENT_REGISTRY, DeploymentBase, resolve_deployment_for
from eden.options.extensions.deployment import DeploymentOptions

A backend registers in DEPLOYMENT_REGISTRY and is bound to a robot options class with bind_deployment; resolve_deployment_for(robot_options) picks the backend for a task's robot. Importing eden.extensions.deployment imports every backend module so those bindings exist; a backend whose SDK is missing is skipped. The MuJoCo backend is robot-agnostic and is looked up by name (eden deploy ... --mujoco), and a composite rig resolves to CompositeDeployment, which drives each part through that part's own binding.

Base class and registry

Base class and registry for real-robot deployment backends.

Attributes

DEPLOYMENT_REGISTRY module-attribute

DEPLOYMENT_REGISTRY = Registry('DEPLOYMENT')

Classes

DeploymentBase

DeploymentBase(env: RLEnvBase, options: DeploymentOptions)

Bases: ConfigurableMixin[DeploymentOptions], ABC

Base class for real-robot deployment backends.

ATTRIBUTE	DESCRIPTION
entity_name	The name of the entity to deploy TYPE: str
control_dt	The control time step TYPE: float | None
control_freq	The control frequency TYPE: float | None
decimation	The decimation factor TYPE: int | None
sync	Whether to synchronize the control loop TYPE: bool
connect_timeout_s	The timeout for connecting to the robot TYPE: float
deploy_kp_kd_scale	The scale of the Kp and Kd for the deployment. Recommend to try from 0.2 (20%), 0.5 (50%), 0.8 (80%), 1.0 (100%) TYPE: float
auto_yaw_align	If True (default), capture the robot's IMU yaw at START (the MuJoCo backend also captures it on :meth:reset's read) and install it on the state wrapper so subsequent observations are computed in the yaw-aligned world frame the policy was trained in. Disable when an external estimator (e.g. mocap) already provides a yaw-consistent base orientation. TYPE: bool
estimate_lin_vel	If True, run an onboard contact-aided Kalman filter (:class:~eden.extensions.deployment.utils.lin_vel_estimator.LinVelKalmanEstimator, the legged_control / BeyondMimic estimator) fusing IMU acceleration with leg-kinematics odometry, and feed the estimated base linear velocity to the observation pipeline — entity.get_vel() / base_lin_vel terms return the estimate instead of zeros. Requires feet_links. TYPE: bool
feet_links	Foot link names for the velocity estimator's leg odometry and contact detection. Defaults to the robot's own RobotOptions.feet_links_name (e.g. G1's ankle-roll links), which is where that knowledge belongs — the feet are a property of the asset, not of the transport, so the estimator works the same on hardware and through the robot-agnostic MuJoCo backend. Set this (--feet_links) only to estimate off a different pair of links, or for a robot whose options declare none. TYPE: tuple[str, ...]
lin_vel_estimator_options	Extra keyword overrides for the estimator (noise parameters, contact_force_threshold, ...). See LinVelKalmanEstimator. TYPE: dict
ignore_action_terms	Names of action terms whose command :class:RobotCommand has no channel for (a suction/welding term) and which are therefore driven over a channel of the operator's own. Listing a term here acknowledges that its command is not sent to the robot; leaving it unlisted raises at construction rather than dropping it silently on hardware. Every name is validated against the deployed config's action terms, and naming a term the payload does carry raises — the flag cannot stop a joint term from being sent (use disable_joints for that). Joint terms never need listing: one driving another entity than entity_name, and one whose effect the robot produces on its own (ActionTerm.sim_surrogate), are both skipped without a flag. See :func:build_term_action_maps. TYPE: tuple[str, ...]
disable_joints	Joint names or fnmatch patterns (e.g. "left_wrist_*", "*_ankle_*_joint") whose kp/kd are forced to 0 in every command, so those joints go free/limp on the robot. The policy still observes them and emits targets; the motors just ignore them (zero gains). For bench debugging (isolate a buzzing joint, free the arms) — do NOT disable load-bearing leg joints on an unsupported robot, it will collapse. TYPE: tuple[str, ...]
gain_ramp_steps	Control steps over which :meth:set_gains moves the commanded kp/kd to a new policy slot's (default 50 = 1 s at 50 Hz), linearly from whatever is being sent at the switch. 0 jumps. A slot switch hands the robot the gains that policy was trained at -- SONIC's 7520-22 hip pitch is ~2.4x Holosoma's 7520-14 -- and a stiffness step of that size can knock a standing robot over. TYPE: int
debug_record	If True, record per control step the command actually sent (target positions + kp/kd) and the measured robot proprioception (joint pos / vel / torque, base quat / ang-vel / lin-acc). For offline diagnosis (e.g. commanded-vs-measured tracking, torque saturation, limit-cycle frequency). Every numeric state.extra entry rides along under an x_ prefix; with estimate_lin_vel that includes the estimate (x_base_lin_vel_estimate) and, per foot, the contact latch (x_contacts) and the solved contact force (x_forces, flattened (n_feet * 3,)). Streamed to disk as it is produced, not held until :meth:close: the runs worth recording are the ones that end badly, and those are the runs where close may never finish. See :class:~eden.extensions.deployment.utils.record.StreamingRecorder — a killed run leaves a chunk directory that :func:~eden.extensions.deployment.utils.record.load_deploy_record reads exactly like the merged file a clean run produces. TYPE: bool
debug_record_path	Destination .npz for debug_record. Defaults to <log_dir>/deploy_record.npz. TYPE: str | None
debug_record_flush_steps	Steps per streamed chunk (default 100 — two seconds at 50 Hz), i.e. the most that a kill can cost. Lower it to lose less and write more often; the writing itself is off the control thread either way. TYPE: int
command_overrides	Command terms to substitute at deploy time, {term_name: CommandTermOptions}. The env's own term is built and stepped by the config it was trained with; a deployment can replace it with one configured for a robot with an operator standing next to it — most usefully a task's random :class:~eden.managers.terms.commands.velocity.UniformVelocityCommand for a :class:~eden.managers.terms.commands.velocity.ConstantVelocityCommand that holds one value, or an :class:~eden.managers.terms.commands.velocity.OperatorVelocityCommand a human drives from a keyboard or gamepad. This is the command-manager counterpart of the entity swap: the substitute is an ordinary command term, so command_manager.compute drives it with no special case anywhere in the deploy loop, and everything reading the command is unchanged. Names are validated against the env's active command terms, and the replacement's command must have the same shape as the term it replaces. Applied transactionally with the entity swap and rolled back with it. TYPE: dict
operator_keyboard	Open the terminal keyboard reader (s / x / i / k + Enter) as an operator input. Default True. Turn it off in a script that reads the keyboard itself -- a pynput teleop device sees every key typed into the terminal too, so a teleop key and a command key would land on the same line (s is both START and a teleop axis on the WASD bindings) -- and give the operator commands to :meth:request_init / :meth:request_start / :meth:request_stop from that device instead. TYPE: bool
target_lpf	First-order low-pass on the commanded joint targets, applied to every payload via :meth:_apply_target_lpf: sent = a*new + (1-a)*prev with a = target_lpf in (0, 1]. 1.0 (default) disables it. Lower values smooth high-frequency command content (e.g. a single-frame policy's half-control-rate limit cycle / motor whine) while leaving kp/kd — and thus the holding stiffness — untouched. Adds command lag (~1/(a*freq)). TYPE: float
init_target	Where the INIT ramp drives the joints before control is handed to the policy. "default" (default) targets RobotOptions.default_dofs_pos -- the robot's own stance, and the only target that is safe from an arbitrary starting pose. "motion" targets frame 0 of the tracked reference clip instead, so a motion-imitation policy starts from the pose it was trained to continue from rather than 100+ degrees away. Both are joint-space ramps over the wire, so hardware and --mujoco do the same thing. "motion" is NOT automatically safe: it drives the legs to whatever shape frame 0 has while the base is held by nothing but the joint PD. For a clip whose first frame is a deep crouch that is an open-loop squat with no balance behind it, and it can topple -- it has in MuJoCo sim-to-sim. Ramp slowly (init_steps), support the robot on the first attempt, and prefer entering through a balance policy slot. "motion_rsi" additionally teleports the base to the clip's frame-0 pose. That is reference-state initialization and is sim-only -- backends that drive real hardware reject it, since nothing on a robot can place its own base. TYPE: str
init_steps	Control cycles the INIT ramp takes from the pose measured at the press to the init_target pose (default 250, 5 s at 50 Hz), so a lying or folded robot rises instead of snapping there under full PD from the first tick. TYPE: int
init_dofs_kp_scale	Scale on the kp sent during that ramp (default 1.0). TYPE: float
init_dofs_kd_scale	Scale on the kd sent during that ramp (default 1.0). TYPE: float
on_motion_end	What a motion-imitation deploy does once the clip reaches its final frame. "continue" (default) keeps handing the policy's actions to the robot -- the reference holds at its last frame and the policy tracks a target that has stopped moving. "stand" disengages the policy and ramps the joints to the robot's default stance over stand_steps, then holds it. "stand" is joint-space only: it detects the end through the same motion clock that runs on hardware and sends the same RobotCommand the wire already carries, so a real robot and --mujoco execute the identical manoeuvre. Nothing about it is sim-only. It is NOT unconditionally safe, though: it drives to the default stance from wherever the clip left the robot, so it suits a clip that ends near a stand and wants a longer stand_steps -- or an operator STOP instead -- for a clip that ends crouched, mid-step, or airborne. "switch" hands control to another loaded policy slot (motion_end_slot) instead, through the same request queue the operator's remote/keyboard uses. That is the only one of the three that actually balances: a fixed joint-space command holds a shape, not a stand, so a biped under "stand" topples within a couple of seconds. Point it at a velocity-tracking / loco policy at zero command. TYPE: str
stand_steps	Control cycles the on_motion_end="stand" ramp takes to reach the default stance (default 100 = 2 s at 50 Hz). Longer is gentler on hardware. TYPE: int
motion_end_slot	1-based policy slot on_motion_end="switch" hands to, numbered as the operator keys 1-9 are. Default 1. The request is dropped with a warning if no such slot is loaded, exactly as an out-of-range operator press would be. TYPE: int
policy_seconds	End the run once the policy has driven the robot this long since its latest START, as the operator's KILL does (eden deploy --mujoco_seconds): the headless counterpart of the unattended INIT and START, so a rollout with no operator ends by itself. Only the ticks the policy drove count -- not STOP's or INIT's, nor START's handover hold -- and a STOP starts the count afresh. Only a simulator (:attr:allow_unattended_start) takes it; on a robot the operator ends the run. None (default): until the operator ends it. TYPE: float | None

Source code in eden/extensions/deployment/base.py

def __init__(self, env: RLEnvBase, options: DeploymentOptions) -> None:
    super().__init__(options)
    if not (0.0 < self.target_lpf <= 1.0):
        raise ValueError(f"target_lpf must be in (0, 1] (1.0 disables), got {self.target_lpf}")
    if self.gain_ramp_steps < 0:
        raise ValueError(f"gain_ramp_steps must be >= 0 (0 jumps), got {self.gain_ramp_steps}")
    if self.policy_seconds is not None and not self.policy_seconds > 0.0:
        raise ValueError(f"policy_seconds must be > 0, or None; got {self.policy_seconds}.")
    if self.policy_seconds is not None and not self.allow_unattended_start:
        raise ValueError(
            f"policy_seconds ends a run by itself, which only a simulator may: {type(self).__name__} drives a "
            "robot, whose run the operator ends (KILL). Drop policy_seconds (--mujoco_seconds)."
        )
    # Ticks the policy has driven since its latest START: ``policy_seconds``' clock.
    self._policy_ticks = 0
    self._check_state_to_observation_override()
    self._env = env
    self.cfg = env.config
    self.robot_options = self._resolve_robot_options(self.cfg, self.entity_name)

    self.control_dt = self._resolve_control_dt()
    if self.control_freq is None and self.control_dt > 0.0:
        self.control_freq = 1.0 / self.control_dt

    self._resolve_dofs()
    self.num_dofs = len(self.dofs_name)
    # Deployment DOF name -> index. Name-mangled (stored as ``_DeploymentBase__dofs_idx``) so a
    # backend's own map cannot replace it: the SDK backends each bind an Eden -> motor-slot map,
    # and one sharing this attribute would break every lookup by DOF name on real hardware.
    self.__dofs_idx = {name: i for i, name in enumerate(self.dofs_name)}
    self._resolve_disabled_joints()
    # The gains every command sends *now*. They start at the deployed robot's and move only
    # through :meth:`set_gains`, which ramps them to the active policy slot's.
    self.default_dof_kp, self.default_dof_kd = self.wire_gains()
    # The :meth:`set_gains` ramp: its (kp, kd) target, and the control steps left to reach it.
    self._gain_target: tuple[np.ndarray, np.ndarray] | None = None
    self._gain_steps_left = 0
    self._init_recording()
    self._lpf_prev: np.ndarray | None = None
    # Last state read off the wire. Declared here because four call sites guard it with
    # ``self.state is not None`` -- an operator INIT before the first read, a hold payload,
    # the end-of-motion latch -- and without this they raise AttributeError instead of
    # taking the branch they are written for.
    self.state: RobotState | None = None
    # The observation manager arrives holding state the training wrapper's ``env.reset()``
    # computed off the SIMULATED robot at its spawn pose; the first read clears it.
    # See :meth:`state_to_observation`.
    self._observations_stale = True
    self._processed_action = torch.zeros(self.num_dofs, device=self._env.device)
    self._term_action_maps = self._build_term_action_maps()
    self._log_unapplied_action_delays()
    self._direct_target_maps: dict[tuple[str, ...], torch.Tensor] = {}
    # Whether anything has ever reached ``_processed_action`` through
    # ``write_dofs_targets``. Zeros are a legal joint target, so the buffer alone cannot say
    # whether it holds a policy's command or its own initializer; see :meth:`step`.
    self._direct_targets_written = False

    self._rate = None
    if self.sync and self.control_freq is not None:
        self._rate = RateLimiter(self.control_freq, name="[deploy] control loop")

    self._resolve_options()
    # Reject an impossible init_target / on_motion_end here, not at the operator's INIT press
    # or at the end of the clip: by then the backend is connected and the robot is powered,
    # and a config error has no business surfacing mid-operation.
    self._validate_init_target()
    self._validate_on_motion_end()

    self._repointed_refs: list[tuple[Any, str]] = []
    self._repointed_command_refs: list[tuple[Any, str]] = []
    self._swapped_commands: dict[str, Any] = {}
    sim_entity = self._swap_in_state_entity()
    try:
        # ``on_motion_end='stand'`` latch: the pose the stand ramp started from, and how
        # far through it we are. Cleared by ``reset_motion_reference`` (deploy reset, INIT,
        # START, policy switch), so re-arming the clip re-engages the policy.
        self._standing_from: np.ndarray | None = None
        self._stand_count = 0
        self._motion_end_switched = False
        self._pending_policy_switches: list[int | str] = []
        # Operator state machine, which every backend runs. Boot in STOP/hold:
        # nothing moves until the operator presses INIT and then START.
        self._inputs: list[Any] = []
        self._warned_no_operator = False
        self._pending_start_events = 0
        self._frozen_watch: dict[str, Any] | None = None
        self._enter_boot_hold()
        self._handover_from_init = False
        # True while :meth:`send_payload` is sending the state machine's own command (the
        # hold, the INIT ramp, the START tick), not the policy's: see :attr:`sending_hold`.
        self._sending_hold = False
        self._lin_vel_warned = False
        self._lin_vel_estimator = self._make_lin_vel_estimator(sim_entity)

        # Command-term substitution, last: a substitute may open an input device, and
        # everything that can still raise has run by now.
        self._apply_command_overrides()
    except BaseException:
        self._undo_swaps(sim_entity)
        raise

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

control_freq class-attribute instance-attribute

control_freq: float | None = None

decimation class-attribute instance-attribute

decimation: int | None = None

sync class-attribute instance-attribute

sync: bool = True

connect_timeout_s class-attribute instance-attribute

connect_timeout_s: float = 5.0

deploy_kp_kd_scale class-attribute instance-attribute

deploy_kp_kd_scale: float = 1.0

auto_yaw_align class-attribute instance-attribute

auto_yaw_align: bool = True

estimate_lin_vel class-attribute instance-attribute

estimate_lin_vel: bool = False

feet_links class-attribute instance-attribute

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

lin_vel_estimator_options class-attribute instance-attribute

lin_vel_estimator_options: dict = {}

ignore_action_terms class-attribute instance-attribute

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

disable_joints class-attribute instance-attribute

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

gain_ramp_steps class-attribute instance-attribute

gain_ramp_steps: int = 50

init_target class-attribute instance-attribute

init_target: str = 'default'

init_steps class-attribute instance-attribute

init_steps: int = 250

init_dofs_kp_scale class-attribute instance-attribute

init_dofs_kp_scale: float = 1.0

init_dofs_kd_scale class-attribute instance-attribute

init_dofs_kd_scale: float = 1.0

on_motion_end class-attribute instance-attribute

on_motion_end: str = 'continue'

stand_steps class-attribute instance-attribute

stand_steps: int = 100

motion_end_slot class-attribute instance-attribute

motion_end_slot: int = 1

debug_record class-attribute instance-attribute

debug_record: bool = False

debug_record_path class-attribute instance-attribute

debug_record_path: str | None = None

debug_record_flush_steps class-attribute instance-attribute

debug_record_flush_steps: int = 100

command_overrides class-attribute instance-attribute

command_overrides: dict = {}

target_lpf class-attribute instance-attribute

target_lpf: float = 1.0

operator_keyboard class-attribute instance-attribute

operator_keyboard: bool = True

policy_seconds class-attribute instance-attribute

policy_seconds: float | None = None

cfg instance-attribute

cfg = env.config

robot_options instance-attribute

robot_options = self._resolve_robot_options(self.cfg, self.entity_name)

control_dt class-attribute instance-attribute

control_dt: float | None = self._resolve_control_dt()

num_dofs instance-attribute

num_dofs = len(self.dofs_name)

state instance-attribute

state: RobotState | None = None

supports_base_placement class-attribute

supports_base_placement: bool = False

sends_command_gains class-attribute

sends_command_gains: bool = True

operator_guide_title class-attribute

operator_guide_title: str = ''

sync_before_commands class-attribute

sync_before_commands: bool = False

advances_episode_clock class-attribute

advances_episode_clock: bool = False

allow_unattended_start class-attribute

allow_unattended_start: bool = False

sending_hold property

sending_hold: bool

Whether the payload :meth:send_payload is sending is the state machine's own, not the policy's.

True for the STOP/boot hold, the INIT ramp and the START tick, the commands that hold or place the robot's joints. A backend whose robot carries a tool the joints do not describe leaves it alone on these: the Franka keeps its gripper at the policy's last command, so a STOP or INIT does not open a grasp. False for a command sent from outside :meth:step.

init_done property

init_done: bool

Whether an INIT ramp has reached its target (and holds it until START or STOP).

policy_active property

policy_active: bool

Whether the action passed to :meth:step reaches the robot this tick.

False until START hands control over, and again after STOP or INIT.

blocks_for_the_period property

blocks_for_the_period: bool

Whether :meth:send_payload itself waits out the control period (and so paces the loop). False here.

A composite rig with such a part lets it pace the rig, as it paces itself deployed alone (:class:~eden.extensions.deployment.composite.CompositeDeployment).

provides_base_lin_vel property

provides_base_lin_vel: bool

Whether base_lin_vel observations read a real value on this deploy, or zeros.

:class:RobotStateEntity holds the base linear velocity at zero unless something supplies it: the onboard estimator (estimate_lin_vel) or a backend that publishes it in RobotState.extra (:meth:_publishes_base_lin_vel -- the MuJoCo backend's ground truth; no hardware backend does, an IMU measures no velocity). A policy or task whose observations read it on a deploy where neither holds is silently-wrong control, which is why the CLI refuses that arrangement at startup rather than letting it run.

Methods:

wire_gains

wire_gains(robot: Any = None) -> tuple[ndarray, ndarray]

The kp/kd this deployment sends for a policy trained on robot, in deployment DOF order.

robot is a policy's pinned :class:RobotOptions (PolicyBase.robot_options()): its default_dofs_kp / default_dofs_kd on every DOF it names, the deployed robot's on the rest, matched by DOF name. None -- a policy pinning no robot, a checkpoint trained in this env -- or a robot declaring no gains (a composite, or an empty map, as G1Dofs23's) is the deployed robot's throughout. Both are then scaled by deploy_kp_kd_scale and zeroed on :attr:disabled_dofs, the operator's choices for the whole run.

Raises ValueError as :func:resolve_pinned_gains does, which resolves the pinned maps.

Source code in eden/extensions/deployment/base.py

def wire_gains(self, robot: Any = None) -> tuple[np.ndarray, np.ndarray]:
    """The kp/kd this deployment sends for a policy trained on ``robot``, in deployment DOF order.

    ``robot`` is a policy's pinned :class:`RobotOptions` (``PolicyBase.robot_options()``): its
    ``default_dofs_kp`` / ``default_dofs_kd`` on every DOF it names, the deployed robot's on
    the rest, matched by DOF name. ``None`` -- a policy pinning no robot, a checkpoint trained
    in this env -- or a robot declaring no gains (a composite, or an empty map, as
    ``G1Dofs23``'s) is the deployed robot's throughout. Both are then scaled by
    ``deploy_kp_kd_scale`` and zeroed on :attr:`disabled_dofs`, the operator's choices for the
    whole run.

    Raises ``ValueError`` as :func:`resolve_pinned_gains` does, which resolves the pinned maps.
    """
    kp, kd = self._robot_dofs_kp.copy(), self._robot_dofs_kd.copy()
    for gains, pinned in zip((kp, kd), resolve_pinned_gains(robot, self.dofs_name), strict=True):
        for name, value in pinned.items():
            gains[self.__dofs_idx[name]] = value
    scale = self._gain_scale()
    kp *= scale
    kd *= scale
    kp[self._disabled_idx] = 0.0
    kd[self._disabled_idx] = 0.0
    return kp, kd

set_gains

set_gains(dofs_kp: ndarray, dofs_kd: ndarray, *, ramp_steps: int | None = None) -> None

Move the kp/kd every command sends to dofs_kp / dofs_kd over ramp_steps control steps.

The targets are wire gains in deployment DOF order -- :meth:wire_gains' output, already scaled and with disabled joints zeroed. ramp_steps defaults to :attr:gain_ramp_steps; 0 jumps, which is only right before :meth:connect, while nothing is on the wire.

The ramp is linear from whatever is being sent now, a ramp still in progress included, and advances once per :meth:step. Every payload reads the current gains -- the policy's, the STOP hold, the INIT ramp, the end-of-clip stand -- so all of them follow it. A policy keeps driving through it: holding the pose instead topples a humanoid, which only its policy balances (see "Each slot runs at its own PD gains" in docs/guides/deployment.md).

Raises on a backend that does not send each command's gains (:attr:sends_command_gains): there the call would change nothing on the robot, which is not what its caller asked for.

Source code in eden/extensions/deployment/base.py

def set_gains(self, dofs_kp: np.ndarray, dofs_kd: np.ndarray, *, ramp_steps: int | None = None) -> None:
    """Move the kp/kd every command sends to ``dofs_kp`` / ``dofs_kd`` over ``ramp_steps`` control steps.

    The targets are wire gains in deployment DOF order -- :meth:`wire_gains`' output, already
    scaled and with disabled joints zeroed. ``ramp_steps`` defaults to :attr:`gain_ramp_steps`;
    ``0`` jumps, which is only right before :meth:`connect`, while nothing is on the wire.

    The ramp is linear from whatever is being sent now, a ramp still in progress included, and
    advances once per :meth:`step`. Every payload reads the current gains -- the policy's, the
    STOP hold, the INIT ramp, the end-of-clip stand -- so all of them follow it. A policy keeps
    driving through it: holding the pose instead topples a humanoid, which only its policy
    balances (see "Each slot runs at its own PD gains" in ``docs/guides/deployment.md``).

    Raises on a backend that does not send each command's gains (:attr:`sends_command_gains`):
    there the call would change nothing on the robot, which is not what its caller asked for.
    """
    if not self.sends_command_gains:
        raise RuntimeError(
            f"{type(self).__name__}.set_gains would change nothing on the robot: this backend does not send "
            "each command's kp/kd (sends_command_gains=False), it programs its gains once at connect or runs "
            "its own loop. Check sends_command_gains before calling it, and set the backend's own gain "
            "options to change what the robot runs at."
        )
    kp = np.array(dofs_kp, dtype=np.float64).reshape(-1)
    kd = np.array(dofs_kd, dtype=np.float64).reshape(-1)
    if kp.shape != (self.num_dofs,) or kd.shape != (self.num_dofs,):
        raise ValueError(
            f"{type(self).__name__}.set_gains takes one kp and one kd per deployment DOF ({self.num_dofs}, "
            f"in dofs_name order); got {kp.shape[0]} kp and {kd.shape[0]} kd. Build them with wire_gains(robot)."
        )
    steps = self.gain_ramp_steps if ramp_steps is None else ramp_steps
    if steps <= 0:
        self._gain_target = None
        self.default_dof_kp, self.default_dof_kd = kp, kd
    else:
        self._gain_target, self._gain_steps_left = (kp, kd), steps

programmed_gains

programmed_gains() -> tuple[ndarray, ndarray] | None

The kp/kd the robot runs whichever slot is active, on a backend without :attr:sends_command_gains.

In deployment DOF order and in :meth:wire_gains' units, so a slot's gains can be compared with them; None when the robot's gains are not in those units (XHand firmware units, a controller's own stiffness) or it has none (a position loop), which is the default. A backend that sends each command's gains runs no fixed set, and is never asked.

Source code in eden/extensions/deployment/base.py

def programmed_gains(self) -> tuple[np.ndarray, np.ndarray] | None:
    """The kp/kd the robot runs whichever slot is active, on a backend without :attr:`sends_command_gains`.

    In deployment DOF order and in :meth:`wire_gains`' units, so a slot's gains can be compared
    with them; ``None`` when the robot's gains are not in those units (XHand firmware units, a
    controller's own stiffness) or it has none (a position loop), which is the default. A
    backend that sends each command's gains runs no fixed set, and is never asked.
    """
    return None

command_gain_dofs

command_gain_dofs() -> ndarray

Per DOF, whether each command's kp/kd reaches its motor: :attr:sends_command_gains, on every DOF.

A composite rig answers per part, since its body may take each command's gains while its hands run their own (:class:~eden.extensions.deployment.composite.CompositeDeployment).

Source code in eden/extensions/deployment/base.py

def command_gain_dofs(self) -> np.ndarray:
    """Per DOF, whether each command's kp/kd reaches its motor: :attr:`sends_command_gains`, on every DOF.

    A composite rig answers per part, since its body may take each command's gains while its
    hands run their own (:class:`~eden.extensions.deployment.composite.CompositeDeployment`).
    """
    return np.full(self.num_dofs, self.sends_command_gains)

write_dofs_targets

write_dofs_targets(targets: Tensor, dofs_name: list[str] | tuple[str, ...]) -> None

Write absolute joint-position targets directly into the processed-action buffer.

Bypass for an extension-policy slot, which converts its policy's raw action with that policy's own controller and writes the resulting targets here, instead of handing raw actions to the env's action manager (which belongs to another policy). targets[i] is name-mapped onto deployment DOF dofs_name[i]; follow with step(None), which builds the payload from this buffer without running the action manager.

Joint positions only — the payload's dofs_vel and dofs_torque fields have no writer on this path, which is what keeps a velocity, torque or velocity-feedforward controller out of a deploy slot.

Raises KeyError if a name is not a deployment DOF.

Source code in eden/extensions/deployment/base.py

def write_dofs_targets(self, targets: torch.Tensor, dofs_name: list[str] | tuple[str, ...]) -> None:
    """Write absolute joint-position targets directly into the processed-action buffer.

    Bypass for an extension-policy slot, which converts its policy's raw action
    with that policy's *own* controller and writes the resulting targets here,
    instead of handing raw actions to the env's action manager (which belongs to
    another policy). ``targets[i]`` is name-mapped onto deployment DOF
    ``dofs_name[i]``; follow with ``step(None)``, which builds the payload from
    this buffer without running the action manager.

    Joint positions only — the payload's ``dofs_vel`` and ``dofs_torque`` fields
    have no writer on this path, which is what keeps a velocity, torque or
    velocity-feedforward controller out of a deploy slot.

    Raises ``KeyError`` if a name is not a deployment DOF.
    """
    key = tuple(dofs_name)
    idx = self._direct_target_maps.get(key)
    if idx is None:
        idx = torch.tensor([self.__dofs_idx[name] for name in key], dtype=torch.long, device=self._env.device)
        self._direct_target_maps[key] = idx
    self._processed_action[idx] = targets.to(device=self._processed_action.device).flatten()
    self._direct_targets_written = True

action_to_payload

action_to_payload(action: Optional[Tensor | dict[str, Tensor]] = None) -> RobotCommand

Source code in eden/extensions/deployment/base.py

def action_to_payload(self, action: Optional[torch.Tensor | dict[str, torch.Tensor]] = None) -> RobotCommand:
    if action is not None:
        self._env.action_manager.compute(action)

        for term, src_idx, dst_idx in self._term_action_maps:
            self._processed_action[dst_idx] = term.processed_action[0, src_idx]
    elif not self._direct_targets_written:
        # :meth:`step` never gets here -- it sends nothing for ``None`` before the first
        # write. A direct caller asking for a payload out of the untouched buffer is a
        # mistake, and an all-zeros command is the wrong thing to hand it.
        raise RuntimeError(
            f"{type(self).__name__}.action_to_payload(None) before anything wrote joint targets. "
            "``action=None`` means 'the active slot already wrote absolute targets via "
            "write_dofs_targets' -- but the buffer is still its all-zeros initializer, and "
            "sending that would drive every joint to 0 rad at the default gains. Have the "
            "policy write its targets first, or pass an action for the action manager to process."
        )

    return RobotCommand(
        dofs_pos=self._processed_action.cpu().numpy(),
        dofs_vel=np.zeros(self.num_dofs),
        dofs_torque=np.zeros(self.num_dofs),
        dofs_kp=self.default_dof_kp,
        dofs_kd=self.default_dof_kd,
    )

request_init

request_init() -> None

INIT, as the operator's key or remote does: ramp from the measured pose to the init_target pose.

Source code in eden/extensions/deployment/base.py

def request_init(self) -> None:
    """INIT, as the operator's key or remote does: ramp from the measured pose to the ``init_target`` pose."""
    self._warned_no_operator = True  # a script pressing the keys is the operator
    self._handle_init()

request_start

request_start() -> None

START, as the operator's key or remote does: hand control to the policy.

A script that drives a deployment without an operator calls this explicitly -- nothing a backend does on its own starts the policy on a robot (:attr:allow_unattended_start).

Source code in eden/extensions/deployment/base.py

def request_start(self) -> None:
    """START, as the operator's key or remote does: hand control to the policy.

    A script that drives a deployment without an operator calls this explicitly -- nothing a
    backend does on its own starts the policy on a robot (:attr:`allow_unattended_start`).
    """
    self._warned_no_operator = True  # a script pressing the keys is the operator
    self._handle_start()

request_stop

request_stop() -> None

STOP, as the operator's key or remote does: disengage the policy and hold the measured pose.

Source code in eden/extensions/deployment/base.py

def request_stop(self) -> None:
    """STOP, as the operator's key or remote does: disengage the policy and hold the measured pose."""
    self._warned_no_operator = True  # a script pressing the keys is the operator
    self._handle_stop()

consume_start_events

consume_start_events() -> int

Return and clear the number of START presses since the last drain.

The deploy loop re-arms the active policy slot on START (reset internal state, rewind command clocks) because :meth:_handle_start re-captures the yaw offset — any policy state accumulated while stepping during STOP/hold references the pre-START world frame.

Source code in eden/extensions/deployment/base.py

def consume_start_events(self) -> int:
    """Return and clear the number of START presses since the last drain.

    The deploy loop re-arms the active policy slot on START (reset internal state, rewind
    command clocks) because :meth:`_handle_start` re-captures the yaw offset — any policy
    state accumulated while stepping during STOP/hold references the pre-START world frame.
    """
    count = self._pending_start_events
    self._pending_start_events = 0
    return count

motion_finished

motion_finished() -> bool

True once every tracked motion reference has played through to its final frame.

A task drives its reference through the motion manager, through a MotionCommand, or through both, and this reads whichever are present — the same two paths :meth:_advance_motion_reference ticks and :meth:reset_motion_reference rewinds. Reading only the manager would leave on_motion_end silently inert on every command-path deploy (SONIC, beyondmimic_moves), which is most of them.

The signals are the clocks the observation terms see, so hardware and --mujoco agree on when the clip is over. False when no reference is present (a plain RL env, or an entity no motion term tracks — a config question, not a reason to park the robot mid-run) and when any present reference is still running.

Costs one device sync per present reference, per control step, until the end fires — a host-side branch needs the value on the host. Callers latch: on_motion_end stops asking once it has acted (_standing_from / _motion_end_switched), and on_motion_end="continue" never asks at all.

Source code in eden/extensions/deployment/base.py

def motion_finished(self) -> bool:
    """True once every tracked motion reference has played through to its final frame.

    A task drives its reference through the motion *manager*, through a ``MotionCommand``,
    or through both, and this reads whichever are present — the same two paths
    :meth:`_advance_motion_reference` ticks and :meth:`reset_motion_reference` rewinds.
    Reading only the manager would leave ``on_motion_end`` silently inert on every
    command-path deploy (SONIC, ``beyondmimic_moves``), which is most of them.

    The signals are the clocks the observation terms see, so hardware and ``--mujoco``
    agree on when the clip is over. ``False`` when *no* reference is present (a plain RL
    env, or an entity no motion term tracks — a config question, not a reason to park the
    robot mid-run) and when any present reference is still running.

    Costs one device sync per present reference, per control step, until the end fires —
    a host-side branch needs the value on the host. Callers latch: ``on_motion_end``
    stops asking once it has acted (``_standing_from`` / ``_motion_end_switched``), and
    ``on_motion_end="continue"`` never asks at all.
    """
    signals = [sig for sig in (self._manager_motion_finished(), self._command_motion_finished()) if sig is not None]
    return bool(signals) and all(signals)

consume_policy_switches

consume_policy_switches() -> list[int | str]

Drain and return pending policy-switch requests (0-based target index or "next").

The deploy loop calls this every step and validates the targets against the loaded slot count (which the deployment does not know). Requests come from an operator remote / keyboard on the hardware backends and from on_motion_end="switch" here, so both reach the loop by the one route — and a MuJoCo sim-to-sim run switches slots exactly as the robot does.

Source code in eden/extensions/deployment/base.py

def consume_policy_switches(self) -> list[int | str]:
    """Drain and return pending policy-switch requests (0-based target index or ``"next"``).

    The deploy loop calls this every step and validates the targets against the loaded slot
    count (which the deployment does not know). Requests come from an operator remote /
    keyboard on the hardware backends and from ``on_motion_end="switch"`` here, so both
    reach the loop by the one route — and a MuJoCo sim-to-sim run switches slots exactly as
    the robot does.
    """
    pending = self._pending_policy_switches
    self._pending_policy_switches = []
    return pending

request_policy_switch

request_policy_switch(target: int | str) -> None

Queue a slot switch for the deploy loop to apply on its next iteration.

Source code in eden/extensions/deployment/base.py

def request_policy_switch(self, target: int | str) -> None:
    """Queue a slot switch for the deploy loop to apply on its next iteration."""
    self._pending_policy_switches.append(target)

step

step(action: Optional[Tensor | dict[str, Tensor]] = None) -> dict[str, Any]

Send one control step's command, then read and return the resulting observation.

action is the policy's raw output, run through the env's action manager by :meth:action_to_payload. It is None for an extension-policy slot, which has already written its targets with :meth:write_dofs_targets: the payload is then built from that buffer alone, with the action manager left out of it.

None before any such write is the third case — a caller that has nothing to send this step — and sends nothing rather than commanding a zero-initialised buffer, which on hardware is every joint driven to 0.0 rather than left alone. Hence the _direct_targets_written flag: an all-zero _processed_action is a legitimate command (every joint to its zero position) and cannot be told from an untouched buffer by inspecting it.

Every backend sends what the operator state machine wants: the hold while stopped, the INIT ramp, and the policy's command only under START, with the end-of-clip handover (on_motion_end) taking precedence over the policy. Sending nothing (the third case above) is :meth:_idle_tick: MuJoCo does not advance and hardware keeps its last command.

Source code in eden/extensions/deployment/base.py

def step(self, action: Optional[torch.Tensor | dict[str, torch.Tensor]] = None) -> dict[str, Any]:
    """Send one control step's command, then read and return the resulting observation.

    ``action`` is the policy's raw output, run through the env's action manager by
    :meth:`action_to_payload`. It is ``None`` for an extension-policy slot, which has
    already written its targets with :meth:`write_dofs_targets`: the payload is then
    built from that buffer alone, with the action manager left out of it.

    ``None`` before any such write is the third case — a caller that has nothing to
    send this step — and sends nothing rather than commanding a zero-initialised buffer,
    which on hardware is every joint driven to 0.0 rather than left alone. Hence the
    ``_direct_targets_written`` flag: an all-zero ``_processed_action`` is a legitimate
    command (every joint to its zero position) and cannot be told from an untouched
    buffer by inspecting it.

    Every backend sends what the operator state machine wants: the hold while stopped, the
    INIT ramp, and the policy's command only under START, with the end-of-clip handover
    (``on_motion_end``) taking precedence over the policy. Sending nothing (the third case
    above) is :meth:`_idle_tick`: MuJoCo does not advance and hardware keeps its last command.
    """
    if self.policy_seconds is not None and self._policy_ticks * self.control_dt >= self.policy_seconds - 1e-9:
        en.logger.info(f"[deploy] policy_seconds={self.policy_seconds:g} reached -- ending the run as KILL does.")
        raise KeyboardInterrupt
    self._before_step()
    self._advance_gain_ramp()
    payload = self._operator_payload(action)
    # Only a tick the policy drove advances its clocks: not STOP/hold, not INIT, and not the
    # START tick's hold, so the policy's first observation reads the clock at 0 and the clip
    # at frame 0, as after an env reset.
    driven = self._use_policy and not self._handover_pending
    self._handover_pending = False
    if payload is not None:
        self._sending_hold = not driven
        try:
            self.send_payload(payload)
        finally:
            self._sending_hold = False
    else:
        self._idle_tick()
    if not self._use_policy:
        self._policy_ticks = 0  # a later START runs policy_seconds afresh
    elif driven and payload is not None:
        self._policy_ticks += 1
    if self._rate is not None:
        self._rate.sleep()
    if self.advances_episode_clock and driven:
        self._env.episode_length_buf += 1
    if self.sync_before_commands:
        self.state = self.read_state()
        self._sync_state(self.state)
        self._update_commands(advance_reference=driven)
    else:
        self._update_commands(advance_reference=driven)
        self.state = self.read_state()
    # A control tick is a step, so stateful observation terms advance once here — the deploy
    # loop's phase 7. ``reset`` reads without advancing, as ``EnvBase.reset`` does.
    if not self._use_policy and payload is not None:
        self._watch_frozen_joints(payload, self.state)
    # Recorded after the observation, not straight after ``read_state``: this tick's velocity
    # estimate and foot contacts join ``state.extra`` inside ``state_to_observation``, and a
    # record taken before it would miss them. ``finally`` so a tick whose observation
    # raises -- the one a post-mortem most wants -- is still recorded, with whatever extras it
    # got as far as (``_record_step`` pads the rest).
    try:
        if self.sync_before_commands:  # synced above, once: sync_scene must not run twice a tick
            return self._observe(self.state, post_physics_refresh=True)
        return self.state_to_observation(self.state, post_physics_refresh=True)
    finally:
        if payload is not None:
            self._record_step(payload, self.state)

reset

reset() -> dict[str, Any]

Boot in STOP/hold and return the first observation: nothing moves until the operator says so.

Every backend runs the same INIT/START/STOP/KILL state machine (see "The operator state machine" in docs/guides/deployment.md). After this, :meth:step holds the measured pose until INIT ramps the joints to the init_target pose and START hands control to the policy -- from the robot's remote, the keyboard, or a script's :meth:request_init / :meth:request_start. With none of them open, a hardware backend holds in STOP for good.

Source code in eden/extensions/deployment/base.py

def reset(self) -> dict[str, Any]:
    """Boot in STOP/hold and return the first observation: nothing moves until the operator says so.

    Every backend runs the same INIT/START/STOP/KILL state machine (see "The operator state
    machine" in ``docs/guides/deployment.md``). After this, :meth:`step` holds the measured pose
    until INIT ramps the joints to the ``init_target`` pose and START hands control to the
    policy -- from the robot's remote, the keyboard, or a script's :meth:`request_init` /
    :meth:`request_start`. With none of them open, a hardware backend holds in STOP for good.
    """
    # Drop the target low-pass state so the new episode warm-starts from its first command
    # instead of smoothing across the reset from the stale pre-reset target (no-op when
    # target_lpf == 1).
    self._lpf_prev = None
    self._policy_ticks = 0
    # A new episode: its first read starts every observation window afresh, as an env reset does.
    self._observations_stale = True
    self._enter_boot_hold()
    self._prepare_reset()
    self.state = self.read_state()
    # Rewind now so the clip waits at frame 0 through STOP and INIT; START rewinds again.
    self.reset_motion_reference()
    self._after_reset_read()
    self._setup_inputs()
    self._print_control_guide()
    return self.state_to_observation(self.state)

reset_motion_reference

reset_motion_reference() -> None

Restart a motion-imitation reference cleanly at frame 0 (no-op for non-mimic envs).

Both reference paths are rewound, for the same reason :meth:_advance_motion_reference advances both: a task drives its reference through the motion manager, through a MotionCommand, or through both, and the deploy loop is the only thing that reaches either one.

• the manager path — :meth:MotionManager.rewind_to_start for each env's currently-assigned clip, then a refill of the reference buffer;
• the command path — :meth:CommandTerm.rewind_reference, the clock-only rewind.

Motion selection is NOT resampled on either (unlike :meth:MotionManager.reset): a deploy replays the same clip deterministically instead of the sim's random start phase (motion_start_from_middle_range), which is correct for a robot that begins from a stand. Called at deploy reset (before its INIT loop), at INIT, at START, and on every policy switch.

Rewinding the command path matters most at the handover the operator gate creates. The clip holds while the robot is in INIT or hold (:meth:step does not advance it), so without this a STOP at frame 500 followed by a START hands the policy a reference 500 frames into the motion while the robot is standing in its INIT stance — a mid-clip pose it is nowhere near, on a biped.

Source code in eden/extensions/deployment/base.py

def reset_motion_reference(self) -> None:
    """Restart a motion-imitation reference cleanly at frame 0 (no-op for non-mimic envs).

    Both reference paths are rewound, for the same reason :meth:`_advance_motion_reference`
    advances both: a task drives its reference through the motion *manager*, through a
    ``MotionCommand``, or through both, and the deploy loop is the only thing that reaches
    either one.

    * the **manager** path — :meth:`MotionManager.rewind_to_start` for each env's
      currently-assigned clip, then a refill of the reference buffer;
    * the **command** path — :meth:`CommandTerm.rewind_reference`, the clock-only rewind.

    Motion selection is NOT resampled on either (unlike :meth:`MotionManager.reset`): a deploy
    replays the same clip deterministically instead of the sim's random start phase
    (``motion_start_from_middle_range``), which is correct for a robot that begins from a
    stand. Called at deploy reset (before its INIT loop), at INIT, at START, and on every
    policy switch.

    Rewinding the command path matters most at the handover the operator gate creates. The
    clip holds while the robot is in INIT or hold (:meth:`step` does not advance it), so
    without this a STOP at frame 500 followed by a START hands the policy a reference 500
    frames into the motion while the robot is standing in its INIT stance — a mid-clip pose it
    is nowhere near, on a biped.
    """
    if self._env.command_manager is not None:
        self._env.command_manager.rewind_reference()
    # Rewinding re-arms the clip, so a latched end-of-motion action ends and the policy
    # drives again -- the handover an operator START or a policy switch produces.
    self._standing_from = None
    self._stand_count = 0
    self._motion_end_switched = False
    mm = self._motion_manager
    if mm is None:
        return
    all_idx = torch.arange(self._env.num_envs, device=self._env.device)
    mm.rewind_to_start(all_idx)
    mm.fill_data()

state_to_observation

state_to_observation(state: RobotState, *, post_physics_refresh: bool = False) -> dict[str, Any]

Sync state onto the env's robot entity and compute the policy's observation dict.

post_physics_refresh is the deploy loop's phase 7. :meth:step passes True so an observation term with per-step state (:meth:ObservationTerm.post_physics_refresh <eden.managers.observation_manager.ObservationTerm.post_physics_refresh>) advances exactly once per control tick, on the freshly synced robot state and before the read — the sim step runs it at phase 7 and a deploy that skipped it would hand the policy a value frozen at its last simulated step, silently (the class of bug _advance_motion_reference documents). The reset paths leave it False: EnvBase.reset runs no phase 7 either, so a term reads as zero there in both places. A command term's post_physics_refresh still does not run on this path (see :meth:CommandTerm.post_physics_refresh <eden.managers.command_manager.CommandTerm.post_physics_refresh>); the observation hook does because the policy's observations are the hardware path.

The first read of an episode -- after construction, :meth:reset or START -- also clears the observation manager's per-step state and the action manager's last actions, in the order an env reset does: this tick's phase 7, then reset(), then post_reset() (a term that rebuilds its state from the current frame, as ParkourProprio backfills its history, needs it), then the read, which backfills every history window from the robot. Without it the policy's first frames come from the training wrapper's env.reset() on the simulated robot at its spawn pose, or, at START, from the hold and INIT ramp in the yaw frame START just replaced. Clearing on the read rather than in each reset covers a backend override that does not chain to this class's, and a loop that drives :meth:step with no reset at all.

Overrides must accept the keyword (and forward it to super()); the constructor refuses one that does not (:meth:_check_state_to_observation_override).

One tick of skew relative to the sim, inherent to this loop's order: :meth:step publishes the new command (_update_commands) before read_state, so a term that integrates command − measured here pairs this tick's measurement with the command just published, where phase 7 in the sim pairs it with the command that was in effect during the physics.

Source code in eden/extensions/deployment/base.py

def state_to_observation(self, state: RobotState, *, post_physics_refresh: bool = False) -> dict[str, Any]:
    """Sync ``state`` onto the env's robot entity and compute the policy's observation dict.

    ``post_physics_refresh`` is the deploy loop's phase 7. :meth:`step` passes ``True`` so an
    observation term with per-step state (:meth:`ObservationTerm.post_physics_refresh
    <eden.managers.observation_manager.ObservationTerm.post_physics_refresh>`) advances exactly once
    per control tick, on the freshly synced robot state and before the read — the sim step runs it
    at phase 7 and a deploy that skipped it would hand the policy a value frozen at its last
    simulated step, silently (the class of bug ``_advance_motion_reference`` documents). The reset
    paths leave it ``False``: ``EnvBase.reset`` runs no phase 7 either, so a term reads as zero
    there in both places. A command term's ``post_physics_refresh`` still does not run on this path
    (see :meth:`CommandTerm.post_physics_refresh
    <eden.managers.command_manager.CommandTerm.post_physics_refresh>`); the observation hook does
    because the policy's observations *are* the hardware path.

    The first read of an episode -- after construction, :meth:`reset` or START -- also clears the
    observation manager's per-step state and the action manager's last actions, in the order an
    env reset does: this tick's phase 7, then ``reset()``, then ``post_reset()`` (a term that
    rebuilds its state from the current frame, as ``ParkourProprio`` backfills its history, needs
    it), then the read, which backfills every history window from the robot. Without it the
    policy's first frames come from the training wrapper's ``env.reset()`` on the *simulated*
    robot at its spawn pose, or, at START, from the hold and INIT ramp in the yaw frame START just
    replaced. Clearing
    on the read rather than in each ``reset`` covers a backend override that does not chain to
    this class's, and a loop that drives :meth:`step` with no reset at all.

    **Overrides must accept the keyword** (and forward it to ``super()``); the constructor refuses
    one that does not (:meth:`_check_state_to_observation_override`).

    One tick of skew relative to the sim, inherent to this loop's order: :meth:`step` publishes the
    new command (``_update_commands``) *before* ``read_state``, so a term that integrates
    ``command − measured`` here pairs this tick's measurement with the command just published,
    where phase 7 in the sim pairs it with the command that was in effect during the physics.
    """
    self._sync_state(state)
    return self._observe(state, post_physics_refresh=post_physics_refresh)

restart_observations

restart_observations() -> dict[str, Any]

Start every observation window afresh from the last reading, as START does; returns the observation.

The deploy loop calls it when it switches policy slots, so the policy switched in reads what an env reset would have handed it, with every history window backfilled from the robot as it is now, a fed-back last action of zero, and the motion reference the switch just rewound. Without it the switched-in policy reads frames observed while another slot drove, with the reference still mid-clip, and its own last action from before it was switched out. The reading is not re-synced and the velocity estimator not re-run: the state is the one the last :meth:step already observed.

Source code in eden/extensions/deployment/base.py

def restart_observations(self) -> dict[str, Any]:
    """Start every observation window afresh from the last reading, as START does; returns the observation.

    The deploy loop calls it when it switches policy slots, so the policy switched in reads what
    an env reset would have handed it, with every history window backfilled from the robot as it
    is now, a fed-back last action of zero, and the motion reference the switch just rewound.
    Without it the switched-in policy reads frames observed while another slot drove, with the
    reference still mid-clip, and its own last action from before it was switched out. The
    reading is not re-synced and the velocity estimator not re-run: the state is the one the last
    :meth:`step` already observed.
    """
    self._observations_stale = False
    self._restart_observed_episode()
    return self._env.observation_manager.compute(update_history=True)

connect abstractmethod

connect() -> None

Source code in eden/extensions/deployment/base.py

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

close abstractmethod

close() -> None

Source code in eden/extensions/deployment/base.py

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

read_state abstractmethod

read_state() -> RobotState

Source code in eden/extensions/deployment/base.py

@abstractmethod
def read_state(self) -> RobotState:
    raise NotImplementedError

send_payload abstractmethod

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/base.py

@abstractmethod
def send_payload(self, payload: RobotCommand) -> None:
    raise NotImplementedError

SoftShutdownMixin

SoftShutdownMixin(options: T)

Bases: ConfigurableMixin, ABC

Settle the robot into a damping state before a backend's close drops its link.

Dropping the link outright leaves the robot holding its last stiff PD target until an onboard comms watchdog trips, and a standing robot can lurch or collapse. :meth:_soft_shutdown instead sends shutdown_damping_steps ticks of kp=0, kd>0: the position servo is off, only velocity damping remains, and the joints settle gently.

With a :class:~eden.extensions.deployment.utils.republisher.CommandRepublisherMixin running, no watchdog can trip while the process lives: a control thread that stalls leaves the loop re-sending its last command at full gains, and the loop reports that rather than acting on it (see "A stalled control thread is reported, not acted on" in docs/guides/deployment.md). This ramp is then the only orderly way down, which is why such a backend's close runs it before _stop_publisher -- the damping command goes out at the same packet rate as any other.

Mixed in ahead of :class:DeploymentBase by the backends that run the ramp, which call :meth:_soft_shutdown from close and implement :meth:_shutdown_hold_pos. The two options live here rather than on :class:DeploymentBase so a backend without the ramp keeps rejecting them instead of accepting and ignoring them.

shutdown_damping_steps: int Control ticks to send the damping command before close releases the link. shutdown_damping_kd_scale: float Scale on kd during that soft shutdown.

Source code in eden/utils/common.py

def __init__(self, options: T) -> None:
    self._uid = gs.UID()
    self._options: T = options

    mirror_options_onto(self, self._options)

    # Deferred, not module-scope — see the NOTE above this module's imports.
    from eden.options._metadata import OPTION_METADATA_KEYS

    # check if there are unused parameters in the options. ``_extra_option_params`` are legitimate
    # options-only fields (not mirrored as attrs), so they are expected to be "unused" here.
    unused_params = set(self._options.keys()) - set(self.get_parameter_names())
    tolerated = {"name", "class_name"} | set(OPTION_METADATA_KEYS) | self._extra_option_params
    # ``params`` is a declared framework field on function-capable term options. An *empty* one is
    # legitimately unused when the same options class configures a class-based term; a non-empty
    # ``params`` on a class-based term would be silently ignored, so it stays invalid.
    if not getattr(self._options, "params", None):
        tolerated.add("params")
    invalid_params = unused_params - tolerated
    if invalid_params:
        # Name the term, not only the class: for a function term the class is the shared
        # ``*FuncWrapper``, which cannot tell the author which of a dozen terms is wrong.
        # Same wording as the construction-time check (``ManagerTermOptions._validate_param_names``).
        term_name = getattr(self._options, "name", None)
        subject = f"term '{term_name}' ({type(self).__name__})" if term_name else type(self).__name__
        raise ValueError(
            invalid_parameters_message(subject, invalid_params, self.get_configure_specs(), self._retired_params)
        )

Attributes

shutdown_damping_steps class-attribute instance-attribute

shutdown_damping_steps: int = 50

shutdown_damping_kd_scale class-attribute instance-attribute

shutdown_damping_kd_scale: float = 1.0

Functions:

bind_deployment

bind_deployment(robot_cls: type, deployment_name: str) -> None

Register robot_cls as driven by the named deployment in :data:DEPLOYMENT_REGISTRY.

Resolution walks the robot's MRO, so subclasses inherit the binding unless they register their own override. Call this once per robot class in the deployment module that supports it; collisions are not detected (the last binding wins).

Source code in eden/extensions/deployment/base.py

def bind_deployment(robot_cls: type, deployment_name: str) -> None:
    """Register ``robot_cls`` as driven by the named deployment in :data:`DEPLOYMENT_REGISTRY`.

    Resolution walks the robot's MRO, so subclasses inherit the binding
    unless they register their own override. Call this once per robot class
    in the deployment module that supports it; collisions are not detected
    (the last binding wins).
    """
    _ROBOT_DEPLOYMENT_BINDINGS[robot_cls] = deployment_name

resolve_deployment_for

resolve_deployment_for(robot_options: RobotOptions | CompositeEntityOptions) -> type[DeploymentBase]

Return the deployment class registered for type(robot_options).

Walks the MRO to honor inheritance (a subclass with no binding falls back to its parent's). Raises KeyError if no ancestor is bound.

A composite rig (:class:~eden.options.entities.CompositeEntityOptions — an arm with a hand welded on, a humanoid with hands) resolves to composite_deployment, which drives each of its parts through that part's own binding.

Source code in eden/extensions/deployment/base.py

def resolve_deployment_for(robot_options: RobotOptions | CompositeEntityOptions) -> type[DeploymentBase]:
    """Return the deployment class registered for ``type(robot_options)``.

    Walks the MRO to honor inheritance (a subclass with no binding falls
    back to its parent's). Raises ``KeyError`` if no ancestor is bound.

    A composite rig (:class:`~eden.options.entities.CompositeEntityOptions` — an arm with a hand
    welded on, a humanoid with hands) resolves to ``composite_deployment``, which drives each of
    its parts through that part's own binding.
    """
    if isinstance(robot_options, CompositeEntityOptions):
        return DEPLOYMENT_REGISTRY.get("composite_deployment")
    cls = type(robot_options)
    for ancestor in cls.__mro__:
        if ancestor in _ROBOT_DEPLOYMENT_BINDINGS:
            name = _ROBOT_DEPLOYMENT_BINDINGS[ancestor]
            return DEPLOYMENT_REGISTRY.get(name)
    available = sorted({k.__name__ for k in _ROBOT_DEPLOYMENT_BINDINGS}) or ["<none>"]
    raise KeyError(
        f"No deployment registered for robot {cls.__name__}. "
        f"Available bindings: {available}. Add one with "
        f"``bind_deployment({cls.__name__}, '<deployment_name>')``."
    )

resolve_pinned_gains

resolve_pinned_gains(robot: Any, dofs_name: Sequence[str]) -> tuple[dict[str, float], dict[str, float]]

A policy's pinned gains on the DOFs of dofs_name it names, as ({dof: kp}, {dof: kd}).

A default_dofs_kp / default_dofs_kd that robot leaves empty (G1Dofs23's, which trains on the asset's gains), or a robot with no dofs_name at all (None, a composite), resolves to {}: it declares no gains, and the deployed robot's apply. :meth:DeploymentBase.wire_gains builds a slot's wire gains from this.

Needs no deployment, so eden deploy runs it for every slot before constructing one: a raise here must land before the entity swap and before an operator device is opened.

Raises ValueError when a non-empty gain map leaves out a joint of dofs_name the robot has (and when a pattern in it matches none of the robot's joints): see "Each slot runs at its own PD gains" in docs/guides/deployment.md.

Source code in eden/extensions/deployment/base.py

def resolve_pinned_gains(robot: Any, dofs_name: Sequence[str]) -> tuple[dict[str, float], dict[str, float]]:
    """A policy's pinned gains on the DOFs of ``dofs_name`` it names, as ``({dof: kp}, {dof: kd})``.

    A ``default_dofs_kp`` / ``default_dofs_kd`` that ``robot`` leaves empty (``G1Dofs23``'s,
    which trains on the asset's gains), or a ``robot`` with no ``dofs_name`` at all (``None``, a
    composite), resolves to ``{}``: it declares no gains, and the deployed robot's apply.
    :meth:`DeploymentBase.wire_gains` builds a slot's wire gains from this.

    Needs no deployment, so ``eden deploy`` runs it for every slot before constructing one: a
    raise here must land before the entity swap and before an operator device is opened.

    Raises ``ValueError`` when a non-empty gain map leaves out a joint of ``dofs_name`` the
    robot has (and when a pattern in it matches none of the robot's joints): see "Each slot
    runs at its own PD gains" in ``docs/guides/deployment.md``.
    """
    robot_dofs = getattr(robot, "dofs_name", None)
    if robot_dofs is None:
        return {}, {}
    robot_dofs = list(robot_dofs)
    deployed = set(dofs_name)
    driven = [name for name in robot_dofs if name in deployed]
    resolved: list[dict[str, float]] = []
    for field in ("default_dofs_kp", "default_dofs_kd"):
        pinned = getattr(robot, field, None)
        if not pinned:  # declares no gains: the deployed robot's
            resolved.append({})
            continue
        idx, _, values = resolve_matching_names_values(pinned, robot_dofs, preserve_order=True)
        covered = {robot_dofs[j]: float(value) for j, value in zip(idx, values, strict=True)}
        if uncovered := [name for name in driven if name not in covered]:
            owner = type(robot).__name__
            raise ValueError(
                f"{owner}'s {field} gives no value for {first_few(uncovered)}, which this deployment drives. A "
                "slot runs at its pinned robot's gains; a joint they leave out took the asset's own gain in "
                "training, which a deployment cannot read. Give every joint a value in the policy's "
                f"robot_options(), e.g. {field}={{**{owner}().{field}, ...}} (pydantic replaces the whole dict, "
                "it does not merge). Do not fall back to 0 or to the deployed robot's gain: either is a guess on "
                "a live robot."
            )
        resolved.append({name: covered[name] for name in driven})
    return resolved[0], resolved[1]

resolve_dofs_values

resolve_dofs_values(data: dict[str, float], dofs_name: list[str]) -> ndarray

Resolve a pattern-keyed per-DOF dict into an array aligned with dofs_name.

:func:resolve_matching_names_values with preserve_order=True returns matches in query-pattern order, so the values must be scattered through the returned indices — zipping them positionally with dofs_name scrambles any dict with wildcard keys. DOFs not covered by any pattern resolve to 0.0 (with a warning: a zero kp/kd reaching a real robot means a limp joint).

Source code in eden/extensions/deployment/base.py

def resolve_dofs_values(data: dict[str, float], dofs_name: list[str]) -> np.ndarray:
    """Resolve a pattern-keyed per-DOF dict into an array aligned with ``dofs_name``.

    :func:`resolve_matching_names_values` with ``preserve_order=True`` returns
    matches in *query-pattern* order, so the values must be scattered through
    the returned indices — zipping them positionally with ``dofs_name``
    scrambles any dict with wildcard keys. DOFs not covered by any pattern
    resolve to 0.0 (with a warning: a zero kp/kd reaching a real robot means a
    limp joint).
    """
    idx, _, values = resolve_matching_names_values(data, dofs_name, preserve_order=True)
    resolved = np.zeros(len(dofs_name), dtype=np.float64)
    resolved[idx] = values
    if len(idx) < len(dofs_name):
        missing = [name for i, name in enumerate(dofs_name) if i not in set(idx)]
        en.logger.warning(f"[deployment] No value matched for DOFs {missing}; defaulting to 0.0.")
    return resolved

Options

Base configuration options for real-robot deployment backends.

Classes

DeploymentOptions

DeploymentOptions(**data)

Bases: DynamicConfigurableOptions

Options for deployment extensions.

PARAMETER	DESCRIPTION
entity_name	The name of the entity to deploy
control_dt	The control time step in seconds. Defaults to None, in which case it is determined by the control frequency, fallback to sim dt * decimation.
control_freq	The control frequency in Hz. Defaults to None, in which case it is determined by 1/control_dt.
decimation	The decimation factor. Defaults to None, in which case sim decimation is used.
sync	Whether to synchronize the control loop. Defaults to True.
connect_timeout_s	The timeout for connecting to the robot. Defaults to 5.0 seconds.
auto_yaw_align	Capture the robot's IMU yaw at START (the MuJoCo backend also captures it on reset's read) and remove it from subsequent world-frame observations so the deployment world frame matches the (yaw-aligned) world frame the policy was trained in. Defaults to True. Disable when an external estimator already provides a yaw-consistent base orientation (e.g. mocap).

Source code in eden/options/options.py

def __init__(self, **data):
    # Seed type metadata as construction data so BaseModel validates it into
    # __pydantic_extra__ (no-op for extra="forbid" classes, which use
    # the serialization envelope instead).
    data.update(option_metadata(self))
    # NOTE: all of the option modifications/validations should be done in the __init__ method
    try:
        BaseModel.__init__(self, **data)
    except ValidationError as exc:
        # A misspelled field on an extra="forbid" class should teach the field set, not just
        # name the typo. Only rewritten when every error is extra_forbidden.
        enriched = _enrich_extra_forbidden(type(self), exc)
        if enriched is None:
            raise
        raise enriched from None

Attributes

entity_name class-attribute instance-attribute

entity_name: str = 'robot'

control_dt class-attribute instance-attribute

control_dt: float | None = None

control_freq class-attribute instance-attribute

control_freq: float | None = None

decimation class-attribute instance-attribute

decimation: int | None = None

sync class-attribute instance-attribute

sync: bool = True

connect_timeout_s class-attribute instance-attribute

connect_timeout_s: float = 5.0

auto_yaw_align class-attribute instance-attribute

auto_yaw_align: bool = True

Hardware backends

Unitree G1

Unitree G1 real-robot deployment backend.

Attributes

G1_JOINT_IDX module-attribute

G1_JOINT_IDX: dict[str, int] = {'left_hip_pitch_joint': 0, 'left_hip_roll_joint': 1, 'left_hip_yaw_joint': 2, 'left_knee_joint': 3, 'left_ankle_pitch_joint': 4, 'left_ankle_roll_joint': 5, 'right_hip_pitch_joint': 6, 'right_hip_roll_joint': 7, 'right_hip_yaw_joint': 8, 'right_knee_joint': 9, 'right_ankle_pitch_joint': 10, 'right_ankle_roll_joint': 11, 'waist_yaw_joint': 12, 'waist_roll_joint': 13, 'waist_pitch_joint': 14, 'left_shoulder_pitch_joint': 15, 'left_shoulder_roll_joint': 16, 'left_shoulder_yaw_joint': 17, 'left_elbow_joint': 18, 'left_wrist_roll_joint': 19, 'left_wrist_pitch_joint': 20, 'left_wrist_yaw_joint': 21, 'right_shoulder_pitch_joint': 22, 'right_shoulder_roll_joint': 23, 'right_shoulder_yaw_joint': 24, 'right_elbow_joint': 25, 'right_wrist_roll_joint': 26, 'right_wrist_pitch_joint': 27, 'right_wrist_yaw_joint': 28}

Classes

UnitreeG1Deployment

UnitreeG1Deployment(env: EnvBase, options: DeploymentOptions)

Bases: _UnitreeDdsDeployment

Drive a Unitree G1 over the unitree_hg DDS low-level interface.

Configurable options (in addition to :class:~eden.extensions.deployment.unitree_dds._UnitreeDdsDeployment's):

publish_rate: float | None Rate at which the held LowCmd_ is re-sent between control ticks (default 500 Hz, Unitree's own low-level rate). None sends one packet per control tick.

The G1's ankle *and* -- on the 3-DOF-waist 29-DOF build -- waist roll/pitch are
parallel A/B linkages, driven here in PR (pitch/roll) mode via ``mode_pr``. That is
the same joint class whose per-packet PD made the LimX Oli's waist buzz at one
packet per 50 Hz tick: the derivative term acts on a 20 ms-old velocity,
which at 25 Hz is anti-damping. 500 Hz is what both vendor references use --
Unitree's own ``g1_low_level_example`` publishes at ``control_dt_ = 0.002``, and
NVIDIA's SONIC G1 deploy runs a dedicated 500 Hz ``LowCommandWriter`` thread beside
its 50 Hz control thread -- so it is the rate the robot is expected to be fed at,
independent of how fast any policy runs.

Unlike the Oli case this has **not** been measured on a G1: the mechanism and the
rate are taken from those two references, not from a hardware fit. ``--publish_rate 0``
sends one packet per tick, which is the A/B to run before concluding anything.

Source code in eden/extensions/deployment/unitree_g1.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    self._mode_machine = None

Attributes

publish_rate class-attribute instance-attribute

publish_rate: float | None = 500.0

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/unitree_g1.py

def connect(self) -> None:
    # Release any running high-level mode first, and re-learn ``mode_machine`` from the new stream.
    self._mode_machine = None
    self._release_motion_mode()
    super().connect()

Functions:

Unitree Go2

Unitree Go2 real-robot deployment backend.

Attributes

GO2_JOINT_IDX module-attribute

GO2_JOINT_IDX: dict[str, int] = {'FR_hip_joint': 0, 'FR_thigh_joint': 1, 'FR_calf_joint': 2, 'FL_hip_joint': 3, 'FL_thigh_joint': 4, 'FL_calf_joint': 5, 'RR_hip_joint': 6, 'RR_thigh_joint': 7, 'RR_calf_joint': 8, 'RL_hip_joint': 9, 'RL_thigh_joint': 10, 'RL_calf_joint': 11}

Classes

UnitreeGo2Deployment

UnitreeGo2Deployment(env: EnvBase, options: DeploymentOptions)

Bases: _UnitreeDdsDeployment

Drive a Unitree Go2 over the unitree_go DDS low-level interface.

Configurable options (in addition to :class:~eden.extensions.deployment.unitree_dds._UnitreeDdsDeployment's):

publish_rate: float | None Rate at which the held LowCmd_ is re-sent between control ticks (default 500 Hz). None sends one packet per control tick.

This is vendor-rate parity, not a fault fix. The Go2's twelve joints are all serial --
its ``LowCmd_`` has no ``mode_pr``, as there is no parallel linkage to transform -- so
the per-packet parallel-linkage fault the republisher exists for on the Oli and G1
does not apply here. 500 Hz is what Unitree's own Go2 low-level example feeds the robot
at (``go2_stand_example.py``: ``dt = 0.002`` and a 2 ms ``RecurrentThread`` writer),
independent of any policy rate; one packet per 50 Hz tick is outside that example's own
``0.001~0.01`` s range. ``--publish_rate 0`` sends one packet per tick.

Source code in eden/extensions/deployment/unitree_dds.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    self._latest_msg: Any | None = None
    self._msg_event = threading.Event()
    self._subscriber = None
    self._publisher = None
    self._low_cmd = None
    self._crc = None
    self._init_republisher()
    # Whether ``_low_cmd`` holds a command some caller actually asked for. Until it does, a
    # re-send would push ``_init_low_cmd``'s limp all-zero packet -- harmless, but nothing
    # goes on the wire before the first ``send_payload``, republisher included.
    self._cmd_ready = False

Attributes

publish_rate class-attribute instance-attribute

publish_rate: float | None = 500.0

Functions:

Unitree Go1

Unitree Go1 real-robot deployment backend (legacy robot_interface UDP SDK).

Mirrors the structure of :mod:eden.extensions.deployment.unitree_go2 — same registry/binding, _dof_index Eden→SDK map, and Eden-order RobotState/RobotCommand contract — but the transport is the legacy unitree_legged_sdk (synchronous UDP.Recv/Send + Safety) rather than the DDS unitree_sdk2py stack Go2/G1 use.

Attributes

GO1_JOINT_IDX module-attribute

GO1_JOINT_IDX: dict[str, int] = {'FR_hip_joint': 0, 'FR_thigh_joint': 1, 'FR_calf_joint': 2, 'FL_hip_joint': 3, 'FL_thigh_joint': 4, 'FL_calf_joint': 5, 'RR_hip_joint': 6, 'RR_thigh_joint': 7, 'RR_calf_joint': 8, 'RL_hip_joint': 9, 'RL_thigh_joint': 10, 'RL_calf_joint': 11}

Classes

UnitreeGo1Deployment

UnitreeGo1Deployment(env: EnvBase, options: DeploymentOptions)

Bases: SoftShutdownMixin, DeploymentBase

Source code in eden/extensions/deployment/unitree_go1.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    self._udp = None
    self._safe = None
    self._cmd = None
    self._state = None

    # Per-joint thermal-warning dedupe (warn once until the motor recovers).
    self._thermal_warned: set[str] = set()

Attributes

robot_ip class-attribute instance-attribute

robot_ip: str = '192.168.123.10'

robot_port class-attribute instance-attribute

robot_port: int = 8007

local_port class-attribute instance-attribute

local_port: int = 8080

motor_mode class-attribute instance-attribute

motor_mode: int = 10

power_protect_factor class-attribute instance-attribute

power_protect_factor: int = 10

motor_temp_warn_c class-attribute instance-attribute

motor_temp_warn_c: int = 80

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/unitree_go1.py

def connect(self) -> None:
    ensure_go1_sdk()

    self._udp = sdk.UDP(LOWLEVEL, self.local_port, self.robot_ip, self.robot_port)
    self._safe = sdk.Safety(sdk.LeggedType.Go1)
    self._cmd = sdk.LowCmd()
    self._state = sdk.LowState()
    self._udp.InitCmdData(self._cmd)
    self._init_low_cmd()

    # Prime the bidirectional UDP link and wait for the first valid low-state
    # packet. The robot only streams state back to a client it has already
    # heard from, so we must SEND while polling: the ``_init_low_cmd`` command
    # is limp (kp=kd=tau=0, so no torque / no motion) and safe to send before
    # we have any state. A Recv-only wait hangs forever whenever the robot
    # isn't already streaming (e.g. after a previous client closed).
    # ``tick`` (ms) stays 0 until the control board is heard from.
    start_time = time.time()
    while True:
        self._udp.SetSend(self._cmd)
        self._udp.Send()
        self._refresh_state()
        if self._state.tick != 0:
            break
        if time.time() - start_time > self.connect_timeout_s:
            raise RuntimeError(
                "Timed out waiting for Unitree Go1 low state. Check the robot is powered, "
                "in low-level mode (L2+A → L2+B → L1+L2+start on the remote), and reachable "
                f"at {self.robot_ip}:{self.robot_port} from local UDP port {self.local_port}."
            )
        time.sleep(_PRIME_DT_S)

close

close() -> None

Source code in eden/extensions/deployment/unitree_go1.py

def close(self) -> None:
    # Flush the debug recording and release the operator input device, then settle to a
    # damping state before dropping the handles, so the robot doesn't hold its last stiff
    # target until the onboard watchdog trips. The legacy SDK has no explicit teardown;
    # nulling the handles frees the UDP socket for a clean reconnect.
    self._teardown_common()
    self._soft_shutdown()
    self._udp = None
    self._safe = None
    self._cmd = None
    self._state = None

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/unitree_go1.py

def read_state(self) -> RobotState:
    self._require_connected()
    self._refresh_state()
    state = self._state

    dofs_pos = np.zeros(self.num_dofs, dtype=np.float64)
    dofs_vel = np.zeros(self.num_dofs, dtype=np.float64)
    dofs_torque = np.zeros(self.num_dofs, dtype=np.float64)
    # Motor health (Eden dof order): recorded to --debug_record and used to warn on
    # thermal derating (a frozen/sagging leg is often an overheated motor cutting its
    # own torque, not a controller/policy issue).
    temps = np.zeros(self.num_dofs, dtype=np.float64)
    modes = np.zeros(self.num_dofs, dtype=np.float64)
    full_dofs_pos = np.zeros(self._num_full_dof, dtype=np.float64)

    for i, motor_idx in enumerate(self._dof_index):
        motor = state.motorState[motor_idx]
        dofs_pos[i] = motor.q
        dofs_vel[i] = motor.dq
        dofs_torque[i] = motor.tauEst
        temps[i] = motor.temperature
        modes[i] = motor.mode

    for i in range(self._num_full_dof):
        full_dofs_pos[i] = state.motorState[i].q

    self._warn_thermal(temps, modes)

    return RobotState(
        stamp=time.time(),
        base_quat=np.array(state.imu.quaternion, dtype=np.float64),  # wxyz
        base_ang_vel=np.array(state.imu.gyroscope, dtype=np.float64),  # body frame
        base_lin_acc=np.array(state.imu.accelerometer, dtype=np.float64),
        dofs_pos=dofs_pos,
        dofs_vel=dofs_vel,
        dofs_torque=dofs_torque,
        full_dofs_pos=full_dofs_pos,
        extra={"motor_temp": temps, "motor_mode": modes, **self._battery_extra()},
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/unitree_go1.py

def send_payload(self, payload: RobotCommand) -> None:
    self._require_connected()

    dofs_pos = self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64))

    for i, motor_idx in enumerate(self._dof_index):
        motor_cmd = self._cmd.motorCmd[motor_idx]
        motor_cmd.q = float(dofs_pos[i])
        motor_cmd.dq = float(payload.dofs_vel[i])
        motor_cmd.tau = float(payload.dofs_torque[i])
        if payload.dofs_kp is not None:
            motor_cmd.Kp = float(payload.dofs_kp[i])
        if payload.dofs_kd is not None:
            motor_cmd.Kd = float(payload.dofs_kd[i])

    # Refresh state first so PowerProtect compares against the current measurement,
    # not the one read at the end of the previous tick -- or none at all: the soft
    # shutdown ramp sends without reading.
    # CRC is computed inside SetSend (no manual CRC, unlike the DDS backends).
    self._refresh_state()
    self._safe.PowerProtect(self._cmd, self._state, self.power_protect_factor)
    self._udp.SetSend(self._cmd)
    self._udp.Send()

Functions:

LimX Oli

LimX Dynamics Oli (HU_D04) real-robot deployment backend.

Talks to the robot through the LimX low-level SDK (limxsdk, the Python wheel shipped in https://github.com/limxdynamics/limxsdk-lowlevel): joint state and IMU arrive on subscription callbacks, commands go out via publishRobotCmd. Command/state arrays are indexed in the robot's motor order, which the backend maps to/from Eden's dofs_name order by motor name read live from the robot (getMotorNames()) — LimX's own docs and shipped controller YAMLs disagree on the head-joint order, so a positional map would be silently wrong.

Commands are sent in PR (pitch/roll, serial-joint) space with the SDK's parallel_solve_required flag set, so the SDK converts to the physical A/B motors of the parallel ankle/waist linkages. This matches Eden's :class:~eden.options.robots.limx_oli.Oli model (the serial HU_D04_01.urdf) and LimX's own RL deployment stack; state read back in this mode is PR-space as well.

Attributes

IS_LIMX_SDK_AVAILABLE module-attribute

IS_LIMX_SDK_AVAILABLE = True

LIMX_OLI_CLAMP_TORQUE_LIMITS module-attribute

LIMX_OLI_CLAMP_TORQUE_LIMITS: dict[str, float] = {'*_ankle_*_joint': 80.0, 'waist_roll_joint': 80.0, 'waist_pitch_joint': 80.0}

Classes

LimxOliDeployment

LimxOliDeployment(env: EnvBase, options: DeploymentOptions)

Bases: CommandRepublisherMixin, SoftShutdownMixin, DeploymentBase

LimX Oli deployment backend on top of :class:DeploymentBase.

Adds, on top of the base options:

• robot_ip: IP the LimX SDK connects to. None (default) resolves to the ROBOT_IP environment variable, then to the real robot's 10.192.1.2; point it at 127.0.0.1 for the LimX MuJoCo simulator.
• shutdown_damping_steps / shutdown_damping_kd_scale: soft shutdown — cycles to hold a damping (kp=0, kd) state before releasing.
• soft_torque_limit: fraction of each joint's torque budget the position term of the implied PD torque may use; every commanded target is clamped to q ± soft_torque_limit * budget / kp around the measured pose. Same 0.95 default as LimX's own walk controller, but without the velocity term theirs includes -- see :meth:_clamp_torque_feasible for the hardware measurements behind that. 0 disables the clamp.
• clamp_torque_limits: DOF-name pattern -> N·m overrides on top of the robot's default_dofs_force_limits for that budget. Defaults to LimX's deploy-time user_torque_limit where it differs from Eden's training limits: 80 N·m on the six parallel ankle/waist PR joints (two 42 N·m A/B motors each). Training keeps 42 -- the shipped action scale derives from it -- so this is a deploy-only budget.
• publish_rate: rate at which the current wire command is re-sent between control ticks (default 1000 Hz, LimX's own rate). Load-bearing on this robot: in PR mode the robot evaluates the PD law of the six parallel-linkage joints (ankle pitch/roll, waist roll/pitch) once per received packet, on that packet's state -- measured on hardware, where their torque fits kp*(q_des-q) - kd*dq[previous packet] at R^2 0.92-0.99 while the serial joints fit the same-instant velocity. At 50 Hz that damping term acts on a 20 ms-old velocity, which is anti-damping at 25 Hz: the waist-roll linkage rings at exactly that frequency once a foot strike kicks it. None sends one packet per control tick, for A/B comparison only.
• motor_name_map: Eden DOF name → SDK motor name overrides for robots whose getMotorNames() vocabulary doesn't match the URDF joint names (unmapped DOFs still match by suffix-normalized name).
• verbose_diagnostics: also log warning/OK robot diagnostics as they change. By default only error-level (2) diagnostics are logged — the robot's diagnostic stream is chatty and the OK/warning transitions are commissioning detail, but a fault must never be silent.

Source code in eden/extensions/deployment/limx_oli.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    self._robot: Any | None = None
    self._init_republisher()
    self._latest_state: Any | None = None
    self._latest_imu: Any | None = None
    self._latest_joy: Any | None = None
    # Last (code, level) logged per diagnostic name, so the continuous
    # diagnostic stream is logged only on change.
    self._last_diagnostic: dict[str, tuple[int, int]] = {}
    self._state_event = threading.Event()
    self._imu_event = threading.Event()
    self._dof_index: list[int] | None = None
    self._num_motors: int | None = None
    self._cmd: Any | None = None

Attributes

robot_ip class-attribute instance-attribute

robot_ip: str | None = None

soft_torque_limit class-attribute instance-attribute

soft_torque_limit: float = 0.95

clamp_torque_limits class-attribute instance-attribute

clamp_torque_limits: dict = dict(LIMX_OLI_CLAMP_TORQUE_LIMITS)

motor_name_map class-attribute instance-attribute

motor_name_map: dict = {}

verbose_diagnostics class-attribute instance-attribute

verbose_diagnostics: bool = False

publish_rate class-attribute instance-attribute

publish_rate: float | None = 1000.0

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/limx_oli.py

def connect(self) -> None:
    ensure_limx_sdk()
    self._latest_state = None
    self._latest_imu = None
    self._latest_joy = None
    self._state_event.clear()
    self._imu_event.clear()

    ip = self.robot_ip or os.environ.get("ROBOT_IP") or _DEFAULT_ROBOT_IP
    robot = LimxRobot(LimxRobotType.Humanoid)
    if not robot.init(ip):
        raise RuntimeError(
            f"Failed to connect to LimX Oli at {ip}. Check the robot is powered and reachable "
            "(real robot: 10.192.1.2 over the robot's network; LimX MuJoCo sim: 127.0.0.1), "
            "or override the target with the ROBOT_IP environment variable "
            "(or the robot_ip deployment option)."
        )

    motor_names = list(robot.getMotorNames())
    num_motors = int(robot.getMotorNumber())
    if len(motor_names) != num_motors:
        raise RuntimeError(
            f"LimX SDK reports {num_motors} motors but {len(motor_names)} motor names: {motor_names}"
        )
    self._dof_index = self._build_dof_index(motor_names)
    self._num_motors = num_motors
    driven = set(self._dof_index)
    idle = [name for i, name in enumerate(motor_names) if i not in driven]
    if idle:
        en.logger.warning(
            f"[deploy] {len(idle)} robot motor(s) are not in the task's DOFs and will be held "
            f"limp (kp=kd=0): {idle}"
        )

    self._cmd = self._init_robot_cmd(motor_names)
    self._last_diagnostic.clear()
    robot.subscribeRobotState(self._on_robot_state)
    robot.subscribeImuData(self._on_imu_data)
    robot.subscribeSensorJoy(self._on_sensor_joy)
    robot.subscribeDiagnosticValue(self._on_diagnostic)
    self._robot = robot

    start_time = time.time()
    while not (self._state_event.is_set() and self._imu_event.is_set()):
        if time.time() - start_time > self.connect_timeout_s:
            missing = [
                name
                for name, event in (("robot state", self._state_event), ("IMU", self._imu_event))
                if not event.is_set()
            ]
            raise RuntimeError(f"Timed out waiting for LimX Oli {' and '.join(missing)} stream(s).")
        time.sleep(0.05)

    self._start_publisher()

close

close() -> None

Source code in eden/extensions/deployment/limx_oli.py

def close(self) -> None:
    # Flush the debug recording and release the operator input device before tearing
    # down (both are in-memory, independent of the SDK link).
    self._teardown_common()
    # Ramp to a damping state before dropping the handle so the robot
    # settles softly instead of losing all commands at once.
    self._soft_shutdown()
    self._stop_publisher()
    # The SDK has no unsubscribe/disconnect API; dropping the handle stops
    # our publishing and the daemon-side link dies with the process.
    self._robot = None
    self._cmd = None
    self._latest_state = None
    self._latest_imu = None
    self._latest_joy = None
    self._state_event.clear()
    self._imu_event.clear()

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/limx_oli.py

def read_state(self) -> RobotState:
    if self._latest_state is None or self._latest_imu is None:
        raise RuntimeError("No LimX Oli robot state / IMU data received yet.")

    msg = self._latest_state
    imu = self._latest_imu
    q = np.asarray(msg.q, dtype=np.float64)
    dq = np.asarray(msg.dq, dtype=np.float64)
    tau = np.asarray(msg.tau, dtype=np.float64)
    if q.shape[0] != self._num_motors:
        raise RuntimeError(
            f"LimX robot state carries {q.shape[0]} motors, expected {self._num_motors} "
            "(robot type / firmware mismatch?)."
        )

    return RobotState(
        stamp=time.time(),
        base_quat=np.asarray(imu.quat, dtype=np.float64),  # wxyz, same as Eden
        base_ang_vel=np.asarray(imu.gyro, dtype=np.float64),
        base_lin_acc=np.asarray(imu.acc, dtype=np.float64),
        dofs_pos=q[self._dof_index],
        dofs_vel=dq[self._dof_index],
        dofs_torque=tau[self._dof_index],
        full_dofs_pos=q,
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/limx_oli.py

def send_payload(self, payload: RobotCommand) -> None:
    if self._robot is None or self._cmd is None:
        raise RuntimeError("LimX Oli deployment is not connected.")

    # Low-pass the commanded targets (no-op unless --target_lpf < 1) before
    # they hit the wire, so every path (init/hold/policy) is smoothed, then
    # clamp them to the position-torque budget (the clamp runs last so the
    # wire command is guaranteed within it).
    payload.dofs_pos = self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64))
    payload.dofs_pos = self._clamp_torque_feasible(payload.dofs_pos, payload.dofs_kp)

    with self._cmd_lock:
        cmd = self._cmd
        cmd.stamp = time.time_ns()
        for i, motor_idx in enumerate(self._dof_index):
            cmd.q[motor_idx] = float(payload.dofs_pos[i])
            cmd.dq[motor_idx] = float(payload.dofs_vel[i])
            cmd.tau[motor_idx] = float(payload.dofs_torque[i])
            if payload.dofs_kp is not None:
                cmd.Kp[motor_idx] = float(payload.dofs_kp[i])
            if payload.dofs_kd is not None:
                cmd.Kd[motor_idx] = float(payload.dofs_kd[i])
        self._robot.publishRobotCmd(cmd)
        self._note_command_published()

Functions:

ensure_limx_sdk

ensure_limx_sdk() -> None

Raise a descriptive error if the LimX SDK is not importable.

Source code in eden/extensions/deployment/limx_oli.py

def ensure_limx_sdk() -> None:
    """Raise a descriptive error if the LimX SDK is not importable."""
    if not IS_LIMX_SDK_AVAILABLE:
        raise RuntimeError(
            "LimX SDK (limxsdk) failed to import. Install the platform wheel shipped in "
            "https://github.com/limxdynamics/limxsdk-lowlevel under python3/<arch>/ "
            "(e.g. `uv pip install python3/amd64/limxsdk-*.whl`). "
            f"Underlying error: {_LIMX_SDK_IMPORT_ERROR!r}"
        )

resolve_clamp_torque_limits

resolve_clamp_torque_limits(robot_options: RobotOptions, dofs_name: list[str], overrides: dict[str, float]) -> ndarray

Per-DOF torque budget for the deploy clamp: the robot's force limits with overrides on top.

overrides maps DOF-name patterns to N·m. It is resolved on its own and scattered onto the force-limit array instead of being merged into one dict, because :func:~eden.utils.string.resolve_matching_names_values refuses a DOF matched by two patterns (waist_*_joint from the robot options and waist_roll_joint from the overrides would collide). A pattern matching no DOF raises, so a typo cannot silently leave a joint at the robot's limit.

Source code in eden/extensions/deployment/limx_oli.py

def resolve_clamp_torque_limits(
    robot_options: RobotOptions, dofs_name: list[str], overrides: dict[str, float]
) -> np.ndarray:
    """Per-DOF torque budget for the deploy clamp: the robot's force limits with ``overrides`` on top.

    ``overrides`` maps DOF-name patterns to N·m. It is resolved on its own and scattered onto the
    force-limit array instead of being merged into one dict, because
    :func:`~eden.utils.string.resolve_matching_names_values` refuses a DOF matched by two patterns
    (``waist_*_joint`` from the robot options and ``waist_roll_joint`` from the overrides would
    collide). A pattern matching no DOF raises, so a typo cannot silently leave a joint at the
    robot's limit.
    """
    limits = resolve_dofs_values(robot_options.default_dofs_force_limits, dofs_name)
    if overrides:
        idx, _, values = resolve_matching_names_values(overrides, dofs_name)
        limits[idx] = values
    return limits

Franka Panda

Franka Panda real-robot deployment backend.

Attributes

PANDA_ARM_JOINT_IDX module-attribute

PANDA_ARM_JOINT_IDX: dict[str, int] = {'joint1': 0, 'joint2': 1, 'joint3': 2, 'joint4': 3, 'joint5': 4, 'joint6': 5, 'joint7': 6}

PANDA_FINGER_JOINTS module-attribute

PANDA_FINGER_JOINTS: tuple[str, ...] = ('finger_joint1', 'finger_joint2')

Classes

FrankaPandaDeployment

FrankaPandaDeployment(env: EnvBase, options: DeploymentOptions)

Bases: DeploymentBase

Deployment backend for Franka Emika Panda / FR3 via panda-py.

The 7-DOF arm is driven through panda-py's real-time :class:controllers.JointPosition controller running at 1 kHz inside libfranka. The Franka Hand gripper is driven through libfranka's blocking Gripper API, dispatched from a worker thread so the per-step :meth:send_payload call never blocks the control loop.

The arm is fixed-base with no IMU, so :meth:read_state returns an identity base_quat, zero base angular velocity and zero linear acceleration.

Requirements

• panda-python must be installed with a libfranka build that matches the controller's reported server version. See https://github.com/JeanElsner/panda-py/releases for per-firmware wheels.
• FCI must be activated in Desk and the joint brakes released.
• A PREEMPT_RT kernel is strongly recommended for stable 1 kHz control; without it expect occasional communication_constraints_violation.

PARAMETER	DESCRIPTION
robot_ip	IP address of the Franka Control box. Defaults to the factory IP.
arm_stiffness	Optional 7-element per-joint gains for the JointPosition controller. If None, panda-py's defaults are used.
arm_damping	Optional 7-element per-joint gains for the JointPosition controller. If None, panda-py's defaults are used.
arm_filter_coeff	Exponential low-pass coefficient applied to the JointPosition setpoint inside panda-py's 1 kHz control loop. 1.0 = no filtering (raw step inputs every set_control call); lower values smooth the trajectory. For policy/teleop at 30-100 Hz the controller sees abrupt setpoint changes between updates which cause the arm to buzz/shake under PD, so the default here is 0.05 (panda-py's value is 1.0). Drop further if you still see oscillation; raise toward 1.0 if you need maximum tracking responsiveness at the cost of smoothness.
velocity_safety_factor	Fraction of Panda's per-joint velocity limits that the deployment will allow per control step. send_payload clamps the commanded delta from current_q so the implied joint velocity is bounded by vel_limit * velocity_safety_factor. Direction is preserved by uniformly scaling the delta (rather than clipping per joint, which would distort EE motion). Without this clamp Differential IK near a high-Jacobian configuration can emit deltas that trigger libfranka's joint_velocity_violation reflex. Default 0.5 is conservative; raise toward 1.0 for more aggressive teleop on a PREEMPT_RT kernel.
use_grasp_on_close	When True, the gripper worker uses libfranka.Gripper.grasp (force-controlled) for closing commands and reserves Gripper.move for opening. grasp is more responsive on held objects (it returns as soon as firmware detects finger contact, instead of waiting out move's stall timeout) and applies continuous holding force, so objects don't slip during arm motion. The classification is by absolute target width: a command below close_width_threshold is treated as a close, anything at or above as an open. Disable to fall back to move-only behavior (e.g. if your task is purely position-mode finger control).
close_width_threshold	Target width (m) below which a command is treated as a close (uses grasp when use_grasp_on_close is set). Default 0.04 is roughly half the Franka Hand's max opening.
gripper_speed	Default gripper move speed (m/s).
gripper_force	Default grasp force (N) for the :meth:grasp helper.
gripper_epsilon_inner	Tolerance (m) for :meth:grasp success on the inner/outer side.
gripper_epsilon_outer	Tolerance (m) for :meth:grasp success on the inner/outer side.
skip_gripper_homing	If True, do not run gripper.homing() even when the encoder reports uncalibrated. Useful when an object is already held.
Like
command
until
velocity
the
it
since

Source code in eden/extensions/deployment/franka_panda.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    # _finger_dof_indices comes from _resolve_options, which super().__init__ ran.
    self._has_gripper: bool = len(self._finger_dof_indices) > 0
    self._gripper_max_width: float = _DEFAULT_MAX_FINGER_WIDTH

    self._panda: Any | None = None
    self._gripper: Any | None = None
    self._controller: Any | None = None
    self._controller_started: bool = False

    # Background gripper command dispatch.
    self._gripper_target_width: float | None = None
    self._gripper_command_lock = threading.Lock()
    self._gripper_thread: threading.Thread | None = None
    self._gripper_stop = threading.Event()

Attributes

sends_command_gains class-attribute

sends_command_gains: bool = False

robot_ip class-attribute instance-attribute

robot_ip: str = '192.168.1.1'

arm_stiffness class-attribute instance-attribute

arm_stiffness: tuple[float, ...] | None = None

arm_damping class-attribute instance-attribute

arm_damping: tuple[float, ...] | None = None

arm_filter_coeff class-attribute instance-attribute

arm_filter_coeff: float = 0.05

velocity_safety_factor class-attribute instance-attribute

velocity_safety_factor: float = 0.5

use_grasp_on_close class-attribute instance-attribute

use_grasp_on_close: bool = True

close_width_threshold class-attribute instance-attribute

close_width_threshold: float = 0.04

gripper_speed class-attribute instance-attribute

gripper_speed: float = 0.1

gripper_force class-attribute instance-attribute

gripper_force: float = 20.0

gripper_epsilon_inner class-attribute instance-attribute

gripper_epsilon_inner: float = 0.02

gripper_epsilon_outer class-attribute instance-attribute

gripper_epsilon_outer: float = 0.02

skip_gripper_homing class-attribute instance-attribute

skip_gripper_homing: bool = False

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/franka_panda.py

def connect(self) -> None:
    if panda_py is None:
        raise RuntimeError(
            "panda-python package is not installed. Install the libfranka-matched "
            "wheel from https://github.com/JeanElsner/panda-py/releases ."
        )

    self._panda = panda_py.Panda(self.robot_ip)
    # Clear any latched controller error from a prior session.
    self._panda.recover()

    if self._has_gripper:
        self._gripper = _libfranka.Gripper(self.robot_ip)
        state = self._gripper.read_once()
        # ``max_width`` reads as ~0 when the encoder hasn't been homed
        # this power cycle. Homing fully opens then closes the fingers
        # to calibrate, so it requires nothing be held.
        if not self.skip_gripper_homing and state.max_width <= 0.01:
            en.logger.info("[franka_panda] Homing gripper to calibrate encoder...")
            self._gripper.homing()
            state = self._gripper.read_once()
        if state.max_width > 0.0:
            self._gripper_max_width = float(state.max_width)

        self._gripper_stop.clear()
        self._gripper_thread = threading.Thread(target=self._gripper_worker, daemon=True)
        self._gripper_thread.start()

    kwargs: dict[str, Any] = {}
    if self.arm_stiffness is not None:
        kwargs["stiffness"] = np.asarray(self.arm_stiffness, dtype=np.float64)
    if self.arm_damping is not None:
        kwargs["damping"] = np.asarray(self.arm_damping, dtype=np.float64)
    # Setpoint low-pass (see ``arm_filter_coeff``). Passed by keyword so a panda-py
    # build that names it differently raises instead of silently ignoring it.
    kwargs["filter_coeff"] = float(self.arm_filter_coeff)
    self._controller = _panda_controllers.JointPosition(**kwargs)

close

close() -> None

Source code in eden/extensions/deployment/franka_panda.py

def close(self) -> None:
    self._teardown_common()
    if self._gripper_thread is not None:
        self._gripper_stop.set()
        self._gripper_thread.join(timeout=2.0)
    self._gripper_thread = None

    if self._panda is not None and self._controller_started:
        try:
            self._panda.stop_controller()
        except Exception as e:
            en.logger.warning(f"[franka_panda] stop_controller failed: {e}")
    self._controller_started = False

    if self._gripper is not None:
        try:
            self._gripper.stop()
        except Exception:
            pass

    self._controller = None
    self._gripper = None
    self._panda = None

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/franka_panda.py

def read_state(self) -> RobotState:
    if self._panda is None:
        raise RuntimeError("Franka Panda deployment is not connected.")

    s = self._panda.get_state()
    dofs_pos = np.zeros(self.num_dofs, dtype=np.float64)
    dofs_vel = np.zeros(self.num_dofs, dtype=np.float64)
    dofs_torque = np.zeros(self.num_dofs, dtype=np.float64)

    q = np.asarray(s.q, dtype=np.float64)
    dq = np.asarray(s.dq, dtype=np.float64)
    tau = np.asarray(s.tau_J, dtype=np.float64)
    for dof_i, joint_i in zip(self._arm_dof_indices, self._arm_joint_indices):
        dofs_pos[dof_i] = q[joint_i]
        dofs_vel[dof_i] = dq[joint_i]
        dofs_torque[dof_i] = tau[joint_i]

    extra: dict[str, Any] = {
        # O_T_EE is column-major (Fortran) in libfranka.
        "O_T_EE": np.asarray(s.O_T_EE, dtype=np.float64).reshape(4, 4, order="F"),
        "robot_mode": str(s.robot_mode),
    }

    if self._has_gripper and self._gripper is not None:
        g = self._gripper.read_once()
        width = float(g.width)
        # Symmetric split between the two finger joints (the Franka Hand
        # constraint ties them together at 2x the per-finger displacement).
        per_finger = 0.5 * width
        for dof_i in self._finger_dof_indices:
            dofs_pos[dof_i] = per_finger
        extra["gripper_width"] = width
        extra["gripper_is_grasped"] = bool(g.is_grasped)
        extra["gripper_max_width"] = float(g.max_width)

    return RobotState(
        stamp=time.time(),
        base_quat=np.array([1.0, 0.0, 0.0, 0.0]),  # identity (wxyz) - fixed-base arm
        base_ang_vel=np.zeros(3),
        base_lin_acc=np.zeros(3),
        dofs_pos=dofs_pos,
        dofs_vel=dofs_vel,
        dofs_torque=dofs_torque,
        extra=extra,
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/franka_panda.py

def send_payload(self, payload: RobotCommand) -> None:
    if self._panda is None or self._controller is None:
        raise RuntimeError("Franka Panda deployment is not connected.")
    if not self._controller_started:
        # The first payload is the boot STOP hold of the measured pose, so the controller
        # starts with the arm's own pose as its setpoint: no jerk.
        self._panda.start_controller(self._controller)
        self._controller_started = True

    # Low-pass the arm on every path (hold / INIT / policy); the payload carries what is sent. Not
    # the fingers: a filtered width ramps over many ticks, the gripper worker dispatches each step
    # of it as a new command, and every new command stops the one in flight -- a closing grasp
    # would be stopped and re-issued at the control rate (:meth:`_gripper_worker`).
    commanded = np.asarray(payload.dofs_pos, dtype=np.float64)
    sent = self._apply_target_lpf(commanded)
    sent[self._finger_dof_indices] = commanded[self._finger_dof_indices]
    payload.dofs_pos = sent
    target_q = self._extract_arm_target(payload.dofs_pos)
    target_q = self._clamp_arm_velocity(target_q)
    self._controller.set_control(target_q)

    # The hold, the INIT ramp and the START tick leave the gripper at the policy's last
    # command: a target from the measured width would stop a force-controlled grasp and
    # re-send it as a plain move, and INIT's default width would open it on what it holds.
    if self._has_gripper and not self.sending_hold:
        self._set_gripper_target(self._extract_target_width(payload.dofs_pos))

grasp

grasp(width: float, force: float | None = None) -> bool

Grasp (synchronously) an object of approximately width meters.

Returns True when the firmware reports the fingers clamped on something within the configured tolerance. Sets the background gripper worker's target to width first, so the worker does not countermand the grasp with a move of its own.

Source code in eden/extensions/deployment/franka_panda.py

def grasp(self, width: float, force: float | None = None) -> bool:
    """Grasp (synchronously) an object of approximately ``width`` meters.

    Returns ``True`` when the firmware reports the fingers clamped on
    something within the configured tolerance. Sets the background
    gripper worker's target to ``width`` first, so the worker does not
    countermand the grasp with a ``move`` of its own.
    """
    if self._gripper is None:
        raise RuntimeError("Gripper is not configured for this deployment.")
    with self._gripper_command_lock:
        self._gripper_target_width = float(width)
    return bool(
        self._gripper.grasp(
            width=float(width),
            speed=self.gripper_speed,
            force=float(force if force is not None else self.gripper_force),
            epsilon_inner=self.gripper_epsilon_inner,
            epsilon_outer=self.gripper_epsilon_outer,
        )
    )

recover

recover() -> None

Clear an error state on the Franka controller (e.g. after a reflex).

Source code in eden/extensions/deployment/franka_panda.py

def recover(self) -> None:
    """Clear an error state on the Franka controller (e.g. after a reflex)."""
    if self._panda is None:
        raise RuntimeError("Franka Panda deployment is not connected.")
    self._panda.recover()

Functions:

RobotEra XHand

RobotEra XHand real-robot deployment backend.

Attributes

XHAND_FINGER_IDX module-attribute

XHAND_FINGER_IDX: dict[str, int] = {'right_hand_thumb_bend_joint': 0, 'right_hand_thumb_rota_joint1': 1, 'right_hand_thumb_rota_joint2': 2, 'right_hand_index_bend_joint': 3, 'right_hand_index_joint1': 4, 'right_hand_index_joint2': 5, 'right_hand_mid_joint1': 6, 'right_hand_mid_joint2': 7, 'right_hand_ring_joint1': 8, 'right_hand_ring_joint2': 9, 'right_hand_pinky_joint1': 10, 'right_hand_pinky_joint2': 11, 'left_hand_thumb_bend_joint': 0, 'left_hand_thumb_rota_joint1': 1, 'left_hand_thumb_rota_joint2': 2, 'left_hand_index_bend_joint': 3, 'left_hand_index_joint1': 4, 'left_hand_index_joint2': 5, 'left_hand_mid_joint1': 6, 'left_hand_mid_joint2': 7, 'left_hand_ring_joint1': 8, 'left_hand_ring_joint2': 9, 'left_hand_pinky_joint1': 10, 'left_hand_pinky_joint2': 11}

Classes

RoboTeraXHandDeployment

RoboTeraXHandDeployment(env: EnvBase, options: DeploymentOptions)

Bases: DeploymentBase

Deployment backend for the RoboTera XHand1 dexterous hand.

Supports both EtherCAT and RS485 communication protocols. For RS485 you must also set serial_port and baud_rate to match your hardware.

The hand is treated as a fixed-base robot: read_state returns an identity quaternion and zero angular velocity because the hand has no IMU.

Joint velocity is a finite difference. The SDK's FingerState_t has a position and no velocity, so read_state differences consecutive positions over the wall-clock time between the two reads (monotonic clock). The first read after connect has no previous sample and reports zero. It is a control-period average, not the instantaneous value the simulator reports, so expect it to be noisier and a period late.

The hand runs its own PD, with the gains below -- not the command's. finger_command carries kp / ki / kd as int16 values in firmware units, which have no known mapping to the N·m/rad of :attr:RobotCommand.dofs_kp (the xhand1 sysid example fits the effective sim-side gains for that reason). So send_payload forwards positions only: RobotCommand.dofs_kp / dofs_kd / dofs_vel / dofs_torque are not sent, and the firmware gains are set once at connect from dofs_kp / dofs_ki / dofs_kd. A payload with kp = 0 therefore changes nothing on the hand. The two base options that work through command gains, deploy_kp_kd_scale and disable_joints, are refused at construction rather than accepted and ignored. Change these options instead. connect logs the gains it programs.

PARAMETER	DESCRIPTION
protocol	Bus to open. RS485 also needs serial_port and baud_rate. TYPE: ('RS485', 'EtherCAT') DEFAULT: "RS485"
serial_port	RS485 device path. TYPE: str
baud_rate	RS485 baud rate. TYPE: int
hand_id	Device ID of the hand to drive. 0 selects the first enumerated hand. TYPE: int
dofs_kp	Firmware PD(I) gains programmed on every finger, in firmware units. TYPE: int
dofs_ki	Firmware PD(I) gains programmed on every finger, in firmware units. TYPE: int
dofs_kd	Firmware PD(I) gains programmed on every finger, in firmware units. TYPE: int
tor_max	Firmware torque limit on every finger, in firmware units. TYPE: int
finger_mode	Firmware finger mode: 0 powerless, 3 position (default), 5 powerful. TYPE: int
crc_retries	Attempts per read_state / send_command before a CRC failure raises. TYPE: int

Source code in eden/extensions/deployment/robotera_xhand.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)

    self._device: Any | None = None
    self._hand_cmd: Any | None = None
    self._active_hand_id: int = self.hand_id
    # Previous position sample and its monotonic stamp, for the finite-difference velocity.
    self._prev_dofs_pos: np.ndarray | None = None
    self._prev_read_t: float | None = None

Attributes

sends_command_gains class-attribute

sends_command_gains: bool = False

protocol class-attribute instance-attribute

protocol: Literal['RS485', 'EtherCAT'] = 'RS485'

serial_port class-attribute instance-attribute

serial_port: str = '/dev/ttyUSB0'

baud_rate class-attribute instance-attribute

baud_rate: int = 3000000

hand_id class-attribute instance-attribute

hand_id: int = 0

dofs_kp class-attribute instance-attribute

dofs_kp: int = 50

dofs_ki class-attribute instance-attribute

dofs_ki: int = 0

dofs_kd class-attribute instance-attribute

dofs_kd: int = 25

tor_max class-attribute instance-attribute

tor_max: int = 300

finger_mode class-attribute instance-attribute

finger_mode: int = 3

crc_retries class-attribute instance-attribute

crc_retries: int = 3

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/robotera_xhand.py

def connect(self) -> None:
    if xhand_control is None:
        raise RuntimeError(
            "xhand_controller package is not installed. Install it before using RoboTeraXHandDeployment."
        )

    self._device = xhand_control.XHandControl()

    if self.protocol == "EtherCAT":
        ports = self._device.enumerate_devices("EtherCAT")
        if not ports:
            raise RuntimeError("No EtherCAT XHand devices found.")
        rsp = self._device.open_ethercat(ports[0])
    elif self.protocol == "RS485":
        rsp = self._device.open_serial(self.serial_port, self.baud_rate)
    else:
        raise ValueError(f"Unknown protocol '{self.protocol}'. Expected 'RS485' or 'EtherCAT'.")

    if rsp.error_code != 0:
        raise RuntimeError(f"Failed to open XHand device: {rsp.error_message}")

    # Wait until the device enumerates at least one hand, then verify that the
    # configured hand_id is present.  When hand_id is 0 (default) we auto-select
    # the first discovered hand for single-hand setups.
    start_time = time.time()
    while True:
        ids = self._device.list_hands_id()
        if ids:
            if self.hand_id in ids:
                self._active_hand_id = self.hand_id
                break
            if self.hand_id == 0:
                self._active_hand_id = ids[0]
                break
            raise RuntimeError(f"Configured hand_id {self.hand_id} not found. Available hand IDs: {ids}")
        if time.time() - start_time > self.connect_timeout_s:
            raise RuntimeError("Timed out waiting for XHand device to enumerate hand IDs.")
        time.sleep(0.05)

    self._hand_cmd = xhand_control.HandCommand_t()
    for i in range(_NUM_JOINTS):
        fc = self._hand_cmd.finger_command[i]
        fc.id = i
        fc.kp = self.dofs_kp
        fc.ki = self.dofs_ki
        fc.kd = self.dofs_kd
        fc.position = 0.0
        fc.tor_max = self.tor_max
        fc.mode = self.finger_mode

    # A fresh link has no previous sample: the first read reports zero velocity.
    self._prev_dofs_pos = None
    self._prev_read_t = None
    en.logger.info(
        f"[deploy] XHand firmware PD: kp={self.dofs_kp} ki={self.dofs_ki} kd={self.dofs_kd} "
        f"tor_max={self.tor_max} (firmware units, on every finger). RobotCommand gains are not "
        "forwarded; set the dofs_kp / dofs_ki / dofs_kd deployment options to change these."
    )

close

close() -> None

Source code in eden/extensions/deployment/robotera_xhand.py

def close(self) -> None:
    self._teardown_common()
    if self._device is not None:
        self._set_finger_mode(0)  # release to powerless before disconnect
    self._device = None
    self._hand_cmd = None

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/robotera_xhand.py

def read_state(self) -> RobotState:
    if self._device is None:
        raise RuntimeError("XHand deployment is not connected.")

    for attempt in range(self.crc_retries):
        err, state = self._device.read_state(self._active_hand_id, True)
        if err.error_code == 0:
            break
        en.logger.warning(f"read_state CRC error (attempt {attempt + 1}/{self.crc_retries}): {err.error_message}")
    else:
        raise RuntimeError(f"Failed to read XHand state after {self.crc_retries} attempts: {err.error_message}")
    read_t = time.perf_counter()

    dofs_pos = np.zeros(self.num_dofs, dtype=np.float64)
    dofs_torque = np.zeros(self.num_dofs, dtype=np.float64)

    joint_diagnostics: list[dict] = []
    for i, finger_idx in enumerate(self._dof_index):
        finger = state.finger_state[finger_idx]
        dofs_pos[i] = finger.position
        dofs_torque[i] = finger.torque
        joint_diagnostics.append(
            {
                "id": finger.id,
                "raw_position": finger.raw_position,
                "sensor_id": finger.sensor_id,
                "temperature": finger.temperature,
                "commboard_err": finger.commboard_err,
                "jonitboard_err": finger.jonitboard_err,
                "tipboard_err": finger.tipboard_err,
            }
        )

    # Collect fingertip tactile data for the five sensor-equipped fingers.
    fingertip_sensors: dict[int, dict] = {}
    for sensor_idx, finger_id in enumerate(_SENSOR_FINGERS):
        sensor_data = state.sensor_data[sensor_idx]
        fingertip_sensors[finger_id] = {
            "calc_pressure": np.array(
                [
                    sensor_data.calc_force.fx,
                    sensor_data.calc_force.fy,
                    sensor_data.calc_force.fz,
                ]
            ),
            "raw_pressure": np.array([[force.fx, force.fy, force.fz] for force in sensor_data.raw_force]),
            "sensor_temperature": sensor_data.calc_temperature,
        }

    dofs_vel = self._finite_difference_dofs_vel(dofs_pos, read_t)

    return RobotState(
        stamp=time.time(),
        # The XHand1 is a fixed-base end-effector with no on-board IMU.
        base_quat=np.array([1.0, 0.0, 0.0, 0.0]),  # identity (wxyz)
        base_ang_vel=np.zeros(3),
        base_lin_acc=np.zeros(3),
        dofs_pos=dofs_pos,
        dofs_vel=dofs_vel,
        dofs_torque=dofs_torque,
        extra={
            "fingertip_sensors": fingertip_sensors,
            "joint_diagnostics": joint_diagnostics,
        },
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/robotera_xhand.py

def send_payload(self, payload: RobotCommand) -> None:
    if self._device is None or self._hand_cmd is None:
        raise RuntimeError("XHand deployment is not connected.")

    # Low-pass every path (hold / INIT / policy); the payload carries what is sent.
    payload.dofs_pos = self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64))
    for i, finger_idx in enumerate(self._dof_index):
        fc = self._hand_cmd.finger_command[finger_idx]
        fc.id = finger_idx
        fc.position = float(payload.dofs_pos[i])

    for attempt in range(self.crc_retries):
        err = self._device.send_command(self._active_hand_id, self._hand_cmd)
        if err.error_code == 0:
            break
        en.logger.warning(f"send_command CRC error (attempt {attempt + 1}/{self.crc_retries}): {err.error_message}")
    else:
        raise RuntimeError(f"Failed to send XHand command after {self.crc_retries} attempts: {err.error_message}")

reset_sensor

reset_sensor(sensor_id: int) -> None

Reset a fingertip sensor by its ID.

Source code in eden/extensions/deployment/robotera_xhand.py

def reset_sensor(self, sensor_id: int) -> None:
    """Reset a fingertip sensor by its ID."""
    if self._device is None:
        raise RuntimeError("XHand deployment is not connected.")
    err = self._device.reset_sensor(self._active_hand_id, sensor_id)
    if err.error_code != 0:
        raise RuntimeError(f"Failed to reset sensor {sensor_id}: {err.error_message}")

set_hand_id

set_hand_id(new_id: int) -> None

Change the hand ID on the device.

Source code in eden/extensions/deployment/robotera_xhand.py

def set_hand_id(self, new_id: int) -> None:
    """Change the hand ID on the device."""
    if self._device is None:
        raise RuntimeError("XHand deployment is not connected.")
    err = self._device.set_hand_id(self._active_hand_id, new_id)
    if err.error_code != 0:
        raise RuntimeError(f"Failed to set hand ID from {self._active_hand_id} to {new_id}: {err.error_message}")
    self._active_hand_id = new_id

Functions:

Wuji Hand

Wuji Hand 1 / Hand 2 real-robot deployment backend.

Both generations are 20-DOF hands (5 fingers x 4 joints) whose SDKs speak joint angles in radians, finger-major (thumb first), in the MJCF joint directions -- no sign or offset conversion. The SDK order is the same for a right and a left hand; the side is a property of the joint names, which :data:WUJI_JOINT_IDX maps for both sides and both naming schemes in Eden:

• {right,left}_finger{1..5}_joint{1..4} (:class:~eden.options.robots.wuji_hand1.WujiHand1_R and the wuji-mjlab task models), and
• {r,l}_{thumb_cmc_flex, ..., pinky_dip} (:class:~eden.options.robots.wuji_hand2beta2.WujiHand2Beta2_R), whose declaration order is the same finger-major order (upstream WUJI_HAND2_TASK_NAMES).

A :class:HandTransport is the wire to one hand: :class:Hand1Transport (wujihandpy) or :class:Hand2Transport (wuji_sdk). It streams targets the way upstream wuji-mjlab does:

• "sync" (upstream reorient): :meth:HandTransport.command blocks for one control period, writing servo_hz linearly interpolated sub-targets from the previous command to the new one.
• "thread" (upstream pen spin): :meth:HandTransport.command starts a new ramp and returns; a background thread streams it at servo_hz and holds the target once the ramp ends.

:class:WujiHandDeployment drives one hand through a transport it builds at :meth:connect, or one handed to it with :meth:WujiHandDeployment.use_transport (a simulator, for sim-to-sim).

Attributes

NUM_JOINTS module-attribute

NUM_JOINTS = 20

EASE_PROFILES module-attribute

EASE_PROFILES = ('cosine', 'smoothstep')

Classes

DeployStop

Bases: RuntimeError

A safety guard ended the run (hardware fault, stale feedback, lost tracking); the message says which.

StreamOptions dataclass

StreamOptions(mode: Literal['sync', 'thread'] = 'sync', servo_hz: float = 200.0, interp_s: float = 0.02)

How targets reach the hand.

PARAMETER	DESCRIPTION
mode	See the module docstring. TYPE: ('sync', 'thread') DEFAULT: "sync"
servo_hz	Rate of the interpolated sub-targets. TYPE: float DEFAULT: 200.0
interp_s	Duration of one ramp: the control period. TYPE: float DEFAULT: 0.02

Attributes

mode class-attribute instance-attribute

mode: Literal['sync', 'thread'] = 'sync'

servo_hz class-attribute instance-attribute

servo_hz: float = 200.0

interp_s class-attribute instance-attribute

interp_s: float = 0.02

Methods:

HandTransport

HandTransport(stream: StreamOptions, joint_names: Sequence[str] | None = None)

The contract every Wuji transport keeps (upstream HandDriverBase): 20 joint angles, SDK order.

PARAMETER	DESCRIPTION
stream	How :meth:command streams a target. TYPE: StreamOptions
joint_names	The 20 joints' names in SDK order, for error messages (:attr:WujiHandDeployment.sdk_joint_names); None names them by SDK index. TYPE: Sequence[str] | None DEFAULT: None

Source code in eden/extensions/deployment/wuji_hand.py

def __init__(self, stream: StreamOptions, joint_names: Sequence[str] | None = None) -> None:
    self.stream = stream
    self.joint_names = _SDK_JOINTS if joint_names is None else tuple(joint_names)
    if len(self.joint_names) != NUM_JOINTS:
        raise ValueError(f"joint_names needs the {NUM_JOINTS} joints in SDK order, got {list(self.joint_names)}")
    self._cmd: np.ndarray | None = None
    self._deadline: float | None = None
    self._lock = threading.Lock()
    self._seg_start: np.ndarray | None = None
    self._seg_target: np.ndarray | None = None
    self._seg_t0 = 0.0
    self._servo_thread: threading.Thread | None = None
    self._servo_stop = threading.Event()
    self._servo_error: BaseException | None = None

Attributes

stream instance-attribute

stream = stream

joint_names instance-attribute

joint_names = _SDK_JOINTS if joint_names is None else tuple(joint_names)

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def connect(self) -> None:
    raise NotImplementedError

write

write(qpos: ndarray) -> None

Send one joint-position target to the hand now.

Source code in eden/extensions/deployment/wuji_hand.py

def write(self, qpos: np.ndarray) -> None:
    """Send one joint-position target to the hand now."""
    raise NotImplementedError

read

read() -> ndarray

The latest measured joint positions.

Source code in eden/extensions/deployment/wuji_hand.py

def read(self) -> np.ndarray:
    """The latest measured joint positions."""
    raise NotImplementedError

fault

fault() -> str | None

A latched hardware fault, described, or None.

Source code in eden/extensions/deployment/wuji_hand.py

def fault(self) -> str | None:
    """A latched hardware fault, described, or ``None``."""
    return None

stale

stale() -> str | None

Why the joint feedback can no longer be trusted (a lost link), or None.

Source code in eden/extensions/deployment/wuji_hand.py

def stale(self) -> str | None:
    """Why the joint feedback can no longer be trusted (a lost link), or ``None``."""
    return None

clock

clock() -> float

The time joint readings are stamped with (wall clock; a simulation's own time in sim-to-sim).

Source code in eden/extensions/deployment/wuji_hand.py

def clock(self) -> float:
    """The time joint readings are stamped with (wall clock; a simulation's own time in sim-to-sim)."""
    return time.perf_counter()

temp_summary

temp_summary() -> str | None

Source code in eden/extensions/deployment/wuji_hand.py

def temp_summary(self) -> str | None:
    return None

disconnect

disconnect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def disconnect(self) -> None:
    raise NotImplementedError

command

command(target: ndarray) -> None

Stream target from the previous command, as upstream does for this task.

Source code in eden/extensions/deployment/wuji_hand.py

def command(self, target: np.ndarray) -> None:
    """Stream ``target`` from the previous command, as upstream does for this task."""
    target = _checked(target, self.joint_names)
    if self._cmd is None:
        self._cmd = self.read().copy()
    if self.stream.mode == "sync":
        self._ramp(self._cmd, target, self.stream.interp_s)
        self._cmd = target.copy()
        return
    self._ensure_servo()
    self._set_segment(None, target)

hold

hold() -> None

Let one control period pass with no new target, holding the last one.

"sync" re-streams the last command across the period through :meth:_ramp, so a tick that sends nothing still lasts one control period (on a simulator's clock too); before any command there is nothing to hold, and it only waits. "thread" returns at once: the servo thread is already holding, and the base rate limiter paces the loop.

Source code in eden/extensions/deployment/wuji_hand.py

def hold(self) -> None:
    """Let one control period pass with no new target, holding the last one.

    ``"sync"`` re-streams the last command across the period through :meth:`_ramp`, so a tick
    that sends nothing still lasts one control period (on a simulator's clock too); before any
    command there is nothing to hold, and it only waits. ``"thread"`` returns at once: the servo
    thread is already holding, and the base rate limiter paces the loop.
    """
    if self.stream.mode != "sync":
        self._raise_if_servo_failed()
        return
    if self._cmd is None:
        time.sleep(self.stream.interp_s)
        return
    self._ramp(self._cmd, self._cmd, self.stream.interp_s)

ease_to

ease_to(goal: ndarray, duration_s: float, rate_hz: float, profile: str, guard: Callable[[], None] | None = None) -> None

Move from the measured pose to goal: upstream's home (cosine, 100 Hz) or start-pose smoothstep.

guard runs before every waypoint, so a safety check that raises stops a multi-second move where it is, holding the last waypoint, instead of only before it starts.

Source code in eden/extensions/deployment/wuji_hand.py

def ease_to(
    self,
    goal: np.ndarray,
    duration_s: float,
    rate_hz: float,
    profile: str,
    guard: Callable[[], None] | None = None,
) -> None:
    """Move from the measured pose to ``goal``: upstream's ``home`` (cosine, 100 Hz) or start-pose smoothstep.

    ``guard`` runs before every waypoint, so a safety check that raises stops a multi-second
    move where it is, holding the last waypoint, instead of only before it starts.
    """
    if not (duration_s >= 0.0 and rate_hz > 0.0):
        raise ValueError(f"ease_to needs duration_s >= 0 and rate_hz > 0; got {duration_s!r}, {rate_hz!r}.")
    goal = _checked(goal, self.joint_names)
    start = self.read().copy()
    n = max(1, round(duration_s * rate_hz))
    path = [start + ease(i / n, profile) * (goal - start) for i in range(1, n + 1)]
    if self.stream.mode == "sync":
        for prev, q in zip([start, *path], path):
            if guard is not None:
                guard()
            self._ramp(prev, q, 1.0 / rate_hz)
            self._cmd = q.copy()
        return
    if self._servo_thread is None:
        self._cmd = start
    for q in path:
        if guard is not None:
            guard()
        self._ensure_servo()
        self._set_segment(q, q)
        time.sleep(1.0 / rate_hz)

stop_servo

stop_servo() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def stop_servo(self) -> None:
    self._servo_stop.set()
    if self._servo_thread is not None:
        self._servo_thread.join(timeout=1.0)
        if self._servo_thread.is_alive():
            en.logger.error("[wuji] servo thread did not stop within 1 s; disconnecting anyway")
    self._servo_thread = None

close

close() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def close(self) -> None:
    self.stop_servo()
    # A reconnect streams from the pose it then measures, not from the previous connection's last target.
    self._cmd = self._deadline = self._servo_error = None
    self.disconnect()

Hand1Transport

Hand1Transport(stream: StreamOptions, effort_limit_a: float = 0.5, lowpass_cutoff_hz: float = 3.0, joint_names: Sequence[str] | None = None, stale_after_s: float = 0.5)

Bases: HandTransport

Wuji Hand 1 via wujihandpy's realtime position controller (upstream WujiHand1Driver).

The hand runs its own position loop behind a low-pass; command gains do not apply. Every :meth:read is a checked bus round-trip (read_joint_actual_position), so the feedback cannot freeze silently the way a subscribed feed can: a read that gets no answer within stale_after_s raises :class:DeployStop, and :meth:stale reports it from then on.

PARAMETER	DESCRIPTION
stream	How targets are streamed. TYPE: StreamOptions
effort_limit_a	Joint current limit, A. TYPE: float DEFAULT: 0.5
lowpass_cutoff_hz	Cutoff of the controller's target low-pass. TYPE: float DEFAULT: 3.0
joint_names	See :class:HandTransport. TYPE: Sequence[str] | None DEFAULT: None
stale_after_s	Timeout of each joint-position read. TYPE: float DEFAULT: 0.5

Source code in eden/extensions/deployment/wuji_hand.py

def __init__(
    self,
    stream: StreamOptions,
    effort_limit_a: float = 0.5,
    lowpass_cutoff_hz: float = 3.0,
    joint_names: Sequence[str] | None = None,
    stale_after_s: float = 0.5,
) -> None:
    super().__init__(stream, joint_names)
    self.effort_limit_a = effort_limit_a
    self.lowpass_cutoff_hz = lowpass_cutoff_hz
    self.stale_after_s = stale_after_s
    self._hand = None
    self._ctrl = None
    self._ctrl_cm = None
    self._stale: str | None = None

Attributes

effort_limit_a instance-attribute

effort_limit_a = effort_limit_a

lowpass_cutoff_hz instance-attribute

lowpass_cutoff_hz = lowpass_cutoff_hz

stale_after_s instance-attribute

stale_after_s = stale_after_s

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def connect(self) -> None:
    self._stale = None
    try:
        import wujihandpy
    except ImportError as e:
        raise ImportError(
            "Wuji Hand 1 needs wujihandpy (upstream pins 1.5.1): pip install wujihandpy==1.5.1"
        ) from e
    self._hand = wujihandpy.Hand()
    self._hand.write_joint_effort_limit(float(self.effort_limit_a))
    # Energized from here on: any failure must de-energize before propagating.
    try:
        self._hand.write_joint_enabled(True)
        self._ctrl_cm = self._hand.realtime_controller(
            enable_upstream=True, filter=wujihandpy.filter.LowPass(cutoff_freq=float(self.lowpass_cutoff_hz))
        )
        self._ctrl = self._ctrl_cm.__enter__()
    except BaseException:
        self.disconnect()
        raise
    en.logger.info(
        f"[wuji] Hand 1 enabled: current limit {self.effort_limit_a:g} A, target low-pass "
        f"{self.lowpass_cutoff_hz:g} Hz, {self.stream.mode} streaming at {self.stream.servo_hz:g} Hz"
    )

write

write(qpos: ndarray) -> None

Source code in eden/extensions/deployment/wuji_hand.py

def write(self, qpos: np.ndarray) -> None:
    self._ctrl.set_joint_target_position(_checked(qpos, self.joint_names).reshape(5, 4))

read

read() -> ndarray

Source code in eden/extensions/deployment/wuji_hand.py

def read(self) -> np.ndarray:
    try:
        q = self._hand.read_joint_actual_position(timeout=self.stale_after_s)
    except Exception as e:
        self._stale = (
            f"read_joint_actual_position(timeout={self.stale_after_s:g}) failed with "
            f"{type(e).__name__}: {e} (USB link lost?)"
        )
        raise DeployStop(f"hand feedback lost: {self._stale}") from e
    return np.asarray(q, dtype=np.float64).flatten()

stale

stale() -> str | None

Source code in eden/extensions/deployment/wuji_hand.py

def stale(self) -> str | None:
    return self._stale

disconnect

disconnect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def disconnect(self) -> None:
    try:
        if self._ctrl_cm is not None:
            self._ctrl_cm.__exit__(None, None, None)
    except BaseException as e:
        en.logger.warning(f"[wuji] Hand 1 realtime_controller close failed: {e}")
    finally:
        self._ctrl = self._ctrl_cm = None
    if self._hand is not None:
        try:
            self._hand.write_joint_enabled(False)
        except BaseException as e:
            en.logger.warning(f"[wuji] Hand 1 disable failed: {e}")

Hand2Gains dataclass

Hand2Gains(kp: list[float], kd: list[float], effort_limit_a: list[float])

MIT impedance and current limit programmed on a Wuji Hand 2 at connect.

PARAMETER	DESCRIPTION
kp	Per-joint stiffness and damping, SDK (finger-major) order. TYPE: list[float]
kd	Per-joint stiffness and damping, SDK (finger-major) order. TYPE: list[float]
effort_limit_a	Per-joint current limit, A. TYPE: list[float]

Attributes

kp instance-attribute

kp: list[float]

kd instance-attribute

kd: list[float]

effort_limit_a instance-attribute

effort_limit_a: list[float]

Methods:

Hand2Transport

Hand2Transport(stream: StreamOptions, gains: Hand2Gains, serial_number: str = '', require_all_joints: bool = True, feedback_timeout_s: float = 2.0, stale_after_s: float = 0.5, joint_names: Sequence[str] | None = None)

Bases: HandTransport

Wuji Hand 2 via wuji_sdk (upstream WujiHand2Driver and pen spin's Hand2Compat).

Connects to serial_number (or the single hand on the bus), programs the MIT gains and the current limit, enables the motors, and streams JointCommand(q, 0, 0). Joint positions arrive on the SDK's joint_states callback; joint_diagnostics feeds :meth:fault, and the age of the joint feed feeds :meth:stale.

PARAMETER	DESCRIPTION
stream	How targets are streamed. TYPE: StreamOptions
gains	Programmed once at connect. TYPE: Hand2Gains
serial_number	The hand; empty connects the only hand on the bus. TYPE: str DEFAULT: ''
require_all_joints	Refuse a hand with fewer than 20 joints online (upstream reorient). Otherwise (pen spin) the joints that report during :meth:connect are the ones :meth:stale watches. TYPE: bool DEFAULT: True
feedback_timeout_s	How long :meth:connect waits for the first fresh reading of every watched joint. TYPE: float DEFAULT: 2.0
stale_after_s	A watched joint (or the whole feed) older than this is stale. TYPE: float DEFAULT: 0.5
joint_names	See :class:HandTransport. TYPE: Sequence[str] | None DEFAULT: None

Source code in eden/extensions/deployment/wuji_hand.py

def __init__(
    self,
    stream: StreamOptions,
    gains: Hand2Gains,
    serial_number: str = "",
    require_all_joints: bool = True,
    feedback_timeout_s: float = 2.0,
    stale_after_s: float = 0.5,
    joint_names: Sequence[str] | None = None,
) -> None:
    super().__init__(stream, joint_names)
    self.gains = gains
    self.serial_number = serial_number
    self.require_all_joints = require_all_joints
    self.feedback_timeout_s = feedback_timeout_s
    self.stale_after_s = stale_after_s
    self.device_name = f"eden_wuji_hand_2_{next(_device_seq)}"
    self._sdk = self._mgr = self._hand = self._pub = self._sub = self._diag_sub = None
    self._state_lock = threading.Lock()
    self._latest = np.zeros(NUM_JOINTS)
    self._joint_t = np.full(NUM_JOINTS, -np.inf)
    self._watched = np.ones(NUM_JOINTS, dtype=bool)
    self._last_state_t = 0.0
    # Every (nid, code) ever reported, in arrival order, never replaced: a joint's later warning must
    # not un-latch the error it reported first. (Insertion-ordered dict: a set would lose the order.)
    self._errors: dict[tuple[int, int], None] = {}
    self._temps: dict[int, float] = {}
    self._dismissed: set[tuple[int, int]] = set()

Attributes

gains instance-attribute

gains = gains

serial_number instance-attribute

serial_number = serial_number

require_all_joints instance-attribute

require_all_joints = require_all_joints

feedback_timeout_s instance-attribute

feedback_timeout_s = feedback_timeout_s

stale_after_s instance-attribute

stale_after_s = stale_after_s

device_name instance-attribute

device_name = f'eden_wuji_hand_2_{next(_device_seq)}'

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def connect(self) -> None:
    from eden.utils.devices.wuji.glove import load_sdk

    self._sdk = wuji_sdk = load_sdk()
    self._mgr = wuji_sdk.SdkManager.instance()
    sn = self.serial_number or self._only_hand_on_the_bus()
    self._hand = self._mgr.connect(
        sn=sn, device_name=self.device_name, options=wuji_sdk.ConnectOptions(enable_bridge=False)
    )
    en.logger.info(f"[wuji] connected Wuji Hand 2 sn={self._hand.serial_number}")
    # Connected from here on: any failure must disable and release the link before propagating.
    try:
        n_online = self._hand.online_joints_count().get()
        if (self.require_all_joints and n_online != NUM_JOINTS) or n_online == 0:
            raise RuntimeError(
                f"Wuji Hand 2 {sn}: {n_online}/{NUM_JOINTS} joints online; refusing to move an incomplete "
                "hand (a missing joint reads back as its default pose). To run with the joints that are "
                "online, WujiHandDeployment.configure(require_all_joints=False)."
            )
        g = self.gains
        self._hand.effort_limit().set([float(a) for a in g.effort_limit_a])
        self._hand.mit_params().set(
            [wuji_sdk.MitParam(float(p), float(d)) for p, d in zip(g.kp, g.kd, strict=True)]
        )
        self._sub = self._hand.joint_states().subscribe_with_callback(self._on_state)
        try:
            self._diag_sub = self._hand.joint_diagnostics().subscribe_with_callback(self._on_diag)
        except Exception as e:
            en.logger.warning(f"[wuji] joint_diagnostics unavailable ({e}); faults limited to stale feedback")
        self._hand.enable()
        time.sleep(_HAND2_ARM_DELAY_S)
        self._pub = self._hand.joint_command().publish()
        self._await_feedback(n_online)
    except BaseException:
        self.disconnect()
        raise
    partial = "" if self._watched.all() else f", {int(self._watched.sum())}/{NUM_JOINTS} joints online"
    en.logger.info(
        f"[wuji] Hand 2 enabled: kp {min(g.kp):.3g}-{max(g.kp):.3g}, kd {min(g.kd):.3g}-{max(g.kd):.3g}, "
        f"current limit {min(g.effort_limit_a):g}-{max(g.effort_limit_a):g} A, "
        f"{self.stream.mode} streaming at {self.stream.servo_hz:g} Hz{partial}"
    )

fault

fault() -> str | None

Source code in eden/extensions/deployment/wuji_hand.py

def fault(self) -> str | None:
    for nid, code in list(self._errors):
        if (nid, code) in self._dismissed:
            continue
        i = _nid_index(nid)
        joint = f"joint {self.joint_names[i]!r} (nid={nid})" if i is not None else f"node nid={nid}"
        detail, severity = f"0x{code:04X}", ""
        try:
            d = self._hand.describe_error(code)
            if isinstance(d, dict):
                severity = str(d.get("severity", ""))
                detail = f"0x{code:04X} {d.get('name')} [{severity}] {d.get('desc')} ({d.get('resolution')})"
        except Exception:
            pass  # the bare code is still reported below
        if "warn" in severity.lower():
            en.logger.warning(f"[wuji] {joint} warning {detail}")
            self._dismissed.add((nid, code))
            continue
        return f"{joint} error {detail}"
    return None

stale

stale() -> str | None

Source code in eden/extensions/deployment/wuji_hand.py

def stale(self) -> str | None:
    now = time.monotonic()
    with self._state_lock:
        ages, last = now - self._joint_t, self._last_state_t
    if (feed_age := now - last) > self.stale_after_s:
        return f"joint_states feed stale for {feed_age:.1f} s (bus or link lost?)"
    if missing := [self.joint_names[i] for i in np.flatnonzero(self._watched & (ages > self.stale_after_s))]:
        return f"joint feedback stale for {missing}"
    return None

temp_summary

temp_summary() -> str | None

Source code in eden/extensions/deployment/wuji_hand.py

def temp_summary(self) -> str | None:
    if not self._temps:
        return None
    t = np.array(list(self._temps.values()))
    return f"{t.min():.1f}-{t.max():.1f} C (mean {t.mean():.1f})"

write

write(qpos: ndarray) -> None

Source code in eden/extensions/deployment/wuji_hand.py

def write(self, qpos: np.ndarray) -> None:
    jc = self._sdk.JointCommand
    self._pub.send([jc(float(x), 0.0, 0.0) for x in _checked(qpos, self.joint_names)])

read

read() -> ndarray

Source code in eden/extensions/deployment/wuji_hand.py

def read(self) -> np.ndarray:
    with self._state_lock:
        return self._latest.copy()

disconnect

disconnect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def disconnect(self) -> None:
    if self._hand is None:
        return
    try:
        self._hand.disable()
    except Exception as e:
        en.logger.warning(f"[wuji] Hand 2 disable failed: {e}")
    for sub in (self._diag_sub, self._sub):
        try:
            if sub is not None:
                sub.close()
        except Exception as e:
            en.logger.warning(f"[wuji] Hand 2 subscription close failed: {e}")
    # This connection only: ``disconnect_all`` would also drop a glove or the other hand.
    try:
        self._mgr.disconnect(self.device_name)
    except Exception as e:
        en.logger.warning(f"[wuji] Hand 2 disconnect failed: {e}")
    self._hand = self._pub = self._sub = self._diag_sub = None

WujiHandDeployment

WujiHandDeployment(env: EnvBase, options: DeploymentOptions)

Bases: DeploymentBase

A Wuji Hand 1 or Hand 2, right or left, on a fixed mount.

The side comes from the robot's joint names (see :data:WUJI_JOINT_IDX); the deployment DOF order may be any permutation of the 20 joints, and is scattered into the SDK's order on the way out and gathered back on the way in. The hand has no IMU: :meth:read_state reports the mount orientation (the robot's default_root_quat) as the base's.

Joint velocity is a finite difference of consecutive reads over the transport's clock, like the XHand's (neither SDK reports one), optionally low-passed (vel_lowpass_hz).

Gains are programmed once at connect, never per command. Hand 2 takes MIT kp / kd (mit_kp / mit_kd, default the payload's nominal gains: the robot's default_dofs_kp / default_dofs_kd times deploy_kp_kd_scale). Hand 1 runs its own position loop and has no gain to set. So disable_joints (which works by zeroing command gains) is refused for both, and deploy_kp_kd_scale / mit_kp / mit_kd for Hand 1.

Loop order. The base control tick (:meth:DeploymentBase.step, the operator state machine every backend runs) with this backend's hooks: :meth:check_safety first, then the command is sent and the period waited out, then the env's episode clock advances (episode_length_buf: the scene is never stepped here, and action warm-ups and clip clocks read it), then the hand is read and :meth:sync_scene -- the hook a subclass uses to write anything else it observes (a camera-tracked object) into the scene -- runs, and only then do the commands publish (sync_before_commands), so a command that grades the new state sees it this tick. In stream_mode="sync" sending the command is the wait (it streams sub-targets across one control period, as upstream does), so the base's rate limiter is not used on top of it: a tick takes one control period, and the inference time is absorbed into it (each period's deadline is chained from the previous one, as upstream anchors its loop before inference). A tick with nothing to send (action=None before a slot has written targets) holds the last target for the period instead (:meth:HandTransport.hold), so it still lasts one period and the episode clock and the reference cannot race ahead at the speed of the loop.

Boot, INIT and START. :meth:reset restarts the episode clock and the velocity estimate and boots in STOP, holding the measured pose. The operator's INIT ramps from the pose measured at the press to init_target over init_steps ticks and holds it; START hands the hand to the policy. :meth:check_safety runs before every one of those ticks, and :meth:ease_to runs :meth:check_hand before every waypoint.

PARAMETER	DESCRIPTION
hand	Hardware generation. None infers it from the robot options (WujiHand1* -> "hand1", WujiHand2Beta2* -> "hand2"); any other robot must set it. TYPE: ('hand1', 'hand2') DEFAULT: "hand1"
serial_number	Hand 2 to connect; empty connects the only hand on the bus. TYPE: str
stream_mode	How targets are streamed (see the module docstring). TYPE: ('sync', 'thread') DEFAULT: "sync"
servo_hz	Rate of the interpolated sub-targets. TYPE: float
effort_limit_a	Joint current limit, A. None: upstream's defaults, 0.5 (Hand 1) / 1.0 (Hand 2). TYPE: float | None
lowpass_cutoff_hz	Hand 1's target low-pass cutoff. TYPE: float
mit_kp	Hand 2's MIT gains: one value for every joint, or a pattern-keyed dict covering all 20. None: the payload's nominal gains. TYPE: float | dict[str, float] | None
mit_kd	Hand 2's MIT gains: one value for every joint, or a pattern-keyed dict covering all 20. None: the payload's nominal gains. TYPE: float | dict[str, float] | None
require_all_joints	Hand 2: refuse a hand with fewer than 20 joints online. TYPE: bool
stop_on_fault	:meth:check_hand raises :class:DeployStop on a latched hand fault. Stale joint feedback stops the run regardless: a policy must never run on a frozen reading. TYPE: bool
stale_after_s	Hand 2: joint feedback older than this is stale. connect_timeout_s bounds the wait for the first fresh reading at connect. Hand 1: the timeout of each joint-position read. TYPE: float
vel_lowpass_hz	First-order low-pass on the finite-difference joint velocity; None uses it raw. TYPE: float | None
clip_targets_to_joint_limits	Clamp every sent target (and every :meth:ease_to goal) to the built robot's joint range. TYPE: bool
auto_yaw_align	Off: the hand has no IMU, so its reported base orientation is the fixed mount's and there is no yaw to align. TYPE: bool

Source code in eden/extensions/deployment/wuji_hand.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    self._transport: HandTransport | None = None
    self._connected = False
    super().__init__(env, options)
    self.reset_velocity_filter()

Attributes

sends_command_gains class-attribute

sends_command_gains: bool = False

sync_before_commands class-attribute

sync_before_commands: bool = True

advances_episode_clock class-attribute

advances_episode_clock: bool = True

hand class-attribute instance-attribute

hand: Literal['hand1', 'hand2'] | None = None

serial_number class-attribute instance-attribute

serial_number: str = ''

stream_mode class-attribute instance-attribute

stream_mode: Literal['sync', 'thread'] = 'sync'

servo_hz class-attribute instance-attribute

servo_hz: float = 200.0

effort_limit_a class-attribute instance-attribute

effort_limit_a: float | None = None

lowpass_cutoff_hz class-attribute instance-attribute

lowpass_cutoff_hz: float = 3.0

mit_kp class-attribute instance-attribute

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

mit_kd class-attribute instance-attribute

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

require_all_joints class-attribute instance-attribute

require_all_joints: bool = True

stop_on_fault class-attribute instance-attribute

stop_on_fault: bool = True

stale_after_s class-attribute instance-attribute

stale_after_s: float = 0.5

vel_lowpass_hz class-attribute instance-attribute

vel_lowpass_hz: float | None = None

clip_targets_to_joint_limits class-attribute instance-attribute

clip_targets_to_joint_limits: bool = False

auto_yaw_align class-attribute instance-attribute

auto_yaw_align: bool = False

transport property

transport: HandTransport

blocks_for_the_period property

blocks_for_the_period: bool

Methods:

programmed_gains

programmed_gains() -> tuple[ndarray, ndarray] | None

Hand 2's MIT gains (mit_kp / mit_kd, else the nominal ones), in a policy's units; Hand 1 has none.

Source code in eden/extensions/deployment/wuji_hand.py

def programmed_gains(self) -> tuple[np.ndarray, np.ndarray] | None:
    """Hand 2's MIT gains (``mit_kp`` / ``mit_kd``, else the nominal ones), in a policy's units; Hand 1 has none."""
    if self.hand != "hand2":
        return None
    return self._mit_kp, self._mit_kd

to_sdk

to_sdk(dofs_values: ndarray) -> ndarray

Scatter a deployment-order vector into the SDK's finger-major order.

Source code in eden/extensions/deployment/wuji_hand.py

def to_sdk(self, dofs_values: np.ndarray) -> np.ndarray:
    """Scatter a deployment-order vector into the SDK's finger-major order."""
    out = np.empty(NUM_JOINTS)
    out[self._dof_index] = np.asarray(dofs_values, dtype=np.float64)
    return out

from_sdk

from_sdk(sdk_values: ndarray) -> ndarray

Gather an SDK-order vector into the deployment DOF order.

Source code in eden/extensions/deployment/wuji_hand.py

def from_sdk(self, sdk_values: np.ndarray) -> np.ndarray:
    """Gather an SDK-order vector into the deployment DOF order."""
    return np.asarray(sdk_values, dtype=np.float64)[self._dof_index]

use_transport

use_transport(transport: HandTransport) -> None

Drive transport (e.g. a simulator for sim-to-sim) instead of the hardware one :meth:connect builds.

Source code in eden/extensions/deployment/wuji_hand.py

def use_transport(self, transport: HandTransport) -> None:
    """Drive ``transport`` (e.g. a simulator for sim-to-sim) instead of the hardware one :meth:`connect` builds."""
    owner = type(self).__name__
    if self._connected:
        raise RuntimeError(f"{owner}.use_transport must be called before connect().")
    if transport.stream.mode != self.stream_mode:
        kind = type(transport).__name__
        raise ValueError(
            f"{owner}.use_transport: {kind} streams {transport.stream.mode!r}, but the deployment is configured "
            f"with stream_mode={self.stream_mode!r} (which decides whether the base rate limiter paces the loop). "
            f"Build it with the deployment's StreamOptions: {kind}(dep.stream, ...)."
        )
    self._transport = transport

connect

connect() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def connect(self) -> None:
    if self._transport is None:
        self._transport = self._make_transport()
    en.logger.info(f"[wuji] {self.side} Wuji {'Hand 1' if self.hand == 'hand1' else 'Hand 2'}")
    self._transport.connect()
    self._connected = True
    self.reset_velocity_filter()

close

close() -> None

Source code in eden/extensions/deployment/wuji_hand.py

def close(self) -> None:
    try:
        self._teardown_common()
    finally:
        # De-energize the hand even if flushing the record or restoring a command raised.
        if self._transport is not None:
            self._transport.close()
        self._connected = False

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/wuji_hand.py

def read_state(self) -> RobotState:
    q = self.from_sdk(self.transport.read())
    t = self.transport.clock()
    if self._prev_q is not None and t > self._prev_t:
        dt = t - self._prev_t
        vel = (q - self._prev_q) / dt
        if self.vel_lowpass_hz:
            a = dt / (dt + 1.0 / (2.0 * np.pi * self.vel_lowpass_hz))
            vel = a * vel + (1.0 - a) * self._vel
        self._vel = vel
    self._prev_q, self._prev_t = q.copy(), t
    return RobotState(
        stamp=t,
        base_quat=self._mount_quat,
        base_ang_vel=np.zeros(3),
        base_lin_acc=np.zeros(3),
        dofs_pos=q,
        dofs_vel=self._vel.copy(),
        dofs_torque=np.zeros(self.num_dofs),
    )

reset_velocity_filter

reset_velocity_filter() -> None

Start the finite-difference velocity afresh from the next read (upstream does, entering a new phase).

Source code in eden/extensions/deployment/wuji_hand.py

def reset_velocity_filter(self) -> None:
    """Start the finite-difference velocity afresh from the next read (upstream does, entering a new phase)."""
    self._prev_q = self._prev_t = None
    self._vel = np.zeros(self.num_dofs)

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/wuji_hand.py

def send_payload(self, payload: RobotCommand) -> None:
    # The payload carries what is sent, so a debug record logs the filtered, clipped target.
    payload.dofs_pos = self._clip(self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64)))
    self.transport.command(self.to_sdk(payload.dofs_pos))

check_hand

check_hand() -> None

Raise :class:DeployStop on stale joint feedback, or a latched hand fault (stop_on_fault).

Source code in eden/extensions/deployment/wuji_hand.py

def check_hand(self) -> None:
    """Raise :class:`DeployStop` on stale joint feedback, or a latched hand fault (``stop_on_fault``)."""
    if stale := self.transport.stale():
        raise DeployStop(f"hand feedback lost: {stale}")
    if self.stop_on_fault and (fault := self.transport.fault()):
        raise DeployStop(f"hand fault: {fault}")

check_safety

check_safety() -> None

Run before every policy tick: :meth:check_hand. Subclasses extend it with their own guards.

Source code in eden/extensions/deployment/wuji_hand.py

def check_safety(self) -> None:
    """Run before every policy tick: :meth:`check_hand`. Subclasses extend it with their own guards."""
    self.check_hand()

sync_scene

sync_scene() -> None

Write what else the run observes into the scene after the hand is read; a no-op here.

Source code in eden/extensions/deployment/wuji_hand.py

def sync_scene(self) -> None:
    """Write what else the run observes into the scene after the hand is read; a no-op here."""

observe

observe() -> dict[str, Any]

Read the hand and return the observation, advancing nothing.

Source code in eden/extensions/deployment/wuji_hand.py

def observe(self) -> dict[str, Any]:
    """Read the hand and return the observation, advancing nothing."""
    self.state = self.read_state()
    return self.state_to_observation(self.state)

ease_to

ease_to(target: ndarray, duration_s: float, rate_hz: float, profile: str = 'cosine') -> None

Move the hand from its measured pose to target (deployment DOF order), "cosine" or "smoothstep".

:meth:check_hand runs before every waypoint. The eased path is not low-passed (it is smooth already), but target_lpf restarts from target, so the next command filters from where the hand was eased to rather than pulling it back toward the target before the ease.

Source code in eden/extensions/deployment/wuji_hand.py

def ease_to(self, target: np.ndarray, duration_s: float, rate_hz: float, profile: str = "cosine") -> None:
    """Move the hand from its measured pose to ``target`` (deployment DOF order), ``"cosine"`` or ``"smoothstep"``.

    :meth:`check_hand` runs before every waypoint. The eased path is not low-passed (it is smooth
    already), but ``target_lpf`` restarts from ``target``, so the next command filters from where
    the hand was eased to rather than pulling it back toward the target before the ease.
    """
    self.check_hand()
    en.logger.info(f"[wuji] easing the hand to its target over {duration_s:g} s ({profile})")
    target = self._clip(np.asarray(target, dtype=np.float64))
    self.transport.ease_to(self.to_sdk(target), duration_s, rate_hz, profile, guard=self.check_hand)
    if self.target_lpf < 1.0:
        self._lpf_prev = target.copy()

Functions:

sdk_index_map

sdk_index_map(names: Sequence[str], owner: str) -> ndarray

Each of names' index in the SDK's joint array, raising (as owner) unless they cover the 20 once each.

Source code in eden/extensions/deployment/wuji_hand.py

def sdk_index_map(names: Sequence[str], owner: str) -> np.ndarray:
    """Each of ``names``' index in the SDK's joint array, raising (as ``owner``) unless they cover the 20 once each."""
    index = resolve_dof_index_map(names, WUJI_JOINT_IDX, owner=owner, what="Wuji SDK joint index", hint=_NAMING_HINT)
    if sorted(index) != list(range(NUM_JOINTS)):
        by_index = {i: [n for n, j in zip(names, index) if j == i] for i in sorted(set(index))}
        missing = sorted(set(range(NUM_JOINTS)) - set(index))
        twice = {i: dup for i, dup in by_index.items() if len(dup) > 1}
        raise ValueError(
            f"{owner}: the joints must cover each of the {NUM_JOINTS} Wuji SDK joints once; SDK joints {missing} "
            f"have no joint and {twice} have several (SDK index: joint names). {_NAMING_HINT}"
        )
    return np.asarray(index)

ease

ease(x: float, profile: str) -> float

Progress along an :data:EASE_PROFILES curve at x in [0, 1]; 0 and 1 at the ends, zero slope at both.

Source code in eden/extensions/deployment/wuji_hand.py

def ease(x: float, profile: str) -> float:
    """Progress along an :data:`EASE_PROFILES` curve at ``x`` in [0, 1]; 0 and 1 at the ends, zero slope at both."""
    if profile not in EASE_PROFILES:
        raise ValueError(f"ease profile must be one of {EASE_PROFILES}; got {profile!r}.")
    return 0.5 * (1.0 - np.cos(np.pi * x)) if profile == "cosine" else x * x * (3.0 - 2.0 * x)

Composite rigs

Deploy a composite rig -- an arm with a hand welded on, and kin -- as one robot.

A composite (:class:~eden.options.entities.CompositeEntityOptions, e.g. :class:~eden.options.robots.FrankaXHand) sits in the task's scene under one name and expands into several entities when the env builds -- the Franka arm as robot, the XHand as hand. Each part is its own piece of hardware with its own SDK, so :class:CompositeDeployment drives each through the backend that part's robot is bound to (:func:resolve_deployment_for) and presents the set as one robot to everything above it: the action manager runs once per tick and its targets are split across the parts, the parts' readings are joined into one :class:RobotState, and one observation is computed from all of them.

The parts' DOFs are laid end to end parent first -- a welded part always after the part it is welded to, otherwise in the rig's declaration order -- so dofs_name -- the layout of every command, of --disable_joints and of --debug_record -- is the arm's joints followed by the hand's.

The rig runs the one operator state machine every backend runs (boot in STOP, INIT, START, STOP, KILL), once, for all of its parts: one START hands the whole rig to the policy.

Each part's wire -- connect, read_state, send_payload, _idle_tick, close -- runs on that part's own thread, every part at once, so a tick's I/O takes as long as its slowest part rather than the sum of them all. The policy tick stays one: the loop waits for every part's reading before it observes, and for every part's send before it moves on.

A part runs as it does deployed alone: its backend's own option defaults, its own wire, its own hooks. The composite adds the joining and changes nothing on a part. The one thing it cannot keep per part is a backend's loop setting that acts on the env every part shares (sync_before_commands, advances_episode_clock): the rig follows one only when every part does, and otherwise runs the standard loop and says which parts it does not follow.

Attributes

FORWARDED_OPTIONS module-attribute

FORWARDED_OPTIONS: tuple[str, ...] = ('connect_timeout_s', 'target_lpf', 'deploy_kp_kd_scale')

PART_HOOKS module-attribute

PART_HOOKS: frozenset[str] = frozenset({'_resolve_robot_options', '_resolve_control_dt', '_resolve_dofs', '_resolve_disabled_joints', '_resolve_options', '_resolve_feet_links', '_validate_init_target', '_validate_on_motion_end', '_build_term_action_maps', '_swap_in_state_entity', '_make_lin_vel_estimator', '_undo_swaps', 'connect', 'close', 'read_state', 'send_payload', '_before_step', '_idle_tick', '_prepare_reset', '_sync_state', '_update_lin_vel_estimate', '_remote_inputs', '_print_control_guide', '_motion_start_dofs', 'programmed_gains', 'command_gain_dofs', 'blocks_for_the_period', 'provides_base_lin_vel', '_publishes_base_lin_vel'})

Classes

CompositeDeployment

CompositeDeployment(env: RLEnvBase, options: DeploymentOptions)

Bases: DeploymentBase

Drive every part of a composite rig through its own backend, as one robot.

eden deploy picks this backend for any composite (:func:resolve_deployment_for), so a task whose scene holds robot=FrankaXHand(...) deploys onto a Franka and an XHand with no extra flags, assuming both are connected. Each part with joints gets the backend its robot is bound to (FrankaPandaDeployment, RoboTeraXHandDeployment), constructed against its own scene entity; a part with none (a mount) is not deployed.

What the composite owns, and the parts do not:

• The loop and the state machine. One action-manager pass and one observation per tick, and one INIT / START / STOP / KILL state machine for the whole rig: it boots in STOP holding every part's measured pose, INIT ramps every joint, and START hands the whole rig to the policy. The operator's inputs are every part's remote (the Oli's joystick, the G1's) plus one keyboard. :meth:connect, :meth:read_state, :meth:send_payload, _idle_tick and :meth:close run on every part at once, each on the part's own thread, and the loop waits for all of them: a tick's I/O takes as long as its slowest part. The scene sync and the other per-tick hooks (_before_step -- a Wuji hand's safety check -- _prepare_reset) run on the loop's thread, part by part, every welded part after its parent (whatever order entity_parts lists them in), so a welded part's FK reads its parent's fresh pose.
• The pace. The parts are built with sync=False and the composite's rate limiter paces the loop -- unless a part's send itself waits out the control period (a Wuji hand's default stream_mode="sync", :attr:~DeploymentBase.blocks_for_the_period). Then that part paces the rig, as it paces itself deployed alone, and the rig runs no rate limiter on top; the parts' sends overlap, so two such hands still take one period, not two.
• The loop settings. A backend's sync_before_commands / advances_episode_clock act on the env every part shares, so they cannot be kept per part: the rig follows one only when every part sets it, and otherwise runs the standard loop -- the order and clock a part without it runs deployed alone -- and warns which parts it does not follow. No part takes on another part's loop.
• The gains. Commands carry the concatenation of the parts' robot gains, scaled by deploy_kp_kd_scale and zeroed on disable_joints. A part whose robot ignores each command's gains (the Franka's controller, the XHand's and Wuji's firmware) cannot go limp that way, so a disabled joint there is refused. Each part takes the rig's scale unless its parts_options sets its own, and its backend also applies it its own way (a Wuji Hand 2 programs its MIT gains from it) or refuses it (an XHand, anything but 1.0).
• The base reading. Base orientation, angular velocity and acceleration come from the rig's root part (its primary entity), as does the velocity estimator (estimate_lin_vel), on the rig's feet (a :class:~eden.options.robots.composite.base.HumanoidOptions' feet_links). A welded part's own base pose is read off FK through the weld, not off its (IMU-less) reading.
• Commands, recording, the motion clock -- everything else :class:DeploymentBase does.

Every other option of a part's backend keeps its own default, as when the part is deployed alone. Each part also runs the composite's :data:FORWARDED_OPTIONS (no backend changes their defaults); parts_options overrides those, and sets each backend's own options, per part::

CompositeDeployment.configure(parts_options={
    "robot": {"robot_ip": "172.16.0.2"},
    "hand": {"protocol": "EtherCAT"},
})

(eden deploy --part_option robot.robot_ip=172.16.0.2 --part_option hand.protocol=EtherCAT.)

Not supported yet, and refused at construction: a part whose backend overrides a method the composite never calls on a part (step, reset, the hold payload -- anything outside :data:PART_HOOKS), two parts with a joint of the same name, and --mujoco sim-to-sim of a composite.

PARAMETER	DESCRIPTION
parts_options	Backend options per part, keyed by the part's scene entity name. A key may be any option of that part's backend except the composite-owned base options. TYPE: dict[str, dict]

Source code in eden/extensions/deployment/composite.py

def __init__(self, env: RLEnvBase, options: DeploymentOptions) -> None:
    #: The deployed parts, by scene entity name, in rig order. Filled by :meth:`_resolve_dofs`.
    self.parts: dict[str, DeploymentBase] = {}
    self._part_slices: dict[str, slice] = {}
    #: Each part's own thread for its wire calls, started on first use (:meth:`_on_part_threads`).
    self._part_threads: dict[str, _PartThread] = {}
    try:
        super().__init__(env, options)
    except BaseException:
        # The parts swap their own entities as they are built, before the composite's own
        # rollback is armed; put back every one that got that far.
        self._undo_part_swaps()
        raise
    if paced := [name for name, part in self.parts.items() if part.blocks_for_the_period]:
        # Each such part waits out the period in its own send, on its own thread, as it does
        # deployed alone (where it runs no rate limiter either); a limiter on top would add
        # whatever the tick spent outside the send to every period.
        self._rate = None
        en.logger.info(
            f"[deploy] Part(s) {paced} wait out the control period in their own send and pace the rig, as they "
            "pace themselves deployed alone; the rig runs no rate limiter of its own."
        )

Attributes

parts_options class-attribute instance-attribute

parts_options: dict = {}

parts instance-attribute

parts: dict[str, DeploymentBase] = {}

provides_base_lin_vel property

provides_base_lin_vel: bool

Methods:

part_backends classmethod

part_backends(rig: CompositeEntityOptions) -> dict[str, type[DeploymentBase]]

The backend each jointed part of rig deploys through, parent first.

Raises as construction would for a part that has none or cannot run inside a rig. eden deploy asks before constructing, to hand a backend-specific flag (--publish_rate) to the parts that declare it.

Source code in eden/extensions/deployment/composite.py

@classmethod
def part_backends(cls, rig: CompositeEntityOptions) -> dict[str, type[DeploymentBase]]:
    """The backend each jointed part of ``rig`` deploys through, parent first.

    Raises as construction would for a part that has none or cannot run inside a rig. ``eden
    deploy`` asks before constructing, to hand a backend-specific flag (``--publish_rate``) to
    the parts that declare it.
    """
    return {name: cls._part_backend(rig, name, options) for name, options in cls._jointed_parts(rig).items()}

connect

connect() -> None

Connect every part at once. If one fails, the ones that came up are closed again, then it raises.

Source code in eden/extensions/deployment/composite.py

def connect(self) -> None:
    """Connect every part at once. If one fails, the ones that came up are closed again, then it raises."""
    _, errors = self._on_part_threads(lambda name, part: part.connect())
    if errors:
        up = {name: part for name, part in self.parts.items() if name not in errors}
        _, unclosed = self._on_part_threads(lambda name, part: part.close(), up)
        for name, exc in unclosed.items():
            en.logger.error(
                f"[deploy] Could not close part '{name}' ({type(self.parts[name]).__name__}) after a failed "
                f"connect: {exc!r}"
            )
        self._raise_part_error(errors, "connecting")

close

close() -> None

Close every part at once, each even if another raises; then raise the first failure.

Source code in eden/extensions/deployment/composite.py

def close(self) -> None:
    """Close every part at once, each even if another raises; then raise the first failure."""
    errors: list[BaseException] = []
    try:
        self._teardown_common()
    except Exception as exc:
        errors.append(exc)
    _, failed = self._on_part_threads(lambda name, part: part.close())
    for name, exc in failed.items():
        en.logger.error(f"[deploy] Closing part '{name}' ({type(self.parts[name]).__name__}) failed: {exc!r}")
        errors.append(exc)
    self._stop_part_threads()
    if errors:
        raise errors[0]

read_state

read_state() -> RobotState

Read every part at once and join the readings, with the base fields from the base part.

Each part keeps its own reading as its state too, which its backend reads back (the Franka clamps its next command against it). A part's extra entries are prefixed with the part's name, so --debug_record keeps them apart.

Source code in eden/extensions/deployment/composite.py

def read_state(self) -> RobotState:
    """Read every part at once and join the readings, with the base fields from the base part.

    Each part keeps its own reading as its ``state`` too, which its backend reads back (the
    Franka clamps its next command against it). A part's ``extra`` entries are prefixed with
    the part's name, so ``--debug_record`` keeps them apart.
    """
    states, errors = self._on_part_threads(lambda name, part: part.read_state())
    self._raise_part_error(errors, "reading its state")
    for name, part in self.parts.items():
        part.state = states[name]
    base = states[self._base_name]
    return RobotState(
        stamp=base.stamp,
        base_quat=base.base_quat,
        base_ang_vel=base.base_ang_vel,
        base_lin_acc=base.base_lin_acc,
        dofs_pos=np.concatenate([state.dofs_pos for state in states.values()]),
        dofs_vel=np.concatenate([state.dofs_vel for state in states.values()]),
        dofs_torque=np.concatenate([state.dofs_torque for state in states.values()]),
        extra={f"{name}.{key}": value for name, state in states.items() for key, value in state.extra.items()},
    )

send_payload

send_payload(payload: RobotCommand) -> None

Send every part its slice of payload at once; payload.dofs_pos then holds what was sent (filtered).

Each part sends it as the rig does: a hold of the rig is a hold of every part (:attr:~DeploymentBase.sending_hold), so a Franka part keeps its gripper where the policy left it.

Source code in eden/extensions/deployment/composite.py

def send_payload(self, payload: RobotCommand) -> None:
    """Send every part its slice of ``payload`` at once; ``payload.dofs_pos`` then holds what was sent (filtered).

    Each part sends it as the rig does: a hold of the rig is a hold of every part
    (:attr:`~DeploymentBase.sending_hold`), so a Franka part keeps its gripper where the policy left it.
    """
    commands = {
        name: RobotCommand(
            dofs_pos=_slice(payload.dofs_pos, part_slice),
            dofs_vel=_slice(payload.dofs_vel, part_slice),
            dofs_torque=_slice(payload.dofs_torque, part_slice),
            dofs_kp=_slice(payload.dofs_kp, part_slice),
            dofs_kd=_slice(payload.dofs_kd, part_slice),
        )
        for name, part_slice in self._part_slices.items()
    }
    for part in self.parts.values():
        part._sending_hold = self._sending_hold
    try:
        _, errors = self._on_part_threads(lambda name, part: part.send_payload(commands[name]))
    finally:
        for part in self.parts.values():
            part._sending_hold = False
    self._raise_part_error(errors, "sending its command")
    sent = np.array(payload.dofs_pos, dtype=np.float64)
    for name, command in commands.items():
        sent[self._part_slices[name]] = command.dofs_pos
    payload.dofs_pos = sent

command_gain_dofs

command_gain_dofs() -> ndarray

Each part's own answer on its slice: a body that takes each command's gains next to hands that do not.

Source code in eden/extensions/deployment/composite.py

def command_gain_dofs(self) -> np.ndarray:
    """Each part's own answer on its slice: a body that takes each command's gains next to hands that do not."""
    return np.concatenate([part.command_gain_dofs() for part in self.parts.values()])

programmed_gains

programmed_gains() -> tuple[ndarray, ndarray] | None

The programmed gains of every part that runs its own, end to end; None if one has none in our units.

A part that takes each command's gains runs no fixed set, so its slice holds what it is commanded now; :meth:command_gain_dofs tells the two apart.

Source code in eden/extensions/deployment/composite.py

def programmed_gains(self) -> tuple[np.ndarray, np.ndarray] | None:
    """The programmed gains of every part that runs its own, end to end; ``None`` if one has none in our units.

    A part that takes each command's gains runs no fixed set, so its slice holds what it is
    commanded now; :meth:`command_gain_dofs` tells the two apart.
    """
    kp, kd = [], []
    for name, part in self.parts.items():
        part_slice = self._part_slices[name]
        gains = (
            (self.default_dof_kp[part_slice], self.default_dof_kd[part_slice])
            if part.sends_command_gains
            else part.programmed_gains()
        )
        if gains is None:
            return None
        kp.append(gains[0])
        kd.append(gains[1])
    return np.concatenate(kp), np.concatenate(kd)

Functions:

MuJoCo sim-to-sim

MuJoCo sim-to-sim deployment backend.

A :class:~eden.extensions.deployment.base.DeploymentBase whose "robot" is a MuJoCo simulation instead of hardware. It runs the exact deploy control loop (eden.utils.cli.tasks._deploy_control_loop) — policy loading, the RobotStateEntity swap, observation assembly, command terms — but reads state from and writes torques to a MuJoCo model rather than a Unitree DDS wire. That makes it a faithful sim-to-sim harness for validating a Genesis-trained policy against an independent physics engine before it ever touches a real robot.

Inspired by Unitree's unitree_mujoco bridge (github.com/unitreerobotics/unitree_mujoco), which stands up a MuJoCo world speaking the Unitree LowState/LowCmd DDS protocol. This backend takes the simpler, robot-agnostic route: it drives MuJoCo directly (no DDS, no unitree_sdk2py), so it works for any robot Eden can describe — the two Unitree families here (G1, Go2) and anything else with a RobotOptions and a loadable MJCF/URDF. The engine details live in :class:~eden.extensions.deployment.utils.mujoco_sim.MujocoSim; this class is the glue between it and the Eden env.

Selected at deploy time with eden deploy ... --mujoco (see deploy_task), which overrides the usual robot→backend binding so any task/robot can be routed through MuJoCo.

Physics can run in lockstep with the control loop (the default: each command advances the simulation exactly one control period, so a rollout is deterministic and a late policy costs nothing) or asynchronously (--async / async_physics=True: the simulation advances on its own thread in wall-clock time, like a robot's controller, so a late policy is late for real). See :class:~eden.extensions.deployment.utils.realtime_physics.RealtimePhysics.

Attributes

Classes

MujocoDeployment

MujocoDeployment(env: EnvBase, options: DeploymentOptions)

Bases: DeploymentBase

Deploy a policy against a MuJoCo simulation of the robot (sim-to-sim).

Configurable options (in addition to :class:DeploymentBase's):

mjcf_path: str | None Explicit model file to load into MuJoCo, bypassing mjcf_source resolution. Point this at any hand-authored MuJoCo scene for one-off sim-to-sim runs. mjcf_source: str Which model to load when mjcf_path is unset (default "eden"):

- ``"eden"``: the robot's own Genesis training asset (``robot_options.file``).
  Always available; the MuJoCo backend reconstructs armature/floor for it.
- ``"menagerie"``: a community-tuned `MuJoCo Menagerie
  <https://github.com/google-deepmind/mujoco_menagerie>`_ model (with the
  armature/damping/contact params MuJoCo needs), fetched from an HF dataset — the
  higher-fidelity, more independent sim-to-sim target. Errors if the robot has no
  Menagerie mapping or the download fails. Selected by ``eden deploy --menagerie``.
- ``"auto"``: Menagerie when the robot is mapped and the download succeeds,
  otherwise fall back to the Eden asset. Not reachable from the CLI, which keeps
  the loaded model deterministic.

menagerie_registry: str HF org for the Menagerie dataset (default "Kashu7100"). menagerie_dataset: str HF dataset name mirroring MuJoCo Menagerie (default "mujoco_menagerie"). sim_dt: float | None MuJoCo integration timestep. None keeps the asset's timestep, or in lockstep, when it does not divide the control period, takes the largest step under it that does. In lockstep each step advances control_dt / sim_dt substeps, so a given sim_dt must divide the control period: one that does not is refused rather than rounded, which would run the physics on a clock ahead of or behind the policy's. integrator: str MuJoCo integrator ("implicitfast" default; also "implicit", "euler", "rk4"). implicitfast keeps a stiff joint PD stable. add_floor: bool Add a ground plane if the model has none (default True). It stands where the scene's ground plane does, raised or tilted alike, with its friction (:func:~eden.extensions.deployment.utils.mujoco_scene.read_floor), but keeps MuJoCo's default solver parameters: the robot's own geoms keep its asset's, and a contact's mix of a Genesis value and a MuJoCo one would be neither engine's. provide_ground_truth_base: bool Feed MuJoCo's exact base position and linear velocity to the observation pipeline (default True). Unlike hardware — where these must be estimated — the simulator knows them, so policies that observe base position / linear velocity work without --estimate_lin_vel. Set False to emulate a hardware-blind base.

Combining it with ``estimate_lin_vel`` is the useful sim-to-sim rehearsal of the
estimator, not a conflict: the estimate is installed *after* the ground-truth
reading, so it is what the policy observes, while the truth stays alongside it in
``state.extra["base_lin_vel"]`` for the estimate
(``state.extra["base_lin_vel_estimate"]``) to be scored against.

imu_acc_noise, imu_gyro_noise: float Standard deviation of white noise added to every accelerometer (m/s^2) and gyro (rad/s) reading, per axis. Default 0: MuJoCo's IMU is otherwise exact, which hides how an IMU-driven estimate behaves on hardware. Under estimate_lin_vel an exact accelerometer barely drifts even with no foot contact, so a contact dropout that sends the real robot's estimate to -0.5 m/s stays at 0.03 m/s in sim. Turn these on to rehearse the estimator against a hardware-like IMU. imu_acc_bias, imu_gyro_bias: tuple[float, float, float] Constant bias added to the accelerometer (m/s^2) and gyro (rad/s) readings, body frame. Default zero. imu_noise_seed: int | None Seed for the IMU noise generator, so a noisy run can be repeated. None (default) draws a fresh seed. show_viewer: bool Open MuJoCo's native passive viewer to watch the sim-to-sim rollout. record_video: str | None Write the rollout from the policy's first START on to this mp4, rendered offscreen from MuJoCo (:class:~eden.extensions.deployment.utils.mujoco_video.MujocoVideo: two views, one frame per control tick, a STOP's holds included). Lockstep only. Headless, it needs MUJOCO_GL=egl. gantry: bool Hang the robot from a simulated overhead hoist -- the sim counterpart of the bring-up rig, so a biped can be held upright while a controller is commissioned instead of falling every time control pauses. See :meth:~eden.extensions.deployment.utils.mujoco_sim.MujocoSim.set_gantry for the rope model; it is a tension-only spring-damper to a fixed anchor, attached above the centre of mass so it produces a real restoring torque rather than just cancelling weight.

Re-rigged on every :meth:`reset` from wherever the robot then stands, so the anchor
follows a re-placed robot instead of yanking it back to the first episode's spot.

gantry_body: tuple[str, ...] Bodies the ropes are tied to, one rope each. Empty (default) uses the robot's own :attr:~eden.options.entities.RobotOptions.gantry_links_name — the two shoulders for a humanoid that declares them — and falls back to the floating base for a robot that declares none. gantry_attach: tuple[float, float, float] | None Attachment point within each rope's body frame, metres. None (default) picks it from how the robot is rigged: the link origin for declared mount links, which for a shoulder sits on the pitch axis at the top of the torso, or 25 cm up for a single base rope, which needs the height to have any moment arm at all. gantry_height: float | None World height of the anchor. Defaults to 1 m above the attachment point. gantry_step: float Metres the operator's + / - keys move the anchor per press. gantry_draw: bool Draw each rope and the force it is pulling with in the --vis viewer (default on). Seeing the pull is most of the value of a simulated rig: a rope that is quietly carrying the robot looks exactly like a robot that is standing, right up until you unrig it. gantry_force_scale: float Metres of force arrow per newton (default 1e-3, so 1 kN draws as 1 m). gantry_stiffness, gantry_damping: Rope spring-damper, N/m and N.s/m. gantry_slack: float Metres of slack. 0 (default) is taut and holds station; a positive value is a safety tether that does nothing until the robot drops that far. gantry_support: float Scale on the rope force, 0-1. Ramp down to hand weight back to the controller. init_steps: int Control cycles the INIT ramp takes to reach the start pose — the sim-to-sim counterpart of the hardware backends' INIT state, inherited from :class:DeploymentBase along with its default of 250 (5 s at 50 Hz). The ramp runs when the operator presses INIT, or on its own in a headless run. The policy does not drive straight from the reset pose: this backend runs the same INIT/START sequence the robot does. init_target: str Inherited from :class:DeploymentBase; this backend is the one that also accepts "motion_rsi", which places joints and base at the tracked reference's frame 0 on reset. "motion" and "default" are the same joint-space ramps here as on the wire. async_physics: bool Run the simulation on its own thread in wall-clock time (default False). Off, physics advances control_dt / sim_dt substeps inside each send_payload and then waits for the next command, so the control loop's timing has no effect on the rollout -- deterministic, and the right mode for comparing two runs. On, the simulation is the robot: it steps at 1 / sim_dt whether or not a new command has arrived, send_payload only replaces the command it holds, and read_state returns the latest state it has published. A policy that overruns the control period therefore leaves the robot executing its previous command for the overrun, and observes a correspondingly older state -- the behaviour hardware has and lockstep cannot show. Selected by eden deploy --mujoco --async.

The simulated robot has no command latency. A task's :class:~eden.managers.modifiers.actions.actuators.ActionDelay models the Genesis actuator's latency, and no deployment backend applies it: each command acts from the substep it arrives on. A policy trained against a delay therefore runs here with less latency than it trained with. This matters most for a fixed delay (both ends of the range equal), which the policy expects on every step. --async adds only the control loop's real lateness, not the trained delay. eden inference --async is the run that keeps the delay.

Being robot-agnostic, this backend knows nothing robot-specific — including the feet the estimate_lin_vel filter needs. It does not have to: the robot declares them itself (RobotOptions.feet_links_name), so eden deploy --mujoco --estimate_lin_vel needs no further flag for any robot that names its feet.

Source code in eden/extensions/deployment/mujoco.py

def __init__(self, env: EnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)
    # A note, not a check, so it stays out of _resolve_options.
    if self.estimate_lin_vel and self.provide_ground_truth_base:
        en.logger.info(
            "[deploy] MuJoCo sim-to-sim: the policy observes the *estimated* base_lin_vel "
            "(estimate_lin_vel overrides the ground-truth reading); the simulator's truth "
            "rides along in state.extra['base_lin_vel'] for comparison. Set "
            "provide_ground_truth_base=False to also blind the base position."
        )
    self._sim: MujocoSim | None = None
    self._physics: RealtimePhysics | None = None
    # The kd the armature currently folds in (``MujocoSim.set_implicit_damping``); set at connect.
    self._damped_kd: np.ndarray | None = None
    self._viewer = None
    self._video: MujocoVideo | None = None
    self._n_substeps = 1
    self._base_pos0 = np.asarray(self.robot_options.default_root_pos, dtype=np.float64)
    self._base_quat0 = np.asarray(self.robot_options.default_root_quat, dtype=np.float64)
    self._imu_rng = np.random.default_rng(self.imu_noise_seed)

Attributes

mjcf_path class-attribute instance-attribute

mjcf_path: str | None = None

mjcf_source class-attribute instance-attribute

mjcf_source: str = 'eden'

menagerie_registry class-attribute instance-attribute

menagerie_registry: str = MENAGERIE_REGISTRY

menagerie_dataset class-attribute instance-attribute

menagerie_dataset: str = MENAGERIE_DATASET

sim_dt class-attribute instance-attribute

sim_dt: float | None = None

integrator class-attribute instance-attribute

integrator: str = 'implicitfast'

add_floor class-attribute instance-attribute

add_floor: bool = True

provide_ground_truth_base class-attribute instance-attribute

provide_ground_truth_base: bool = True

gantry class-attribute instance-attribute

gantry: bool = False

gantry_body class-attribute instance-attribute

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

gantry_attach class-attribute instance-attribute

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

gantry_height class-attribute instance-attribute

gantry_height: float | None = None

gantry_step class-attribute instance-attribute

gantry_step: float = 0.05

gantry_draw class-attribute instance-attribute

gantry_draw: bool = True

gantry_force_scale class-attribute instance-attribute

gantry_force_scale: float = 0.001

gantry_stiffness class-attribute instance-attribute

gantry_stiffness: float = 5000.0

gantry_damping class-attribute instance-attribute

gantry_damping: float = 200.0

gantry_slack class-attribute instance-attribute

gantry_slack: float = 0.0

gantry_support class-attribute instance-attribute

gantry_support: float = 1.0

imu_acc_noise class-attribute instance-attribute

imu_acc_noise: float = 0.0

imu_gyro_noise class-attribute instance-attribute

imu_gyro_noise: float = 0.0

imu_acc_bias class-attribute instance-attribute

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

imu_gyro_bias class-attribute instance-attribute

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

imu_noise_seed class-attribute instance-attribute

imu_noise_seed: int | None = None

show_viewer class-attribute instance-attribute

show_viewer: bool = False

record_video class-attribute instance-attribute

record_video: str | None = None

supports_base_placement class-attribute

supports_base_placement: bool = True

allow_unattended_start class-attribute

allow_unattended_start: bool = True

operator_guide_title class-attribute

operator_guide_title: str = 'MUJOCO SIM-TO-SIM DEPLOY CONTROLS  (keyboard — same sequence as the robot)'

async_physics class-attribute instance-attribute

async_physics: bool = False

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/mujoco.py

def connect(self) -> None:
    asset = self._resolve_asset()
    floor = None
    if self.add_floor:
        # The ground's pose and friction; MuJoCo's solver parameters, as the robot's geoms keep the asset's.
        floor = dataclasses.replace(read_floor(self._env), sol_params=None)
    self._sim = self._load_sim(asset, self.sim_dt, floor)
    if self.sim_dt is None and not self.async_physics and not divides(self.control_dt, self._sim.sim_dt):
        # Lockstep ends every tick on a step: an asset's timestep off the control period gives way to
        # the largest under it that divides it, as a composite rig's default does.
        fit = fitting_timestep(self.control_dt, self._sim.sim_dt)
        en.logger.info(
            f"[deploy] MuJoCo sim-to-sim: the asset's timestep {self._sim.sim_dt:g} s does not divide the control "
            f"period {self.control_dt:g} s; running at {fit:g} s, the largest under it that does "
            "(--mujoco_sim_dt sets another)."
        )
        self._sim = self._load_sim(asset, fit, floor)
    say_floor_caveat(self._sim.floor)
    # Genesis integrates the PD damping implicitly (its implicit integrators) or folds kd*dt into
    # the armature (``euler``, RigidEntity.post_build); match that here, or a hand-rolled PD on the
    # zero-armature training MJCF diverges.
    self._fold_implicit_damping()
    if self.async_physics:  # the physics thread keeps its own clock, at 1 / sim_dt
        self._n_substeps = max(1, round(self.control_dt / self._sim.sim_dt))
    else:
        self._n_substeps = steps_per_tick(self.control_dt, self._sim.sim_dt)
    self._sim.reset(self.default_dof_pos, self._base_pos0, self._base_quat0)
    mode = "asynchronous, physics on its own thread" if self.async_physics else "lockstep with the control loop"
    en.logger.info(
        f"[deploy] MuJoCo sim-to-sim: {asset} | sim_dt={self._sim.sim_dt:g}s "
        f"× {self._n_substeps} substeps/step | integrator={self.integrator} | {mode}"
    )
    if self.show_viewer:
        import mujoco.viewer

        self._viewer = mujoco.viewer.launch_passive(self._sim.model, self._sim.data)
    if self.async_physics:
        overlay = self._draw_gantry_overlay if self.gantry and self.gantry_draw else None
        self._physics = RealtimePhysics(
            self._sim, self._hold_default_pose(), viewer=self._viewer, viewer_overlay=overlay
        )
        self._physics.start()
    if self.record_video is not None:
        from eden.extensions.deployment.utils.mujoco_video import MujocoVideo

        self._video = MujocoVideo(
            self._sim.model,
            self._sim.data,
            self.record_video,
            fps=1.0 / self.control_dt,
            framed_bodies=self._sim.robot_body_ids(),
            markers=kinematic_markers(self._env),
        )

close

close() -> None

Source code in eden/extensions/deployment/mujoco.py

def close(self) -> None:
    if self._physics is not None:
        # Stop the robot's clock before anything else, and let it grade itself: a physics
        # thread that fell behind wall time silently stretched every control period.
        self._physics.stop()
        log = en.logger.warning if self._physics.reanchors else en.logger.info
        log(self._physics.summary())
        self._physics = None
    self._teardown_common()
    if self._video is not None:
        self._video.close()
        self._video = None
    if self._viewer is not None:
        try:
            self._viewer.close()
        except Exception:  # noqa: BLE001 — best-effort on the way down
            pass
        self._viewer = None
    self._sim = None

rig_gantry

rig_gantry(enabled: bool) -> None

Rig or release the gantry mid-run, anchored over where the robot stands now.

The switch a caller driving its own loop needs to hand the robot's weight back to the controller (and to catch it again); gantry itself only sets how a reset rigs it.

Source code in eden/extensions/deployment/mujoco.py

def rig_gantry(self, enabled: bool) -> None:
    """Rig or release the gantry mid-run, anchored over where the robot stands now.

    The switch a caller driving its own loop needs to hand the robot's weight back to the
    controller (and to catch it again); ``gantry`` itself only sets how a reset rigs it.
    """
    self.gantry = enabled
    self._rig_gantry()
    if not enabled and self._viewer is not None:
        # Nothing redraws the overlay once the gantry is off; drop the last-drawn ropes.
        self._viewer.user_scn.ngeom = 0

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/mujoco.py

def read_state(self) -> RobotState:
    assert self._sim is not None, "connect() must be called before read_state()."
    if self._physics is not None:
        # The latest state the physics thread published, stamped when it was published:
        # in this mode the state has an age, as a wire's does.
        r = self._physics.read()
        stamp = r["stamp"]
    else:
        r = self._sim.read()
        stamp = time.time()
    extra: dict[str, Any] = {}
    if self.provide_ground_truth_base and not self.robot_options.is_fixed_base:
        # The simulator knows the true base pose/velocity; hand them to the
        # observation pipeline so base-position / base-lin-vel terms work without
        # an estimator (RobotStateEntity yaw-aligns them like any world-frame read).
        extra["base_pos"] = r["base_pos"]
        extra["base_lin_vel"] = r["base_lin_vel"]
    return RobotState(
        stamp=stamp,
        base_quat=r["base_quat"],  # wxyz
        base_ang_vel=self._corrupt_imu(r["base_ang_vel"], self.imu_gyro_noise, self.imu_gyro_bias),  # body (gyro)
        base_lin_acc=self._corrupt_imu(r["base_lin_acc"], self.imu_acc_noise, self.imu_acc_bias),  # specific force
        dofs_pos=r["dofs_pos"],
        dofs_vel=r["dofs_vel"],
        dofs_torque=r["dofs_torque"],
        extra=extra,
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/mujoco.py

def send_payload(self, payload: RobotCommand) -> None:
    assert self._sim is not None, "connect() must be called before send_payload()."
    # As every hardware backend does on its wire, so a --target_lpf rehearsed here is the one the robot gets.
    payload.dofs_pos = self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64))
    kp = payload.dofs_kp if payload.dofs_kp is not None else self.default_dof_kp
    kd = payload.dofs_kd if payload.dofs_kd is not None else self.default_dof_kd
    # The armature models the damping Genesis folded in at build from the robot the active slot
    # trained on, so it follows the slot's gains (``default_dof_kd``, which a slot switch ramps
    # through ``set_gains``) -- not a payload's own kd, such as the INIT ramp's
    # ``init_dofs_kd_scale``, which was never part of any training robot.
    refold = not np.array_equal(self.default_dof_kd, self._damped_kd)
    if self._physics is not None:
        # Hand the command to the robot's clock; it takes effect at the thread's next
        # substep, and the physics keeps running whether or not another one follows.
        command = PDCommand.of(payload.dofs_pos, payload.dofs_vel, kp, kd, payload.dofs_torque)
        if not refold:
            self._physics.set_command(command)
            return
        # One exclusive section for both, so no burst runs the new damping with the old
        # command or the old damping with the new one.
        with self._physics.exclusive():
            self._fold_implicit_damping()
            self._physics.set_command(command)
        return
    if refold:
        self._fold_implicit_damping()
    self._sim.step_pd(
        target_pos=payload.dofs_pos,
        target_vel=payload.dofs_vel,
        kp=kp,
        kd=kd,
        tau_ff=payload.dofs_torque,
        n_substeps=self._n_substeps,
    )
    if self._video is not None:
        self._video.tick(self.policy_active, marker_poses(self._env, self._video.markers))
    if self._viewer is not None and self._viewer.is_running():
        if self.gantry and self.gantry_draw:
            self._draw_gantry_overlay(self._viewer)
        self._viewer.sync()

Functions:

MuJoCo sim-to-sim of a composite rig, through the composite deployment itself.

eden deploy --mujoco on a scene whose --entity_name is a composite rig (KukaSharpa, FrankaXHand, OliWuji) runs :class:MujocoCompositeDeployment: the very :class:~eden.extensions.deployment.composite.CompositeDeployment that drives the rig's hardware -- one backend per part, each on its own thread, one state machine, the parts' readings joined into one observation -- with every part on a :class:MujocoPartDeployment in place of its hardware backend. The parts share one MuJoCo model of the whole rig (:class:~eden.extensions.deployment.utils.mujoco_rig.MujocoRig) plus the scene's props (a table, a tool), so a manipulation policy acts on a simulated object and observes it.

The MuJoCo model is assembled from the built Genesis env, so it is the rig the policy trained on:

• each part's own asset, welded where the env welds it -- the mount read off the env's forward kinematics, not re-derived -- and checked link by link against that kinematics at connect;
• each part's armature and joint damping as built, its material friction (default_geoms_friction per link), its links' contact solver parameters as built (a set_contact_sol_params startup event included) and gravity compensation;
• each prop at the pose the env spawned it, with its friction and contact parameters.

Each tick every part writes its command to the model on its own thread, then the composite steps the physics once for the control period (lockstep, as MujocoDeployment), reads the parts back on their threads, and writes each prop's pose into its env entity -- what a mocap system does on hardware -- before the observation is computed. Where hardware has a camera instead (an AprilTag pose estimate), prop_pos_noise / prop_rot_noise / prop_camera_hz / prop_latency_s make what the policy reads of each free prop one: noisy, late and at the camera's rate (the exact pose stays in the state as <prop>.true_pos / <prop>.true_quat for --debug_record). record_video renders the rollout from MuJoCo, as a single robot's backend does, and the base's policy_seconds ends a headless run. The env is read into the rig's description by :mod:~eden.extensions.deployment.utils.mujoco_scene; this class keeps the lifecycle, the checks and the camera.

Attributes

Classes

MujocoPartDeployment

MujocoPartDeployment(env: RLEnvBase, options: DeploymentOptions)

Bases: DeploymentBase

One part of a composite rig on the rig's shared MuJoCo model -- the sim's stand-in for a part's hardware backend.

Built only by :class:MujocoCompositeDeployment, which binds it to the model at connect. Its wire is the model: :meth:send_payload buffers the part's PD command (the composite steps the physics once every part has sent), :meth:read_state reads the part's joints, and for the rig's root its base -- the IMU, and the ground-truth pose and velocity of a free base when provide_ground_truth_base, as MujocoDeployment gives a single robot.

Source code in eden/extensions/deployment/mujoco_composite.py

def __init__(self, env: RLEnvBase, options: DeploymentOptions) -> None:
    # Before the base constructor, which asks :meth:`_publishes_base_lin_vel` already.
    self._rig: MujocoRig | None = None
    super().__init__(env, options)

Attributes

provide_ground_truth_base class-attribute instance-attribute

provide_ground_truth_base: bool = True

allow_unattended_start class-attribute

allow_unattended_start: bool = True

Methods:

bind

bind(rig: MujocoRig) -> None

Wire this part to the rig's model (:meth:MujocoCompositeDeployment.connect).

Source code in eden/extensions/deployment/mujoco_composite.py

def bind(self, rig: MujocoRig) -> None:
    """Wire this part to the rig's model (:meth:`MujocoCompositeDeployment.connect`)."""
    self._rig = rig

connect

connect() -> None

Source code in eden/extensions/deployment/mujoco_composite.py

def connect(self) -> None:
    self._require_rig()

close

close() -> None

Source code in eden/extensions/deployment/mujoco_composite.py

def close(self) -> None:
    self._teardown_common()

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/mujoco_composite.py

def read_state(self) -> RobotState:
    r = self._require_rig().read_part(self.entity_name)
    extra: dict[str, Any] = {}
    if self.provide_ground_truth_base and self._is_free_root():
        extra["base_pos"] = r["base_pos"]
        extra["base_lin_vel"] = r["base_lin_vel"]
    # A welded part has no IMU; its base is read off the forward kinematics through the weld.
    return RobotState(
        stamp=time.time(),
        base_quat=r.get("base_quat", np.array([1.0, 0.0, 0.0, 0.0])),
        base_ang_vel=r.get("base_ang_vel", np.zeros(3)),
        base_lin_acc=r.get("base_lin_acc", np.zeros(3)),
        dofs_pos=r["dofs_pos"],
        dofs_vel=r["dofs_vel"],
        dofs_torque=r["dofs_torque"],
        extra=extra,
    )

send_payload

send_payload(payload: RobotCommand) -> None

Source code in eden/extensions/deployment/mujoco_composite.py

def send_payload(self, payload: RobotCommand) -> None:
    kp = self.default_dof_kp if payload.dofs_kp is None else payload.dofs_kp
    kd = self.default_dof_kd if payload.dofs_kd is None else payload.dofs_kd
    # Each part low-passes its own slice, as its hardware backend does on its wire (``target_lpf``), and
    # leaves what it sent in the payload, which the composite joins into the rig's (``--debug_record``).
    payload.dofs_pos = self._apply_target_lpf(np.asarray(payload.dofs_pos, dtype=np.float64))
    self._require_rig().set_command(
        self.entity_name, payload.dofs_pos, payload.dofs_vel, kp, kd, payload.dofs_torque
    )

MujocoCompositeDeployment

MujocoCompositeDeployment(env: RLEnvBase, options: DeploymentOptions)

Bases: CompositeDeployment

A composite rig deployed onto MuJoCo: :class:CompositeDeployment with a :class:MujocoPartDeployment per part.

Options (in addition to :class:CompositeDeployment's):

sim_dt: float | None MuJoCo timestep. None (default) takes the largest step of at most 2 ms that divides the control period exactly, so each tick advances the physics by exactly one control period. A given one must divide it too (eden deploy --mujoco_sim_dt): one that does not is refused. integrator: str "implicitfast" (default), "implicit", "euler" or "rk4". cone: str Friction cone, "elliptic" (default -- grasps hold far better on it) or "pyramidal". props: tuple[str, ...] | None Scene entities to simulate besides the rig (eden deploy --mujoco_props). None (default) takes every colliding entity of the scene that is not the rig or the ground. Each must be a box primitive or a rigid body loaded from a URDF / MJCF, without joints of its own: only its pose reaches the env, which a mesh file's (MuJoCo cannot load one as a body) or an articulated object's joints would not. Anything else is refused, naming it. A kinematic entity (a goal marker: not simulated, no contacts) is never a prop; record_video draws it. prop_pos_noise, prop_rot_noise: float A camera's error on every free prop's pose the policy reads: per-axis white noise on the position (m, standard deviation), and a rotation about a random axis by a normal angle (rad, standard deviation). Default 0. With either on, the quaternion's sign is random too, as a pose estimator's is. Only what the policy observes changes; MuJoCo's physics does not, and the exact pose stays in the state as <prop>.true_pos / <prop>.true_quat. prop_camera_hz: float | None The camera's frame rate: a free prop's observed pose updates at it and holds between frames, on the simulation's clock (a tick that sends nothing advances neither). None (default): every control tick. prop_latency_s: float How old a frame is when it arrives, rounded half up to whole control ticks (with a warning when that changes it). Default 0. prop_noise_seed: int | None Seed for the camera model's noise, so a run can be repeated. None draws a fresh one. record_video: str | None Write the rollout from the policy's first START on to this mp4, rendered offscreen from MuJoCo (two views side by side, one frame per control tick, a STOP's holds included so the video never skips). Headless, it needs MUJOCO_GL=egl. kinematics_tolerance: float Largest link-pose disagreement (metres plus radians) between the MuJoCo model and the env's forward kinematics accepted at connect. Above it the assembly is wrong and connect raises. show_viewer: bool Open MuJoCo's passive viewer on the rig.

Source code in eden/extensions/deployment/mujoco_composite.py

def __init__(self, env: RLEnvBase, options: DeploymentOptions) -> None:
    super().__init__(env, options)
    self._rig: MujocoRig | None = None
    self._viewer = None
    self._n_substeps = 1
    self._damped_kd: np.ndarray | None = None
    # The yaw the props were last written into the env under (the env spawned them under none).
    self._props_yaw = 0.0
    self._prop_names = self._resolve_props()
    # Where each prop spawned, captured before any tick writes the simulated pose over it.
    self._prop_home = {name: entity_pose(self._env, name) for name in self._prop_names}
    self._camera = _PropCamera(
        pos_noise=self.prop_pos_noise,
        rot_noise=self.prop_rot_noise,
        hz=self.prop_camera_hz,
        latency_ticks=self._latency_ticks(),
        control_dt=self.control_dt,
        seed=self.prop_noise_seed,
    )
    self._video: MujocoVideo | None = None
    # Control periods the physics has advanced since the reset: the camera's clock.
    self._sim_ticks = 0

Attributes

sim_dt class-attribute instance-attribute

sim_dt: float | None = None

integrator class-attribute instance-attribute

integrator: str = 'implicitfast'

cone class-attribute instance-attribute

cone: str = 'elliptic'

props class-attribute instance-attribute

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

prop_pos_noise class-attribute instance-attribute

prop_pos_noise: float = 0.0

prop_rot_noise class-attribute instance-attribute

prop_rot_noise: float = 0.0

prop_camera_hz class-attribute instance-attribute

prop_camera_hz: float | None = None

prop_latency_s class-attribute instance-attribute

prop_latency_s: float = 0.0

prop_noise_seed class-attribute instance-attribute

prop_noise_seed: int | None = None

record_video class-attribute instance-attribute

record_video: str | None = None

kinematics_tolerance class-attribute instance-attribute

kinematics_tolerance: float = 0.0001

show_viewer class-attribute instance-attribute

show_viewer: bool = False

allow_unattended_start class-attribute

allow_unattended_start: bool = True

operator_guide_title class-attribute

operator_guide_title: str = 'MUJOCO RIG SIM-TO-SIM DEPLOY CONTROLS  (keyboard — same sequence as the robot)'

Methods:

connect

connect() -> None

Source code in eden/extensions/deployment/mujoco_composite.py

def connect(self) -> None:
    from eden.extensions.deployment.utils.mujoco_rig import MujocoRig

    # The largest step of at most 2 ms that divides the control period, unless one is given.
    sim_dt = fitting_timestep(self.control_dt, 0.002) if self.sim_dt is None else float(self.sim_dt)
    self._rig = MujocoRig(
        self._rig_parts(),
        rig_props(self._env, {name: self._prop_home[name] for name in self._prop_names}),
        sim_dt=sim_dt,
        integrator=self.integrator,
        cone=self.cone,
        floor=read_floor(self._env),
    )
    say_floor_caveat(self._rig.floor)
    self._check_kinematics()
    self._check_contact_floor()
    self._n_substeps = steps_per_tick(self.control_dt, self._rig.sim_dt)
    for part in self.parts.values():
        part.bind(self._rig)
    super().connect()
    self._reset_sim()
    en.logger.info(
        f"[deploy] MuJoCo rig sim-to-sim: parts {list(self.parts)}, props {self._prop_names} | "
        f"sim_dt={self._rig.sim_dt:.6g}s x {self._n_substeps} substeps/step | integrator={self.integrator} | "
        f"cone={self.cone} | lockstep with the control loop"
    )
    if self.show_viewer:
        import mujoco.viewer

        self._viewer = mujoco.viewer.launch_passive(self._rig.model, self._rig.data)
    if self.record_video is not None:
        from eden.extensions.deployment.utils.mujoco_video import MujocoVideo

        # Framed on the rig's bodies and its free props where they start.
        rig = self._rig
        framed = [rig.prop_body_id(name) for name, prop in rig.props.items() if prop.free] + rig.part_body_ids()
        self._video = MujocoVideo(
            rig.model,
            rig.data,
            self.record_video,
            fps=1.0 / self.control_dt,
            framed_bodies=framed,
            markers=kinematic_markers(self._env),
        )

close

close() -> None

Source code in eden/extensions/deployment/mujoco_composite.py

def close(self) -> None:
    try:
        super().close()
    finally:
        if self._video is not None:
            self._video.close()
            self._video = None
        if self._viewer is not None:
            try:
                self._viewer.close()
            except Exception:  # noqa: BLE001 — best-effort on the way down
                pass
            self._viewer = None

send_payload

send_payload(payload: RobotCommand) -> None

Every part writes its command (each on its thread), then the physics advances one control period.

Source code in eden/extensions/deployment/mujoco_composite.py

def send_payload(self, payload: RobotCommand) -> None:
    """Every part writes its command (each on its thread), then the physics advances one control period."""
    if self._rig is None:
        raise RuntimeError("MujocoCompositeDeployment: connect() must be called before send_payload().")
    if not np.array_equal(self.default_dof_kd, self._damped_kd):
        self._fold_implicit_damping()
    super().send_payload(payload)
    self._rig.step(self._n_substeps)
    self._sim_ticks += 1
    if self._viewer is not None and self._viewer.is_running():
        self._viewer.sync()
    if self._video is not None:
        self._video.tick(self.policy_active, marker_poses(self._env, self._video.markers))

read_state

read_state() -> RobotState

Source code in eden/extensions/deployment/mujoco_composite.py

def read_state(self) -> RobotState:
    state = super().read_state()
    for name in self._prop_names:
        pos, quat = self._rig.prop_pose(name)
        if self._rig.props[name].free and not self._camera.exact:
            # What the policy reads is the camera's; the exact pose stays beside it for a post-mortem.
            state.extra[f"{name}.true_pos"], state.extra[f"{name}.true_quat"] = pos, quat
            pos, quat = self._camera.observe(name, self._sim_ticks, pos, quat)
        state.extra[f"{name}.pos"] = pos
        state.extra[f"{name}.quat"] = quat
    return state

The MuJoCo world itself (MujocoSim, MujocoRig, the MujocoFloor both add), what both MuJoCo backends read off the built env (mujoco_scene.py), the offscreen recording (MujocoVideo, mujoco_video.py), the Eden-robot-to-Menagerie map (menagerie_path_for), the state and command dataclasses (RobotState, RobotCommand) and the loop helpers (RateLimiter, StreamingRecorder, LinVelKalmanEstimator) live in eden/extensions/deployment/utils/, which has no __init__.py, so they are not rendered here; read them in the source.

See also

• Deployment guide
• Policy export
• System identification