Skip to content

Recorder Terms

Recorder terms capture rollout data (actions, observations, RL transitions, entity states, sensor readings, system-identification signals) for the recorder manager to write to an episode file. The metrics and recorders guide and the datasets and recording guide explain the file layout; the term catalog lists every parameter.

from eden import recorders

The base class RecorderTerm is on the managers page.

Actions

Recorder term capturing per-step actions.

Attributes

Classes

ActionRecorder 

ActionRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

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

Methods:

record_pre_step 
record_pre_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/actions.py 
12
13
def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    return {"action": self._env.action_manager.action.clone()}

Observations

Recorder term capturing the observations each action was chosen from.

Attributes

Classes

ObservationRecorder 

ObservationRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record observation groups at the start of each step, as obs/<group>.

The start of a step is when the policy — or a scripted demonstrator — reads the observations and picks the action :class:~eden.managers.terms.recorders.actions.ActionRecorder records for that same step, so row t of obs and action form one (o_t, a_t) pair: the shape a behavior-cloning dataset needs.

PARAMETER DESCRIPTION
groups

Observation groups to record; every group the env computes when None.

TYPE: list[str] | None

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

Attributes

groups class-attribute instance-attribute 
groups: list[str] | None = None

Methods:

build 
build() -> None
Source code in eden/managers/terms/recorders/observations.py 
27
28
29
30
31
32
33
34
def build(self) -> None:
    available = list(self._env.observation_manager.group_term_names)
    unknown = [group for group in self.groups or [] if group not in available]
    if unknown:
        raise KeyError(
            f"ObservationRecorder.groups names observation groups {unknown} that this env does not have; "
            f"available groups: {available}."
        )
record_pre_step 
record_pre_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/observations.py 
36
37
38
39
40
41
42
43
44
def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    obs = self._env.obs_buf
    if obs is None:
        raise RuntimeError(
            "ObservationRecorder recorded a step before the env computed any observations; reset the env "
            "(env.reset()) before stepping it."
        )
    groups = self.groups if self.groups is not None else list(obs)
    return {"obs": {group: _clone(obs[group]) for group in groups}}

RL transitions

Recorder term capturing RL step transitions.

Attributes

Classes

RLStepRecorder 

RLStepRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record per-step RL buffers (reward, done, terminated, timeout, timestamp).

Reads RLEnvBase step buffers after rewards and terminations are computed. Opt in by registering on recorder_options for RL tasks; not applicable to non-RL EnvBase flows (the buffers don't exist there).

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

Attributes

include_reward class-attribute instance-attribute 
include_reward: bool = True
include_done class-attribute instance-attribute 
include_done: bool = True
include_terminated class-attribute instance-attribute 
include_terminated: bool = True
include_timeout class-attribute instance-attribute 
include_timeout: bool = True
include_timestamp class-attribute instance-attribute 
include_timestamp: bool = True

Methods:

record_post_step 
record_post_step() -> dict[str, Tensor]
Source code in eden/managers/terms/recorders/rl.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
def record_post_step(self) -> dict[str, torch.Tensor]:
    out: dict[str, torch.Tensor] = {}
    if self.include_reward and hasattr(self._env, "reward_buf"):
        out["reward"] = self._env.reward_buf.unsqueeze(-1)
    if self.include_done and hasattr(self._env, "reset_buf"):
        out["done"] = self._env.reset_buf.to(dtype=torch.bool).unsqueeze(-1)
    if self.include_terminated and hasattr(self._env, "reset_terminated"):
        out["terminated"] = self._env.reset_terminated.to(dtype=torch.bool).unsqueeze(-1)
    if self.include_timeout and hasattr(self._env, "reset_timeouts"):
        out["timeout"] = self._env.reset_timeouts.to(dtype=torch.bool).unsqueeze(-1)
    if self.include_timestamp and hasattr(self._env, "episode_length_buf") and hasattr(self._env, "dt"):
        out["timestamp"] = self._env.episode_length_buf.to(dtype=torch.float32).unsqueeze(-1) * float(self._env.dt)
    return out

Entity states

Recorder terms capturing initial and per-step entity states.

Attributes

Classes

InitialStateRecorder 

InitialStateRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record per-episode initial entity states right after reset.

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

Attributes

include_qpos class-attribute instance-attribute 
include_qpos: bool = True
include_dofs_vel class-attribute instance-attribute 
include_dofs_vel: bool = True
include_pos_quat class-attribute instance-attribute 
include_pos_quat: bool = True

Methods:

record_post_reset 
record_post_reset(envs_idx=None) -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/states.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def record_post_reset(self, envs_idx=None) -> dict[str, torch.Tensor | dict]:
    if not self._env.entities:
        return {}

    entities_state: dict[str, dict[str, torch.Tensor]] = {}
    for entity_name, entity in self._env.entities.items():
        entity_state: dict[str, torch.Tensor] = {}

        if self.include_qpos:
            entity_state["qpos"] = entity.get_qpos(qs_idx_local=entity.qs_idx_local, envs_idx=envs_idx).clone()
        if self.include_dofs_vel:
            entity_state["dofs_vel"] = entity.get_dofs_vel(envs_idx=envs_idx).clone()
        if self.include_pos_quat:
            entity_state["pos"] = entity.get_pos(envs_idx=envs_idx).clone()
            entity_state["quat"] = entity.get_quat(envs_idx=envs_idx).clone()

        if entity_state:
            entities_state[entity_name] = entity_state

    if not entities_state:
        return {}

    return {"initial_state": {"entities": entities_state}}

StateRecorder 

StateRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record the state of the environment at the beginning of each step.

entity_name is one entity, or several — the two arms of a two-arm task — whose states are concatenated in that order along the last axis of each key (qpos, dofs_vel, pos, quat), so a dataset keeps one state/<key> per key whatever the number of arms.

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

Attributes

entity_name class-attribute instance-attribute 
entity_name: str | list[str] = 'robot'
include_qpos class-attribute instance-attribute 
include_qpos: bool = True
include_dofs_vel class-attribute instance-attribute 
include_dofs_vel: bool = True
include_pos_quat class-attribute instance-attribute 
include_pos_quat: bool = True

Methods:

build 
build() -> None
Source code in eden/managers/terms/recorders/states.py 
57
58
59
60
61
62
63
def build(self) -> None:
    names = [self.entity_name] if isinstance(self.entity_name, str) else list(self.entity_name)
    if not names:
        raise ValueError("StateRecorder.entity_name lists no entity; name the entity (or entities) to record.")
    for name in names:
        self.resolve_entity(name)
    self._entity_names = names
record_pre_step 
record_pre_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/states.py 
65
66
67
def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    states = [self._entity_state(self.resolve_entity(name)) for name in self._entity_names]
    return {"state": {key: torch.cat([state[key] for state in states], dim=-1) for key in states[0]}}

Sensors

Recorder term capturing sensor readings.

Attributes

Classes

SensorRecorder 

SensorRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record sensor outputs during simulation.

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

Attributes

use_ground_truth class-attribute instance-attribute 
use_ground_truth: bool = False
record_post_reset_snapshot class-attribute instance-attribute 
record_post_reset_snapshot: bool = True

Methods:

record_post_reset 
record_post_reset(envs_idx=None) -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/sensors.py 
47
48
49
50
def record_post_reset(self, envs_idx=None) -> dict[str, torch.Tensor | dict]:
    if not self.record_post_reset_snapshot:
        return {}
    return self._read_sensors(envs_idx=envs_idx)
record_pre_step 
record_pre_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/sensors.py 
52
53
def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    return self._read_sensors()

System identification

Recorder term capturing data for system identification.

Attributes

Classes

SysIDRecorder 

SysIDRecorder(env: EnvBase, options: _TermOptionsT)

Bases: RecorderTerm

Record the proprioceptive trace needed to rebuild a sysid Trajectory.

Captures the signals under the sysid/ namespace so that eden.extensions.sysid.Trajectory.from_recorder_episode can reload an episode without any extra metadata.

PARAMETER DESCRIPTION
entity_name

Entity to record. Must exist in env.entities.
include_torque

Record get_dofs_control_force() under sysid/dofs_control_force.
include_base

Record the base pose (pos/quat) at reset and the base quat / body-frame ang_vel each step. Set False for fixed-base arms.
Source code in eden/managers/base.py 
223
224
225
def __init__(self, env: EnvBase, options: _TermOptionsT):
    ManagerTermBase.__init__(self, env=env)
    ConfigurableMixin.__init__(self, options=options)

Attributes

entity_name class-attribute instance-attribute 
entity_name: str = 'robot'
include_torque class-attribute instance-attribute 
include_torque: bool = True
include_base class-attribute instance-attribute 
include_base: bool = True

Methods:

record_post_reset 
record_post_reset(envs_idx=None) -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/sysid.py 
33
34
35
36
37
38
39
40
41
42
def record_post_reset(self, envs_idx=None) -> dict[str, torch.Tensor | dict]:
    entity = self.resolve_entity(self.entity_name)
    initial_state: dict[str, torch.Tensor] = {
        "qpos": entity.get_qpos(qs_idx_local=entity.qs_idx_local, envs_idx=envs_idx).clone(),
        "dofs_vel": entity.get_dofs_vel(envs_idx=envs_idx).clone(),
    }
    if self.include_base:
        initial_state["pos"] = entity.get_pos(envs_idx=envs_idx).clone()
        initial_state["quat"] = entity.get_quat(envs_idx=envs_idx).clone()
    return {"sysid": {"initial_state": initial_state}}
record_pre_step 
record_pre_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/sysid.py 
44
45
46
47
def record_pre_step(self) -> dict[str, torch.Tensor | dict]:
    action = self._env.action_manager.action.clone()
    stamp = (self._env.episode_length_buf.float() * self._env.dt).unsqueeze(-1)
    return {"sysid": {"action": action, "stamp": stamp}}
record_post_step 
record_post_step() -> dict[str, Tensor | dict]
Source code in eden/managers/terms/recorders/sysid.py 
49
50
51
52
53
54
55
56
57
58
59
60
def record_post_step(self) -> dict[str, torch.Tensor | dict]:
    entity = self.resolve_entity(self.entity_name)
    out: dict[str, torch.Tensor] = {
        "dofs_pos": entity.get_dofs_pos().clone(),
        "dofs_vel": entity.get_dofs_vel().clone(),
    }
    if self.include_torque:
        out["dofs_control_force"] = entity.get_dofs_control_force().clone()
    if self.include_base:
        out["base_quat"] = entity.get_quat().clone()
        out["base_ang_vel"] = entity.get_ang(frame="body").clone()
    return {"sysid": out}

See also