embodichain.lab.gym.envs.managers

Managers that orchestrate collections of functors (observation, reward, event, action, dataset) running at specific points in the environment step loop.

Each manager owns a typed @configclass config whose attributes are FunctorCfg instances; the config attribute name becomes the functor’s unique identifier.

Overview

Managers orchestrate collections of functors that run at specific points in the environment step loop. Each manager owns a typed @configclass config whose attributes are FunctorCfg (or subclass) instances; at init the manager resolves every func reference, validates argument signatures against params, resolves SceneEntityCfg targets to scene indices, and groups functors by mode. The config attribute name becomes the functor’s unique identifier within its manager.

The five manager types are ObservationManager (compute(obs)), RewardManager (compute(obs, action, info)), EventManager (apply(mode, env_ids), the home of all randomization functors), ActionManager (process_actions(actions)), and DatasetManager (step/save for LeRobot recording).

Submodules

randomization

Domain-randomization event functors (physics, visual, spatial, geometry).

Classes

FunctorCfg	Configuration for a functor.
SceneEntityCfg	Configuration for a scene entity that is used by the manager's functor.
EventCfg	Configuration for a event functor.
ObservationCfg	Configuration for an observation functor.
RewardCfg	Configuration for a reward functor.
ActionTermCfg	Configuration for an action term.
DatasetFunctorCfg	Configuration for dataset collection functors.
Functor	Base class for Functor.
ManagerBase	Base class for all managers.
EventManager	Manager for orchestrating operations based on different simulation events.
ObservationManager	Manager for orchestrating operations based on different simulation observations.
RewardManager	Manager for orchestrating reward computation in reinforcement learning tasks.
ActionManager	Manager for processing actions sent to the environment.
DatasetManager	Manager for orchestrating dataset collection and saving using functors.
ActionTerm	Base class for action terms.
DeltaQposTerm	Delta joint position action: current_qpos + scale * action -> qpos.
QposTerm	Absolute joint position action: scale * action -> qpos.
QposDenormalizedTerm	Normalized action in [range[0], range[1]] -> denormalize to joint limits -> qpos.
QposNormalizedTerm	Normalize action from qpos limits -> [range[0], range[1]].
EefPoseTerm	End-effector pose (6D or 7D) -> IK -> qpos.
QvelTerm	Joint velocity action: scale * action -> qvel.
QfTerm	Joint force/torque action: scale * action -> qf.
LeRobotRecorder	Functor for recording episodes in LeRobot format.
AsyncLeRobotRecorder	LeRobot recorder that saves episodes on a background thread.

Functions

observations.get_rigid_object_pose(env, obs, ...)	Get the world poses of the rigid objects in the environment.
observations.normalize_robot_joint_data(env, ...)	Normalize the robot joint positions to the range of [0, 1] based on the joint limits.
observations.compute_semantic_mask(env, obs, ...)	Compute the semantic mask for the specified scene entity.
observations.compute_exteroception	Compute the exteroception for the observation space.
events.replace_assets_from_group	Replace assets in the environment from a specified group of assets.
record.record_camera_data	Record camera data in the environment.
rewards.distance_between_objects(env, obs, ...)	Reward based on distance between two rigid objects.
rewards.success_reward(env, obs, action, info)	Sparse bonus reward when task succeeds.
rewards.distance_to_target(env, obs, action, ...)	Reward based on absolute distance to a virtual target pose.
randomization.visual.randomize_light(env, ...)	Randomize light properties by adding, scaling, or setting random values.
randomization.visual.randomize_camera_intrinsics(...)	Randomize camera intrinsic properties by adding, scaling, or setting random values.
randomization.visual.randomize_visual_material	Randomize the visual material properties of a RigidObject or Articulation.
randomization.spatial.get_random_pose(...[, ...])	Generate a random pose based on the initial position and rotation.
randomization.spatial.randomize_rigid_object_pose(...)	Randomize the pose of a rigid object in the environment.
randomization.spatial.randomize_robot_eef_pose(...)	Randomize the initial end-effector pose of a robot in the environment.
randomization.spatial.randomize_robot_qpos(...)	Randomize the initial joint positions of a robot in the environment.

Configuration Classes

class embodichain.lab.gym.envs.managers.FunctorCfg

Configuration for a functor.

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
params	The parameters to be passed to the function as keyword arguments.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.gym.envs.managers.SceneEntityCfg

Configuration for a scene entity that is used by the manager’s functor.

This class is used to specify the name of the scene entity that is queried from the SimulationManager and passed to the manager’s functor.

Attributes:

body_ids	The indices of the bodies from the asset required by the functor.
body_names	The names of the bodies from the asset required by the functor.
control_parts	The names of the control parts from the asset(only support for robot) required by the functor.
joint_ids	The indices of the joints from the asset required by the functor.
joint_names	The names of the joints from the scene entity.
link_names	The names of the links from the asset required by the functor.
preserve_order	Whether to preserve indices ordering to match with that in the specified joint, body, or object collection names.
uid	The name of the scene entity.

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
resolve(scene)	Resolves the scene entity and converts the joint and body names to indices.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

body_ids: list[int] | slice

The indices of the bodies from the asset required by the functor. Defaults to slice(None), which means all the bodies in the asset.

If body_names is specified, this is filled in automatically on initialization of the manager.

body_names: str | list[str] | None

The names of the bodies from the asset required by the functor. Defaults to None.

The names can be either body names or a regular expression matching the body names.

These are converted to body indices on initialization of the manager and passed to the functor function as a list of body indices under body_ids.

control_parts: str | list[str] | None

The names of the control parts from the asset(only support for robot) required by the functor. Defaults to None.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

joint_ids: list[int] | slice

The indices of the joints from the asset required by the functor. Defaults to slice(None), which means all the joints in the asset (if present).

If joint_names is specified, this is filled in automatically on initialization of the manager.

joint_names: str | list[str] | None

The names of the joints from the scene entity. Defaults to None.

The names can be either joint names or a regular expression matching the joint names.

These are converted to joint indices on initialization of the manager and passed to the functor as a list of joint indices under joint_ids.

link_names: str | list[str] | None

The names of the links from the asset required by the functor. Defaults to None.

The names can be either link names or a regular expression matching the link names.

preserve_order: bool

Whether to preserve indices ordering to match with that in the specified joint, body, or object collection names. Defaults to False.

If False, the ordering of the indices are sorted in ascending order (i.e. the ordering in the entity’s joints, bodies, or object in the object collection). Otherwise, the indices are preserved in the order of the specified joint, body, or object collection names.

For more details, see the isaaclab.utils.string.resolve_matching_names() function.

Note

This attribute is only used when joint_names, body_names are specified.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

resolve(scene)

Resolves the scene entity and converts the joint and body names to indices.

This function examines the scene entity from the SimulationManager and resolves the indices and names of the joints and bodies. It is an expensive operation as it resolves regular expressions and should be called only once.

Parameters:

scene (SimulationManager) – The interactive scene instance.

Raises:

• ValueError – If the scene entity is not found.

• ValueError – If both joint_names and joint_ids are specified and are not consistent.

• ValueError – If both body_names and body_ids are specified and are not consistent.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

uid: str

The name of the scene entity.

This is the name defined in the scene configuration file. See the SimulationManagerCfg class for more details.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.gym.envs.managers.EventCfg

Configuration for a event functor.

The event functor is used to trigger events in the environment at specific times or under specific conditions. The mode attribute determines when the functor is applied. - startup: The functor is applied when the environment is started. - interval: The functor is applied at each env step. - reset: The functor is applied when the environment is reset.

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
interval_step	The number of environment step after which the functor is applied.
is_global	Whether the event should be tracked on a per-environment basis.
mode	The mode in which the event functor is applied.
params	The parameters to be passed to the function as keyword arguments.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

interval_step: int

The number of environment step after which the functor is applied. Defaults to 4.

is_global: bool

Whether the event should be tracked on a per-environment basis. Defaults to False.

If True, the same interval step is used for all the environment instances. If False, the interval step is sampled independently for each environment instance and the functor is applied when the current step hits the interval step for that instance.

Note

This is only used if the mode is "interval".

mode: Literal['startup', 'interval', 'reset']

The mode in which the event functor is applied.

Note

The mode name "interval" is a special mode that is handled by the manager Hence, its name is reserved and cannot be used for other modes.

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.gym.envs.managers.ObservationCfg

Configuration for an observation functor.

The observation functor is used to compute observations for the environment. The mode attribute determines whether the observation is already present in the observation space or not.

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
mode	The mode for the observation computation.
name	The name of the observation.
params	The parameters to be passed to the function as keyword arguments.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

mode: Literal['modify', 'add']

The mode for the observation computation.

• modify: The observation is already present in the observation space, updated the value in-place.

• add: The observation is not present in the observation space, add a new entry to the observation space.

name: str

The name of the observation.

The name can be a new key to observation space, eg:

• object_position: shape of (num_envs, 3)

• robot/eef_pose: shape of (num_envs, 7) or (num_envs, 4, 4)

• sensor/cam_high/mask: shape of (num_envs, H, W)

or a existing key to modify, eg:

• robot/qpos: shape of (num_envs, num_dofs)

/ is used to separate different levels of hierarchy in the observation dictionary.

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.gym.envs.managers.RewardCfg

Configuration for a reward functor.

The reward functor is used to compute rewards for the environment. The mode attribute determines how the reward is combined with existing rewards.

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
mode	The mode for the reward computation.
params	The parameters to be passed to the function as keyword arguments.
weight	The weight multiplier for this reward term.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

mode: Literal['add', 'replace']

The mode for the reward computation.

• add: The reward is added to the existing total reward.

• replace: The reward replaces the total reward (useful for single reward functions).

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

weight: float

The weight multiplier for this reward term.

This value is used to scale the reward before adding it to the total reward. Default is 1.0 (no scaling).

class embodichain.lab.gym.envs.managers.ActionTermCfg

Configuration for an action term.

The action term is used to preprocess raw actions from the policy into the format expected by the robot (e.g., qpos, qvel, qf).

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
mode	The mode for the action term.
params	The parameters to be passed to the function as keyword arguments.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

mode: Literal['pre', 'post']

The mode for the action term.

• pre: Preprocess raw action from policy (default). This is applied before the action is sent to the robot control.

• post: Postprocess the action after it has been processed by another term. This is useful for applying additional transformations like noise, clipping, or filtering to the output actions.

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.gym.envs.managers.DatasetFunctorCfg

Configuration for dataset collection functors.

Dataset functors are called with mode=”save” which handles both: - Recording observation-action pairs on every step - Auto-saving episodes when dones=True

Methods:

copy(**kwargs)	Return a new object replacing specified fields with new values.
replace(**kwargs)	Return a new object replacing specified fields with new values.
to_dict()	Convert an object into dictionary recursively.
validate([prefix])	Check the validity of configclass object.

Attributes:

extra	Extra metadata about the functor.
func	The function or class to be called for the functor.
params	The parameters to be passed to the function as keyword arguments.
save_failed_episodes	Whether to save failed episodes.

copy(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

extra: dict[str, Any]

Extra metadata about the functor. Defaults to an empty dict.

This can be used to store additional configuration information such as the output shape of observation functors, which can be used for pre-allocating buffers.

For observation functors, common keys include:

• shape: A tuple defining the output shape of the functor (excluding num_envs dimension).

func: Callable | Functor

The function or class to be called for the functor.

The function must take the environment object as the first argument. The remaining arguments are specified in the params attribute.

It also supports callable classes, i.e. classes that implement the __call__() method. In this case, the class should inherit from the Functor class and implement the required methods.

params: dict[str, Any | SceneEntityCfg]

The parameters to be passed to the function as keyword arguments. Defaults to an empty dict.

Note

If the value is a SceneEntityCfg object, the manager will query the scene entity from the SimulationManager and process the entity’s joints and bodies as specified in the SceneEntityCfg object.

replace(**kwargs)

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

• obj (object) – The object to replace.

• **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

save_failed_episodes: bool

Whether to save failed episodes.

If enabled for any dataset functor, all save-mode dataset functors receive both successful and failed episodes. During run-env expert generation, a non-empty failed or truncated result is committed and counts toward max_episodes instead of being discarded and retried. Empty plans and execution exceptions remain uncommitted.

to_dict()

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

• obj (object) – The object to check.

• prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

Base Classes

class embodichain.lab.gym.envs.managers.Functor

Bases: ABC

Base class for Functor.

Functor implementations can be functions or classes. If the functor is a class, it should inherit from this base class and implement the required methods.

Each manager is implemented as a class that inherits from the ManagerBase class. Each manager class should also have a corresponding configuration class that defines the configuration functors for the manager. Each functor should the FunctorCfg class or its subclass.

Example pseudo-code for creating a manager:

from embodichain.utils import configclass
from embodichain.lab.gym.managers import ManagerBase
from embodichain.lab.gym.managers FunctorCfg

@configclass
class MyManagerCfg:

    functor1: FunctorCfg = FunctorCfg(...)
    functor2: FunctorCfg = FunctorCfg(...)
    functor3: FunctorCfg = FunctorCfg(...)

# define manager instance
my_manager = ManagerBase(cfg=ManagerCfg(), env=env)

Methods:

__init__(cfg, env)	Initialize the functor.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

Attributes:

device	Device on which to perform computations.
num_envs	Number of environments.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property device: str

Device on which to perform computations.

property num_envs: int

Number of environments.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.ManagerBase

Bases: ABC

Base class for all managers.

Methods:

__init__(cfg, env)	Initialize the manager.
find_functors(name_keys)	Find functors in the manager based on the names.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
reset([env_ids])	Resets the manager and returns logging information for the current time-step.

Attributes:

active_functors	Name of active functors.
device	Device on which to perform computations.
num_envs	Number of environments.

__init__(cfg, env)

Initialize the manager.

This function is responsible for parsing the configuration object and creating the functors.

If the simulation is not playing, the scene entities are not resolved immediately. Instead, the resolution is deferred until the simulation starts. This is done to ensure that the scene entities are resolved even if the manager is created after the simulation has already started.

Parameters:

• cfg (object) – The configuration object. If None, the manager is initialized without any functors.

• env (EmbodiedEnv) – The environment instance.

abstract property active_functors: list[str] | dict[str, list[str]]

Name of active functors.

property device: str

Device on which to perform computations.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

property num_envs: int

Number of environments.

reset(env_ids=None)

Resets the manager and returns logging information for the current time-step.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids for which to log data. Defaults None, which logs data for all environments.

Return type:

dict[str, float]

Returns:

Dictionary containing the logging information.

Managers

class embodichain.lab.gym.envs.managers.EventManager

Bases: ManagerBase

Manager for orchestrating operations based on different simulation events.

The event manager applies operations to the environment based on different simulation events. For example, changing the masses of objects or their friction coefficients during initialization/ reset, or applying random pushes to the robot at a fixed interval of steps. The user can specify several modes of events to fine-tune the behavior based on when to apply the event.

The event functors are parsed from a config class containing the manager’s settings and each functor’s parameters. Each event functor should instantiate the EventCfg class.

Event functors can be grouped by their mode. The mode is a user-defined string that specifies when the event functor should be applied. This provides the user complete control over when event functors should be applied.

For a typical training process, you may want to apply events in the following modes:

• “prestartup”: Event is applied once at the beginning of the training before the simulation starts. This is used to randomize USD-level properties of the simulation stage.

• “startup”: Event is applied once at the beginning of the training once simulation is started.

• “reset”: Event is applied at every reset.

• “interval”: Event is applied at pre-specified intervals of time.

However, you can also define your own modes and use them in the training process as you see fit. For this you will need to add the triggering of that mode in the environment implementation as well.

Note

The triggering of operations corresponding to the mode "interval" are the only mode that are directly handled by the manager itself. The other modes are handled by the environment implementation.

Methods:

__init__(cfg, env)	Initialize the event manager.
apply(mode[, env_ids])	Calls each event functor in the specified mode.
find_functors(name_keys)	Find functors in the manager based on the names.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
get_functor(functor_name)	Retrieve a functor from the configuration by its name.
get_functor_cfg(functor_name)	Gets the configuration for the specified functor.
reset([env_ids])	Resets the manager and returns logging information for the current time-step.
set_functor_cfg(functor_name, cfg)	Sets the configuration of the specified functor into the manager.
set_seed(seed)	Set the event seed and rewind all deterministic event streams.

Attributes:

active_functors	Name of active event functors.
available_modes	Modes of events.
device	Device on which to perform computations.
num_envs	Number of environments.
seed	Return the base seed used for event-functor randomization.

__init__(cfg, env)

Initialize the event manager.

Parameters:

• cfg (object) – A configuration object or dictionary (dict[str, EventCfg]).

• env (EmbodiedEnv) – An environment object.

property active_functors: dict[str, list[str]]

Name of active event functors.

The keys are the modes of event and the values are the names of the event functors.

apply(mode, env_ids=None)

Calls each event functor in the specified mode.

This function iterates over all the event functors in the specified mode and calls the function corresponding to the functor. The function is called with the environment instance and the environment indices to apply the event to.

For the “interval” mode, the function is called when the time interval has passed. This requires specifying the time step of the environment.

For the “reset” mode, the function is called when the mode is “reset” and the total number of environment steps that have happened since the last trigger of the function is equal to its configured parameter for the number of environment steps between resets.

Parameters:

• mode (str) – The mode of event.

• env_ids (Optional[Sequence[int]]) – The indices of the environments to apply the event to. Defaults to None, in which case the event is applied to all environments when applicable.

Raises:

• ValueError – If the mode is "interval" and the environment indices are provided. This is an undefined behavior as the environment indices are computed based on the time left for each environment.

• ValueError – If the mode is "reset" and the total number of environment steps that have happened is not provided.

property available_modes: list[str]

Modes of events.

property device: str

Device on which to perform computations.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

get_functor(functor_name)

Retrieve a functor from the configuration by its name.

Parameters:

functor_name (str) – The name of the functor to retrieve.

Returns:

The functor if it exists in the configuration, otherwise None.

get_functor_cfg(functor_name)

Gets the configuration for the specified functor.

The method finds the functor by name by searching through all the modes. It then returns the configuration of the functor with the first matching name.

Parameters:

functor_name (str) – The name of the event functor.

Return type:

EventCfg

Returns:

The configuration of the event functor.

Raises:

ValueError – If the functor name is not found.

property num_envs: int

Number of environments.

reset(env_ids=None)

Resets the manager and returns logging information for the current time-step.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids for which to log data. Defaults None, which logs data for all environments.

Return type:

dict[str, float]

Returns:

Dictionary containing the logging information.

property seed: int | None

Return the base seed used for event-functor randomization.

set_functor_cfg(functor_name, cfg)

Sets the configuration of the specified functor into the manager.

The method finds the functor by name by searching through all the modes. It then updates the configuration of the functor with the first matching name.

Parameters:

• functor_name (str) – The name of the event functor.

• cfg (EventCfg) – The configuration for the event functor.

Raises:

ValueError – If the functor name is not found.

set_seed(seed)

Set the event seed and rewind all deterministic event streams.

Parameters:

seed (int | None) – Base task-environment seed. None disables scoped event randomization and preserves the process-global RNG behavior.

Raises:

TypeError – If seed is not an integer or None.

Return type:

None

class embodichain.lab.gym.envs.managers.ObservationManager

Bases: ManagerBase

Manager for orchestrating operations based on different simulation observations.

The default observation space will contain two observation groups:

• robot: Contains the default observations related to the robot.

  • qpos: The joint positions of the robot.

  • qvel: The joint velocities of the robot.

  • qf: The joint forces of the robot.

• sensor: Contains the observations related to the sensors which are enabled in the environment.

The observation manager offers two modes of operation:

• modify: This mode perform data fetching and modification on existing observation data.

• add: This mode perform new observation computation and add new observation data to the observation space.

Methods:

__init__(cfg, env)	Initialize the observation manager.
compute(obs)	Calls each observation functor in the specified mode.
find_functors(name_keys)	Find functors in the manager based on the names.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
get_functor_cfg(functor_name)	Gets the configuration for the specified functor.
reset([env_ids])	Resets the manager and returns logging information for the current time-step.

Attributes:

active_functors	Name of active observation functors.
device	Device on which to perform computations.
num_envs	Number of environments.

__init__(cfg, env)

Initialize the observation manager.

Parameters:

• cfg (object) – A configuration object or dictionary (dict[str, ObservationCfg]).

• env (EmbodiedEnv) – An environment object.

property active_functors: dict[str, list[str]]

Name of active observation functors.

The keys are the modes of observation and the values are the names of the observation functors.

compute(obs)

Calls each observation functor in the specified mode.

This function iterates over all the observation functors in the specified mode and calls the function corresponding to the functor. The function is called with the environment instance and the environment indices to apply the observation to.

Parameters:

obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation data to apply the observation to.

Return type:

TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]

Returns:

The modified observation data.

Raises:

ValueError – If the mode is not supported.

property device: str

Device on which to perform computations.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

get_functor_cfg(functor_name)

Gets the configuration for the specified functor.

The method finds the functor by name by searching through all the modes. It then returns the configuration of the functor with the first matching name.

Parameters:

functor_name (str) – The name of the observation functor.

Return type:

ObservationCfg

Returns:

The configuration of the observation functor.

Raises:

ValueError – If the functor name is not found.

property num_envs: int

Number of environments.

reset(env_ids=None)

Resets the manager and returns logging information for the current time-step.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids for which to log data. Defaults None, which logs data for all environments.

Return type:

dict[str, float]

Returns:

Dictionary containing the logging information.

class embodichain.lab.gym.envs.managers.RewardManager

Bases: ManagerBase

Manager for orchestrating reward computation in reinforcement learning tasks.

The reward manager computes rewards based on the current state of the environment and actions. It supports multiple reward terms that can be combined through weighted summation.

The reward manager offers two modes of operation:

• add: This mode computes a reward term and adds it to the total reward (weighted by the term’s weight).

• replace: This mode replaces the total reward with the computed value (useful for single reward functions).

Note: The config key is used as the unique identifier and display name for each reward functor.

Methods:

__init__(cfg, env)	Initialize the reward manager.
compute(obs, action, info)	Compute the total reward by calling each reward functor.
find_functors(name_keys)	Find functors in the manager based on the names.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
get_functor_cfg(functor_name)	Gets the configuration for the specified functor.
reset([env_ids])	Reset reward terms that are stateful (implemented as classes).

Attributes:

active_functors	Name of active reward functors.
device	Device on which to perform computations.
num_envs	Number of environments.

__init__(cfg, env)

Initialize the reward manager.

Parameters:

• cfg (object) – A configuration object or dictionary (dict[str, RewardCfg]).

• env (EmbodiedEnv) – An environment object.

property active_functors: dict[str, list[str]]

Name of active reward functors.

The keys are the modes of reward computation and the values are the names of the reward functors.

compute(obs, action, info)

Compute the total reward by calling each reward functor.

This function iterates over all the reward functors and calls them to compute individual reward terms. The terms are then combined according to their mode and weight.

Parameters:

• obs (EnvObs) – The observation from the environment.

• action (EnvAction) – The action applied to the robot.

• info (dict) – Additional information dictionary.

Returns:

• total_reward: The total reward for each environment (shape: [num_envs]).

• reward_info: A dictionary mapping reward term names to their values for logging.

Return type:

A tuple containing

Raises:

ValueError – If the mode is not supported.

property device: str

Device on which to perform computations.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

get_functor_cfg(functor_name)

Gets the configuration for the specified functor.

The method finds the functor by name by searching through all the modes. It then returns the configuration of the functor with the first matching name.

Parameters:

functor_name (str) – The name of the reward functor.

Return type:

RewardCfg

Returns:

The configuration of the reward functor.

Raises:

ValueError – If the functor name is not found.

property num_envs: int

Number of environments.

reset(env_ids=None)

Reset reward terms that are stateful (implemented as classes).

Parameters:

env_ids (Optional[Sequence[int]]) – The environment indices to reset. If None, all environments are reset.

Return type:

dict[str, float]

Returns:

An empty dictionary (no logging needed for reset).

class embodichain.lab.gym.envs.managers.ActionManager

Bases: ManagerBase

Manager for processing actions sent to the environment.

The action manager handles the interpretation and preprocessing of raw actions from the policy into the format expected by the robot. It supports a single active action term per environment (matching current RL usage).

Methods:

__init__(cfg, env)	Initialize the action manager.
convert_policy_action_to_env_action(action)	Convert raw action from policy into robot control format.
find_functors(name_keys)	Find functors in the manager based on the names.
get_action_dim_by_mode(mode)	Get total action dimension for terms of a specific mode.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
get_term(name)	Get action term by name.
get_terms_by_mode(mode)	Get action terms filtered by mode.
process_action(action[, mode])	Process raw action from policy into robot control format.
reset([env_ids])	Resets the manager and returns logging information for the current time-step.

Attributes:

active_functors	Name of active action terms.
device	Device on which to perform computations.
num_envs	Number of environments.
total_action_dim	Total dimension of actions (sum of all term dimensions).

__init__(cfg, env)

Initialize the action manager.

Parameters:

• cfg (object) – A configuration object or dictionary (dict[str, ActionTermCfg]).

• env (EmbodiedEnv) – The environment instance.

property active_functors: list[str]

Name of active action terms.

convert_policy_action_to_env_action(action)

Convert raw action from policy into robot control format.

This is a convenience method for processing a raw action tensor through the active terms. It assumes the input action is ordered according to the active terms and concatenated into a single tensor.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, total_action_dim).

Return type:

Union[Tensor, TensorDict[str, Tensor]]

Returns:

Processed action tensor ready for robot control, shape depends on active terms.

property device: str

Device on which to perform computations.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_action_dim_by_mode(mode)

Get total action dimension for terms of a specific mode.

Parameters:

mode (Literal['pre', 'post']) – The mode to filter by (“pre” or “post”).

Return type:

int

Returns:

Sum of action dimensions for terms with the specified mode.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

get_term(name)

Get action term by name.

Return type:

ActionTerm

get_terms_by_mode(mode)

Get action terms filtered by mode.

Parameters:

mode (Literal['pre', 'post']) – The mode to filter by (“pre” or “post”).

Return type:

list[tuple[str, ActionTerm]]

Returns:

List of (name, term) tuples for terms with the specified mode.

property num_envs: int

Number of environments.

process_action(action, mode='pre')

Process raw action from policy into robot control format.

Supports: 1. Tensor input: Passed to the active (first) term of the specified mode. 2. Dict/TensorDict input: Uses key matching term name; raises an error if no match.

Parameters:

• action (Union[Tensor, TensorDict[str, Tensor]]) – Raw action from policy (tensor or dict).

• mode (Literal['pre', 'post']) – The processing mode - “pre” for preprocessing (default) or “post” for postprocessing. When “post”, only terms with mode=”post” are applied.

Return type:

Union[Tensor, TensorDict[str, Tensor]]

Returns:

TensorDict action ready for robot control.

reset(env_ids=None)

Resets the manager and returns logging information for the current time-step.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids for which to log data. Defaults None, which logs data for all environments.

Return type:

dict[str, float]

Returns:

Dictionary containing the logging information.

property total_action_dim: int

Total dimension of actions (sum of all term dimensions).

class embodichain.lab.gym.envs.managers.DatasetManager

Bases: ManagerBase

Manager for orchestrating dataset collection and saving using functors.

The dataset manager supports multiple dataset formats through a functor system: - LeRobot format (via LeRobotRecorder) - HDF5 format (via HDF5Recorder) - Zarr format (via ZarrRecorder) - Custom formats (via user-defined functors)

Each functor’s step() method is called once per environment step and handles: - Recording observation-action pairs - Detecting episode completion (dones=True) - Auto-saving completed episodes

Example configuration:

>>> from embodichain.lab.gym.envs.managers.cfg import DatasetFunctorCfg
>>> from embodichain.lab.gym.envs.managers.datasets import LeRobotRecorder
>>>
>>> @configclass
>>> class MyEnvCfg:
>>>     dataset: dict = {
>>>         "lerobot": DatasetFunctorCfg(
>>>             func=LeRobotRecorder,
>>>             save_failed_episodes=True,
>>>             params={
>>>                 "robot_meta": {...},
>>>                 "instruction": {"lang": "pick and place"},
>>>                 "extra": {"scene_type": "kitchen"},
>>>                 "save_path": "/data/datasets"
>>>             }
>>>         )
>>>     }

Methods:

__init__(cfg, env)	Initialize the dataset manager.
apply(mode[, env_ids])	Apply dataset functors for the specified mode.
clear_cache()	Clear cached data from all dataset functors (for online training).
close()	Finalize all dataset functors; repeated calls are safe.
finalize()	Finalize every dataset functor exactly once.
find_functors(name_keys)	Find functors in the manager based on the names.
get_active_iterable_functors(env_idx)	Returns the active functors as iterable sequence of tuples.
get_cached_data()	Get cached data from all dataset functors (for online training).
get_functor_cfg(functor_name)	Gets the configuration for the specified functor.
reset([env_ids])	Reset all dataset functors.

Attributes:

active_functors	Name of active dataset functors by mode.
available_modes	List of available modes for the dataset manager.
device	Device on which to perform computations.
num_envs	Number of environments.
save_failed_episodes	Whether any configured dataset recorder should keep failed episodes.

__init__(cfg, env)

Initialize the dataset manager.

Parameters:

• cfg (object) – Configuration object containing dataset functor configurations.

• env (EmbodiedEnv) – The environment instance.

property active_functors: dict[str, list[str]]

Name of active dataset functors by mode.

The keys are the modes and the values are the names of the dataset functors.

apply(mode, env_ids=None)

Apply dataset functors for the specified mode.

This method saves completed episodes by reading data from the environment’s episode buffers. It should be called before clearing the buffers during reset.

Parameters:

• mode (str) – The mode to apply (currently only “save” is supported).

• env_ids (Union[Sequence[int], Tensor, None]) – The indices of the environments to apply the functor to. Defaults to None, in which case the functor is applied to all environments.

Return type:

None

property available_modes: list[str]

List of available modes for the dataset manager.

clear_cache()

Clear cached data from all dataset functors (for online training).

Iterates through all functors and clears their cache if they support online training mode (have clear_cache method).

Return type:

int

Returns:

Total number of cached items cleared across all functors.

close()

Finalize all dataset functors; repeated calls are safe.

Return type:

Optional[str]

property device: str

Device on which to perform computations.

finalize()

Finalize every dataset functor exactly once.

Finalization is a storage barrier only; individual recorders must not implicitly commit live episode buffers here. All functors are attempted even when one fails, and their failures are reported together.

Return type:

Optional[str]

Returns:

Path to the first finalized dataset, or None if none was returned.

Raises:

RuntimeError – If one or more functors fail to finalize.

find_functors(name_keys)

Find functors in the manager based on the names.

This function searches the manager for functors based on the names. The names can be specified as regular expressions or a list of regular expressions. The search is performed on the active functors in the manager.

Please check the resolve_matching_names() function for more information on the name matching.

Parameters:

name_keys (Union[str, Sequence[str]]) – A regular expression or a list of regular expressions to match the functor names.

Return type:

list[str]

Returns:

A list of functor names that match the input keys.

get_active_iterable_functors(env_idx)

Returns the active functors as iterable sequence of tuples.

The first element of the tuple is the name of the functor and the second element is the raw value(s) of the functor.

Return type:

Sequence[tuple[str, Sequence[float]]]

Returns:

The active functors.

get_cached_data()

Get cached data from all dataset functors (for online training).

Iterates through all functors and collects cached data from those that support online training mode (have get_cached_data method).

Return type:

list[Dict[str, Any]]

Returns:

List of cached data dictionaries from all functors.

get_functor_cfg(functor_name)

Gets the configuration for the specified functor.

Parameters:

functor_name (str) – The name of the dataset functor.

Return type:

DatasetFunctorCfg

Returns:

The configuration of the dataset functor.

Raises:

ValueError – If the functor name is not found.

property num_envs: int

Number of environments.

reset(env_ids=None)

Reset all dataset functors.

Parameters:

env_ids (Union[Sequence[int], Tensor, None]) – The environment ids. Defaults to None.

Return type:

dict[str, float]

Returns:

Empty dict (no logging info).

property save_failed_episodes: bool

Whether any configured dataset recorder should keep failed episodes.

Action Terms

class embodichain.lab.gym.envs.managers.ActionTerm

Bases: Functor

Base class for action terms.

The action term is responsible for processing the raw actions sent to the environment and converting them to the format expected by the robot (e.g., qpos, qvel, qf).

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

abstract property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

abstract property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

abstract process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Union[Tensor, TensorDict[str, Tensor]]

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.DeltaQposTerm

Bases: ActionTerm

Delta joint position action: current_qpos + scale * action -> qpos.

This action term adds a scaled delta to the current joint positions. Useful for relative position control where the policy outputs position offsets.

Parameters:

scale – Scaling factor for the action. Defaults to 1.0.

Example

>>> cfg = ActionTermCfg(func=DeltaQposTerm, params={"scale": 0.1})
>>> term = DeltaQposTerm(cfg, env)
>>> action = torch.ones(num_envs, dof) * 2.0
>>> result = term.process_action(action)
>>> # result["qpos"] = current_qpos + 0.1 * action

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.QposTerm

Bases: ActionTerm

Absolute joint position action: scale * action -> qpos.

This action term directly uses the scaled action as target joint positions. Useful for absolute position control.

Parameters:

scale – Scaling factor for the action. Defaults to 1.0.

Example

>>> cfg = ActionTermCfg(func=QposTerm, params={"scale": 1.0})
>>> term = QposTerm(cfg, env)
>>> action = torch.ones(num_envs, dof) * 0.5
>>> result = term.process_action(action)
>>> # result["qpos"] = 0.5 * action

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.QposDenormalizedTerm

Bases: ActionTerm

Normalized action in [range[0], range[1]] -> denormalize to joint limits -> qpos.

The policy outputs normalized actions in the range [range[0], range[1]] which are then mapped to the joint’s position limits.

The policy output is scaled by params.scale before denormalization. With scale=1.0 (default), action in [range[0], range[1]] maps to [low, high]. With scale<1.0, the effective range shrinks toward the center (e.g. scale=0.5 maps to 25%-75% of joint range). Use scale=1.0 for standard normalized control.

Parameters:

• scale – Scaling factor applied before denormalization. Defaults to 1.0.

• joint_ids – List of joint IDs to apply the action to. Defaults to all active joints.

• range – The range of the normalized action. Defaults to [-1.0, 1.0].

Example

>>> cfg = ActionTermCfg(func=QposDenormalizedTerm, params={"scale": 1.0})
>>> term = QposDenormalizedTerm(cfg, env)
>>> action = torch.tensor([[-1.0, 1.0], [0.0, 0.0]])  # min/max per joint
>>> result = term.process_action(action)
>>> # Maps [-1, 1] to [qpos_limits_low, qpos_limits_high]

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.QposNormalizedTerm

Bases: ActionTerm

Normalize action from qpos limits -> [range[0], range[1]].

Map joint positions to a normalized range [range[0], range[1]] based on the joint limits. This is the usually used for post processing the output of action.

Parameters:

• joint_ids – List of joint IDs to apply the action to. Defaults to all active joints.

• range – The range of the normalized action. Defaults to [0.0, 1.0].

Example

>>> cfg = ActionTermCfg(func=QposNormalizedTerm)
>>> term = QposNormalizedTerm(cfg, env)
>>> action = torch.tensor([[-1.0, 1.0], [0.0, 0.0]])  # min/max per joint
>>> result = term.process_action(action)
>>> # Maps [-1, 1] to [0, 1] based on joint limits

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.EefPoseTerm

Bases: ActionTerm

End-effector pose (6D or 7D) -> IK -> qpos.

The policy outputs a target end-effector pose which is converted to joint positions using inverse kinematics.

Supports two pose representations: - 6D: position (3) + Euler angles (3) - 7D: position (3) + quaternion (4)

On IK failure, falls back to current_qpos for that env. Returns ik_success in the TensorDict so reward/observation can penalize or condition on IK failures.

Parameters:

• scale – Scaling factor for the pose. Defaults to 1.0.

• pose_dim – Dimension of the pose (6 for Euler, 7 for quaternion). Defaults to 7.

Example

>>> cfg = ActionTermCfg(func=EefPoseTerm, params={"scale": 1.0, "pose_dim": 7})
>>> term = EefPoseTerm(cfg, env)
>>> # 7D: position (3) + quaternion (4)
>>> action = torch.zeros(num_envs, 7)
>>> action[:, :3] = 0.1  # target position
>>> action[:, 3] = 1.0   # quaternion w
>>> result = term.process_action(action)
>>> # result["qpos"] = IK solution
>>> # result["ik_success"] = bool tensor indicating IK success

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Union[Tensor, TensorDict[str, Tensor]]

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.QvelTerm

Bases: ActionTerm

Joint velocity action: scale * action -> qvel.

This action term outputs target joint velocities. Useful for velocity control tasks.

Parameters:

scale – Scaling factor for the action. Defaults to 1.0.

Example

>>> cfg = ActionTermCfg(func=QvelTerm, params={"scale": 0.2})
>>> term = QvelTerm(cfg, env)
>>> action = torch.ones(num_envs, dof)
>>> result = term.process_action(action)
>>> # result["qvel"] = 0.2 * action

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

class embodichain.lab.gym.envs.managers.QfTerm

Bases: ActionTerm

Joint force/torque action: scale * action -> qf.

This action term outputs target joint forces/torques. Useful for impedance control or force-based tasks.

Parameters:

scale – Scaling factor for the action. Defaults to 1.0.

Example

>>> cfg = ActionTermCfg(func=QfTerm, params={"scale": 10.0})
>>> term = QfTerm(cfg, env)
>>> action = torch.ones(num_envs, dof)
>>> result = term.process_action(action)
>>> # result["qf"] = 10.0 * action

Attributes:

SUPPORTED_TYPES	The supported action types.
action_dim	Dimension of the action term (policy output dimension).
device	Device on which to perform computations.
input_key	The output type of the action term, which determines how the processed action is applied to the robot.
num_envs	Number of environments.

Methods:

__init__(cfg, env)	Initialize the action term.
process_action(action)	Process raw action from policy into robot control format.
reset([env_ids])	Resets the functor.
serialize()	General serialization call.

SUPPORTED_TYPES = ['qpos', 'qvel', 'qf', 'eef_pose']

The supported action types. Each term must specify one of these as its output type, which determines how the processed action is applied to the robot.

__init__(cfg, env)

Initialize the action term.

Parameters:

• cfg (ActionTermCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

property action_dim: int

Dimension of the action term (policy output dimension).

property device: str

Device on which to perform computations.

property input_key: str

The output type of the action term, which determines how the processed action is applied to the robot.

Must be one of the supported types defined in SUPPORTED_TYPES.

property num_envs: int

Number of environments.

process_action(action)

Process raw action from policy into robot control format.

Parameters:

action (Tensor) – Raw action tensor from policy, shape (num_envs, action_dim).

Return type:

Tensor

Returns:

Processed action tensor ready for robot control, shape depends on input_key.

reset(env_ids=None)

Resets the functor.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, in which case all environments are considered.

Return type:

None

serialize()

General serialization call. Includes the configuration dict.

Return type:

dict

Observation Functions

Classes:

compute_exteroception	Compute the exteroception for the observation space.
get_articulation_joint_drive	Get the joint drive properties of the articulation in the environment with caching.
get_rigid_object_physics_attributes	Get the physics attributes of the rigid object in the environment with caching.

Functions:

compute_semantic_mask(env, obs, entity_cfg, ...)	Compute the semantic mask for the specified scene entity.
get_object_body_scale(env, obs, entity_cfg)	Get the body scale of the objects in the environment.
get_object_pose(env, obs, entity_cfg[, ...])	Get the arena poses of the objects in the environment.
get_object_uid(env, obs, entity_cfg)	Get the user IDs of the objects in the environment.
get_rigid_object_pose(env, obs, entity_cfg)	Get the world poses of the rigid objects in the environment.
get_rigid_object_velocity(env, obs, entity_cfg)	Get the world velocities of the rigid objects in the environment.
get_robot_eef_pose(env, obs[, part_name, ...])	Get robot end-effector pose using forward kinematics.
get_sensor_intrinsics(env, obs, entity_cfg)	Get the intrinsic matrix of a sensor (camera).
get_sensor_pose_in_robot_frame(env, obs, ...)	Get the pose of a sensor in the robot's base coordinate frame.
normalize_robot_joint_data(env, data, joint_ids)	Normalize the robot joint positions to the range of [0, 1] based on the joint limits.
target_position(env, obs[, target_pose_key])	Get virtual target position from env state.

class embodichain.lab.gym.envs.managers.observations.compute_exteroception

Compute the exteroception for the observation space.

The exteroception is currently defined as a set of keypoints around a reference pose, which are prjected from 3D space to 2D image plane. The reference pose can derive from the following sources:

• Pose from robot control part (e.g., end-effector, usually tcp pose)

• Object affordance pose (e.g., handle pose of a mug or a pick pose of a cube)

Therefore, the exteroception are defined in the camera-like sensor, for example. descriptor = {

“cam_high”: [

{

“type”: “affordance”, “obj_uid”: “obj1”, “key”: “grasp_pose”, “is_arena_coord”: True

}, {

“type”: “affordance”, “obj_uid”: “obj1”, “key”: “place_pose”,

}, {

“type”: “robot”, “control_part”: “left_arm”,

}, {

“type”: “robot”, “control_part”: “right_arm”,

}

}

Explanation of the parameters:

• The key of the dictionary is the sensor uid.

• The value is another dictionary, where the key is the source type, and the value is a dictionary of parameters.

• For affordance source type, the parameters are:

  • obj_uid: The uid of the object to get the affordance pose from.

  • key: The key of the affordance pose in the affordance data.

  • is_arena_coord: Whether the affordance pose is in the arena coordinate system. Default is False.

• For robot source type, the parameters are:

  • control_part: The control part of the robot to get the pose from.

Methods:

__init__(cfg, env)	Initialize the functor.
expand_pose(pose, x_interval, y_interval, ...)	Expand pose with keypoints along x and y axes.
shift_pose(pose, axis, shift)	Shift the pose along the specified axis by the given amount.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

static expand_pose(pose, x_interval, y_interval, kpnts_number, ref_pose=None)

Expand pose with keypoints along x and y axes.

Parameters:

• pose (Tensor) – The original pose tensor of shape (B, 4, 4).

• x_interval (float) – The interval for expanding along x-axis.

• y_interval (float) – The interval for expanding along y-axis.

• kpnts_number (int) – Number of keypoints to generate for each axis.

• ref_pose (Tensor) – Reference pose tensor of shape (B, 4, 4). If None, uses identity matrix.

Return type:

Tensor

Returns:

Expanded poses tensor of shape (B, 1 + 2*kpnts_number, 4, 4).

static shift_pose(pose, axis, shift)

Shift the pose along the specified axis by the given amount.

Parameters:

• pose (Tensor) – The original pose tensor of shape (B, 4, 4).

• axis (int) – The axis along which to shift (0 for x, 1 for y, 2 for z).

• shift (float) – The amount to shift along the specified axis.

Return type:

Tensor

embodichain.lab.gym.envs.managers.observations.compute_semantic_mask(env, obs, entity_cfg, foreground_uids, is_right=False)

Compute the semantic mask for the specified scene entity.

Note

The semantic mask is defined as (B, H, W, len(SemanticMask)) where these channels represents: - background channel: the instance id of the background is set to 1 (0 if not background) - foreground channel: the instance id of the foreground objects is set to 1 (0 if not foreground) - robot left-side channel: the instance id of the robot left-side is set to 1 - robot right-side channel: the instance id of the robot right-side is set to 1

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

• foreground_uids (Sequence[str]) – The list of uids for the foreground objects.

• is_right (bool) – Whether to use the right camera for stereo cameras. Default is False. Only applicable if the sensor is a StereoCamera.

Return type:

Tensor

Returns:

A tensor of shape (num_envs, height, width) representing the semantic mask.

class embodichain.lab.gym.envs.managers.observations.get_articulation_joint_drive

Get the joint drive properties of the articulation in the environment with caching.

This functor retrieves and caches joint drive properties (stiffness, damping, max_effort, max_velocity, friction) for articulations (including robots). The cache is cleared when the environment resets, ensuring fresh values are fetched at the start of each episode.

If the articulation with the specified UID does not exist in the environment, a zero tensor will be returned for each attribute.

The cached data is stored per entity UID. When called, if data is cached, it returns a clone of the cached tensor to prevent accidental modifications.

Note

Joint drive properties are typically constant during an episode, so caching improves performance by avoiding repeated queries.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

Methods:

__init__(cfg, env)	Initialize the joint drive functor.
reset([env_ids])	Clear the cached joint drive properties.

__init__(cfg, env)

Initialize the joint drive functor.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

reset(env_ids=None)

Clear the cached joint drive properties.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, which clears all cache.

Return type:

None

embodichain.lab.gym.envs.managers.observations.get_object_body_scale(env, obs, entity_cfg)

Get the body scale of the objects in the environment.

If the object with the specified UID does not exist in the environment, a zero tensor will be returned.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 3) representing the body scale of the objects.

embodichain.lab.gym.envs.managers.observations.get_object_pose(env, obs, entity_cfg, to_matrix=True)

Get the arena poses of the objects in the environment.

If the object with the specified UID does not exist in the environment, a zero tensor will be returned.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

• to_matrix (bool) – Whether to return the pose as a 4x4 transformation matrix. If False, returns as (position, quaternion).

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 7) or (num_envs, 4, 4) representing the world poses of the objects.

embodichain.lab.gym.envs.managers.observations.get_object_uid(env, obs, entity_cfg)

Get the user IDs of the objects in the environment.

If the object with the specified UID does not exist in the environment, a zero tensor will be returned.

Note

• If asset is RigidObject, the user IDs is shaped as (num_envs,)

• If asset is Articulation or Robot, the user IDs is shaped as (num_envs, num_links) and ordered by

  link_names in the configuration.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

Return type:

Tensor

Returns:

A tensor of shape (num_envs,) representing the user IDs of the objects.

class embodichain.lab.gym.envs.managers.observations.get_rigid_object_physics_attributes

Get the physics attributes of the rigid object in the environment with caching.

This functor retrieves and caches physics attributes (mass, friction, damping, inertia) for rigid objects. The cache is cleared when the environment resets, ensuring fresh values are fetched at the start of each episode.

If the rigid object with the specified UID does not exist in the environment, a zero tensor will be returned for each attribute.

The cached data is stored per entity UID. When called, if data is cached, it returns a clone of the cached tensor to prevent accidental modifications.

Note

Physics attributes are typically constant during an episode, so caching improves performance by avoiding repeated queries to the physics engine.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

Methods:

__init__(cfg, env)	Initialize the physics attributes functor.
reset([env_ids])	Clear the cached physics attributes.

__init__(cfg, env)

Initialize the physics attributes functor.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

reset(env_ids=None)

Clear the cached physics attributes.

Parameters:

env_ids (Optional[Sequence[int]]) – The environment ids. Defaults to None, which clears all cache.

Return type:

None

embodichain.lab.gym.envs.managers.observations.get_rigid_object_pose(env, obs, entity_cfg, to_matrix=True)

Get the world poses of the rigid objects in the environment.

If the rigid object with the specified UID does not exist in the environment, a zero tensor will be returned.

Note

This method will be deprecated in the future and replaced by get_object_pose as the distinction between rigid objects and general objects is being removed. Please use get_object_pose instead when possible.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

• to_matrix (bool) – Whether to return the pose as a 4x4 transformation matrix. If False, returns as (position, quaternion).

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 7) or (num_envs, 4, 4) representing the world poses of the rigid objects.

embodichain.lab.gym.envs.managers.observations.get_rigid_object_velocity(env, obs, entity_cfg)

Get the world velocities of the rigid objects in the environment.

If the rigid object with the specified UID does not exist in the environment, a zero tensor will be returned.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity.

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 6) representing the linear and angular velocities of the rigid objects.

embodichain.lab.gym.envs.managers.observations.get_robot_eef_pose(env, obs, part_name=None, position_only=False)

Get robot end-effector pose using forward kinematics.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• part_name (str | None) – The name of the control part. If None, uses default part.

• position_only (bool) – If True, returns only position (3D). If False, returns full pose (4x4 matrix).

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 3) if position_only=True, or (num_envs, 4, 4) otherwise.

embodichain.lab.gym.envs.managers.observations.get_sensor_intrinsics(env, obs, entity_cfg, is_right=False)

Get the intrinsic matrix of a sensor (camera).

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the sensor entity.

• is_right (bool) – Whether to return the right camera intrinsics for stereo cameras. Defaults to False (left camera). Ignored for monocular cameras.

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 3, 3) representing the camera intrinsics.

embodichain.lab.gym.envs.managers.observations.get_sensor_pose_in_robot_frame(env, obs, entity_cfg, robot_uid=None, is_right=False)

Get the pose of a sensor in the robot’s base coordinate frame.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – The observation dictionary.

• entity_cfg (SceneEntityCfg) – The configuration of the sensor entity.

• robot_uid (str | None) – The uid of the robot. If None, uses the default robot from env.

• is_right (bool) – Whether to return the right camera intrinsics for stereo cameras. Defaults to False (left camera). Ignored for monocular cameras.

Return type:

Tensor

Returns:

A tensor of shape (num_envs, 4, 4) representing the sensor pose in robot coordinates as a transformation matrix.

embodichain.lab.gym.envs.managers.observations.normalize_robot_joint_data(env, data, joint_ids, limit='qpos_limits', range=[0.0, 1.0])

Normalize the robot joint positions to the range of [0, 1] based on the joint limits.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• obs – The observation dictionary.

• joint_ids (Sequence[int]) – The indices of the joints to be normalized.

• limit (Literal['qpos_limits', 'qvel_limits']) – The type of joint limits to be used for normalization. Options are: - qpos_limits: Use the joint position limits for normalization. - qvel_limits: Use the joint velocity limits for normalization.

• range (Sequence[float]) – The range to normalize the joint data to. Defaults to [0.0, 1.0].

Return type:

Tensor

embodichain.lab.gym.envs.managers.observations.target_position(env, obs, target_pose_key='goal_pose')

Get virtual target position from env state.

Reads target pose from env.{target_pose_key} (set by randomize_target_pose event). Returns zeros if not yet initialized (e.g., during env initialization before reset).

Parameters:

• env (EmbodiedEnv) – The environment instance

• obs (TensorDict[str, Union[Tensor, TensorDict[str, Tensor]]]) – Observation dict (unused, for API compatibility)

• target_pose_key (str) – Key for target pose in env (default: “goal_pose”)

Return type:

Tensor

Returns:

Target position tensor of shape (num_envs, 3). Returns zeros if target_pose_key is not found (e.g., before first reset).

Reward Functions

Common reward functors for reinforcement learning tasks.

Functions:

action_smoothness_penalty(env, obs, action, info)	Penalize large action changes between consecutive timesteps.
distance_between_objects(env, obs, action, info)	Reward based on distance between two rigid objects.
distance_to_target(env, obs, action, info[, ...])	Reward based on absolute distance to a virtual target pose.
incremental_distance_to_target(env, obs, ...)	Incremental reward for progress toward a virtual target pose.
joint_limit_penalty(env, obs, action, info)	Penalize robot joints that are close to their position limits.
joint_velocity_penalty(env, obs, action, info)	Penalize high joint velocities to encourage smooth motion.
orientation_alignment(env, obs, action, info)	Reward rotational alignment between two rigid objects.
reaching_behind_object(env, obs, action, info)	Reward for positioning end-effector behind object for pushing.
success_reward(env, obs, action, info)	Sparse bonus reward when task succeeds.

embodichain.lab.gym.envs.managers.rewards.action_smoothness_penalty(env, obs, action, info)

Penalize large action changes between consecutive timesteps.

Encourages smooth control commands by penalizing sudden changes in actions. Reads the previous action from the RL rollout_buffer (action / done). Returns zeros when that buffer is unavailable (e.g. evaluation).

Return type:

Tensor

Returns:

Penalty tensor of shape (num_envs,). Zero on first step (no previous action), negative on subsequent steps (larger change = more negative).

Example

```json {

“func”: “action_smoothness_penalty”, “weight”: 0.01, “params”: {}

}

embodichain.lab.gym.envs.managers.rewards.distance_between_objects(env, obs, action, info, source_entity_cfg=None, target_entity_cfg=None, exponential=False, sigma=1.0)

Reward based on distance between two rigid objects.

Encourages the source object to get closer to the target object. Can use either linear negative distance or exponential Gaussian-shaped reward.

Parameters:

• source_entity_cfg (SceneEntityCfg) – Configuration for the source object (e.g., {“uid”: “cube”})

• target_entity_cfg (SceneEntityCfg) – Configuration for the target object (e.g., {“uid”: “goal_sphere”})

• exponential (bool) – If True, use exponential reward exp(-d²/2σ²), else use -distance

• sigma (float) – Standard deviation for exponential reward (controls reward spread)

Return type:

Tensor

Returns:

Reward tensor of shape (num_envs,). Higher when objects are closer. - Linear mode: ranges from -inf to 0 (0 when objects touch) - Exponential mode: ranges from 0 to 1 (1 when objects touch)

Example

```json {

“func”: “distance_between_objects”, “weight”: 0.5, “params”: {

“source_entity_cfg”: {“uid”: “cube”}, “target_entity_cfg”: {“uid”: “target”}, “exponential”: true, “sigma”: 0.2

}

}

embodichain.lab.gym.envs.managers.rewards.distance_to_target(env, obs, action, info, source_entity_cfg=None, target_pose_key='target_pose', exponential=False, sigma=1.0, use_xy_only=False)

Reward based on absolute distance to a virtual target pose.

Encourages an object to get closer to a target pose specified in the info dict. Unlike incremental_distance_to_target, this provides direct distance-based reward.

Parameters:

• source_entity_cfg (SceneEntityCfg) – Configuration for the object (e.g., {“uid”: “cube”})

• target_pose_key (str) – Key in info dict for target pose (default: “target_pose”) Can be (num_envs, 3) position or (num_envs, 4, 4) transform

• exponential (bool) – If True, use exponential reward exp(-d²/2σ²), else use -distance

• sigma (float) – Standard deviation for exponential reward (default: 1.0)

• use_xy_only (bool) – If True, ignore z-axis and only consider horizontal distance

Return type:

Tensor

Returns:

Reward tensor of shape (num_envs,). - Linear mode: -distance (negative, approaches 0 when close) - Exponential mode: exp(-d²/2σ²) (0 to 1, approaches 1 when close)

Example

```json {

“func”: “distance_to_target”, “weight”: 0.5, “params”: {

“source_entity_cfg”: {“uid”: “cube”}, “target_pose_key”: “goal_pose”, “exponential”: false, “use_xy_only”: true

}

}

embodichain.lab.gym.envs.managers.rewards.incremental_distance_to_target(env, obs, action, info, source_entity_cfg=None, target_pose_key='target_pose', tanh_scale=10.0, positive_weight=1.0, negative_weight=1.0, use_xy_only=False)

Incremental reward for progress toward a virtual target pose.

Rewards the robot for getting closer to the target compared to previous timestep. Stores previous distance in env._reward_states for comparison. Uses tanh shaping to normalize rewards and supports asymmetric weighting for approach vs. retreat.

Parameters:

• source_entity_cfg (SceneEntityCfg) – Configuration for the object (e.g., {“uid”: “cube”})

• target_pose_key (str) – Key for target pose in env (default: “target_pose”) Reads from env._{target_pose_key} set by randomize_target_pose event Can be (num_envs, 3) position or (num_envs, 4, 4) transform

• tanh_scale (float) – Scaling for tanh normalization (higher = more sensitive, default: 10.0)

• positive_weight (float) – Multiplier for reward when getting closer (default: 1.0)

• negative_weight (float) – Multiplier for penalty when moving away (default: 1.0)

• use_xy_only (bool) – If True, ignore z-axis and only consider horizontal distance

Returns:

• Positive when getting closer (scaled by positive_weight)

• Negative when moving away (scaled by negative_weight)

• Magnitude bounded by tanh function

Return type:

Reward tensor of shape (num_envs,). Zero on first call, then

Note

This function maintains state using env._reward_states[f”prev_dist_{uid}_{key}”]. State is automatically reset when the environment resets.

Example

```json {

“func”: “incremental_distance_to_target”, “weight”: 1.0, “params”: {

“source_entity_cfg”: {“uid”: “cube”}, “target_pose_key”: “goal_pose”, “tanh_scale”: 10.0, “positive_weight”: 2.0, “negative_weight”: 0.5, “use_xy_only”: true

}

}

embodichain.lab.gym.envs.managers.rewards.joint_limit_penalty(env, obs, action, info, robot_uid='robot', joint_ids=slice(None, None, None), margin=0.1)

Penalize robot joints that are close to their position limits.

Prevents joints from reaching their physical limits, which can cause instability or singularities. Penalty increases as joints approach limits within the margin.

Parameters:

• robot_uid (str) – Robot entity UID in simulation (default: “robot”)

• joint_ids (slice | list[int]) – Joint indices to monitor (default: all joints)

• margin (float) – Normalized distance threshold (0 to 1). Penalty applied when joint is within this fraction of its range from either limit. Example: 0.1 means penalty when within 10% of limits.

Return type:

Tensor

Returns:

Penalty tensor of shape (num_envs,). Always negative or zero. Sum of penalties across all monitored joints.

Example

```json {

“func”: “joint_limit_penalty”, “weight”: 0.01, “params”: {

“robot_uid”: “robot”, “joint_ids”: [0, 1, 2, 3, 4, 5], “margin”: 0.1

}

}

embodichain.lab.gym.envs.managers.rewards.joint_velocity_penalty(env, obs, action, info, robot_uid='robot', joint_ids=None, part_name=None)

Penalize high joint velocities to encourage smooth motion.

Computes the L2 norm of joint velocities and returns negative value as penalty. Useful for preventing jerky or unstable robot movements.

Parameters:

• robot_uid (str) – Robot entity UID in simulation (default: “robot”)

• joint_ids (slice | list[int] | None) – Specific joint indices to penalize. Takes priority over part_name. Example: [0, 1, 2] or slice(0, 6)

• part_name (str | None) – Control part name (e.g., “arm”). Used only if joint_ids is None. Will penalize all joints in the specified part.

Return type:

Tensor

Returns:

Penalty tensor of shape (num_envs,). Always negative or zero. Magnitude increases with joint velocity (larger velocity = more negative).

Example

```json {

“func”: “joint_velocity_penalty”, “weight”: 0.001, “params”: {

“robot_uid”: “robot”, “part_name”: “arm”

}

}

embodichain.lab.gym.envs.managers.rewards.orientation_alignment(env, obs, action, info, source_entity_cfg=None, target_entity_cfg=None)

Reward rotational alignment between two rigid objects.

Encourages the source object’s orientation to match the target object’s orientation. Uses rotation matrix trace to measure alignment.

Parameters:

• source_entity_cfg (SceneEntityCfg) – Configuration for the source object (e.g., {“uid”: “cube”})

• target_entity_cfg (SceneEntityCfg) – Configuration for the target object (e.g., {“uid”: “reference”})

Return type:

Tensor

Returns:

Reward tensor of shape (num_envs,). Ranges from -1 to 1. - 1.0: Perfect alignment (same orientation) - 0.0: 90° rotation difference - -1.0: 180° rotation difference (opposite orientation)

Example

```json {

“func”: “orientation_alignment”, “weight”: 0.5, “params”: {

“source_entity_cfg”: {“uid”: “object”}, “target_entity_cfg”: {“uid”: “goal_object”}

}

}

embodichain.lab.gym.envs.managers.rewards.reaching_behind_object(env, obs, action, info, object_cfg=None, target_pose_key='goal_pose', behind_offset=0.015, height_offset=0.015, distance_scale=5.0, part_name=None)

Reward for positioning end-effector behind object for pushing.

Encourages the robot’s end-effector to reach a position behind the object along the object-to-goal direction. Useful for push manipulation tasks.

Parameters:

• object_cfg (SceneEntityCfg) – Configuration for the object to push (e.g., {“uid”: “cube”})

• target_pose_key (str) – Key in info dict for goal pose (default: “goal_pose”) Can be (num_envs, 3) position or (num_envs, 4, 4) transform

• behind_offset (float) – Distance behind object to reach (in meters, default: 0.015)

• height_offset (float) – Additional height above object (in meters, default: 0.015)

• distance_scale (float) – Scaling factor for tanh function (higher = steeper, default: 5.0)

• part_name (str) – Robot part name for FK computation (e.g., “arm”)

Return type:

Tensor

Returns:

Reward tensor of shape (num_envs,). Ranges from 0 to 1. - 1.0: End-effector at ideal pushing position - 0.0: End-effector far from ideal position

Example

```json {

“func”: “reaching_behind_object”, “weight”: 0.1, “params”: {

“object_cfg”: {“uid”: “cube”}, “target_pose_key”: “goal_pose”, “behind_offset”: 0.015, “height_offset”: 0.015, “distance_scale”: 5.0, “part_name”: “arm”

}

}

embodichain.lab.gym.envs.managers.rewards.success_reward(env, obs, action, info)

Sparse bonus reward when task succeeds.

Provides a fixed reward when the task success condition is met. Reads success status from info[‘success’] which should be set by the environment.

Return type:

Tensor

Returns:

Reward tensor of shape (num_envs,). - 1.0 when successful - 0.0 when not successful or if ‘success’ key missing

Note

The environment’s get_info() must populate info[‘success’] with a boolean tensor indicating success status for each environment.

Example

```json {

“func”: “success_reward”, “weight”: 10.0, “params”: {}

}

Event Functions

Functions:

create_rigid_constraint(env, env_ids, ...[, ...])	Attach two rigid objects via a fixed constraint for the given env_ids.
drop_rigid_object_group_sequentially(env, ...)	Drop rigid object group from a specified height sequentially in the environment.
register_entity_attrs(env, env_ids, entity_cfg)	Register the atrributes of an entity to the env.registration dict.
remove_rigid_constraint(env, env_ids, name)	Remove the named constraint for the given env_ids.
set_detached_uids_for_env_reset(env, ...)	Set the UIDs of objects that are detached from automatic reset in the environment.
wait_for_dynamic_objects_to_settle(env, env_ids)	Advance physics until selected dynamic objects remain stationary.

Classes:

prepare_extra_attr
replace_assets_from_group	Replace assets in the environment from a specified group of assets.

embodichain.lab.gym.envs.managers.events.create_rigid_constraint(env, env_ids, obj_a_cfg, obj_b_cfg, name, local_frame_a=None, local_frame_b=None)

Attach two rigid objects via a fixed constraint for the given env_ids.

Registered under a custom event mode (e.g. "attach"); the task triggers it with env.event_manager.apply(mode="attach", env_ids=...). Delegates to SimulationManager.create_rigid_constraint().

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Tensor | None) – Target environment indices. None -> all envs.

• obj_a_cfg (SceneEntityCfg) – SceneEntityCfg pointing at the first RigidObject.

• obj_b_cfg (SceneEntityCfg) – SceneEntityCfg pointing at the second RigidObject.

• name (str) – Base constraint name; per-arena names derived by the sim layer.

• local_frame_a (ndarray | None) – Local joint frame on object A. None -> identity (object A’s origin). Accepts (4,4) or (N,4,4).

• local_frame_b (ndarray | None) – Local joint frame on object B. None -> computed per env as inv(pose_B) @ pose_A so the constraint welds the objects at their current relative pose. Accepts (4,4) or (N,4,4).

Raises:

RuntimeError – If either entity is not a RigidObject.

Return type:

None

embodichain.lab.gym.envs.managers.events.drop_rigid_object_group_sequentially(env, env_ids, entity_cfg, drop_position=[0.0, 0.0, 1.0], position_range=([-0.1, -0.1, 0.0], [0.1, 0.1, 0.0]), physics_step=2)

Drop rigid object group from a specified height sequentially in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the event.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• drop_position (List[float]) – The base position from which to drop the objects. Default is [0.0, 0.0, 1.0].

• position_range (Tuple[List[float], List[float]]) – The range for randomizing the drop position around the base position.

• physics_step (int) – The number of physics steps to simulate after dropping the objects. Default is 2.

Return type:

None

class embodichain.lab.gym.envs.managers.events.prepare_extra_attr

Methods:

__init__(cfg, env)

Initializes the event manager with the given configuration and environment.

__init__(cfg, env)

Initializes the event manager with the given configuration and environment.

Parameters:

• cfg (FunctorCfg) – The configuration object for the functor.

• env (EmbodiedEnv) – The embodied environment instance.

extra_attrs

A dictionary to hold additional attributes.

Type:

dict

embodichain.lab.gym.envs.managers.events.register_entity_attrs(env, env_ids, entity_cfg, registration='affordance_datas', attrs=[], prefix=True)

Register the atrributes of an entity to the env.registration dict.

TODO: Currently this method only support 1 env or multi-envs that reset() together,

as it’s behavior is to update a overall dict every time it’s called.

In the future, asynchronously reset mode shall be supported.

Parameters:

• env (EmbodiedEnv) – The environment the entity is in.

• env_ids (torch.Tensor | None) – The ids of the envs that the entity should be registered.

• entity_cfg (SceneEntityCfg) – The config of the entity.

• attrs (List[str]) – The list of entity attributes that asked to be registered.

• registration (str, optional) – The env’s registration string where the attributes should be injected to.

embodichain.lab.gym.envs.managers.events.remove_rigid_constraint(env, env_ids, name)

Remove the named constraint for the given env_ids.

Delegates to SimulationManager.remove_rigid_constraint(). Idempotent: warns (via the sim layer) if the constraint is not found.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Tensor | None) – Target environment indices. None -> all envs.

• name (str) – Base constraint name to remove.

Return type:

None

class embodichain.lab.gym.envs.managers.events.replace_assets_from_group

Replace assets in the environment from a specified group of assets.

The group of assets can be defined in the following ways:

• A directory containing multiple asset files.

• A json file listing multiple assets with their properties. (not supported yet)

• … (other methods can be added in the future)

Methods:

__init__(cfg, env)

Initialize the term.

__init__(cfg, env)

Initialize the term.

Parameters:

• cfg (FunctorCfg) – The configuration of the functor.

• env (EmbodiedEnv) – The environment instance.

Raises:

ValueError – If the asset is not a RigidObject or an Articulation.

embodichain.lab.gym.envs.managers.events.set_detached_uids_for_env_reset(env, env_ids, uids)

Set the UIDs of objects that are detached from automatic reset in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the event.

• uids (list[str]) – The list of UIDs to be detached from automatic reset.

Return type:

None

embodichain.lab.gym.envs.managers.events.wait_for_dynamic_objects_to_settle(env, env_ids, entity_cfgs=None, linear_velocity_threshold=0.03, angular_velocity_threshold=0.2, min_steps=10, max_steps=240, check_interval_steps=2, required_stable_checks=3, timeout_behavior='warn', allow_partial_envs=False)

Advance physics until selected dynamic objects remain stationary.

The functor waits at least min_steps and then polls every check_interval_steps. Every selected body in every selected environment must remain below both velocity thresholds for required_stable_checks consecutive polls. It never clears dynamics and never advances beyond max_steps.

When entity_cfgs is None, all dynamic rigid objects, rigid object groups, and non-robot articulations are selected automatically. Static and kinematic entities are ignored during automatic discovery.

Attention

SimulationManager.update() advances the entire vectorized physics world. Partial env_ids are therefore rejected by default. Set allow_partial_envs=True only when advancing non-target environments is acceptable.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Tensor | Sequence[int] | slice | None) – Target environment IDs. None or slice(None) selects all environments.

• entity_cfgs (Sequence[SceneEntityCfg] | None) – Explicit settle targets. None discovers all supported dynamic non-robot entities.

• linear_velocity_threshold (float) – Maximum stable linear speed in meters per second.

• angular_velocity_threshold (float) – Maximum stable angular speed in radians per second.

• min_steps (int) – Physics steps to run before the first stability check.

• max_steps (int) – Maximum total number of physics steps to run.

• check_interval_steps (int) – Physics steps between stability checks.

• required_stable_checks (int) – Consecutive stable checks required before return.

• timeout_behavior (Literal['warn', 'raise']) – "warn" to log and continue or "raise" to raise TimeoutError when max_steps is reached.

• allow_partial_envs (bool) – Whether to permit a partial environment selection despite whole-world physics advancement.

Raises:

• IndexError – If an environment ID is outside the valid range.

• RuntimeError – If a dynamic target has no readable body data.

• TimeoutError – If objects do not settle and timeout_behavior is "raise".

• TypeError – If a parameter or entity configuration has the wrong type.

• ValueError – If parameters, targets, or environment selection are invalid.

Return type:

None

Recording Functions

Classes:

record_camera_data	Record camera data in the environment.
record_camera_data_async	Record camera data for multiple environments, merge and save as a single video at episode end.
validation_cameras	This functor creates validation cameras during initialization and captures their data when called.

class embodichain.lab.gym.envs.managers.record.record_camera_data

Record camera data in the environment. The camera is usually setup with third-person view, and is used to record the scene during the episode. It is helpful for debugging and visualization.

Note

Currently, the functor is implemented in interval’ mode such that, it can only save the recorded frames when in :meth:`env.step() function call. For example: `python env.step() # perform multiple steps in the same episode env.reset() env.step()  # the video of the first episode will be saved here. ` The final episode frames will not be saved in the current implementation. We may improve it in the future.

Methods:

__init__(cfg, env)	Initialize the functor.
close()	Finalize the recorder; repeated calls are safe.
discard_and_clear([env_ids])	Discard recorded frames without creating an episode video.
finalize()	Discard uncommitted frames and close the recorder exactly once.
save_and_clear([env_ids])	Save recorded frames as video and clear the buffer.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The configuration of the functor.

• env (EmbodiedEnv) – The environment instance.

Raises:

ValueError – If the asset is not a RigidObject or an Articulation.

close()

Finalize the recorder; repeated calls are safe.

Return type:

None

discard_and_clear(env_ids=None)

Discard recorded frames without creating an episode video.

Return type:

None

finalize()

Discard uncommitted frames and close the recorder exactly once.

Return type:

None

save_and_clear(env_ids=None)

Save recorded frames as video and clear the buffer.

This method is called from EmbodiedEnv._initialize_episode() to ensure frames are saved before the episode is reset. This avoids the issue where the final episode’s frames are lost because the save previously relied on detecting a reset inside __call__().

Return type:

None

class embodichain.lab.gym.envs.managers.record.record_camera_data_async

Record camera data for multiple environments, merge and save as a single video at episode end.

Methods:

__init__(cfg, env)	Initialize the functor.
discard_and_clear([env_ids])	Discard live frames while preserving already committed episodes.
finalize()	Flush committed frame sets and reject incomplete committed batches.
save_and_clear([env_ids])	Commit selected rows immediately instead of waiting for a later step.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The configuration of the functor.

• env (EmbodiedEnv) – The environment instance.

Raises:

ValueError – If the asset is not a RigidObject or an Articulation.

discard_and_clear(env_ids=None)

Discard live frames while preserving already committed episodes.

Return type:

None

finalize()

Flush committed frame sets and reject incomplete committed batches.

Return type:

None

save_and_clear(env_ids=None)

Commit selected rows immediately instead of waiting for a later step.

Return type:

None

class embodichain.lab.gym.envs.managers.record.validation_cameras

This functor creates validation cameras during initialization and captures their data when called. The cameras are created once and reused for subsequent calls.

Methods:

__init__(cfg, env)

Initialize the functor.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The configuration object.

• env (EmbodiedEnv) – The environment instance.

Dataset Recording

Dataset manager for orchestrating dataset collection functors.

Classes:

DatasetManager

Manager for orchestrating dataset collection and saving using functors.

class embodichain.lab.gym.envs.managers.dataset_manager.DatasetManager

Manager for orchestrating dataset collection and saving using functors.

The dataset manager supports multiple dataset formats through a functor system: - LeRobot format (via LeRobotRecorder) - HDF5 format (via HDF5Recorder) - Zarr format (via ZarrRecorder) - Custom formats (via user-defined functors)

Each functor’s step() method is called once per environment step and handles: - Recording observation-action pairs - Detecting episode completion (dones=True) - Auto-saving completed episodes

Example configuration:

>>> from embodichain.lab.gym.envs.managers.cfg import DatasetFunctorCfg
>>> from embodichain.lab.gym.envs.managers.datasets import LeRobotRecorder
>>>
>>> @configclass
>>> class MyEnvCfg:
>>>     dataset: dict = {
>>>         "lerobot": DatasetFunctorCfg(
>>>             func=LeRobotRecorder,
>>>             save_failed_episodes=True,
>>>             params={
>>>                 "robot_meta": {...},
>>>                 "instruction": {"lang": "pick and place"},
>>>                 "extra": {"scene_type": "kitchen"},
>>>                 "save_path": "/data/datasets"
>>>             }
>>>         )
>>>     }

Methods:

__init__(cfg, env)	Initialize the dataset manager.
apply(mode[, env_ids])	Apply dataset functors for the specified mode.
clear_cache()	Clear cached data from all dataset functors (for online training).
close()	Finalize all dataset functors; repeated calls are safe.
finalize()	Finalize every dataset functor exactly once.
get_cached_data()	Get cached data from all dataset functors (for online training).
get_functor_cfg(functor_name)	Gets the configuration for the specified functor.
reset([env_ids])	Reset all dataset functors.

Attributes:

active_functors	Name of active dataset functors by mode.
available_modes	List of available modes for the dataset manager.
save_failed_episodes	Whether any configured dataset recorder should keep failed episodes.

__init__(cfg, env)

Initialize the dataset manager.

Parameters:

• cfg (object) – Configuration object containing dataset functor configurations.

• env (EmbodiedEnv) – The environment instance.

property active_functors: dict[str, list[str]]

Name of active dataset functors by mode.

The keys are the modes and the values are the names of the dataset functors.

apply(mode, env_ids=None)

Apply dataset functors for the specified mode.

This method saves completed episodes by reading data from the environment’s episode buffers. It should be called before clearing the buffers during reset.

Parameters:

• mode (str) – The mode to apply (currently only “save” is supported).

• env_ids (Union[Sequence[int], Tensor, None]) – The indices of the environments to apply the functor to. Defaults to None, in which case the functor is applied to all environments.

Return type:

None

property available_modes: list[str]

List of available modes for the dataset manager.

clear_cache()

Clear cached data from all dataset functors (for online training).

Iterates through all functors and clears their cache if they support online training mode (have clear_cache method).

Return type:

int

Returns:

Total number of cached items cleared across all functors.

close()

Finalize all dataset functors; repeated calls are safe.

Return type:

Optional[str]

finalize()

Finalize every dataset functor exactly once.

Finalization is a storage barrier only; individual recorders must not implicitly commit live episode buffers here. All functors are attempted even when one fails, and their failures are reported together.

Return type:

Optional[str]

Returns:

Path to the first finalized dataset, or None if none was returned.

Raises:

RuntimeError – If one or more functors fail to finalize.

get_cached_data()

Get cached data from all dataset functors (for online training).

Iterates through all functors and collects cached data from those that support online training mode (have get_cached_data method).

Return type:

list[Dict[str, Any]]

Returns:

List of cached data dictionaries from all functors.

get_functor_cfg(functor_name)

Gets the configuration for the specified functor.

Parameters:

functor_name (str) – The name of the dataset functor.

Return type:

DatasetFunctorCfg

Returns:

The configuration of the dataset functor.

Raises:

ValueError – If the functor name is not found.

reset(env_ids=None)

Reset all dataset functors.

Parameters:

env_ids (Union[Sequence[int], Tensor, None]) – The environment ids. Defaults to None.

Return type:

dict[str, float]

Returns:

Empty dict (no logging info).

property save_failed_episodes: bool

Whether any configured dataset recorder should keep failed episodes.

Dataset functors for collecting and saving episode data.

Classes:

LeRobotRecorder

Functor for recording episodes in LeRobot format.

class embodichain.lab.gym.envs.managers.datasets.LeRobotRecorder

Functor for recording episodes in LeRobot format.

This functor handles:

• Recording observation-action pairs during episodes

• Converting data to LeRobot format

• Saving episodes when they complete

Methods:

__init__(cfg, env)	Initialize the LeRobot dataset recorder.
close()	Finalize the recorder; repeated calls are safe.
finalize()	Finalize resources without implicitly committing a partial episode.

Attributes:

dataset_path

Path to the dataset directory.

__init__(cfg, env)

Initialize the LeRobot dataset recorder.

Parameters:

• cfg (DatasetFunctorCfg) – Functor configuration containing params: - save_path: Root directory for saving datasets - robot_meta: Robot metadata for dataset - instruction: Optional task instruction - extra: Optional extra metadata - use_videos: Whether to save videos - image_writer_threads: Number of threads for image writing - image_writer_processes: Number of processes for image writing

• env (EmbodiedEnv) – The environment instance

close()

Finalize the recorder; repeated calls are safe.

Return type:

Optional[str]

property dataset_path: str

Path to the dataset directory.

finalize()

Finalize resources without implicitly committing a partial episode.

Episodes are committed only when __call__() is invoked by an explicit reset(save_data=True). Closing the environment therefore leaves any still-live rollout buffer uncommitted.

Return type:

Optional[str]

Returns:

The finalized dataset path, or None when no dataset exists.

Raises:

RuntimeError – If one or more dataset resources cannot be finalized.

Asynchronous LeRobot recorder for parallel environments.

This module provides AsyncLeRobotRecorder, which decouples episode saving from the simulation loop. It is the recommended recorder when running many parallel environments that complete episodes together: instead of blocking env.reset() while episodes are converted and flushed to disk, the completed episode buffers are cloned and handed to a background worker thread, so the simulator can keep stepping.

Classes:

AsyncLeRobotRecorder

LeRobot recorder that saves episodes on a background thread.

class embodichain.lab.gym.envs.managers.async_datasets.AsyncLeRobotRecorder

LeRobot recorder that saves episodes on a background thread.

Drop-in replacement for LeRobotRecorder selected via the dataset config "func": "AsyncLeRobotRecorder". It shares the same on-disk format and feature-building logic; only the timing of the save differs.

Why this helps parallel environments

In the synchronous recorder, LeRobotRecorder.__call__() runs inside env.reset() (via DatasetManager.apply) and blocks the simulator while it iterates add_frame + save_episode for every finished env. With num_envs=N all finishing at once, the sim stalls for the sum of all episodes’ save time every reset.

This recorder instead, on each apply:

1. Reads each env’s rollout-buffer slice (obs/actions).

2. Clones the slice to CPU (detached from the live buffer).

3. Clones frame annotations and episode/segment metadata with the payload.

4. Pushes the detached payload onto a queue.

5. Returns immediately - the sim is free to reset and keep stepping.

A single daemon worker thread drains the queue and runs the standard LeRobotRecorder._persist_episode_payload() on each cloned payload.

Correctness

• No concurrent dataset access. LeRobotDataset is not thread-safe; only the worker thread ever calls add_frame / save_episode / mutates curr_episode. The main thread only enqueues and, at close, drains.

• No buffer race. The slice is cloned in the caller thread before the buffer is cleared on reset, so the worker never reads memory that the sim is overwriting.

• Ordering. A single worker preserves FIFO episode order, so episode_index assignment is deterministic.

• Drain on close. finalize() joins the worker before the parent flushes the image writer and finalizes the dataset.

Note

The clone copies each episode’s camera frames into host RAM. Memory use is bounded by how far the worker falls behind (typically it keeps up, since per-frame PNG write is the only heavy step and can itself be offloaded via image_writer_threads). For very high resolutions or many envs, monitor RSS.

type cfg:

param cfg:

DatasetFunctorCfg with the same params as LeRobotRecorder. The image_writer_threads / image_writer_processes params are honored and combine with the background worker (two levels of async: episode conversion off the sim thread, PNG writes off the worker thread).

type env:

EmbodiedEnv

param env:

The environment instance.

Methods:

__init__(cfg, env)	Initialize the LeRobot dataset recorder.
finalize()	Drain committed writes, finalize storage, and surface all failures.

__init__(cfg, env)

Initialize the LeRobot dataset recorder.

Parameters:

• cfg – Functor configuration containing params: - save_path: Root directory for saving datasets - robot_meta: Robot metadata for dataset - instruction: Optional task instruction - extra: Optional extra metadata - use_videos: Whether to save videos - image_writer_threads: Number of threads for image writing - image_writer_processes: Number of processes for image writing

• env (EmbodiedEnv) – The environment instance

finalize()

Drain committed writes, finalize storage, and surface all failures.

The queue is a durability barrier for episodes explicitly committed by reset(save_data=True). A live rollout that was never enqueued is not saved during close.

Return type:

Optional[str]

Returns:

The finalized dataset path.

Raises:

RuntimeError – If any queued episode or dataset resource failed.

Randomization

Domain-randomization event functors (physics, visual, spatial, geometry).

Implemented as event functors registered via EventCfg and dispatched by EventManager at startup / reset / interval modes.

Submodules

physics
visual
spatial
geometry

Physics

Functions:

randomize_articulation_mass(env, env_ids, ...)	Randomize the mass of articulation links in the environment.
randomize_rigid_object_center_of_mass(env, ...)	Randomize the center of mass of rigid objects in the environment.
randomize_rigid_object_mass(env, env_ids, ...)	Randomize the mass of rigid objects in the environment.

embodichain.lab.gym.envs.managers.randomization.physics.randomize_articulation_mass(env, env_ids, entity_cfg, mass_range, link_names=None, relative=False)

Randomize the mass of articulation links in the environment.

Uses regular expression matching to select which links to randomize.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | list[int]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration for the scene entity.

• mass_range (tuple[float, float] | dict[str, tuple[float, float]]) – The range (min, max) to sample the mass from. If a dict, keys are link names and values are per-link mass ranges. When a dict is provided, link_names is ignored and the dict keys are used instead.

• link_names (str | list[str] | None) – A regex pattern or list of regex patterns to match link names. If None, all links are randomized. Ignored when mass_range is a dict. Defaults to None.

• relative (bool) – Whether to apply the mass change relative to the current mass. Defaults to False.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.physics.randomize_rigid_object_center_of_mass(env, env_ids, entity_cfg, com_pos_offset_range)

Randomize the center of mass of rigid objects in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | list[int]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration for the scene entity.

• com_pos_offset_range (tuple[list[float], list[float]]) – The range (min, max) to sample the center of mass offset from.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.physics.randomize_rigid_object_mass(env, env_ids, entity_cfg, mass_range, relative=False)

Randomize the mass of rigid objects in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | list[int]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration for the scene entity.

• mass_range (tuple[float, float]) – The range (min, max) to sample the mass from.

• relative (bool) – Whether to apply the mass change relative to the initial mass. Defaults to False.

Return type:

None

Visual

Functions:

randomize_camera_extrinsics(env, env_ids, ...)	Randomize camera extrinsic properties (position and orientation).
randomize_camera_intrinsics(env, env_ids, ...)	Randomize camera intrinsic properties by adding, scaling, or setting random values.
randomize_emission_light(env, env_ids[, ...])	Randomize emission light properties by adding, scaling, or setting random values.
randomize_light(env, env_ids, entity_cfg[, ...])	Randomize light properties by adding, scaling, or setting random values.
set_rigid_object_group_visual_material(env, ...)	Set a rigid object group's visual material (deterministic, non-random).
set_rigid_object_visual_material(env, ...)	Set a rigid object's visual material (deterministic, non-random).

Classes:

randomize_indirect_lighting	Randomize the environment's indirect (IBL) lighting or emissive light.
randomize_visual_material	Randomize the visual material properties of a RigidObject or Articulation.

embodichain.lab.gym.envs.managers.randomization.visual.randomize_camera_extrinsics(env, env_ids, entity_cfg, pos_range=None, euler_range=None, eye_range=None, target_range=None, up_range=None)

Randomize camera extrinsic properties (position and orientation).

Behavior: - If extrinsics config has a parent field (attach mode), pos_range/euler_range are used to perturb the initial pose (pos, quat),

and set_local_pose is called to attach the camera to the parent node. In this case, pose is related to parent.

• If extrinsics config uses eye/target/up (no parent), eye_range/target_range/up_range are used to perturb the initial eye, target, up vectors,

  and look_at is called to set the camera orientation.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Optional[Tensor]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• pos_range (tuple[list[float], list[float]] | None) – Position perturbation range (attach mode).

• euler_range (tuple[list[float], list[float]] | None) – Euler angle perturbation range (attach mode).

• eye_range (tuple[list[float], list[float]] | None) – Eye position perturbation range (look_at mode).

• target_range (tuple[list[float], list[float]] | None) – Target position perturbation range (look_at mode).

• up_range (tuple[list[float], list[float]] | None) – Up vector perturbation range (look_at mode).

Return type:

None

embodichain.lab.gym.envs.managers.randomization.visual.randomize_camera_intrinsics(env, env_ids, entity_cfg, focal_x_range=None, focal_y_range=None, cx_range=None, cy_range=None)

Randomize camera intrinsic properties by adding, scaling, or setting random values.

This function allows randomizing camera intrinsic parameters in the scene. The function samples random values from the given distribution parameters and adds, scales, or sets the values into the physics simulation based on the operation.

The distribution parameters are tuples of two elements each, representing the lower and upper bounds of the distribution for the focal length (fx, fy) and principal point (cx, cy) components of the camera intrinsics. The function samples random values for each component independently.

Attention

This function applies the same intrinsic properties for all the environments.

focal_x_range and focal_y_range are values added to the camera’s current fx and fy values. focal_xy_range is a combined range for both fx and fy, where the range is specified as [[fx_min, fy_min], [fx_max, fy_max]]. cx_range and cy_range are values added to the camera’s current cx and cy values.

Tip

This function uses CPU tensors to assign camera intrinsic properties.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Union[torch.Tensor, None]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• focal_x_range (tuple[float, float] | None) – The range for the focal length x randomization.

• focal_y_range (tuple[float, float] | None) – The range for the focal length y randomization.

• cx_range (tuple[float, float] | None) – The range for the principal point x randomization.

• cy_range (tuple[float, float] | None) – The range for the principal point y randomization.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.visual.randomize_emission_light(env, env_ids, color_range=None, intensity_range=None)

Randomize emission light properties by adding, scaling, or setting random values.

This function allows randomizing emission light properties in the scene. The function samples random values from the given distribution parameters and adds, scales, or sets the values into the physics simulation based on the operation.

The distribution parameters are lists of two elements each, representing the lower and upper bounds of the distribution for the r, g, b components of the light color and intensity. The function samples random values for each component independently. :rtype: None

Attention

This function applied the same emission light properties for all the environments.

color_range is the absolute r, g, b value set on the emission light. intensity_range is the absolute intensity value set on the emission light.

class embodichain.lab.gym.envs.managers.randomization.visual.randomize_indirect_lighting

Randomize the environment’s indirect (IBL) lighting or emissive light.

This functor operates in one of two mutually exclusive modes:

• HDR mode — path is provided. A random .hdr file is chosen from the folder on every call and applied via set_indirect_lighting().

• Emissive mode — emissive_color_range and/or emissive_intensity_range are provided. The emissive light color and intensity are sampled uniformly on every call and applied via set_emission_light().

Providing both path and emissive parameters simultaneously is an error.

Attention

This functor applies the same lighting to all environments.

Tip

The path parameter is resolved via get_data_path(), so it supports absolute paths, data-root-relative paths, and dataset-class paths (e.g. "EnvMapHDR").

emissive_color_range is a pair of [r, g, b] lists representing the lower and upper bounds for sampling the emissive color, e.g. [[0.8, 0.8, 0.8], [1.0, 1.0, 1.0]].

emissive_intensity_range is a [min, max] pair for the emissive intensity scalar, e.g. [80.0, 150.0].

Methods:

__init__(cfg, env)

Initialize the functor.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) –

  The configuration of the functor.

  • HDR mode: set params["path"] to a folder of .hdr files.

  • Emissive mode: set params["emissive_color_range"] (pair of RGB lists) and/or params["emissive_intensity_range"] (pair of floats).

• env (EmbodiedEnv) – The environment instance.

Raises:

ValueError – If both HDR and emissive params are provided, or if neither is provided.

embodichain.lab.gym.envs.managers.randomization.visual.randomize_light(env, env_ids, entity_cfg, position_range=None, color_range=None, intensity_range=None, direction_range=None)

Randomize light properties by adding, scaling, or setting random values.

This function allows randomizing light properties in the scene. The function samples random values from the given distribution parameters and adds, scales, or sets the values into the physics simulation based on the operation.

The distribution parameters are lists of two elements each, representing the lower and upper bounds of the distribution for the x, y, and z components of the light properties. The function samples random values for each component independently.

Attention

This function applied the same light properties for all the environments.

position_range is the x, y, z value added into light’s cfg.init_pos. color_range is the absolute r, g, b value set to the light object. intensity_range is the value added into light’s cfg.intensity. direction_range is the x, y, z value added into light’s cfg.direction. (Only applicable for "sun", "direction", "spot", "rect", and "mesh" light types.)

Tip

This function uses CPU tensors to assign light properties.

Warning

position_range is ignored for global scene lights ("sun", "direction") because they are infinite-distance lights with no meaningful position. Use direction_range instead for these light types.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (Union[torch.Tensor, None]) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• position_range (tuple[list[float], list[float]] | None) – The range for the position randomization.

• color_range (tuple[list[float], list[float]] | None) – The range for the color randomization.

• intensity_range (tuple[float, float] | None) – The range for the intensity randomization.

• direction_range (tuple[list[float], list[float]] | None) – The range for the direction randomization. Only applicable for directional light types ("sun", "direction", "spot", "rect", "mesh").

Return type:

None

class embodichain.lab.gym.envs.managers.randomization.visual.randomize_visual_material

Randomize the visual material properties of a RigidObject or Articulation.

Supported properties are base color, base-color texture, metallic factor, roughness, and index of refraction. Textures from texture_path are preloaded during initialization. The default ground plane can be selected with entity_cfg.uid="default_plane".

By default, the functor retains the asset’s original material instances and creates working instances from their existing templates. Set fallback_to_new=True to force the legacy new-material path.

Methods:

__init__(cfg, env)	Initialize the term.
gen_random_base_color_texture(width, height)	Generate a random base color texture.

__init__(cfg, env)

Initialize the term.

Parameters:

• cfg (FunctorCfg) – The configuration of the functor.

• env (EmbodiedEnv) – The environment instance.

Raises:

ValueError – If the asset is not a RigidObject or an Articulation.

static gen_random_base_color_texture(width, height)

Generate a random base color texture.

Parameters:

• width (int) – The width of the texture.

• height (int) – The height of the texture.

Return type:

Tensor

Returns:

A torch tensor representing the random base color texture with shape (height, width, 4).

embodichain.lab.gym.envs.managers.randomization.visual.set_rigid_object_group_visual_material(env, env_ids, entity_cfg, mat_cfg)

Set a rigid object group’s visual material (deterministic, non-random).

This helper exists to support configs that want fixed colors/materials during reset.

Parameters:

• env (EmbodiedEnv) – Environment instance.

• env_ids (Tensor | None) – Target env ids. If None, applies to all envs.

• entity_cfg (SceneEntityCfg) – Scene entity config (must point to a rigid object).

• mat_cfg (Union[VisualMaterialCfg, Dict]) – Visual material configuration. Can be a VisualMaterialCfg object or a dict. If a dict is provided, it will be converted to VisualMaterialCfg using from_dict(). If uid is not specified in mat_cfg, it will default to “{entity_uid}_mat”.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.visual.set_rigid_object_visual_material(env, env_ids, entity_cfg, mat_cfg)

Set a rigid object’s visual material (deterministic, non-random).

This helper exists to support configs that want fixed colors/materials during reset.

Parameters:

• env (EmbodiedEnv) – Environment instance.

• env_ids (Optional[Tensor]) – Target env ids. If None, applies to all envs.

• entity_cfg (SceneEntityCfg) – Scene entity config (must point to a rigid object).

• mat_cfg (Union[VisualMaterialCfg, Dict]) – Visual material configuration. Can be a VisualMaterialCfg object or a dict. If a dict is provided, it will be converted to VisualMaterialCfg using from_dict(). If uid is not specified in mat_cfg, it will default to “{entity_uid}_mat”.

Return type:

None

Spatial

Functions:

get_random_pose(init_pos, init_rot[, ...])	Generate a random pose based on the initial position and rotation.
randomize_articulation_root_pose(env, ...[, ...])	Randomize the root pose of an articulation in the environment.
randomize_rigid_object_pose(env, env_ids, ...)	Randomize the pose of a rigid object in the environment.
randomize_robot_eef_pose(env, env_ids, ...)	Randomize the initial end-effector pose of a robot in the environment.
randomize_robot_qpos(env, env_ids, entity_cfg)	Randomize the initial joint positions of a robot in the environment.
randomize_target_pose(env, env_ids, ...[, ...])	Randomize a virtual target pose and store in env state.

Classes:

planner_grid_cell_sampler	Sample grid cells for object placement without replacement.
randomize_anchor_height	Randomize the height of an anchor object and shift other objects by the same delta.
sample_rigid_object_pose_from_workspace	Place a rigid object at a position sampled from a robot workspace.

embodichain.lab.gym.envs.managers.randomization.spatial.get_random_pose(init_pos, init_rot, position_range=None, rotation_range=None, relative_position=True, relative_rotation=False)

Generate a random pose based on the initial position and rotation.

Parameters:

• init_pos (torch.Tensor) – The initial position tensor of shape (num_instance, 3).

• init_rot (torch.Tensor) – The initial rotation tensor of shape (num_instance, 3, 3).

• position_range (tuple[list[float], list[float]] | None) – The range for the position randomization.

• rotation_range (tuple[list[float], list[float]] | None) – The range for the rotation randomization. The rotation is represented as Euler angles (roll, pitch, yaw) in degree.

• relative_position (bool) – Whether to randomize the position relative to the initial position. Default is True.

• relative_rotation (bool) – Whether to randomize the rotation relative to the initial rotation. Default is False.

Returns:

The generated random pose tensor of shape (num_instance, 4, 4).

Return type:

torch.Tensor

class embodichain.lab.gym.envs.managers.randomization.spatial.planner_grid_cell_sampler

Sample grid cells for object placement without replacement.

This functor divides a planar region into a regular 2D grid and samples cells to place objects. Each sampled cell will be marked as occupied and will not be resampled until the grid is reset.

The sampler places objects at the center of selected grid cells, with the z-position set to a reference height.

Methods:

__init__(cfg, env)	Initialize the GridCellSampler functor.
reset([env_ids])	Reset the grid sampling state.

__init__(cfg, env)

Initialize the GridCellSampler functor.

Parameters:

• cfg (FunctorCfg) – The configuration of the functor.

• env (EmbodiedEnv) – The environment instance.

reset(env_ids=None)

Reset the grid sampling state.

Parameters:

env_ids (Tensor | None) – The environment IDs to reset. If None, resets all environments.

Return type:

None

class embodichain.lab.gym.envs.managers.randomization.spatial.randomize_anchor_height

Randomize the height of an anchor object and shift other objects by the same delta.

This functor samples a per-environment height delta, moves the anchor object relative to its configured initial position, and adds the same delta to the Z component of every other included object while preserving XY and rotation.

The functor is configured through FunctorCfg parameters, following the same pattern as planner_grid_cell_sampler.

Methods:

__init__(cfg, env)

Initialize the functor.

__init__(cfg, env)

Initialize the functor.

Parameters:

• cfg (FunctorCfg) – The functor configuration.

• env (EmbodiedEnv) – The environment instance.

embodichain.lab.gym.envs.managers.randomization.spatial.randomize_articulation_root_pose(env, env_ids, entity_cfg, position_range=None, rotation_range=None, relative_position=True, relative_rotation=False, physics_update_step=-1)

Randomize the root pose of an articulation in the environment.

This function randomizes the position and/or rotation of an articulation’s root link. The articulation’s root is the base frame that all other links are attached to.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• position_range (tuple[list[float], list[float]] | None) – The range for the position randomization. Format: [[x_min, y_min, z_min], [x_max, y_max, z_max]].

• rotation_range (tuple[list[float], list[float]] | None) – The range for the rotation randomization. The rotation is represented as Euler angles (roll, pitch, yaw) in degrees.

• relative_position (bool) – Whether to randomize the position relative to the articulation’s initial position. Default is True.

• relative_rotation (bool) – Whether to randomize the rotation relative to the articulation’s initial rotation. Default is False.

• physics_update_step (int) – The number of physics update steps to apply after randomization. Default is -1 (no update).

Return type:

None

Note

This function is similar to randomize_rigid_object_pose() but operates on articulations (multi-link rigid body systems) rather than single rigid objects.

embodichain.lab.gym.envs.managers.randomization.spatial.randomize_rigid_object_pose(env, env_ids, entity_cfg, position_range=None, rotation_range=None, relative_position=True, relative_rotation=False, physics_update_step=-1)

Randomize the pose of a rigid object in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• position_range (tuple[list[float], list[float]] | None) – The range for the position randomization.

• rotation_range (tuple[list[float], list[float]] | None) – The range for the rotation randomization. The rotation is represented as Euler angles (roll, pitch, yaw) in degree.

• relative_position (bool) – Whether to randomize the position relative to the object’s initial position. Default is True.

• relative_rotation (bool) – Whether to randomize the rotation relative to the object’s initial rotation. Default is False.

• physics_update_step (int) – The number of physics update steps to apply after randomization. Default is -1 (no update).

Return type:

None

embodichain.lab.gym.envs.managers.randomization.spatial.randomize_robot_eef_pose(env, env_ids, entity_cfg, position_range=None, rotation_range=None)

Randomize the initial end-effector pose of a robot in the environment.

Note

• The position and rotation are performed randomization in a relative manner.

• The current state of eef pose is computed based on the current joint positions of the robot.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the randomization.

• robot_name (str) – The name of the robot.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• position_range (tuple[list[float], list[float]] | None) – The range for the position randomization.

• rotation_range (tuple[list[float], list[float]] | None) – The range for the rotation randomization. The rotation is represented as Euler angles (roll, pitch, yaw) in degree.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.spatial.randomize_robot_qpos(env, env_ids, entity_cfg, qpos_range=None, relative_qpos=True, joint_ids=None)

Randomize the initial joint positions of a robot in the environment.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the randomization.

• entity_cfg (SceneEntityCfg) – The configuration of the scene entity to randomize.

• qpos_range (tuple[list[float], list[float]] | None) – The range for the joint position randomization.

• relative_qpos (bool) – Whether to randomize the joint positions relative to the current joint positions. Default is True.

• joint_ids (List[int] | None) – The list of joint IDs to randomize. If None, all joints will be randomized.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.spatial.randomize_target_pose(env, env_ids, position_range, rotation_range=None, relative_position=False, relative_rotation=False, reference_entity_cfg=None, store_key='target_pose')

Randomize a virtual target pose and store in env state.

This function generates random target poses without requiring a physical object in the scene. The generated poses are stored as a public attribute in env for use by observations and rewards.

Parameters:

• env (EmbodiedEnv) – The environment instance.

• env_ids (torch.Tensor | None) – The environment IDs to apply the randomization.

• position_range (tuple[list[float], list[float]]) – The range for the position randomization.

• rotation_range (tuple[list[float], list[float]] | None) – The range for the rotation randomization. The rotation is represented as Euler angles (roll, pitch, yaw) in degree.

• relative_position (bool) – Whether to randomize the position relative to a reference entity. Default is False.

• relative_rotation (bool) – Whether to randomize the rotation relative to a reference entity. Default is False.

• reference_entity_cfg (SceneEntityCfg | None) – The reference entity for relative randomization. If None and relative mode is True, uses world origin.

• store_key (str) – The key to store the target pose in env state. Default is “target_pose”. The pose will be stored as a public attribute env.{store_key}.

Return type:

None

class embodichain.lab.gym.envs.managers.randomization.spatial.sample_rigid_object_pose_from_workspace

Place a rigid object at a position sampled from a robot workspace.

The workspace supplies kinematically reachable end-effector samples. This functor consumes their Cartesian positions for object placement while leaving scene collision and motion-planning validation to task-specific logic.

Methods:

__init__(cfg, env)

Initialize the workspace object sampler.

__init__(cfg, env)

Initialize the workspace object sampler.

Parameters:

• cfg (FunctorCfg) – Functor configuration.

• env (EmbodiedEnv) – Environment instance.

Geometry

Functions:

randomize_rigid_object_body_scale(env, ...)	Deprecated.
randomize_rigid_object_scale(env, env_ids, ...)	Randomize a rigid object's body scale factors (multiplicative, not absolute size).
randomize_rigid_objects_scale(env, env_ids, ...)	Randomize body scale factors for multiple rigid objects.

embodichain.lab.gym.envs.managers.randomization.geometry.randomize_rigid_object_body_scale(env, env_ids, entity_cfg, scale_range=None, same_scale_all_axes=True)

Deprecated. Use randomize_rigid_object_scale + scale_factor_range.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.geometry.randomize_rigid_object_scale(env, env_ids, entity_cfg, scale_factor_range=None, same_scale_all_axes=True)

Randomize a rigid object’s body scale factors (multiplicative, not absolute size).

Parameters:

• env (EmbodiedEnv) – Environment instance.

• env_ids (Optional[Tensor]) – Target env ids. If None, applies to all envs.

• entity_cfg (SceneEntityCfg) – Scene entity config of the rigid object.

• scale_factor_range (tuple[list[float], list[float]] | None) – If same_scale_all_axes is True, should be [[s_min], [s_max]]. Otherwise [[sx_min, sy_min, sz_min], [sx_max, sy_max, sz_max]].

• same_scale_all_axes (bool) – Whether to use same factor on x/y/z.

Return type:

None

embodichain.lab.gym.envs.managers.randomization.geometry.randomize_rigid_objects_scale(env, env_ids, entity_cfgs, scale_factor_range=None, same_scale_all_axes=True, shared_sample=False)

Randomize body scale factors for multiple rigid objects.

Parameters:

• env (EmbodiedEnv) – Environment instance.

• env_ids (Optional[Tensor]) – Target env ids. If None, applies to all envs.

• entity_cfgs (List[SceneEntityCfg]) – List of scene entity configs (rigid objects).

• scale_factor_range (tuple[list[float], list[float]] | None) – Scale factor sampling range.

• same_scale_all_axes (bool) – Whether to use same factor on x/y/z.

• shared_sample (bool) – If True, sample one scale per-env and apply to all objects (sync). If False, each object samples its own scales independently.

Return type:

None