Skip to content

Metric Terms

Metric terms grade a rollout: each reports a value per env with a direction (MetricDirection, higher or lower is better), a mode (MetricMode) and a success_threshold, and the metric manager combines the ones that contribute to success. Rewards and terminations can read a metric instead of recomputing it. The metrics and recorders guide explains them; the term catalog lists every parameter.

from eden import metrics
from eden.constants import MetricDirection, MetricMode

The base classes MetricTerm and MetricTermFuncWrapper are on the managers page.

Proximity

Proximity and relative-position metric terms between entities.

Attributes

Classes

EeNearEntity 

EeNearEntity(env: EnvBase, options: _TermOptionsT)

Bases: MetricTerm

Check if the mean end-effector position is within threshold of a target entity.

This is a class-based metric that caches link indices at build() time for efficiency.

PARAMETER DESCRIPTION
robot_name

Name of the robot entity.
ee_link_names

Link names whose mean position is treated as the EE position.
entity_name

Name of the target entity.
threshold

Maximum allowed distance.
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

robot_name class-attribute instance-attribute 
robot_name: str = 'robot'

ee_link_names: list[str] = []
entity_name class-attribute instance-attribute 
entity_name: str = 'obj'
threshold class-attribute instance-attribute 
threshold: float = 0.05

Methods:

build 
build()
Source code in eden/managers/terms/metrics/proximity.py 
246
247
248
249
250
def build(self):
    self._robot = self.resolve_entity(self.robot_name)
    self._target = self.resolve_entity(self.entity_name)
    _, ee_idx = self._robot.find_named_links_idx_local(self.ee_link_names, preserve_order=True)
    self._ee_idx_local = ee_idx
compute 
compute() -> Tensor
Source code in eden/managers/terms/metrics/proximity.py 
252
253
254
255
256
def compute(self) -> torch.Tensor:
    ee_pos = self._robot.get_links_pos(ls_idx_local=self._ee_idx_local).mean(dim=1)
    obj_pos = self._target.get_pos()
    dist = torch.linalg.vector_norm(ee_pos - obj_pos, dim=1)
    return dist < self.threshold

Functions:

object_a_is_on_b 

object_a_is_on_b(env: EnvBase, *, entity_a_name: str, entity_b_name: str, xy_threshold: float = 0.03, height_threshold: float = 0.04, height_diff: float = 0.0)

Check if object_a is on object_b by comparing height and xy distance of object origins.

This function can be used as a termination term or a metric term.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_a_name

The name of the entity_a.

TYPE: str

entity_b_name

The name of the entity_b.

TYPE: str

xy_threshold

The threshold for the xy distance.

TYPE: float DEFAULT: 0.03

height_threshold

The threshold for the height distance.

TYPE: float DEFAULT: 0.04

height_diff

The expected height difference.

TYPE: float DEFAULT: 0.0

RETURNS DESCRIPTION
success

A boolean tensor indicating if the object_a is on the object_b.

TYPE: Tensor

Source code in eden/managers/terms/metrics/proximity.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def object_a_is_on_b(
    env: EnvBase,
    *,
    entity_a_name: str,
    entity_b_name: str,
    xy_threshold: float = 0.03,
    height_threshold: float = 0.04,
    height_diff: float = 0.0,
):
    """Check if object_a is on object_b by comparing height and xy distance of object origins.

    This function can be used as a termination term or a metric term.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_a_name: str
        The name of the entity_a.
    entity_b_name: str
        The name of the entity_b.
    xy_threshold: float
        The threshold for the xy distance.
    height_threshold: float
        The threshold for the height distance.
    height_diff: float
        The expected height difference.

    Returns
    -------
    success: torch.Tensor
        A boolean tensor indicating if the object_a is on the object_b.
    """
    object_a: Entity = env.get_entity(entity_a_name)
    object_b: Entity = env.get_entity(entity_b_name)

    pos_diff = object_a.get_pos() - object_b.get_pos()
    height_dist = torch.linalg.vector_norm(pos_diff[:, 2:], dim=1)
    xy_dist = torch.linalg.vector_norm(pos_diff[:, :2], dim=1)

    success = torch.logical_and(
        pos_diff[:, 2] > 0.0,  # object_a is above object_b
        torch.logical_and(xy_dist < xy_threshold, (height_dist - height_diff) < height_threshold),
    )
    return success

object_a_within_b_xy 

object_a_within_b_xy(env: EnvBase, *, entity_a_name: str, entity_b_name: str, min_fraction: float = 0.8) -> Tensor

min_fraction of entity_a's footprint lies over entity_b's.

Both footprints are the xy extent of the entity's world-frame bounding box (its collision geometry), so a cup over a cutting board, or a scissor dropped into a container, passes whatever its height: pair it with a height or contact check to require that it rests there. PolaRiS's is_within_xy rubric checker (eden.tasks.polaris) is this test.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_a_name

The entity that must lie over the other.

TYPE: str

entity_b_name

The entity it must lie over.

TYPE: str

min_fraction

Smallest share of entity_a's footprint area that must overlap entity_b's, in (0, 1].

TYPE: float DEFAULT: 0.8

Source code in eden/managers/terms/metrics/proximity.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@METRIC_TERM_REGISTRY.register()
def object_a_within_b_xy(
    env: EnvBase,
    *,
    entity_a_name: str,
    entity_b_name: str,
    min_fraction: float = 0.8,
) -> torch.Tensor:
    """*min_fraction* of entity_a's footprint lies over entity_b's.

    Both footprints are the xy extent of the entity's world-frame bounding box (its collision geometry),
    so a cup over a cutting board, or a scissor dropped into a container, passes whatever its height:
    pair it with a height or contact check to require that it rests there. PolaRiS's ``is_within_xy``
    rubric checker (``eden.tasks.polaris``) is this test.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_a_name: str
        The entity that must lie over the other.
    entity_b_name: str
        The entity it must lie over.
    min_fraction: float
        Smallest share of entity_a's footprint area that must overlap entity_b's, in ``(0, 1]``.
    """
    a = env.get_entity(entity_a_name).get_AABB()[..., :2]  # (num_envs, 2 [low, high], 2)
    b = env.get_entity(entity_b_name).get_AABB()[..., :2]
    overlap = (torch.minimum(a[:, 1], b[:, 1]) - torch.maximum(a[:, 0], b[:, 0])).clamp(min=0.0).prod(dim=1)
    area = (a[:, 1] - a[:, 0]).prod(dim=1)
    return overlap >= min_fraction * area

entity_distance 

entity_distance(env: EnvBase, *, entity_a_name: str, entity_b_name: str) -> Tensor

Distance between two entity root positions.

Returns a float tensor of shape (num_envs,) with the Euclidean distance. Registered with direction=LIB (lower is better).

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_a_name

Name of the first entity.

TYPE: str

entity_b_name

Name of the second entity.

TYPE: str

Source code in eden/managers/terms/metrics/proximity.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@METRIC_TERM_REGISTRY.register(direction=MetricDirection.LIB)
def entity_distance(
    env: EnvBase,
    *,
    entity_a_name: str,
    entity_b_name: str,
) -> torch.Tensor:
    """Distance between two entity root positions.

    Returns a float tensor of shape ``(num_envs,)`` with the Euclidean
    distance. Registered with ``direction=LIB`` (lower is better).

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_a_name: str
        Name of the first entity.
    entity_b_name: str
        Name of the second entity.
    """
    entity_a: Entity = env.get_entity(entity_a_name)
    entity_b: Entity = env.get_entity(entity_b_name)
    pos_diff = entity_a.get_pos() - entity_b.get_pos()
    return torch.linalg.vector_norm(pos_diff, dim=1)

entity_near_target 

entity_near_target(env: EnvBase, *, entity_name: str, target_pos: Vec3FType | Tensor, threshold: float = 0.05) -> Tensor

Check if an entity is within threshold of a fixed world position.

Returns a per-env bool tensor (True if within threshold).

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the entity to check.

TYPE: str

target_pos

Target world position [x, y, z]. A plain list is converted to a tensor once per distinct value (device_constant) and reused.

TYPE: Vec3FType | Tensor

threshold

Maximum allowed distance.

TYPE: float DEFAULT: 0.05

Source code in eden/managers/terms/metrics/proximity.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def entity_near_target(
    env: EnvBase,
    *,
    entity_name: str,
    target_pos: Vec3FType | torch.Tensor,
    threshold: float = 0.05,
) -> torch.Tensor:
    """Check if an entity is within *threshold* of a fixed world position.

    Returns a per-env bool tensor (True if within threshold).

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the entity to check.
    target_pos: list[float] | Tensor
        Target world position ``[x, y, z]``.  A plain list is converted to a
        tensor once per distinct value (``device_constant``) and reused.
    threshold: float
        Maximum allowed distance.
    """
    entity: Entity = env.get_entity(entity_name)
    if not isinstance(target_pos, torch.Tensor):
        target_pos = device_constant(target_pos, env.device).unsqueeze(0)
    dist = torch.linalg.vector_norm(entity.get_pos() - target_pos, dim=1)
    return dist < threshold

entity_distance_xy 

entity_distance_xy(env: EnvBase, *, entity_a_name: str, entity_b_name: str) -> Tensor

Horizontal (XY) distance between two entity root positions.

Like :func:entity_distance but ignores the Z axis. Useful for placement tasks where objects sit at different heights (e.g. an apple on a plate).

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_a_name

Name of the first entity.

TYPE: str

entity_b_name

Name of the second entity.

TYPE: str

Source code in eden/managers/terms/metrics/proximity.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@METRIC_TERM_REGISTRY.register(direction=MetricDirection.LIB)
def entity_distance_xy(
    env: EnvBase,
    *,
    entity_a_name: str,
    entity_b_name: str,
) -> torch.Tensor:
    """Horizontal (XY) distance between two entity root positions.

    Like :func:`entity_distance` but ignores the Z axis.  Useful for
    placement tasks where objects sit at different heights (e.g. an apple
    on a plate).

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_a_name: str
        Name of the first entity.
    entity_b_name: str
        Name of the second entity.
    """
    entity_a: Entity = env.get_entity(entity_a_name)
    entity_b: Entity = env.get_entity(entity_b_name)
    pos_diff = entity_a.get_pos()[:, :2] - entity_b.get_pos()[:, :2]
    return torch.linalg.vector_norm(pos_diff, dim=1)

entity_near_entity_xy 

entity_near_entity_xy(env: EnvBase, *, entity_a_name: str, entity_b_name: str, threshold: float = 0.05) -> Tensor

Check if two entities are within threshold horizontal (XY) distance.

Returns a per-env bool tensor, True when the XY distance between root positions is below threshold. Useful for placement tasks where objects naturally sit at different heights.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_a_name

Name of the first entity.

TYPE: str

entity_b_name

Name of the second entity.

TYPE: str

threshold

Maximum allowed horizontal distance.

TYPE: float DEFAULT: 0.05

Source code in eden/managers/terms/metrics/proximity.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def entity_near_entity_xy(
    env: EnvBase,
    *,
    entity_a_name: str,
    entity_b_name: str,
    threshold: float = 0.05,
) -> torch.Tensor:
    """Check if two entities are within *threshold* horizontal (XY) distance.

    Returns a per-env bool tensor, True when the XY distance between root
    positions is below *threshold*.  Useful for placement tasks where objects
    naturally sit at different heights.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_a_name: str
        Name of the first entity.
    entity_b_name: str
        Name of the second entity.
    threshold: float
        Maximum allowed horizontal distance.
    """
    entity_a: Entity = env.get_entity(entity_a_name)
    entity_b: Entity = env.get_entity(entity_b_name)
    pos_diff = entity_a.get_pos()[:, :2] - entity_b.get_pos()[:, :2]
    dist = torch.linalg.vector_norm(pos_diff, dim=1)
    return dist < threshold

Containment

Containment metric terms (object above height / inside region).

Attributes

Classes

Functions:

object_above_height 

object_above_height(env: EnvBase, *, entity_name: str, height: float, relative: bool = False) -> Tensor

Check if an entity's root z-position is above height.

Returns a per-env bool tensor (True if above).

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the entity.

TYPE: str

height

Minimum height threshold.

TYPE: float

relative

Frame height is measured in. False (the default) reads the solver frame, which Genesis re-origins onto a rigid body's inertial frame; True reads the frame the asset was authored in. Pass True whenever height was reasoned from the asset's own geometry: the two frames differ by the body's centre-of-mass offset, and because that offset is expressed in the body frame, the reported z swings by it as the body tilts.

Only a rigid entity has the second frame, so True against a deformable raises from :meth:~eden.entities.base.Entity._reject_relative. Unlike the observation terms, which the manager warm-computes while sizing their buffers, this is a metric term and so first runs on step 1 rather than at env construction.

TYPE: bool DEFAULT: False

Source code in eden/managers/terms/metrics/containment.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def object_above_height(
    env: EnvBase,
    *,
    entity_name: str,
    height: float,
    relative: bool = False,
) -> torch.Tensor:
    """Check if an entity's root z-position is above *height*.

    Returns a per-env bool tensor (True if above).

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the entity.
    height: float
        Minimum height threshold.
    relative: bool
        Frame *height* is measured in. ``False`` (the default) reads the solver frame, which Genesis
        re-origins onto a rigid body's inertial frame; ``True`` reads the frame the asset was
        authored in. Pass ``True`` whenever *height* was reasoned from the asset's own geometry: the
        two frames differ by the body's centre-of-mass offset, and because that offset is expressed
        in the body frame, the reported z swings by it as the body tilts.

        Only a rigid entity has the second frame, so ``True`` against a deformable raises from
        :meth:`~eden.entities.base.Entity._reject_relative`. Unlike the observation terms, which the
        manager warm-computes while sizing their buffers, this is a metric term and so first runs on
        step 1 rather than at env construction.
    """
    entity: Entity = env.get_entity(entity_name)
    return entity.get_pos(relative=relative)[:, 2] > height

object_bottom_above_height 

object_bottom_above_height(env: EnvBase, *, entity_name: str, height: float) -> Tensor

Lowest point of the entity above height.

Unlike :func:object_above_height, this does not depend on where the asset puts its origin (a scanned mesh's origin may sit at its base, its centre or off to one side), so height=0.03 with the table at z = 0 reads "lifted 3 cm off the table" for any object.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the entity.

TYPE: str

height

World z [m] the entity's bounding box must clear.

TYPE: float

Source code in eden/managers/terms/metrics/containment.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@METRIC_TERM_REGISTRY.register()
def object_bottom_above_height(
    env: EnvBase,
    *,
    entity_name: str,
    height: float,
) -> torch.Tensor:
    """Lowest point of the entity above *height*.

    Unlike :func:`object_above_height`, this does not depend on where the asset puts its origin (a
    scanned mesh's origin may sit at its base, its centre or off to one side), so ``height=0.03`` with
    the table at ``z = 0`` reads "lifted 3 cm off the table" for any object.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the entity.
    height: float
        World ``z`` [m] the entity's bounding box must clear.
    """
    entity: Entity = env.get_entity(entity_name)
    return entity.get_AABB()[:, 0, 2] > height

object_in_region 

object_in_region(env: EnvBase, *, entity_name: str, region_min: list[float], region_max: list[float], relative: bool = False) -> Tensor

Check if an entity's root position is inside an axis-aligned box.

Returns a per-env bool tensor (True if inside, bounds inclusive).

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the entity.

TYPE: str

region_min

Lower corner of the box [x_min, y_min, z_min].

TYPE: list[float]

region_max

Upper corner of the box [x_max, y_max, z_max].

TYPE: list[float]

relative

Frame the box is expressed in — see :func:object_above_height, including why True needs a rigid entity and only says so on step 1. Pass True when the corners were reasoned from the asset's own geometry rather than measured in the solver frame.

TYPE: bool DEFAULT: False

Source code in eden/managers/terms/metrics/containment.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def object_in_region(
    env: EnvBase,
    *,
    entity_name: str,
    region_min: list[float],
    region_max: list[float],
    relative: bool = False,
) -> torch.Tensor:
    """Check if an entity's root position is inside an axis-aligned box.

    Returns a per-env bool tensor (True if inside, bounds inclusive).

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the entity.
    region_min: list[float]
        Lower corner of the box ``[x_min, y_min, z_min]``.
    region_max: list[float]
        Upper corner of the box ``[x_max, y_max, z_max]``.
    relative: bool
        Frame the box is expressed in — see :func:`object_above_height`, including why ``True``
        needs a rigid entity and only says so on step 1. Pass ``True`` when the corners were
        reasoned from the asset's own geometry rather than measured in the solver frame.
    """
    entity: Entity = env.get_entity(entity_name)
    pos = entity.get_pos(relative=relative)
    # Built once per distinct box, not per call: an upload per step is a GPU->CPU sync.
    lo = device_constant(region_min, env.device).unsqueeze(0)
    hi = device_constant(region_max, env.device).unsqueeze(0)
    inside = (pos >= lo).all(dim=1) & (pos <= hi).all(dim=1)
    return inside

Grasping

Grasp metric terms (object lifted, is-grasping).

Attributes

Classes

IsGrasping 

IsGrasping(env: EnvBase, options: MetricTermOptions)

Bases: MetricTerm

Detect a parallel-jaw grasp via contact force alignment.

Returns a per-env bool, True when both gripper pads have contact forces aligned with their closing direction above force_threshold. This correctly detects a grasp even when the fingers cannot fully close (e.g. when an object is between them).

Shares the force test of :func:~eden.managers.terms.utils.parallel_jaw_is_grasping with :class:~eden.managers.terms.actions.welding_actions.ParallelJawWelding, but sums each pad's contacts over every link it touches. A grasp spanning two links of one object counts, and so do pads pressing two different objects. The weld action instead requires both pads on the one link it would weld, so this metric can read True in a step that welds nothing.

PARAMETER DESCRIPTION
robot_name

Name of the robot entity.

TYPE: str

left_gripper_link_name

Link name of the left gripper pad (e.g. "left_gripper_pad").

TYPE: str

right_gripper_link_name

Link name of the right gripper pad (e.g. "right_gripper_pad").

TYPE: str

force_threshold

Minimum projected contact force magnitude on each pad to count as grasping. Lower values are more sensitive.

TYPE: float

Source code in eden/managers/terms/metrics/grasp.py 
78
79
80
81
82
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._left_link_idx: int = -1
    self._right_link_idx: int = -1
    self._gripper_idx_local: list[int] = []

Attributes

robot_name class-attribute instance-attribute 
robot_name: str = 'robot'

left_gripper_link_name: str = 'left_gripper_pad'

right_gripper_link_name: str = 'right_gripper_pad'
force_threshold class-attribute instance-attribute 
force_threshold: float = 1.0

Methods:

build 
build()
Source code in eden/managers/terms/metrics/grasp.py 
84
85
86
87
88
89
90
91
92
def build(self):
    entity = self.resolve_entity(self.robot_name)
    self._entity = entity
    self._left_link_idx = entity.get_link(self.left_gripper_link_name).idx
    self._right_link_idx = entity.get_link(self.right_gripper_link_name).idx
    self._gripper_idx_local = [
        entity.get_link(self.left_gripper_link_name).idx_local,
        entity.get_link(self.right_gripper_link_name).idx_local,
    ]
compute 
compute() -> Tensor
Source code in eden/managers/terms/metrics/grasp.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def compute(self) -> torch.Tensor:
    contact_info = self._entity.get_contacts(exclude_self_contact=True)
    return parallel_jaw_is_grasping(
        self._entity,
        contact_info,
        self._left_link_idx,
        self._right_link_idx,
        self._gripper_idx_local,
        self.force_threshold,
    )

Functions:

object_lifted 

object_lifted(env: EnvBase, *, entity_name: str, min_height: float) -> Tensor

Check if an entity has been lifted above min_height.

Returns a per-env bool tensor, True when the entity's root z-position (solver frame) exceeds min_height.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the entity.

TYPE: str

min_height

World-z height threshold [m].

TYPE: float

Source code in eden/managers/terms/metrics/grasp.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def object_lifted(
    env: EnvBase,
    *,
    entity_name: str,
    min_height: float,
) -> torch.Tensor:
    """Check if an entity has been lifted above *min_height*.

    Returns a per-env bool tensor, True when the entity's root z-position
    (solver frame) exceeds *min_height*.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the entity.
    min_height: float
        World-z height threshold [m].
    """
    entity: Entity = env.get_entity(entity_name)
    return entity.get_pos()[:, 2] > min_height

Articulation

Articulation metric terms (e.g. joint angle at target).

Attributes

Classes

Functions:

joint_angle_at_target 

joint_angle_at_target(env: EnvBase, *, entity_name: str, joint_name: str, target_angle: float, threshold: float = 0.05) -> Tensor

Check if a single DOF is within threshold of target_angle.

Returns a per-env bool tensor, True when the joint position satisfies |current - target_angle| < threshold.

PARAMETER DESCRIPTION
env

The environment instance.

TYPE: EnvBase

entity_name

Name of the articulated entity.

TYPE: str

joint_name

Name of the DOF to query.

TYPE: str

target_angle

Target joint position (radians or meters depending on joint type).

TYPE: float

threshold

Maximum allowed deviation.

TYPE: float DEFAULT: 0.05

Source code in eden/managers/terms/metrics/articulation.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@TERMINATION_TERM_REGISTRY.register()
@METRIC_TERM_REGISTRY.register()
def joint_angle_at_target(
    env: EnvBase,
    *,
    entity_name: str,
    joint_name: str,
    target_angle: float,
    threshold: float = 0.05,
) -> torch.Tensor:
    """Check if a single DOF is within *threshold* of *target_angle*.

    Returns a per-env bool tensor, True when the joint position satisfies
    ``|current - target_angle| < threshold``.

    Parameters
    ----------
    env: EnvBase
        The environment instance.
    entity_name: str
        Name of the articulated entity.
    joint_name: str
        Name of the DOF to query.
    target_angle: float
        Target joint position (radians or meters depending on joint type).
    threshold: float
        Maximum allowed deviation.
    """
    entity: Entity = env.get_entity(entity_name)
    _, dof_idx = entity.find_named_dofs_idx_local([joint_name], preserve_order=True)
    dof_pos = entity.get_dofs_pos(dofs_idx_local=dof_idx)  # (num_envs, 1)
    error = torch.abs(dof_pos[:, 0] - target_angle)
    return error < threshold

Velocity-tracking error

Velocity-tracking error metric terms.

Attributes

Classes

Functions:

error_vel_xy 

error_vel_xy(env: RLEnvBase, *, command_name: str = 'velocity')
Source code in eden/managers/terms/metrics/error.py 
16
17
18
19
20
21
22
23
24
25
26
27
@METRIC_TERM_REGISTRY.register(
    is_cumulative=True,
    direction=MetricDirection.LIB,
    metric_mode=MetricMode.RESET,
)
def error_vel_xy(
    env: RLEnvBase,
    *,
    command_name: str = "velocity",
):
    command: CommandTerm = env.command_manager.get_term(command_name)
    return torch.norm(command.vel_command_b[:, :2] - command.robot.get_vel()[:, :2], dim=-1)

error_vel_yaw 

error_vel_yaw(env: RLEnvBase, *, command_name: str = 'velocity')
Source code in eden/managers/terms/metrics/error.py 
30
31
32
33
34
35
36
37
38
39
40
41
@METRIC_TERM_REGISTRY.register(
    is_cumulative=True,
    direction=MetricDirection.LIB,
    metric_mode=MetricMode.RESET,
)
def error_vel_yaw(
    env: RLEnvBase,
    *,
    command_name: str = "velocity",
):
    command: CommandTerm = env.command_manager.get_term(command_name)
    return torch.abs(command.vel_command_b[:, 2] - command.robot.get_ang()[:, 2])

Boolean combinators

Boolean combinators (AND/OR/NOT) over metric terms.

Attributes

Classes

MetricTermAND 

MetricTermAND(env: EnvBase, options: MetricTermOptions)

Bases: _BooleanMetricTermBase

Metric term that returns the element-wise minimum of its child terms.

All child terms must be "passing" for this term to pass.

PARAMETER DESCRIPTION
terms

Child term configurations (from .configure()).

TYPE: list[MetricTermOptions]

Example

::

MetricTermAND.configure(
    terms=[
        EeNearEntity.configure(robot_name="robot", ...),
        IsGrasping.configure(robot_name="robot", ...),
    ],
)
Source code in eden/managers/terms/metrics/boolean.py 
26
27
28
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._child_terms: list[MetricTerm] = []

MetricTermOR 

MetricTermOR(env: EnvBase, options: MetricTermOptions)

Bases: _BooleanMetricTermBase

Metric term that returns the element-wise maximum of its child terms.

Any child term passing is sufficient for this term to pass.

PARAMETER DESCRIPTION
terms

Child term configurations (from .configure()).

TYPE: list[MetricTermOptions]

Example

::

MetricTermOR.configure(
    terms=[
        entity_near_target.configure(entity_name="apple", ...),
        entity_near_target.configure(entity_name="orange", ...),
    ],
)
Source code in eden/managers/terms/metrics/boolean.py 
26
27
28
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._child_terms: list[MetricTerm] = []

MetricTermNOT 

MetricTermNOT(env: EnvBase, options: MetricTermOptions)

Bases: MetricTerm

Metric term that inverts a single child term (1.0 - value).

PARAMETER DESCRIPTION
term

The child term configuration (from .configure()).

TYPE: MetricTermOptions

Example

::

MetricTermNOT.configure(
    term=IsGrasping.configure(robot_name="robot", ...),
)
Source code in eden/managers/terms/metrics/boolean.py 
128
129
130
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._child_term: MetricTerm | None = None

Attributes

term class-attribute instance-attribute 
term: MetricTermOptions = None

Methods:

build 
build()
Source code in eden/managers/terms/metrics/boolean.py 
132
133
134
135
136
137
def build(self):
    if self.term is None:
        raise ValueError("MetricTermNOT requires a 'term' option")
    if not isinstance(self.term, MetricTermOptions):
        raise TypeError(f"'term' must be a MetricTermOptions (from .configure()), got {type(self.term).__name__}")
    self._child_term = build_term(METRIC_TERM_REGISTRY, self._env, self.term, label=f"{type(self).__name__}.term")
reset 
reset(envs_idx: Tensor | None = None) -> None
Source code in eden/managers/terms/metrics/boolean.py 
139
140
141
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    if self._child_term is not None:
        self._child_term.reset(envs_idx=envs_idx)
compute 
compute() -> Tensor
Source code in eden/managers/terms/metrics/boolean.py 
143
144
def compute(self) -> torch.Tensor:
    return 1.0 - self._child_term.compute().float()

Functions:

Sequences and rubrics

Sequential and rubric metric terms: sub-metrics met in order, or in any order their dependencies allow.

Attributes

Classes

SequentialMetricTerm 

SequentialMetricTerm(env: EnvBase, options: MetricTermOptions)

Bases: MetricTerm

Metric that tracks ordered sub-goals (phases).

Each phase is a predicate that must be satisfied before the next phase is evaluated. The metric value is the fraction of phases completed (completed / total). With the default success_threshold=1.0 and direction="hib", the episode is considered successful once all phases are complete.

Phase advancement is per-environment -- different parallel environments can be in different phases.

PARAMETER DESCRIPTION
phases

Ordered list of phase definitions. Each PhaseOptions must contain one of:

  • term -- a MetricTermOptions from .configure().
  • ref -- a string referencing a sibling term in MetricManagerOptions.

And optionally:

  • threshold -- (default 1.0) minimum value returned by the phase predicate that counts as "passed".
  • hold -- if True, reuse the phase's own term as hold predicate.
  • hold_term -- a separate MetricTermOptions for the hold predicate (overrides hold).
  • hold_until -- the last phase (name or index, inclusive) through which the hold stays active; default: until the task completes.
  • hold_guard -- name of another held phase; this hold is enforced only where that phase's hold is also failing.

TYPE: list[PhaseOptions]

allow_skip

If True, an env may advance through multiple phases in a single compute() call. If False (default), at most one phase transition happens per call.

TYPE: bool

Example

::

SequentialMetricTerm.configure(
    phases=[
        PhaseOptions(
            name="reach",
            term=EeNearEntity.configure(
                robot_name="robot",
                ee_link_names=["finger_l", "finger_r"],
                entity_name="apple",
                threshold=0.15,
            ),
            threshold=0.5,
        ),
        PhaseOptions(
            name="grasp",
            term=IsGrasping.configure(
                robot_name="robot",
                left_gripper_link_name="left_gripper_pad",
                right_gripper_link_name="right_gripper_pad",
                force_threshold=0.5,
            ),
            threshold=1.0,
            hold=True,
            hold_until="place",
        ),
        PhaseOptions(
            name="place",
            term=entity_near_target.configure(entity_name="apple", target_pos=[0.5, 0.0, 0.8]),
        ),
    ],
    allow_skip=False,
)
Source code in eden/managers/terms/metrics/sequence.py 
105
106
107
108
109
110
111
112
113
114
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._phase_buf = torch.zeros(env.num_envs, dtype=torch.long, device=env.device)
    self._resolved_phases: list[tuple[MetricTerm, float]] = []
    self._phase_names: list[str] = []
    self._phase_name_to_idx: dict[str, int] = {}
    self._child_terms: list[MetricTerm] = []
    self._hold_conditions: list[_HoldTuple | None] = []
    self._has_any_holds: bool = False
    self._phase_tensors: list[torch.Tensor] = []

Attributes

phases class-attribute instance-attribute 
phases: list[PhaseOptions] = []
allow_skip class-attribute instance-attribute 
allow_skip: bool = False
phase_names property 
phase_names: list[str]

Names of all phases (read-only).

Methods:

resolve_refs 
resolve_refs(manager_options) -> None

Resolve ref phases by looking up sibling MetricTermOptions.

Called by MetricManager._prepare_terms() in a second pass after all terms have been built.

Source code in eden/managers/terms/metrics/sequence.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def resolve_refs(self, manager_options) -> None:
    """Resolve ``ref`` phases by looking up sibling MetricTermOptions.

    Called by ``MetricManager._prepare_terms()`` in a second pass after
    all terms have been built.
    """
    for i, phase in enumerate(self.phases):
        if phase.ref is not None and self._resolved_phases[i][0] is None:
            sibling_options = getattr(manager_options, phase.ref, None)
            if sibling_options is None:
                raise ValueError(f"Phase {i} ref='{phase.ref}' not found in MetricManagerOptions")
            term_instance = self._instantiate_term(sibling_options, f"phases[{i}].ref ('{phase.ref}')")
            self._resolved_phases[i] = (term_instance, phase.threshold)

    # Now resolve hold conditions (requires all phases to be resolved)
    self._resolve_all_holds()
build 
build()
Source code in eden/managers/terms/metrics/sequence.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def build(self):
    for i, phase in enumerate(self.phases):
        if not isinstance(phase, PhaseOptions):
            raise TypeError(
                f"Phase {i} must be a PhaseOptions instance, got {type(phase).__name__}. "
                f"Use PhaseOptions(term=MyTerm.configure(...), ...) instead of dict."
            )

        if phase.term is not None and phase.ref is not None:
            raise ValueError(f"Phase {i}: 'term' and 'ref' are mutually exclusive")
        if phase.term is None and phase.ref is None:
            raise ValueError(f"Phase {i}: must specify either 'term' or 'ref'")

        # Derive phase name
        name = phase.name
        if not name:
            if phase.term is not None:
                name = phase.term.name
            elif phase.ref is not None:
                name = phase.ref
            else:
                name = f"phase_{i}"
        self._phase_names.append(name)
        self._phase_name_to_idx[name] = i

        # Instantiate term (or defer if ref)
        if phase.term is not None:
            term_instance = self._instantiate_term(phase.term, f"phases[{i}].term")
            self._resolved_phases.append((term_instance, phase.threshold))
        else:
            # ref — will be resolved in resolve_refs()
            self._resolved_phases.append((None, phase.threshold))

    # If no refs, resolve holds now; otherwise MetricManager will call resolve_refs()
    has_refs = any(p.ref is not None for p in self.phases)
    if not has_refs:
        self._resolve_all_holds()
get_phase_params 
get_phase_params(phase: int | str) -> dict

Return the mutable params dict for a phase (by index or name).

Only meaningful for function-based terms that have a params attribute. Modifying the returned dict changes future compute() calls.

Source code in eden/managers/terms/metrics/sequence.py 
294
295
296
297
298
299
300
301
302
def get_phase_params(self, phase: int | str) -> dict:
    """Return the mutable params dict for a phase (by index or name).

    Only meaningful for function-based terms that have a ``params`` attribute.
    Modifying the returned dict changes future ``compute()`` calls.
    """
    idx = self._resolve_phase_ref(phase)
    term_instance = self._resolved_phases[idx][0]
    return term_instance.params
get_phase_term 
get_phase_term(phase: int | str) -> MetricTerm

Return the underlying term instance for a phase (by index or name).

Source code in eden/managers/terms/metrics/sequence.py 
304
305
306
307
def get_phase_term(self, phase: int | str) -> MetricTerm:
    """Return the underlying term instance for a phase (by index or name)."""
    idx = self._resolve_phase_ref(phase)
    return self._resolved_phases[idx][0]
reset 
reset(envs_idx: Tensor | None = None) -> None
Source code in eden/managers/terms/metrics/sequence.py 
318
319
320
321
322
323
324
325
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    if envs_idx is None:
        self._phase_buf.zero_()
    else:
        self._phase_buf[envs_idx] = 0
    # Reset child terms as well
    for child in self._child_terms:
        child.reset(envs_idx=envs_idx)
compute 
compute() -> Tensor
Source code in eden/managers/terms/metrics/sequence.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def compute(self) -> torch.Tensor:
    num_phases = len(self._resolved_phases)
    if num_phases == 0:
        return torch.ones(self._env.num_envs, device=self.device)

    # ------------------------------------------------------------------
    # Hold-condition check: revert to earliest failing hold
    # ------------------------------------------------------------------
    if self._has_any_holds:
        # Track the earliest failing phase per env (num_phases = "no failure")
        revert_target = torch.full((self._env.num_envs,), num_phases, dtype=torch.long, device=self.device)

        # Pre-compute hold values so guards can reference them.
        # hold_failed[p] is a per-env bool tensor (True = hold failing).
        hold_failed: list[torch.Tensor | None] = [None] * num_phases
        for p, hold in enumerate(self._hold_conditions):
            if hold is None:
                continue
            candidates = (self._phase_buf > p) & (self._phase_buf < hold.until_idx)
            if not candidates.any():
                continue
            h_value = hold.term.compute()
            h_failed = h_value < hold.threshold
            hold_failed[p] = candidates & h_failed

            if en.logger.level <= logging.DEBUG:
                for e in range(self._env.num_envs):
                    if candidates[e]:
                        en.logger.debug(
                            f"  [env {e}] hold check phase {p} ({self._phase_names[p]}): "
                            f"value={h_value[e].item():.4f}  threshold={hold.threshold}  "
                            f"failed={bool(h_failed[e])}"
                        )

        # Apply hold failures, respecting guards.
        for p, hold in enumerate(self._hold_conditions):
            if hold is None or hold_failed[p] is None:
                continue
            failing = hold_failed[p]
            if not failing.any():
                continue

            # If this hold has a guard, suppress it for envs where the
            # guard's hold is still passing (i.e. guard not failing).
            if hold.guard_idx is not None:
                guard_failing = hold_failed[hold.guard_idx]
                if guard_failing is None:
                    # Guard hold wasn't even evaluated (no envs in window) → guard is passing → suppress all
                    if en.logger.level <= logging.DEBUG:
                        en.logger.debug(
                            f"  hold on phase {p} ({self._phase_names[p]}) fully suppressed "
                            f"by guard '{self._phase_names[hold.guard_idx]}' (guard passing)"
                        )
                    continue
                # Only enforce this hold where the guard is also failing
                failing = failing & guard_failing
                if not failing.any():
                    if en.logger.level <= logging.DEBUG:
                        en.logger.debug(
                            f"  hold on phase {p} ({self._phase_names[p]}) suppressed "
                            f"by guard '{self._phase_names[hold.guard_idx]}' (guard passing for all envs)"
                        )
                    continue

            revert_target = torch.where(failing & (p < revert_target), self._phase_tensors[p], revert_target)

        # Apply reverts
        needs_revert = revert_target < num_phases
        if needs_revert.any():
            if en.logger.level <= logging.INFO:
                for e in range(self._env.num_envs):
                    if needs_revert[e]:
                        old_p = self._phase_buf[e].item()
                        new_p = revert_target[e].item()
                        en.logger.info(
                            f"[env {e}] HOLD FAILED: reverting from phase {old_p} ({self._phase_names[old_p]}) "
                            f"→ phase {new_p} ({self._phase_names[new_p]})"
                        )
            self._phase_buf = torch.where(needs_revert, revert_target, self._phase_buf)

    # ------------------------------------------------------------------
    # Phase advancement
    # ------------------------------------------------------------------
    prev_phases = self._phase_buf.clone()

    # When allow_skip is False, use a snapshot so each env advances at
    # most one phase per compute().  When True, read _phase_buf directly
    # so envs that just advanced are re-evaluated for the next phase.
    phase_view = self._phase_buf if self.allow_skip else prev_phases

    for p, (term_instance, threshold) in enumerate(self._resolved_phases):
        mask = phase_view == p
        if not mask.any():
            continue
        value = term_instance.compute()
        passed = value >= threshold
        advance = mask & passed
        self._phase_buf[advance] = p + 1

        if en.logger.level <= logging.DEBUG:
            for e in range(self._env.num_envs):
                if mask[e]:
                    en.logger.debug(
                        f"  [env {e}] phase {p} ({self._phase_names[p]}): "
                        f"value={value[e].item():.4f}  threshold={threshold}  "
                        f"passed={bool(passed[e])}"
                    )

    # Clamp to [0, num_phases]
    self._phase_buf.clamp_(max=num_phases)

    if en.logger.level <= logging.INFO:
        for e in range(self._env.num_envs):
            if self._phase_buf[e] != prev_phases[e]:
                old_p = prev_phases[e].item()
                new_p = self._phase_buf[e].item()
                old_name = self._phase_names[old_p]
                if new_p >= num_phases:
                    en.logger.info(
                        f"[env {e}] Phase {old_p} ({old_name}) DONE → task COMPLETE ({new_p}/{num_phases})"
                    )
                else:
                    en.logger.info(
                        f"[env {e}] Phase {old_p} ({old_name}) DONE → now at phase {new_p} ({self._phase_names[new_p]})  [{new_p}/{num_phases}]"
                    )

    return self._phase_buf.float() / num_phases

RubricMetricTerm 

RubricMetricTerm(env: EnvBase, options: MetricTermOptions)

Bases: MetricTerm

Fraction of rubric criteria met, each after its dependencies.

A criterion is reached the first step its predicate succeeds — by the term's own is_success, so its direction and success_threshold decide (entity_distance with success_threshold=0.05 is met within 5 cm) — while every criterion in its after list has already been reached, and stays reached until the env resets: a cube let go over the tray counts even if it bounces off later. The value is the fraction of criteria reached, so with the default success_threshold=1.0 the task succeeds once all of them are. Unlike :class:SequentialMetricTerm, criteria with no dependency between them are reached in any order (both cubes' "reach" before either "lift", or one cube's whole chain first) — the progress score of PolaRiS's Rubric, which :mod:eden.tasks.polaris reproduces.

Criteria are evaluated in list order every step, so after names earlier criteria only, and a criterion can be reached in the same step as the ones it depends on.

PARAMETER DESCRIPTION
criteria

The criteria, each a named predicate (term) and the earlier criteria it needs (after).

TYPE: list[CriterionOptions]

Example

::

RubricMetricTerm.configure(
    criteria=[
        CriterionOptions(name="reach", term=EeNearEntity.configure(robot_name="gripper", ...)),
        CriterionOptions(name="lift", term=object_above_height.configure(...), after=["reach"]),
        CriterionOptions(name="in_bowl", term=object_a_within_b_xy.configure(...), after=["lift"]),
    ]
)
Source code in eden/managers/terms/metrics/sequence.py 
493
494
495
496
497
def __init__(self, env: EnvBase, options: MetricTermOptions):
    super().__init__(env=env, options=options)
    self._terms: list[MetricTerm] = []
    self._after: list[list[int]] = []
    self._reached = torch.zeros(env.num_envs, len(self.criteria), dtype=torch.bool, device=env.device)

Attributes

criteria class-attribute instance-attribute 
criteria: list[CriterionOptions] = []
criterion_names property 
criterion_names: list[str]

Methods:

build 
build()
Source code in eden/managers/terms/metrics/sequence.py 
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
def build(self):
    if not self.criteria:
        raise ValueError(
            "RubricMetricTerm needs at least one criterion: pass criteria=[CriterionOptions(...), ...]."
        )
    index: dict[str, int] = {}
    for i, criterion in enumerate(self.criteria):
        if not isinstance(criterion, CriterionOptions):
            raise TypeError(
                f"criteria[{i}] must be a CriterionOptions, got {type(criterion).__name__}; write "
                f"CriterionOptions(name=..., term=MyTerm.configure(...)) rather than a dict."
            )
        if criterion.name in index:
            raise ValueError(
                f"RubricMetricTerm criterion name '{criterion.name}' is used twice; names must be unique."
            )
        unknown = [name for name in criterion.after if name not in index]
        if unknown:
            raise ValueError(
                f"RubricMetricTerm criterion '{criterion.name}' lists after={unknown}, which are not earlier "
                f"criteria; earlier ones: {list(index) or 'none'}. Move the criterion below the ones it depends on."
            )
        term = build_term(
            METRIC_TERM_REGISTRY, self._env, criterion.term, label=f"{type(self).__name__}.criteria[{i}]"
        )
        if term.is_cumulative:
            raise ValueError(
                f"RubricMetricTerm criterion '{criterion.name}' has is_cumulative=True, but a criterion is "
                f"judged on each step's value, never on a sum over the episode; configure its term with "
                f"is_cumulative=False."
            )
        self._after.append([index[name] for name in criterion.after])
        self._terms.append(term)
        index[criterion.name] = i
reset 
reset(envs_idx: Tensor | None = None) -> None
Source code in eden/managers/terms/metrics/sequence.py 
538
539
540
541
542
543
544
def reset(self, envs_idx: torch.Tensor | None = None) -> None:
    if envs_idx is None:
        self._reached.zero_()
    else:
        self._reached[envs_idx] = False
    for term in self._terms:
        term.reset(envs_idx=envs_idx)
reached 
reached() -> Tensor

(num_envs, num_criteria) bool: the criteria each env has reached this episode (a copy).

Source code in eden/managers/terms/metrics/sequence.py 
546
547
548
def reached(self) -> torch.Tensor:
    """``(num_envs, num_criteria)`` bool: the criteria each env has reached this episode (a copy)."""
    return self._reached.clone()
compute 
compute() -> Tensor
Source code in eden/managers/terms/metrics/sequence.py 
550
551
552
553
554
555
556
def compute(self) -> torch.Tensor:
    for i, (term, after) in enumerate(zip(self._terms, self._after)):
        met = term.is_success(term.compute())
        if after:
            met &= self._reached[:, after].all(dim=1)
        self._reached[:, i] |= met
    return self._reached.float().mean(dim=1)

Functions:

See also