embodichain.lab.gym.utils#

Environment utilities: the registration system (register_env, make), Gymnasium integration helpers, miscellaneous utilities, and EnvProfiler for step/reset timing.

Overview#

Utilities for the environment framework: the registration system (register_env() decorator, make() factory, and the EnvSpec/TimeLimitWrapper helpers), Gymnasium integration helpers, miscellaneous environment utilities, and the EnvProfiler for per-step / per-reset and per-functor timing.

Registration System#

class embodichain.lab.gym.utils.registration.EnvSpec[source]#

Bases: object

Methods:

__init__(uid, cls[, max_episode_steps, ...])

A specification for a Embodied environment.

Attributes:

gym_spec

Return a gym EnvSpec for this env

__init__(uid, cls, max_episode_steps=None, default_kwargs=None, task_program_registration=None, task_program_adapter_factory=None, supports_rl=False)[source]#

A specification for a Embodied environment.

property gym_spec#

Return a gym EnvSpec for this env

embodichain.lab.gym.utils.registration.register(name, cls, max_episode_steps=None, default_kwargs=None, task_program_registration=None, task_program_adapter_factory=None, supports_rl=False)[source]#

Register a Embodied environment.

embodichain.lab.gym.utils.registration.register_env(uid, max_episode_steps=None, override=False, *, supports_rl=False, task_program_registration=None, task_program_adapter_factory=None, **kwargs)[source]#

A decorator to register Embodied environments.

Parameters:
  • uid (str) – unique id of the environment.

  • max_episode_steps (int) – maximum number of steps in an episode.

  • override (bool) – whether to override the environment if it is already registered.

  • supports_rl (bool) – Whether the environment has a supported RL training path.

Notes

  • max_episode_steps is processed differently from other keyword arguments in gym. gym.make wraps the env with gym.wrappers.TimeLimit to limit the maximum number of steps.

  • gym.EnvSpec uses kwargs instead of **kwargs!

embodichain.lab.gym.utils.registration.make(env_id, **kwargs)[source]#

Instantiate a Embodied environment.

Parameters:
  • env_id (str) – Environment ID.

  • as_gym (bool, optional) – Add TimeLimit wrapper as gym.

  • **kwargs – Keyword arguments to pass to the environment.

embodichain.lab.gym.utils.registration.get_env_spec(env_id)[source]#

Return one registered environment specification or fail closed.

Return type:

EnvSpec

embodichain.lab.gym.utils.registration.build_env(env_id, base_env_cfg)[source]#

Create an environment from a registered env id.

A thin convenience wrapper around make() that deep-copies the base config so callers can safely mutate the resulting environment’s cfg without affecting shared defaults. This helper used to live in the task package; it now lives with the registry so that core code paths such as RL training do not need to depend on an official task package.

Parameters:
  • env_id (str) – Registered environment id (see register_env()).

  • base_env_cfg (EmbodiedEnvCfg) – Base environment configuration to instantiate with.

Returns:

The instantiated environment.

embodichain.lab.gym.utils.registration.make_vec(env_id, **kwargs)[source]#
embodichain.lab.gym.utils.registration.register_env_function(cls, uid, override=False, max_episode_steps=None, *, supports_rl=False, task_program_registration=None, task_program_adapter_factory=None, **kwargs)[source]#
class embodichain.lab.gym.utils.registration.TimeLimitWrapper[source]#

Bases: Wrapper

like the standard gymnasium timelimit wrapper but fixes truncated variable to be a batched array

Methods:

__init__(env, max_episode_steps)

Wraps an environment to allow a modular transformation of the step() and reset() methods.

step(action)

Uses the step() of the env that can be overwritten to change the returned data.

__init__(env, max_episode_steps)[source]#

Wraps an environment to allow a modular transformation of the step() and reset() methods.

Parameters:

env (Env) – The environment to wrap

step(action)[source]#

Uses the step() of the env that can be overwritten to change the returned data.

embodichain.lab.gym.utils.registration.REGISTERED_ENVS: Dict[str, EnvSpec] = {'EmbodiedEnv-v1': <embodichain.lab.gym.utils.registration.EnvSpec object>}#

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:

d = {} for k, v in iterable:

d[k] = v

dict(**kwargs) -> new dictionary initialized with the name=value pairs

in the keyword argument list. For example: dict(one=1, two=2)

Task Package Discovery#

embodichain.lab.gym.utils.registration.discover_task_packages()[source]#

Import all registered task packages via embodichain.tasks entry_points.

Each task package recursively imports its task modules, which triggers @register_envgym.register(). After this call, all tasks from all installed packages are available in gymnasium’s global registry.

Return type:

list[str]

Returns:

List of entry point names that were successfully imported.

embodichain.lab.gym.utils.registration.execute_init_hooks()[source]#

Execute all registered init hooks via embodichain.init entry_points.

Hooks are called in entry_points declaration order. An exception from one hook does not prevent others from executing.

Each entry point value must be in the format "module.path:function_name". The function must accept no arguments and return None.

Return type:

list[str]

Returns:

List of hook names that were executed successfully.

Utility Modules#

Gymnasium Utilities#

Miscellaneous#

Profiling#

Backward-compatible environment aliases for the simulation profiler.

The implementation lives in embodichain.lab.sim.profiler so standalone simulation code and Gym environments can share the same profiler instance.