Teleoperation Devices

Input-device backends for teleoperation and for driving a policy by hand: keyboard, gamepad, SpaceMouse and GUI sliders for SE(2)/SE(3) commands, the GELLO leader arm, the Meta Quest and PICO headsets, and the Wuji Glove. Each device imports its driver lazily, so its optional extra is needed only when you use it. The teleoperation guide covers the eden gello, eden metaquest, eden pico, eden wuji and eden teleop pipelines.

from eden.utils.devices.se2 import make_se2_device

Base class

Base class for teleoperation input devices.

Classes

DeviceBase

DeviceBase()

Bases: RBC

Base class for all devices.

Source code in eden/utils/devices/base.py

def __init__(self):
    self._callbacks = {}

Methods:

add_callback

add_callback(key: str, callback: Callable) -> None

Source code in eden/utils/devices/base.py

def add_callback(self, key: str, callback: Callable) -> None:
    self._callbacks[key] = callback

remove_callback

remove_callback(key: str) -> Callable

Source code in eden/utils/devices/base.py

def remove_callback(self, key: str) -> Callable:
    return self._callbacks.pop(key, None)

reset

reset() -> None

Source code in eden/utils/devices/base.py

def reset(self) -> None:
    pass

get_raw_data abstractmethod

get_raw_data() -> Tensor

Source code in eden/utils/devices/base.py

@abstractmethod
def get_raw_data(self) -> torch.Tensor:
    pass

get_command abstractmethod

get_command() -> Tensor

Source code in eden/utils/devices/base.py

@abstractmethod
def get_command(self) -> torch.Tensor:
    pass

SE(2) and SE(3) devices

make_se2_device(name) builds a keyboard or gamepad SE(2) device by name; OperatorVelocityCommand (a velocity command term) reads one.

Building an SE(2) input device by name, and mapping its deflection to a velocity.

Both :class:~eden.utils.devices.keyboard.TeleopSE2Keyboard and :class:~eden.utils.devices.gamepad.SE2Gamepad emit a normalized (vx, vy, wz) deflection in [-1, 1]; the consumer owns the conversion to m/s and rad/s, the contract :class:~eden.utils.devices.keyboard.SE3Keyboard already documents. This module holds the two halves of that contract shared by every consumer: naming a device, and scaling its deflection into a velocity range.

A held key is full deflection, from the first press -- the same thing a stick pushed to its stop means. That is why the keyboard here is TeleopSE2Keyboard and not its SE2Keyboard base, whose tanh smoothing turns one press into 0.76 and only approaches full deflection as OS key-repeat accumulates, making the commanded speed depend on the operator's key-repeat rate.

The keyboard bindings are arrows and , / . -- deliberately not letters or digits. pynput listens to the keyboard globally rather than to the focused window, so it also sees the keys the operator types at the terminal to drive the deployment state machine (:mod:eden.extensions.deployment.utils.operator_input reads s START, x STOP, i INIT, k KILL, n and 1-9 from stdin). On WASD bindings, typing START would also command the robot backwards.

======================= ================= ================= Command Key (+ve axis) Key (-ve axis) ======================= ================= ================= Forward / backward vx Up Down Strafe vy Left Right Turn wz , . ======================= ================= =================

Attributes

DEVICES module-attribute

DEVICES: tuple[str, ...] = ('keyboard', 'gamepad')

GAMEPAD_LAYOUTS module-attribute

GAMEPAD_LAYOUTS: tuple[str, ...] = ('ps4', 'xbox')

Functions:

teleop_key_bindings

teleop_key_bindings() -> dict

SE(2) key bindings that do not collide with the deploy operator keys.

See the module docstring for the table and for why letters and digits are unusable. Requires pynput (the all extra), since the arrow keys are pynput key objects rather than characters.

Source code in eden/utils/devices/se2.py

def teleop_key_bindings() -> dict:
    """SE(2) key bindings that do not collide with the deploy operator keys.

    See the module docstring for the table and for why letters and digits are
    unusable. Requires ``pynput`` (the ``all`` extra), since the arrow keys are
    ``pynput`` key objects rather than characters.
    """
    from eden.utils.devices.keyboard import KeyBinding, Keyboard

    Keyboard._require_pynput()
    from pynput.keyboard import Key

    return {
        Key.up: KeyBinding(0, +1),
        Key.down: KeyBinding(0, -1),
        Key.left: KeyBinding(1, +1),
        Key.right: KeyBinding(1, -1),
        ",": KeyBinding(2, +1),
        ".": KeyBinding(2, -1),
    }

make_se2_device

make_se2_device(device: str, options: dict | None = None)

Build a normalized SE(2) input device by name.

PARAMETER	DESCRIPTION
device	One of :data:DEVICES. TYPE: str
options	Device options. layout ("ps4" | "xbox") picks the gamepad axis scaling; every other key reaches the device constructor. TYPE: dict | None DEFAULT: None

RETURNS	DESCRIPTION
DeviceBase	The device, already listening. vel_max / ang_max are pinned to 1.0 so it emits a deflection, not a velocity -- :func:scale_se2_deflection owns the units.

Source code in eden/utils/devices/se2.py

def make_se2_device(device: str, options: dict | None = None):
    """Build a normalized SE(2) input device by name.

    Parameters
    ----------
    device: str
        One of :data:`DEVICES`.
    options: dict | None
        Device options. ``layout`` ("ps4" | "xbox") picks the gamepad axis scaling;
        every other key reaches the device constructor.

    Returns
    -------
    DeviceBase
        The device, already listening. ``vel_max`` / ``ang_max`` are pinned to ``1.0``
        so it emits a deflection, not a velocity -- :func:`scale_se2_deflection` owns
        the units.
    """
    options = dict(options or {})
    if device not in DEVICES:
        raise ValueError(f"Unknown SE(2) device {device!r}; expected one of {list(DEVICES)}.")

    if device == "keyboard":
        # TeleopSE2Keyboard, not SE2Keyboard: see the module docstring for why a ramped
        # deflection is the wrong contract for a velocity command.
        from eden.utils.devices.keyboard import TeleopSE2Keyboard

        options.pop("layout", None)
        return TeleopSE2Keyboard(key_bindings=teleop_key_bindings(), vel_max=1.0, ang_max=1.0, **options)

    from eden.utils.devices.gamepad import (
        DEADZONE_PS4,
        DEADZONE_XBOX,
        OFFSET_PS4,
        OFFSET_XBOX,
        SCALE_PS4,
        SCALE_XBOX,
        SE2Gamepad,
    )

    layout = options.pop("layout", "ps4")
    if layout not in GAMEPAD_LAYOUTS:
        raise ValueError(f"Unknown gamepad layout {layout!r}; expected one of {list(GAMEPAD_LAYOUTS)}.")
    scale, offset, deadzone = {
        "ps4": (SCALE_PS4, OFFSET_PS4, DEADZONE_PS4),
        "xbox": (SCALE_XBOX, OFFSET_XBOX, DEADZONE_XBOX),
    }[layout]
    return SE2Gamepad(vel_max=1.0, ang_max=1.0, scale=scale, offset=offset, deadzone=deadzone, **options)

scale_se2_deflection

scale_se2_deflection(deflection: Tensor, low: Tensor, high: Tensor) -> Tensor

Map a normalized [-1, 1] deflection into [low, high] per axis.

Full deflection maps to the larger side of each axis and the result is clamped into the range, so an asymmetric range (e.g. (-2.0, 3.0)) reaches both ends and saturates early on the short side rather than overshooting it. Clamping is what bounds an operator to the velocities a policy was trained on, so it is applied to the device reading too -- a miscalibrated stick cannot deflect past 1.0 and out of the range that way.

Source code in eden/utils/devices/se2.py

def scale_se2_deflection(deflection: torch.Tensor, low: torch.Tensor, high: torch.Tensor) -> torch.Tensor:
    """Map a normalized ``[-1, 1]`` deflection into ``[low, high]`` per axis.

    Full deflection maps to the *larger* side of each axis and the result is clamped
    into the range, so an asymmetric range (e.g. ``(-2.0, 3.0)``) reaches both ends and
    saturates early on the short side rather than overshooting it. Clamping is what
    bounds an operator to the velocities a policy was trained on, so it is applied to
    the device reading too -- a miscalibrated stick cannot deflect past 1.0 and out of
    the range that way.
    """
    return torch.clamp(deflection.clamp(-1.0, 1.0) * torch.maximum(high.abs(), low.abs()), min=low, max=high)

Keyboard teleoperation device (SE2/SE3).

Attributes

IS_PYNPUT_INSTALLED module-attribute

IS_PYNPUT_INSTALLED = True

PYNPUT_IMPORT_ERROR module-attribute

PYNPUT_IMPORT_ERROR: Exception | None = None

Classes

KeyBinding dataclass

KeyBinding(index: int, value: int, reset_on_release: bool = True, is_toggle: bool = False)

Attributes

index instance-attribute

index: int

value instance-attribute

value: int

reset_on_release class-attribute instance-attribute

reset_on_release: bool = True

is_toggle class-attribute instance-attribute

is_toggle: bool = False

Methods:

Keyboard

Keyboard(key_bindings: dict[str, KeyBinding] = None)

Bases: DeviceBase

Keyboard device.

PARAMETER	DESCRIPTION
key_bindings	The key bindings for the keyboard. TYPE: dict[str, KeyBinding] DEFAULT: None

Source code in eden/utils/devices/keyboard.py

def __init__(self, key_bindings: dict[str, KeyBinding] = None):
    super().__init__()
    self._require_pynput()

    self.key_bindings = key_bindings or {
        "w": KeyBinding(0, +1),
        "s": KeyBinding(0, -1),
        "a": KeyBinding(1, +1),
        "d": KeyBinding(1, -1),
        "q": KeyBinding(2, +1),
        "e": KeyBinding(2, -1),
        Key.space: KeyBinding(3, 1, is_toggle=True),
    }

    if self.key_bindings:
        max_index = max(v.index for v in self.key_bindings.values())
        self._readings = torch.zeros(max_index + 1, dtype=torch.float32)
    else:
        self._readings = torch.zeros(0, dtype=torch.float32)
    self._is_toggle_mask = torch.zeros_like(self._readings, dtype=torch.bool)
    for key_binding in self.key_bindings.values():
        if key_binding.is_toggle:
            self._is_toggle_mask[key_binding.index] = True
    self.lock = threading.Lock()
    self.active_keys = set()

    self.listener = keyboard.Listener(on_press=self._on_press, on_release=self._on_release)
    self.listener.daemon = True
    self.listener.start()

Attributes

key_bindings instance-attribute

key_bindings = key_bindings or {'w': KeyBinding(0, +1), 's': KeyBinding(0, -1), 'a': KeyBinding(1, +1), 'd': KeyBinding(1, -1), 'q': KeyBinding(2, +1), 'e': KeyBinding(2, -1), Key.space: KeyBinding(3, 1, is_toggle=True)}

lock instance-attribute

lock = threading.Lock()

active_keys instance-attribute

active_keys = set()

listener instance-attribute

listener = keyboard.Listener(on_press=self._on_press, on_release=self._on_release)

Methods:

reset

reset() -> None

Source code in eden/utils/devices/keyboard.py

def reset(self) -> None:
    with self.lock:
        self.active_keys.clear()
        self._readings.fill_(0)

stop

stop() -> None

Stop listening and zero the readings. Idempotent.

Mirrors :meth:~eden.utils.devices.gamepad.Gamepad.stop so a caller can release either device without knowing which it holds.

Source code in eden/utils/devices/keyboard.py

def stop(self) -> None:
    """Stop listening and zero the readings. Idempotent.

    Mirrors :meth:`~eden.utils.devices.gamepad.Gamepad.stop` so a caller can
    release either device without knowing which it holds.
    """
    self.listener.stop()
    self.reset()

get_raw_data

get_raw_data() -> Tensor

Source code in eden/utils/devices/keyboard.py

def get_raw_data(self) -> torch.Tensor:
    return self._readings

get_command

get_command() -> Tensor

Source code in eden/utils/devices/keyboard.py

def get_command(self) -> torch.Tensor:
    return self._readings

SE2Keyboard

SE2Keyboard(key_bindings: dict[str, KeyBinding] = None, vel_max: float = 1.0, ang_max: float = 1.0)

Bases: Keyboard

A keyboard controller for sending SE(2) commands as velocity commands.

PARAMETER	DESCRIPTION
key_bindings	The key bindings for the keyboard. TYPE: dict[str, KeyBinding] DEFAULT: None
vel_max	The maximum value of the velocity commands. TYPE: float DEFAULT: 1.0
ang_max	The maximum value of the angular velocity commands. TYPE: float DEFAULT: 1.0
Command
Move
Move
Rotate

Source code in eden/utils/devices/keyboard.py

def __init__(
    self,
    key_bindings: dict[str, KeyBinding] = None,
    vel_max: float = 1.0,
    ang_max: float = 1.0,
):
    key_bindings = key_bindings or self.default_key_bindings()
    super().__init__(key_bindings)
    self.limit = torch.tensor(
        [
            vel_max,
        ]
        * 2
        + [
            ang_max,
        ]
        * 1,
        dtype=torch.float32,
    )

Attributes

limit instance-attribute

limit = torch.tensor([vel_max] * 2 + [ang_max] * 1, dtype=torch.float32)

Methods:

default_key_bindings classmethod

default_key_bindings()

Source code in eden/utils/devices/keyboard.py

@classmethod
def default_key_bindings(cls):
    return {
        "8": KeyBinding(0, +1),
        "2": KeyBinding(0, -1),
        "4": KeyBinding(1, +1),
        "6": KeyBinding(1, -1),
        "7": KeyBinding(2, +1),
        "9": KeyBinding(2, -1),
    }

wasd_key_bindings classmethod

wasd_key_bindings()

Source code in eden/utils/devices/keyboard.py

@classmethod
def wasd_key_bindings(cls):
    return {
        "w": KeyBinding(0, +1),
        "s": KeyBinding(0, -1),
        "a": KeyBinding(1, +1),
        "d": KeyBinding(1, -1),
        "q": KeyBinding(2, +1),
        "e": KeyBinding(2, -1),
    }

get_command

get_command() -> Tensor

Source code in eden/utils/devices/keyboard.py

def get_command(self) -> torch.Tensor:
    return torch.where(
        self._is_toggle_mask,
        self._readings,
        torch.tanh(self._readings) * self.limit,
    )

TeleopSE2Keyboard

TeleopSE2Keyboard(key_bindings: dict[str, KeyBinding] = None, vel_max: float = 1.0, ang_max: float = 1.0)

Bases: SE2Keyboard

:class:SE2Keyboard whose deflection saturates instead of ramping.

SE2Keyboard smooths its readings through tanh. Because :meth:Keyboard._on_press accumulates (+=) and the OS delivers key-repeat while a key is held, that makes the reading a ramp: one press is tanh(1) = 0.76, and full deflection is only approached as repeats pile up — so the commanded velocity depends on the machine's key-repeat rate and on how long the operator held the key. Fine for nudging a camera, wrong for a velocity command a robot acts on, where the same gesture has to mean the same speed everywhere.

Here a held key is full deflection from the first press, exactly like a gamepad stick pushed to its stop, and releasing still returns to zero (reset_on_release). The ceiling that deflection maps to is the trained range, which :func:~eden.utils.devices.se2.scale_se2_deflection owns.

Source code in eden/utils/devices/keyboard.py

def __init__(
    self,
    key_bindings: dict[str, KeyBinding] = None,
    vel_max: float = 1.0,
    ang_max: float = 1.0,
):
    key_bindings = key_bindings or self.default_key_bindings()
    super().__init__(key_bindings)
    self.limit = torch.tensor(
        [
            vel_max,
        ]
        * 2
        + [
            ang_max,
        ]
        * 1,
        dtype=torch.float32,
    )

Methods:

get_command

get_command() -> Tensor

Source code in eden/utils/devices/keyboard.py

def get_command(self) -> torch.Tensor:
    return torch.where(
        self._is_toggle_mask,
        self._readings,
        self._readings.clamp(-1.0, 1.0) * self.limit,
    )

SE3Keyboard

SE3Keyboard(key_bindings: dict[str, KeyBinding] = None, pos_max: float = 1.0, rot_max: float = 1.0)

Bases: Keyboard

A keyboard controller for sending SE(3) commands as delta poses and binary command (open/close).

PARAMETER	DESCRIPTION
key_bindings	The key bindings for the keyboard. TYPE: dict[str, KeyBinding] DEFAULT: None
pos_max	Gain on the three position axes. Defaults to 1.0, i.e. a normalized delta pose: the task-space controller downstream owns the conversion to metres via its position_scale. Lower it only to clamp the device below full deflection, and expect the physical step to be the product of the two. TYPE: float DEFAULT: 1.0
rot_max	Gain on the three rotation axes, on the same normalized contract as pos_max — the controller's orientation_scale owns radians. TYPE: float DEFAULT: 1.0

Key bindings

============================== ================= ================= Description Key (+ve axis) Key (-ve axis) ============================== ================= ================= Toggle gripper (open/close) Space Move along x-axis 8 2 Move along y-axis 4 6 Move along z-axis 7 9 Rotate along x-axis Up Down Rotate along y-axis Right Left Rotate along z-axis 1 3 ============================== ================= =================

Source code in eden/utils/devices/keyboard.py

def __init__(
    self,
    key_bindings: dict[str, KeyBinding] = None,
    pos_max: float = 1.0,
    rot_max: float = 1.0,
):
    key_bindings = key_bindings or self.default_key_bindings()
    super().__init__(key_bindings)
    self.limit = torch.tensor(
        [
            pos_max,
        ]
        * 3
        + [
            rot_max,
        ]
        * 3
        + [1.0],
        dtype=torch.float32,
    )

Attributes

limit instance-attribute

limit = torch.tensor([pos_max] * 3 + [rot_max] * 3 + [1.0], dtype=torch.float32)

Methods:

default_key_bindings classmethod

default_key_bindings()

Source code in eden/utils/devices/keyboard.py

@classmethod
def default_key_bindings(cls):
    cls._require_pynput()
    return {
        "8": KeyBinding(0, +1),
        "2": KeyBinding(0, -1),
        "4": KeyBinding(1, +1),
        "6": KeyBinding(1, -1),
        "7": KeyBinding(2, +1),
        "9": KeyBinding(2, -1),
        Key.up: KeyBinding(3, +1),
        Key.down: KeyBinding(3, -1),
        Key.right: KeyBinding(4, +1),
        Key.left: KeyBinding(4, -1),
        "1": KeyBinding(5, +1),
        "3": KeyBinding(5, -1),
        Key.space: KeyBinding(6, 0, is_toggle=True),
    }

wasd_key_bindings classmethod

wasd_key_bindings()

Source code in eden/utils/devices/keyboard.py

@classmethod
def wasd_key_bindings(cls):
    cls._require_pynput()
    return {
        "w": KeyBinding(0, +1),
        "s": KeyBinding(0, -1),
        "a": KeyBinding(1, +1),
        "d": KeyBinding(1, -1),
        "q": KeyBinding(2, +1),
        "e": KeyBinding(2, -1),
        Key.up: KeyBinding(3, +1),
        Key.down: KeyBinding(3, -1),
        Key.right: KeyBinding(4, +1),
        Key.left: KeyBinding(4, -1),
        "z": KeyBinding(5, +1),
        "c": KeyBinding(5, -1),
        Key.space: KeyBinding(6, 0, is_toggle=True),
    }

get_command

get_command() -> Tensor

Source code in eden/utils/devices/keyboard.py

def get_command(self) -> torch.Tensor:
    # NOTE: apply a hyperbolic tangent to the readings to smooth the keyboard reading rate
    return torch.where(
        self._is_toggle_mask,
        self._readings,
        torch.tanh(self._readings) * self.limit,
    )

Gamepad teleoperation device (SE2/SE3).

Attributes

IS_INPUTS_INSTALLED module-attribute

IS_INPUTS_INSTALLED = True

SCALE_PS4 module-attribute

SCALE_PS4 = 1.0 / 128

SCALE_XBOX module-attribute

SCALE_XBOX = 1.0 / 32768

OFFSET_PS4 module-attribute

OFFSET_PS4 = 1.0

OFFSET_XBOX module-attribute

OFFSET_XBOX = 0.0

DEADZONE_PS4 module-attribute

DEADZONE_PS4 = 0.005

DEADZONE_XBOX module-attribute

DEADZONE_XBOX = 0.005

gamepad module-attribute

gamepad = SE3Gamepad()

cmd module-attribute

cmd = gamepad.get_command()

Classes

GamepadBinding dataclass

GamepadBinding(index: int, scale: float = 1.0, offset: float = 0.0, deadzone: float = 0.0, is_toggle: bool = False)

Attributes

index instance-attribute

index: int

scale class-attribute instance-attribute

scale: float = 1.0

offset class-attribute instance-attribute

offset: float = 0.0

deadzone class-attribute instance-attribute

deadzone: float = 0.0

is_toggle class-attribute instance-attribute

is_toggle: bool = False

Methods:

Gamepad

Gamepad(bindings: dict[str, GamepadBinding] = None)

Bases: DeviceBase

Gamepad device.

PARAMETER	DESCRIPTION
bindings	The bindings for the gamepad. TYPE: dict[str, GamepadBinding] DEFAULT: None

Source code in eden/utils/devices/gamepad.py

def __init__(self, bindings: dict[str, GamepadBinding] = None):
    if not IS_INPUTS_INSTALLED:
        raise ImportError("inputs is not installed. Please install it with `pip install inputs`.")
    if not input_devices.gamepads:
        # ``get_gamepad`` raises per call when nothing is plugged in, and the reader
        # thread swallows that; without this check the device looks alive and reports a
        # steady zero command forever. Say so here instead, while the caller can act.
        raise RuntimeError(
            "No gamepad found. Connect one (and check it is readable, e.g. it appears in "
            "/dev/input) before constructing a Gamepad device."
        )

    super().__init__()
    self.bindings = bindings or {}
    self.connected = True

    # Determine size of readings vector based on max index in bindings
    if self.bindings:
        max_index = max(b.index for b in self.bindings.values())
        self._readings = torch.zeros(max_index + 1, dtype=torch.float32)
        self._is_toggle_mask = torch.zeros_like(self._readings, dtype=torch.bool)
        for binding in self.bindings.values():
            if binding.is_toggle:
                self._is_toggle_mask[binding.index] = True
    else:
        self._readings = torch.zeros(0, dtype=torch.float32)
        self._is_toggle_mask = torch.zeros(0, dtype=torch.bool)

    self.lock = Lock()
    self.running = True
    self.thread = Thread(target=self._read_inputs, daemon=True)
    self.thread.start()

Attributes

bindings instance-attribute

bindings = bindings or {}

connected instance-attribute

connected = True

lock instance-attribute

lock = Lock()

running instance-attribute

running = True

thread instance-attribute

thread = Thread(target=self._read_inputs, daemon=True)

Methods:

stop

stop()

Source code in eden/utils/devices/gamepad.py

def stop(self):
    self.running = False

reset

reset()

Source code in eden/utils/devices/gamepad.py

def reset(self):
    with self.lock:
        self._readings.fill_(0)

get_raw_data

get_raw_data() -> Tensor

Source code in eden/utils/devices/gamepad.py

def get_raw_data(self) -> torch.Tensor:
    with self.lock:
        return self._readings.clone()

get_command

get_command() -> Tensor

Source code in eden/utils/devices/gamepad.py

def get_command(self) -> torch.Tensor:
    return self.get_raw_data()

SE2Gamepad

SE2Gamepad(bindings: dict[str, GamepadBinding] = None, vel_max: float = 1.0, ang_max: float = 1.0, scale: float = SCALE_PS4, offset: float = OFFSET_PS4, deadzone: float = DEADZONE_PS4)

Bases: Gamepad

A gamepad controller for sending SE(2) commands as velocity commands.

Default bindings (inputs event codes; axis scaling defaults to a PS4 pad's 0-255 range, pass scale=SCALE_XBOX, offset=OFFSET_XBOX for an Xbox pad): - Left Stick Y: Linear Velocity X - Left Stick X: Linear Velocity Y - Right Stick X: Angular Velocity Z

Source code in eden/utils/devices/gamepad.py

def __init__(
    self,
    bindings: dict[str, GamepadBinding] = None,
    vel_max: float = 1.0,
    ang_max: float = 1.0,
    scale: float = SCALE_PS4,
    offset: float = OFFSET_PS4,
    deadzone: float = DEADZONE_PS4,
):
    bindings = bindings or self.default_bindings(scale, offset, deadzone)
    super().__init__(bindings)

    # Multipliers on the normalized axes, not clamps.
    self.vel_max = vel_max
    self.ang_max = ang_max

Attributes

vel_max instance-attribute

vel_max = vel_max

ang_max instance-attribute

ang_max = ang_max

Methods:

default_bindings classmethod

default_bindings(scale: float, offset: float = 0.0, deadzone: float = 0.0)

Source code in eden/utils/devices/gamepad.py

@classmethod
def default_bindings(cls, scale: float, offset: float = 0.0, deadzone: float = 0.0):
    return {
        "ABS_Y": GamepadBinding(0, scale=-scale, deadzone=deadzone, offset=offset),  # Y axis (inverted often)
        "ABS_X": GamepadBinding(1, scale=-scale, deadzone=deadzone, offset=offset),  # X axis
        "ABS_RX": GamepadBinding(2, scale=-scale, deadzone=deadzone, offset=offset),  # Angular Z
    }

get_command

get_command() -> Tensor

Source code in eden/utils/devices/gamepad.py

def get_command(self) -> torch.Tensor:
    readings = self.get_raw_data()
    cmd = torch.zeros(3, dtype=torch.float32)
    # Assuming index 0 is VX, 1 is VY, 2 is WZ based on default bindings
    if readings.shape[0] >= 3:
        cmd[0] = readings[0] * self.vel_max
        cmd[1] = readings[1] * self.vel_max
        cmd[2] = readings[2] * self.ang_max
    return cmd

SE3Gamepad

SE3Gamepad(bindings: dict[str, GamepadBinding] = None, pos_max: float = 1.0, rot_max: float = 1.0, scale: float = SCALE_PS4, offset: float = OFFSET_PS4, deadzone: float = DEADZONE_PS4)

Bases: Gamepad

A gamepad controller for sending SE(3) commands as delta poses and binary command (open/close).

Default bindings (inputs event codes; axis scaling defaults to a PS4 pad's 0-255 range, pass scale=SCALE_XBOX, offset=OFFSET_XBOX for an Xbox pad): - Left Stick Y: X axis - Left Stick X: Y axis - Right Stick Y: Z axis - Right Stick X: Yaw (Rot Z) - D-Pad Up/Down: Pitch (Rot Y) - D-Pad Left/Right: Roll (Rot X) - A Button: Toggle Gripper

pos_max and rot_max default to 1.0, i.e. this device emits a normalized delta pose and the task-space controller downstream owns the conversion to metres and radians via its position_scale / orientation_scale. Lower them only to clamp the sticks below full deflection, and expect the physical step to be the product of the two.

Source code in eden/utils/devices/gamepad.py

def __init__(
    self,
    bindings: dict[str, GamepadBinding] = None,
    pos_max: float = 1.0,
    rot_max: float = 1.0,
    scale: float = SCALE_PS4,
    offset: float = OFFSET_PS4,
    deadzone: float = DEADZONE_PS4,
):
    bindings = bindings or self.default_bindings(scale, offset, deadzone)
    super().__init__(bindings)
    self.pos_max = pos_max
    self.rot_max = rot_max

Attributes

pos_max instance-attribute

pos_max = pos_max

rot_max instance-attribute

rot_max = rot_max

Methods:

default_bindings classmethod

default_bindings(scale: float, offset: float = 0.0, deadzone: float = 0.0)

Source code in eden/utils/devices/gamepad.py

@classmethod
def default_bindings(cls, scale: float, offset: float = 0.0, deadzone: float = 0.0):
    return {
        "ABS_Y": GamepadBinding(0, scale=-scale, deadzone=deadzone, offset=offset),  # Tx
        "ABS_X": GamepadBinding(1, scale=-scale, deadzone=deadzone, offset=offset),  # Ty
        "ABS_RY": GamepadBinding(2, scale=-scale, deadzone=deadzone, offset=offset),  # Tz
        # Rotation
        "ABS_HAT0X": GamepadBinding(3, scale=1.0, deadzone=0.0),  # Rx (D-pad is -1/0/1 usually)
        "ABS_HAT0Y": GamepadBinding(4, scale=-1.0, deadzone=0.0),  # Ry
        "ABS_RX": GamepadBinding(5, scale=-scale, deadzone=deadzone, offset=offset),  # Rz
        # Gripper
        "BTN_SOUTH": GamepadBinding(6, is_toggle=True),  # A button
    }

get_command

get_command() -> Tensor

Source code in eden/utils/devices/gamepad.py

def get_command(self) -> torch.Tensor:
    readings = self.get_raw_data()
    cmd = torch.zeros(7, dtype=torch.float32)

    if readings.shape[0] >= 7:
        # Translation
        cmd[:3] = readings[:3] * self.pos_max
        # Rotation
        cmd[3:6] = readings[3:6] * self.rot_max
        # Gripper
        cmd[6] = readings[6]

    return cmd

SpaceMouse teleoperation device (SE2/SE3).

Attributes

IS_SPNAV_INSTALLED module-attribute

IS_SPNAV_INSTALLED = True

sm module-attribute

sm = Spacemouse(deadzone=0.3, is_toggle_btn0=True, is_toggle_btn1=False)

cmd module-attribute

cmd = sm.get_command()

btns module-attribute

btns = [sm.is_button_pressed(i) for i in range(2)]

Classes

Spacemouse

Spacemouse(pos_max: float = 1.0, rot_max: float = 1.0, deadzone: float | tuple = 0.3, polling_rate: int = 200, is_toggle_btn0: bool = False, is_toggle_btn1: bool = False)

Bases: DeviceBase

Spacemouse device.

PARAMETER	DESCRIPTION
pos_max	Gain on the three position axes. Defaults to 1.0, i.e. a normalized delta pose: the task-space controller downstream owns the conversion to metres via its position_scale. Lower it only to clamp the device below full deflection, and expect the physical step to be the product of the two. TYPE: float DEFAULT: 1.0
rot_max	Gain on the three rotation axes, on the same normalized contract as pos_max — the controller's orientation_scale owns radians. TYPE: float DEFAULT: 1.0
deadzone	The deadzone of the spacemouse reading. Default is 0.3. TYPE: float | tuple DEFAULT: 0.3
polling_rate	The polling rate of the spacemouse. Default is 200. TYPE: int DEFAULT: 200
is_toggle_btn0	Whether the first button is a toggle button. Default is False. TYPE: bool DEFAULT: False
is_toggle_btn1	Whether the second button is a toggle button. Default is False. TYPE: bool DEFAULT: False

Source code in eden/utils/devices/spacemouse.py

def __init__(
    self,
    pos_max: float = 1.0,
    rot_max: float = 1.0,
    deadzone: float | tuple = 0.3,
    polling_rate: int = 200,
    is_toggle_btn0: bool = False,
    is_toggle_btn1: bool = False,
):
    if not IS_SPNAV_INSTALLED:
        error_message = """
        spnav is not installed.
        Please install it with the following commands:
        ```bash
        sudo apt install libspnav-dev spacenavd
        sudo systemctl start spacenavd
        pip install https://github.com/cheng-chi/spnav/archive/c1c938ebe3cc542db4685e0d13850ff1abfdb943.tar.gz
        ```
        Then reboot the system before using the spacemouse.
        """
        raise ImportError(error_message)

    super().__init__()

    self.pos_max = pos_max
    self.rot_max = rot_max
    self.polling_rate = polling_rate
    self.is_toggle_btn0 = is_toggle_btn0
    self.is_toggle_btn1 = is_toggle_btn1

    if isinstance(deadzone, (int, float)):
        self.deadzone = np.full(6, fill_value=deadzone, dtype=np.float32)
    else:
        self.deadzone = np.array(deadzone, dtype=np.float32)
    assert (self.deadzone >= 0).all()

    self.lock = threading.Lock()
    self.stop_event = threading.Event()

    self.motion_event = SpnavMotionEvent([0, 0, 0], [0, 0, 0], 0)
    self.button_state = defaultdict(lambda: False)

    # Coordinate transform from Spacemouse to Eden (Z-up)
    self.tx_zup_spnav = torch.tensor([[0.0, 0.0, -1.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]], device="cpu")

    self.thread = threading.Thread(target=self._run_worker)
    self.thread.daemon = True
    self.thread.start()

Attributes

pos_max instance-attribute

pos_max = pos_max

rot_max instance-attribute

rot_max = rot_max

polling_rate instance-attribute

polling_rate = polling_rate

is_toggle_btn0 instance-attribute

is_toggle_btn0 = is_toggle_btn0

is_toggle_btn1 instance-attribute

is_toggle_btn1 = is_toggle_btn1

deadzone instance-attribute

deadzone = np.full(6, fill_value=deadzone, dtype=np.float32)

lock instance-attribute

lock = threading.Lock()

stop_event instance-attribute

stop_event = threading.Event()

motion_event instance-attribute

motion_event = SpnavMotionEvent([0, 0, 0], [0, 0, 0], 0)

button_state instance-attribute

button_state = defaultdict(lambda: False)

tx_zup_spnav instance-attribute

tx_zup_spnav = torch.tensor([[0.0, 0.0, -1.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]], device='cpu')

thread instance-attribute

thread = threading.Thread(target=self._run_worker)

Methods:

get_raw_data

get_raw_data() -> Tensor

Return the raw normalized state without coordinate transformation.

Source code in eden/utils/devices/spacemouse.py

def get_raw_data(self) -> torch.Tensor:
    """Return the raw normalized state without coordinate transformation."""
    with self.lock:
        me = self.motion_event
        state = np.array(me.translation + me.rotation, dtype=np.float32) / 500.0

    is_dead = (-self.deadzone < state) & (state < self.deadzone)
    state[is_dead] = 0.0
    return torch.from_numpy(state)

get_command

get_command() -> Tensor

Return the transformed command in Eden coordinate system (Z-up).

Source code in eden/utils/devices/spacemouse.py

def get_command(self) -> torch.Tensor:
    """Return the transformed command in Eden coordinate system (Z-up)."""
    state = self.get_raw_data()

    tf_state = torch.zeros_like(state)
    tf_state[:3] = (self.tx_zup_spnav @ state[:3]) * self.pos_max
    tf_state[3:] = (self.tx_zup_spnav @ state[3:]) * self.rot_max

    return tf_state

is_button_pressed

is_button_pressed(button_id: int) -> bool

Source code in eden/utils/devices/spacemouse.py

def is_button_pressed(self, button_id: int) -> bool:
    with self.lock:
        return self.button_state[button_id]

close

close()

Source code in eden/utils/devices/spacemouse.py

def close(self):
    self.stop_event.set()
    if self.thread.is_alive():
        self.thread.join(timeout=1.0)

SE2Spacemouse

SE2Spacemouse(pos_max: float = 0.1, rot_max: float = 0.1, **kwargs)

Bases: Spacemouse

A spacemouse controller for sending SE(2) commands as velocity commands.

Unlike the delta-pose devices, this one feeds a velocity command directly rather than a task-space controller, so it keeps physical units, capped by default at 0.1 m/s and 0.1 rad/s.

PARAMETER	DESCRIPTION
pos_max	The maximum linear velocity, in m/s. TYPE: float DEFAULT: 0.1
rot_max	The maximum angular velocity, in rad/s. TYPE: float DEFAULT: 0.1
**kwargs	Forwarded to :class:Spacemouse. DEFAULT: {}

Source code in eden/utils/devices/spacemouse.py

def __init__(self, pos_max: float = 0.1, rot_max: float = 0.1, **kwargs):
    super().__init__(pos_max=pos_max, rot_max=rot_max, **kwargs)

Methods:

get_command

get_command() -> Tensor

Source code in eden/utils/devices/spacemouse.py

def get_command(self) -> torch.Tensor:
    state = self.get_raw_data()

    tf_state = torch.zeros(3)
    tf_state[:2] = (self.tx_zup_spnav @ state[3:5]) * self.rot_max

    return tf_state

SE3Spacemouse

SE3Spacemouse(pos_max: float = 1.0, rot_max: float = 1.0, deadzone: float | tuple = 0.3, polling_rate: int = 200, is_toggle_btn0: bool = False, is_toggle_btn1: bool = False)

Bases: Spacemouse

A spacemouse controller for sending SE(3) commands as delta poses and binary command (open/close).

Source code in eden/utils/devices/spacemouse.py

def __init__(
    self,
    pos_max: float = 1.0,
    rot_max: float = 1.0,
    deadzone: float | tuple = 0.3,
    polling_rate: int = 200,
    is_toggle_btn0: bool = False,
    is_toggle_btn1: bool = False,
):
    if not IS_SPNAV_INSTALLED:
        error_message = """
        spnav is not installed.
        Please install it with the following commands:
        ```bash
        sudo apt install libspnav-dev spacenavd
        sudo systemctl start spacenavd
        pip install https://github.com/cheng-chi/spnav/archive/c1c938ebe3cc542db4685e0d13850ff1abfdb943.tar.gz
        ```
        Then reboot the system before using the spacemouse.
        """
        raise ImportError(error_message)

    super().__init__()

    self.pos_max = pos_max
    self.rot_max = rot_max
    self.polling_rate = polling_rate
    self.is_toggle_btn0 = is_toggle_btn0
    self.is_toggle_btn1 = is_toggle_btn1

    if isinstance(deadzone, (int, float)):
        self.deadzone = np.full(6, fill_value=deadzone, dtype=np.float32)
    else:
        self.deadzone = np.array(deadzone, dtype=np.float32)
    assert (self.deadzone >= 0).all()

    self.lock = threading.Lock()
    self.stop_event = threading.Event()

    self.motion_event = SpnavMotionEvent([0, 0, 0], [0, 0, 0], 0)
    self.button_state = defaultdict(lambda: False)

    # Coordinate transform from Spacemouse to Eden (Z-up)
    self.tx_zup_spnav = torch.tensor([[0.0, 0.0, -1.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]], device="cpu")

    self.thread = threading.Thread(target=self._run_worker)
    self.thread.daemon = True
    self.thread.start()

Methods:

get_command

get_command() -> Tensor

Return the transformed command in Eden coordinate system (Z-up).

Source code in eden/utils/devices/spacemouse.py

def get_command(self) -> torch.Tensor:
    """Return the transformed command in Eden coordinate system (Z-up)."""
    state = self.get_raw_data()

    tf_state = torch.zeros(7)
    tf_state[:3] = (self.tx_zup_spnav @ state[:3]) * self.pos_max
    tf_state[3:6] = (self.tx_zup_spnav @ state[3:6]) * self.rot_max
    tf_state[6] = 1.0 if self.is_button_pressed(0) else 0.0

    return tf_state

ImGui slider overlay for manual DOF teleoperation.

Classes

SliderGUI

SliderGUI(viewer: Viewer, dofs_name: list[str], dofs_pos_limits: Tensor, reset_callback: Callable[[], None] | None = None, initial_position: Tensor | None = None, plugin: ImGuiOverlayPlugin | None = None)

Bases: DeviceBase

ImGui-based slider device for teleoperation.

Attaches an :class:~genesis.ext.pyrender.overlay.ImGuiOverlayPlugin to the Genesis viewer and registers a custom side panel with one slider per named DoF plus a Reset button. The current slider values are polled via :meth:get_command, shape (1, n_dofs).

PARAMETER	DESCRIPTION
viewer	The viewer to attach the overlay to, typically env.scene.viewer. The viewer must already be built — the plugin is registered immediately. TYPE: Viewer
dofs_name	Labels for each slider (used as the ImGui label text). TYPE: list[str]
dofs_pos_limits	Tensor of shape (n_dofs, 2) with [lo, hi] for each DoF. ±inf is clipped to ±pi to keep sliders bounded. TYPE: Tensor
reset_callback	Optional callback invoked on the viewer thread when the Reset button is pressed. TYPE: Callable[[], None] | None DEFAULT: None
initial_position	Initial slider values (also used as the Reset target). If None, defaults to zero clipped into dofs_pos_limits. TYPE: Tensor | None DEFAULT: None
plugin	Existing overlay plugin to register the panel on. If None, the viewer's first :class:ImGuiOverlayPlugin is reused, and a new one is added only when it has none, so several devices share one overlay. TYPE: ImGuiOverlayPlugin | None DEFAULT: None

Source code in eden/utils/devices/gui_slider.py

def __init__(
    self,
    viewer: Viewer,
    dofs_name: list[str],
    dofs_pos_limits: torch.Tensor,
    reset_callback: Callable[[], None] | None = None,
    initial_position: torch.Tensor | None = None,
    plugin: ImGuiOverlayPlugin | None = None,
):
    from genesis.ext.pyrender.overlay import ImGuiOverlayPlugin

    super().__init__()

    self.dofs_name = list(dofs_name)
    limits = dofs_pos_limits.detach().cpu().numpy().astype(np.float32, copy=True)
    limits[limits == -np.inf] = -np.pi
    limits[limits == +np.inf] = +np.pi
    self.dofs_pos_limits = limits

    if initial_position is None:
        defaults = np.zeros(len(self.dofs_name), dtype=np.float32)
    else:
        defaults = initial_position.detach().cpu().numpy().astype(np.float32, copy=True)
    self._defaults = np.clip(defaults, limits[:, 0], limits[:, 1])
    self._readings = self._defaults.copy()
    self._reset_callback = reset_callback
    self._viewer = viewer

    if plugin is None:
        plugin = next(
            (p for p in viewer.plugins if isinstance(p, ImGuiOverlayPlugin)),
            None,
        )
    if plugin is None:
        plugin = viewer.add_plugin(ImGuiOverlayPlugin())
    self._plugin = plugin
    self._plugin.register_panel(self._draw_panel, section="side")
    en.logger.info("SliderGUI ImGui panel registered.")

Attributes

dofs_name instance-attribute

dofs_name = list(dofs_name)

dofs_pos_limits instance-attribute

dofs_pos_limits = limits

Methods:

get_raw_data

get_raw_data() -> Tensor

Source code in eden/utils/devices/gui_slider.py

def get_raw_data(self) -> torch.Tensor:
    return torch.from_numpy(self._readings.copy()).unsqueeze(0)

get_command

get_command() -> Tensor

Source code in eden/utils/devices/gui_slider.py

def get_command(self) -> torch.Tensor:
    return torch.from_numpy(self._readings.copy()).unsqueeze(0)

set_command

set_command(values: Tensor | ndarray) -> None

Overwrite the slider readings programmatically (clipped into the slider limits).

For syncing the panel to an externally-known pose — e.g. a real-robot teleop keeping the sliders glued to the measured joint positions until control is handed over, so the first commanded target equals the current pose instead of jumping to wherever the sliders were left.

PARAMETER	DESCRIPTION
values	New slider values, shape (n_dofs,) or (1, n_dofs). TYPE: Tensor | ndarray

Source code in eden/utils/devices/gui_slider.py

def set_command(self, values: torch.Tensor | np.ndarray) -> None:
    """Overwrite the slider readings programmatically (clipped into the slider limits).

    For syncing the panel to an externally-known pose — e.g. a real-robot
    teleop keeping the sliders glued to the measured joint positions until
    control is handed over, so the first commanded target equals the
    current pose instead of jumping to wherever the sliders were left.

    Parameters
    ----------
    values: torch.Tensor | np.ndarray
        New slider values, shape ``(n_dofs,)`` or ``(1, n_dofs)``.
    """
    if isinstance(values, torch.Tensor):
        values = values.detach().cpu().numpy()
    arr = np.asarray(values, dtype=np.float32).reshape(-1)
    if arr.shape[0] != len(self.dofs_name):
        raise ValueError(f"set_command expects {len(self.dofs_name)} values, got {arr.shape[0]}")
    self._readings[:] = np.clip(arr, self.dofs_pos_limits[:, 0], self.dofs_pos_limits[:, 1])

close

close() -> None

Source code in eden/utils/devices/gui_slider.py

def close(self) -> None:
    # The ImGui plugin lifecycle is owned by the viewer; nothing to clean up here.
    pass

GELLO leader arm

GELLO leader-arm runtime and configuration.

Attributes

Classes

GelloLeaderConfig dataclass

GelloLeaderConfig(joint_ids: Sequence[int], joint_offsets_rad: Sequence[float], joint_signs: Sequence[int], gripper_id: int | None = None, gripper_open_deg: float | None = None, gripper_close_deg: float | None = None, baudrate: int = 57600)

Calibration config for a single-arm Gello leader.

ATTRIBUTE	DESCRIPTION
joint_ids	Dynamixel IDs for the arm DOFs, in controller order. TYPE: Sequence[int]
joint_offsets_rad	Offset (rad) subtracted from each raw joint reading. Typically multiples of pi/2. TYPE: Sequence[float]
joint_signs	+1 or -1 per arm DOF to match the follower's positive rotation direction. TYPE: Sequence[int]
gripper_id	(Optional) Dynamixel ID for the gripper DOF. TYPE: int | None
gripper_open_deg	(Optional) Gripper position in degrees when fully open. TYPE: float | None
gripper_close_deg	(Optional) Gripper position in degrees when fully closed. TYPE: float | None
baudrate	Serial baudrate for Dynamixel communication. TYPE: int

Attributes

joint_ids instance-attribute

joint_ids: Sequence[int]

joint_offsets_rad instance-attribute

joint_offsets_rad: Sequence[float]

joint_signs instance-attribute

joint_signs: Sequence[int]

gripper_id class-attribute instance-attribute

gripper_id: int | None = None

gripper_open_deg class-attribute instance-attribute

gripper_open_deg: float | None = None

gripper_close_deg class-attribute instance-attribute

gripper_close_deg: float | None = None

baudrate class-attribute instance-attribute

baudrate: int = 57600

has_gripper property

has_gripper: bool

num_arm_dofs property

num_arm_dofs: int

num_dofs property

num_dofs: int

all_ids property

all_ids: list[int]

Methods:

to_dict

to_dict() -> dict[str, Any]

Source code in eden/utils/devices/gello/leader.py

def to_dict(self) -> dict[str, Any]:
    data = asdict(self)
    data["joint_ids"] = [int(v) for v in self.joint_ids]
    data["joint_offsets_rad"] = [float(v) for v in self.joint_offsets_rad]
    data["joint_signs"] = [int(v) for v in self.joint_signs]
    return data

save

save(path: str | Path) -> Path

Save config to path. Format is inferred from the extension (.json, .yaml, .yml).

Source code in eden/utils/devices/gello/leader.py

def save(self, path: str | Path) -> Path:
    """Save config to ``path``. Format is inferred from the extension (.json, .yaml, .yml)."""
    path = Path(path)
    text = _dump_config(self.to_dict(), path.suffix.lower())
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")
    return path

from_dict classmethod

from_dict(data: dict[str, Any]) -> GelloLeaderConfig

Source code in eden/utils/devices/gello/leader.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> GelloLeaderConfig:
    return cls(**data)

load classmethod

load(path: str | Path) -> GelloLeaderConfig

Load config from path. Format is inferred from the extension (.json, .yaml, .yml).

Source code in eden/utils/devices/gello/leader.py

@classmethod
def load(cls, path: str | Path) -> GelloLeaderConfig:
    """Load config from ``path``. Format is inferred from the extension (.json, .yaml, .yml)."""
    path = Path(path)
    return cls.from_dict(_load_config(path.read_text(encoding="utf-8"), path.suffix.lower()))

GelloLeader

GelloLeader(port: str, config: GelloLeaderConfig, start_dofs: ndarray | Sequence[float] | None = None)

Gello leader agent: reads calibrated joint positions from Dynamixel hardware.

The arm DOFs are returned in radians after applying (raw - offset) * sign. When a gripper is configured, the gripper DOF is normalized to [0, 1] with 0 == open and 1 == closed.

Source code in eden/utils/devices/gello/leader.py

def __init__(
    self,
    port: str,
    config: GelloLeaderConfig,
    start_dofs: np.ndarray | Sequence[float] | None = None,
):
    self._config = config
    self._driver = DynamixelDriver(ids=config.all_ids, port=port, baudrate=config.baudrate)
    self._driver.set_torque_mode(False)

    arm_offsets = np.asarray(config.joint_offsets_rad, dtype=np.float64)
    arm_signs = np.asarray(config.joint_signs, dtype=np.float64)
    if config.has_gripper:
        self._offsets = np.concatenate([arm_offsets, np.zeros(1)])
        self._signs = np.concatenate([arm_signs, np.ones(1)])
        self._gripper_range_rad = (
            float(np.deg2rad(config.gripper_open_deg)),
            float(np.deg2rad(config.gripper_close_deg)),
        )
    else:
        self._offsets = arm_offsets
        self._signs = arm_signs
        self._gripper_range_rad = None

    if start_dofs is not None:
        self._align_offsets_to_start(np.asarray(start_dofs, dtype=np.float64).reshape(-1))

Attributes

num_dofs property

num_dofs: int

config property

config: GelloLeaderConfig

Methods:

get_command

get_command() -> ndarray

Return calibrated arm DOFs (rad) concatenated with a normalized gripper value (if any).

Source code in eden/utils/devices/gello/leader.py

def get_command(self) -> np.ndarray:
    """Return calibrated arm DOFs (rad) concatenated with a normalized gripper value (if any)."""
    raw = self._driver.get_positions()
    dofs = (raw - self._offsets) * self._signs
    if self._gripper_range_rad is not None:
        dofs[-1] = self._normalize_gripper(dofs[-1])
    return dofs

close

close()

Source code in eden/utils/devices/gello/leader.py

def close(self):
    self._driver.close()

Functions:

run_gello_leader

run_gello_leader(argv: list[str] | None = None)

Source code in eden/utils/devices/gello/leader.py

def run_gello_leader(argv: list[str] | None = None):
    from argparse import ArgumentParser
    from pathlib import Path

    from eden.utils.devices.gello.zmq_client import GelloClient

    description = (
        f"{colorize('Gello leader arm teleoperation tool.', ANSI_BOLD, ANSI_CYAN)}\n"
        "\n"
        "Run this tool to continuously read calibrated joint positions from Dynamixel hardware\n"
        "and send them to a GelloZMQServer over ZMQ. The arm DOFs are returned in radians after\n"
        "applying (raw - offset) * sign. When a gripper is configured, the gripper DOF is\n"
        "normalized to [0, 1] with 0 == open and 1 == closed."
    )
    epilog = (
        f"{colorize('Examples:', ANSI_BOLD, ANSI_YELLOW)}\n"
        "  eden gello leader \\\n"
        f"      {colorize('-c', ANSI_GREEN)} configs/gello_ur5e.yaml \\\n"
        f"      {colorize('--port', ANSI_GREEN)} /dev/ttyUSB0 \\\n"
        f"      {colorize('--zmq-host', ANSI_GREEN)} 127.0.0.1 \\\n"
        f"      {colorize('--zmq-port', ANSI_GREEN)} 6001"
    )

    parser = ArgumentParser(
        prog="eden gello leader",
        description=description,
        epilog=epilog,
        formatter_class=ColorHelpFormatter,
    )
    parser.add_argument(
        "-c",
        "--config",
        type=Path,
        required=True,
        help="Path to the GelloLeaderConfig file. First run 'eden gello calibrate' to create a config file.",
    )
    parser.add_argument("--port", type=str, required=True, help="Dynamixel serial port (e.g. /dev/ttyUSB0).")
    parser.add_argument("--zmq-host", type=str, help="ZMQ host (e.g. 127.0.0.1).", default="127.0.0.1")
    parser.add_argument("--zmq-port", type=int, help="ZMQ port (e.g. 6001).", default=6001)

    args = parser.parse_args(argv)

    config = GelloLeaderConfig.load(args.config)
    leader = GelloLeader(port=args.port, config=config, start_dofs=np.zeros(config.num_dofs))
    client = GelloClient(host=args.zmq_host, port=args.zmq_port)

    assert client.num_dofs() == leader.num_dofs

    print(f"{colorize('Gello leader arm teleoperation tool started.', ANSI_BOLD, ANSI_CYAN)}")

    try:
        while True:
            client.command_joint_state(leader.get_command())
            time.sleep(1 / 100)  # 100 Hz
    finally:
        leader.close()
        client.close()
        print(f"{colorize('Gello leader arm teleoperation tool stopped.', ANSI_BOLD, ANSI_CYAN)}")

GELLO leader-arm joint-offset calibration.

Attributes

Classes

Functions:

get_offset

get_offset(current_dofs: Sequence[float] | ndarray, start_dofs: Sequence[float] | ndarray, signs: Sequence[float] | ndarray, max_pi_multiple: int = 8) -> ndarray

Search per-joint offsets (multiples of pi/2) that align current DOFs with known start DOFs.

For each joint i, picks the offset that minimizes abs(signs[i] * (current_dofs[i] - offset) - start_dofs[i]) over offset in [-max_pi_multiple * pi, +max_pi_multiple * pi] stepped by pi/2.

Source code in eden/utils/devices/gello/calibration.py

def get_offset(
    current_dofs: Sequence[float] | np.ndarray,
    start_dofs: Sequence[float] | np.ndarray,
    signs: Sequence[float] | np.ndarray,
    max_pi_multiple: int = 8,
) -> np.ndarray:
    """Search per-joint offsets (multiples of pi/2) that align current DOFs with known start DOFs.

    For each joint i, picks the offset that minimizes
    ``abs(signs[i] * (current_dofs[i] - offset) - start_dofs[i])``
    over ``offset`` in ``[-max_pi_multiple * pi, +max_pi_multiple * pi]`` stepped by pi/2.
    """
    if max_pi_multiple <= 0:
        raise ValueError("'max_pi_multiple' must be greater than 0.")

    current_arr = np.asarray(current_dofs, dtype=np.float64).reshape(-1)
    start_arr = np.asarray(start_dofs, dtype=np.float64).reshape(-1)
    sign_arr = np.asarray(signs, dtype=np.float64).reshape(-1)

    if start_arr.size == 0:
        raise ValueError("'start_dofs' must not be empty.")
    if start_arr.size != sign_arr.size or start_arr.size != current_arr.size:
        raise ValueError("'current_dofs', 'start_dofs', and 'signs' must have the same length.")
    if not np.all(np.isin(sign_arr, (-1.0, 1.0))):
        raise ValueError("'signs' must only contain -1 or 1.")

    candidates = np.linspace(
        -max_pi_multiple * np.pi,
        max_pi_multiple * np.pi,
        max_pi_multiple * 4 + 1,
    )

    errors = np.abs(sign_arr[:, None] * (current_arr[:, None] - candidates[None, :]) - start_arr[:, None])
    return candidates[np.argmin(errors, axis=1)]

format_offsets_as_pi_half_terms

format_offsets_as_pi_half_terms(offsets: Sequence[float] | ndarray) -> list[str]

Return each offset formatted as k*np.pi/2.

Source code in eden/utils/devices/gello/calibration.py

def format_offsets_as_pi_half_terms(offsets: Sequence[float] | np.ndarray) -> list[str]:
    """Return each offset formatted as ``k*np.pi/2``."""
    arr = np.asarray(offsets, dtype=np.float64).reshape(-1)
    return [f"{int(np.round(v / (np.pi / 2.0)))}*np.pi/2" for v in arr]

estimate_gripper_range_deg

estimate_gripper_range_deg(current_gripper_rad: float, open_delta_deg: float = -0.2, close_delta_deg: float = -42.0) -> tuple[float, float]

Estimate gripper open/close positions (degrees) from the current resting position.

The gripper is assumed to be held near its open position during calibration.

Source code in eden/utils/devices/gello/calibration.py

def estimate_gripper_range_deg(
    current_gripper_rad: float,
    open_delta_deg: float = -0.2,
    close_delta_deg: float = -42.0,
) -> tuple[float, float]:
    """Estimate gripper open/close positions (degrees) from the current resting position.

    The gripper is assumed to be held near its open position during calibration.
    """
    current_deg = float(np.rad2deg(float(current_gripper_rad)))
    return current_deg + float(open_delta_deg), current_deg + float(close_delta_deg)

calibrate_gello_leader

calibrate_gello_leader(argv: list[str] | None = None)

Source code in eden/utils/devices/gello/calibration.py

def calibrate_gello_leader(argv: list[str] | None = None):
    from argparse import ArgumentParser
    from pathlib import Path

    from eden.utils.devices.gello.dynamixel_driver import DynamixelDriver
    from eden.utils.devices.gello.leader import GelloLeaderConfig

    description = (
        f"{colorize('Gello leader arm calibration tool.', ANSI_BOLD, ANSI_CYAN)}\n"
        "\n"
        "Hold the leader arm at a known joint configuration (passed via "
        f"{colorize('--start-dofs', ANSI_GREEN)},\n"
        "in radians), then run this tool to compute per-joint offsets and a gripper\n"
        "open/close range. The result is printed as a ready-to-paste\n"
        f"{colorize('GelloLeaderConfig(...)', ANSI_MAGENTA)}."
    )
    epilog = (
        f"{colorize('Examples:', ANSI_BOLD, ANSI_YELLOW)}\n"
        f"  {colorize('# 6-DoF arm (IDs 1..6) at zero pose, with a gripper on ID 7 (default N+1):', ANSI_DIM)}\n"
        "  eden gello calibrate \\\n"
        f"      {colorize('--port', ANSI_GREEN)} /dev/ttyUSB0 \\\n"
        f"      {colorize('--start-dofs', ANSI_GREEN)} 0 0 0 0 0 0 \\\n"
        f"      {colorize('--signs', ANSI_GREEN)} 1 1 -1 1 1 1 \\\n"
        f"      {colorize('--gripper', ANSI_GREEN)}\\\n"
        f"      {colorize('--output', ANSI_GREEN)} configs/gello_ur5e.yaml\n"
        "\n"
        f"  {colorize('# Custom Dynamixel IDs and high-speed baudrate, no gripper:', ANSI_DIM)}\n"
        "  eden gello calibrate \\\n"
        f"      {colorize('--port', ANSI_GREEN)} /dev/ttyUSB0 \\\n"
        f"      {colorize('--start-dofs', ANSI_GREEN)} 0 0 0 0 0 0 \\\n"
        f"      {colorize('--signs', ANSI_GREEN)} 1 1 1 1 1 1 \\\n"
        f"      {colorize('--joint-ids', ANSI_GREEN)} 1 2 4 6 7 8 \\\n"
        f"      {colorize('--baudrate', ANSI_GREEN)} 1000000"
    )
    parser = ArgumentParser(
        prog="eden gello calibrate",
        description=description,
        epilog=epilog,
        formatter_class=ColorHelpFormatter,
    )
    parser.add_argument("--port", type=str, required=True, help="Dynamixel serial port (e.g. /dev/ttyUSB0).")
    parser.add_argument(
        "--start-dofs", type=float, nargs="+", required=True, help="Known arm DOFs (rad) at calibration pose."
    )
    parser.add_argument("--signs", type=float, nargs="+", required=True, help="Sign (-1 or 1) per arm DOF.")
    parser.add_argument(
        "--joint-ids", type=int, nargs="+", default=None, help="Dynamixel IDs for arm DOFs. Default: 1..N."
    )
    parser.add_argument("--gripper", action="store_true", help="Whether the gripper is attached.")
    parser.add_argument("--gripper-id", type=int, default=None, help="Dynamixel ID for the gripper. Default: N+1.")
    parser.add_argument("--baudrate", type=int, default=57600)
    parser.add_argument("--warmup-reads", type=int, default=10, help="Read iterations before sampling offsets.")
    parser.add_argument("--max-pi-multiple", type=int, default=8)
    parser.add_argument("--gripper-open-delta-deg", type=float, default=-0.2)
    parser.add_argument("--gripper-close-delta-deg", type=float, default=-42.0)
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        default=None,
        help="Save the resulting GelloLeaderConfig to this path (format inferred from .json, .yaml, or .yml).",
    )
    args = parser.parse_args(argv)

    if args.output is not None and args.output.suffix.lower() not in (".json", ".yaml", ".yml"):
        parser.error(f"--output must end in .json, .yaml, or .yml (got {args.output.suffix!r}).")

    start_dofs = np.asarray(args.start_dofs, dtype=np.float64)
    signs = np.asarray(args.signs, dtype=np.float64)
    if start_dofs.size != signs.size:
        raise ValueError("'--start-dofs' and '--signs' must have the same length.")

    num_arm = start_dofs.size
    joint_ids = args.joint_ids if args.joint_ids is not None else list(range(1, num_arm + 1))
    if len(joint_ids) != num_arm:
        raise ValueError(f"'--joint-ids' must have length {num_arm}, got {len(joint_ids)}.")

    ids = list(joint_ids)
    if args.gripper:
        gripper_id = int(args.gripper_id) if args.gripper_id is not None else num_arm + 1
        if gripper_id in ids:
            raise ValueError("'--gripper-id' must not overlap with arm joint IDs.")
        ids.append(gripper_id)

    driver = DynamixelDriver(ids=ids, port=args.port, baudrate=args.baudrate)
    try:
        for _ in range(max(0, args.warmup_reads)):
            driver.get_positions()

        positions = driver.get_positions()
        arm_positions = positions[:num_arm]
        offsets = get_offset(
            current_dofs=arm_positions,
            start_dofs=start_dofs,
            signs=signs,
            max_pi_multiple=args.max_pi_multiple,
        )

        def _label(text: str) -> str:
            return colorize(text, ANSI_CYAN)

        def _value(text: str) -> str:
            return colorize(text, ANSI_YELLOW)

        def _key(text: str) -> str:
            return colorize(text, ANSI_GREEN)

        print(f"{_label('current arm dofs (rad)    :')} {_value(str([f'{v:.6f}' for v in arm_positions.tolist()]))}")
        print(f"{_label('best offsets (rad)        :')} {_value(str([f'{v:.6f}' for v in offsets.tolist()]))}")
        print(f"{_label('best offsets (pi/2 terms) :')} {_value(str(format_offsets_as_pi_half_terms(offsets)))}")

        gripper_block = ""
        if args.gripper:
            open_deg, close_deg = estimate_gripper_range_deg(
                current_gripper_rad=float(positions[-1]),
                open_delta_deg=args.gripper_open_delta_deg,
                close_delta_deg=args.gripper_close_delta_deg,
            )
            print(f"{_label('gripper current (deg)     :')} {_value(f'{np.rad2deg(positions[-1]):.6f}')}")
            print(f"{_label('gripper open (deg)        :')} {_value(f'{open_deg:.6f}')}")
            print(f"{_label('gripper close (deg)       :')} {_value(f'{close_deg:.6f}')}")
            gripper_block = (
                f"    {_key('gripper_id')}={_value(str(gripper_id))},\n"
                f"    {_key('gripper_open_deg')}={_value(f'{open_deg:.6f}')},\n"
                f"    {_key('gripper_close_deg')}={_value(f'{close_deg:.6f}')},\n"
            )

        config = GelloLeaderConfig(
            joint_ids=[int(v) for v in joint_ids],
            joint_offsets_rad=[float(v) for v in offsets],
            joint_signs=[int(s) for s in signs],
            gripper_id=gripper_id if args.gripper else None,
            gripper_open_deg=(open_deg if args.gripper else None),
            gripper_close_deg=(close_deg if args.gripper else None),
            baudrate=args.baudrate,
        )

        print(f"\n{colorize('GelloLeaderConfig(', ANSI_BOLD, ANSI_MAGENTA)}")
        print(f"    {_key('joint_ids')}={_value(str(tuple(config.joint_ids)))},")
        print(f"    {_key('joint_offsets_rad')}={_value(str(tuple(config.joint_offsets_rad)))},")
        print(f"    {_key('joint_signs')}={_value(str(tuple(config.joint_signs)))},")
        if gripper_block:
            print(gripper_block, end="")
        print(f"    {_key('baudrate')}={_value(str(config.baudrate))},")
        print(colorize(")", ANSI_BOLD, ANSI_MAGENTA))

        if args.output is not None:
            saved = config.save(args.output)
            print(f"\n{colorize('Saved config to:', ANSI_BOLD, ANSI_GREEN)} {saved}")
    finally:
        driver.close()

Dynamixel servo driver for the GELLO leader arm.

Attributes

ADDR_TORQUE_ENABLE module-attribute

ADDR_TORQUE_ENABLE = 64

ADDR_GOAL_POSITION module-attribute

ADDR_GOAL_POSITION = 116

LEN_GOAL_POSITION module-attribute

LEN_GOAL_POSITION = 4

ADDR_PRESENT_POSITION module-attribute

ADDR_PRESENT_POSITION = 132

LEN_PRESENT_POSITION module-attribute

LEN_PRESENT_POSITION = 4

TORQUE_ENABLE module-attribute

TORQUE_ENABLE = 1

TORQUE_DISABLE module-attribute

TORQUE_DISABLE = 0

Classes

DynamixelDriver

DynamixelDriver(ids: Sequence[int], port: str, baudrate: int = 57600)

Minimal Dynamixel position driver for Gello leader arms.

Reads joint positions (rad) and, when torque is enabled, commands goal positions.

Source code in eden/utils/devices/gello/dynamixel_driver.py

def __init__(self, ids: Sequence[int], port: str, baudrate: int = 57600):
    if len(ids) == 0:
        raise ValueError("'ids' must not be empty.")

    try:
        from dynamixel_sdk.group_sync_read import GroupSyncRead
        from dynamixel_sdk.group_sync_write import GroupSyncWrite
        from dynamixel_sdk.packet_handler import PacketHandler
        from dynamixel_sdk.port_handler import PortHandler
        from dynamixel_sdk.robotis_def import COMM_SUCCESS
        from dynamixel_sdk.robotis_def import DXL_HIBYTE, DXL_HIWORD, DXL_LOBYTE, DXL_LOWORD
    except ImportError as exc:
        raise ImportError("dynamixel_sdk is required. Install with `pip install dynamixel-sdk`.") from exc

    self._ids = tuple(int(i) for i in ids)
    self._port_handler = PortHandler(port)
    self._packet_handler = PacketHandler(2.0)
    self._group_read = GroupSyncRead(
        self._port_handler, self._packet_handler, ADDR_PRESENT_POSITION, LEN_PRESENT_POSITION
    )
    self._group_write = GroupSyncWrite(
        self._port_handler, self._packet_handler, ADDR_GOAL_POSITION, LEN_GOAL_POSITION
    )
    self._comm_success = COMM_SUCCESS
    self._byte_pack = (DXL_LOBYTE, DXL_HIBYTE, DXL_LOWORD, DXL_HIWORD)
    self._torque_enabled = False

    if not self._port_handler.openPort():
        raise RuntimeError(f"Failed to open Dynamixel port: {port}")
    if not self._port_handler.setBaudRate(int(baudrate)):
        raise RuntimeError(f"Failed to set Dynamixel baudrate: {baudrate}")

    for dxl_id in self._ids:
        if not self._group_read.addParam(dxl_id):
            raise RuntimeError(f"Failed to add group read param for ID {dxl_id}")

Attributes

ids property

ids: tuple[int, ...]

Methods:

set_torque_mode

set_torque_mode(enable: bool)

Source code in eden/utils/devices/gello/dynamixel_driver.py

def set_torque_mode(self, enable: bool):
    target = TORQUE_ENABLE if enable else TORQUE_DISABLE
    for dxl_id in self._ids:
        comm_result, error = self._packet_handler.write1ByteTxRx(
            self._port_handler, dxl_id, ADDR_TORQUE_ENABLE, target
        )
        self._raise_if_comm_failed(comm_result, error, f"set torque mode for ID {dxl_id}")
    self._torque_enabled = bool(enable)

get_positions

get_positions() -> ndarray

Source code in eden/utils/devices/gello/dynamixel_driver.py

def get_positions(self) -> np.ndarray:
    comm_result = self._group_read.txRxPacket()
    if comm_result != self._comm_success:
        raise RuntimeError(f"Failed to read joint positions: {self._packet_handler.getTxRxResult(comm_result)}")

    positions = np.zeros(len(self._ids), dtype=np.float64)
    for i, dxl_id in enumerate(self._ids):
        if not self._group_read.isAvailable(dxl_id, ADDR_PRESENT_POSITION, LEN_PRESENT_POSITION):
            raise RuntimeError(f"Position data is not available for ID {dxl_id}")
        raw = self._group_read.getData(dxl_id, ADDR_PRESENT_POSITION, LEN_PRESENT_POSITION)
        signed = raw - (1 << 32) if raw & (1 << 31) else raw
        positions[i] = signed * np.pi / 2048.0
    return positions

set_positions

set_positions(joint_angles: Sequence[float])

Source code in eden/utils/devices/gello/dynamixel_driver.py

def set_positions(self, joint_angles: Sequence[float]):
    if not self._torque_enabled:
        raise RuntimeError("Torque must be enabled before commanding joint positions.")

    joint_angles_arr = np.asarray(joint_angles, dtype=np.float64).reshape(-1)
    if joint_angles_arr.size != len(self._ids):
        raise ValueError(f"Expected {len(self._ids)} joint angles, got {joint_angles_arr.size}.")

    lo_byte, hi_byte, lo_word, hi_word = self._byte_pack
    try:
        for dxl_id, angle in zip(self._ids, joint_angles_arr):
            value = int(np.round(angle * 2048.0 / np.pi)) & 0xFFFFFFFF
            data = [
                lo_byte(lo_word(value)),
                hi_byte(lo_word(value)),
                lo_byte(hi_word(value)),
                hi_byte(hi_word(value)),
            ]
            if not self._group_write.addParam(dxl_id, data):
                raise RuntimeError(f"Failed to add write param for ID {dxl_id}")
        comm_result = self._group_write.txPacket()
    finally:
        self._group_write.clearParam()

    if comm_result != self._comm_success:
        raise RuntimeError(f"Failed to write joint positions: {self._packet_handler.getTxRxResult(comm_result)}")

close

close()

Source code in eden/utils/devices/gello/dynamixel_driver.py

def close(self):
    handler = getattr(self, "_port_handler", None)
    if handler is not None:
        handler.closePort()

ZMQ server bridging the GELLO leader arm to the sim follower.

Attributes

ZMQ_AVAILABLE module-attribute

ZMQ_AVAILABLE = True

Classes

GelloZMQServer

GelloZMQServer(env: EnvBase, arm_entity_name: str, arm_dofs_name: list[str], ee_link_name: str, *, gripper_entity_name: str | None = None, gripper_dofs_name: list[str] | None = None, host: str = 'localhost', port: int = 5001, verbose: bool = False)

Bases: DeviceBase

ZMQ RPC server exposing a sim arm (and optional two-DOF gripper) as a GELLO follower.

A :class:~eden.utils.devices.gello.zmq_client.GelloClient (e.g. eden gello leader) sends command_joint_state requests, which :meth:get_command returns; the observation methods read the follower's state from env.

PARAMETER	DESCRIPTION
env	The environment to use. TYPE: EnvBase
arm_entity_name	The name of the arm entity. TYPE: str
arm_dofs_name	The names of the arm DOFs. TYPE: list[str]
ee_link_name	The name of the end-effector link. TYPE: str
gripper_entity_name	(Optional) The name of the gripper entity. TYPE: str | None DEFAULT: None
gripper_dofs_name	(Optional) The names of the gripper DOFs. TYPE: list[str] | None DEFAULT: None
host	The host to use. TYPE: str DEFAULT: 'localhost'
port	The port to use. TYPE: int DEFAULT: 5001
verbose	Whether to log verbose output. TYPE: bool DEFAULT: False

Source code in eden/utils/devices/gello/zmq_server.py

def __init__(
    self,
    env: EnvBase,
    arm_entity_name: str,
    arm_dofs_name: list[str],
    ee_link_name: str,
    *,
    gripper_entity_name: str | None = None,
    gripper_dofs_name: list[str] | None = None,
    host: str = "localhost",
    port: int = 5001,
    verbose: bool = False,
):
    if not ZMQ_AVAILABLE:
        raise ImportError(
            "pyzmq (import name `zmq`) is not installed. "
            f"Install it with Eden's 'gello' extra, {install_hint('gello')}; or alone with `pip install pyzmq`."
        )

    self._host = host
    self._port = port
    self._verbose = verbose

    self._env = env
    self._arm_entity: RigidEntity = env.entities[arm_entity_name]
    _, self._arm_dofs_idx = self._arm_entity.find_named_dofs_idx_local(arm_dofs_name, preserve_order=True)
    self._ee_link: RigidLink = self._arm_entity.get_link(ee_link_name)
    if gripper_entity_name is not None:
        self._gripper_entity: RigidEntity = env.entities[gripper_entity_name]
        gripper_dofs_name, self._gripper_dofs_idx = self._gripper_entity.find_named_dofs_idx_local(
            gripper_dofs_name, preserve_order=True
        )
        assert len(self._gripper_dofs_idx) == 2, "Gripper must have exactly 2 DOFs."

        gripper_open = torch.zeros(2)
        gripper_close = torch.zeros(2)
        open_dofs_pos = getattr(self._gripper_entity._options, "open_dofs_pos", None)
        close_dofs_pos = getattr(self._gripper_entity._options, "close_dofs_pos", None)

        if isinstance(open_dofs_pos, float):
            gripper_open[:] = open_dofs_pos
        elif isinstance(open_dofs_pos, dict):
            for dofs_name, pos in open_dofs_pos.items():
                # Robot-level configs may include both left/right gripper keys. Skip entries that do not
                # match this server's gripper scope, as decided by the lookup itself (an exact joint name is
                # literal against every joint, which a separate probe over the scope alone would not see).
                try:
                    _dofs_name, dofs_idx = self._gripper_entity.find_named_dofs_idx_local(
                        dofs_name, name_scope=gripper_dofs_name
                    )
                except ValueError:
                    continue
                dofs_idx = [gripper_dofs_name.index(name) for name in _dofs_name]
                gripper_open[dofs_idx] = pos
        else:
            raise ValueError(f"Unsupported open_dofs_pos type: {type(open_dofs_pos)}.")

        if isinstance(close_dofs_pos, float):
            gripper_close[:] = close_dofs_pos
        elif isinstance(close_dofs_pos, dict):
            for dofs_name, pos in close_dofs_pos.items():
                # Robot-level configs may include both left/right gripper keys. Skip entries that do not
                # match this server's gripper scope, as decided by the lookup itself (an exact joint name is
                # literal against every joint, which a separate probe over the scope alone would not see).
                try:
                    _dofs_name, dofs_idx = self._gripper_entity.find_named_dofs_idx_local(
                        dofs_name, name_scope=gripper_dofs_name
                    )
                except ValueError:
                    continue
                dofs_idx = [gripper_dofs_name.index(name) for name in _dofs_name]
                gripper_close[dofs_idx] = pos
        else:
            raise ValueError(f"Unsupported close_dofs_pos type: {type(close_dofs_pos)}.")

        self._gripper_open = gripper_open
        self.max_opening = (gripper_close - gripper_open).abs().sum(dim=-1)
    else:
        self._gripper_entity = None

    self._num_dofs = len(self._arm_dofs_idx) + int(self._gripper_entity is not None)

    self._lifecycle_lock = threading.Lock()
    self._stop_event = threading.Event()
    self._server_ready = threading.Event()
    self._server_startup_done = threading.Event()
    self._server_thread: threading.Thread | None = None
    self._server_startup_error: Exception | None = None
    self._server_state: _ServerState = _ServerState.IDLE
    self._command_lock = threading.Lock()
    self._command = (
        np.asarray(self._get_observations_for_rpc(dofs_pos_only=True), dtype=np.float32).reshape(-1).copy()
    )

Attributes

max_opening instance-attribute

max_opening = (gripper_close - gripper_open).abs().sum(dim=-1)

host property

host: str

port property

port: int

endpoint property

endpoint: str

Methods:

start

start()

Source code in eden/utils/devices/gello/zmq_server.py

def start(self):
    startup_timeout_seconds = 10.0

    with self._lifecycle_lock:
        if self._server_state == _ServerState.STOPPED:
            raise RuntimeError("Gello RPC server has already been stopped and cannot be restarted.")

        if self._server_state == _ServerState.RUNNING:
            thread = self._server_thread
            if thread is not None and thread.is_alive() and self._server_ready.is_set():
                return
            self._server_state = _ServerState.IDLE
            self._server_thread = None

        if self._server_state == _ServerState.IDLE:
            self._server_startup_error = None
            self._server_startup_done.clear()
            self._server_ready.clear()
            self._stop_event.clear()
            self._server_state = _ServerState.STARTING

            thread = threading.Thread(target=self._serve, daemon=True)
            try:
                thread.start()
            except Exception:
                self._server_state = _ServerState.IDLE
                raise
            self._server_thread = thread

    if not self._server_startup_done.wait(timeout=startup_timeout_seconds):
        raise RuntimeError(
            f"Gello RPC server startup timed out after {startup_timeout_seconds} seconds "
            f"for endpoint {self.endpoint!r}."
        )

    with self._lifecycle_lock:
        if self._server_state == _ServerState.STOPPED:
            raise RuntimeError("Gello RPC server was stopped during startup.")
        if self._server_startup_error is not None:
            self._server_state = _ServerState.IDLE
            self._server_thread = None
            raise RuntimeError(
                f"Failed to start Gello RPC server on {self.endpoint}."
            ) from self._server_startup_error
        if not self._server_ready.is_set():
            self._server_state = _ServerState.IDLE
            self._server_thread = None
            raise RuntimeError("Gello RPC server thread exited before becoming ready.")
        self._server_state = _ServerState.RUNNING

stop

stop()

Source code in eden/utils/devices/gello/zmq_server.py

def stop(self):
    with self._lifecycle_lock:
        if self._server_state == _ServerState.STOPPED:
            return
        self._server_state = _ServerState.STOPPED
        self._stop_event.set()
        thread = self._server_thread

    self._wake_server_for_shutdown()

    if thread is not None and thread.is_alive():
        thread.join(timeout=2.0)

    self._server_ready.clear()
    self._server_startup_done.set()

get_raw_data

get_raw_data() -> Tensor

Source code in eden/utils/devices/gello/zmq_server.py

def get_raw_data(self) -> torch.Tensor:
    with self._command_lock:
        return torch.from_numpy(self._command.copy())

get_command

get_command() -> Tensor

Source code in eden/utils/devices/gello/zmq_server.py

def get_command(self) -> torch.Tensor:
    return self.get_raw_data()

Functions:

ZMQ client for the GELLO teleoperation server.

Attributes

ZMQ_AVAILABLE module-attribute

ZMQ_AVAILABLE = True

Classes

GelloClient

GelloClient(host: str = '127.0.0.1', port: int = 6001, timeout_ms: int = 2000)

Minimal ZMQ REQ client for :class:GelloZMQServer.

Exposes the four RPC methods served by the Gello follower server: num_dofs, get_joint_positions, get_observations, and command_joint_state.

Source code in eden/utils/devices/gello/zmq_client.py

def __init__(self, host: str = "127.0.0.1", port: int = 6001, timeout_ms: int = 2000):
    if not ZMQ_AVAILABLE:
        raise ImportError("pyzmq is not installed. Install it with `pip install pyzmq`.")
    if timeout_ms <= 0:
        raise ValueError("'timeout_ms' must be greater than 0.")

    self._context = zmq.Context()
    self._socket = self._context.socket(zmq.REQ)
    self._socket.setsockopt(zmq.LINGER, 0)
    self._socket.setsockopt(zmq.RCVTIMEO, int(timeout_ms))
    self._socket.setsockopt(zmq.SNDTIMEO, int(timeout_ms))
    self._socket.connect(f"tcp://{host}:{port}")

Methods:

num_dofs

num_dofs() -> int

Source code in eden/utils/devices/gello/zmq_client.py

def num_dofs(self) -> int:
    return int(self._request({"method": "num_dofs"}))

get_joint_positions

get_joint_positions() -> ndarray

Source code in eden/utils/devices/gello/zmq_client.py

def get_joint_positions(self) -> np.ndarray:
    result = self._request({"method": "get_joint_positions"})
    return np.asarray(result, dtype=np.float64).reshape(-1)

get_observations

get_observations() -> dict[str, ndarray]

Source code in eden/utils/devices/gello/zmq_client.py

def get_observations(self) -> dict[str, np.ndarray]:
    return self._request({"method": "get_observations"})

command_joint_state

command_joint_state(joint_state: Sequence[float] | ndarray) -> None

Source code in eden/utils/devices/gello/zmq_client.py

def command_joint_state(self, joint_state: Sequence[float] | np.ndarray) -> None:
    arr = np.asarray(joint_state, dtype=np.float32).reshape(-1)
    self._request({"method": "command_joint_state", "args": {"joint_state": arr}})

close

close()

Source code in eden/utils/devices/gello/zmq_client.py

def close(self):
    socket = getattr(self, "_socket", None)
    if socket is not None:
        socket.close(0)
        self._socket = None
    context = getattr(self, "_context", None)
    if context is not None:
        context.term()
        self._context = None

Meta Quest

A WebXR bridge process (televuer) streams controller poses to Eden over ZMQ.

XR bridge process for Meta Quest teleoperation.

Owns the televuer XR session: it spins up a Vuer WebXR server that the Meta Quest connects to over Wi-Fi, captures head / wrist poses plus either the hand-joint poses or the motion-controller state, and displays the stereo frames that Eden renders.

WebXR exposes hand tracking and motion controllers as mutually exclusive input sources, so the bridge runs in one of two modes (--tracking): * hand : head/wrist poses + 25 hand-joint positions + pinch/squeeze * controller : head/wrist poses + per-hand trigger / grip / A,B buttons / thumbstick (button + 2D axis)

Data flow (see protocol.py): * publishes head/wrist + hand-or-controller state -> Eden (PUB on POSE_PORT) * receives stereo RGB frames <- Eden (PULL on FRAME_PORT) -> headset

Each published dict carries a "mode" key ("hand" or "controller") so the Eden side knows which fields to expect.

All published poses are converted from the OpenXR basis (x right, y up, z back) into the Genesis/robot basis (x forward, y left, z up) so Eden can consume them directly in its world frame.

Requires the metaquest extra and televuer, which is not on PyPI (uv sync --inexact --extra metaquest --group metaquest-git in a checkout).

Run: eden metaquest bridge # hand tracking (default) eden metaquest bridge --tracking controller # motion controllers # or, equivalently: uv run python -m eden.utils.devices.metaquest.xr_bridge

Functions:

to_robot_se3

to_robot_se3(m_xr)

OpenXR-basis 4x4 pose -> robot/Genesis-basis 4x4 (z up, x forward).

Source code in eden/utils/devices/metaquest/xr_bridge.py

def to_robot_se3(m_xr):
    """OpenXR-basis 4x4 pose -> robot/Genesis-basis 4x4 (z up, x forward)."""
    return T_ROBOT_OPENXR @ np.asarray(m_xr) @ T_OPENXR_ROBOT

to_robot_points

to_robot_points(p_xr)

OpenXR-basis (N, 3) points -> robot/Genesis-basis (N, 3).

Source code in eden/utils/devices/metaquest/xr_bridge.py

def to_robot_points(p_xr):
    """OpenXR-basis (N, 3) points -> robot/Genesis-basis (N, 3)."""
    return (R_ROBOT_OPENXR @ np.asarray(p_xr).T).T

build_pose_msg

build_pose_msg(tv: TeleVuer)

Snapshot the current hand-tracking XR state into a JSON dict (robot frame).

Source code in eden/utils/devices/metaquest/xr_bridge.py

def build_pose_msg(tv: TeleVuer):
    """Snapshot the current hand-tracking XR state into a JSON dict (robot frame)."""
    msg = {
        "mode": "hand",
        "ready": bool(tv.motion_data_ready),
        "head": to_robot_se3(tv.head_pose).tolist(),
        "left_wrist": to_robot_se3(tv.left_arm_pose).tolist(),
        "right_wrist": to_robot_se3(tv.right_arm_pose).tolist(),
        "left_hand_pos": to_robot_points(tv.left_hand_positions).tolist(),
        "right_hand_pos": to_robot_points(tv.right_hand_positions).tolist(),
        # pinchValue is a distance (~0 when pinching); expose both the bool and value.
        "left_pinch": bool(tv.left_hand_pinch),
        "right_pinch": bool(tv.right_hand_pinch),
        "left_pinch_value": float(tv.left_hand_pinchValue),
        "right_pinch_value": float(tv.right_hand_pinchValue),
        "left_squeeze": float(tv.left_hand_squeezeValue),
        "right_squeeze": float(tv.right_hand_squeezeValue),
    }
    return msg

build_controller_msg

build_controller_msg(tv: TeleVuer)

Snapshot the current motion-controller XR state into a JSON dict (robot frame).

left_wrist / right_wrist are the controller grip poses (already in the robot frame). Button / trigger / grip booleans and analog values plus the thumbstick click + 2D axis come straight from the WebXR gamepad mapping:

• *_trigger_value / *_squeeze_value: 0.0 (released) -> 1.0 (fully pressed)
• *_thumbstick_value: (x, y) in [-1, 1], +x right, +y back (see televuer)

Source code in eden/utils/devices/metaquest/xr_bridge.py

def build_controller_msg(tv: TeleVuer):
    """Snapshot the current motion-controller XR state into a JSON dict (robot frame).

    ``left_wrist`` / ``right_wrist`` are the controller grip poses (already in the
    robot frame). Button / trigger / grip booleans and analog values plus the
    thumbstick click + 2D axis come straight from the WebXR gamepad mapping:

      * ``*_trigger_value`` / ``*_squeeze_value``: 0.0 (released) -> 1.0 (fully pressed)
      * ``*_thumbstick_value``: (x, y) in [-1, 1], +x right, +y back (see televuer)
    """
    msg = {
        "mode": "controller",
        "ready": bool(tv.motion_data_ready),
        "head": to_robot_se3(tv.head_pose).tolist(),
        "left_wrist": to_robot_se3(tv.left_arm_pose).tolist(),
        "right_wrist": to_robot_se3(tv.right_arm_pose).tolist(),
    }
    for side in ("left", "right"):
        msg.update(
            {
                f"{side}_trigger": bool(getattr(tv, f"{side}_ctrl_trigger")),
                f"{side}_trigger_value": float(getattr(tv, f"{side}_ctrl_triggerValue")),
                f"{side}_squeeze": bool(getattr(tv, f"{side}_ctrl_squeeze")),
                f"{side}_squeeze_value": float(getattr(tv, f"{side}_ctrl_squeezeValue")),
                f"{side}_a_button": bool(getattr(tv, f"{side}_ctrl_aButton")),
                f"{side}_b_button": bool(getattr(tv, f"{side}_ctrl_bButton")),
                f"{side}_thumbstick": bool(getattr(tv, f"{side}_ctrl_thumbstick")),
                f"{side}_thumbstick_value": [float(x) for x in getattr(tv, f"{side}_ctrl_thumbstickValue")],
            }
        )
    return msg

main

main(argv=None)

Run the XR bridge loop (parses argv or sys.argv when None).

Source code in eden/utils/devices/metaquest/xr_bridge.py

def main(argv=None):
    """Run the XR bridge loop (parses ``argv`` or ``sys.argv`` when ``None``)."""
    ap = argparse.ArgumentParser(description="Eden <-> Quest XR bridge (televuer side)")
    ap.add_argument("--width", type=int, default=1280, help="full stereo frame width (both eyes)")
    ap.add_argument("--height", type=int, default=480, help="frame height")
    ap.add_argument("--fps", type=float, default=30.0, help="headset display fps")
    ap.add_argument("--display-mode", default="immersive", choices=["immersive", "ego", "pass-through"])
    ap.add_argument(
        "--tracking",
        default="hand",
        choices=["hand", "controller"],
        help="XR input source: hand joints (default) or motion controllers (mutually exclusive in WebXR)",
    )
    args = ap.parse_args(argv)

    use_hand_tracking = args.tracking == "hand"
    build_msg = build_pose_msg if use_hand_tracking else build_controller_msg

    # The bridge runs televuer, not Genesis, so en.init() (and its en.logger) never
    # runs here -- stand up Eden's logger directly.
    logger = getattr(en, "logger", None) or en.Logger("INFO")

    tv = TeleVuer(
        use_hand_tracking=use_hand_tracking,
        binocular=True,
        img_shape=(args.height, args.width),
        display_fps=args.fps,
        display_mode=args.display_mode,
        zmq=True,  # in-process shared-memory frame display (no external image server)
        webrtc=False,
    )

    ctx = zmq.Context.instance()
    frame_sock = P.frame_pull(ctx)
    pose_sock = P.pose_pub(ctx)

    # Eden-styled banner: ~<...>~ mint, ~~<...>~~ mint-underline, ~~~~<...>~~~~ bold.
    rule = "━" * 64
    logger.info(f"~<{rule}>~")
    logger.info(f" 🍏 ~~~~<Meta Quest XR bridge>~~~~  ({args.tracking} tracking)  —  waiting for the headset")
    logger.info(" On the Quest: join this host's Wi-Fi, then open in the browser ~<→>~")
    for ip in list_host_ips() or ["<host-ip>"]:
        logger.info(f"     ~~<https://{ip}:8012/?ws=wss://{ip}:8012>~~")
    logger.info(" Tap ~<Virtual Reality>~ and allow all prompts")
    logger.info(" (first run: accept the self-signed cert — Advanced ~<→>~ Proceed)")
    logger.info(f" ZMQ   poses PUB ~<:{P.POSE_PORT}>~   frames PULL ~<:{P.FRAME_PORT}>~")
    logger.info(f"~<{rule}>~")

    running = {"v": True}

    def stop(*_):
        running["v"] = False

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    period = 1.0 / max(args.fps, 1.0)
    frames_shown = 0
    last_log = time.time()
    try:
        while running["v"]:
            loop_start = time.time()

            # Newest frame from Eden -> headset. Eden sends RGB; render_to_xr treats
            # its input as BGR (it does BGR2RGB internally), so flip channels here.
            frame = P.recv_frame(frame_sock, flags=zmq.NOBLOCK)
            if frame is not None:
                tv.render_to_xr(np.ascontiguousarray(frame[:, :, ::-1]))
                frames_shown += 1

            # Publish current XR pose state to Eden.
            P.send_pose(pose_sock, build_msg(tv))

            now = time.time()
            if now - last_log > 5.0:
                state = "tracking" if tv.motion_data_ready else "waiting for headset"
                logger.info(f"{state} | frames shown: {frames_shown}")
                last_log = now

            dt = time.time() - loop_start
            if dt < period:
                time.sleep(period - dt)
    finally:
        logger.info("shutting down...")
        tv.close()

Wire protocol for the Eden <-> XR (televuer) ZMQ bridge.

Two localhost channels, both single-part so ZMQ_CONFLATE (keep-latest) works:

• frames : Eden PUSH -> bridge PULL (FRAME_PORT) stereo RGB uint8 (H, W, 3)
• poses : bridge PUB -> Eden SUB (POSE_PORT) head/wrist/hand JSON state

This module is imported by both processes (the XR bridge and Eden), so it must depend only on numpy + pyzmq -- never on televuer or eden.

Pose matrices published here are already expressed in the Genesis/robot world frame: z up, x forward, y left, metres, origin at the XR tracking origin.

Attributes

FRAME_PORT module-attribute

FRAME_PORT = 5556

POSE_PORT module-attribute

POSE_PORT = 5557

Functions:

send_frame

send_frame(sock, rgb)

Send one stereo RGB frame. rgb is (H, W, 3) uint8 (left|right halves).

Source code in eden/utils/devices/metaquest/protocol.py

def send_frame(sock, rgb):
    """Send one stereo RGB frame. ``rgb`` is (H, W, 3) uint8 (left|right halves)."""
    rgb = np.ascontiguousarray(rgb, dtype=np.uint8)
    h, w = rgb.shape[:2]
    header = np.array([h, w], dtype=np.uint32).tobytes()
    sock.send(header + rgb.tobytes(), copy=False)

recv_frame

recv_frame(sock, flags=0)

Receive one frame, or return None if flags=zmq.NOBLOCK and none waiting.

Source code in eden/utils/devices/metaquest/protocol.py

def recv_frame(sock, flags=0):
    """Receive one frame, or return None if ``flags=zmq.NOBLOCK`` and none waiting."""
    try:
        buf = sock.recv(flags=flags)
    except zmq.Again:
        return None
    h, w = np.frombuffer(buf[:8], dtype=np.uint32)
    return np.frombuffer(buf[8:], dtype=np.uint8).reshape(int(h), int(w), 3)

send_pose

send_pose(sock, data: dict)

Publish one pose state dict as JSON.

Source code in eden/utils/devices/metaquest/protocol.py

def send_pose(sock, data: dict):
    """Publish one pose state dict as JSON."""
    sock.send_string(json.dumps(data))

recv_pose

recv_pose(sock, flags=0)

Receive the latest pose dict, or None if flags=zmq.NOBLOCK and none waiting.

Source code in eden/utils/devices/metaquest/protocol.py

def recv_pose(sock, flags=0):
    """Receive the latest pose dict, or None if ``flags=zmq.NOBLOCK`` and none waiting."""
    try:
        return json.loads(sock.recv_string(flags=flags))
    except zmq.Again:
        return None

frame_pull

frame_pull(ctx, port=FRAME_PORT)

Bridge side: PULL that keeps only the newest frame.

Source code in eden/utils/devices/metaquest/protocol.py

def frame_pull(ctx, port=FRAME_PORT):
    """Bridge side: PULL that keeps only the newest frame."""
    s = ctx.socket(zmq.PULL)
    s.setsockopt(zmq.CONFLATE, 1)
    s.setsockopt(zmq.RCVHWM, 1)
    s.bind(f"tcp://*:{port}")
    return s

frame_push

frame_push(ctx, host='127.0.0.1', port=FRAME_PORT)

Eden side: PUSH that drops frames rather than blocking when the bridge lags.

Source code in eden/utils/devices/metaquest/protocol.py

def frame_push(ctx, host="127.0.0.1", port=FRAME_PORT):
    """Eden side: PUSH that drops frames rather than blocking when the bridge lags."""
    s = ctx.socket(zmq.PUSH)
    s.setsockopt(zmq.SNDHWM, 1)
    s.setsockopt(zmq.CONFLATE, 1)
    s.connect(f"tcp://{host}:{port}")
    return s

pose_pub

pose_pub(ctx, port=POSE_PORT)

Bridge side: PUB of pose state.

Source code in eden/utils/devices/metaquest/protocol.py

def pose_pub(ctx, port=POSE_PORT):
    """Bridge side: PUB of pose state."""
    s = ctx.socket(zmq.PUB)
    s.bind(f"tcp://*:{port}")
    return s

pose_sub

pose_sub(ctx, host='127.0.0.1', port=POSE_PORT)

Eden side: SUB that keeps only the newest pose.

Source code in eden/utils/devices/metaquest/protocol.py

def pose_sub(ctx, host="127.0.0.1", port=POSE_PORT):
    """Eden side: SUB that keeps only the newest pose."""
    s = ctx.socket(zmq.SUB)
    s.setsockopt(zmq.CONFLATE, 1)
    s.setsockopt_string(zmq.SUBSCRIBE, "")
    s.connect(f"tcp://{host}:{port}")
    return s

Clutch-relative pose mapping: XR controller pose -> end-effector target.

Ported from Dream-Machines-Robotics/vr-teleop-kit (Apache-2.0), with the upstream mujoco quaternion calls swapped for Genesis' numpy-backed wxyz helpers (genesis.utils.geom) — no quaternion math is reimplemented here. The algorithm and its hardware-tuned defaults are unchanged; see tests/utils/devices/metaquest/test_pose_mapping.py for the ported sanity checks (T1-T11).

ClutchPoseMapper holds the "engage" state — the controller pose and the EE pose captured at the moment the operator pressed the clutch (grip) button. While the clutch is held, the EE target is the engaged EE pose composed with an effective controller delta, rotated from the XR world frame into the arm's base frame. On release the mapper disengages and target() returns None until the next engage, so the operator can reposition their hand without moving the robot (clutch-relative mapping, like lifting a mouse off the desk).

Reach limit (the interesting part). When the caller passes the arm's current EE pose into target(), the position/orientation channels become INCREMENTAL: per-tick controller deltas accumulate into an effective delta, and anything past pos_reach_limit / rot_reach_limit of the current EE pose is absorbed (a "slipping clutch"). This kills two failure modes the upstream authors hit on hardware:

• an absolute orientation map pushed far past a joint stop builds up the ~180° error where the shortest-way direction flips sign -> cap-speed shaking;
• 350° CW and 10° CCW are the same orientation, so an absolute map eventually "snaps" the wrist around the other way. The incremental one never can.

The absorbed overshoot does not come back when the operator reverses (mouse at the edge of the screen: reversing bites immediately). Trade-off: with scale_rotation != 1 the per-increment gain is a rate gain, so curved hand paths are path-dependent on SO(3); re-clutching realigns. 0/None on either reach limit restores the legacy absolute-delta-since-engage mapping.

Frames: all quaternions are wxyz. R maps XR-world vectors into the arm base (v_armbase = R @ v_xr); for an Eden bridge that already publishes poses in the Genesis/robot world frame, R is identity.

Classes

ClutchPoseMapper dataclass

ClutchPoseMapper(R: ndarray = (lambda: eye(3))(), scale: float = 1.0, scale_rotation: float = 1.0, rotation_pivot: ndarray | None = None, rot_reach_limit: float | None = 0.6, pos_reach_limit: float | None = 0.25)

One-handed clutch-relative controller -> EE mapping.

Attributes

R class-attribute instance-attribute

R: ndarray = field(default_factory=lambda: np.eye(3))

scale class-attribute instance-attribute

scale: float = 1.0

scale_rotation class-attribute instance-attribute

scale_rotation: float = 1.0

rotation_pivot class-attribute instance-attribute

rotation_pivot: ndarray | None = None

rot_reach_limit class-attribute instance-attribute

rot_reach_limit: float | None = 0.6

pos_reach_limit class-attribute instance-attribute

pos_reach_limit: float | None = 0.25

engaged property

engaged: bool

Methods:

set_R

set_R(R: ndarray) -> None

Replace the XR->arm-base rotation. Typically called per engage with a yaw-corrected R so the operator can turn their body between engagements and still have 'controller forward' mean 'robot forward'.

Source code in eden/utils/devices/metaquest/pose_mapping.py

def set_R(self, R: np.ndarray) -> None:
    """Replace the XR->arm-base rotation. Typically called per engage with a
    yaw-corrected R so the operator can turn their body between engagements and
    still have 'controller forward' mean 'robot forward'.
    """
    self.R = np.asarray(R, dtype=float).copy()
    self._R_quat = R_to_quat(self.R)
    self._R_quat_conj = inv_quat(self._R_quat)

engage

engage(controller_pos_xr: ndarray, controller_quat_xr: ndarray, ee_pos_armbase: ndarray, ee_quat_armbase: ndarray, pivot_armbase: ndarray | None = None) -> None

Capture the engage frame. Call on the rising clutch edge. If pivot_armbase is supplied, rotation deltas during this engagement pivot around that point instead of around the EE.

Source code in eden/utils/devices/metaquest/pose_mapping.py

def engage(
    self,
    controller_pos_xr: np.ndarray,
    controller_quat_xr: np.ndarray,
    ee_pos_armbase: np.ndarray,
    ee_quat_armbase: np.ndarray,
    pivot_armbase: np.ndarray | None = None,
) -> None:
    """Capture the engage frame. Call on the rising clutch edge. If
    ``pivot_armbase`` is supplied, rotation deltas during this engagement
    pivot around that point instead of around the EE.
    """
    self._ctrl_engage_pos = np.array(controller_pos_xr, float, copy=True)
    self._ctrl_engage_quat = np.array(controller_quat_xr, float, copy=True)
    self._ee_engage_pos = np.array(ee_pos_armbase, float, copy=True)
    self._ee_engage_quat = np.array(ee_quat_armbase, float, copy=True)
    self.rotation_pivot = None if pivot_armbase is None else np.array(pivot_armbase, float, copy=True)
    self._ctrl_prev_quat = self._ctrl_engage_quat.copy()
    self._ctrl_prev_pos = self._ctrl_engage_pos.copy()
    self._d_quat_eff = np.array([1.0, 0.0, 0.0, 0.0])
    self._d_pos_eff = np.zeros(3)
    self._engaged = True

disengage

disengage() -> None

Source code in eden/utils/devices/metaquest/pose_mapping.py

def disengage(self) -> None:
    self._engaged = False

target

target(controller_pos_xr: ndarray, controller_quat_xr: ndarray, ee_pos_armbase: ndarray | None = None, ee_quat_armbase: ndarray | None = None) -> tuple[ndarray, ndarray] | None

Compute the current EE target in arm-base frame, or None if disengaged.

ee_pos_armbase / ee_quat_armbase are the arm's CURRENT EE pose; passing them enables the pos/rot reach limits (see the class docstring). Without them the legacy absolute mapping applies unchanged.

Source code in eden/utils/devices/metaquest/pose_mapping.py

def target(
    self,
    controller_pos_xr: np.ndarray,
    controller_quat_xr: np.ndarray,
    ee_pos_armbase: np.ndarray | None = None,
    ee_quat_armbase: np.ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray] | None:
    """Compute the current EE target in arm-base frame, or None if disengaged.

    ``ee_pos_armbase`` / ``ee_quat_armbase`` are the arm's CURRENT EE pose;
    passing them enables the pos/rot reach limits (see the class docstring).
    Without them the legacy absolute mapping applies unchanged.
    """
    if not self._engaged:
        return None
    assert self._ctrl_engage_pos is not None
    assert self._ctrl_engage_quat is not None
    assert self._ee_engage_pos is not None
    assert self._ee_engage_quat is not None

    # Position delta. Reach-limited path accumulates per-tick increments
    # (vectors commute, so per-tick-scaled increments compose to exactly the
    # absolute scaled delta until the reach limit absorbs). Legacy path: the
    # absolute delta from the engage frame.
    p_now = np.asarray(controller_pos_xr, float)
    pos_limited = ee_pos_armbase is not None and bool(self.pos_reach_limit)
    if pos_limited:
        assert self._ctrl_prev_pos is not None
        self._d_pos_eff = self._d_pos_eff + self.R @ (self.scale * (p_now - self._ctrl_prev_pos))
        d_pos_arm = self._d_pos_eff
    else:
        d_pos_arm = self.R @ (self.scale * (p_now - self._ctrl_engage_pos))
    self._ctrl_prev_pos = p_now.copy()

    q_now = np.asarray(controller_quat_xr, float)
    rot_limited = ee_quat_armbase is not None and bool(self.rot_reach_limit)
    if rot_limited:
        # Incremental path: accumulate this tick's controller rotation
        # increment (a degree or two — never direction-ambiguous) into the
        # effective delta. At scale_rotation=1 the increments telescope to
        # exactly the absolute delta until the reach limit clamps; at other
        # scales it is a rate gain (path-dependent on curved SO(3) loops).
        assert self._ctrl_prev_quat is not None
        if float(np.dot(q_now, self._ctrl_prev_quat)) < 0.0:
            q_now = -q_now  # hemisphere-align: q and -q are the same rotation
        inc = transform_quat_by_quat(inv_quat(self._ctrl_prev_quat), q_now)  # q_now ⊗ prev⁻¹
        self._ctrl_prev_quat = q_now.copy()
        if self.scale_rotation != 1.0:
            inc = rotvec_to_quat(self.scale_rotation * quat_to_rotvec(inc))
        # express the XR-frame increment in arm frame: R ⊗ inc ⊗ R⁻¹
        inc_arm = transform_quat_by_quat(self._R_quat_conj, transform_quat_by_quat(inc, self._R_quat))
        d_quat_arm = transform_quat_by_quat(self._d_quat_eff, inc_arm)  # inc_arm ⊗ d_quat_eff
        d_quat_arm /= np.linalg.norm(d_quat_arm)
        self._d_quat_eff = d_quat_arm
    else:
        # Legacy absolute path: rotation delta in XR world (now * engage⁻¹),
        # conjugated by R_quat to express in arm base. Keep the incremental
        # state fresh so a live reach-limit toggle doesn't see stale prev.
        if self._ctrl_prev_quat is not None:
            if float(np.dot(q_now, self._ctrl_prev_quat)) < 0.0:
                q_now = -q_now
            self._ctrl_prev_quat = q_now.copy()
        d_quat_xr = transform_quat_by_quat(inv_quat(self._ctrl_engage_quat), q_now)  # q_now ⊗ engage⁻¹
        if self.scale_rotation != 1.0:
            d_quat_xr = rotvec_to_quat(self.scale_rotation * quat_to_rotvec(d_quat_xr))
        # R ⊗ d_quat_xr ⊗ R⁻¹
        d_quat_arm = transform_quat_by_quat(self._R_quat_conj, transform_quat_by_quat(d_quat_xr, self._R_quat))

    target_quat = transform_quat_by_quat(self._ee_engage_quat, d_quat_arm)  # d_quat_arm ⊗ ee_engage

    if rot_limited:
        # Clamp the target to within rot_reach_limit of the arm's current
        # orientation and ABSORB the excess into the effective delta.
        ee_q = np.asarray(ee_quat_armbase, float)
        e = quat_to_rotvec(transform_quat_by_quat(inv_quat(ee_q), target_quat))  # target ⊗ ee⁻¹
        e_norm = float(np.linalg.norm(e))
        if e_norm > self.rot_reach_limit:
            e *= self.rot_reach_limit / e_norm
            target_quat = transform_quat_by_quat(ee_q, rotvec_to_quat(e))  # rotvec_to_quat(e) ⊗ ee
            self._d_quat_eff = transform_quat_by_quat(inv_quat(self._ee_engage_quat), target_quat)
            d_quat_arm = self._d_quat_eff

    # A pivot turns the rotation into a swing of the EE around it; with
    # rotation_pivot None the offset term vanishes (legacy in-place rotation).
    if self.rotation_pivot is not None:
        offset = self._ee_engage_pos - self.rotation_pivot
        target_pos = self.rotation_pivot + transform_by_quat(offset, d_quat_arm) + d_pos_arm
    else:
        target_pos = self._ee_engage_pos + d_pos_arm

    # Position reach limit: clamp toward the current EE position and ABSORB
    # the excess (mouse-at-screen-edge: reversal moves the target immediately
    # instead of after retracing the overshoot).
    if pos_limited:
        ee_p = np.asarray(ee_pos_armbase, float)
        dp = target_pos - ee_p
        dp_norm = float(np.linalg.norm(dp))
        if dp_norm > self.pos_reach_limit:
            clamped = ee_p + dp * (self.pos_reach_limit / dp_norm)
            self._d_pos_eff = self._d_pos_eff + (clamped - target_pos)
            target_pos = clamped

    return target_pos, target_quat

PICO

The XRoboToolkit tracking client, the bridge process, and the Remote Vision video link that streams Eden's stereo frames to the headset.

PICO tracking client: XRoboToolkit PC Service state -> Genesis/robot frame.

The PICO headset does not talk to Eden directly. It streams its tracking state to the XRoboToolkit PC Service running on this host, and the service exposes it through the xrobotoolkit_sdk extension module. :class:PicoXrClient wraps that module and hands Eden ready-to-use state:

• poses as 4x4 matrices in the Genesis/robot world frame (z up, x forward, y left, metres, origin at the headset's tracking origin),
• controller buttons / triggers / grips / thumbsticks by name,
• the PICO-only extras: paired motion trackers and, in Full-body mode, the 24 SMPL-ordered body joints.

The service reports poses in the OpenXR basis (x right, y up, -z forward), so every pose is conjugated into the robot basis exactly as Eden's Meta Quest bridge does — :data:R_ROBOT_XR here is the same matrix as televuer's R_ROBOT_OPENXR and as XRoboToolkit's own R_HEADSET_TO_WORLD.

:func:xr_pose_to_se3 and friends are module-level and SDK-free so the frame conversion is testable without a headset; only :class:PicoXrClient needs the extension module, which it imports on construction.

Install: build xrobotoolkit_sdk from https://github.com/XR-Robotics/XRoboToolkit-PC-Service-Pybind (it needs the PC Service's libPXREARobotSDK.so; setup_ubuntu.sh fetches and builds both). The service's shared library is preloaded from its install directory when the dynamic linker can't find it, so no LD_LIBRARY_PATH export is needed.

Attributes

R_ROBOT_XR module-attribute

R_ROBOT_XR = np.array([[0.0, 0.0, -1.0], [-1.0, 0.0, 0.0], [0.0, 1.0, 0.0]])

T_ROBOT_XR module-attribute

T_ROBOT_XR = np.eye(4)

T_XR_ROBOT module-attribute

T_XR_ROBOT = T_ROBOT_XR.T

UNTRACKED module-attribute

UNTRACKED = np.zeros((4, 4))

BODY_JOINT_NAMES module-attribute

BODY_JOINT_NAMES = ('pelvis', 'left_hip', 'right_hip', 'spine1', 'left_knee', 'right_knee', 'spine2', 'left_ankle', 'right_ankle', 'spine3', 'left_foot', 'right_foot', 'neck', 'left_collar', 'right_collar', 'head', 'left_shoulder', 'right_shoulder', 'left_elbow', 'right_elbow', 'left_wrist', 'right_wrist', 'left_hand', 'right_hand')

N_BODY_JOINTS module-attribute

N_BODY_JOINTS = len(BODY_JOINT_NAMES)

N_HAND_JOINTS module-attribute

N_HAND_JOINTS = 26

Classes

PicoXrClient

PicoXrClient(tracker_roles: dict[str, str] | None = None)

Live PICO tracking state, converted into Eden's world frame.

Constructing the client initialises the SDK (which connects to the PC Service on this host); call :meth:close when done. Every getter is a cheap read of the state the service pushes in the background, so calling them once per sim step is fine.

Source code in eden/utils/devices/pico/xr_client.py

def __init__(self, tracker_roles: dict[str, str] | None = None) -> None:
    """Parameters
    ----------
        tracker_roles: optional ``{serial_number: role}`` map that labels motion
            trackers by body placement (e.g. ``{"PA123...": "waist"}``). The
            service reports trackers only by serial, so this is how a tracker
            becomes ``left_foot`` / ``waist`` / ... in :meth:`trackers`. Unmapped
            trackers fall back to their serial.
    """
    self._sdk = load_sdk()
    self._sdk.init()
    self._tracker_roles = dict(tracker_roles or {})

    sdk = self._sdk
    # Name -> getter tables. The names match XRoboToolkit's own vocabulary
    # (XrClient.get_*_by_name) so recipes written against the upstream
    # samples carry over unchanged.
    self._poses = {
        "head": sdk.get_headset_pose,
        "left_controller": sdk.get_left_controller_pose,
        "right_controller": sdk.get_right_controller_pose,
    }
    self._values = {
        "left_trigger": sdk.get_left_trigger,
        "right_trigger": sdk.get_right_trigger,
        "left_grip": sdk.get_left_grip,
        "right_grip": sdk.get_right_grip,
    }
    self._buttons = {
        "X": sdk.get_X_button,
        "Y": sdk.get_Y_button,
        "A": sdk.get_A_button,
        "B": sdk.get_B_button,
        "left_menu_button": sdk.get_left_menu_button,
        "right_menu_button": sdk.get_right_menu_button,
        "left_axis_click": sdk.get_left_axis_click,
        "right_axis_click": sdk.get_right_axis_click,
    }
    self._axes = {"left": sdk.get_left_axis, "right": sdk.get_right_axis}
    self._hand_active = {"left": sdk.get_left_hand_is_active, "right": sdk.get_right_hand_is_active}
    self._hand_states = {"left": sdk.get_left_hand_tracking_state, "right": sdk.get_right_hand_tracking_state}

Attributes

tracking property

tracking: bool

True once the headset reports a pose (i.e. the app is connected and sending).

Methods:

pose

pose(name: str) -> NDArray[float64]

Robot-frame 4x4 pose of head / left_controller / right_controller.

Source code in eden/utils/devices/pico/xr_client.py

def pose(self, name: str) -> npt.NDArray[np.float64]:
    """Robot-frame 4x4 pose of ``head`` / ``left_controller`` / ``right_controller``."""
    return xr_pose_to_se3(self._get(self._poses, name, "pose")())

value

value(name: str) -> float

Analog trigger/grip reading in [0, 1] (e.g. "left_grip").

Source code in eden/utils/devices/pico/xr_client.py

def value(self, name: str) -> float:
    """Analog trigger/grip reading in ``[0, 1]`` (e.g. ``"left_grip"``)."""
    return float(self._get(self._values, name, "analog input")())

button

button(name: str) -> bool

Button state (e.g. "A", "left_menu_button", "right_axis_click").

Source code in eden/utils/devices/pico/xr_client.py

def button(self, name: str) -> bool:
    """Button state (e.g. ``"A"``, ``"left_menu_button"``, ``"right_axis_click"``)."""
    return bool(self._get(self._buttons, name, "button")())

axis

axis(side: str) -> tuple[float, float]

Thumbstick (x, y) in [-1, 1] for "left" or "right".

Source code in eden/utils/devices/pico/xr_client.py

def axis(self, side: str) -> tuple[float, float]:
    """Thumbstick ``(x, y)`` in ``[-1, 1]`` for ``"left"`` or ``"right"``."""
    x, y = self._get(self._axes, side, "thumbstick")()
    return float(x), float(y)

timestamp_ns

timestamp_ns() -> int

Service timestamp of the latest controller/headset sample, in nanoseconds.

Source code in eden/utils/devices/pico/xr_client.py

def timestamp_ns(self) -> int:
    """Service timestamp of the latest controller/headset sample, in nanoseconds."""
    return int(self._sdk.get_time_stamp_ns())

hand_joints

hand_joints(side: str) -> NDArray[float64] | None

Robot-frame (26, 3) hand-joint positions, or None when hand tracking is off.

This is the OpenXR 26-joint layout (palm included), not the 25-joint WebXR one the Meta Quest bridge publishes — index it with the OpenXR XrHandJointEXT order.

Source code in eden/utils/devices/pico/xr_client.py

def hand_joints(self, side: str) -> npt.NDArray[np.float64] | None:
    """Robot-frame ``(26, 3)`` hand-joint positions, or None when hand tracking is off.

    This is the OpenXR 26-joint layout (palm included), *not* the 25-joint
    WebXR one the Meta Quest bridge publishes — index it with the OpenXR
    ``XrHandJointEXT`` order.
    """
    if not self._get(self._hand_active, side, "hand")():
        return None
    state = np.asarray(self._hand_states[side](), dtype=float)
    if state.shape != (N_HAND_JOINTS, 7):
        raise ValueError(f"expected ({N_HAND_JOINTS}, 7) hand joint poses from the service, got {state.shape}")
    return xr_points_to_robot(state[:, :3])

trackers

trackers() -> dict[str, dict]

PICO motion trackers (up to 5) in the robot frame, keyed by role or serial.

Each value is {"serial", "role", "pos" (3), "quat" (4), "vel" (6), "acc" (6)}; role is None for a tracker with no tracker_roles entry, and such a tracker is keyed by its serial instead. vel/acc are (vx, vy, vz, wx, wy, wz) in the robot frame.

Source code in eden/utils/devices/pico/xr_client.py

def trackers(self) -> dict[str, dict]:
    """PICO motion trackers (up to 5) in the robot frame, keyed by role or serial.

    Each value is ``{"serial", "role", "pos" (3), "quat" (4), "vel" (6),
    "acc" (6)}``; ``role`` is ``None`` for a tracker with no ``tracker_roles``
    entry, and such a tracker is keyed by its serial instead. ``vel``/``acc`` are
    ``(vx, vy, vz, wx, wy, wz)`` in the robot frame.
    """
    count = int(self._sdk.num_motion_data_available())
    if count == 0:
        return {}
    poses = self._sdk.get_motion_tracker_pose()
    vels = self._sdk.get_motion_tracker_velocity()
    accs = self._sdk.get_motion_tracker_acceleration()
    serials = self._sdk.get_motion_tracker_serial_numbers()
    out: dict[str, dict] = {}
    for i in range(count):
        pq = xr_pose_to_pos_quat(poses[i])
        if pq is None:
            continue
        serial = str(serials[i])
        role = self._tracker_roles.get(serial)
        out[role or serial] = {
            "serial": serial,
            "role": role,
            "pos": pq[0].tolist(),
            "quat": pq[1].tolist(),
            "vel": xr_vel6_to_robot(vels[i]).tolist(),
            "acc": xr_vel6_to_robot(accs[i]).tolist(),
        }
    return out

body_joints

body_joints(poses: ndarray | None = None) -> dict[str, dict] | None

Full-body joints as {name: {pos, quat, vel, acc}}, or None when unavailable.

Requires Full-body mode in the PICO app with calibrated motion trackers; keys are :data:BODY_JOINT_NAMES (left_wrist, left_ankle, left_foot, pelvis, ...). vel/acc are (linear, angular) 6-vectors in the robot frame. A joint the service has not solved yet reports a degenerate quaternion and is published as identity at the origin. poses reuses an already-read (24, 7) service-frame array so one snapshot needs a single SDK pose read; velocities/accelerations are always read live.

Source code in eden/utils/devices/pico/xr_client.py

def body_joints(self, poses: np.ndarray | None = None) -> dict[str, dict] | None:
    """Full-body joints as ``{name: {pos, quat, vel, acc}}``, or None when unavailable.

    Requires Full-body mode in the PICO app with calibrated motion trackers;
    keys are :data:`BODY_JOINT_NAMES` (``left_wrist``, ``left_ankle``,
    ``left_foot``, ``pelvis``, ...). ``vel``/``acc`` are ``(linear, angular)``
    6-vectors in the robot frame. A joint the service has not solved yet reports
    a degenerate quaternion and is published as identity at the origin.
    ``poses`` reuses an already-read ``(24, 7)`` service-frame array so one
    snapshot needs a single SDK pose read; velocities/accelerations are
    always read live.
    """
    if poses is None:
        if not self._sdk.is_body_data_available():
            return None
        poses = np.asarray(self._sdk.get_body_joints_pose(), dtype=float)
    vels = np.asarray(self._sdk.get_body_joints_velocity(), dtype=float)
    accs = np.asarray(self._sdk.get_body_joints_acceleration(), dtype=float)
    for name, arr, cols in (("pose", poses, 7), ("velocity", vels, 6), ("acceleration", accs, 6)):
        if arr.shape != (N_BODY_JOINTS, cols):
            raise ValueError(f"expected ({N_BODY_JOINTS}, {cols}) body {name} from the service, got {arr.shape}")
    body: dict[str, dict] = {}
    for i, joint in enumerate(BODY_JOINT_NAMES):
        pq = xr_pose_to_pos_quat(poses[i])
        body[joint] = {
            "pos": (pq[0] if pq is not None else np.zeros(3)).tolist(),
            "quat": (pq[1] if pq is not None else np.array([1.0, 0.0, 0.0, 0.0])).tolist(),
            "vel": xr_vel6_to_robot(vels[i]).tolist(),
            "acc": xr_vel6_to_robot(accs[i]).tolist(),
        }
    return body

body_raw

body_raw() -> tuple[int, list[list[float]]] | None

Untouched service-frame body poses: (timestamp_ns, (24, 7) xyz+xyzw lists).

Deliberately no frame conversion, no untracked-joint substitution, no wxyz flip — for consumers that apply their own frame math to exactly what the SDK reports (the SONIC teleop pipelines). The timestamp is get_time_stamp_ns (not the body-specific stamp) read immediately before the poses, matching the in-process reader's dedupe order (eden.extensions.teleop.sonic.inputs.PicoReader).

Source code in eden/utils/devices/pico/xr_client.py

def body_raw(self) -> tuple[int, list[list[float]]] | None:
    """Untouched service-frame body poses: ``(timestamp_ns, (24, 7) xyz+xyzw lists)``.

    Deliberately no frame conversion, no untracked-joint substitution, no
    wxyz flip — for consumers that apply their own frame math to exactly
    what the SDK reports (the SONIC teleop pipelines). The timestamp is
    ``get_time_stamp_ns`` (not the body-specific stamp) read immediately
    before the poses, matching the in-process reader's dedupe order
    (``eden.extensions.teleop.sonic.inputs.PicoReader``).
    """
    if not self._sdk.is_body_data_available():
        return None
    stamp_ns = int(self._sdk.get_time_stamp_ns())
    poses = np.asarray(self._sdk.get_body_joints_pose(), dtype=float)
    if poses.shape != (N_BODY_JOINTS, 7):
        raise ValueError(f"expected ({N_BODY_JOINTS}, 7) body poses from the service, got {poses.shape}")
    return stamp_ns, poses.tolist()

state

state(derived_body: bool = True) -> dict

Snapshot every input as a JSON-safe dict, in the Genesis/robot frame.

The controller keys deliberately match what eden metaquest bridge --tracking controller publishes (head, {side}_wrist, {side}_trigger[_value], {side}_squeeze[_value], {side}_a_button/_b_button, {side}_thumbstick[_value]), so an Eden-side teleop loop runs against either headset unchanged. PICO's extras are additive: {side}_menu_button, trackers (see :meth:trackers), body_ready and body (see :meth:body_joints), plus the lossless body_raw / body_raw_timestamp_ns passthrough (see :meth:body_raw).

derived_body=False skips the per-joint body dict and trackers (a consumer that only reads body_raw — the SONIC operator — saves most of the per-snapshot SDK work; body_ready then reflects the raw read).

Source code in eden/utils/devices/pico/xr_client.py

def state(self, derived_body: bool = True) -> dict:
    """Snapshot every input as a JSON-safe dict, in the Genesis/robot frame.

    The controller keys deliberately match what ``eden metaquest bridge
    --tracking controller`` publishes (``head``, ``{side}_wrist``,
    ``{side}_trigger[_value]``, ``{side}_squeeze[_value]``,
    ``{side}_a_button``/``_b_button``, ``{side}_thumbstick[_value]``), so an
    Eden-side teleop loop runs against either headset unchanged. PICO's extras
    are additive: ``{side}_menu_button``, ``trackers`` (see :meth:`trackers`),
    ``body_ready`` and ``body`` (see :meth:`body_joints`), plus the lossless
    ``body_raw`` / ``body_raw_timestamp_ns`` passthrough (see :meth:`body_raw`).

    ``derived_body=False`` skips the per-joint ``body`` dict and ``trackers``
    (a consumer that only reads ``body_raw`` — the SONIC operator — saves
    most of the per-snapshot SDK work; ``body_ready`` then reflects the raw
    read).
    """
    # Read the head pose once so ``ready`` describes the matrix we publish.
    head = self.pose("head")
    state = {
        "device": "pico",
        "mode": "controller",
        "ready": bool(np.abs(head).sum() > 1e-6),
        "head": head.tolist(),
        "left_wrist": self.pose("left_controller").tolist(),
        "right_wrist": self.pose("right_controller").tolist(),
        "timestamp_ns": self.timestamp_ns(),
    }
    # X/Y (left) and A/B (right) are the primary/secondary buttons of their
    # controller, which is what the Meta Quest bridge calls a/b per side.
    primary = {"left": "X", "right": "A"}
    secondary = {"left": "Y", "right": "B"}
    for side in ("left", "right"):
        trigger = self.value(f"{side}_trigger")
        squeeze = self.value(f"{side}_grip")
        state.update(
            {
                f"{side}_trigger": trigger > 0.5,
                f"{side}_trigger_value": trigger,
                f"{side}_squeeze": squeeze > 0.5,
                f"{side}_squeeze_value": squeeze,
                f"{side}_a_button": self.button(primary[side]),
                f"{side}_b_button": self.button(secondary[side]),
                f"{side}_menu_button": self.button(f"{side}_menu_button"),
                f"{side}_thumbstick": self.button(f"{side}_axis_click"),
                f"{side}_thumbstick_value": list(self.axis(side)),
            }
        )

    raw = self.body_raw()
    if raw is not None:
        state["body_raw_timestamp_ns"], state["body_raw"] = raw
    if derived_body:
        state["trackers"] = self.trackers()
        # Reuse the raw read: one SDK pose read per snapshot.
        body = self.body_joints(poses=None if raw is None else np.asarray(raw[1], dtype=float))
        state["body_ready"] = body is not None
        if body is not None:
            state["body"] = body  # {joint_name: {pos, quat, vel, acc}}
    else:
        state["body_ready"] = raw is not None
    return state

close

close() -> None

Disconnect from the PC Service.

Source code in eden/utils/devices/pico/xr_client.py

def close(self) -> None:
    """Disconnect from the PC Service."""
    self._sdk.close()

Functions:

load_sdk

load_sdk()

Import xrobotoolkit_sdk, raising a build hint instead of a bare ImportError.

The pybind module links against the PC Service's libPXREARobotSDK.so, which lives outside the default loader path. When the import fails on that library, it is preloaded from the service's install directory and the import retried — no LD_LIBRARY_PATH export needed.

Source code in eden/utils/devices/pico/xr_client.py

def load_sdk():
    """Import ``xrobotoolkit_sdk``, raising a build hint instead of a bare ImportError.

    The pybind module links against the PC Service's ``libPXREARobotSDK.so``,
    which lives outside the default loader path. When the import fails on that
    library, it is preloaded from the service's install directory and the
    import retried — no ``LD_LIBRARY_PATH`` export needed.
    """
    try:
        return _import_sdk()
    except ImportError as exc:
        if _PXREA_LIB in str(exc) and _preload_pxrea():
            try:
                return _import_sdk()
            except ImportError as exc2:  # pragma: no cover - depends on the local build
                raise ImportError(_SDK_HINT) from exc2
        raise ImportError(_SDK_HINT) from exc

xr_pose_to_se3

xr_pose_to_se3(pose: ArrayLike) -> NDArray[float64]

Convert one service pose (x, y, z, qx, qy, qz, qw) to a robot-frame 4x4.

Returns :data:UNTRACKED (all zeros) when the quaternion is degenerate, which is what the service reports for a device it has not seen yet. Eden-side consumers test np.abs(mat).sum() < 1e-6 for "tracking dropped".

Source code in eden/utils/devices/pico/xr_client.py

def xr_pose_to_se3(pose: npt.ArrayLike) -> npt.NDArray[np.float64]:
    """Convert one service pose ``(x, y, z, qx, qy, qz, qw)`` to a robot-frame 4x4.

    Returns :data:`UNTRACKED` (all zeros) when the quaternion is degenerate, which
    is what the service reports for a device it has not seen yet. Eden-side
    consumers test ``np.abs(mat).sum() < 1e-6`` for "tracking dropped".
    """
    values = np.asarray(pose, dtype=float).reshape(-1)
    if values.shape != (7,):
        raise ValueError(f"expected a 7-element (x, y, z, qx, qy, qz, qw) pose, got shape {values.shape}")
    quat = values[[6, 3, 4, 5]]  # the service reports xyzw; Genesis is wxyz
    norm = float(np.linalg.norm(quat))
    if norm < 1e-8:
        return UNTRACKED.copy()
    return T_ROBOT_XR @ trans_quat_to_T(values[:3], quat / norm) @ T_XR_ROBOT

xr_pose_to_pos_quat

xr_pose_to_pos_quat(pose: ArrayLike) -> tuple[NDArray[float64], NDArray[float64]] | None

Robot-frame (pos (3,), wxyz quat (4,)) for one service pose, or None if untracked.

Source code in eden/utils/devices/pico/xr_client.py

def xr_pose_to_pos_quat(pose: npt.ArrayLike) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]] | None:
    """Robot-frame ``(pos (3,), wxyz quat (4,))`` for one service pose, or None if untracked."""
    mat = xr_pose_to_se3(pose)
    if np.abs(mat).sum() < 1e-6:
        return None
    pos, quat = T_to_trans_quat(mat)
    return np.asarray(pos, dtype=float), np.asarray(quat, dtype=float)

xr_points_to_robot

xr_points_to_robot(points: ArrayLike) -> NDArray[float64]

Rotate (N, 3) XR-frame points into the Genesis/robot frame.

Source code in eden/utils/devices/pico/xr_client.py

def xr_points_to_robot(points: npt.ArrayLike) -> npt.NDArray[np.float64]:
    """Rotate ``(N, 3)`` XR-frame points into the Genesis/robot frame."""
    pts = np.asarray(points, dtype=float)
    if pts.ndim != 2 or pts.shape[1] != 3:
        raise ValueError(f"expected (N, 3) points, got shape {pts.shape}")
    return pts @ R_ROBOT_XR.T  # (R @ pts.T).T

xr_vel6_to_robot

xr_vel6_to_robot(vel6: ArrayLike) -> NDArray[float64]

Rotate a 6-vector (vx, vy, vz, wx, wy, wz) velocity/accel into the robot frame.

Both the linear and angular halves are free vectors, so each rotates by :data:R_ROBOT_XR (a proper rotation, so angular pseudovectors transform the same way). Used for the motion-tracker and body-joint velocities/accelerations.

Source code in eden/utils/devices/pico/xr_client.py

def xr_vel6_to_robot(vel6: npt.ArrayLike) -> npt.NDArray[np.float64]:
    """Rotate a 6-vector ``(vx, vy, vz, wx, wy, wz)`` velocity/accel into the robot frame.

    Both the linear and angular halves are free vectors, so each rotates by
    :data:`R_ROBOT_XR` (a proper rotation, so angular pseudovectors transform the
    same way). Used for the motion-tracker and body-joint velocities/accelerations.
    """
    v = np.asarray(vel6, dtype=float).reshape(-1)
    if v.shape != (6,):
        raise ValueError(f"expected a 6-element (linear, angular) vector, got shape {v.shape}")
    return np.concatenate([R_ROBOT_XR @ v[:3], R_ROBOT_XR @ v[3:]])

PICO teleoperation bridge: XRoboToolkit PC Service <-> Eden.

Owns everything device-side of a PICO teleop session so the Eden process only speaks ZMQ: it reads the headset's tracking state from the PC Service (via xrobotoolkit_sdk), publishes it in the Genesis/robot world frame, and streams Eden's rendered stereo frames back to the headset as H.264 over the app's Remote Vision link.

Data flow (see :mod:eden.utils.devices.pico.protocol for the headset link and :mod:eden.utils.devices.metaquest.protocol for the ZMQ channels)::

PICO ──Wi-Fi──▶ XRoboToolkit PC Service ──▶ eden pico bridge
                                             │  poses  (PUB :5557) ──▶ Eden
  headset decoder ◀──H.264 / TCP─────────────┤  frames (PULL :5556) ◀──┘

The published state dict uses the same schema as eden metaquest bridge --tracking controller, so an Eden-side teleop loop runs against either headset unchanged; PICO's extras (menu buttons, motion trackers, full-body joints) are additive keys. See :meth:eden.utils.devices.pico.xr_client.PicoXrClient.state.

Requires the pico extra (uv sync --inexact --extra pico) plus a locally built xrobotoolkit_sdk — see :mod:eden.utils.devices.pico.xr_client.

Run: eden pico bridge # or, equivalently: uv run python -m eden.utils.devices.pico.bridge

Attributes

Classes

Functions:

main

main(argv=None)

Run the PICO bridge loop (parses argv or sys.argv when None).

Source code in eden/utils/devices/pico/bridge.py

def main(argv=None):
    """Run the PICO bridge loop (parses ``argv`` or ``sys.argv`` when ``None``)."""
    ap = argparse.ArgumentParser(description="Eden <-> PICO teleop bridge (XRoboToolkit side)")
    ap.add_argument("--fps", type=float, default=30.0, help="pose publish rate (frames are forwarded as they arrive)")
    ap.add_argument(
        "--control-address",
        default=f"0.0.0.0:{CONTROL_PORT}",
        help="host:port to listen on for the app's OPEN_CAMERA / CLOSE_CAMERA commands",
    )
    ap.add_argument("--pose-port", type=int, default=P.POSE_PORT, help="ZMQ PUB port for the tracking state")
    ap.add_argument("--frame-port", type=int, default=P.FRAME_PORT, help="ZMQ PULL port for Eden's rendered frames")
    ap.add_argument("--no-video", action="store_true", help="publish tracking only; do not stream frames back")
    ap.add_argument(
        "--no-derived-body",
        action="store_true",
        help="skip the derived per-joint body dict and trackers; publish the lossless body_raw only",
    )
    ap.add_argument(
        "--tracker-role",
        action="append",
        default=[],
        metavar="SERIAL=ROLE",
        help="label a motion tracker by serial, e.g. --tracker-role PA1234=waist "
        "(repeatable; unmapped trackers are logged so you can add them)",
    )
    args = ap.parse_args(argv)

    tracker_roles = {}
    for spec in args.tracker_role:
        serial, sep, role = spec.partition("=")
        if not (serial and sep and role):
            ap.error(f"--tracker-role expects SERIAL=ROLE, got {spec!r}")
        tracker_roles[serial] = role

    # The bridge does not run Genesis, so en.init() (and its en.logger) never
    # runs here -- stand up Eden's logger directly.
    logger = getattr(en, "logger", None) or en.Logger("INFO")

    client = PicoXrClient(tracker_roles=tracker_roles)  # raises with a build hint if the SDK is missing
    video = None if args.no_video else PicoVideoLink(args.control_address, logger=logger)

    ctx = zmq.Context.instance()
    # No video, no frame channel: nothing would consume the frames, and not
    # binding keeps the port free for whoever shares this host (the SONIC
    # operator's command PUB uses the same default number, 5556).
    frame_sock = None if video is None else P.frame_pull(ctx, port=args.frame_port)
    pose_sock = P.pose_pub(ctx, port=args.pose_port)

    # Eden-styled banner: ~<...>~ mint, ~~<...>~~ mint-underline, ~~~~<...>~~~~ bold.
    rule = "━" * 64
    logger.info(f"~<{rule}>~")
    logger.info(" 🥽 ~~~~<PICO XR bridge>~~~~  (XRoboToolkit)  —  waiting for the headset")
    logger.info(" On the PICO: open ~<XRoboToolkit>~, point it at this host and tap ~<Connect>~")
    for ip in list_host_ips() or ["<host-ip>"]:
        logger.info(f"     ~~<{ip}>~~")
    logger.info(" Status must read ~<Working>~; then enable ~<Send>~ under Data & Control")
    if video is not None:
        logger.info(f" Stereo view: ~<Remote Vision>~ ~<→>~ ~<Listen>~ (Eden listens on {args.control_address})")
    frames_note = "frames off (--no-video)" if frame_sock is None else f"frames PULL ~<:{args.frame_port}>~"
    logger.info(f" ZMQ   poses PUB ~<:{args.pose_port}>~   {frames_note}")
    logger.info(f"~<{rule}>~")

    if video is not None:
        video.start()

    running = {"v": True}

    def stop(*_):
        running["v"] = False

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    period = 1.0 / max(args.fps, 1.0)
    frames_forwarded = 0
    last_log = time.time()
    warned_at = 0.0
    seen_serials: set[str] = set()
    # Probe once up front: a fundamentally wrong SDK contract fails loud here, at
    # startup, rather than deep in the loop below.
    derived_body = not args.no_derived_body
    last_state = client.state(derived_body=derived_body)
    try:
        while running["v"]:
            loop_start = time.time()

            # Newest frame from Eden -> headset. Eden renders RGB and the encoder
            # takes RGB, so no channel flip is needed here.
            if video is not None:
                frame = P.recv_frame(frame_sock, flags=zmq.NOBLOCK)
                if frame is not None:
                    video.push(frame)
                    frames_forwarded += 1

            # Transient device noise (a momentarily malformed read) must not tear
            # down a live teleop session: log it at most every few seconds and
            # republish the last good state (the consumers CONFLATE to latest).
            try:
                last_state = client.state(derived_body=derived_body)
            except (ValueError, RuntimeError) as exc:
                now = time.time()
                if now - warned_at > 5.0:
                    logger.warning(f"skipping a bad PICO read: {exc}")
                    warned_at = now
            P.send_pose(pose_sock, last_state)

            # Surface each new unlabeled tracker once so the operator can map it.
            for tr in last_state.get("trackers", {}).values():
                if tr.get("role") is None and tr["serial"] not in seen_serials:
                    seen_serials.add(tr["serial"])
                    logger.info(
                        f"motion tracker {tr['serial']} is unlabeled — map it with --tracker-role {tr['serial']}=<role>"
                    )

            now = time.time()
            if now - last_log > 5.0:
                logger.info(_status(last_state, video, frames_forwarded))
                last_log = now

            dt = time.time() - loop_start
            if dt < period:
                time.sleep(period - dt)
    finally:
        logger.info("shutting down...")
        if video is not None:
            video.stop()
        client.close()

Stream Eden's rendered stereo frames to a PICO headset over the Remote Vision link.

:class:PicoVideoLink owns both sockets of the link documented in :mod:eden.utils.devices.pico.protocol:

• it listens on :13579 for the OPEN_CAMERA / CLOSE_CAMERA commands the XRoboToolkit app sends when the operator taps Listen / stops,
• on OPEN_CAMERA it connects out to the decoder address the request names and pushes H.264 (or H.265) frames until the app closes the stream.

Two daemon threads do the work — one accepting control connections, one encoding — so the caller only ever does::

link = PicoVideoLink()
link.start()
while ...:
    link.push(rgb)   # (H, W, 3) uint8 RGB, left|right side-by-side
link.stop()

push() is non-blocking and latest-frame-wins: it drops into a single slot that the encoder thread drains, so a slow encoder never stalls the sim loop and the headset always shows the newest frame rather than a growing backlog.

Frames are encoded at the size the headset asked for; a differently sized frame is rescaled (with a one-time warning) because the app's decoder is configured from the OPEN_CAMERA request, not from the stream.

The encoder/sender half is ported from SIMPLE's src/simple/teleop/pico/ (MIT, https://github.com/physical-superintelligence-lab/SIMPLE).

Classes

PicoVideoLink

PicoVideoLink(address: str = f'0.0.0.0:{CONTROL_PORT}', logger=None)

Control server + video sender for one PICO headset.

Source code in eden/utils/devices/pico/video.py

def __init__(self, address: str = f"0.0.0.0:{P.CONTROL_PORT}", logger=None) -> None:
    host, _, port = address.rpartition(":")
    self._host = host or "0.0.0.0"
    self._port = int(port)
    self._logger = logger

    self._stop = threading.Event()
    self._control_thread: threading.Thread | None = None
    self._encode_thread: threading.Thread | None = None
    self._client: socket.socket | None = None
    self._sender: _VideoSender | None = None
    self._latest = _LatestFrame()

Attributes

streaming property

streaming: bool

True while a headset is connected and frames are being encoded.

frames_sent property

frames_sent: int

Frames encoded and written to the headset since the stream opened.

Methods:

start

start() -> None

Start accepting control connections (returns immediately).

Source code in eden/utils/devices/pico/video.py

def start(self) -> None:
    """Start accepting control connections (returns immediately)."""
    if self._control_thread is not None:
        raise RuntimeError("PicoVideoLink.start() called twice; create a new link instead")
    self._stop.clear()
    self._control_thread = threading.Thread(target=self._serve, name="PicoVideoControl", daemon=True)
    self._control_thread.start()

stop

stop() -> None

Tear down the stream, the control connection, and both threads.

Source code in eden/utils/devices/pico/video.py

def stop(self) -> None:
    """Tear down the stream, the control connection, and both threads."""
    self._stop.set()
    self._close_stream()
    self._close_client()
    for thread in (self._control_thread, self._encode_thread):
        if thread is not None:
            thread.join(timeout=3.0)
    self._control_thread = None
    self._encode_thread = None

push

push(rgb: NDArray[uint8]) -> None

Offer one rendered frame ((H, W, 3) uint8 RGB) to the headset.

Non-blocking and latest-wins; a no-op while no headset is streaming.

Source code in eden/utils/devices/pico/video.py

def push(self, rgb: npt.NDArray[np.uint8]) -> None:
    """Offer one rendered frame (``(H, W, 3)`` uint8 RGB) to the headset.

    Non-blocking and latest-wins; a no-op while no headset is streaming.
    """
    if self.streaming:
        self._latest.put(np.ascontiguousarray(rgb, dtype=np.uint8))

Wire protocol of the Eden <-> PICO (XRoboToolkit) Remote Vision video link.

The XRoboToolkit PICO app drives both directions; Eden is the server on the control channel and the client on the video channel:

• control (headset -> Eden), TCP :13579::

  [4B BE uint32 body_len][body]
  

  where body is a little-endian NetworkDataProtocol record::

  [4B LE int32 cmd_len][command][4B LE int32 data_len][data]
  

  The app sends OPEN_CAMERA (data = a CameraRequestData payload, see below) when the operator taps Listen, and CLOSE_CAMERA (no data) when they stop. Eden acknowledges each command with the same framing, echoing the command name and an empty data field.

• video (Eden -> headset), TCP to the ip/port the OPEN_CAMERA request names — that is where the app's MediaDecoder listens::

  [4B BE uint32 frame_size][H.264 / H.265 NAL data]
  

CameraRequestData (the OPEN_CAMERA payload) is::

    [0xCA 0xFE]        magic
    [0x01]             version
    [7 x 4B LE int32]  width, height, fps, bitrate, enable_mv_hevc,
                       render_mode, port
    [1B len + bytes]   camera   (the app's source profile, e.g. "ZED")
    [1B len + bytes]   ip       (the headset's own address)

This module is pure bytes <-> objects (stdlib only), so the framing is unit-testable without a headset; the sockets and the encoder live in :mod:eden.utils.devices.pico.video.

Ported from SIMPLE's src/simple/teleop/pico/protocol.py (MIT, https://github.com/physical-superintelligence-lab/SIMPLE), which is where the byte layout above was reverse-engineered from the PICO app.

Attributes

CONTROL_PORT module-attribute

CONTROL_PORT = 13579

MAGIC module-attribute

MAGIC = b'\xca\xfe'

VERSION module-attribute

VERSION = 1

OPEN_CAMERA module-attribute

OPEN_CAMERA = 'OPEN_CAMERA'

CLOSE_CAMERA module-attribute

CLOSE_CAMERA = 'CLOSE_CAMERA'

HEADER_SIZE module-attribute

HEADER_SIZE = struct.calcsize(_FRAME_HEADER)

Classes

CameraRequest dataclass

CameraRequest(width: int, height: int, fps: int, bitrate: int, hevc: bool, render_mode: int, port: int, camera: str, ip: str)

A decoded OPEN_CAMERA payload: what the headset wants and where to send it.

width/height are the full side-by-side frame size (both eyes), and ip/port address the decoder listening inside the PICO app.

Attributes

width instance-attribute

width: int

height instance-attribute

height: int

fps instance-attribute

fps: int

bitrate instance-attribute

bitrate: int

hevc instance-attribute

hevc: bool

render_mode instance-attribute

render_mode: int

port instance-attribute

port: int

camera instance-attribute

camera: str

ip instance-attribute

ip: str

Methods:

Functions:

encode_message

encode_message(command: str, data: bytes = b'') -> bytes

Frame one control message (length header included) ready to sendall.

Source code in eden/utils/devices/pico/protocol.py

def encode_message(command: str, data: bytes = b"") -> bytes:
    """Frame one control message (length header included) ready to ``sendall``."""
    cmd = command.encode("utf-8")
    body = struct.pack("<i", len(cmd)) + cmd + struct.pack("<i", len(data)) + data
    return struct.pack(_FRAME_HEADER, len(body)) + body

decode_message

decode_message(body: bytes) -> tuple[str, bytes]

Split a control-message body into (command, data).

body is the payload after the 4-byte length header (see :func:unpack_length). Raises ValueError on a malformed record — the caller should log it and keep reading rather than tear the connection down, since the next message may be fine.

Source code in eden/utils/devices/pico/protocol.py

def decode_message(body: bytes) -> tuple[str, bytes]:
    """Split a control-message body into ``(command, data)``.

    ``body`` is the payload *after* the 4-byte length header (see
    :func:`unpack_length`). Raises ``ValueError`` on a malformed record — the
    caller should log it and keep reading rather than tear the connection down,
    since the next message may be fine.
    """
    if len(body) < 8:
        raise ValueError(f"control body too small: {len(body)} bytes (need >= 8)")

    (cmd_len,) = struct.unpack_from("<i", body, 0)
    if cmd_len < 0 or 4 + cmd_len > len(body):
        raise ValueError(f"invalid command length {cmd_len} in a {len(body)}-byte body")
    command = body[4 : 4 + cmd_len].decode("utf-8").rstrip("\x00")

    offset = 4 + cmd_len
    if offset + 4 > len(body):
        raise ValueError(f"command {command!r} is missing its data-length field")
    (data_len,) = struct.unpack_from("<i", body, offset)
    offset += 4
    if data_len < 0 or offset + data_len > len(body):
        raise ValueError(f"invalid data length {data_len} for command {command!r}")

    return command, body[offset : offset + data_len]

unpack_length

unpack_length(header: bytes) -> int

Decode a 4-byte big-endian length header (control body or video frame).

Source code in eden/utils/devices/pico/protocol.py

def unpack_length(header: bytes) -> int:
    """Decode a 4-byte big-endian length header (control body or video frame)."""
    if len(header) != HEADER_SIZE:
        raise ValueError(f"length header must be {HEADER_SIZE} bytes, got {len(header)}")
    return int(struct.unpack(_FRAME_HEADER, header)[0])

decode_camera_request

decode_camera_request(data: bytes) -> CameraRequest

Decode an OPEN_CAMERA payload. Raises ValueError if malformed.

Source code in eden/utils/devices/pico/protocol.py

def decode_camera_request(data: bytes) -> CameraRequest:
    """Decode an ``OPEN_CAMERA`` payload. Raises ``ValueError`` if malformed."""
    if len(data) < 3 + 28:
        raise ValueError(f"camera request too small: {len(data)} bytes (need >= 31)")
    if data[:2] != MAGIC:
        raise ValueError(f"bad camera-request magic: {data[:2].hex()} (expected {MAGIC.hex()})")
    if data[2] != VERSION:
        raise ValueError(f"unsupported camera-request version: {data[2]} (expected {VERSION})")

    width, height, fps, bitrate, enable_mv_hevc, render_mode, port = struct.unpack_from("<7i", data, 3)
    camera, offset = _decode_string(data, 3 + 28)
    ip, offset = _decode_string(data, offset)
    return CameraRequest(
        width=width,
        height=height,
        fps=fps,
        bitrate=bitrate,
        hevc=bool(enable_mv_hevc),
        render_mode=render_mode,
        port=port,
        camera=camera,
        ip=ip,
    )

encode_camera_request

encode_camera_request(request: CameraRequest) -> bytes

Encode a :class:CameraRequest back into its wire payload.

Eden never sends this — the PICO app does. It exists so tests (and a fake headset client) can exercise the decoder against the documented layout.

Source code in eden/utils/devices/pico/protocol.py

def encode_camera_request(request: CameraRequest) -> bytes:
    """Encode a :class:`CameraRequest` back into its wire payload.

    Eden never sends this — the PICO app does. It exists so tests (and a fake
    headset client) can exercise the decoder against the documented layout.
    """
    head = MAGIC + bytes([VERSION])
    ints = struct.pack(
        "<7i",
        request.width,
        request.height,
        request.fps,
        request.bitrate,
        int(request.hevc),
        request.render_mode,
        request.port,
    )
    return head + ints + _encode_string(request.camera) + _encode_string(request.ip)

pack_video_frame

pack_video_frame(nal: bytes) -> bytes

Frame one encoded video packet for the headset's decoder socket.

Source code in eden/utils/devices/pico/protocol.py

def pack_video_frame(nal: bytes) -> bytes:
    """Frame one encoded video packet for the headset's decoder socket."""
    return struct.pack(_FRAME_HEADER, len(nal)) + nal

Wuji Glove

Wuji Glove teleoperation device: live hand pose + tactile, in Eden's hand frame.

The glove is a plain SDK device — it streams over USB/Ethernet to wuji_sdk running in this process, so unlike the Meta Quest / PICO headsets there is no bridge process to stand up. :class:WujiGlove connects, subscribes to the streams you ask for, and keeps the newest frame of each so every getter is a cheap read that a sim loop can call once per step.

What it exposes

• hand pose — the 21 MediaPipe landmarks (:func:~eden.utils.devices.wuji.protocol.skeleton_to_arrays), the IK solver's 21 joint angles, and the 5 fingertip poses.
• tactile — the 24x31 palm-and-finger pressure sheet, its zone-grouped view, the calibrated binary/residual contact streams, and the 3D tactile point cloud.

Poses are returned in Eden's hand frame (x forward along the fingers, y left, z up out of the back of the hand) by default, so an axis means the same physical direction on either glove — the vendor's +y flips between dorsal and ventral by side. See :mod:eden.utils.devices.wuji.protocol for the basis change and the data layouts. Pass frame="wuji" to any getter for the vendor's raw wrist frame instead.

Install the SDK with the wuji extra (uv sync --inexact --extra wuji); it ships manylinux wheels only, so this is Linux x86_64 / aarch64.

Example

.. code-block:: python

from eden.utils.devices.wuji import WujiGlove

glove = WujiGlove(side="right", streams=("hand_skeleton", "tactile"))
keypoints = glove.get_command()          # (21, 3) torch tensor, Eden hand frame
pressure = glove.tactile()               # (24, 31), masked taxels are nan
glove.close()

Attributes

STREAMS module-attribute

STREAMS = ('hand_skeleton', 'hand_joint_angles', 'tip_poses', 'emf_poses', 'tactile', 'tactile_zones', 'tactile_binary', 'tactile_residual', 'tactile_point_cloud')

DEFAULT_STREAMS module-attribute

DEFAULT_STREAMS = ('hand_skeleton', 'hand_joint_angles', 'tactile')

STREAM_RATE_SOURCE module-attribute

STREAM_RATE_SOURCE = {'emf_poses': 'emf_poses', 'hand_skeleton': 'emf_poses', 'hand_joint_angles': 'emf_poses', 'tip_poses': 'emf_poses', 'tactile': 'tactile', 'tactile_zones': 'tactile_zones', 'tactile_binary': 'tactile', 'tactile_residual': 'tactile', 'tactile_point_cloud': 'tactile'}

RATE_SOURCES module-attribute

RATE_SOURCES = tuple(dict.fromkeys(STREAM_RATE_SOURCE.values()))

TACTILE_SHEETS module-attribute

TACTILE_SHEETS = ('tactile', 'tactile_binary', 'tactile_residual')

CALIBRATED_STREAMS module-attribute

CALIBRATED_STREAMS = ('tactile_binary', 'tactile_residual')

STATE_KEYS module-attribute

STATE_KEYS = {'hand_skeleton': ('keypoints', 'keypoint_quat', 'keypoint_confidence'), 'tip_poses': ('fingertip_pos', 'fingertip_quat', 'fingertip_confidence'), 'emf_poses': ('emf_pos', 'emf_quat', 'emf_confidence'), 'hand_joint_angles': ('joint_angles', 'joint_angle_confidence'), 'tactile_point_cloud': ('tactile_points', 'tactile_pressure')}

DEVICE_NAME_PREFIX module-attribute

DEVICE_NAME_PREFIX = 'eden_wuji_glove'

Classes

WujiGlove

WujiGlove(*, sn: str | None = None, side: str | None = None, device_name: str | None = None, streams: Sequence[str] = DEFAULT_STREAMS, rate_hz: int | None = None, hand_model_path: str | None = None, exclusive: bool = False)

Bases: DeviceBase

A connected Wuji Glove, streaming hand pose and tactile into Eden.

Constructing the device scans for gloves, connects to one, and opens a background subscription per requested stream; call :meth:close when done (or use it as a context manager). Every getter returns None until that stream's first frame has arrived, so a teleop loop can hold its last pose instead of blocking.

PARAMETER	DESCRIPTION
sn	Serial number of the glove to connect to. None scans and picks the only glove on the bus (or the one matching side). TYPE: str | None DEFAULT: None
side	"left" or "right". Used to pick between two connected gloves, and checked against the glove's own hand_side either way, so a mismatch raises at construction rather than silently flipping the radial axis of every pose. TYPE: str | None DEFAULT: None
device_name	Handle this connection is registered under inside the SDK. It selects no particular glove, but it does identify the connection to disconnect, so it must be unique per live device — None (the default) takes a fresh one from :func:_next_device_name. Pass your own only if something outside Eden has to resolve the same connection. TYPE: str | None DEFAULT: None
streams	Which of :data:STREAMS to subscribe to. Defaults to :data:DEFAULT_STREAMS. TYPE: Sequence[str] DEFAULT: DEFAULT_STREAMS
rate_hz	Lower the device push rate of the rate-adjustable sources (:data:RATE_SOURCES) that streams needs. None keeps the device default (120 Hz). The device quantizes the request and the applied rate is logged. TYPE: int | None DEFAULT: None
hand_model_path	Custom hand URDF for the glove's online IK, applied before subscribing. Needs a named (non-default) SDK user; without one the SDK's built-in model is used. TYPE: str | None DEFAULT: None
exclusive	Take the glove for this process alone. By default the SDK bridges the device so Wuji Studio / a recorder can read it at the same time. TYPE: bool DEFAULT: False

Notes

tactile_binary and tactile_residual are produced by a contact model trained per SDK user and glove serial. Subscribing to them before running the vendor's tactile calibration yields an all-zero sheet, not an error, so this device logs a warning when you ask for one.

Source code in eden/utils/devices/wuji/glove.py

def __init__(
    self,
    *,
    sn: str | None = None,
    side: str | None = None,
    device_name: str | None = None,
    streams: Sequence[str] = DEFAULT_STREAMS,
    rate_hz: int | None = None,
    hand_model_path: str | None = None,
    exclusive: bool = False,
):
    super().__init__()

    if not streams:
        raise ValueError(
            f"'streams' is empty — the device would carry no data; pass at least one of {list(STREAMS)}"
        )
    unknown = [s for s in streams if s not in STREAMS]
    if unknown:
        raise ValueError(f"unknown glove stream(s) {unknown}; expected a subset of {list(STREAMS)}")
    if side is not None:
        P.check_side(side)
    if rate_hz is not None and rate_hz <= 0:
        raise ValueError(f"'rate_hz' must be positive (or None for the device default), got {rate_hz}")

    self._sdk = load_sdk()
    self._manager = self._sdk.SdkManager.instance()
    self._device_name = _next_device_name() if device_name is None else device_name
    # Seeded before the connect so teardown and __str__ can read them unconditionally,
    # including on the paths where the connect itself is what failed.
    self._sn = "?"
    self._side = "?"
    self._streams = tuple(dict.fromkeys(streams))  # de-duplicate, keep order
    self._latest: dict[str, Any] = {}
    self._lock = threading.Lock()
    self._subscriptions: list[Any] = []
    self._rate_handles: list[Any] = []
    self._closed = False
    self._connected = False

    options = self._sdk.ConnectOptions(enable_bridge=not exclusive)
    self._glove = self._connect(sn, side, options)
    self._sn = str(self._glove.serial_number)
    self._side = P.check_side(str(self._glove.hand_side().get()).lower())
    if side is not None and self._side != side:
        self.close()  # not `disconnect`: mark the device closed so __del__ doesn't retry
        raise RuntimeError(f"glove {self._sn} reports hand_side {self._side!r}, but {side!r} was requested")

    en.logger.info(f"Wuji Glove {self._sn} connected ({self._side} hand, FW {self._glove.version().get()})")

    if hand_model_path is not None:
        self._glove.hand_model_path().set(str(hand_model_path))
        en.logger.info(f"Wuji Glove {self._sn} online-IK hand model: {self._glove.hand_model_path().get()}")

    for stream in CALIBRATED_STREAMS:
        if stream in self._streams:
            en.logger.warning(
                f"'{stream}' comes from a contact model trained per SDK user and glove serial. "
                "Without a tactile calibration for this pair the stream stays at zero — run the "
                "vendor's tactile calibration (wuji-sdk examples/python/wuji_glove/7.tactile_calibration.py)."
            )

    try:
        if rate_hz is not None:
            self._set_rate(rate_hz)
        for stream in self._streams:
            resource = getattr(self._glove, stream)()
            self._subscriptions.append(resource.subscribe_with_callback(partial(self._on_frame, stream)))
    except BaseException:
        self.close()
        raise

Attributes

serial_number property

serial_number: str

Serial number of the connected glove.

side property

side: str

"left" or "right", as the glove reports it.

streams property

streams: tuple[str, ...]

The streams this device subscribed to.

tracking property

tracking: bool

True once a hand-pose frame has arrived (the glove is worn and solving).

emf_poses is left out on purpose: the raw coil poses stream whether or not the IK has solved a hand, so they answer "is the glove powered", not "is it tracking".

Methods:

keypoints

keypoints(frame: str = 'hand') -> Tensor | None

(21, 3) MediaPipe landmark positions in metres, or None before the first frame.

Landmarks are in :data:~eden.utils.devices.wuji.protocol.HAND_LANDMARKS order, wrist-relative, which is exactly what DexRetargeter and the vendor's RetargetSession consume.

Source code in eden/utils/devices/wuji/glove.py

def keypoints(self, frame: str = "hand") -> torch.Tensor | None:
    """``(21, 3)`` MediaPipe landmark positions in metres, or None before the first frame.

    Landmarks are in :data:`~eden.utils.devices.wuji.protocol.HAND_LANDMARKS` order,
    wrist-relative, which is exactly what ``DexRetargeter`` and the vendor's
    ``RetargetSession`` consume.
    """
    poses = self.keypoint_poses(frame)
    return None if poses is None else poses[0]

keypoint_poses

keypoint_poses(frame: str = 'hand') -> tuple[Tensor, Tensor, Tensor] | None

(pos (21, 3), quat (21, 4) wxyz, confidence (21,)), or None before the first frame.

Source code in eden/utils/devices/wuji/glove.py

def keypoint_poses(self, frame: str = "hand") -> tuple[torch.Tensor, torch.Tensor, torch.Tensor] | None:
    """``(pos (21, 3), quat (21, 4) wxyz, confidence (21,))``, or None before the first frame."""
    return self._decode("hand_skeleton", P.skeleton_to_arrays, self._side, frame)

joint_angle_state

joint_angle_state() -> tuple[Tensor, Tensor] | None

(angles (21,) in radians, confidence (5,)), or None before the first frame.

Both halves come out of one decode; :meth:joint_angles is the view onto the angles alone, which is what a retargeter reads.

Source code in eden/utils/devices/wuji/glove.py

def joint_angle_state(self) -> tuple[torch.Tensor, torch.Tensor] | None:
    """``(angles (21,) in radians, confidence (5,))``, or None before the first frame.

    Both halves come out of one decode; :meth:`joint_angles` is the view onto the
    angles alone, which is what a retargeter reads.
    """
    return self._decode("hand_joint_angles", P.pack_joint_angles)

joint_angles

joint_angles() -> Tensor | None

The IK solver's (21,) joint angles in radians, or None before the first frame.

Packed finger-major in :data:~eden.utils.devices.wuji.protocol.FINGERS order: 5 thumb DOFs (CMC x3, MCP, IP) then 4 per finger (MCP x2, PIP, DIP). This is the human hand's DOFs, not a robot's — retarget from :meth:keypoints to drive a robot hand.

Source code in eden/utils/devices/wuji/glove.py

def joint_angles(self) -> torch.Tensor | None:
    """The IK solver's ``(21,)`` joint angles in radians, or None before the first frame.

    Packed finger-major in :data:`~eden.utils.devices.wuji.protocol.FINGERS` order:
    5 thumb DOFs (CMC x3, MCP, IP) then 4 per finger (MCP x2, PIP, DIP). This is the
    *human* hand's DOFs, not a robot's — retarget from :meth:`keypoints` to drive a
    robot hand.
    """
    state = self.joint_angle_state()
    return None if state is None else state[0]

fingertips

fingertips(frame: str = 'hand') -> tuple[Tensor, Tensor, Tensor] | None

(pos (5, 3), quat (5, 4) wxyz, confidence (5,)) fingertip poses, or None.

Thumb first (:data:~eden.utils.devices.wuji.protocol.FINGERS order), from the glove's IK->FK chain.

Source code in eden/utils/devices/wuji/glove.py

def fingertips(self, frame: str = "hand") -> tuple[torch.Tensor, torch.Tensor, torch.Tensor] | None:
    """``(pos (5, 3), quat (5, 4) wxyz, confidence (5,))`` fingertip poses, or None.

    Thumb first (:data:`~eden.utils.devices.wuji.protocol.FINGERS` order), from the
    glove's IK->FK chain.
    """
    return self._decode("tip_poses", P.fingertips_to_arrays, self._side, frame)

emf_poses

emf_poses(frame: str = 'hand') -> tuple[Tensor, Tensor, Tensor] | None

(pos (5, 3), quat (5, 4) wxyz, confidence (5,)) raw EMF coil poses, or None.

The unfiltered sensor product behind the skeleton. Confidence below 0.9 is the vendor's threshold for "unreliable".

Source code in eden/utils/devices/wuji/glove.py

def emf_poses(self, frame: str = "hand") -> tuple[torch.Tensor, torch.Tensor, torch.Tensor] | None:
    """``(pos (5, 3), quat (5, 4) wxyz, confidence (5,))`` raw EMF coil poses, or None.

    The unfiltered sensor product behind the skeleton. Confidence below 0.9 is the
    vendor's threshold for "unreliable".
    """
    return self._decode("emf_poses", P.emf_poses_to_arrays, self._side, frame)

tactile

tactile() -> Tensor | None

(24, 31) calibrated pressure in [0, 1], or None before the first frame.

Row-major over the palm-and-finger sheet. The cells the sheet does not populate are nan (see :func:~eden.utils.devices.wuji.protocol.tactile_grid), so reduce with torch.nanmean / nan_to_num rather than plain mean.

Source code in eden/utils/devices/wuji/glove.py

def tactile(self) -> torch.Tensor | None:
    """``(24, 31)`` calibrated pressure in ``[0, 1]``, or None before the first frame.

    Row-major over the palm-and-finger sheet. The cells the sheet does not populate
    are ``nan`` (see :func:`~eden.utils.devices.wuji.protocol.tactile_grid`), so reduce
    with ``torch.nanmean`` / ``nan_to_num`` rather than plain ``mean``.
    """
    return self._grid("tactile")

tactile_binary

tactile_binary() -> Tensor | None

(24, 31) contact mask (1 = contact, nan = masked taxel), or None.

Needs a tactile calibration for this SDK user + glove serial; without one the sheet reads all zero.

Source code in eden/utils/devices/wuji/glove.py

def tactile_binary(self) -> torch.Tensor | None:
    """``(24, 31)`` contact mask (1 = contact, ``nan`` = masked taxel), or None.

    Needs a tactile calibration for this SDK user + glove serial; without one the
    sheet reads all zero.
    """
    return self._grid("tactile_binary")

tactile_residual

tactile_residual() -> Tensor | None

(24, 31) signed baseline-relative contact signal, or None.

Positive = pressed harder than the calibrated baseline, ~0 = no contact. The continuous signal behind :meth:tactile_binary; you own the contact threshold. Masked taxels are nan. Needs the same calibration.

Source code in eden/utils/devices/wuji/glove.py

def tactile_residual(self) -> torch.Tensor | None:
    """``(24, 31)`` signed baseline-relative contact signal, or None.

    Positive = pressed harder than the calibrated baseline, ~0 = no contact. The
    continuous signal behind :meth:`tactile_binary`; you own the contact threshold.
    Masked taxels are ``nan``. Needs the same calibration.
    """
    return self._grid("tactile_residual")

tactile_taxels

tactile_taxels(stream: str = 'tactile', flat: bool = False) -> tuple[Tensor, Tensor] | None

(index, value) for just the taxels stream actually measured, or None.

The node-wise view: one entry per live taxel, dead ones dropped rather than nan-masked. index is (N, 2) (row, col) into the 24 x 31 sheet, or (N,) row-major ids when flat=True; value is that taxel's reading. stream picks which sheet to read — tactile, tactile_binary or tactile_residual — and must be one this device subscribed to.

See :func:~eden.utils.devices.wuji.protocol.tactile_taxels for why N is not the point cloud's point count.

Source code in eden/utils/devices/wuji/glove.py

def tactile_taxels(self, stream: str = "tactile", flat: bool = False) -> tuple[torch.Tensor, torch.Tensor] | None:
    """``(index, value)`` for just the taxels ``stream`` actually measured, or None.

    The node-wise view: one entry per live taxel, dead ones dropped rather than
    ``nan``-masked. ``index`` is ``(N, 2)`` ``(row, col)`` into the ``24 x 31`` sheet,
    or ``(N,)`` row-major ids when ``flat=True``; ``value`` is that taxel's reading.
    ``stream`` picks which sheet to read — ``tactile``, ``tactile_binary`` or
    ``tactile_residual`` — and must be one this device subscribed to.

    See :func:`~eden.utils.devices.wuji.protocol.tactile_taxels` for why ``N`` is not
    the point cloud's point count.
    """
    if stream not in TACTILE_SHEETS:
        raise ValueError(f"'{stream}' is not a tactile sheet; expected one of {list(TACTILE_SHEETS)}")
    frame = self._take(stream)
    if frame is None:
        return None
    index, value = P.tactile_taxels(frame.data, flat)
    return torch.from_numpy(index), torch.from_numpy(value).float()

tactile_zones

tactile_zones() -> dict[str, Tensor] | None

Pressure grouped by hand region — {"palm", "thumb", ..., "pinky"} — or None.

Each value is a 1-D tensor of that zone's taxels (lengths in :data:~eden.utils.devices.wuji.protocol.TACTILE_ZONE_SIZES), padding as nan. Cheaper to reason about than the full sheet when you only need "is the thumb touching something".

Source code in eden/utils/devices/wuji/glove.py

def tactile_zones(self) -> dict[str, torch.Tensor] | None:
    """Pressure grouped by hand region — ``{"palm", "thumb", ..., "pinky"}`` — or None.

    Each value is a 1-D tensor of that zone's taxels (lengths in
    :data:`~eden.utils.devices.wuji.protocol.TACTILE_ZONE_SIZES`), padding as ``nan``.
    Cheaper to reason about than the full sheet when you only need "is the thumb
    touching something".
    """
    zones = self._take("tactile_zones")
    if zones is None:
        return None
    return {k: torch.from_numpy(v).float() for k, v in P.tactile_zones_to_dict(zones).items()}

tactile_points

tactile_points(frame: str = 'hand') -> tuple[Tensor, Tensor] | None

(points (N, 3), pressure (N,)) for the 3D tactile point cloud, or None.

The 526 real taxels placed on the live hand skeleton, so contact can be drawn in the scene or fed to a contact-conditioned policy.

Source code in eden/utils/devices/wuji/glove.py

def tactile_points(self, frame: str = "hand") -> tuple[torch.Tensor, torch.Tensor] | None:
    """``(points (N, 3), pressure (N,))`` for the 3D tactile point cloud, or None.

    The 526 real taxels placed on the live hand skeleton, so contact can be drawn in
    the scene or fed to a contact-conditioned policy.
    """
    cloud = self._take("tactile_point_cloud")
    if cloud is None:
        return None
    points, pressure = P.decode_point_cloud(cloud)
    if P.check_frame(frame) == "hand":
        points = P.wuji_points_to_hand(points, self._side)
    return torch.from_numpy(points).float(), torch.from_numpy(pressure).float()

get_raw_data

get_raw_data() -> Tensor

(21, 3) landmarks in the glove's own wrist frame, untransformed.

Zeros until the first hand_skeleton frame arrives, so the shape is stable for a caller that stacks readings.

Source code in eden/utils/devices/wuji/glove.py

def get_raw_data(self) -> torch.Tensor:
    """``(21, 3)`` landmarks in the glove's own wrist frame, untransformed.

    Zeros until the first ``hand_skeleton`` frame arrives, so the shape is stable for
    a caller that stacks readings.
    """
    return self._keypoints_or_zeros("wuji")

get_command

get_command() -> Tensor

(21, 3) landmarks in Eden's hand frame — the teleoperation command.

Feed straight into a retargeter (DexRetargeter, wuji_sdk.RetargetSession) to drive a robot hand. Zeros before the first frame, matching :meth:get_raw_data.

Source code in eden/utils/devices/wuji/glove.py

def get_command(self) -> torch.Tensor:
    """``(21, 3)`` landmarks in Eden's hand frame — the teleoperation command.

    Feed straight into a retargeter (``DexRetargeter``, ``wuji_sdk.RetargetSession``)
    to drive a robot hand. Zeros before the first frame, matching :meth:`get_raw_data`.
    """
    return self._keypoints_or_zeros("hand")

state

state(frame: str = 'hand') -> dict

JSON-safe snapshot of every subscribed stream, for logging or a ZMQ bridge.

Only the streams this device subscribed to appear. Tactile sheets are published as the raw flat 744-value payload with the vendor's :data:~eden.utils.devices.wuji.protocol.TACTILE_INVALID sentinel intact rather than the nan-masked grid, because nan is not valid JSON.

Source code in eden/utils/devices/wuji/glove.py

def state(self, frame: str = "hand") -> dict:
    """JSON-safe snapshot of every subscribed stream, for logging or a ZMQ bridge.

    Only the streams this device subscribed to appear. Tactile sheets are published
    as the raw flat 744-value payload with the vendor's
    :data:`~eden.utils.devices.wuji.protocol.TACTILE_INVALID` sentinel intact rather
    than the ``nan``-masked grid, because ``nan`` is not valid JSON.
    """
    P.check_frame(frame)
    state: dict[str, Any] = {
        "device": "wuji_glove",
        "sn": self._sn,
        "side": self._side,
        "frame": frame,
        "ready": self.tracking,
    }

    # Every stream that decodes to tensors, as (stream, a thunk reading it). A thunk
    # rather than a getter plus args, so the one stream with no frame of its own —
    # joint angles, already anatomical — sits in the same table as the rest.
    for stream, read in (
        ("hand_skeleton", lambda: self.keypoint_poses(frame)),
        ("tip_poses", lambda: self.fingertips(frame)),
        ("emf_poses", lambda: self.emf_poses(frame)),
        ("hand_joint_angles", self.joint_angle_state),
        ("tactile_point_cloud", lambda: self.tactile_points(frame)),
    ):
        if stream in self._streams and (tensors := read()) is not None:
            # strict: STATE_KEYS is the published wire format, so an arity drift has to
            # raise here rather than ship a snapshot quietly missing a field.
            state.update(zip(STATE_KEYS[stream], (t.tolist() for t in tensors), strict=True))

    for stream in TACTILE_SHEETS:
        if stream in self._streams and (sheet := self._take(stream)) is not None:
            state[stream] = [float(v) for v in sheet.data]

    if "tactile_zones" in self._streams and (zones := self._take("tactile_zones")) is not None:
        state["tactile_zones"] = {zone: [float(v) for v in getattr(zones, zone)] for zone in P.TACTILE_ZONE_SIZES}

    return state

reset

reset() -> None

Drop every buffered frame, so the getters report None until fresh data lands.

Source code in eden/utils/devices/wuji/glove.py

def reset(self) -> None:
    """Drop every buffered frame, so the getters report None until fresh data lands."""
    with self._lock:
        self._latest.clear()

close

close() -> None

Close every subscription and disconnect the glove. Idempotent.

Source code in eden/utils/devices/wuji/glove.py

def close(self) -> None:
    """Close every subscription and disconnect the glove. Idempotent."""
    if self._closed:
        return
    self._closed = True
    for handle in (*self._subscriptions, *self._rate_handles):
        try:
            handle.close()
        except Exception as exc:  # pragma: no cover - teardown of a dead link
            en.logger.warning(f"Wuji Glove {self._sn}: failed to close a subscription: {exc}")
    self._subscriptions.clear()
    self._rate_handles.clear()
    if not self._connected:
        return  # the connect itself is what failed; there is no link to release
    try:
        self._manager.disconnect(self._device_name)
    except Exception as exc:  # pragma: no cover - teardown of a dead link
        en.logger.warning(f"Wuji Glove {self._sn}: failed to disconnect: {exc}")

Functions:

load_sdk

load_sdk()

Import wuji_sdk, raising an install hint instead of a bare ImportError.

Source code in eden/utils/devices/wuji/glove.py

def load_sdk():
    """Import ``wuji_sdk``, raising an install hint instead of a bare ImportError."""
    try:
        import wuji_sdk
    except ImportError as exc:  # pragma: no cover - depends on the optional extra
        raise ImportError(_SDK_HINT) from exc
    return wuji_sdk

scan_gloves

scan_gloves() -> list[dict]

List every Wuji Glove on the bus as {"sn", "address", "transport"} dicts.

A connection-free probe: use it to find a serial to pass to :class:WujiGlove. The glove's side is not discoverable here — it takes a connection to read.

Source code in eden/utils/devices/wuji/glove.py

def scan_gloves() -> list[dict]:
    """List every Wuji Glove on the bus as ``{"sn", "address", "transport"}`` dicts.

    A connection-free probe: use it to find a serial to pass to :class:`WujiGlove`. The
    glove's side is not discoverable here — it takes a connection to read.
    """
    sdk = load_sdk()
    manager = sdk.SdkManager.instance()
    return [
        {"sn": str(d.sn), "address": str(d.address), "transport": str(d.transport_type)}
        for d in _gloves_among(manager.scan(), sdk)
    ]

Wuji Glove data layout: stream shapes, landmark tables, and the basis change into Eden.

Everything here is pure numpy and duck-typed over the SDK message objects, so the whole decode path is testable without a glove (or without wuji_sdk installed). :mod:eden.utils.devices.wuji.glove owns the connection and the subscriptions.

Frames

Wuji publishes hand data in the wrist frame (l_wrist / r_wrist), which the vendor defines anatomically (docs, Coordinate Frames)::

x = radial     (ulnar -> radial, i.e. toward the thumb)
z = proximal   (distal -> proximal, i.e. along the bone toward the elbow)
y = right-hand rule, so it MIRRORS between hands:
    left glove  +y = dorsal (back of the hand)
    right glove +y = ventral (palm)

So +y means the opposite physical direction on the two gloves. Eden pins every axis to anatomy instead, in the Genesis basis — x forward along the fingers, y left, z up — so :data:R_HAND_WUJI maps::

x_eden = -z_wuji                                  (distal, out along the fingers)
z_eden = +y_wuji (left) / -y_wuji (right)         (dorsal, out of the back of the hand)
y_eden = z_eden x x_eden = -x_wuji (left) / +x_wuji (right)

Both per-side matrices are proper rotations (det = +1), and the per-side sign flip is exactly the vendor's own note that "left-hand glove data is mirrored from the right hand, with the Y direction inverted". Note what this does not buy: y is radial on a right hand and ulnar on a left one, because a left hand is a mirrored right hand and no right-handed frame can make the two agree. x (distal) and z (dorsal) do agree, which is what a retargeter and a pose-plotting recipe actually read.

Tactile

The tactile matrix is a fixed 24 x 31 row-major grid of 744 taxels, only some of which are real; the rest are permanently pinned to :data:TACTILE_INVALID (-1.0). Two counts describe "real" and they disagree by eight — read :data:N_ACTIVE_TAXELS and :data:N_UNMASKED_TAXELS before hard-coding either, here or in prose. That sentinel is a number, so grid.mean() over a raw frame is silently wrong — :func:tactile_grid therefore turns masked taxels into nan, which fails loudly instead. Use :func:tactile_valid for the boolean mask, or :func:tactile_taxels for the node-wise view: one (index, value) entry per taxel that actually measured something, with the dead ones dropped rather than masked.

Attributes

FINGERS module-attribute

FINGERS = ('thumb', 'index', 'middle', 'ring', 'pinky')

HAND_LANDMARKS module-attribute

HAND_LANDMARKS = ('wrist', 'thumb_cmc', 'thumb_mcp', 'thumb_ip', 'thumb_tip', 'index_finger_mcp', 'index_finger_pip', 'index_finger_dip', 'index_finger_tip', 'middle_finger_mcp', 'middle_finger_pip', 'middle_finger_dip', 'middle_finger_tip', 'ring_finger_mcp', 'ring_finger_pip', 'ring_finger_dip', 'ring_finger_tip', 'pinky_mcp', 'pinky_pip', 'pinky_dip', 'pinky_tip')

N_LANDMARKS module-attribute

N_LANDMARKS = len(HAND_LANDMARKS)

FINGERTIP_LANDMARKS module-attribute

FINGERTIP_LANDMARKS = tuple(i for i, name in enumerate(HAND_LANDMARKS) if name.endswith('_tip'))

HAND_CONNECTIONS module-attribute

HAND_CONNECTIONS = ((0, 1), (0, 5), (0, 17), (5, 9), (9, 13), (13, 17), (1, 2), (2, 3), (3, 4), (5, 6), (6, 7), (7, 8), (9, 10), (10, 11), (11, 12), (13, 14), (14, 15), (15, 16), (17, 18), (18, 19), (19, 20))

JOINT_ANGLE_SLOTS module-attribute

JOINT_ANGLE_SLOTS = 5

FINGER_NUM_DOFS module-attribute

FINGER_NUM_DOFS = (5, 4, 4, 4, 4)

N_HAND_DOFS module-attribute

N_HAND_DOFS = sum(FINGER_NUM_DOFS)

TACTILE_ROWS module-attribute

TACTILE_ROWS = 24

TACTILE_COLS module-attribute

TACTILE_COLS = 31

N_TAXELS module-attribute

N_TAXELS = TACTILE_ROWS * TACTILE_COLS

TACTILE_INVALID module-attribute

TACTILE_INVALID = -1.0

N_ACTIVE_TAXELS module-attribute

N_ACTIVE_TAXELS = 526

N_UNMASKED_TAXELS module-attribute

N_UNMASKED_TAXELS = 534

TACTILE_ZONE_SIZES module-attribute

TACTILE_ZONE_SIZES = {'palm': 308, 'thumb': 50, 'index': 45, 'middle': 60, 'ring': 54, 'pinky': 55}

FLOAT32_BYTES module-attribute

FLOAT32_BYTES = 4

POINT_CLOUD_FIELDS module-attribute

POINT_CLOUD_FIELDS = ('x', 'y', 'z', 'pressure')

POINT_CLOUD_STRIDE module-attribute

POINT_CLOUD_STRIDE = FLOAT32_BYTES * len(POINT_CLOUD_FIELDS)

POINT_FIELD_FLOAT32 module-attribute

POINT_FIELD_FLOAT32 = 7

R_HAND_WUJI module-attribute

R_HAND_WUJI = {'left': np.array([[0.0, 0.0, -1.0], [-1.0, 0.0, 0.0], [0.0, 1.0, 0.0]]), 'right': np.array([[0.0, 0.0, -1.0], [1.0, 0.0, 0.0], [0.0, -1.0, 0.0]])}

Q_HAND_WUJI module-attribute

Q_HAND_WUJI = {side: R_to_quat(R) for side, R in R_HAND_WUJI.items()}

SIDES module-attribute

SIDES = tuple(R_HAND_WUJI)

FRAMES module-attribute

FRAMES = ('hand', 'wuji')

SIDE_FROM_SN_LETTER module-attribute

SIDE_FROM_SN_LETTER = {'J': 'left', 'K': 'right'}

HAND_TO_MANO module-attribute

HAND_TO_MANO = np.array([[0.0, 0.0, 1.0], [0.0, 1.0, 0.0], [-1.0, 0.0, 0.0]])

Functions:

side_from_sn

side_from_sn(sn: str) -> str | None

Guess a glove's side from its serial number, or None if the letter is unfamiliar.

Lets a caller connect to the likely glove first instead of probing every one, without ever trusting the guess: whoever uses it must still confirm against hand_side().

Source code in eden/utils/devices/wuji/protocol.py

def side_from_sn(sn: str) -> str | None:
    """Guess a glove's side from its serial number, or None if the letter is unfamiliar.

    Lets a caller connect to the likely glove first instead of probing every one, without
    ever trusting the guess: whoever uses it must still confirm against ``hand_side()``.
    """
    letter = sn[3].upper() if len(sn) > 3 else ""
    return SIDE_FROM_SN_LETTER.get(letter)

check_side

check_side(side: str) -> str

Return side if it is "left" or "right", else raise naming both.

Source code in eden/utils/devices/wuji/protocol.py

def check_side(side: str) -> str:
    """Return ``side`` if it is ``"left"`` or ``"right"``, else raise naming both."""
    return _one_of(side, SIDES, "glove side")

check_frame

check_frame(frame: str) -> str

Return frame if it is "hand" or "wuji", else raise naming both.

Source code in eden/utils/devices/wuji/protocol.py

def check_frame(frame: str) -> str:
    """Return ``frame`` if it is ``"hand"`` or ``"wuji"``, else raise naming both."""
    return _one_of(frame, FRAMES, "frame")

wuji_points_to_hand

wuji_points_to_hand(points: ArrayLike, side: str) -> NDArray[float64]

Rotate (N, 3) wrist-frame points into the Eden hand frame for side.

Source code in eden/utils/devices/wuji/protocol.py

def wuji_points_to_hand(points: npt.ArrayLike, side: str) -> npt.NDArray[np.float64]:
    """Rotate ``(N, 3)`` wrist-frame points into the Eden hand frame for ``side``."""
    pts = np.asarray(points, dtype=float)
    if pts.ndim != 2 or pts.shape[1] != 3:
        raise ValueError(f"expected (N, 3) points, got shape {pts.shape}")
    return transform_by_R(pts, R_HAND_WUJI[check_side(side)])

wuji_pose_to_hand

wuji_pose_to_hand(position: ArrayLike, orientation_xyzw: ArrayLike, side: str) -> tuple[NDArray[float64], NDArray[float64]]

Convert one wrist-frame pose into the Eden hand frame.

PARAMETER	DESCRIPTION
position	(x, y, z) in metres, as wuji_sdk.Pose.position reports it. TYPE: ArrayLike
orientation_xyzw	The pose's wuji_sdk.Quaternion as (x, y, z, w) — the SDK's scalar-last order, not Genesis' wxyz. TYPE: ArrayLike
side	"left" or "right". TYPE: str

RETURNS	DESCRIPTION
tuple	(pos (3,), quat (4,)) in the hand frame, the quaternion wxyz as everything Genesis-facing expects. A degenerate (zero-norm) input quaternion — what the glove reports for a pose it has not solved — comes back as identity.

Source code in eden/utils/devices/wuji/protocol.py

def wuji_pose_to_hand(
    position: npt.ArrayLike, orientation_xyzw: npt.ArrayLike, side: str
) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Convert one wrist-frame pose into the Eden hand frame.

    Parameters
    ----------
    position:
        ``(x, y, z)`` in metres, as ``wuji_sdk.Pose.position`` reports it.
    orientation_xyzw:
        The pose's ``wuji_sdk.Quaternion`` as ``(x, y, z, w)`` — the SDK's scalar-last
        order, *not* Genesis' wxyz.
    side:
        ``"left"`` or ``"right"``.

    Returns
    -------
    tuple
        ``(pos (3,), quat (4,))`` in the hand frame, the quaternion **wxyz** as
        everything Genesis-facing expects. A degenerate (zero-norm) input quaternion —
        what the glove reports for a pose it has not solved — comes back as identity.
    """
    side = check_side(side)
    pos = np.asarray(position, dtype=float).reshape(-1)
    if pos.shape != (3,):
        raise ValueError(f"expected a 3-element position, got shape {pos.shape}")
    out_pos, out_quat = _poses_to_hand(pos[None], _quat_wxyz(orientation_xyzw)[None], side)
    return out_pos[0], out_quat[0]

keypoints_to_mano

keypoints_to_mano(keypoints: ArrayLike) -> NDArray[float64]

Hand-frame (21, 3) landmarks -> wrist-origin MANO keypoints for a retargeter.

Feed the result to DexRetargeter.extract_ref_value exactly as you would a MediaPipe detection's joint_pos. Pass keypoints in Eden's hand frame (the default :meth:~eden.utils.devices.wuji.glove.WujiGlove.keypoints output), not the vendor's wrist frame.

Source code in eden/utils/devices/wuji/protocol.py

def keypoints_to_mano(keypoints: npt.ArrayLike) -> npt.NDArray[np.float64]:
    """Hand-frame ``(21, 3)`` landmarks -> wrist-origin MANO keypoints for a retargeter.

    Feed the result to ``DexRetargeter.extract_ref_value`` exactly as you would a
    MediaPipe detection's ``joint_pos``. Pass keypoints in Eden's hand frame (the default
    :meth:`~eden.utils.devices.wuji.glove.WujiGlove.keypoints` output), not the vendor's
    wrist frame.
    """
    kp = np.asarray(keypoints, dtype=float)
    if kp.shape != (N_LANDMARKS, 3):
        raise ValueError(f"expected ({N_LANDMARKS}, 3) hand-frame keypoints, got shape {kp.shape}")
    return (kp - kp[0]) @ HAND_TO_MANO

skeleton_to_arrays

skeleton_to_arrays(skeleton: Any, side: str, frame: str = 'hand') -> tuple[NDArray[float64], NDArray[float64], NDArray[float64]]

Decode a HandSkeleton frame into (pos (21, 3), quat (21, 4), confidence (21,)).

Landmarks come back in :data:HAND_LANDMARKS (MediaPipe) order, positions in metres. frame="hand" rotates into Eden's hand basis; frame="wuji" leaves the vendor's wrist basis alone.

Source code in eden/utils/devices/wuji/protocol.py

def skeleton_to_arrays(
    skeleton: Any, side: str, frame: str = "hand"
) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Decode a ``HandSkeleton`` frame into ``(pos (21, 3), quat (21, 4), confidence (21,))``.

    Landmarks come back in :data:`HAND_LANDMARKS` (MediaPipe) order, positions in metres.
    ``frame="hand"`` rotates into Eden's hand basis; ``frame="wuji"`` leaves the vendor's
    wrist basis alone.
    """
    return _poses_to_arrays(skeleton.joints, side, frame, N_LANDMARKS, "skeleton joints")

fingertips_to_arrays

fingertips_to_arrays(tips: Any, side: str, frame: str = 'hand') -> tuple[NDArray[float64], NDArray[float64], NDArray[float64]]

Decode a FingertipPoses frame into (pos (5, 3), quat (5, 4), confidence (5,)).

Fingers are in :data:FINGERS order (thumb first).

Source code in eden/utils/devices/wuji/protocol.py

def fingertips_to_arrays(
    tips: Any, side: str, frame: str = "hand"
) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Decode a ``FingertipPoses`` frame into ``(pos (5, 3), quat (5, 4), confidence (5,))``.

    Fingers are in :data:`FINGERS` order (thumb first).
    """
    return _poses_to_arrays(tips.poses, side, frame, len(FINGERS), "fingertip poses")

emf_poses_to_arrays

emf_poses_to_arrays(emf: Any, side: str, frame: str = 'hand') -> tuple[NDArray[float64], NDArray[float64], NDArray[float64]]

Decode an EmfPoseArray frame into (pos (5, 3), quat (5, 4), confidence (5,)).

These are the raw receiver-coil poses in :data:FINGERS order. They are reported in the emf_tx (back-of-hand transmitter) frame, which is a pure translation off the wrist frame, so the rotation applied here is the wrist one — the coil offset is not removed. Confidence below 0.9 means the vendor considers the estimate unreliable.

Source code in eden/utils/devices/wuji/protocol.py

def emf_poses_to_arrays(
    emf: Any, side: str, frame: str = "hand"
) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Decode an ``EmfPoseArray`` frame into ``(pos (5, 3), quat (5, 4), confidence (5,))``.

    These are the raw receiver-coil poses in :data:`FINGERS` order. They are reported in
    the ``emf_tx`` (back-of-hand transmitter) frame, which is a pure translation off the
    wrist frame, so the rotation applied here is the wrist one — the coil offset is *not*
    removed. Confidence below 0.9 means the vendor considers the estimate unreliable.
    """
    return _poses_to_arrays(emf.poses, side, frame, len(FINGERS), "EMF poses")

pack_joint_angles

pack_joint_angles(angles: Any) -> tuple[NDArray[float64], NDArray[float64]]

Pack a HandJointAngles frame's 5x5 padded array into the 21 real DOFs.

Returns (angles (21,) in radians, confidence (5,)). The thumb contributes all five of its slots, every other finger the first four (:data:FINGER_NUM_DOFS); the padding slot is dropped rather than passed on as a fake zero-angle joint.

Source code in eden/utils/devices/wuji/protocol.py

def pack_joint_angles(angles: Any) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Pack a ``HandJointAngles`` frame's 5x5 padded array into the 21 real DOFs.

    Returns ``(angles (21,) in radians, confidence (5,))``. The thumb contributes all
    five of its slots, every other finger the first four (:data:`FINGER_NUM_DOFS`); the
    padding slot is dropped rather than passed on as a fake zero-angle joint.
    """
    fingers = angles.fingers
    if len(fingers) != len(FINGERS):
        raise ValueError(f"expected {len(FINGERS)} fingers of joint angles, got {len(fingers)}")
    packed = []
    conf = np.empty(len(FINGERS))
    for i, (source, finger, n_dofs) in enumerate(zip(fingers, FINGERS, FINGER_NUM_DOFS)):
        row = np.asarray(source.angles, dtype=float).reshape(-1)
        if row.shape != (JOINT_ANGLE_SLOTS,):
            raise ValueError(f"expected {JOINT_ANGLE_SLOTS} angle slots for the {finger}, got {row.shape[0]}")
        packed.append(row[:n_dofs])
        conf[i] = float(source.confidence)
    return np.concatenate(packed), conf  # FINGER_NUM_DOFS sums to N_HAND_DOFS

tactile_valid

tactile_valid(data: Sequence[float]) -> NDArray[bool_]

(24, 31) boolean mask of the taxels this frame actually measured.

Source code in eden/utils/devices/wuji/protocol.py

def tactile_valid(data: Sequence[float]) -> npt.NDArray[np.bool_]:
    """``(24, 31)`` boolean mask of the taxels this frame actually measured."""
    return (_tactile_flat(data) != TACTILE_INVALID).reshape(TACTILE_ROWS, TACTILE_COLS)

tactile_grid

tactile_grid(data: Sequence[float]) -> NDArray[float64]

Reshape a flat tactile payload into the (24, 31) row-major grid, masked to nan.

Works for every sheet-shaped stream — tactile (pressure in [0, 1]), tactile_binary (1 = contact), and tactile_residual (signed, baseline-relative). The cells the sheet does not populate arrive as :data:TACTILE_INVALID and become nan here, so a stray .mean() reports nan instead of a plausible-looking number biased toward -1.

Source code in eden/utils/devices/wuji/protocol.py

def tactile_grid(data: Sequence[float]) -> npt.NDArray[np.float64]:
    """Reshape a flat tactile payload into the ``(24, 31)`` row-major grid, masked to ``nan``.

    Works for every sheet-shaped stream — ``tactile`` (pressure in ``[0, 1]``),
    ``tactile_binary`` (1 = contact), and ``tactile_residual`` (signed, baseline-relative).
    The cells the sheet does not populate arrive as :data:`TACTILE_INVALID` and become
    ``nan`` here, so a stray ``.mean()`` reports ``nan`` instead of a plausible-looking
    number biased toward -1.
    """
    grid = _tactile_flat(data).reshape(TACTILE_ROWS, TACTILE_COLS).copy()
    grid[grid == TACTILE_INVALID] = np.nan
    return grid

tactile_taxels

tactile_taxels(data: Sequence[float], flat: bool = False) -> tuple[NDArray[int64], NDArray[float64]]

One entry per taxel this frame actually measured: (index, value).

The node-wise view of a sheet-shaped stream. :func:tactile_grid keeps the full 24 x 31 layout with the dead taxels as nan; this drops them, so value is every real reading and nothing else — iterate it, reduce it, or feed it somewhere without first writing the nan filter yourself.

PARAMETER	DESCRIPTION
data	The stream's flat 744-value payload (TactileFrame.data and friends). TYPE: Sequence[float]
flat	False (default) indexes as (N, 2) (row, col) into the grid; True as (N,) row-major ids in [0, 744). TYPE: bool DEFAULT: False

RETURNS

DESCRIPTION

tuple

(index, value). index is an integer array — kept separate from value rather than packed into one float array, so a grid coordinate can never be read as a pressure. N is whatever the sheet reports this frame, which on firmware 0.11.1 is :data:N_UNMASKED_TAXELS and is deliberately not :data:N_ACTIVE_TAXELS — see those constants before joining against the point cloud, which carries the smaller count.

Source code in eden/utils/devices/wuji/protocol.py

def tactile_taxels(data: Sequence[float], flat: bool = False) -> tuple[npt.NDArray[np.int64], npt.NDArray[np.float64]]:
    """One entry per taxel this frame actually measured: ``(index, value)``.

    The node-wise view of a sheet-shaped stream. :func:`tactile_grid` keeps the full
    ``24 x 31`` layout with the dead taxels as ``nan``; this drops them, so ``value`` is
    every real reading and nothing else — iterate it, reduce it, or feed it somewhere
    without first writing the ``nan`` filter yourself.

    Parameters
    ----------
    data:
        The stream's flat 744-value payload (``TactileFrame.data`` and friends).
    flat:
        ``False`` (default) indexes as ``(N, 2)`` ``(row, col)`` into the grid;
        ``True`` as ``(N,)`` row-major ids in ``[0, 744)``.

    Returns
    -------
    tuple
        ``(index, value)``. ``index`` is an integer array — kept separate from ``value``
        rather than packed into one float array, so a grid coordinate can never be read
        as a pressure. ``N`` is whatever the sheet reports this frame, which on firmware
        0.11.1 is :data:`N_UNMASKED_TAXELS` and is deliberately *not*
        :data:`N_ACTIVE_TAXELS` — see those constants before joining against the point
        cloud, which carries the smaller count.
    """
    values = _tactile_flat(data)
    live = np.flatnonzero(values != TACTILE_INVALID)
    index = live if flat else np.stack(np.unravel_index(live, (TACTILE_ROWS, TACTILE_COLS)), axis=1)
    return index.astype(np.int64), values[live]

tactile_zones_to_dict

tactile_zones_to_dict(zones: Any) -> dict[str, NDArray[float64]]

Decode a TactileZones frame into {zone: (n,) pressures}, masked to nan.

Zones are :data:TACTILE_ZONE_SIZES (palm plus one per finger); the length of each array is fixed by the hardware, and a length mismatch raises rather than silently reindexing a caller's per-zone slice.

Source code in eden/utils/devices/wuji/protocol.py

def tactile_zones_to_dict(zones: Any) -> dict[str, npt.NDArray[np.float64]]:
    """Decode a ``TactileZones`` frame into ``{zone: (n,) pressures}``, masked to ``nan``.

    Zones are :data:`TACTILE_ZONE_SIZES` (palm plus one per finger); the length of each
    array is fixed by the hardware, and a length mismatch raises rather than silently
    reindexing a caller's per-zone slice.
    """
    out = {}
    for zone, size in TACTILE_ZONE_SIZES.items():
        # `np.array`, not `asarray`: masking is in place, and an SDK frame that already
        # hands back a float64 array would otherwise have its sentinels overwritten —
        # the same frame `WujiGlove.state` publishes raw. `tactile_grid` copies too.
        values = np.array(getattr(zones, zone), dtype=float).reshape(-1)
        if values.shape != (size,):
            raise ValueError(f"expected {size} taxels in the {zone} zone, got {values.shape[0]}")
        values[values == TACTILE_INVALID] = np.nan
        out[zone] = values
    return out

decode_point_cloud

decode_point_cloud(cloud: Any) -> tuple[NDArray[float64], NDArray[float64]]

Decode a PointCloud frame into (points (N, 3), pressure (N,)).

The payload is a packed byte buffer, so the field descriptors are validated against :data:POINT_CLOUD_FIELDS / :data:POINT_CLOUD_STRIDE first: a firmware change to the layout must raise here rather than reinterpret the bytes into plausible garbage. Points are in the glove's wrist frame — pass them through :func:wuji_points_to_hand for the Eden hand frame.

Source code in eden/utils/devices/wuji/protocol.py

def decode_point_cloud(cloud: Any) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Decode a ``PointCloud`` frame into ``(points (N, 3), pressure (N,))``.

    The payload is a packed byte buffer, so the field descriptors are validated against
    :data:`POINT_CLOUD_FIELDS` / :data:`POINT_CLOUD_STRIDE` first: a firmware change to
    the layout must raise here rather than reinterpret the bytes into plausible garbage.
    Points are in the glove's wrist frame — pass them through :func:`wuji_points_to_hand`
    for the Eden hand frame.
    """
    stride = int(cloud.point_stride)
    layout = [(str(f.name), int(f.offset), int(f.type)) for f in cloud.fields]
    expected = [(name, FLOAT32_BYTES * i, POINT_FIELD_FLOAT32) for i, name in enumerate(POINT_CLOUD_FIELDS)]
    if stride != POINT_CLOUD_STRIDE or layout != expected:
        raise ValueError(
            f"unexpected tactile point-cloud layout: stride={stride} fields={layout}; "
            f"this decoder handles stride={POINT_CLOUD_STRIDE} fields={expected} "
            "(x, y, z, pressure as float32)"
        )
    raw = np.frombuffer(bytes(cloud.data), dtype=np.float32)
    if raw.size % len(POINT_CLOUD_FIELDS):
        raise ValueError(
            f"point-cloud payload of {raw.size * FLOAT32_BYTES} bytes is not a whole number "
            f"of {POINT_CLOUD_STRIDE}-byte points"
        )
    values = raw.reshape(-1, len(POINT_CLOUD_FIELDS)).astype(float)
    # The header counts the points too; disagreeing with the payload means a truncated or
    # over-long frame, which would otherwise decode into a plausible-looking short cloud.
    count = int(cloud.point_count)
    if count != len(values):
        raise ValueError(f"point cloud says point_count={count} but carries {len(values)} points")
    return values[:, :3], values[:, 3]

eden wuji — find a Wuji Glove and watch its hand-pose / tactile streams live.

The first thing to run after plugging a glove in: it proves the SDK sees the device, that the glove is solving a hand pose, and that the tactile sheet responds to a touch, without building a Genesis scene.

Run::

eden wuji scan                                  # list the gloves on the bus
eden wuji monitor                               # live status, Ctrl-C to stop
eden wuji monitor --side right --seconds 30
eden wuji monitor --streams hand_skeleton tactile_zones --rate-hz 30
# or, equivalently:
uv run python -m eden.utils.devices.wuji.monitor monitor

Attributes

REFRESH_HZ module-attribute

REFRESH_HZ = 5.0

LOADED_THRESHOLD module-attribute

LOADED_THRESHOLD = 0.05

Classes

Functions:

main

main(argv=None) -> int

Run eden wuji scan or eden wuji monitor (parses argv, or sys.argv when None).

Source code in eden/utils/devices/wuji/monitor.py

def main(argv=None) -> int:
    """Run ``eden wuji scan`` or ``eden wuji monitor`` (parses ``argv``, or ``sys.argv`` when ``None``)."""
    args = _build_parser().parse_args(argv)
    return _run_scan() if args.command == "scan" else _run_monitor(args)

See also

• Teleoperation guide
• Teleoperation extensions