Scripted Demonstrations

Scripted demonstrators produce demo-like trajectories, the kind a person teleoperating would record: smooth minimum-jerk-timed motion through randomized keyframes, a per-episode speed, pauses, off-centre grasps, retries and deliberate mistakes the demonstrator recovers from. They are a port of OCBench's scripted policies onto Eden's batched envs. SCRIPTED_DEMO_REGISTRY says which demonstrator drives which task, and generate_demos (what eden datagen runs) rolls one out in parallel envs and records every episode. The scripted demonstrations guide covers eden datagen and writing a demonstrator.

from eden.extensions.scripted_demos.generate import generate_demos
from eden.extensions.scripted_demos.registry import SCRIPTED_DEMO_REGISTRY

Generation and registry

Roll a task's scripted demonstrator out in parallel envs and record every episode (eden datagen).

OCBench's data_gen_scripts/generate_ocbench.py, on Eden's recorder: the env records itself (:class:~eden.managers.recorder_manager.RecorderManager), so a demonstration is written in the same HDF5 layout as a teleoperated one and eden lerobot convert reads either.

A demonstrator is either a keyframe controller (:class:~eden.extensions.scripted_demos.controller.ScriptedPickPlaceController) or a tool routine (:class:~eden.extensions.scripted_demos.routines.RoutineDemonstrator, or one per arm composed by a :class:~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator); all expose reset / update_subgoals / act. A two-arm task records both arms' state, side by side. A routine demonstrator also keeps a per-episode trace, and the report carries the first failed episode's — the place to start when the success rate is low.

An episode ends when the demonstrator has finished the task — its success is the task's own success metric at that step — or when the env ends it (time limit, an object off the table). Every ended episode is exported with its success flag; export_mode decides whether the failed ones are kept. Generation stops once num_episodes have ended; the episodes still running then are dropped rather than written cut short.

Attributes

DEFAULT_EPISODE_LENGTH_S module-attribute

DEFAULT_EPISODE_LENGTH_S = 90.0

Classes

GenerationReport dataclass

GenerationReport(dataset_path: str, episodes: int = 0, successes: int = 0, cut_short: int = 0, steps: int = 0, wall_s: float = 0.0, subgoals: dict[str, int] = dict(), failure_trace: list[str] = list())

What a generation run produced.

ATTRIBUTE	DESCRIPTION
dataset_path	The dataset file the successful (or all) episodes went to. TYPE: str
episodes	Episodes that ended (exported or not, per export_mode). TYPE: int
successes	Of those, the ones the task's success metric passed. TYPE: int
cut_short	Episodes that ended unsuccessfully before the demonstrator finished: the time limit, or a task termination (an object off the table). A task that ends its episode on success (Wipe) cuts nothing short. TYPE: int
steps	Control steps recorded across all ended episodes. TYPE: int
wall_s	Wall-clock duration of the rollout. TYPE: float
subgoals	Subgoals started, by kind (goal, clear, mistake) — episodes still running when generation stopped included. For a routine demonstrator: motions ended, by <tool>:<status>. TYPE: dict[str, int]
failure_trace	A routine demonstrator's trace of the first episode that ended unsuccessfully (empty for a keyframe demonstrator, or when every episode succeeded). TYPE: list[str]

Attributes

dataset_path instance-attribute

dataset_path: str

episodes class-attribute instance-attribute

episodes: int = 0

successes class-attribute instance-attribute

successes: int = 0

cut_short class-attribute instance-attribute

cut_short: int = 0

steps class-attribute instance-attribute

steps: int = 0

wall_s class-attribute instance-attribute

wall_s: float = 0.0

subgoals class-attribute instance-attribute

subgoals: dict[str, int] = field(default_factory=dict)

failure_trace class-attribute instance-attribute

failure_trace: list[str] = field(default_factory=list)

success_rate property

success_rate: float

Methods:

render

render() -> str

Source code in eden/extensions/scripted_demos/generate.py

def render(self) -> str:
    return (
        f"{self.episodes} episodes, {self.successes} successful ({100 * self.success_rate:.1f} %), "
        f"{self.cut_short} cut short; {self.steps} steps in {self.wall_s:.0f} s; "
        f"subgoals {self.subgoals} -> {self.dataset_path}"
    )

Functions:

recorder_options

recorder_options(output_dir: str, filename: str, export_mode: DatasetExportMode, *, state_entity: str | list[str]) -> RecorderManagerOptions

What a demonstration records: observations, actions, rewards / dones / timestamps, the arm's state, and each episode's initial state.

obs/<group> and action are the behavior-cloning pairs; state/* is state_entity's joint and root state — every arm's, concatenated in order, when it lists several — which eden lerobot convert reads as observation.state; initial_state lets an episode be replayed from its first frame.

Source code in eden/extensions/scripted_demos/generate.py

def recorder_options(
    output_dir: str, filename: str, export_mode: DatasetExportMode, *, state_entity: str | list[str]
) -> RecorderManagerOptions:
    """What a demonstration records: observations, actions, rewards / dones / timestamps, the arm's state, and each
    episode's initial state.

    ``obs/<group>`` and ``action`` are the behavior-cloning pairs; ``state/*`` is ``state_entity``'s
    joint and root state — every arm's, concatenated in order, when it lists several — which
    ``eden lerobot convert`` reads as ``observation.state``; ``initial_state`` lets an episode be
    replayed from its first frame.
    """
    return RecorderManagerOptions(
        file_handler_options=HDF5FileHandler.configure(),
        dataset_export_dir_path=output_dir,
        dataset_filename=filename,
        dataset_export_mode=export_mode,
        override=True,
        observations=ObservationRecorder.configure(),
        actions=ActionRecorder.configure(),
        rl_step=RLStepRecorder.configure(),
        state=StateRecorder.configure(entity_name=state_entity),
        initial_state=InitialStateRecorder.configure(),
    )

generate_demos

generate_demos(config: EdenRLConfig, task_name: str, *, num_episodes: int, output_dir: str, filename: str = 'demos', export_mode: DatasetExportMode = EXPORT_ALL, seed: int = 0, p_mistake: float | None = None, episode_length_s: float | None = None, camera_only: bool = False, show_viewer: bool = False) -> GenerationReport

Generate num_episodes scripted demonstrations of task_name into output_dir/filename.

PARAMETER	DESCRIPTION
config	The task config (mods applied); its recorder options and episode length are replaced. TYPE: EdenRLConfig
task_name	Key of the task's demonstrator in :data:~eden.extensions.scripted_demos.registry.SCRIPTED_DEMO_REGISTRY. TYPE: str
num_episodes	Episodes to end before stopping. TYPE: int
output_dir	Where the dataset goes (filename without extension). TYPE: str
filename	Where the dataset goes (filename without extension). TYPE: str
export_mode	Keep every episode, the successful ones only, or both in separate files. Not EXPORT_NONE: successes are counted off what is exported. TYPE: DatasetExportMode DEFAULT: EXPORT_ALL
seed	Seeds the env and the demonstrator. TYPE: int DEFAULT: 0
p_mistake	Probability a subgoal is a deliberate mistake; the demonstrator's default when None. TYPE: float DEFAULT: None
episode_length_s	Time limit [s] of an episode; the demonstrator's own episode_length_s (or :data:DEFAULT_EPISODE_LENGTH_S) when None. TYPE: float DEFAULT: None
camera_only	Run the demonstrator without the simulator's object state (privileged=False): it sees through the task's cameras, which then render every env (env_separate_rigid). Only demonstrators that take a privileged argument can. TYPE: bool DEFAULT: False
show_viewer	Open the Genesis viewer. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
GenerationReport	Counts and where the dataset went.

Source code in eden/extensions/scripted_demos/generate.py

def generate_demos(
    config: EdenRLConfig,
    task_name: str,
    *,
    num_episodes: int,
    output_dir: str,
    filename: str = "demos",
    export_mode: DatasetExportMode = DatasetExportMode.EXPORT_ALL,
    seed: int = 0,
    p_mistake: float | None = None,
    episode_length_s: float | None = None,
    camera_only: bool = False,
    show_viewer: bool = False,
) -> GenerationReport:
    """Generate ``num_episodes`` scripted demonstrations of ``task_name`` into ``output_dir/filename``.

    Parameters
    ----------
    config : EdenRLConfig
        The task config (mods applied); its recorder options and episode length are replaced.
    task_name : str
        Key of the task's demonstrator in :data:`~eden.extensions.scripted_demos.registry.SCRIPTED_DEMO_REGISTRY`.
    num_episodes : int
        Episodes to end before stopping.
    output_dir, filename : str
        Where the dataset goes (``filename`` without extension).
    export_mode : DatasetExportMode
        Keep every episode, the successful ones only, or both in separate files. Not
        ``EXPORT_NONE``: successes are counted off what is exported.
    seed : int
        Seeds the env and the demonstrator.
    p_mistake : float, optional
        Probability a subgoal is a deliberate mistake; the demonstrator's default when None.
    episode_length_s : float, optional
        Time limit [s] of an episode; the demonstrator's own ``episode_length_s`` (or
        :data:`DEFAULT_EPISODE_LENGTH_S`) when None.
    camera_only : bool
        Run the demonstrator without the simulator's object state (``privileged=False``): it sees
        through the task's cameras, which then render every env (``env_separate_rigid``). Only
        demonstrators that take a ``privileged`` argument can.
    show_viewer : bool
        Open the Genesis viewer.

    Returns
    -------
    GenerationReport
        Counts and where the dataset went.
    """
    if export_mode == DatasetExportMode.EXPORT_NONE:
        raise ValueError("export_mode=EXPORT_NONE would generate demonstrations and write none of them.")
    if num_episodes <= 0:
        raise ValueError(f"num_episodes must be a positive integer, got {num_episodes}.")
    if task_name not in SCRIPTED_DEMO_REGISTRY:
        available = ", ".join(sorted(SCRIPTED_DEMO_REGISTRY.keys())) or "none"
        raise KeyError(
            f"Task '{task_name}' has no scripted demonstrator. Tasks with one: {available}. A demonstrator is a factory "
            f"registered in SCRIPTED_DEMO_REGISTRY under the task's name (see eden/tasks/robosuite/scripted_demos.py)."
        )
    from eden.envs.base import env_class_for_config

    factory = SCRIPTED_DEMO_REGISTRY.get(task_name)
    factory_kwargs = {"seed": seed}
    if p_mistake is not None:
        factory_kwargs["p_mistake"] = p_mistake
    if camera_only:
        if "privileged" not in inspect.signature(factory).parameters:
            camera_capable = sorted(
                name  # the registry iterates (name, factory) pairs
                for name, candidate in SCRIPTED_DEMO_REGISTRY
                if "privileged" in inspect.signature(candidate).parameters
            )
            raise ValueError(
                f"Task '{task_name}''s demonstrator reads the simulator's object state and has no camera-only mode; "
                f"tasks that do: {camera_capable or 'none'}. Drop --camera_only for this task."
            )
        factory_kwargs["privileged"] = False
        if config.env_options.num_envs > 1 and not config.env_options.env_separate_rigid:
            # A camera shared by the envs renders env 0 only; a camera-only demonstrator would see one scene everywhere.
            config.env_options.env_separate_rigid = True
            en.logger.info("--camera_only: rendering every env's own view (env_options.env_separate_rigid = True).")
    if episode_length_s is None:
        episode_length_s = getattr(factory, "episode_length_s", DEFAULT_EPISODE_LENGTH_S)
    if not episode_length_s > 0:
        raise ValueError(f"episode_length_s must be a positive number of seconds, got {episode_length_s}.")
    config.env_options.episode_length_s = episode_length_s
    arms = [rig.arm_name for rig in find_gripper_rigs(config.scene_options)]
    if not arms:
        raise TypeError(
            f"Task '{task_name}' holds no arm + two-finger gripper a scripted demonstrator could drive; scene entities: "
            f"{list(config.scene_options.keys())}."
        )
    state_entity = arms[0] if len(arms) == 1 else arms
    config.recorder_options = recorder_options(output_dir, filename, export_mode, state_entity=state_entity)
    env = env_class_for_config(config).from_config(config, show_viewer=show_viewer, eval_mode=True)
    env.build()
    env.reset(seed=seed)
    if env.metric_manager is None or not hasattr(env.metric_manager, "success_buf"):
        raise ValueError(
            f"Task '{task_name}' declares no success metric, so a generated episode cannot be labelled; give it a "
            f"metric term with contributes_to_success=True."
        )

    controller = factory(env, **factory_kwargs)
    controller.reset()

    recorder = env.recorder_manager
    report = GenerationReport(dataset_path=recorder_dataset_path(config.recorder_options))
    episode_steps = torch.zeros(env.num_envs, dtype=torch.long, device=env.device)
    # The generator's own time limit: a task need not declare a time_out termination (SIMPLER's
    # declare none), and a demonstrator that cannot finish must not hold its env forever.
    max_episode_steps = math.ceil(episode_length_s / env.dt)
    # Successful episodes exported per env so far: an env-ended episode's success is read off the
    # exporter, since the env clears its success buffer when it resets the env inside the step.
    exported_successes = [0] * env.num_envs
    recorder.start_recording()
    start = time.time()
    try:
        while report.episodes < num_episodes:
            done = controller.update_subgoals()
            timed_out = (episode_steps >= max_episode_steps) & ~done
            report.cut_short += int(timed_out.sum())
            ending = done | timed_out
            if bool(ending.any()):
                recorder.set_episode_success(ending, env.metric_manager.success_buf[ending])
                _keep_failure_trace(report, controller, ending & ~env.metric_manager.success_buf)
                _tally(report, ending, episode_steps)
                env.reset(ending)
                _count_exported_successes(recorder, ending, exported_successes)
                controller.reset(ending)
                episode_steps[ending] = 0
                if report.episodes >= num_episodes:
                    break

            try:
                env.step(controller.act())
            except gs.GenesisException as error:
                if "not finite" not in str(error) or not config.env_options.use_gjk_collision:
                    raise
                raise RuntimeError(
                    f"Genesis's GJK narrowphase produced a non-finite contact while generating '{task_name}' demos. "
                    f"Thin convex collision geoms (a WidowX's finger pads, say) pressed against an object trigger "
                    f"it, so a long scripted rollout eventually does. Rerun with MPR collision: eden datagen "
                    f"--task {task_name} --override 'env_options: {{use_gjk_collision: false}}'. Catching this "
                    f"and resetting the env does not help: the solver state of every env is invalid after it."
                ) from error
            episode_steps += 1

            ended = env.reset_buf
            if bool(ended.any()):  # the env has already reset these, and exported them with their success
                succeeded = _count_exported_successes(recorder, ended, exported_successes)
                report.cut_short += int((ended & ~succeeded).sum())
                _keep_failure_trace(report, controller, ended & ~succeeded)
                _tally(report, ended, episode_steps)
                controller.reset(ended)
                episode_steps[ended] = 0
    finally:
        recorder.end_recording(export_unfinished=False)
        # Counted by the exporter: the env clears its success buffer when it resets an env, so the
        # flag an env-ended episode carried survives only in what was exported.
        report.successes = recorder.exported_successful_episode_count
        recorder.close()
    report.wall_s = time.time() - start
    report.subgoals = dict(controller.subgoal_counts)
    en.logger.info(f"Scripted demos for '{task_name}': {report.render()}")
    if report.failure_trace:
        en.logger.info("Trace of the first unsuccessful episode:\n" + "\n".join(report.failure_trace))
    return report

recorder_dataset_path

recorder_dataset_path(options: RecorderManagerOptions) -> str

The file a recorder with options writes its (successful) episodes to.

Source code in eden/extensions/scripted_demos/generate.py

def recorder_dataset_path(options: RecorderManagerOptions) -> str:
    """The file a recorder with ``options`` writes its (successful) episodes to."""
    handler = FILE_HANDLER_REGISTRY.get(options.file_handler_options.name)()
    return handler.resolve_path(os.path.join(options.dataset_export_dir_path, options.dataset_filename))

Which scripted demonstrator drives which task: SCRIPTED_DEMO_REGISTRY, keyed by task name.

An entry is a factory (env, *, seed, p_mistake) -> demonstrator — a keyframe :class:~eden.extensions.scripted_demos.controller.ScriptedPickPlaceController, a tool-routine :class:~eden.extensions.scripted_demos.routines.RoutineDemonstrator, or a two-arm :class:~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator — registered under the task's :data:~eden.tasks.registry.TASK_REGISTRY name, next to the task it scripts (eden/tasks/<suite>/scripted_demos.py) — list that module in populated_by below. A factory that also takes privileged has a camera-only mode (eden datagen --camera_only), and one whose episodes need longer than the generator's default declares an episode_length_s attribute.

Attributes

SCRIPTED_DEMO_REGISTRY module-attribute

SCRIPTED_DEMO_REGISTRY = Registry('SCRIPTED_DEMO', populated_by=('eden.tasks.robosuite.scripted_demos', 'eden.tasks.simpler.scripted_demos', 'eden.tasks.polaris.scripted_demos', 'eden.tasks.whiteboard.scripted_demos', 'eden.tasks.art.calligraphy.scripted_demos', 'eden.tasks.art.painting.scripted_demos'))

Classes

Keyframe trajectories and pick-and-place primitives

Batched keyframe trajectories with demo-like timing: the motion layer of OCBench's scripted policies.

Port of ocbench/oracles/primitives/primitive.py (Primitive._compute_smooth_plan, _smooth_keyframe_times, current_plan_action), vectorized over environments.

A primitive samples a handful of keyframes — tool poses, times and a gripper value — and the trajectory between them is what makes a scripted rollout read like a human demonstration rather than a rule-based controller:

• Position follows a cubic Hermite spline. The tangent is zero at stop keyframes (where a person pauses: before closing the gripper, at the place pose, at the end) and passes through the others at the neighbours' average velocity, clamped to 1.25x the faster neighbouring segment.
• Rotation and gripper ease with the quintic smoothstep u^3 (10 - 15u + 6u^2).
• Timing is stretched so no segment exceeds a peak speed: a minimum-jerk profile peaks at 1.875 d / T, so a segment of length d gets at least T = 1.875 d / v_peak.
• Gates freeze the plan clock until the tool is within a tolerance of a keyframe, so the plan never runs ahead of an arm that lags behind it.
• Wobble adds a small sideways bulge 16 u^2 (1 - u)^2 ((1 - u) c_1 + u c_2) to the vertical approach and retreat phases, which a human never executes as a straight line.

Every tensor is batched over environments in the first dimension and lives on the env device; a plan is written for a subset of environments with :meth:KeyframeTrajectory.assign and read for all of them with :meth:KeyframeTrajectory.evaluate.

Attributes

MIN_JERK_PEAK_TO_MEAN module-attribute

MIN_JERK_PEAK_TO_MEAN = 1.875

TANGENT_SPEED_LIMIT module-attribute

TANGENT_SPEED_LIMIT = 1.25

Classes

Keyframes dataclass

Keyframes(times: Tensor, pos: Tensor, quat: Tensor, grasp: Tensor, stop: Tensor, valid: Tensor)

A batch of keyframe plans, one row per environment.

Optional keyframes (OCBench's pick_hold, or the release half of a pick that only lifts) are authored in place and switched off with valid; :meth:compacted moves the valid ones to the front, so the spline is built over exactly the keyframes each environment uses.

ATTRIBUTE	DESCRIPTION
times	(B, K) absolute keyframe times [s], non-decreasing over the valid keyframes. TYPE: Tensor
pos	(B, K, 3) tool positions in the world frame. TYPE: Tensor
quat	(B, K, 4) tool orientations (wxyz) in the world frame. TYPE: Tensor
grasp	(B, K) gripper command in [0, 1] (0 open, 1 closed). TYPE: Tensor
stop	(B, K) bool, keyframes the tool comes to rest at (zero Hermite tangent). TYPE: Tensor
valid	(B, K) bool, keyframes this environment's plan uses. The first must be valid. TYPE: Tensor

Attributes

times instance-attribute

times: Tensor

pos instance-attribute

pos: Tensor

quat instance-attribute

quat: Tensor

grasp instance-attribute

grasp: Tensor

stop instance-attribute

stop: Tensor

valid instance-attribute

valid: Tensor

Methods:

compacted

compacted() -> Keyframes

The valid keyframes first, the tail padded with the last valid one (so it adds no motion and no time).

Source code in eden/extensions/scripted_demos/trajectory.py

def compacted(self) -> Keyframes:
    """The valid keyframes first, the tail padded with the last valid one (so it adds no motion and no time)."""
    num_kf = self.times.shape[1]
    # A stable sort on ``~valid`` keeps the valid keyframes in their authored order.
    order = torch.argsort((~self.valid).to(torch.int8), dim=1, stable=True)
    num_valid = self.valid.sum(dim=1)
    last = (num_valid - 1).clamp(min=0).unsqueeze(1)
    slots = torch.arange(num_kf, device=self.times.device).unsqueeze(0)
    # Tail slots read the last valid keyframe.
    gather = torch.where(slots < num_valid.unsqueeze(1), order, order.gather(1, last))

    def take(x: torch.Tensor) -> torch.Tensor:
        index = gather.view(*gather.shape, *([1] * (x.dim() - 2))).expand_as(x)
        return x.gather(1, index)

    tail = slots >= num_valid.unsqueeze(1)
    return Keyframes(
        times=take(self.times),
        pos=take(self.pos),
        quat=take(self.quat),
        grasp=take(self.grasp),
        stop=take(self.stop) | tail,
        valid=~tail,
    )

filled

filled() -> Keyframes

Every invalid keyframe replaced by the last valid one before it (time, pose and grasp).

Keeps the authored layout, so a primitive can still read a keyframe's time by its slot after :meth:retimed — an invalid slot then reports its predecessor's time.

Source code in eden/extensions/scripted_demos/trajectory.py

def filled(self) -> Keyframes:
    """Every invalid keyframe replaced by the last valid one before it (time, pose and grasp).

    Keeps the authored layout, so a primitive can still read a keyframe's time by its slot
    after :meth:`retimed` — an invalid slot then reports its predecessor's time.
    """
    num_kf = self.times.shape[1]
    slots = torch.arange(num_kf, device=self.times.device).expand_as(self.times)
    source = torch.where(self.valid, slots, 0).cummax(dim=1).values

    def take(x: torch.Tensor) -> torch.Tensor:
        return x.gather(1, source.view(*source.shape, *([1] * (x.dim() - 2))).expand_as(x))

    return Keyframes(take(self.times), take(self.pos), take(self.quat), take(self.grasp), self.stop, self.valid)

retimed

retimed(peak_speed: Tensor) -> Keyframes

Stretch every segment that would exceed peak_speed ((B,) m/s) under a minimum-jerk profile.

OCBench Primitive._smooth_keyframe_times. Expects a :meth:filled or :meth:compacted plan, where an unused keyframe duplicates a used one: zero length and zero duration, so it stays put and the segments around it are timed as if it were absent.

Source code in eden/extensions/scripted_demos/trajectory.py

def retimed(self, peak_speed: torch.Tensor) -> Keyframes:
    """Stretch every segment that would exceed ``peak_speed`` (``(B,)`` m/s) under a minimum-jerk profile.

    OCBench ``Primitive._smooth_keyframe_times``. Expects a :meth:`filled` or :meth:`compacted`
    plan, where an unused keyframe duplicates a used one: zero length and zero duration, so it
    stays put and the segments around it are timed as if it were absent.
    """
    dt = self.times.diff(dim=1)
    dist = (self.pos[:, 1:] - self.pos[:, :-1]).norm(dim=-1)
    min_dt = MIN_JERK_PEAK_TO_MEAN * dist / peak_speed.unsqueeze(1)
    dt = torch.maximum(dt, min_dt)
    times = torch.cat([self.times[:, :1], self.times[:, :1] + dt.cumsum(dim=1)], dim=1)
    return Keyframes(times, self.pos, self.quat, self.grasp, self.stop, self.valid)

KeyframeTrajectory

KeyframeTrajectory(num_envs: int, num_keyframes: int, device: device)

The current keyframe plan of every environment, evaluated at a per-environment plan time.

PARAMETER	DESCRIPTION
num_envs	Number of environments. TYPE: int
num_keyframes	Keyframes per plan (the primitive's layout, before compaction). TYPE: int
device	Device of every buffer. TYPE: device

Source code in eden/extensions/scripted_demos/trajectory.py

def __init__(self, num_envs: int, num_keyframes: int, device: torch.device):
    if num_keyframes < 2:
        raise ValueError(
            f"A keyframe trajectory needs at least 2 keyframes per plan, got num_keyframes={num_keyframes}."
        )
    self.num_envs = num_envs
    self.num_keyframes = num_keyframes
    self.device = device
    self.times = torch.zeros(num_envs, num_keyframes, device=device)
    self.pos = torch.zeros(num_envs, num_keyframes, 3, device=device)
    self.quat = torch.zeros(num_envs, num_keyframes, 4, device=device)
    self.quat[..., 0] = 1.0
    self.grasp = torch.zeros(num_envs, num_keyframes, device=device)
    self.tangents = torch.zeros(num_envs, num_keyframes, 3, device=device)
    #: ``(N,)`` time of the last valid keyframe: the plan is complete once the plan clock reaches it.
    self.duration = torch.zeros(num_envs, device=device)

Attributes

num_envs instance-attribute

num_envs = num_envs

num_keyframes instance-attribute

num_keyframes = num_keyframes

device instance-attribute

device = device

times instance-attribute

times = torch.zeros(num_envs, num_keyframes, device=device)

pos instance-attribute

pos = torch.zeros(num_envs, num_keyframes, 3, device=device)

quat instance-attribute

quat = torch.zeros(num_envs, num_keyframes, 4, device=device)

grasp instance-attribute

grasp = torch.zeros(num_envs, num_keyframes, device=device)

tangents instance-attribute

tangents = torch.zeros(num_envs, num_keyframes, 3, device=device)

duration instance-attribute

duration = torch.zeros(num_envs, device=device)

Methods:

assign

assign(envs_idx: Tensor, keyframes: Keyframes) -> None

Write a new plan for envs_idx (long indices). keyframes need not be compacted.

Source code in eden/extensions/scripted_demos/trajectory.py

def assign(self, envs_idx: torch.Tensor, keyframes: Keyframes) -> None:
    """Write a new plan for ``envs_idx`` (long indices). ``keyframes`` need not be compacted."""
    if keyframes.times.shape != (len(envs_idx), self.num_keyframes):
        raise ValueError(
            f"Keyframes of shape {tuple(keyframes.times.shape)} do not fit a trajectory of "
            f"{self.num_keyframes} keyframes for {len(envs_idx)} envs; expected ({len(envs_idx)}, {self.num_keyframes})."
        )
    if not bool(keyframes.valid[:, 0].all()):
        raise ValueError("The first keyframe of every plan must be valid: it is where the tool starts.")
    compact = keyframes.compacted()
    self.times[envs_idx] = compact.times
    self.pos[envs_idx] = compact.pos
    self.quat[envs_idx] = normalize(compact.quat)
    self.grasp[envs_idx] = compact.grasp
    self.tangents[envs_idx] = hermite_tangents(compact)
    self.duration[envs_idx] = compact.times[:, -1]

evaluate

evaluate(t: Tensor) -> tuple[Tensor, Tensor, Tensor]

Tool position (N, 3), orientation (N, 4) and gripper command (N,) at plan time t (N,).

Source code in eden/extensions/scripted_demos/trajectory.py

def evaluate(self, t: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
    """Tool position ``(N, 3)``, orientation ``(N, 4)`` and gripper command ``(N,)`` at plan time ``t`` ``(N,)``."""
    t = torch.minimum(t, self.duration).unsqueeze(1)
    seg = (torch.searchsorted(self.times, t, right=True) - 1).clamp(0, self.num_keyframes - 2)
    nxt = seg + 1

    def at(x: torch.Tensor, index: torch.Tensor) -> torch.Tensor:
        return x.gather(1, index.view(*index.shape, *([1] * (x.dim() - 2))).expand(-1, 1, *x.shape[2:])).squeeze(1)

    t0, t1 = at(self.times, seg), at(self.times, nxt)
    dt = (t1 - t0).clamp(min=_EPS)
    u = ((t.squeeze(1) - t0) / dt).clamp(0.0, 1.0)
    u2, u3 = u * u, u * u * u
    h00, h10, h01, h11 = (2 * u3 - 3 * u2 + 1), (u3 - 2 * u2 + u), (-2 * u3 + 3 * u2), (u3 - u2)
    pos = (
        h00.unsqueeze(-1) * at(self.pos, seg)
        + (h10 * dt).unsqueeze(-1) * at(self.tangents, seg)
        + h01.unsqueeze(-1) * at(self.pos, nxt)
        + (h11 * dt).unsqueeze(-1) * at(self.tangents, nxt)
    )
    eased = smoothstep5(u)
    quat = normalize(quat_slerp(at(self.quat, seg), at(self.quat, nxt), eased.unsqueeze(-1)))
    grasp = (1.0 - eased) * at(self.grasp, seg) + eased * at(self.grasp, nxt)
    return pos, quat, grasp

Wobble

Wobble(num_envs: int, num_windows: int, device: device)

Sideways bulges over a few plan-time windows (OCBench CubePrimitive's _vertical_phases).

Inside window [t_0, t_1] the tool's xy is offset by 16 u^2 (1 - u)^2 ((1 - u) c_1 + u c_2) with u the window's progress, so the offset is zero, with zero slope, at both ends.

PARAMETER	DESCRIPTION
num_envs	Number of environments. TYPE: int
num_windows	Windows per plan. TYPE: int
device	Device of every buffer. TYPE: device

Source code in eden/extensions/scripted_demos/trajectory.py

def __init__(self, num_envs: int, num_windows: int, device: torch.device):
    self.windows = torch.zeros(num_envs, num_windows, 2, device=device)
    self.curves = torch.zeros(num_envs, num_windows, 2, 2, device=device)

Attributes

windows instance-attribute

windows = torch.zeros(num_envs, num_windows, 2, device=device)

curves instance-attribute

curves = torch.zeros(num_envs, num_windows, 2, 2, device=device)

Methods:

assign

assign(envs_idx: Tensor, windows: Tensor, curves: Tensor) -> None

Set (B, W, 2) windows [t_0, t_1] and (B, W, 2, 2) control offsets (c_1, c_2) for envs_idx.

Source code in eden/extensions/scripted_demos/trajectory.py

def assign(self, envs_idx: torch.Tensor, windows: torch.Tensor, curves: torch.Tensor) -> None:
    """Set ``(B, W, 2)`` windows ``[t_0, t_1]`` and ``(B, W, 2, 2)`` control offsets ``(c_1, c_2)`` for
    ``envs_idx``.
    """
    self.windows[envs_idx] = windows
    self.curves[envs_idx] = curves

offset

offset(t: Tensor) -> Tensor

The (N, 2) xy offset at plan time t (N,) (the first window containing t wins, as in OCBench).

Source code in eden/extensions/scripted_demos/trajectory.py

def offset(self, t: torch.Tensor) -> torch.Tensor:
    """The ``(N, 2)`` xy offset at plan time ``t`` ``(N,)`` (the first window containing ``t`` wins, as in
    OCBench).
    """
    t0, t1 = self.windows[..., 0], self.windows[..., 1]
    t = t.unsqueeze(1)
    inside = (t >= t0) & (t <= t1)
    first = inside & (inside.cumsum(dim=1) == 1)
    u = ((t - t0) / (t1 - t0).clamp(min=_EPS)).clamp(0.0, 1.0).unsqueeze(-1)
    bulge = 16.0 * u**2 * (1.0 - u) ** 2 * ((1.0 - u) * self.curves[..., 0, :] + u * self.curves[..., 1, :])
    return (bulge * first.unsqueeze(-1)).sum(dim=1)

Functions:

smoothstep5

smoothstep5(u: Tensor) -> Tensor

Quintic smoothstep u^3 (10 - 15u + 6u^2) on u clamped to [0, 1] (zero velocity and acceleration at both ends).

Source code in eden/extensions/scripted_demos/trajectory.py

def smoothstep5(u: torch.Tensor) -> torch.Tensor:
    """Quintic smoothstep ``u^3 (10 - 15u + 6u^2)`` on ``u`` clamped to ``[0, 1]`` (zero velocity and acceleration at
    both ends).
    """
    u = u.clamp(0.0, 1.0)
    return u**3 * (10.0 - 15.0 * u + 6.0 * u * u)

hermite_tangents

hermite_tangents(keyframes: Keyframes) -> Tensor

Per-keyframe position tangents (B, K, 3) for a :meth:~Keyframes.compacted plan.

Zero at the first keyframe, at the last valid one, and at every stop keyframe; elsewhere the central difference, clamped to :data:TANGENT_SPEED_LIMIT x the faster neighbouring segment.

Source code in eden/extensions/scripted_demos/trajectory.py

def hermite_tangents(keyframes: Keyframes) -> torch.Tensor:
    """Per-keyframe position tangents ``(B, K, 3)`` for a :meth:`~Keyframes.compacted` plan.

    Zero at the first keyframe, at the last valid one, and at every stop keyframe; elsewhere the
    central difference, clamped to :data:`TANGENT_SPEED_LIMIT` x the faster neighbouring segment.
    """
    times, pos = keyframes.times, keyframes.pos
    tangents = torch.zeros_like(pos)
    if pos.shape[1] < 3:
        return tangents
    t_prev, t_mid, t_next = times[:, :-2], times[:, 1:-1], times[:, 2:]
    p_prev, p_mid, p_next = pos[:, :-2], pos[:, 1:-1], pos[:, 2:]
    central = (p_next - p_prev) / (t_next - t_prev).clamp(min=_EPS).unsqueeze(-1)
    prev_speed = (p_mid - p_prev).norm(dim=-1) / (t_mid - t_prev).clamp(min=_EPS)
    next_speed = (p_next - p_mid).norm(dim=-1) / (t_next - t_mid).clamp(min=_EPS)
    max_speed = TANGENT_SPEED_LIMIT * torch.maximum(prev_speed, next_speed)
    speed = central.norm(dim=-1)
    central = central * torch.where(speed > max_speed, max_speed / speed.clamp(min=_EPS), 1.0).unsqueeze(-1)
    # The last valid keyframe is followed by padding (or nothing), so ``valid[i + 1]`` is False there.
    rests = keyframes.stop[:, 1:-1] | ~keyframes.valid[:, 2:]
    tangents[:, 1:-1] = torch.where(rests.unsqueeze(-1), 0.0, central)
    return tangents

Demo-like pick-and-place keyframes: a batched port of OCBench's CubePrimitive.

ocbench/oracles/primitives/cube.py (compute_keyframes). One call samples a whole pick -> carry -> place -> retreat plan for a batch of environments. Everything that makes it read like a person rather than a script is sampled per plan:

• a grasp yaw that is usually the face nearest the current wrist yaw but sometimes a quarter turn off, a wrist tilt of up to 25 deg that carries (attenuated) into the place,
• approach heights, contact offsets and a slightly off-centre place approach,
• an optional pause before lifting, and when the fingers close (on arrival or after a beat),
• a carry midpoint that bows to one side and, a third of the time, overshoots the target,
• +-3 % jitter on every keyframe time.

A failed pick is retried from wherever the hand is (:attr:PickPlaceRequest.num_retries), and every random offset shrinks by 0.75 per retry — a person is more careful the second time.

Poses here are effector rotations in OCBench's sense — the rotation of a gripper that points straight down when it is the identity — so yaw, roll and pitch read as the wrist's heading and tilt. :func:tool_quat maps one to the tool link's world orientation through the rig's down_quat.

Attributes

NUM_SLOTS module-attribute

NUM_SLOTS = len(Slot)

STOP_SLOTS module-attribute

STOP_SLOTS = (Slot.INITIAL, Slot.PICK_HOLD, Slot.PLACE_START, Slot.PLACE_END, Slot.FINAL)

WOBBLE_WINDOWS module-attribute

WOBBLE_WINDOWS = ((Slot.PICK, Slot.PICK_START), (Slot.PICK_HOLD, Slot.POSTPICK), (Slot.PLACE, Slot.PLACE_START), (Slot.PLACE_END, Slot.POSTPLACE))

GATE_SLOTS module-attribute

GATE_SLOTS = (Slot.PICK, Slot.PICK_START, Slot.PLACE, Slot.PLACE_START)

CONTACT_GATE_SLOTS module-attribute

CONTACT_GATE_SLOTS = (Slot.PICK_START, Slot.PLACE_START)

GRIPPER_PERIOD module-attribute

GRIPPER_PERIOD = math.pi

Classes

Slot

Bases: IntEnum

Keyframe layout of a pick-and-place plan (OCBench's keyframe names, in time order).

Attributes

INITIAL class-attribute instance-attribute

INITIAL = 0

PICK class-attribute instance-attribute

PICK = 1

PICK_START class-attribute instance-attribute

PICK_START = 2

PICK_END class-attribute instance-attribute

PICK_END = 3

PICK_HOLD class-attribute instance-attribute

PICK_HOLD = 4

POSTPICK class-attribute instance-attribute

POSTPICK = 5

CLEARANCE class-attribute instance-attribute

CLEARANCE = 6

PLACE class-attribute instance-attribute

PLACE = 7

PLACE_START class-attribute instance-attribute

PLACE_START = 8

PLACE_END class-attribute instance-attribute

PLACE_END = 9

POSTPLACE class-attribute instance-attribute

POSTPLACE = 10

FINAL class-attribute instance-attribute

FINAL = 11

PickPlaceParams dataclass

PickPlaceParams(peak_speed: float = 0.3, descent_height: tuple[float, float] = (0.06, 0.22), descent_xy: tuple[float, float] = (0.016, 0.06), max_tilt: float = radians(25.0), p_pick_hold: float = 0.4, p_overshoot: float = 0.35, gate_tolerance: float = 0.04, contact_gate_tolerance: float = 0.01, retry_noise_decay: float = 0.75)

The tunables of :func:sample_pick_place, OCBench's values by default.

ATTRIBUTE	DESCRIPTION
peak_speed	Peak tool speed [m/s] of a nominal-speed segment; the plan's segments are stretched so a minimum-jerk profile does not exceed peak_speed / segment_dt. TYPE: float
descent_height	Height [m] of the hover pose above the pick / place pose, along the approach axis. TYPE: tuple[float, float]
descent_xy	Sideways offset [m] of the place hover pose from straight above the target. TYPE: tuple[float, float]
max_tilt	Largest wrist tilt [rad] at the pick. TYPE: float
p_pick_hold	Probability of a pause between closing on the object and lifting it. TYPE: float
p_overshoot	Probability the carry midpoint overshoots the target instead of sitting between the two. TYPE: float
gate_tolerance	Distance [m] the tool must be within before the plan clock passes a hover gate. TYPE: float
contact_gate_tolerance	The same at the grasp and release poses (:data:CONTACT_GATE_SLOTS). TYPE: float
retry_noise_decay	Every random offset is scaled by retry_noise_decay ** num_retries. TYPE: float

Attributes

peak_speed class-attribute instance-attribute

peak_speed: float = 0.3

descent_height class-attribute instance-attribute

descent_height: tuple[float, float] = (0.06, 0.22)

descent_xy class-attribute instance-attribute

descent_xy: tuple[float, float] = (0.016, 0.06)

max_tilt class-attribute instance-attribute

max_tilt: float = math.radians(25.0)

p_pick_hold class-attribute instance-attribute

p_pick_hold: float = 0.4

p_overshoot class-attribute instance-attribute

p_overshoot: float = 0.35

gate_tolerance class-attribute instance-attribute

gate_tolerance: float = 0.04

contact_gate_tolerance class-attribute instance-attribute

contact_gate_tolerance: float = 0.01

retry_noise_decay class-attribute instance-attribute

retry_noise_decay: float = 0.75

Methods:

PickPlaceRequest dataclass

PickPlaceRequest(tool_pos: Tensor, tool_rot: Tensor, object_pos: Tensor, object_yaw: Tensor, goal_pos: Tensor, goal_yaw: Tensor, final_pos: Tensor, final_yaw: Tensor, release: Tensor, segment_dt: Tensor, num_retries: Tensor, yaw_window: Tensor, object_symmetry: int = 4)

What to pick and where to put it, for a batch of B environments.

ATTRIBUTE	DESCRIPTION
tool_pos	(B, 3) current tool-point position. TYPE: Tensor
tool_rot	(B, 4) current effector rotation (wxyz, identity = pointing straight down). TYPE: Tensor
object_pos, object_yaw	(B, 3) / (B,) object frame position and heading. TYPE: Tensor
goal_pos, goal_yaw	(B, 3) / (B,) where the object frame should end up. TYPE: Tensor
final_pos, final_yaw	(B, 3) / (B,) where the hand goes once it has let go. TYPE: Tensor
release	(B,) bool. False ends the plan holding the object at the goal (a lift), dropping the place hover, the release and the retreat. TYPE: Tensor
segment_dt	(B,) seconds per nominal segment — the inverse of the demonstrator's speed. TYPE: Tensor
num_retries	(B,) long, failed picks so far for this object. TYPE: Tensor
yaw_window	(B, 2) effector headings [rad] the wrist can reach from where it is now (low, high). Every planned heading is moved into it by a symmetry of what it grasps — see :func:fit_yaw. (-inf, inf) when the wrist is unlimited. TYPE: Tensor
object_symmetry	The object looks the same every 2 pi / object_symmetry of heading (4 for a cube): the grasp heading and the place heading are free up to that period. TYPE: int

Attributes

tool_pos instance-attribute

tool_pos: Tensor

tool_rot instance-attribute

tool_rot: Tensor

object_pos instance-attribute

object_pos: Tensor

object_yaw instance-attribute

object_yaw: Tensor

goal_pos instance-attribute

goal_pos: Tensor

goal_yaw instance-attribute

goal_yaw: Tensor

final_pos instance-attribute

final_pos: Tensor

final_yaw instance-attribute

final_yaw: Tensor

release instance-attribute

release: Tensor

segment_dt instance-attribute

segment_dt: Tensor

num_retries instance-attribute

num_retries: Tensor

yaw_window instance-attribute

yaw_window: Tensor

object_symmetry class-attribute instance-attribute

object_symmetry: int = 4

Methods:

PickPlacePlan dataclass

PickPlacePlan(keyframes: Keyframes, gate_times: Tensor, gate_pos: Tensor, gate_tolerance: Tensor, gate_valid: Tensor, wobble_windows: Tensor, wobble_curves: Tensor, pick_check_time: Tensor)

A sampled plan: keyframes in effector rotations, plus its gates, wobble and pick check.

ATTRIBUTE	DESCRIPTION
keyframes	:meth:~Keyframes.filled and retimed, in the :class:Slot layout; quat holds effector rotations (map through :func:tool_quat before driving the arm). TYPE: Keyframes
gate_times, gate_pos, gate_tolerance	(B, G) / (B, G, 3) / (B, G) the :data:GATE_SLOTS keyframes and how close the tool must come to each. TYPE: Tensor
gate_valid	(B, G) bool, gates this plan uses. TYPE: Tensor
wobble_windows, wobble_curves	(B, W, 2) / (B, W, 2, 2) for :class:~eden.extensions.scripted_demos.trajectory.Wobble. TYPE: Tensor
pick_check_time	(B,) plan time (postpick) from which the pick check may run; the controller then also waits for the hand to rise before judging it. TYPE: Tensor

Attributes

keyframes instance-attribute

keyframes: Keyframes

gate_times instance-attribute

gate_times: Tensor

gate_pos instance-attribute

gate_pos: Tensor

gate_tolerance instance-attribute

gate_tolerance: Tensor

gate_valid instance-attribute

gate_valid: Tensor

wobble_windows instance-attribute

wobble_windows: Tensor

wobble_curves instance-attribute

wobble_curves: Tensor

pick_check_time instance-attribute

pick_check_time: Tensor

Methods:

Functions:

effector_rot

effector_rot(yaw: Tensor, roll: Tensor | float = 0.0, pitch: Tensor | float = 0.0) -> Tensor

Effector rotation (B, 4) from heading and tilt: Rz(yaw) Ry(pitch) Rx(roll), OCBench's to_pose.

Source code in eden/extensions/scripted_demos/primitives.py

def effector_rot(
    yaw: torch.Tensor, roll: torch.Tensor | float = 0.0, pitch: torch.Tensor | float = 0.0
) -> torch.Tensor:
    """Effector rotation ``(B, 4)`` from heading and tilt: ``Rz(yaw) Ry(pitch) Rx(roll)``, OCBench's ``to_pose``."""
    roll = torch.as_tensor(roll, device=yaw.device).expand_as(yaw)
    pitch = torch.as_tensor(pitch, device=yaw.device).expand_as(yaw)
    return quat_from_euler_xyz(roll, pitch, yaw)

tool_quat

tool_quat(rot: Tensor, down_quat: Tensor) -> Tensor

World orientation of the tool link for effector rotation rot (down_quat is the tool's straight-down pose).

Source code in eden/extensions/scripted_demos/primitives.py

def tool_quat(rot: torch.Tensor, down_quat: torch.Tensor) -> torch.Tensor:
    """World orientation of the tool link for effector rotation ``rot`` (``down_quat`` is the tool's straight-down
    pose).
    """
    return quat_mul(rot, down_quat.expand_as(rot))

fit_yaw

fit_yaw(yaw: Tensor, window: Tensor, period: float) -> Tensor

Shift each heading by a whole number of period into its (low, high) window.

The shift that lands inside the window and is smallest wins; if none does (a window narrower than the period), the one closest to it. This is what keeps a 7-DOF arm's last joint off its stop — OCBench's UR5e wrist turns +-2 pi, a Panda's +-2.9 rad: the planner is free to grasp a cube by any of its faces and to finish with the fingers either way round, so it picks the equivalent heading the wrist can still reach.

Source code in eden/extensions/scripted_demos/primitives.py

def fit_yaw(yaw: torch.Tensor, window: torch.Tensor, period: float) -> torch.Tensor:
    """Shift each heading by a whole number of ``period`` into its ``(low, high)`` window.

    The shift that lands inside the window and is smallest wins; if none does (a window narrower
    than the period), the one closest to it. This is what keeps a 7-DOF arm's last joint off its
    stop — OCBench's UR5e wrist turns +-2 pi, a Panda's +-2.9 rad: the planner is free to grasp a
    cube by any of its faces and to finish with the fingers either way round, so it picks the
    equivalent heading the wrist can still reach.
    """
    shifts = torch.arange(-4, 5, device=yaw.device, dtype=yaw.dtype) * period
    candidates = yaw.unsqueeze(-1) + shifts  # (B, 9)
    low, high = window[:, :1], window[:, 1:]
    outside = (low - candidates).clamp(min=0.0) + (candidates - high).clamp(min=0.0)
    score = outside + 1e-3 * shifts.abs()
    return candidates.gather(1, score.argmin(dim=1, keepdim=True)).squeeze(1)

sample_pick_place

sample_pick_place(request: PickPlaceRequest, *, generator: Generator, workspace: tuple[Tensor, Tensor], params: PickPlaceParams | None = None) -> PickPlacePlan

Sample a demo-like pick-and-place plan per environment (OCBench CubePrimitive.compute_keyframes).

PARAMETER	DESCRIPTION
request	The batch to plan for. TYPE: PickPlaceRequest
generator	Source of every random draw, on the request's device. TYPE: Generator
workspace	(3,) low / high corners the carry midpoint is clipped to. TYPE: tuple[Tensor, Tensor]
params	Tunables; OCBench's values by default. TYPE: PickPlaceParams DEFAULT: None

RETURNS	DESCRIPTION
PickPlacePlan	The plan, in the :class:Slot layout.

Source code in eden/extensions/scripted_demos/primitives.py

def sample_pick_place(
    request: PickPlaceRequest,
    *,
    generator: torch.Generator,
    workspace: tuple[torch.Tensor, torch.Tensor],
    params: PickPlaceParams | None = None,
) -> PickPlacePlan:
    """Sample a demo-like pick-and-place plan per environment (OCBench ``CubePrimitive.compute_keyframes``).

    Parameters
    ----------
    request : PickPlaceRequest
        The batch to plan for.
    generator : torch.Generator
        Source of every random draw, on the request's device.
    workspace : tuple[torch.Tensor, torch.Tensor]
        ``(3,)`` low / high corners the carry midpoint is clipped to.
    params : PickPlaceParams, optional
        Tunables; OCBench's values by default.

    Returns
    -------
    PickPlacePlan
        The plan, in the :class:`Slot` layout.
    """
    p = params or PickPlaceParams()
    batch, device = request.tool_pos.shape[0], request.tool_pos.device
    rng = _Sampler(generator, batch, device)
    dt = request.segment_dt
    ns = p.retry_noise_decay ** request.num_retries.to(torch.float32)  # OCBench's ``noise_scale``
    ns2 = ns.unsqueeze(-1)
    up = torch.tensor([0.0, 0.0, 1.0], device=device).expand(batch, 3)

    pos = torch.zeros(batch, NUM_SLOTS, 3, device=device)
    rot = torch.zeros(batch, NUM_SLOTS, 4, device=device)
    valid = torch.ones(batch, NUM_SLOTS, dtype=torch.bool, device=device)

    use_hold = rng.bernoulli(p.p_pick_hold * ns)
    release = request.release
    valid[:, Slot.PICK_HOLD] = use_hold
    for slot in (Slot.PLACE, Slot.PLACE_END, Slot.POSTPLACE, Slot.FINAL):
        valid[:, slot] = release

    # --- grasp heading and tilt -----------------------------------------------------------------
    # The object face nearest the current wrist heading (a cube is 4-fold symmetric), usually; a
    # quarter turn off either way 30 % of the time.
    tool_yaw = yaw_from_quat(request.tool_rot)
    face = 2 * math.pi / request.object_symmetry
    quarter = math.pi / 2
    nearest_face = face * torch.round(wrap_to_pi(tool_yaw - request.object_yaw) / face)
    yaw_offset = nearest_face + rng.choice((-quarter, 0.0, quarter), (0.15, 0.7, 0.15)) + rng.uniform(-0.04, 0.04) * ns
    pick_yaw = fit_yaw(request.object_yaw + yaw_offset, request.yaw_window, face)
    # The object turns by the goal heading's nearest equivalent (at most half a face), not by up
    # to a full turn because the two headings were read off different branches of atan2.
    goal_turn = request.goal_yaw - request.object_yaw
    goal_turn = goal_turn - face * torch.round(goal_turn / face)
    place_yaw = fit_yaw(pick_yaw + goal_turn, request.yaw_window, face)
    tilt_angle = rng.uniform(-math.pi, math.pi)
    tilt_mag = rng.uniform(0.0, p.max_tilt) * ns
    pick_tilt = tilt_mag.unsqueeze(-1) * torch.stack([tilt_angle.cos(), tilt_angle.sin()], dim=-1)
    place_tilt = 0.2 * pick_tilt + rng.uniform(-0.06, 0.06, 2) * ns2

    pick_rot = effector_rot(pick_yaw, pick_tilt[:, 0], pick_tilt[:, 1])
    place_rot = effector_rot(place_yaw, place_tilt[:, 0], place_tilt[:, 1])

    # --- pick: hover, descend with a small contact offset, close, (pause), lift -----------------
    pick_axis = transform_by_quat(up, pick_rot)
    contact_angle = rng.uniform(-math.pi, math.pi)
    zeros = torch.zeros_like(contact_angle)
    contact_dir = transform_by_quat(torch.stack([contact_angle.cos(), contact_angle.sin(), zeros], dim=-1), pick_rot)
    contact_shift = (rng.uniform(0.0, 0.012) * ns).unsqueeze(-1)
    start_lift = (rng.uniform(0.0, 0.012) * ns).unsqueeze(-1)
    end_depth = (rng.uniform(-0.004, 0.006) * ns).unsqueeze(-1)
    pick_start_pos = request.object_pos + start_lift * pick_axis - 0.5 * contact_shift * contact_dir
    pick_end_pos = request.object_pos + end_depth * pick_axis + 0.5 * contact_shift * contact_dir
    pick_end_tilt = pick_tilt + rng.uniform(-0.035, 0.035, 2) * ns2
    pick_end_rot = effector_rot(pick_yaw + rng.uniform(-0.06, 0.06) * ns, pick_end_tilt[:, 0], pick_end_tilt[:, 1])
    pick_hover = pick_start_pos + rng.uniform(*p.descent_height).unsqueeze(-1) * pick_axis

    pos[:, Slot.INITIAL], rot[:, Slot.INITIAL] = request.tool_pos, request.tool_rot
    pos[:, Slot.PICK], rot[:, Slot.PICK] = pick_hover, pick_rot
    pos[:, Slot.PICK_START], rot[:, Slot.PICK_START] = pick_start_pos, pick_rot
    pos[:, Slot.PICK_END], rot[:, Slot.PICK_END] = pick_end_pos, pick_end_rot
    pos[:, Slot.PICK_HOLD], rot[:, Slot.PICK_HOLD] = pick_end_pos, pick_end_rot
    pos[:, Slot.POSTPICK], rot[:, Slot.POSTPICK] = pick_hover, pick_rot

    # --- place: hover slightly off to one side, descend, open, retreat --------------------------
    place_forward = torch.stack([place_yaw.cos(), place_yaw.sin()], dim=-1)
    place_side = torch.stack([-place_yaw.sin(), place_yaw.cos()], dim=-1) * rng.sign().unsqueeze(-1)
    place_offset = (rng.uniform(*p.descent_xy) * ns).unsqueeze(-1) * place_side + (
        rng.uniform(-0.02, 0.02) * ns
    ).unsqueeze(-1) * place_forward
    place_z = request.goal_pos[:, 2] + rng.uniform(*p.descent_height)
    place_hover = torch.cat([request.goal_pos[:, :2] + place_offset, place_z.unsqueeze(-1)], dim=-1)

    pos[:, Slot.PLACE], rot[:, Slot.PLACE] = place_hover, place_rot
    pos[:, Slot.PLACE_START], rot[:, Slot.PLACE_START] = request.goal_pos, place_rot
    pos[:, Slot.PLACE_END], rot[:, Slot.PLACE_END] = request.goal_pos, place_rot
    pos[:, Slot.POSTPLACE], rot[:, Slot.POSTPLACE] = place_hover, place_rot
    final_yaw = fit_yaw(request.final_yaw, request.yaw_window, GRIPPER_PERIOD)
    pos[:, Slot.FINAL], rot[:, Slot.FINAL] = request.final_pos, effector_rot(final_yaw)

    # --- carry midpoint: bowed to one side, sometimes overshooting ------------------------------
    # A lift has no place hover: the carry heads for the goal itself.
    carry_end = torch.where(release.unsqueeze(-1), place_hover, request.goal_pos)
    diff = carry_end - pick_hover
    xy_norm = diff[:, :2].norm(dim=-1, keepdim=True)
    random_heading = rng.uniform(-math.pi, math.pi)
    fallback_side = torch.stack([-random_heading.sin(), random_heading.cos()], dim=-1)
    side = torch.where(
        xy_norm > 1e-6, torch.stack([-diff[:, 1], diff[:, 0]], dim=-1) / xy_norm.clamp(min=1e-6), fallback_side
    )
    overshoot = rng.bernoulli(p.p_overshoot)
    alpha = torch.where(
        overshoot,
        rng.uniform(1.15, 1.35) * ns + (1.0 - ns),
        (0.5 + rng.uniform(-0.08, 0.08) * ns).clamp(0.15, 0.85),
    )
    side_offset = (rng.uniform(0.04, 0.14) * ns * rng.sign()).unsqueeze(-1)
    mid_xy = (
        pick_hover[:, :2] + alpha.unsqueeze(-1) * diff[:, :2] + side_offset * side + rng.uniform(-0.02, 0.02, 2) * ns2
    )
    mid_z = torch.maximum(pick_hover[:, 2], carry_end[:, 2]) + rng.uniform(0.03, 0.12)
    low, high = workspace
    mid_pos = torch.maximum(torch.minimum(torch.cat([mid_xy, mid_z.unsqueeze(-1)], dim=-1), high), low)
    mid_yaw = pick_yaw + alpha * (place_yaw - pick_yaw) + rng.uniform(-0.4, 0.4) * ns
    a = alpha.unsqueeze(-1)
    mid_tilt = (1.0 - a) * pick_tilt + a * place_tilt + rng.uniform(-0.05, 0.05, 2) * ns2
    pos[:, Slot.CLEARANCE], rot[:, Slot.CLEARANCE] = mid_pos, effector_rot(mid_yaw, mid_tilt[:, 0], mid_tilt[:, 1])

    # --- timing ---------------------------------------------------------------------------------
    dur = torch.zeros(batch, NUM_SLOTS, device=device)
    dur[:, Slot.PICK] = 1.0
    dur[:, Slot.PICK_START] = 1.8
    dur[:, Slot.PICK_END] = rng.uniform(0.75, 1.25)
    dur[:, Slot.PICK_HOLD] = rng.uniform(0.1, 0.22)
    dur[:, Slot.POSTPICK] = rng.uniform(0.75, 1.05)
    dur[:, Slot.CLEARANCE] = 1.0
    dur[:, Slot.PLACE] = 1.0
    dur[:, Slot.PLACE_START] = 1.5
    dur[:, Slot.PLACE_END] = rng.uniform(0.75, 1.1)
    dur[:, Slot.POSTPLACE] = rng.uniform(0.8, 1.15)
    dur[:, Slot.FINAL] = 1.0
    times = (dur * valid * dt.unsqueeze(-1)).cumsum(dim=1)
    jitter = rng.uniform(-1.0, 1.0, NUM_SLOTS) * 0.03 * dt.unsqueeze(-1)
    jitter[:, Slot.INITIAL] = 0.0
    times = times + jitter * valid

    # --- gripper: closes on arrival or a beat later, opens at the place -------------------------
    # The gripper value ramps over the segment *ending* at its keyframe. OCBench also closes at
    # pick_start (10 % without a pause), ramping over the descent — fine for its continuous gripper,
    # but a binary one fires halfway down, above the object. Dropped here: closing ramps over the
    # contact segment (pick_start -> pick_end) or the pause (pick_end -> pick_hold), with the hand at
    # the object either way.
    close_slot = torch.where(
        use_hold, rng.choice((Slot.PICK_END, Slot.PICK_HOLD), (0.65, 0.35)), int(Slot.PICK_END)
    ).long()
    open_slot = torch.where(release, int(Slot.PLACE_END), NUM_SLOTS).long()
    slots = torch.arange(NUM_SLOTS, device=device)
    grasp = ((slots >= close_slot.unsqueeze(-1)) & (slots < open_slot.unsqueeze(-1))).float()

    stop = torch.zeros(batch, NUM_SLOTS, dtype=torch.bool, device=device)
    stop[:, list(STOP_SLOTS)] = True
    keyframes = Keyframes(times=times, pos=pos, quat=rot, grasp=grasp, stop=stop, valid=valid).filled()
    keyframes = keyframes.retimed(peak_speed=p.peak_speed / dt)

    # --- gates, wobble, pick check (read off the retimed layout by slot) ------------------------
    t = keyframes.times
    gate_slots = list(GATE_SLOTS)
    windows = torch.stack([torch.stack([t[:, a], t[:, b]], dim=-1) for a, b in WOBBLE_WINDOWS], dim=1)
    window_valid = torch.stack([valid[:, b] & (valid[:, a] | (a == Slot.PICK_HOLD)) for a, b in WOBBLE_WINDOWS], dim=1)
    curves = rng.uniform(-0.025, 0.025, len(WOBBLE_WINDOWS), 2, 2) * ns.view(-1, 1, 1, 1)
    curves = curves * window_valid.view(batch, -1, 1, 1)
    return PickPlacePlan(
        keyframes=keyframes,
        gate_times=t[:, gate_slots],
        gate_pos=keyframes.pos[:, gate_slots],
        gate_tolerance=torch.tensor(
            [p.contact_gate_tolerance if slot in CONTACT_GATE_SLOTS else p.gate_tolerance for slot in GATE_SLOTS],
            device=device,
        ).expand(batch, -1),
        gate_valid=valid[:, gate_slots],
        wobble_windows=windows,
        wobble_curves=curves,
        pick_check_time=t[:, Slot.POSTPICK],
    )

Pick-and-place controller

Scripted pick-and-place demonstrator for an arm + two-finger-gripper rig, batched over environments.

Port of OCBench's controller / primitive split (ocbench/oracles/controllers/controller.py, ocbench/oracles/primitives/primitive.py): a task subclass decides what to move next (:meth:ScriptedPickPlaceController.choose_subgoals); this base turns each decision into a demo-like pick-and-place plan (:func:~eden.extensions.scripted_demos.primitives.sample_pick_place) and drives the rig along it:

• Speed. Every episode draws one demonstrator speed (segment_dt uniform in speed_range), which scales every segment of every plan in it.
• Plan clock and gates. Each env has its own plan time, advanced by the control step and frozen at each gate (the pick / place hovers and the grasp / release poses, :data:~eden.extensions.scripted_demos.primitives.GATE_SLOTS) until the tool has caught up.
• Retries. If the object does not rise with the hand on the way up from the grasp, the pick is re-planned from wherever the hand is, up to max_pick_retries times.
• Actions. The planned tool pose becomes the 6-D delta-pose command of the rig's DifferentialIKController and the plan's gripper value the BinaryJointController command, whatever the task names them (:mod:~eden.extensions.scripted_demos.rig) — the task's own action space, so a recorded action replays as-is.

Object state is read straight off the scene (OCBench's privileged/ observations), so a demonstration never depends on the noise the policy observations carry.

The control flow is batched torch on the env device. Only the subgoal decision is a per-env Python loop on a CPU snapshot: it runs once per finished plan, not per step, and keeps the task logic as readable as OCBench's.

Attributes

PICK_CHECK_RISE module-attribute

PICK_CHECK_RISE = 0.75

PICK_FOLLOW_FRACTION module-attribute

PICK_FOLLOW_FRACTION = 0.5

SETTLE_SPEED module-attribute

SETTLE_SPEED = 0.01

SETTLE_TIME module-attribute

SETTLE_TIME = 0.6

SETTLE_REACH module-attribute

SETTLE_REACH = 0.04

Classes

Subgoal dataclass

Subgoal(object_index: int, goal_pos: ndarray, goal_yaw: float, release: bool = True, kind: str = 'goal')

One pick-and-place decision for one environment.

ATTRIBUTE	DESCRIPTION
object_index	Index into :attr:ScriptedPickPlaceController.object_names. TYPE: int
goal_pos	(3,) world position the object's frame should end at. TYPE: ndarray
goal_yaw	Heading [rad] the object should end at. TYPE: float
release	Open the gripper at the goal and retreat (a place) — or keep holding it there (a lift). TYPE: bool
kind	Free-form label (goal, clear, mistake, ...), counted in the generation report. TYPE: str

Attributes

object_index instance-attribute

object_index: int

goal_pos instance-attribute

goal_pos: ndarray

goal_yaw instance-attribute

goal_yaw: float

release class-attribute instance-attribute

release: bool = True

kind class-attribute instance-attribute

kind: str = 'goal'

Methods:

ObjectSnapshot dataclass

ObjectSnapshot(pos: ndarray, yaw: ndarray)

CPU copy of the task objects' poses for the environments being decided for.

ATTRIBUTE	DESCRIPTION
pos	(B, M, 3) object frame positions (relative=True: the authored frame). TYPE: ndarray
yaw	(B, M) object headings [rad]. TYPE: ndarray

Attributes

pos instance-attribute

pos: ndarray

yaw instance-attribute

yaw: ndarray

Methods:

ScriptedPickPlaceController

ScriptedPickPlaceController(env: EnvBase, object_names: list[str], *, rig: str = 'robot', seed: int = 0, speed_range: tuple[float, float] = (0.7, 1.5), workspace: tuple[tuple[float, float, float], tuple[float, float, float]], effector_bounds: tuple[tuple[float, float, float], tuple[float, float, float]], effector_yaw_range: tuple[float, float] = (-pi / 2, pi / 2), max_pick_retries: int = 2, max_overrun_s: float = 10.0, gate_timeout_s: float = 3.0, max_step_translation: float = 0.05, max_step_rotation: float = 0.3, params: PickPlaceParams | None = None)

Bases: ABC

Base of the scripted demonstrators: plans, clocks and acts; a subclass picks the subgoals.

PARAMETER	DESCRIPTION
env	A built environment whose scene holds an arm + two-finger gripper at rig (a :class:~eden.options.robots.ManipulatorOptions composite, or a robot carrying its own fingers — see :func:~eden.extensions.scripted_demos.rig.gripper_rig), moved by one DifferentialIKController and one BinaryJointController (:func:~eden.extensions.scripted_demos.rig.find_rig_actions). TYPE: EnvBase
object_names	Scene entities the subgoals refer to, by index. TYPE: list[str]
rig	Scene entry holding the rig. TYPE: str DEFAULT: 'robot'
seed	Seed of the controller's own random draws (plans and subgoals). TYPE: int DEFAULT: 0
speed_range	Range of segment_dt [s], one draw per episode (OCBench: 0.7–1.5). TYPE: tuple[float, float] DEFAULT: (0.7, 1.5)
workspace	Low / high corners every tool target is clipped to. The low z is the lowest the fingertips may go: the tool point's floor is raised by :attr:tip_reach, measured on the built rig. TYPE: tuple[tuple[float, float, float], tuple[float, float, float]]
effector_bounds	Low / high corners the hand retreats to after a place. TYPE: tuple[tuple[float, float, float], tuple[float, float, float]]
effector_yaw_range	Heading range [rad] of that retreat pose, about the rig's straight-down pose. TYPE: tuple[float, float] DEFAULT: (-pi / 2, pi / 2)
max_pick_retries	Failed picks re-planned per subgoal before giving up on it. TYPE: int DEFAULT: 2
max_overrun_s	A plan still running this long [s] past its own duration (its clock held back at gates) is abandoned and a new subgoal chosen. Relative to the plan, not absolute: a slow demonstrator's plan legitimately runs long, and cutting it off mid-place ends an episode with the object still in the hand. TYPE: float DEFAULT: 10.0
gate_timeout_s	A hand held at a gate this long [s] without reaching it is re-planned from where it is. TYPE: float DEFAULT: 3.0
max_step_translation	Largest end-effector move asked for in one control step, per axis [m] and in angle [rad] (OCBench's ee_action_delta clip). Binding where the task's IK scale is generous (SIMPLER, Polaris: 0.5 m per action unit); the robosuite scale (0.05) is already tighter. TYPE: float DEFAULT: 0.05
max_step_rotation	Largest end-effector move asked for in one control step, per axis [m] and in angle [rad] (OCBench's ee_action_delta clip). Binding where the task's IK scale is generous (SIMPLER, Polaris: 0.5 m per action unit); the robosuite scale (0.05) is already tighter. TYPE: float DEFAULT: 0.05
params	Primitive tunables; OCBench's by default. TYPE: PickPlaceParams DEFAULT: None

Source code in eden/extensions/scripted_demos/controller.py

def __init__(
    self,
    env: EnvBase,
    object_names: list[str],
    *,
    rig: str = "robot",
    seed: int = 0,
    speed_range: tuple[float, float] = (0.7, 1.5),
    workspace: tuple[tuple[float, float, float], tuple[float, float, float]],
    effector_bounds: tuple[tuple[float, float, float], tuple[float, float, float]],
    effector_yaw_range: tuple[float, float] = (-math.pi / 2, math.pi / 2),
    max_pick_retries: int = 2,
    max_overrun_s: float = 10.0,
    gate_timeout_s: float = 3.0,
    max_step_translation: float = 0.05,
    max_step_rotation: float = 0.3,
    params: PickPlaceParams | None = None,
):
    if speed_range[0] <= 0 or speed_range[0] > speed_range[1]:
        raise ValueError(f"speed_range must be 0 < low <= high seconds per segment, got {speed_range}.")
    self.env = env
    self.object_names = list(object_names)
    self.rig_slot = rig
    self.speed_range = speed_range
    self.max_pick_retries = max_pick_retries
    self.max_overrun_s = max_overrun_s
    self.gate_timeout_steps = math.ceil(gate_timeout_s / env.dt)
    self.params = params or PickPlaceParams()
    self.num_envs = env.num_envs
    self.device = env.device
    self.generator = torch.Generator(device=self.device).manual_seed(seed)
    self.np_rng = np.random.default_rng(seed)
    #: Reads the rig and turns tool poses into the task's actions; resolved here, so a bad rig fails at
    #: construction.
    self.driver = RigDriver(
        env,
        rig,
        workspace=workspace,
        max_step_translation=max_step_translation,
        max_step_rotation=max_step_rotation,
    )
    self.rig = self.driver.rig
    self.down_quat = self.driver.down_quat
    self.tip_reach = self.driver.tip_reach
    self._workspace = self.driver.workspace
    self._effector_bounds = (
        torch.tensor(effector_bounds[0], device=self.device, dtype=torch.float32),
        torch.tensor(effector_bounds[1], device=self.device, dtype=torch.float32),
    )
    self._effector_yaw_range = effector_yaw_range
    #: Subgoal kinds started so far, for the generation report.
    self.subgoal_counts: dict[str, int] = {}
    self._build()

Attributes

env instance-attribute

env = env

object_names instance-attribute

object_names = list(object_names)

rig_slot instance-attribute

rig_slot = rig

speed_range instance-attribute

speed_range = speed_range

max_pick_retries instance-attribute

max_pick_retries = max_pick_retries

max_overrun_s instance-attribute

max_overrun_s = max_overrun_s

gate_timeout_steps instance-attribute

gate_timeout_steps = math.ceil(gate_timeout_s / env.dt)

params instance-attribute

params = params or PickPlaceParams()

num_envs instance-attribute

num_envs = env.num_envs

device instance-attribute

device = env.device

generator instance-attribute

generator = torch.Generator(device=self.device).manual_seed(seed)

np_rng instance-attribute

np_rng = np.random.default_rng(seed)

driver instance-attribute

driver = RigDriver(env, rig, workspace=workspace, max_step_translation=max_step_translation, max_step_rotation=max_step_rotation)

rig instance-attribute

rig = self.driver.rig

down_quat instance-attribute

down_quat = self.driver.down_quat

tip_reach instance-attribute

tip_reach = self.driver.tip_reach

subgoal_counts instance-attribute

subgoal_counts: dict[str, int] = {}

Methods:

tool_pose

tool_pose() -> tuple[Tensor, Tensor]

Tool-point position (N, 3) and effector rotation (N, 4) (identity = pointing straight down).

Source code in eden/extensions/scripted_demos/controller.py

def tool_pose(self) -> tuple[torch.Tensor, torch.Tensor]:
    """Tool-point position ``(N, 3)`` and effector rotation ``(N, 4)`` (identity = pointing straight down)."""
    return self.driver.tool_pose()

object_poses

object_poses() -> tuple[Tensor, Tensor]

Positions (N, M, 3) and headings (N, M) of :attr:object_names, in their authored frames.

Source code in eden/extensions/scripted_demos/controller.py

def object_poses(self) -> tuple[torch.Tensor, torch.Tensor]:
    """Positions ``(N, M, 3)`` and headings ``(N, M)`` of :attr:`object_names`, in their authored frames."""
    pos = torch.stack([obj.get_pos(relative=True) for obj in self._objects], dim=1)
    yaw = torch.stack([yaw_from_quat(obj.get_quat(relative=True)) for obj in self._objects], dim=1)
    return pos, yaw

yaw_window

yaw_window() -> Tensor

(N, 2) effector headings the arm's last joint can still reach (:meth:RigDriver.yaw_window).

Source code in eden/extensions/scripted_demos/controller.py

def yaw_window(self) -> torch.Tensor:
    """``(N, 2)`` effector headings the arm's last joint can still reach (:meth:`RigDriver.yaw_window`)."""
    return self.driver.yaw_window()

reset

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

Start a fresh episode for envs_idx (bool mask or long indices; all when None).

Call it for every env after the env itself has been reset — including once for all of them before the first :meth:act.

Source code in eden/extensions/scripted_demos/controller.py

def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    """Start a fresh episode for ``envs_idx`` (bool mask or long indices; all when None).

    Call it for every env after the env itself has been reset — including once for all of
    them before the first :meth:`act`.
    """
    if envs_idx is None:
        envs_idx = torch.arange(self.num_envs, device=self.device)
    elif envs_idx.dtype == torch.bool:
        envs_idx = envs_idx.nonzero(as_tuple=False).flatten()
    if len(envs_idx) == 0:
        return
    low, high = self.speed_range
    u = torch.rand(len(envs_idx), generator=self.generator, device=self.device)
    self.segment_dt[envs_idx] = low + (high - low) * u
    self.active[envs_idx] = False
    self.done[envs_idx] = False
    self.reset_task_state(envs_idx.tolist())

reset_task_state

reset_task_state(env_ids: list[int]) -> None

Clear a subclass's per-episode bookkeeping for env_ids. Default: nothing.

Source code in eden/extensions/scripted_demos/controller.py

def reset_task_state(self, env_ids: list[int]) -> None:
    """Clear a subclass's per-episode bookkeeping for ``env_ids``. Default: nothing."""

choose_subgoals abstractmethod

choose_subgoals(env_ids: list[int], objects: ObjectSnapshot) -> list[Subgoal | None]

Decide what each of env_ids moves next; None when its task is complete.

Called whenever an env has no plan (episode start) or has finished one.

Source code in eden/extensions/scripted_demos/controller.py

@abstractmethod
def choose_subgoals(self, env_ids: list[int], objects: ObjectSnapshot) -> list[Subgoal | None]:
    """Decide what each of ``env_ids`` moves next; ``None`` when its task is complete.

    Called whenever an env has no plan (episode start) or has finished one.
    """

update_subgoals

update_subgoals() -> Tensor

Retire finished plans and start the next subgoals; returns :attr:done.

:meth:act calls it first, so calling it is optional — and idempotent within a step. Call it before :meth:act to learn which envs are complete before acting, so their episode can end on the step that completed it rather than one hold step later.

Source code in eden/extensions/scripted_demos/controller.py

def update_subgoals(self) -> torch.Tensor:
    """Retire finished plans and start the next subgoals; returns :attr:`done`.

    :meth:`act` calls it first, so calling it is optional — and idempotent within a step. Call
    it before :meth:`act` to learn which envs are complete *before* acting, so their episode can
    end on the step that completed it rather than one hold step later.
    """
    finished = self.active & ((self.plan_time >= self.trajectory.duration) | (self._steps >= self._step_budget))
    needs_subgoal = (~self.active | finished) & ~self.done
    if bool(needs_subgoal.any()):
        self._start_subgoals(needs_subgoal.nonzero(as_tuple=False).flatten())
    return self.done

act

act() -> dict[str, Tensor]

The action dict for this control step: the IK term's (N, 6) command and the gripper term's (N, 1).

Source code in eden/extensions/scripted_demos/controller.py

def act(self) -> dict[str, torch.Tensor]:
    """The action dict for this control step: the IK term's ``(N, 6)`` command and the gripper term's ``(N, 1)``."""
    self.update_subgoals()
    # A plan installed since the last act() issues its first keyframe before its clock moves.
    running = self.active & ~self._fresh
    self.plan_time += self.env.dt * running
    self._steps += running

    tool_pos, _ = self.tool_pose()
    blocked, gate_pos = self._apply_gates(tool_pos)
    self._gate_wait = torch.where(blocked, self._gate_wait + 1, 0)
    stalled = self._gate_wait >= self.gate_timeout_steps
    if bool(stalled.any()):
        # The hand cannot reach the gate pose it is waiting at: start the pick-and-place over from here.
        self._plan(stalled.nonzero(as_tuple=False).flatten())
        blocked = blocked & ~stalled
    self._check_picks()

    pos, rot, grasp = self.trajectory.evaluate(self.plan_time)
    pos = torch.cat([pos[:, :2] + self.wobble.offset(self.plan_time), pos[:, 2:]], dim=-1)
    pos = torch.where(blocked.unsqueeze(-1), gate_pos, pos)
    self._fresh[:] = False
    # Envs whose task is complete hold still until the generator resets them.
    return self.driver.command(pos, rot, grasp, hold=self.done)

Functions:

Block-world demonstrators

Block-world demonstrators: build a tower, or lift a block — with OCBench's deliberate mistakes.

Port of ocbench/oracles/controllers/block.py (BlockController) onto :class:~eden.extensions.scripted_demos.controller.ScriptedPickPlaceController. The objects are boxes resting on a table; a subgoal moves one of them.

• Goal subgoals do the task: the lowest block of the tower that is not yet in place goes on the one below it — after whatever covers either of them has been moved out of the way.
• Mistakes. With probability p_mistake a subgoal is replaced by a wrong one: a random uncovered block goes to a random free spot on the table or, half the time, onto another uncovered block. The demonstrator recovers on its own (it re-plans from the resulting state), which is what makes the data suboptimal in the way human play data is: the high-level plan is noisy, not just the motion.

The decision logic is a per-env Python loop over a CPU snapshot, as in OCBench — it runs once per completed subgoal, not per step.

Attributes

REST_TOLERANCE module-attribute

REST_TOLERANCE = 0.015

STACK_XY_TOLERANCE module-attribute

STACK_XY_TOLERANCE = 0.02

Classes

BlockWorldController

BlockWorldController(env: EnvBase, block_names: list[str], *, table_z: float, spot_bounds: tuple[tuple[float, float], tuple[float, float]], min_spot_distance: float = 0.08, p_mistake: float = 0.1, **kwargs)

Bases: ScriptedPickPlaceController

Shared block-world geometry: sizes, what covers what, free table spots, and mistakes.

PARAMETER	DESCRIPTION
env	The built environment. TYPE: EnvBase
block_names	Scene entities that are boxes (:class:~eden.options.entities.BoxOptions) on the table. TYPE: list[str]
table_z	Height [m] of the table top. TYPE: float
spot_bounds	Low / high xy corners of the free spots blocks are moved to. TYPE: tuple[tuple[float, float], tuple[float, float]]
min_spot_distance	A free spot keeps at least this distance [m] from every other block. TYPE: float DEFAULT: 0.08
p_mistake	Probability a subgoal is replaced by a mistake. TYPE: float DEFAULT: 0.1
**kwargs	Forwarded to :class:~eden.extensions.scripted_demos.controller.ScriptedPickPlaceController. DEFAULT: {}

Source code in eden/extensions/scripted_demos/block_world.py

def __init__(
    self,
    env: EnvBase,
    block_names: list[str],
    *,
    table_z: float,
    spot_bounds: tuple[tuple[float, float], tuple[float, float]],
    min_spot_distance: float = 0.08,
    p_mistake: float = 0.1,
    **kwargs,
):
    if not 0.0 <= p_mistake <= 1.0:
        raise ValueError(f"p_mistake must be a probability in [0, 1], got {p_mistake}.")
    self.table_z = table_z
    self.spot_bounds = (np.asarray(spot_bounds[0], dtype=np.float64), np.asarray(spot_bounds[1], dtype=np.float64))
    self.min_spot_distance = min_spot_distance
    self.p_mistake = p_mistake
    self.half_heights = np.array([_box_half_extent(env, name) for name in block_names])
    super().__init__(env, block_names, **kwargs)

Attributes

table_z instance-attribute

table_z = table_z

spot_bounds instance-attribute

spot_bounds = (np.asarray(spot_bounds[0], dtype=np.float64), np.asarray(spot_bounds[1], dtype=np.float64))

min_spot_distance instance-attribute

min_spot_distance = min_spot_distance

p_mistake instance-attribute

p_mistake = p_mistake

half_heights instance-attribute

half_heights = np.array([_box_half_extent(env, name) for name in block_names])

Methods:

goal_subgoal

goal_subgoal(env_id: int, pos: ndarray, yaw: ndarray) -> Subgoal | None

The next step toward the task for one env, or None when it is complete.

Source code in eden/extensions/scripted_demos/block_world.py

def goal_subgoal(self, env_id: int, pos: np.ndarray, yaw: np.ndarray) -> Subgoal | None:
    """The next step toward the task for one env, or ``None`` when it is complete."""
    raise NotImplementedError

choose_subgoals

choose_subgoals(env_ids: list[int], objects: ObjectSnapshot) -> list[Subgoal | None]

Source code in eden/extensions/scripted_demos/block_world.py

def choose_subgoals(self, env_ids: list[int], objects: ObjectSnapshot) -> list[Subgoal | None]:
    subgoals: list[Subgoal | None] = []
    for row, env_id in enumerate(env_ids):
        pos, yaw = objects.pos[row], objects.yaw[row]
        subgoal = self.goal_subgoal(env_id, pos, yaw)
        if subgoal is not None and self.np_rng.uniform() < self.p_mistake:
            subgoal = self._mistake(pos) or subgoal
        subgoals.append(subgoal)
    return subgoals

TowerController

TowerController(env: EnvBase, block_names: list[str], *, table_z: float, spot_bounds: tuple[tuple[float, float], tuple[float, float]], min_spot_distance: float = 0.08, p_mistake: float = 0.1, **kwargs)

Bases: BlockWorldController

Stack block_names into one tower, bottom to top, wherever the bottom block stands.

robosuite Stack is the two-block case (["cubeB", "cubeA"]); OCBench's stack tasks are the longer ones.

Source code in eden/extensions/scripted_demos/block_world.py

def __init__(
    self,
    env: EnvBase,
    block_names: list[str],
    *,
    table_z: float,
    spot_bounds: tuple[tuple[float, float], tuple[float, float]],
    min_spot_distance: float = 0.08,
    p_mistake: float = 0.1,
    **kwargs,
):
    if not 0.0 <= p_mistake <= 1.0:
        raise ValueError(f"p_mistake must be a probability in [0, 1], got {p_mistake}.")
    self.table_z = table_z
    self.spot_bounds = (np.asarray(spot_bounds[0], dtype=np.float64), np.asarray(spot_bounds[1], dtype=np.float64))
    self.min_spot_distance = min_spot_distance
    self.p_mistake = p_mistake
    self.half_heights = np.array([_box_half_extent(env, name) for name in block_names])
    super().__init__(env, block_names, **kwargs)

Methods:

goal_subgoal

goal_subgoal(env_id: int, pos: ndarray, yaw: ndarray) -> Subgoal | None

Source code in eden/extensions/scripted_demos/block_world.py

def goal_subgoal(self, env_id: int, pos: np.ndarray, yaw: np.ndarray) -> Subgoal | None:
    num_blocks = len(pos)
    level = 0
    while level < num_blocks and (
        self._on_table(pos, 0) if level == 0 else self._stacked_on(pos, level, level - 1)
    ):
        level += 1
    if level == num_blocks:
        return None
    if level == 0:
        # The base sits on another block: uncover it, then put it down.
        return self._uncover(pos, 0) or self._move_to_free_spot(pos, 0, kind="clear")
    base = level - 1
    for block in (base, level):
        blocking = [other for other in self._covering(pos, block) if other != level]
        if blocking:
            top = max(blocking, key=lambda other: pos[other, 2])
            return self._move_to_free_spot(pos, top, kind="clear")
    goal = pos[base] + np.array([0.0, 0.0, self.half_heights[base] + self.half_heights[level]])
    return Subgoal(level, goal, float(yaw[base]), kind="goal")

LiftController

LiftController(env: EnvBase, block_names: list[str], *, lift_height=(0.1, 0.2), lift_xy=0.05, **kwargs)

Bases: BlockWorldController

Pick one block and hold it lift_height above the table (robosuite Lift).

The lift itself has no OCBench counterpart; mistakes are the block-world ones — the block is put down somewhere else first and the lift starts over from there.

PARAMETER	DESCRIPTION
lift_height	Range [m] the block's bottom is held above the table. TYPE: tuple[float, float] DEFAULT: (0.1, 0.2)
lift_xy	Largest horizontal drift [m] of the held pose from where the block was picked. TYPE: float DEFAULT: 0.05
**kwargs	Forwarded to :class:BlockWorldController. DEFAULT: {}

Source code in eden/extensions/scripted_demos/block_world.py

def __init__(self, env: EnvBase, block_names: list[str], *, lift_height=(0.1, 0.2), lift_xy=0.05, **kwargs):
    if len(block_names) != 1:
        raise ValueError(f"LiftController lifts one block, got block_names={block_names}.")
    self.lift_height = lift_height
    self.lift_xy = lift_xy
    super().__init__(env, block_names, **kwargs)

Attributes

lift_height instance-attribute

lift_height = lift_height

lift_xy instance-attribute

lift_xy = lift_xy

Methods:

goal_subgoal

goal_subgoal(env_id: int, pos: ndarray, yaw: ndarray) -> Subgoal | None

Source code in eden/extensions/scripted_demos/block_world.py

def goal_subgoal(self, env_id: int, pos: np.ndarray, yaw: np.ndarray) -> Subgoal | None:
    # Done once the block is held off the table; a dropped block is lifted again.
    if pos[0, 2] - (self.table_z + self.half_heights[0]) > 0.5 * self.lift_height[0]:
        return None
    return Subgoal(
        0, self._held_pose(pos), float(yaw[0] + self.np_rng.uniform(-0.3, 0.3)), release=False, kind="goal"
    )

Functions:

The robot a demonstrator drives

The arm + two-finger gripper a scripted demonstrator drives, from whichever robot a scene holds there.

A demonstrator needs a handful of facts about the robot: which entity carries the arm DOFs and the link the IK term drives, which entity carries the fingers, and where the tool point (the pad midpoint) sits in that link's frame. Two shapes of scene entry supply them:

• a composite :class:~eden.options.robots.ManipulatorOptions (robosuite's Panda, UR5e + Robotiq, DROID): an arm entity and a welded gripper entity, the tool point carried through the weld;
• a single :class:~eden.options.entities.RobotOptions whose asset includes its own fingers (SIMPLER's WidowX250): the fingers are the DOFs of open_dofs_pos, the arm every other actuated DOF, the IK link ee_links_name[0], and the tool point the robot's tcp_offset.

:func:gripper_rig reads either into one :class:GripperRig, so nothing downstream asks which shape it is. The action terms are found on the built env by what they drive (:func:find_rig_actions), not by name: tasks name them robot_arm (robosuite) or ee_controller (SIMPLER, Polaris).

:class:RigDriver is what moves the rig: it reads the tool point's pose off the built robot and turns a target tool pose and gripper value into the two terms' commands — the task's own action space, so a recorded action replays as-is. Every demonstrator drives the robot through it (the keyframe controller and the tool routines alike).

Attributes

MAX_COMMAND_RATIO module-attribute

MAX_COMMAND_RATIO = 0.9

WRIST_MARGIN module-attribute

WRIST_MARGIN = 0.35

MIN_WRIST_VERTICALITY module-attribute

MIN_WRIST_VERTICALITY = 0.5

Classes

GripperRig dataclass

GripperRig(slot: str, arm_name: str, gripper_name: str, ee_link: str, arm_dofs: tuple[str, ...], gripper_dofs: tuple[str, ...], tcp_offset: tuple[float, float, float])

The facts a scripted demonstrator reads off an arm + two-finger gripper.

ATTRIBUTE	DESCRIPTION
slot	Scene entry the rig was read from. TYPE: str
arm_name	Entity carrying the arm DOFs and :attr:ee_link. TYPE: str
gripper_name	Entity carrying the finger DOFs — :attr:arm_name itself for a single-body robot. TYPE: str
ee_link	Link the IK term drives, on :attr:arm_name. TYPE: str
arm_dofs	Arm DOFs, base to wrist; the last is the joint that turns the hand about its approach axis. TYPE: tuple[str, ...]
gripper_dofs	Finger DOFs. TYPE: tuple[str, ...]
tcp_offset	Tool point (pad midpoint, gripper open) in the :attr:ee_link frame (m). Its direction is the hand's approach axis. TYPE: tuple[float, float, float]

Attributes

slot instance-attribute

slot: str

arm_name instance-attribute

arm_name: str

gripper_name instance-attribute

gripper_name: str

ee_link instance-attribute

ee_link: str

arm_dofs instance-attribute

arm_dofs: tuple[str, ...]

gripper_dofs instance-attribute

gripper_dofs: tuple[str, ...]

tcp_offset instance-attribute

tcp_offset: tuple[float, float, float]

Methods:

RigActions dataclass

RigActions(arm: str, gripper: str)

The two action terms that move a rig: an IK term on its arm and a binary term on its fingers.

Attributes

arm instance-attribute

arm: str

gripper instance-attribute

gripper: str

Methods:

RigDriver

RigDriver(env: EnvBase, slot: str, *, workspace: tuple[tuple[float, float, float], tuple[float, float, float]], max_step_translation: float = 0.05, max_step_rotation: float = 0.3, exclusive: bool = True)

Reads a rig's tool pose and turns target tool poses into the task's IK + gripper commands, batched over envs.

Effector rotations are relative to the rig's straight-down pose (:attr:down_quat): identity points the approach axis at the floor with heading 0, and :func:~eden.extensions.scripted_demos.primitives.effector_rot builds one from a heading and a tilt.

PARAMETER	DESCRIPTION
env	A built environment holding the rig at slot (:func:gripper_rig), moved by one DifferentialIKController and one BinaryJointController (:func:find_rig_actions). TYPE: EnvBase
slot	Scene entry holding the rig. TYPE: str
workspace	Low / high corners every tool target is clipped to. The low z is the lowest the fingertips may go: the tool point's floor is raised by :attr:tip_reach, measured on the built rig. TYPE: tuple[tuple[float, float, float], tuple[float, float, float]]
max_step_translation	Largest end-effector move asked for in one control step, per axis [m] and in angle [rad] (OCBench's ee_action_delta clip). TYPE: float DEFAULT: 0.05
max_step_rotation	Largest end-effector move asked for in one control step, per axis [m] and in angle [rad] (OCBench's ee_action_delta clip). TYPE: float DEFAULT: 0.05
exclusive	Whether the rig's two action terms must be all the env has (:func:find_rig_actions); a multi-arm demonstrator drives each arm with False. TYPE: bool DEFAULT: True

Source code in eden/extensions/scripted_demos/rig.py

def __init__(
    self,
    env: EnvBase,
    slot: str,
    *,
    workspace: tuple[tuple[float, float, float], tuple[float, float, float]],
    max_step_translation: float = 0.05,
    max_step_rotation: float = 0.3,
    exclusive: bool = True,
):
    self.env = env
    self.device = device = env.device
    n = env.num_envs
    self.rig = rig = gripper_rig(env.scene_options, slot)
    tcp = torch.tensor(rig.tcp_offset, device=device, dtype=torch.float32)
    if float(tcp.norm()) < 1e-3:
        raise ValueError(
            f"Rig '{rig.slot}' has tcp_offset {rig.tcp_offset}: the tool point must lie off '{rig.ee_link}' — its "
            f"direction is the approach axis the demonstrator points at the floor."
        )
    self.arm: RigidEntity = env.entities[rig.arm_name]
    self.gripper: RigidEntity = env.entities[rig.gripper_name]
    self._ee_link_idx = torch.tensor([self.arm.get_link(rig.ee_link).idx_local], device=device)
    self._tcp = tcp.expand(n, 3)

    actions = find_rig_actions(env, rig, exclusive=exclusive)
    self.arm_term_name, self.gripper_term_name = actions.arm, actions.gripper
    self._scale = env.action_manager.get_term(actions.arm).effective_scale
    # The arm's last joint: its range bounds the headings a plan may ask for (dofs_name order
    # throughout — get_dofs_pos / get_dofs_limit / get_jacobian all use it).
    wrist = rig.arm_dofs[-1]
    if wrist not in self.arm.dofs_name:
        raise KeyError(
            f"Rig '{rig.slot}' lists arm DOF '{wrist}', which '{rig.arm_name}' does not have: "
            f"{list(self.arm.dofs_name)}."
        )
    self._wrist = list(self.arm.dofs_name).index(wrist)
    gripper_term = env.action_manager.get_term(actions.gripper)
    self._finger_dofs_idx = gripper_term.genesis_dofs_idx
    # Clamped to the joint range: a close target past it (a 2F-85 driven beyond its stop) would
    # leave empty fingers short of "closed", and read as holding something.
    low, high = (limit[:1] for limit in self.gripper.get_dofs_limit(dofs_idx_local=self._finger_dofs_idx))
    self._finger_open = torch.clamp(gripper_term.open_targets, low, high)
    self._finger_close = torch.clamp(gripper_term.close_targets, low, high)
    self.max_step_translation = max_step_translation
    self.max_step_rotation = max_step_rotation

    #: ``(1, 4)`` world orientation of the IK link with the hand pointing straight down, heading 0.
    self.down_quat = self._measure_down_quat(tcp)
    #: How far [m] the fingers reach past the tool point along the approach axis.
    self.tip_reach = self._measure_tip_reach(tcp)
    # The workspace floor bounds the fingertips, not the tool point between the pads: a 3 cm cube's
    # centre is lower than a WidowX's fingers reach past their pads, so a grasp aimed there drives
    # the fingers — and the cube in them — into the table.
    #: Low / high ``(3,)`` corners of the tool point's reachable box (the floor raised by :attr:`tip_reach`).
    self.workspace = (
        torch.tensor(workspace[0], device=device, dtype=torch.float32),
        torch.tensor(workspace[1], device=device, dtype=torch.float32),
    )
    self.workspace[0][2] += self.tip_reach

Attributes

env instance-attribute

env = env

device instance-attribute

device = env.device

rig instance-attribute

rig = gripper_rig(env.scene_options, slot)

arm instance-attribute

arm: RigidEntity = env.entities[rig.arm_name]

gripper instance-attribute

gripper: RigidEntity = env.entities[rig.gripper_name]

max_step_translation instance-attribute

max_step_translation = max_step_translation

max_step_rotation instance-attribute

max_step_rotation = max_step_rotation

down_quat instance-attribute

down_quat = self._measure_down_quat(tcp)

tip_reach instance-attribute

tip_reach = self._measure_tip_reach(tcp)

workspace instance-attribute

workspace = (torch.tensor(workspace[0], device=device, dtype=torch.float32), torch.tensor(workspace[1], device=device, dtype=torch.float32))

finger_axis_yaw cached property

finger_axis_yaw: float

Heading [rad] of the line the fingers close along, in the effector frame, in [0, pi).

Read off the built fingers the first time it is asked (the keyframe controller never asks): the contact links' box centres, projected on the plane across the approach axis — their principal direction is the closing line — relative to the effector's heading, so a target heading turns it with the hand.

Methods:

finger_links

finger_links() -> list

The finger links with collision geometry: every such link of a separate gripper entity, or the links the finger DOFs move.

Source code in eden/extensions/scripted_demos/rig.py

def finger_links(self) -> list:
    """The finger links with collision geometry: every such link of a separate gripper entity, or the links the
    finger DOFs move.
    """
    rig = self.rig
    if rig.gripper_name != rig.arm_name:
        links = [link for link in self.gripper.links if link.n_geoms]
    else:
        joints = {joint.name: joint for joint in self.gripper.joints}
        links = [joints[dof].link for dof in rig.gripper_dofs if dof in joints and joints[dof].link.n_geoms]
    if not links:
        raise ValueError(
            f"Rig '{rig.slot}': no finger link of '{rig.gripper_name}' has collision geometry to measure; "
            f"finger DOFs {list(rig.gripper_dofs)}."
        )
    return links

contact_links

contact_links() -> list

The links whose touch is the fingers': :meth:finger_links less a separate gripper's root (its palm / mount).

Source code in eden/extensions/scripted_demos/rig.py

def contact_links(self) -> list:
    """The links whose touch is the fingers': :meth:`finger_links` less a separate gripper's root (its palm /
    mount).
    """
    if self.rig.gripper_name == self.rig.arm_name:
        return self.finger_links()
    root = self.gripper.links[0]
    links = [link for link in self.finger_links() if link.idx != root.idx]
    if not links:
        raise ValueError(
            f"Rig '{self.rig.slot}': only the root link '{root.name}' of '{self.rig.gripper_name}' has collision "
            f"geometry, so no finger contact can be told from the palm's."
        )
    return links

ee_pos

ee_pos() -> Tensor

(N, 3) world position of the IK link.

Source code in eden/extensions/scripted_demos/rig.py

def ee_pos(self) -> torch.Tensor:
    """``(N, 3)`` world position of the IK link."""
    return self.arm.get_links_pos(ls_idx_local=self._ee_link_idx)[:, 0]

ee_quat

ee_quat() -> Tensor

(N, 4) world orientation of the IK link.

Source code in eden/extensions/scripted_demos/rig.py

def ee_quat(self) -> torch.Tensor:
    """``(N, 4)`` world orientation of the IK link."""
    return self.arm.get_links_quat(ls_idx_local=self._ee_link_idx)[:, 0]

tool_pose

tool_pose() -> tuple[Tensor, Tensor]

Tool-point position (N, 3) and effector rotation (N, 4) (identity = pointing straight down).

Source code in eden/extensions/scripted_demos/rig.py

def tool_pose(self) -> tuple[torch.Tensor, torch.Tensor]:
    """Tool-point position ``(N, 3)`` and effector rotation ``(N, 4)`` (identity = pointing straight down)."""
    ee_quat = self.ee_quat()
    pos = self.ee_pos() + transform_by_quat(self._tcp, ee_quat)
    return pos, quat_mul(ee_quat, inv_quat(self.down_quat).expand_as(ee_quat))

gripper_closure

gripper_closure() -> Tensor

(N,) how far the fingers are from open (0) toward their closed targets (1), averaged over finger DOFs.

Fingers closed on an object stop short of 1 — the proprioceptive "am I holding something".

Source code in eden/extensions/scripted_demos/rig.py

def gripper_closure(self) -> torch.Tensor:
    """``(N,)`` how far the fingers are from open (0) toward their closed targets (1), averaged over finger DOFs.

    Fingers closed on an object stop short of 1 — the proprioceptive "am I holding something".
    """
    q = self.gripper.get_dofs_pos(dofs_idx_local=self._finger_dofs_idx)
    span = self._finger_close - self._finger_open
    moving = span.abs() > 1e-6
    fraction = torch.where(moving, (q - self._finger_open) / torch.where(moving, span, 1.0), 0.0)
    return fraction.sum(-1) / moving.sum(-1).clamp(min=1)

finger_touches

finger_touches(min_force: float) -> Tensor

(N, L) bool over the scene's global link indices: which links a finger presses on with at least min_force [N].

The robot's own links never count.

Source code in eden/extensions/scripted_demos/rig.py

def finger_touches(self, min_force: float) -> torch.Tensor:
    """``(N, L)`` bool over the scene's global link indices: which links a finger presses on with at least
    ``min_force`` [N].

    The robot's own links never count.
    """
    contacts = self.gripper.get_contacts()
    link_a, link_b, valid = contacts["link_a"], contacts["link_b"], contacts["valid_mask"]
    force = contacts["force_a"].norm(dim=-1)
    finger_a, finger_b = torch.isin(link_a, self._contact_links_idx), torch.isin(link_b, self._contact_links_idx)
    rig_a, rig_b = torch.isin(link_a, self._rig_links_idx), torch.isin(link_b, self._rig_links_idx)
    pressing = valid & (force >= min_force)
    touches = torch.zeros(len(valid), self._num_scene_links, dtype=torch.long, device=self.device)
    for other, touching in ((link_b, pressing & finger_a & ~rig_b), (link_a, pressing & finger_b & ~rig_a)):
        # Padding rows hold whatever the buffer held: clamp their index, add nothing for them.
        index = torch.where(touching, other, 0).long().clamp(0, self._num_scene_links - 1)
        touches.scatter_add_(1, index, touching.long())
    return touches > 0

finger_contact

finger_contact(min_force: float) -> Tensor

(N,) bool: a finger link touches something that is not the robot, pressing at least min_force [N].

Source code in eden/extensions/scripted_demos/rig.py

def finger_contact(self, min_force: float) -> torch.Tensor:
    """``(N,)`` bool: a finger link touches something that is not the robot, pressing at least ``min_force`` [N]."""
    return self.finger_touches(min_force).any(dim=-1)

pinching

pinching(entity) -> Tensor

(N,) bool: finger links on both sides of the closing line touch entity — it is held between them.

Two links of one finger (a 2F-85's pad and its knuckle) touching it do not count.

Source code in eden/extensions/scripted_demos/rig.py

def pinching(self, entity) -> torch.Tensor:
    """``(N,)`` bool: finger links on *both* sides of the closing line touch ``entity`` — it is held between them.

    Two links of one finger (a 2F-85's pad and its knuckle) touching it do not count.
    """
    contacts = self.gripper.get_contacts(with_entity=entity._entity)
    link_a, link_b, valid = contacts["link_a"], contacts["link_b"], contacts["valid_mask"]
    tool_pos, rot = self.tool_pose()
    heading = yaw_from_quat(rot) + self.finger_axis_yaw
    closing = torch.stack([torch.cos(heading), torch.sin(heading), torch.zeros_like(heading)], dim=-1)
    left = right = torch.zeros(len(tool_pos), dtype=torch.bool, device=self.device)
    for link in self.contact_links():
        touching = (valid & ((link_a == link.idx) | (link_b == link.idx))).any(dim=-1)
        centre = (
            link.get_AABB().reshape(-1, 2, 3).mean(dim=1).to(self.device)
        )  # (N, 3); a frame can sit on the joint axis
        side = ((centre - tool_pos) * closing).sum(-1)
        left = left | (touching & (side > 0))
        right = right | (touching & (side < 0))
    return left & right

yaw_window

yaw_window() -> Tensor

(N, 2) effector headings reachable before the last arm joint comes within :data:WRIST_MARGIN of a stop.

Linearized about the current pose: a heading change d psi turns the last joint by d psi / k, where k is the vertical component of its world axis (about +-1 for a wrist pointing down). Re-measure it at every plan, so the redundant arm's drift is accounted for.

Source code in eden/extensions/scripted_demos/rig.py

def yaw_window(self) -> torch.Tensor:
    """``(N, 2)`` effector headings reachable before the last arm joint comes within :data:`WRIST_MARGIN` of a stop.

    Linearized about the current pose: a heading change ``d psi`` turns the last joint by
    ``d psi / k``, where ``k`` is the vertical component of its world axis (about +-1 for a wrist
    pointing down). Re-measure it at every plan, so the redundant arm's drift is accounted for.
    """
    q = self.arm.get_dofs_pos()[:, self._wrist]
    low, high = (limit[:, self._wrist] for limit in self.arm.get_dofs_limit())
    k = self.arm.get_jacobian(self.arm.get_link(self.rig.ee_link))[:, 5, self._wrist]
    _, rot = self.tool_pose()
    yaw = yaw_from_quat(rot)
    a = yaw + k * (low + WRIST_MARGIN - q)
    b = yaw + k * (high - WRIST_MARGIN - q)
    window = torch.stack([torch.minimum(a, b), torch.maximum(a, b)], dim=-1)
    unbounded = (k.abs() < MIN_WRIST_VERTICALITY) | ~torch.isfinite(window).all(dim=-1)
    inf = torch.full_like(window, math.inf)
    return torch.where(unbounded.unsqueeze(-1), torch.stack([-inf[:, 0], inf[:, 1]], dim=-1), window)

clip

clip(tool_pos: Tensor) -> Tensor

tool_pos clipped into :attr:workspace.

Source code in eden/extensions/scripted_demos/rig.py

def clip(self, tool_pos: torch.Tensor) -> torch.Tensor:
    """``tool_pos`` clipped into :attr:`workspace`."""
    low, high = self.workspace
    return torch.maximum(torch.minimum(tool_pos, high), low)

command

command(tool_pos: Tensor, rot: Tensor, grasp: Tensor, hold: Tensor | None = None) -> dict[str, Tensor]

The action dict that moves the tool point toward tool_pos / rot with gripper grasp.

PARAMETER	DESCRIPTION
tool_pos	(N, 3) target tool-point position; clipped into :attr:workspace. TYPE: Tensor
rot	(N, 4) target effector rotation. TYPE: Tensor
grasp	(N,) gripper value, 0 open to 1 closed. TYPE: Tensor
hold	(N,) bool: these envs' arms get a zero command (hold still). TYPE: Tensor DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Tensor]	The IK term's (N, 6) command and the gripper term's (N, 1), keyed by term name.

Source code in eden/extensions/scripted_demos/rig.py

def command(
    self, tool_pos: torch.Tensor, rot: torch.Tensor, grasp: torch.Tensor, hold: torch.Tensor | None = None
) -> dict[str, torch.Tensor]:
    """The action dict that moves the tool point toward ``tool_pos`` / ``rot`` with gripper ``grasp``.

    Parameters
    ----------
    tool_pos : torch.Tensor
        ``(N, 3)`` target tool-point position; clipped into :attr:`workspace`.
    rot : torch.Tensor
        ``(N, 4)`` target effector rotation.
    grasp : torch.Tensor
        ``(N,)`` gripper value, 0 open to 1 closed.
    hold : torch.Tensor, optional
        ``(N,)`` bool: these envs' arms get a zero command (hold still).

    Returns
    -------
    dict[str, torch.Tensor]
        The IK term's ``(N, 6)`` command and the gripper term's ``(N, 1)``, keyed by term name.
    """
    tool_pos = self.clip(tool_pos)
    target_quat = tool_quat(rot, self.down_quat)
    target_ee = tool_pos - transform_by_quat(self._tcp, target_quat)
    ee_quat = self.ee_quat()
    translation = (target_ee - self.ee_pos()).clamp(-self.max_step_translation, self.max_step_translation)
    rotation = axis_angle_from_quat(quat_mul(target_quat, inv_quat(ee_quat)))
    angle = rotation.norm(dim=-1, keepdim=True)
    rotation = rotation * (self.max_step_rotation / angle.clamp(min=self.max_step_rotation))
    delta = torch.cat([translation, rotation], dim=-1)
    ratio = (delta / self._scale).clamp(-MAX_COMMAND_RATIO, MAX_COMMAND_RATIO)
    arm = torch.atanh(ratio)
    if hold is not None:
        arm = torch.where(hold.unsqueeze(-1), 0.0, arm)
    gripper = (2.0 * grasp - 1.0).unsqueeze(-1)
    return {self.arm_term_name: arm, self.gripper_term_name: gripper}

Functions:

gripper_rig

gripper_rig(scene_options: SceneOptions, slot: str) -> GripperRig

Read the arm + gripper at scene entry slot (a composite rig or a robot with its own fingers).

RAISES	DESCRIPTION
KeyError	slot is not a scene entry.
TypeError	The entry is neither shape, or a single-body robot does not declare a fact the rig needs.

Source code in eden/extensions/scripted_demos/rig.py

def gripper_rig(scene_options: SceneOptions, slot: str) -> GripperRig:
    """Read the arm + gripper at scene entry ``slot`` (a composite rig or a robot with its own fingers).

    Raises
    ------
    KeyError
        ``slot`` is not a scene entry.
    TypeError
        The entry is neither shape, or a single-body robot does not declare a fact the rig needs.
    """
    options = getattr(scene_options, slot, None)
    if options is None:
        raise KeyError(
            f"Scene has no entry '{slot}' to read an arm + gripper from; scene entities: {entity_entries(scene_options)}."
        )
    if isinstance(options, ManipulatorOptions):
        return GripperRig(
            slot=slot,
            arm_name=options.arm_name,
            gripper_name=options.gripper_name,
            ee_link=options.ee_link,
            arm_dofs=tuple(options.arm_dofs),
            gripper_dofs=tuple(options.gripper_dofs),
            tcp_offset=tuple(options.tcp_offset),
        )
    if isinstance(options, RobotOptions):
        return _single_body_rig(slot, options)
    raise TypeError(
        f"Scene entry '{slot}' is a {type(options).__name__}; a scripted demonstrator drives an arm + two-finger "
        f"gripper — a ManipulatorOptions composite, or a RobotOptions whose asset carries its own fingers."
    )

entity_entries

entity_entries(scene_options: SceneOptions) -> list[str]

Names of the scene's entity entries — its attachments_dict and other plain fields left out.

Source code in eden/extensions/scripted_demos/rig.py

def entity_entries(scene_options: SceneOptions) -> list[str]:
    """Names of the scene's entity entries — its ``attachments_dict`` and other plain fields left out."""
    return [
        name
        # .keys(), not iteration: SceneOptions is a pydantic model, which iterates (name, value) pairs.
        for name in scene_options.keys()  # noqa: SIM118
        if isinstance(getattr(scene_options, name), (BaseEntityOptions, CompositeEntityOptions))
    ]

find_gripper_rigs

find_gripper_rigs(scene_options: SceneOptions) -> list[GripperRig]

Every arm + gripper in the scene, in scene-entry order — one for a single-arm task, two for a two-arm one.

Source code in eden/extensions/scripted_demos/rig.py

def find_gripper_rigs(scene_options: SceneOptions) -> list[GripperRig]:
    """Every arm + gripper in the scene, in scene-entry order — one for a single-arm task, two for a two-arm one."""
    rigs = []
    for slot in entity_entries(scene_options):
        try:
            rigs.append(gripper_rig(scene_options, slot))
        except TypeError:
            continue
    return rigs

find_gripper_rig

find_gripper_rig(scene_options: SceneOptions) -> GripperRig

The one arm + gripper in the scene, when the caller does not name its slot.

RAISES	DESCRIPTION
TypeError	The scene holds none, or several — name the slot then.

Source code in eden/extensions/scripted_demos/rig.py

def find_gripper_rig(scene_options: SceneOptions) -> GripperRig:
    """The one arm + gripper in the scene, when the caller does not name its slot.

    Raises
    ------
    TypeError
        The scene holds none, or several — name the slot then.
    """
    rigs = find_gripper_rigs(scene_options)
    if len(rigs) != 1:
        found = [rig.slot for rig in rigs]
        raise TypeError(
            f"Expected exactly one arm + gripper in the scene, found {len(rigs)} ({found}); pass the slot explicitly "
            f"(gripper_rig(scene_options, '<slot>'))."
        )
    return rigs[0]

find_rig_actions

find_rig_actions(env: EnvBase, rig: GripperRig, *, exclusive: bool = True) -> RigActions

Find the DifferentialIKController driving rig.ee_link and the BinaryJointController on its fingers.

Matched on what each term drives — entity, link, DOFs — not on its name. With exclusive (the default) every action term of the env must be one of the two: a demonstrator commands both each step and has nothing to send a third. A multi-arm demonstrator passes exclusive=False for each arm and checks that the arms' terms together cover every term (:class:~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator).

Source code in eden/extensions/scripted_demos/rig.py

def find_rig_actions(env: EnvBase, rig: GripperRig, *, exclusive: bool = True) -> RigActions:
    """Find the ``DifferentialIKController`` driving ``rig.ee_link`` and the ``BinaryJointController`` on its fingers.

    Matched on what each term drives — entity, link, DOFs — not on its name. With ``exclusive`` (the
    default) every action term of the env must be one of the two: a demonstrator commands both each
    step and has nothing to send a third. A multi-arm demonstrator passes ``exclusive=False`` for each
    arm and checks that the arms' terms together cover every term
    (:class:`~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator`).
    """
    manager = env.action_manager
    terms = {name: manager.get_term(name) for name in manager.active_terms}
    arm = [
        name
        for name, term in terms.items()
        if isinstance(term, DifferentialIKController)
        and term.entity_name == rig.arm_name
        and term.ee_link_name == rig.ee_link
    ]
    gripper = [
        name
        for name, term in terms.items()
        if isinstance(term, BinaryJointController)
        and term.entity_name == rig.gripper_name
        and set(term.dofs_name) == set(rig.gripper_dofs)
    ]
    layout = {name: f"{type(term).__name__}(entity={term.entity_name})" for name, term in terms.items()}
    if len(arm) != 1 or len(gripper) != 1:
        raise TypeError(
            f"A scripted demonstrator drives rig '{rig.slot}' through one DifferentialIKController on "
            f"'{rig.arm_name}'/'{rig.ee_link}' and one BinaryJointController on '{rig.gripper_name}' "
            f"{list(rig.gripper_dofs)}; found {len(arm)} and {len(gripper)} among the action terms {layout}."
        )
    extra = sorted(set(terms) - {arm[0], gripper[0]})
    if exclusive and extra:
        raise TypeError(
            f"Action terms {extra} drive something other than rig '{rig.slot}'; a scripted demonstrator commands "
            f"only its IK term '{arm[0]}' and gripper term '{gripper[0]}', and every term needs a command each step. "
            f"Action terms: {layout}."
        )
    return RigActions(arm=arm[0], gripper=gripper[0])

Tool routines

A routine is a scripted policy written for one robot as a generator of motions (move_to, move_until_contact, close_gripper, ...), run batched over every env.

Robot tools: the calls a scripted routine makes — motions it waits on, and readings it takes at once.

A routine (:mod:~eden.extensions.scripted_demos.routines) is written for one environment, as a plain Python generator, against a :class:Tools handle:

• Motions (:meth:Tools.move_to, :meth:Tools.move_until_contact, :meth:Tools.close_gripper, ...) return a :class:Motion. The routine yields it and gets an :class:Outcome back once the motion has ended — reached, timed out, stopped by contact. Every env's motion runs batched on the env device, so a routine reads like a script for one robot and runs on thousands.
• Readings return at once. The robot's own state (:meth:Tools.tool_pos, :meth:Tools.holding) is always available. What it knows about objects comes either from the simulator (:meth:Tools.object_footprint, :meth:Tools.grasped, :meth:Tools.link_pose — privileged) or from a camera (:meth:Tools.locate). A routine run with privileged=False is refused the first kind, by name, so a camera-only demonstrator cannot read the simulator state by accident.

Positions are world-frame metres. Headings are effector headings: 0 is the rig's straight-down pose with heading 0 (:attr:~eden.extensions.scripted_demos.rig.RigDriver.down_quat).

Attributes

STATUSES module-attribute

STATUSES = (REACHED, CONTACT, DONE, TIMEOUT, BLOCKED, NO_CONTACT)

TOP_BAND module-attribute

TOP_BAND = 0.02

Classes

MotionKind

Bases: IntEnum

What a :class:Motion does; the executor dispatches on it, batched.

Attributes

MOVE class-attribute instance-attribute

MOVE = 0

GUARDED class-attribute instance-attribute

GUARDED = 1

GRIPPER class-attribute instance-attribute

GRIPPER = 2

WAIT class-attribute instance-attribute

WAIT = 3

Motion dataclass

Motion(kind: MotionKind, label: str, target: tuple[float, float, float] | None = None, yaw: float | None = None, rot: tuple[float, float, float, float] | None = None, direction: tuple[float, float, float] | None = None, distance: float = 0.0, speed: float = 0.0, tolerance: float = 0.01, duration_s: float = 0.0, timeout_s: float = 2.0, grasp: float | None = None)

One motion a routine yields. Build it with a :class:Tools method, not by hand.

ATTRIBUTE	DESCRIPTION
kind	What the executor does with it. TYPE: MotionKind
label	How the trace names it (move_to(0.10, -0.25, 0.95)). TYPE: str
target	Tool-point position to reach (MOVE). TYPE: tuple[float, float, float] | None
yaw	Effector heading to reach (MOVE); None keeps the current one. TYPE: float | None
rot	Full effector rotation to reach (MOVE), wxyz relative to the straight-down pose — a tilted hand; set instead of yaw. TYPE: tuple[float, float, float, float] | None
direction	Unit direction of a guarded move. TYPE: tuple[float, float, float] | None
distance	How far [m] a guarded move goes before giving up. TYPE: float
speed	Peak tool speed [m/s] of a move, or the constant speed of a guarded one. TYPE: float
tolerance	A move has reached its target once the tool point is this close [m]. TYPE: float
duration_s	How long a gripper motion or a wait lasts [s]. TYPE: float
timeout_s	How long past its own duration a move may run before it ends TIMEOUT [s]. TYPE: float
grasp	Gripper value from this motion on, 0 open / 1 closed; None leaves it. TYPE: float | None

Attributes

kind instance-attribute

kind: MotionKind

label instance-attribute

label: str

target class-attribute instance-attribute

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

yaw class-attribute instance-attribute

yaw: float | None = None

rot class-attribute instance-attribute

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

direction class-attribute instance-attribute

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

distance class-attribute instance-attribute

distance: float = 0.0

speed class-attribute instance-attribute

speed: float = 0.0

tolerance class-attribute instance-attribute

tolerance: float = 0.01

duration_s class-attribute instance-attribute

duration_s: float = 0.0

timeout_s class-attribute instance-attribute

timeout_s: float = 2.0

grasp class-attribute instance-attribute

grasp: float | None = None

Methods:

Outcome dataclass

Outcome(motion: str, status: str, elapsed_s: float, error_m: float)

What a motion reports when it ends — the value of the routine's yield.

ATTRIBUTE	DESCRIPTION
motion	The motion's label. TYPE: str
status	One of :data:STATUSES. TYPE: str
elapsed_s	How long the motion ran [s]. TYPE: float
error_m	Distance [m] from the tool point to the motion's target when it ended (0 for gripper / wait). TYPE: float

Attributes

motion instance-attribute

motion: str

status instance-attribute

status: str

elapsed_s instance-attribute

elapsed_s: float

error_m instance-attribute

error_m: float

ok property

ok: bool

The motion did what it was asked: a move reached, a guarded move touched, a gripper / wait finished.

Methods:

Footprint dataclass

Footprint(center: ndarray, yaw: float, half_length: float, half_width: float, top_z: float, bottom_z: float, top_center: ndarray, num_points: int, centroid: ndarray | None = None)

Where an object is and how it lies, from points on it — its collision mesh, or what a camera saw of it.

ATTRIBUTE	DESCRIPTION
center	(3,): the centre of the horizontal footprint, at mid-height. TYPE: ndarray
yaw	Heading [rad] of the footprint's long axis, in (-pi/2, pi/2]. TYPE: float
half_length, half_width	Half extents [m] along the long and the short horizontal axis. TYPE: float
top_z, bottom_z	Highest and lowest point [m]. TYPE: float
top_center	(3,): centroid of the points within :data:TOP_BAND of the top — where a hand coming from above meets the object, which is off :attr:center when the object leans. TYPE: ndarray
num_points	How many points it was measured from. TYPE: int
centroid	(2,): mean xy of the points — off :attr:center toward where they crowd, the body of a mug rather than its handle (:func:~eden.extensions.scripted_demos.put_on_routine.body_centre). TYPE: ndarray | None

Attributes

center instance-attribute

center: ndarray

yaw instance-attribute

yaw: float

half_length instance-attribute

half_length: float

half_width instance-attribute

half_width: float

top_z instance-attribute

top_z: float

bottom_z instance-attribute

bottom_z: float

top_center instance-attribute

top_center: ndarray

num_points instance-attribute

num_points: int

centroid class-attribute instance-attribute

centroid: ndarray | None = None

height property

height: float

Methods:

PrivilegedToolError

Bases: PermissionError

A camera-only routine (privileged=False) called a tool that reads the simulator's state.

Tools

Tools(executor: RoutineDemonstrator, env_id: int, rng: Generator)

The tools one environment's routine calls. Handed to the routine by the executor; one per env.

PARAMETER	DESCRIPTION
executor	Runs the motions and serves the readings, batched over envs. TYPE: RoutineDemonstrator
env_id	The environment this handle speaks for. TYPE: int
rng	This env's random stream, re-seeded per episode: draw a routine's variations from it. TYPE: Generator

Source code in eden/extensions/scripted_demos/tools.py

def __init__(self, executor: RoutineDemonstrator, env_id: int, rng: np.random.Generator):
    self._executor = executor
    self.env_id = env_id
    self.rng = rng

Attributes

env_id instance-attribute

env_id = env_id

rng instance-attribute

rng = rng

privileged property

privileged: bool

Whether this routine may read the simulator's state (:meth:object_footprint, :meth:grasped, :meth:link_pose).

speed property

speed: float

This episode's peak tool speed [m/s]: the demonstrator's peak_speed times a per-episode draw.

tip_reach property

tip_reach: float

How far [m] the fingertips reach past the tool point (the pad midpoint) along the approach axis.

finger_axis_yaw property

finger_axis_yaw: float

Heading [rad] the fingers close along, relative to the effector heading (modulo pi).

time_s property

time_s: float

Time [s] since this env's episode started.

shared property

shared: dict

A dict this env's routine keeps for the episode, shared with the other arm's routine.

Under a :class:~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator both routines of an env get the same one, which is how two arms hand off: one writes a key, the other waits for it (while not tools.shared.get(key): yield tools.wait(0.1)).

Methods:

note

note(text: str) -> None

Add a line to this env's trace (:meth:RoutineDemonstrator.trace): what the routine decided and why.

Source code in eden/extensions/scripted_demos/tools.py

def note(self, text: str) -> None:
    """Add a line to this env's trace (:meth:`RoutineDemonstrator.trace`): what the routine decided and why."""
    self._executor.add_note(self.env_id, text)

tool_pos

tool_pos() -> ndarray

(3,) position of the tool point (the pad midpoint).

Source code in eden/extensions/scripted_demos/tools.py

def tool_pos(self) -> np.ndarray:
    """``(3,)`` position of the tool point (the pad midpoint)."""
    return self._executor.read("tool_pos")[self.env_id]

tool_yaw

tool_yaw() -> float

Current effector heading [rad].

Source code in eden/extensions/scripted_demos/tools.py

def tool_yaw(self) -> float:
    """Current effector heading [rad]."""
    return float(self._executor.read("tool_yaw")[self.env_id])

heading_window

heading_window() -> tuple[float, float]

Lowest and highest effector heading [rad] the arm's last joint can turn the hand to from here.

Linearized about the current pose (:meth:~eden.extensions.scripted_demos.rig.RigDriver.yaw_window); re-read it after a move. A routine about to turn something it holds by a quarter turn reads it to pick the direction there is room for — :meth:move_to would otherwise realize a turn past the joint's stop as the equivalent turn the other way, which a two-finger grasp allows but the held object does not.

Source code in eden/extensions/scripted_demos/tools.py

def heading_window(self) -> tuple[float, float]:
    """Lowest and highest effector heading [rad] the arm's last joint can turn the hand to from here.

    Linearized about the current pose (:meth:`~eden.extensions.scripted_demos.rig.RigDriver.yaw_window`);
    re-read it after a move. A routine about to turn something it holds by a quarter turn reads it
    to pick the direction there is room for — :meth:`move_to` would otherwise realize a turn past
    the joint's stop as the equivalent turn the other way, which a two-finger grasp allows but the
    held object does not.
    """
    low, high = self._executor.read("yaw_window")[self.env_id]
    return float(low), float(high)

tool_rotation

tool_rotation() -> ndarray

(4,) current effector rotation, wxyz relative to the straight-down pose — what move_to(rot=...) asks for.

The whole rotation, where :meth:tool_yaw is its heading only: what a held object's pose is measured against to turn it with the hand.

Source code in eden/extensions/scripted_demos/tools.py

def tool_rotation(self) -> np.ndarray:
    """``(4,)`` current effector rotation, wxyz relative to the straight-down pose — what ``move_to(rot=...)``
    asks for.

    The whole rotation, where :meth:`tool_yaw` is its heading only: what a held object's pose is
    measured against to turn it with the hand.
    """
    return self._executor.read("tool_rot")[self.env_id]

gripper_closure

gripper_closure() -> float

How far the fingers are from open (0) toward fully closed (1).

Source code in eden/extensions/scripted_demos/tools.py

def gripper_closure(self) -> float:
    """How far the fingers are from open (0) toward fully closed (1)."""
    return float(self._executor.read("gripper_closure")[self.env_id])

holding

holding() -> bool

The fingers are commanded closed and stopped short of closing: something is between them.

Proprioceptive — what the robot feels, not what the simulator knows — so camera-only routines may call it. Ask it after :meth:close_gripper has finished.

Source code in eden/extensions/scripted_demos/tools.py

def holding(self) -> bool:
    """The fingers are commanded closed and stopped short of closing: something is between them.

    Proprioceptive — what the robot feels, not what the simulator knows — so camera-only routines
    may call it. Ask it after :meth:`close_gripper` has finished.
    """
    closure = self.gripper_closure()
    return self._executor.grasp_command(self.env_id) > 0.5 and closure < self._executor.holding_closure

object_footprint

object_footprint(name: str) -> Footprint

Footprint of entity name from its collision geometry (privileged).

Source code in eden/extensions/scripted_demos/tools.py

def object_footprint(self, name: str) -> Footprint:
    """Footprint of entity ``name`` from its collision geometry (privileged)."""
    self._require_privileged("object_footprint", f"locate('{name}', camera=...)")
    return self._executor.object_footprint(name, self.env_id)

grasped

grasped(name: str) -> bool

Fingers on both sides of the closing line touch entity name (privileged).

Source code in eden/extensions/scripted_demos/tools.py

def grasped(self, name: str) -> bool:
    """Fingers on both sides of the closing line touch entity ``name`` (privileged)."""
    self._require_privileged("grasped", "holding()")
    return bool(self._executor.grasped(name)[self.env_id])

link_pose

link_pose(name: str, link: str | None = None) -> tuple[ndarray, ndarray]

World position (3,) and wxyz orientation (4,) of entity name's link (privileged).

None reads the entity's root. Poses are in the frame the asset was authored in (relative=True), so a point given in an MJCF body's frame — a site — is pos + R(quat) @ offset. For articulated fixtures (a door's panel and handle) whose pose a footprint cannot tell.

Source code in eden/extensions/scripted_demos/tools.py

def link_pose(self, name: str, link: str | None = None) -> tuple[np.ndarray, np.ndarray]:
    """World position ``(3,)`` and wxyz orientation ``(4,)`` of entity ``name``'s ``link`` (privileged).

    ``None`` reads the entity's root. Poses are in the frame the asset was authored in
    (``relative=True``), so a point given in an MJCF body's frame — a site — is
    ``pos + R(quat) @ offset``. For articulated fixtures (a door's panel and handle) whose pose a
    footprint cannot tell.
    """
    self._require_privileged("link_pose", f"locate('{name}', camera=...)")
    return self._executor.link_pose(name, link, self.env_id)

locate

locate(name: str, camera: str | None = None) -> Footprint | None

Footprint of what camera sees of entity name; None if it sees too little of it.

The camera's segmentation stands in for a perfect text-prompted segmenter: the points are the depth of the pixels showing name, so an occluded side or a hidden object reads as it would to a real camera. camera defaults to the demonstrator's first camera.

Source code in eden/extensions/scripted_demos/tools.py

def locate(self, name: str, camera: str | None = None) -> Footprint | None:
    """Footprint of what ``camera`` sees of entity ``name``; ``None`` if it sees too little of it.

    The camera's segmentation stands in for a perfect text-prompted segmenter: the points are the
    depth of the pixels showing ``name``, so an occluded side or a hidden object reads as it would
    to a real camera. ``camera`` defaults to the demonstrator's first camera.
    """
    return self._executor.locate(name, camera, self.env_id)

move_to

move_to(pos: Sequence[float], yaw: float | None = None, *, rot: Sequence[float] | None = None, speed: float | None = None, tolerance: float = 0.01, timeout_s: float = 2.0) -> Motion

Move the tool point to pos (and heading yaw) along a straight, minimum-jerk path.

yaw is turned by a half turn (a two-finger gripper's symmetry) into the range the arm's last joint can reach. rot instead asks for a full effector rotation (wxyz, relative to the straight-down pose; :meth:aim builds one), reached as given. Ends REACHED within tolerance [m], or TIMEOUT timeout_s after the path's own duration.

Source code in eden/extensions/scripted_demos/tools.py

def move_to(
    self,
    pos: Sequence[float],
    yaw: float | None = None,
    *,
    rot: Sequence[float] | None = None,
    speed: float | None = None,
    tolerance: float = 0.01,
    timeout_s: float = 2.0,
) -> Motion:
    """Move the tool point to ``pos`` (and heading ``yaw``) along a straight, minimum-jerk path.

    ``yaw`` is turned by a half turn (a two-finger gripper's symmetry) into the range the arm's
    last joint can reach. ``rot`` instead asks for a full effector rotation (wxyz, relative to the
    straight-down pose; :meth:`aim` builds one), reached as given. Ends ``REACHED`` within
    ``tolerance`` [m], or ``TIMEOUT`` ``timeout_s`` after the path's own duration.
    """
    if tolerance <= 0 or timeout_s < 0:
        raise ValueError(f"move_to needs tolerance > 0 and timeout_s >= 0, got {tolerance} and {timeout_s}.")
    if yaw is not None and rot is not None:
        raise ValueError(f"move_to takes a heading (yaw={yaw}) or a full rotation (rot={tuple(rot)}), not both.")
    target = _vector(pos, "pos")
    rotation = None if rot is None else _quaternion(rot)
    label = f"move_to({_fmt(target)}" + (f", yaw={yaw:.2f}" if yaw is not None else "")
    label += (f", rot=({_fmt(rotation)})" if rotation is not None else "") + ")"
    return Motion(
        kind=MotionKind.MOVE,
        label=label,
        target=target,
        yaw=None if yaw is None else float(yaw),
        rot=rotation,
        speed=self._speed(speed),
        tolerance=tolerance,
        timeout_s=timeout_s,
    )

aim

aim(approach: Sequence[float], closing: Sequence[float]) -> tuple[float, float, float, float]

The effector rotation (for :meth:move_to's rot) pointing the hand along approach and closing along closing.

closing is made perpendicular to approach; both are world-frame directions.

Source code in eden/extensions/scripted_demos/tools.py

def aim(self, approach: Sequence[float], closing: Sequence[float]) -> tuple[float, float, float, float]:
    """The effector rotation (for :meth:`move_to`'s ``rot``) pointing the hand along ``approach`` and closing
    along ``closing``.

    ``closing`` is made perpendicular to ``approach``; both are world-frame directions.
    """
    return aim_rotation(approach, closing, self.finger_axis_yaw)

move_by

move_by(offset: Sequence[float], *, speed: float | None = None, tolerance: float = 0.01) -> Motion

Move the tool point by offset from where it is now, keeping its heading.

Source code in eden/extensions/scripted_demos/tools.py

def move_by(self, offset: Sequence[float], *, speed: float | None = None, tolerance: float = 0.01) -> Motion:
    """Move the tool point by ``offset`` from where it is now, keeping its heading."""
    target = tuple(float(p + o) for p, o in zip(self.tool_pos(), _vector(offset, "offset")))
    motion = self.move_to(target, speed=speed, tolerance=tolerance)
    return Motion(**{**motion.__dict__, "label": f"move_by({_fmt(_vector(offset, 'offset'))})"})

move_until_contact

move_until_contact(direction: Sequence[float] = (0.0, 0.0, -1.0), max_distance: float = 0.1, *, speed: float = 0.03) -> Motion

Move the tool point along direction at speed until a finger touches something.

Ends CONTACT when a finger link newly presses on anything that is not the robot (a touch already there when it starts — a held object — does not count; lowering a held object until it lands ends BLOCKED), BLOCKED when the hand stops short of the command without a finger touching (the palm, the arm, the workspace floor), NO_CONTACT after max_distance [m] with nothing touched.

Source code in eden/extensions/scripted_demos/tools.py

def move_until_contact(
    self, direction: Sequence[float] = (0.0, 0.0, -1.0), max_distance: float = 0.1, *, speed: float = 0.03
) -> Motion:
    """Move the tool point along ``direction`` at ``speed`` until a finger touches something.

    Ends ``CONTACT`` when a finger link newly presses on anything that is not the robot (a touch
    already there when it starts — a held object — does not count; lowering a held object until it
    lands ends ``BLOCKED``), ``BLOCKED`` when the hand stops short of the command without a finger
    touching (the palm, the arm, the workspace floor), ``NO_CONTACT`` after ``max_distance`` [m]
    with nothing touched.
    """
    d = np.asarray(_vector(direction, "direction"))
    norm = float(np.linalg.norm(d))
    if norm < 1e-9:
        raise ValueError(f"move_until_contact needs a non-zero direction, got {tuple(direction)}.")
    if max_distance <= 0 or speed <= 0:
        raise ValueError(f"move_until_contact needs max_distance > 0 and speed > 0, got {max_distance}, {speed}.")
    unit = tuple(float(v) for v in d / norm)
    return Motion(
        kind=MotionKind.GUARDED,
        label=f"move_until_contact({_fmt(unit)}, max {max_distance:.3f} m)",
        direction=unit,
        distance=float(max_distance),
        speed=float(speed),
    )

close_gripper

close_gripper(duration_s: float = 0.6) -> Motion

Close the fingers and hold the arm still for duration_s [s].

Source code in eden/extensions/scripted_demos/tools.py

def close_gripper(self, duration_s: float = 0.6) -> Motion:
    """Close the fingers and hold the arm still for ``duration_s`` [s]."""
    return Motion(kind=MotionKind.GRIPPER, label="close_gripper()", duration_s=_positive(duration_s), grasp=1.0)

open_gripper

open_gripper(duration_s: float = 0.5) -> Motion

Open the fingers and hold the arm still for duration_s [s].

Source code in eden/extensions/scripted_demos/tools.py

def open_gripper(self, duration_s: float = 0.5) -> Motion:
    """Open the fingers and hold the arm still for ``duration_s`` [s]."""
    return Motion(kind=MotionKind.GRIPPER, label="open_gripper()", duration_s=_positive(duration_s), grasp=0.0)

wait

wait(duration_s: float) -> Motion

Hold the arm still for duration_s [s].

Source code in eden/extensions/scripted_demos/tools.py

def wait(self, duration_s: float) -> Motion:
    """Hold the arm still for ``duration_s`` [s]."""
    return Motion(kind=MotionKind.WAIT, label=f"wait({duration_s:.2f})", duration_s=_positive(duration_s))

Functions:

footprint_from_points

footprint_from_points(points: ndarray) -> Footprint

The :class:Footprint of (P, 3) points: principal horizontal axes, and the extents along them.

RAISES	DESCRIPTION
ValueError	Fewer than 3 points — nothing to measure an orientation from.

Source code in eden/extensions/scripted_demos/tools.py

def footprint_from_points(points: np.ndarray) -> Footprint:
    """The :class:`Footprint` of ``(P, 3)`` points: principal horizontal axes, and the extents along them.

    Raises
    ------
    ValueError
        Fewer than 3 points — nothing to measure an orientation from.
    """
    points = np.asarray(points, dtype=np.float64).reshape(-1, 3)
    if len(points) < 3:
        raise ValueError(f"A footprint needs at least 3 points to measure; got {len(points)}.")
    xy = points[:, :2]
    mean = xy.mean(axis=0)
    _, _, vt = np.linalg.svd(xy - mean, full_matrices=False)
    long_axis = vt[0]
    yaw = math.atan2(long_axis[1], long_axis[0])
    # Fold into (-pi/2, pi/2]: the long axis has no direction.
    if yaw <= -math.pi / 2:
        yaw += math.pi
    elif yaw > math.pi / 2:
        yaw -= math.pi
    axes = np.array([[math.cos(yaw), math.sin(yaw)], [-math.sin(yaw), math.cos(yaw)]])
    local = (xy - mean) @ axes.T
    low, high = local.min(axis=0), local.max(axis=0)
    if high[1] - low[1] > high[0] - low[0]:
        # The axis of most variance need not be the longest (uneven vertices, a partial camera view):
        # the long axis is the one of larger extent.
        yaw = yaw + math.pi / 2 if yaw <= 0 else yaw - math.pi / 2
        axes = np.array([[math.cos(yaw), math.sin(yaw)], [-math.sin(yaw), math.cos(yaw)]])
        local = (xy - mean) @ axes.T
        low, high = local.min(axis=0), local.max(axis=0)
    centre_xy = mean + ((low + high) / 2) @ axes
    top, bottom = float(points[:, 2].max()), float(points[:, 2].min())
    half = (high - low) / 2
    top_center = points[points[:, 2] >= top - TOP_BAND].mean(axis=0)
    return Footprint(
        center=np.array([centre_xy[0], centre_xy[1], (top + bottom) / 2]),
        yaw=yaw,
        half_length=float(half[0]),
        half_width=float(half[1]),
        top_z=top,
        bottom_z=bottom,
        top_center=top_center,
        num_points=len(points),
        centroid=mean,
    )

aim_rotation

aim_rotation(approach: Sequence[float], closing: Sequence[float], finger_axis_yaw: float) -> tuple[float, float, float, float]

Effector rotation (wxyz, relative to the straight-down pose) whose approach axis is approach and closing line closing.

In the straight-down pose with heading 0 the approach axis is world -z and the fingers close along heading finger_axis_yaw; the rotation maps that pair onto the requested one.

RAISES	DESCRIPTION
ValueError	A zero direction, or closing parallel to approach.

Source code in eden/extensions/scripted_demos/tools.py

def aim_rotation(
    approach: Sequence[float], closing: Sequence[float], finger_axis_yaw: float
) -> tuple[float, float, float, float]:
    """Effector rotation (wxyz, relative to the straight-down pose) whose approach axis is ``approach`` and closing
    line ``closing``.

    In the straight-down pose with heading 0 the approach axis is world ``-z`` and the fingers close
    along heading ``finger_axis_yaw``; the rotation maps that pair onto the requested one.

    Raises
    ------
    ValueError
        A zero direction, or ``closing`` parallel to ``approach``.
    """
    a = np.asarray(_vector(approach, "approach"))
    c = np.asarray(_vector(closing, "closing"))
    if np.linalg.norm(a) < 1e-9:
        raise ValueError(f"aim needs a non-zero approach direction, got {tuple(approach)}.")
    a = a / np.linalg.norm(a)
    c = c - (c @ a) * a
    if np.linalg.norm(c) < 1e-6:
        raise ValueError(f"aim's closing direction {tuple(closing)} is parallel to its approach {tuple(approach)}.")
    c = c / np.linalg.norm(c)
    source_a = np.array([0.0, 0.0, -1.0])
    source_c = np.array([math.cos(finger_axis_yaw), math.sin(finger_axis_yaw), 0.0])
    source = np.stack([source_a, source_c, np.cross(source_a, source_c)], axis=1)
    target = np.stack([a, c, np.cross(a, c)], axis=1)
    rotation = torch.as_tensor(target @ source.T, dtype=torch.float64)
    return _quaternion(R_to_quat(rotation).tolist())

Routine demonstrators: a scripted policy written for one robot, run batched over every env.

A routine is a generator function that drives one environment's robot through its :class:~eden.extensions.scripted_demos.tools.Tools handle. It yields motions and gets each one's :class:~eden.extensions.scripted_demos.tools.Outcome back, so checks and retries are plain Python::

def pick(tools):
    cube = tools.object_footprint("cube")
    above = cube.center + (0.0, 0.0, 0.10)
    yield tools.move_to(above, yaw=cube.yaw - tools.finger_axis_yaw + math.pi / 2)
    yield tools.move_to(cube.center)
    yield tools.close_gripper()
    yield tools.move_to(above)
    if not tools.holding():
        tools.note("missed the cube")

:class:RoutineDemonstrator gives each env its own instance of the routine and runs their motions together: the routines are resumed only when an env's motion has ended (once per motion, not per step), and every motion in flight is evaluated as batched torch on the env device and turned into the task's own actions by :class:~eden.extensions.scripted_demos.rig.RigDriver. It has the demonstrator interface eden datagen drives (reset / update_subgoals / act), so a routine registered for a task in :data:~eden.extensions.scripted_demos.registry.SCRIPTED_DEMO_REGISTRY generates datasets like the keyframe demonstrators do.

Every motion and every :meth:~eden.extensions.scripted_demos.tools.Tools.note lands in the env's :meth:~RoutineDemonstrator.trace — the record to read when a routine fails, one line per motion with how it ended.

Attributes

Routine module-attribute

Routine = Callable[[Tools], Generator[Motion, Outcome, Any]]

HEADING_TOLERANCE module-attribute

HEADING_TOLERANCE = 0.15

ROTATION_TOLERANCE module-attribute

ROTATION_TOLERANCE = 0.15

TILTED module-attribute

TILTED = 0.05

MIN_MOVE_S module-attribute

MIN_MOVE_S = 0.3

PEAK_TURN_RATE module-attribute

PEAK_TURN_RATE = 1.2

BLOCK_LAG module-attribute

BLOCK_LAG = 0.015

STALL_SPEED module-attribute

STALL_SPEED = 0.01

STALL_TIME module-attribute

STALL_TIME = 0.3

GUARD_SETTLE_S module-attribute

GUARD_SETTLE_S = 0.5

HOLD_TARGET_WITHIN module-attribute

HOLD_TARGET_WITHIN = 0.02

MIN_LOCATE_POINTS module-attribute

MIN_LOCATE_POINTS = 20

Classes

RoutineDemonstrator

RoutineDemonstrator(env: EnvBase, routine: Routine, *, rig: str = 'robot', seed: int = 0, workspace: tuple[tuple[float, float, float], tuple[float, float, float]], peak_speed: float = 0.12, speed_range: tuple[float, float] = (0.8, 1.25), max_step_translation: float = 0.05, max_step_rotation: float = 0.3, privileged: bool = True, cameras: tuple[str, ...] = (), contact_force: float = 1.0, holding_closure: float = 0.9, exclusive: bool = True)

Runs one routine per environment, batched, through an arm + two-finger-gripper rig.

PARAMETER	DESCRIPTION
env	A built environment holding the rig (:class:~eden.extensions.scripted_demos.rig.RigDriver). TYPE: EnvBase
routine	Called with each env's :class:~eden.extensions.scripted_demos.tools.Tools at the start of every episode; the env's task is complete when the generator returns. A routine may list the scene entities it looks at as an entity_names attribute (the put-on routine does), and each must then be in the scene. TYPE: Routine
rig	Scene entry holding the rig. TYPE: str DEFAULT: 'robot'
seed	Seed of the per-episode random streams (:attr:Tools.rng, the speed draw). TYPE: int DEFAULT: 0
workspace	Low / high corners every tool target is clipped to; the low z bounds the fingertips. TYPE: tuple[tuple[float, float, float], tuple[float, float, float]]
peak_speed	Nominal peak tool speed [m/s] of a move. Size it to what the task's action realizes per step (docs/guides/scripted-demos.md, Speed and saturation). TYPE: float DEFAULT: 0.12
speed_range	Each episode's moves run at peak_speed times a factor drawn uniformly from this range. TYPE: tuple[float, float] DEFAULT: (0.8, 1.25)
max_step_translation	Largest end-effector move per control step ([m] per axis, [rad]). TYPE: float DEFAULT: 0.05
max_step_rotation	Largest end-effector move per control step ([m] per axis, [rad]). TYPE: float DEFAULT: 0.05
privileged	Whether routines may read the simulator's object state. False refuses :meth:Tools.object_footprint / :meth:Tools.grasped, leaving the robot's own state and :meth:Tools.locate. TYPE: bool DEFAULT: True
cameras	Cameras :meth:Tools.locate may look through, the first being its default. Each must render every env (env_options.env_separate_rigid or the batch renderer) when there are several. TYPE: tuple[str, ...] DEFAULT: ()
contact_force	Smallest force [N] on a finger link that stops a guarded move. TYPE: float DEFAULT: 1.0
holding_closure	Fingers commanded closed that stop below this closure (0 open, 1 closed) are holding something. TYPE: float DEFAULT: 0.9
exclusive	Whether this rig's IK and gripper terms must be every action term of the env. A multi-arm demonstrator runs one routine demonstrator per arm with False (:class:~eden.extensions.scripted_demos.multi_arm.MultiArmDemonstrator). TYPE: bool DEFAULT: True

Source code in eden/extensions/scripted_demos/routines.py

def __init__(
    self,
    env: EnvBase,
    routine: Routine,
    *,
    rig: str = "robot",
    seed: int = 0,
    workspace: tuple[tuple[float, float, float], tuple[float, float, float]],
    peak_speed: float = 0.12,
    speed_range: tuple[float, float] = (0.8, 1.25),
    max_step_translation: float = 0.05,
    max_step_rotation: float = 0.3,
    privileged: bool = True,
    cameras: tuple[str, ...] = (),
    contact_force: float = 1.0,
    holding_closure: float = 0.9,
    exclusive: bool = True,
):
    if not callable(routine):
        raise TypeError(f"routine must be a generator function taking a Tools handle, got {routine!r}.")
    if peak_speed <= 0 or not 0 < speed_range[0] <= speed_range[1]:
        raise ValueError(
            f"peak_speed must be positive and speed_range 0 < low <= high, got {peak_speed} and {speed_range}."
        )
    self.env = env
    self.routine = routine
    self.privileged = privileged
    self.peak_speed = peak_speed
    self.speed_range = speed_range
    self.contact_force = contact_force
    self.holding_closure = holding_closure
    self.seed = seed
    self.num_envs = n = env.num_envs
    self.device = device = env.device
    self.dt = env.dt
    self.driver = RigDriver(
        env,
        rig,
        workspace=workspace,
        max_step_translation=max_step_translation,
        max_step_rotation=max_step_rotation,
        exclusive=exclusive,
    )
    self.cameras = tuple(cameras)
    #: Host copies of batched readings and camera frames, valid for the current step.
    self._cache: dict[Any, Any] = {}
    self._check_cameras()
    self._check_entities()
    self._stall_steps_needed = math.ceil(STALL_TIME / self.dt)

    # Batched motion state, one row per env.
    self._kind = torch.full((n,), _KIND_NONE, dtype=torch.long, device=device)
    self._elapsed = torch.zeros(n, device=device)
    self._duration = torch.zeros(n, device=device)
    self._timeout = torch.zeros(n, device=device)
    self._tolerance = torch.zeros(n, device=device)
    self._start_pos = torch.zeros(n, 3, device=device)
    self._start_rot = torch.zeros(n, 4, device=device)
    self._target_pos = torch.zeros(n, 3, device=device)
    self._target_rot = torch.zeros(n, 4, device=device)
    self._direction = torch.zeros(n, 3, device=device)
    self._distance = torch.zeros(n, device=device)
    self._speed = torch.zeros(n, device=device)
    self._grasp = torch.zeros(n, device=device)
    self._stall = torch.zeros(n, dtype=torch.long, device=device)
    self._prev_tool_pos = torch.zeros(n, 3, device=device)
    self._fresh = torch.zeros(n, dtype=torch.bool, device=device)
    self._tilted = torch.zeros(n, dtype=torch.bool, device=device)
    # The effector rotation last asked for: a move without a heading, a gripper motion or a wait
    # keeps it, so tilt and heading error do not build up from one motion to the next.
    self._held_rot = torch.zeros(n, 4, device=device)
    self._has_held_rot = torch.zeros(n, dtype=torch.bool, device=device)
    # Likewise the last move's target position, held by the gripper motions and waits after it.
    self._held_pos = torch.zeros(n, 3, device=device)
    self._has_held_pos = torch.zeros(n, dtype=torch.bool, device=device)
    # What the fingers touched (any force) when a guarded move started, by scene link: only a
    # touch of something else ends it — a held object's grip force comes and goes.
    self._touched_at_start: torch.Tensor | None = None
    #: ``(N,)`` bool: the routine has returned, this env's task is complete.
    self.done = torch.zeros(n, dtype=torch.bool, device=device)

    # Per-env Python state, touched only when an env's motion ends.
    self._routines: list[Generator[Motion, Outcome, Any] | None] = [None] * n
    self._motions: list[Motion | None] = [None] * n
    self._motion_start_s = np.zeros(n)
    self._traces: list[list[str]] = [[] for _ in range(n)]
    self._episode_steps = np.zeros(n, dtype=np.int64)
    self._episodes = np.zeros(n, dtype=np.int64)
    self._speed_scale = np.ones(n)
    self._grasp_np = np.zeros(n)
    self._blackboards: list[dict] = [{} for _ in range(n)]
    #: Motions ended so far, by ``<tool>:<status>`` — the generation report's tally.
    self.subgoal_counts: dict[str, int] = {}

Attributes

env instance-attribute

env = env

routine instance-attribute

routine = routine

privileged instance-attribute

privileged = privileged

peak_speed instance-attribute

peak_speed = peak_speed

speed_range instance-attribute

speed_range = speed_range

contact_force instance-attribute

contact_force = contact_force

holding_closure instance-attribute

holding_closure = holding_closure

seed instance-attribute

seed = seed

num_envs instance-attribute

num_envs = env.num_envs

device instance-attribute

device = env.device

dt instance-attribute

dt = env.dt

driver instance-attribute

driver = RigDriver(env, rig, workspace=workspace, max_step_translation=max_step_translation, max_step_rotation=max_step_rotation, exclusive=exclusive)

cameras instance-attribute

cameras = tuple(cameras)

done instance-attribute

done = torch.zeros(n, dtype=torch.bool, device=device)

subgoal_counts instance-attribute

subgoal_counts: dict[str, int] = {}

Methods:

reset

reset(envs_idx: Tensor | None = None, *, blackboards: dict[int, dict] | None = None) -> None

Start a fresh routine for envs_idx (bool mask or long indices; all when None).

Call it for every env after the env itself has been reset — including once for all of them before the first :meth:act, and after every env.step for the envs it auto-reset (env.reset_buf: a task's time limit or termination). A routine left running across an env reset keeps acting on the scene it saw before. eden datagen does both.

blackboards maps an env id to the dict its routine sees as :attr:Tools.shared this episode; a multi-arm demonstrator hands both arms of an env the same one. A fresh, empty dict otherwise.

Source code in eden/extensions/scripted_demos/routines.py

def reset(self, envs_idx: torch.Tensor | None = None, *, blackboards: dict[int, dict] | None = None) -> None:
    """Start a fresh routine for ``envs_idx`` (bool mask or long indices; all when None).

    Call it for every env after the env itself has been reset — including once for all of them
    before the first :meth:`act`, and after every ``env.step`` for the envs it auto-reset
    (``env.reset_buf``: a task's time limit or termination). A routine left running across an
    env reset keeps acting on the scene it saw before. ``eden datagen`` does both.

    ``blackboards`` maps an env id to the dict its routine sees as :attr:`Tools.shared` this
    episode; a multi-arm demonstrator hands both arms of an env the same one. A fresh, empty
    dict otherwise.
    """
    env_ids = self._env_ids(envs_idx)
    if not env_ids:
        return
    low, high = self.speed_range
    for env_id in env_ids:
        if self._routines[env_id] is not None:
            self._routines[env_id].close()  # run its finally blocks now, not at garbage collection
        self._episodes[env_id] += 1
        rng = np.random.default_rng((self.seed, env_id, int(self._episodes[env_id])))
        self._speed_scale[env_id] = rng.uniform(low, high)
        self._routines[env_id] = self.routine(Tools(self, env_id, rng))
        self._motions[env_id] = None
        self._traces[env_id] = []
        self._episode_steps[env_id] = 0
        self._grasp_np[env_id] = 0.0
        self._blackboards[env_id] = {} if blackboards is None else blackboards[env_id]
    idx = torch.tensor(env_ids, device=self.device)
    self._kind[idx] = _KIND_NONE
    self._grasp[idx] = 0.0
    self.done[idx] = False
    self._fresh[idx] = False
    self._has_held_rot[idx] = False
    self._has_held_pos[idx] = False
    self._cache.clear()

update_subgoals

update_subgoals() -> Tensor

End the motions that are over, resume their routines, and start their next motions; returns :attr:done.

:meth:act calls it first; calling it before is optional and idempotent within a step.

Source code in eden/extensions/scripted_demos/routines.py

def update_subgoals(self) -> torch.Tensor:
    """End the motions that are over, resume their routines, and start their next motions; returns :attr:`done`.

    :meth:`act` calls it first; calling it before is optional and idempotent within a step.
    """
    status = self._motion_status()
    idle = (self._kind == _KIND_NONE) & ~self.done
    resume = (idle | (status >= 0)) & ~self.done
    if bool(resume.any()):
        env_ids = resume.nonzero(as_tuple=False).flatten().tolist()
        self._resume(env_ids, tensor_to_array(status).tolist())
    return self.done

act

act() -> dict[str, Tensor]

The action dict for this control step, keyed by the task's IK and gripper term names.

Source code in eden/extensions/scripted_demos/routines.py

def act(self) -> dict[str, torch.Tensor]:
    """The action dict for this control step, keyed by the task's IK and gripper term names."""
    self.update_subgoals()
    moving = self._kind >= 0
    t = self._elapsed
    # MOVE: minimum-jerk along the straight line; GUARDED: constant speed along its direction;
    # GRIPPER / WAIT: hold where the motion started (the previous move's target when within
    # HOLD_TARGET_WITHIN of it), at the rotation last asked for.
    s = smoothstep5(t / self._duration.clamp(min=1e-6)).unsqueeze(-1)
    move_pos = self._start_pos + (self._target_pos - self._start_pos) * s
    move_rot = quat_slerp(self._start_rot, self._target_rot, s)
    move_rot = move_rot / move_rot.norm(dim=-1, keepdim=True).clamp(min=1e-9)
    travel = torch.minimum(self._speed * t, self._distance).unsqueeze(-1)
    guarded_pos = self._start_pos + self._direction * travel

    is_move = (self._kind == MotionKind.MOVE).unsqueeze(-1)
    is_guarded = (self._kind == MotionKind.GUARDED).unsqueeze(-1)
    pos = torch.where(is_move, move_pos, torch.where(is_guarded, guarded_pos, self._target_pos))
    rot = torch.where(is_move, move_rot, self._target_rot)

    # Stall tracking for guarded moves: the hand trails the command and has (nearly) stopped.
    tool_pos, _ = self.driver.tool_pose()
    # Against the unclipped command: a hand held at the workspace floor is blocked as much as one on the table.
    lag = (guarded_pos - tool_pos).norm(dim=-1)
    still = (tool_pos - self._prev_tool_pos).norm(dim=-1) < STALL_SPEED * self.dt
    stalling = is_guarded.squeeze(-1) & ~self._fresh & (lag > BLOCK_LAG) & still
    self._stall = torch.where(stalling, self._stall + 1, 0)
    self._prev_tool_pos = tool_pos

    command = self.driver.command(pos, rot, self._grasp, hold=self.done | ~moving)
    self._elapsed += self.dt * moving
    self._fresh[:] = False
    self._episode_steps += 1
    self._cache.clear()
    return command

trace

trace(env_id: int) -> list[str]

This episode's record for env_id: one line per ended motion and per note, in order.

Source code in eden/extensions/scripted_demos/routines.py

def trace(self, env_id: int) -> list[str]:
    """This episode's record for ``env_id``: one line per ended motion and per note, in order."""
    return list(self._traces[env_id])

add_note

add_note(env_id: int, text: str) -> None

Source code in eden/extensions/scripted_demos/routines.py

def add_note(self, env_id: int, text: str) -> None:
    self._traces[env_id].append(f"{self.episode_time(env_id):6.2f} s  # {text}")

episode_time

episode_time(env_id: int) -> float

Source code in eden/extensions/scripted_demos/routines.py

def episode_time(self, env_id: int) -> float:
    return float(self._episode_steps[env_id]) * self.dt

episode_speed

episode_speed(env_id: int) -> float

Source code in eden/extensions/scripted_demos/routines.py

def episode_speed(self, env_id: int) -> float:
    return self.peak_speed * float(self._speed_scale[env_id])

grasp_command

grasp_command(env_id: int) -> float

Source code in eden/extensions/scripted_demos/routines.py

def grasp_command(self, env_id: int) -> float:
    return float(self._grasp_np[env_id])

blackboard

blackboard(env_id: int) -> dict

The dict env_id's routine shares this episode (:attr:Tools.shared).

Source code in eden/extensions/scripted_demos/routines.py

def blackboard(self, env_id: int) -> dict:
    """The dict ``env_id``'s routine shares this episode (:attr:`Tools.shared`)."""
    return self._blackboards[env_id]

read

read(key: str) -> ndarray

A batched robot reading as a host array, read once per step: tool_pos, tool_rot, tool_yaw, gripper_closure, yaw_window.

Source code in eden/extensions/scripted_demos/routines.py

def read(self, key: str) -> np.ndarray:
    """A batched robot reading as a host array, read once per step: ``tool_pos``, ``tool_rot``, ``tool_yaw``,
    ``gripper_closure``, ``yaw_window``.
    """
    if key not in self._cache:
        if key == "tool_pos":
            value = self.driver.tool_pose()[0]
        elif key == "tool_rot":
            value = self.driver.tool_pose()[1]
        elif key == "tool_yaw":
            value = yaw_from_quat(self.driver.tool_pose()[1])
        elif key == "gripper_closure":
            value = self.driver.gripper_closure()
        elif key == "yaw_window":
            value = self.driver.yaw_window()
        else:
            raise KeyError(
                f"No robot reading '{key}'; readings: tool_pos, tool_rot, tool_yaw, gripper_closure, yaw_window."
            )
        self._cache[key] = tensor_to_array(value)
    return self._cache[key]

object_footprint

object_footprint(name: str, env_id: int) -> Footprint

Footprint of entity name in env_id from its collision vertices (the vertices read once per step).

Source code in eden/extensions/scripted_demos/routines.py

def object_footprint(self, name: str, env_id: int) -> Footprint:
    """Footprint of entity ``name`` in ``env_id`` from its collision vertices (the vertices read once per step)."""
    key = ("verts", name)
    if key not in self._cache:
        entity = self._entity(name)
        # A kinematic entity's links carry no collision geoms (and no ``n_geoms`` at all).
        verts = [link.get_verts() for link in entity.links if getattr(link, "n_geoms", 0)]
        if not verts:
            raise ValueError(
                f"Entity '{name}' has no collision geometry to measure a footprint from; read its pose with "
                f"link_pose('{name}') instead."
            )
        batched = [v if v.ndim == 3 else v.unsqueeze(0).expand(self.num_envs, -1, -1) for v in verts]
        self._cache[key] = tensor_to_array(torch.cat(batched, dim=1))
    return footprint_from_points(self._cache[key][env_id])

grasped

grasped(name: str) -> ndarray

(N,) bool: fingers on both sides of the closing line touch entity name.

Source code in eden/extensions/scripted_demos/routines.py

def grasped(self, name: str) -> np.ndarray:
    """``(N,)`` bool: fingers on both sides of the closing line touch entity ``name``."""
    key = ("grasped", name)
    if key not in self._cache:
        self._cache[key] = tensor_to_array(self.driver.pinching(self._entity(name)))
    return self._cache[key]

link_pose

link_pose(name: str, link: str | None, env_id: int) -> tuple[ndarray, ndarray]

Authored-frame pose of entity name's link (its root when None) in env_id, read once per step.

Source code in eden/extensions/scripted_demos/routines.py

def link_pose(self, name: str, link: str | None, env_id: int) -> tuple[np.ndarray, np.ndarray]:
    """Authored-frame pose of entity ``name``'s ``link`` (its root when None) in ``env_id``, read once per step."""
    key = ("link_pose", name, link)
    if key not in self._cache:
        entity = self._entity(name)
        if link is None:
            pos, quat = entity.get_pos(relative=True), entity.get_quat(relative=True)
        else:
            links = [candidate.name for candidate in entity.links]
            if link not in links:
                raise KeyError(f"Entity '{name}' has no link '{link}'; its links: {links}.")
            idx = torch.tensor([entity.get_link(link).idx_local], device=self.device)
            pos = entity.get_links_pos(ls_idx_local=idx, relative=True)[:, 0]
            quat = entity.get_links_quat(ls_idx_local=idx, relative=True)[:, 0]
        pos, quat = (v if v.ndim == 2 else v.unsqueeze(0).expand(self.num_envs, -1) for v in (pos, quat))
        self._cache[key] = (tensor_to_array(pos), tensor_to_array(quat))
    pos, quat = self._cache[key]
    return pos[env_id], quat[env_id]

locate

locate(name: str, camera: str | None, env_id: int) -> Footprint | None

Footprint of the pixels of camera showing entity name in env_id; None below :data:MIN_LOCATE_POINTS.

Source code in eden/extensions/scripted_demos/routines.py

def locate(self, name: str, camera: str | None, env_id: int) -> Footprint | None:
    """Footprint of the pixels of ``camera`` showing entity ``name`` in ``env_id``; None below
    :data:`MIN_LOCATE_POINTS`.
    """
    if not self.cameras:
        raise ValueError(
            "locate() looks through a camera, and this demonstrator was given none: pass "
            "cameras=('<camera name>', ...) to RoutineDemonstrator; the scene's cameras: "
            f"{sorted(self.env.cameras)}."
        )
    camera = camera or self.cameras[0]
    if camera not in self.cameras:
        raise KeyError(
            f"locate() camera '{camera}' is not one of this demonstrator's cameras {list(self.cameras)}."
        )
    segmentation, points = self._camera_frame(camera)
    entity = self._entity(name)
    ids = self.env.entity_segmentation_ids.get(entity.idx)
    if not ids:
        raise KeyError(
            f"locate('{name}'): no segmentation id belongs to entity '{name}' (idx {entity.idx}), so no pixel could "
            f"ever show it; the ids map to entities {sorted(self.env.entity_segmentation_ids)}. Check the scene's "
            f"segmentation level — reading it as 'not seen' would hide the mistake."
        )
    mask = np.isin(segmentation[env_id], ids)
    seen = points[env_id][mask]
    if len(seen) < MIN_LOCATE_POINTS:
        return None
    return footprint_from_points(seen)

Functions:

A tool routine that puts an object on or into its target (eden datagen --task simpler_put_carrot_on_plate, …).

Written against :class:~eden.extensions.scripted_demos.tools.Tools, for one robot, as the routines of :mod:eden.extensions.scripted_demos.routines are: look at the object, grip it from above, carry it over the goal and lower it on, then look again. The loop is goal-directed, so the episode's deliberate mistakes (the object put down off the goal first, with probability p_mistake) and a missed or dropped pick are recovered from by looking again.

The goal is where the object's own centre must come to rest, from :class:Goal: over a target entity's footprint (the towel, the plate), into a target container (a bowl, a tray), or inside a fixed box (the sink's basket, which is part of the fixed sink). The same routine runs privileged (footprints off the collision meshes) or camera-only (:meth:~eden.extensions.scripted_demos.tools.Tools.locate); only :func:_look chooses.

What differs between robots is the :class:Hand: where its open fingers reach and where it moves to clear the camera's view. The SIMPLER Bridge tasks (WidowX250) and the Polaris DROID tasks (Franka + Robotiq 2F85) each pass their own.

Attributes

SETTLE_STEP_S module-attribute

SETTLE_STEP_S = 0.4

SETTLE_DRIFT module-attribute

SETTLE_DRIFT = 0.003

SETTLE_MAX_S module-attribute

SETTLE_MAX_S = 4.0

REAIM_DISTANCE module-attribute

REAIM_DISTANCE = 0.008

HOVER_ABOVE module-attribute

HOVER_ABOVE = (0.04, 0.08)

HOVER_CLEARANCE module-attribute

HOVER_CLEARANCE = 0.02

CARRY_CLEARANCE module-attribute

CARRY_CLEARANCE = (0.03, 0.05)

GRASP_DEPTH module-attribute

GRASP_DEPTH = 0.02

FLOOR_GAP module-attribute

FLOOR_GAP = 0.002

GRASP_SLIDE module-attribute

GRASP_SLIDE = 0.15

GRASP_TOLERANCE module-attribute

GRASP_TOLERANCE = 0.005

MAX_GRASP_SLIDE module-attribute

MAX_GRASP_SLIDE = 0.8

FINGER_CLEARANCE module-attribute

FINGER_CLEARANCE = 0.012

GRASP_PAUSE module-attribute

GRASP_PAUSE = (0.2, 0.5)

RELEASE_GAP module-attribute

RELEASE_GAP = 0.01

PLACED_TOLERANCE module-attribute

PLACED_TOLERANCE = 0.02

PLACED_HEIGHT module-attribute

PLACED_HEIGHT = 0.02

PLACE_JITTER module-attribute

PLACE_JITTER = 0.008

PLACE_TOLERANCE module-attribute

PLACE_TOLERANCE = 0.015

MAX_PICK_ATTEMPTS module-attribute

MAX_PICK_ATTEMPTS = 3

MAX_PLACEMENTS module-attribute

MAX_PLACEMENTS = 4

RIM_DEPTH module-attribute

RIM_DEPTH = 0.02

RIM_INSET module-attribute

RIM_INSET = 0.004

ROUND_SLACK module-attribute

ROUND_SLACK = 0.01

SCRUB_STROKES module-attribute

SCRUB_STROKES = 3

CONTAINER_WALL module-attribute

CONTAINER_WALL = 0.02

SCRUB_REACH module-attribute

SCRUB_REACH = 0.06

Classes

Hand dataclass

Hand(finger_reach: tuple[float, float], clear_view_pos: tuple[float, float, float])

What the routine knows about the robot's hand that the rig does not measure.

ATTRIBUTE	DESCRIPTION
finger_reach	Distances [m] from the grasp point, along the closing axis, of an open finger's inner and outer faces. TYPE: tuple[float, float]
clear_view_pos	Where the hand goes to clear the camera's view when the object cannot be seen. TYPE: tuple[float, float, float]

Attributes

finger_reach instance-attribute

finger_reach: tuple[float, float]

clear_view_pos instance-attribute

clear_view_pos: tuple[float, float, float]

Methods:

Goal dataclass

Goal(target: str | None = None, region: tuple[tuple[float, float, float], tuple[float, float, float]] | None = None, object_yaw: float | None = None, obstacle_top: float = 0.0, mistake_bounds: tuple[tuple[float, float], tuple[float, float]] = ((0.2, -0.15), (0.35, 0.15)), mistake_floor: float = 0.0, container: bool = False, target_body: bool = False, max_reach: float | None = None, avoid: tuple[str, ...] = (), scrub: float = 0.0)

Where the object's centre must come to rest.

Exactly one of target and region is given.

A target is something to put the object on (the object counts as placed with its centre within :data:PLACED_TOLERANCE of the target's) unless container: then the object is let go above the target's rim and counts as placed with its centre anywhere over the target's footprint, inset by the object's half width — where a bowl or a tray holds it.

ATTRIBUTE	DESCRIPTION
target	Scene entity to put the object on: the goal is its footprint's centre and top. TYPE: str | None
region	Low / high corners of a fixed box the object goes into: the goal is its centre in xy, and its low z is the floor the object is lowered onto. TYPE: tuple[tuple[float, float, float], tuple[float, float, float]] | None
object_yaw	Heading [rad] the object's long axis is turned to (modulo pi) while carried; None keeps the heading it was picked at. TYPE: float | None
obstacle_top	Highest point [m] between the object and the goal — a rim the hand and the object pass over. TYPE: float
mistake_bounds	Low / high xy corners of where a deliberate mistake puts the object down instead. TYPE: tuple[tuple[float, float], tuple[float, float]]
mistake_floor	Height [m] of the surface inside mistake_bounds. TYPE: float
container	The target is a container the object is dropped into (see above). TYPE: bool
target_body	Aim at the round body of a target with a handle (a frying pan) instead of its footprint's centre, which the handle drags off it: the end of the footprint's long side its points crowd. TYPE: bool
max_reach	Container goals: the aim stays within this horizontal distance [m] of the robot's base (the world origin) where the container allows — the far side of a large one is out of the arm's reach — and otherwise comes as close to it as the container allows. TYPE: float | None
avoid	Container goals: entities already in the container (a tomato in the tray) the aim keeps clear of. TYPE: tuple[str, ...]
scrub	Lower the object onto the goal and rub it back and forth this far [m] to either side, :data:SCRUB_STROKES times, before letting it go (a sponge on a pan); 0 does not. TYPE: float

Attributes

target class-attribute instance-attribute

target: str | None = None

region class-attribute instance-attribute

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

object_yaw class-attribute instance-attribute

object_yaw: float | None = None

obstacle_top class-attribute instance-attribute

obstacle_top: float = 0.0

mistake_bounds class-attribute instance-attribute

mistake_bounds: tuple[tuple[float, float], tuple[float, float]] = ((0.2, -0.15), (0.35, 0.15))

mistake_floor class-attribute instance-attribute

mistake_floor: float = 0.0

container class-attribute instance-attribute

container: bool = False

target_body class-attribute instance-attribute

target_body: bool = False

max_reach class-attribute instance-attribute

max_reach: float | None = None

avoid class-attribute instance-attribute

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

scrub class-attribute instance-attribute

scrub: float = 0.0

Methods:

Functions:

put_on_routine

put_on_routine(source: str, goal: Goal, *, hand: Hand, p_mistake: float, camera: str | None, grasp: str = 'pinch', clutter: tuple[str, ...] = ()) -> Routine

The routine that puts source at goal; camera-only when camera is given.

grasp is how the object is gripped from above: "pinch" across its short side, or "rim" by the wall of a hollow object too wide to pinch (a cup, a roll of tape), one finger inside it. A pinch keeps the open fingers off the goal's target and off the clutter entities around the object (a tilted plate next to a cube), which the hand would come down on and stop short.

Source code in eden/extensions/scripted_demos/put_on_routine.py

def put_on_routine(
    source: str,
    goal: Goal,
    *,
    hand: Hand,
    p_mistake: float,
    camera: str | None,
    grasp: str = "pinch",
    clutter: tuple[str, ...] = (),
) -> Routine:
    """The routine that puts ``source`` at ``goal``; camera-only when ``camera`` is given.

    ``grasp`` is how the object is gripped from above: ``"pinch"`` across its short side, or ``"rim"``
    by the wall of a hollow object too wide to pinch (a cup, a roll of tape), one finger inside it.
    A pinch keeps the open fingers off the goal's target and off the ``clutter`` entities around the
    object (a tilted plate next to a cube), which the hand would come down on and stop short.
    """
    if grasp not in ("pinch", "rim"):
        raise ValueError(f"put_on_routine grasp must be 'pinch' or 'rim', got {grasp!r}.")
    if clutter and grasp == "rim":
        raise ValueError(
            f"clutter keeps a pinch's open fingers off nearby objects, and a rim grasp does not use it; got "
            f"grasp='rim' with clutter={clutter}. Drop clutter, or use grasp='pinch'."
        )

    def put_on(tools: Tools) -> Generator[Motion, Outcome, None]:
        mistake = tools.rng.random() < p_mistake
        for _ in range(MAX_PLACEMENTS):
            item = yield from _settled_look(tools, source, camera, hand)
            if item is None:
                tools.note(f"{source}: not seen, gave up")
                return
            target = _look(tools, goal.target, camera) if goal.target is not None else None
            if goal.target is not None and target is None:
                tools.note(f"{goal.target}: not seen, gave up")
                return
            centre, top = _goal_centre_and_top(goal, target)
            if _placed(item, goal, centre, top, target):
                return
            others = [other for other in (_look(tools, name, camera) for name in goal.avoid) if other is not None]
            centre = _aim(goal, target, item, centre, others)
            if mistake:
                mistake = False
                centre, top = _mistake_spot(tools, goal, item, centre), goal.mistake_floor
                tools.note(f"{source} -> mistake spot ({centre[0]:.3f}, {centre[1]:.3f})")
            else:
                centre = centre + tools.rng.uniform(-PLACE_JITTER, PLACE_JITTER, size=2)
                tools.note(f"{source} -> goal ({centre[0]:.3f}, {centre[1]:.3f})")
            if not (yield from _pick_and_place(tools, source, item, centre, top, goal, camera, hand, grasp, clutter)):
                tools.note(f"{source}: every pick missed")
        tools.note(f"gave up after {MAX_PLACEMENTS} placements")

    put_on.entity_names = (source, *((goal.target,) if goal.target is not None else ()), *goal.avoid, *clutter)
    return put_on

body_centre

body_centre(footprint: Footprint) -> tuple[ndarray, float]

The (2,) centre and radius [m] of the round body of an object with a handle (a mug, a pan).

The body is the end of the footprint's long side its points crowd (a handle is thin), as wide as the footprint's short side. A footprint about as long as wide is all body.

Source code in eden/extensions/scripted_demos/put_on_routine.py

def body_centre(footprint: Footprint) -> tuple[np.ndarray, float]:
    """The ``(2,)`` centre and radius [m] of the round body of an object with a handle (a mug, a pan).

    The body is the end of the footprint's long side its points crowd (a handle is thin), as wide as
    the footprint's short side. A footprint about as long as wide is all body.
    """
    spare = footprint.half_length - footprint.half_width
    if spare < ROUND_SLACK or footprint.centroid is None:
        return np.asarray(footprint.center[:2]), footprint.half_width
    along = np.array([math.cos(footprint.yaw), math.sin(footprint.yaw)])
    side = 1.0 if (footprint.centroid - footprint.center[:2]) @ along >= 0 else -1.0
    return footprint.center[:2] + side * spare * along, footprint.half_width

Multi-arm demonstrators: one routine per arm, run side by side in every env, handing off through a blackboard.

A two-arm task is scripted as two routines, each written for its own arm against its own :class:~eden.extensions.scripted_demos.tools.Tools handle, exactly like a single-arm routine. They run concurrently — each arm resumes its routine when its own motion ends — and coordinate through :attr:Tools.shared <eden.extensions.scripted_demos.tools.Tools.shared>, a dict both routines of an env share for the episode. A hand-off is one routine writing a key and the other waiting for it::

def giver(tools):
    ...  # carry the object to the hand-off pose
    tools.shared["offered"] = True
    while not tools.shared.get("taken"):
        yield tools.wait(0.1)
    yield tools.open_gripper()


def taker(tools):
    while not tools.shared.get("offered"):
        yield tools.wait(0.1)
    ...  # close on the object
    tools.shared["taken"] = True

:class:MultiArmDemonstrator runs one :class:~eden.extensions.scripted_demos.routines.RoutineDemonstrator per arm, merges their actions into the task's, and has the interface eden datagen drives (reset / update_subgoals / act / done / subgoal_counts / trace). An env's task is complete when every arm's routine has returned; an arm that finishes first holds still.

Attributes

Classes

MultiArmDemonstrator

MultiArmDemonstrator(env: EnvBase, routines: dict[str, Routine], *, seed: int = 0, **arm_options: Any)

Runs one routine per arm in every env, batched, with a blackboard per env shared between the arms.

PARAMETER	DESCRIPTION
env	A built environment holding each arm at its scene entry, each moved by one IK term and one gripper term, and no other action term. TYPE: EnvBase
routines	Scene entry of each arm -> the routine that drives it. Each is called with that arm's :class:~eden.extensions.scripted_demos.tools.Tools at the start of every episode. TYPE: dict[str, Routine]
seed	Seeds every arm's per-episode random streams (each arm its own). TYPE: int DEFAULT: 0
**arm_options	Passed to every arm's :class:~eden.extensions.scripted_demos.routines.RoutineDemonstrator (workspace, peak_speed, holding_closure, ...). An option that differs per arm is a dict keyed by the arm's scene entry, e.g. workspace={"robot0": ..., "robot1": ...}. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
ValueError	Fewer than two arms, or a per-arm option that does not name exactly the arms.
TypeError	An action term no arm drives.

Source code in eden/extensions/scripted_demos/multi_arm.py

def __init__(self, env: EnvBase, routines: dict[str, Routine], *, seed: int = 0, **arm_options: Any):
    if len(routines) < 2:
        raise ValueError(
            f"A MultiArmDemonstrator drives two or more arms, got routines for {sorted(routines)}; a single arm "
            f"is a RoutineDemonstrator(env, routine, rig='<slot>')."
        )
    for name, value in arm_options.items():
        if isinstance(value, dict) and set(value) != set(routines):
            raise ValueError(
                f"Per-arm option '{name}' is given for {sorted(value)}, but the arms are {sorted(routines)}; key it "
                f"by every arm's scene entry, or pass one value for all of them."
            )
    self.env = env
    self.arms: dict[str, RoutineDemonstrator] = {
        slot: RoutineDemonstrator(
            env,
            routine,
            rig=slot,
            seed=seed + i * _ARM_SEED_STRIDE,
            exclusive=False,
            **{name: value[slot] if isinstance(value, dict) else value for name, value in arm_options.items()},
        )
        for i, (slot, routine) in enumerate(routines.items())
    }
    self._check_action_terms()
    self.num_envs = env.num_envs
    self.device = env.device

Attributes

env instance-attribute

env = env

arms instance-attribute

arms: dict[str, RoutineDemonstrator] = {slot: RoutineDemonstrator(env, routine, rig=slot, seed=seed + i * _ARM_SEED_STRIDE, exclusive=False, **{name: value[slot] if isinstance(value, dict) else value for name, value in arm_options.items()}) for i, (slot, routine) in enumerate(routines.items())}

num_envs instance-attribute

num_envs = env.num_envs

device instance-attribute

device = env.device

done property

done: Tensor

(N,) bool: every arm's routine has returned — this env's task is complete.

subgoal_counts property

subgoal_counts: dict[str, int]

Motions ended so far, by <arm>/<tool>:<status> — the generation report's tally.

Methods:

reset

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

Start fresh routines on every arm for envs_idx, the arms of each env sharing one new blackboard.

Call it when :meth:~eden.extensions.scripted_demos.routines.RoutineDemonstrator.reset would be called.

Source code in eden/extensions/scripted_demos/multi_arm.py

def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    """Start fresh routines on every arm for ``envs_idx``, the arms of each env sharing one new blackboard.

    Call it when :meth:`~eden.extensions.scripted_demos.routines.RoutineDemonstrator.reset` would be called.
    """
    if envs_idx is None:
        env_ids = list(range(self.num_envs))
    else:
        env_ids = (
            envs_idx.nonzero(as_tuple=False).flatten() if envs_idx.dtype == torch.bool else envs_idx
        ).tolist()
    boards = {env_id: {} for env_id in env_ids}
    for arm in self.arms.values():
        arm.reset(envs_idx, blackboards=boards)

update_subgoals

update_subgoals() -> Tensor

Advance every arm's routines (see :meth:RoutineDemonstrator.update_subgoals); returns :attr:done.

Source code in eden/extensions/scripted_demos/multi_arm.py

def update_subgoals(self) -> torch.Tensor:
    """Advance every arm's routines (see :meth:`RoutineDemonstrator.update_subgoals`); returns :attr:`done`."""
    for arm in self.arms.values():
        arm.update_subgoals()
    return self.done

act

act() -> dict[str, Tensor]

The action dict for this control step: every arm's IK and gripper commands.

Source code in eden/extensions/scripted_demos/multi_arm.py

def act(self) -> dict[str, torch.Tensor]:
    """The action dict for this control step: every arm's IK and gripper commands."""
    actions: dict[str, torch.Tensor] = {}
    for arm in self.arms.values():
        actions.update(arm.act())
    return actions

trace

trace(env_id: int) -> list[str]

This episode's record for env_id: every arm's lines, prefixed with the arm, in time order.

Source code in eden/extensions/scripted_demos/multi_arm.py

def trace(self, env_id: int) -> list[str]:
    """This episode's record for ``env_id``: every arm's lines, prefixed with the arm, in time order."""
    lines = [(slot, line) for slot, arm in self.arms.items() for line in arm.trace(env_id)]

    def at(entry: tuple[str, str]) -> float:
        match = _TRACE_TIME.match(entry[1])
        return float(match.group(1)) if match else 0.0

    return [f"[{slot}] {line}" for slot, line in sorted(lines, key=at)]

See also

• Scripted demonstrations guide
• Datasets and recording guide