unilab.envs.manager_based_rl_env

Community-compatible Manager-Based lifecycle with a Torch public contract.

Functions

make_manager_based_rl_env(cfg[, num_envs, ...])

Construct the generic Registry-owned Manager-Based production runtime.

Classes

ManagerBasedRLEnv

alias of ManagerBasedRlEnv

ManagerBasedRLEnvCfg

alias of ManagerBasedRlEnvCfg

ManagerBasedRlEnv

Manager-Based runtime with a Torch public lifecycle.

ManagerBasedRlEnvCfg

Configuration for the Manager-Based Torch environment.
unilab.envs.manager_based_rl_env.ManagerBasedRLEnv

alias of ManagerBasedRlEnv

unilab.envs.manager_based_rl_env.ManagerBasedRLEnvCfg

alias of ManagerBasedRlEnvCfg

class unilab.envs.manager_based_rl_env.ManagerBasedRlEnv[source]

Bases: TorchEnv

Manager-Based runtime with a Torch public lifecycle.

P1 keeps Manager terms, Entity state, reset transactions, and recorder scratch on an explicit NumPy host boundary. Public action/reset inputs, backend control, state, observations, reward, termination, truncation, and final observations are Torch tensors. This is a migration boundary, not a durable NumPy fallback: unsupported tensor capabilities fail closed.

Parameters:
is_vector_env = True
event_manager: EventManager
command_manager: CommandManager | NullCommandManager
action_manager: ActionManager
observation_manager: ObservationManager
termination_manager: TerminationManager
reward_manager: RewardManager
curriculum_manager: CurriculumManager | NullCurriculumManager
metrics_manager: MetricsManager | NullMetricsManager
recorder_manager: RecorderManager | NullRecorderManager
__init__(cfg, backend, num_envs)[source]
Parameters:
tensor_reset_pose_bounds: torch.Tensor | None
tensor_reset_velocity_bounds: torch.Tensor | None
tensor_reset_bounds_key: tuple[Any, ...] | None
obs_buf: dict[str, torch.Tensor]
extras: dict[str, Any]
event_term_mask_buffer: ResetMaskBuffer
property physics_dt: float
property step_dt: float
property max_episode_length_s: float
property max_episode_length: int
property obs_groups_spec: dict[str, int]

Map from observation group name to its dimension.

property action_space: Space

Action space

property unwrapped: ManagerBasedRlEnv
get_playback_debug_overlays()[source]

Aggregate playback overlays from command terms that provide one.

Terms opt in by implementing playback_debug_overlay_getter(), which returns a per-frame getter following the unisim.backend.base.DebugOverlayGetter contract. The returned getter merges primitives per env across all providing terms. Returns None when no term provides an overlay.

Return type:

DebugOverlayGetter | None

init_state()[source]

Initialize environment and return initial state

Return type:

TorchEnvState

step(actions)[source]

Step the environment with owner-declared action carrier.

The abstract base stays runtime-neutral. Concrete owners validate the declared carrier; the Torch lifecycle accepts Torch tensors.

Parameters:

actions (Tensor)

Return type:

TorchEnvState

apply_action(actions, state)[source]

Transform an action tensor into a backend control tensor.

Parameters:
Return type:

Tensor

update_state(state)[source]

Compute tensor observations, reward, and termination state.

Parameters:

state (TorchEnvState)

Return type:

TorchEnvState

publish_reward_log()[source]

Publish the latest device reward means as reward/<term> entries.

Return type:

dict[str, float]

property algo_capabilities: Any

Expose cold-path metadata to algorithm owners.

Per-dimension action bounds and action-order joint names are reserved fields. They intentionally stay unset until an algorithm consumes them; the Manager-Based action owner, not the RL runtime, owns their source.

reset(env_ids=None, *, seed=None, options=None)[source]

Execute the selected-row (env_ids) Manager reset lifecycle.

Parameters:
Return type:

tuple[dict[str, Tensor], dict[str, Any]]

get_observations()[source]
Return type:

dict[str, Tensor]

set_episode_length_buf(values)[source]

Overwrite per-env episode counters (cold path, runner-init only).

episode_length_buf mirrors state.info["steps"]: each step writes info["steps"] + 1 into the buffer and reset() zeroes both, so a direct assignment must update the two together. RL runners (e.g. RSL-RL init_at_random_ep_len) call this once before learning starts to stagger initial episode lengths across envs.

Parameters:

values (Tensor)

Return type:

None

seed(seed=-1)[source]
Parameters:

seed (int)

Return type:

int

import_training_state(state)[source]

Restore the authoritative counter and its manager-derived counters.

Parameters:

state (Mapping[str, Any])

Return type:

None

close()[source]

Clean up environment resources

Return type:

None

class unilab.envs.manager_based_rl_env.ManagerBasedRlEnvCfg[source]

Bases: EnvCfg

Configuration for the Manager-Based Torch environment.

Production task owners declare these fields in Hydra. The Registry materializes them into this plain typed config on the cold path; Python factories do not mirror task-specific manager or term declarations.

Parameters:
observations: dict[str, ObservationGroupCfg | None]
actions: dict[str, ActionTermCfg | None]
events: dict[str, EventTermCfg | None]
rewards: dict[str, RewardTermCfg | None]
terminations: dict[str, TerminationTermCfg | None]
commands: dict[str, CommandTermCfg | None]
curriculum: dict[str, CurriculumTermCfg | None]
metrics: dict[str, MetricsTermCfg | None]
recorders: dict[str, RecorderTermCfg | None]
reset_owners: dict[str, ResetOwnerCfg | None]
collector_tensor_device: str | None = None
seed: int | None = None
is_finite_horizon: bool = False
auto_reset: bool = True
scale_rewards_by_dt: bool = True
policy_observation_group: str = 'policy'
critic_observation_group: str | None = None
validate()[source]

validate the config

Return type:

None

__init__(scene=None, fixed_model_variants=None, sim_dt=0.01, max_episode_seconds=None, ctrl_dt=0.01, render_spacing=1.0, render_offset_mode='grid', drake_backend_mode='batch', drake_nthread=0, superdex_num_workers=0, superdex_execution_mode='batch', superdex_assets_root=None, superdex_effort_limits=None, superdex_allow_contact_approximation=False, motrix_max_iterations=None, cpu_ids=None, mjwarp_nconmax=None, mjwarp_njmax=None, newton_device=None, newton_nconmax=None, newton_njmax=None, newton_capacity_check_steps=1, newton_use_cuda_graph=True, isaacgym_device_id=None, isaacgym_worker_timeout_s=None, genesis_device_id=None, genesis_integrator=None, genesis_constraint_solver=None, genesis_friction_cone=None, genesis_solver_iterations=None, isaacsim_device_id=None, isaacsim_worker_timeout_s=None, isaacsim_render_mode=None, isaacsim_render_width=1280, isaacsim_render_height=720, isaacsim_tensor_cuda_ipc=False, isaacsim_share_friction_materials=False, isaacsim_env_spacing=None, observations=<factory>, actions=<factory>, events=<factory>, rewards=<factory>, terminations=<factory>, commands=<factory>, curriculum=<factory>, metrics=<factory>, recorders=<factory>, reset_owners=<factory>, collector_tensor_device=None, seed=None, is_finite_horizon=False, auto_reset=True, scale_rewards_by_dt=True, policy_observation_group='policy', critic_observation_group=None)
Parameters:
unilab.envs.manager_based_rl_env.make_manager_based_rl_env(cfg, num_envs=1, backend_type='mujoco')[source]

Construct the generic Registry-owned Manager-Based production runtime.

Parameters:
Return type:

ManagerBasedRlEnv