unilab.tasks.locomotion.g1.manager_terms.G1WalkManagerBasedEnv

class unilab.tasks.locomotion.g1.manager_terms.G1WalkManagerBasedEnv[source]

Bases: ManagerBasedRlEnv

Manager-Based G1 walk runtime.

Parameters:

Methods

__init__(cfg, backend, num_envs)

apply_action(actions, state)

Transform an action tensor into a backend control tensor.

capture_play_video_frame()

Capture one detached RGB video frame through the env contract.

close()

Clean up environment resources

export_training_state()

get_backend_runtime_diagnostics()

Serialize the backend's tensor runtime diagnostics on the cold path.

get_observations()

get_physics_state_layout()

Return the backend physics-state playback layout.

get_physics_state_snapshot()

Return a detached physics snapshot for playback/video export.

get_playback_debug_overlays()

Aggregate playback overlays from command terms that provide one.

get_playback_mocap_state([env_index])

Return detached (mocap_pos, mocap_quat) playback state.

get_playback_model([env_index])

Return the backend playback model for one vectorized environment.

get_scene_visual_model_file()

Return the backend visual model file on the playback cold path.

import_training_state(state)

Restore the authoritative counter and its manager-derived counters.

init_play_renderer([render_spacing, ...])

Initialize backend-native playback rendering when available.

init_state()

Initialize environment and return initial state

publish_reward_log()

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

render([mode])

Render the current state to an RGB array through the play renderer.

reset([env_ids, seed, options])

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

resolve_play_render_plan(*, ...)

Resolve high-level playback mode through the concrete backend.

run_playback(*, initialize, step, num_steps)

Execute playback through the concrete backend.

run_playback_mode(*, play_render_mode, ...)

Resolve configured playback mode and execute it through the backend contract.

seed([seed])

set_autoreset(enabled)

set_episode_length_buf(values)

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

set_nan_guard(guard)

Attach the owner-provided diagnostic guard.

step(actions)

Step the environment with owner-declared action carrier.

update_state(state)

Compute tensor observations, reward, and termination state.

Attributes

action_space

Action space

algo_capabilities

Expose cold-path metadata to algorithm owners.

backend

Public backend owner used by Manager terms and tensor contracts.

cfg

The configuration of the environment

device

is_vector_env

max_episode_length

max_episode_length_s

num_envs

return the size of the env if it is vectorized

obs_groups_spec

Map from observation group name to its dimension.

observation_space

Observation space

physics_dt

play_capabilities

Return env-facing play/render capabilities.

produces_completion_receipts

RL env-contract declaration of the backend's step receipts.

state

Current environment state (None before first reset)

step_dt

unwrapped

event_manager

command_manager

action_manager

observation_manager

termination_manager

reward_manager

curriculum_manager

metrics_manager

recorder_manager

tensor_reset_pose_bounds

tensor_reset_velocity_bounds

tensor_reset_bounds_key

event_term_mask_buffer

__init__(cfg, backend, num_envs)
Parameters:
property action_space: Space

Action space

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.

apply_action(actions, state)

Transform an action tensor into a backend control tensor.

Parameters:
Return type:

Tensor

property backend: SimBackend

Public backend owner used by Manager terms and tensor contracts.

capture_play_video_frame()

Capture one detached RGB video frame through the env contract.

Return type:

ndarray

property cfg: EnvCfg

The configuration of the environment

close()

Clean up environment resources

Return type:

None

property device: device
export_training_state()
Return type:

dict[str, Any]

get_backend_runtime_diagnostics()

Serialize the backend’s tensor runtime diagnostics on the cold path.

This is the collection boundary consumed by training runtimes (the unilab.rl off-policy collector publishes it into the run manifest). The payload is plain JSON: available and backend_type plus, when available, one diagnostics entry per backend optimization with requested/enabled/disable_reason. Backends that declare no optional tensor runtime optimization, or whose diagnostic probe fails, record an explicit unavailable marker instead of a null field.

Return type:

dict[str, Any]

get_observations()
Return type:

dict[str, Tensor]

get_physics_state_layout()

Return the backend physics-state playback layout.

Return type:

Any

get_physics_state_snapshot()

Return a detached physics snapshot for playback/video export.

Return type:

ndarray

get_playback_debug_overlays()

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

get_playback_mocap_state(env_index=0)

Return detached (mocap_pos, mocap_quat) playback state.

Parameters:

env_index (int)

Return type:

tuple[ndarray, ndarray]

get_playback_model(env_index=None)

Return the backend playback model for one vectorized environment.

Parameters:

env_index (int | None)

Return type:

Any

get_scene_visual_model_file()

Return the backend visual model file on the playback cold path.

Return type:

str | None

import_training_state(state)

Restore the authoritative counter and its manager-derived counters.

Parameters:

state (Mapping[str, Any])

Return type:

None

init_play_renderer(render_spacing=None, render_offset_mode=None, *, headless=False, capture=False, width=1280, height=720, camera_kwargs=None)

Initialize backend-native playback rendering when available.

Parameters:
Return type:

None

init_state()

Initialize environment and return initial state

Return type:

TorchEnvState

is_vector_env = True
property max_episode_length: int
property max_episode_length_s: float
property num_envs: int

return the size of the env if it is vectorized

property obs_groups_spec: dict[str, int]

Map from observation group name to its dimension.

property observation_space: Space

Observation space

property physics_dt: float
property play_capabilities: EnvPlayCapabilities

Return env-facing play/render capabilities.

property produces_completion_receipts: bool

RL env-contract declaration of the backend’s step receipts.

True when the bound backend publishes an event-backed async completion receipt on every step (replay CUDA ingress then fails closed without one). False for synchronous producers whose step outputs are already ordered at return, where no receipt exists and none is awaited. This is a cold-path declaration; step keeps publishing whatever receipt the backend returns.

publish_reward_log()

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

Return type:

dict[str, float]

render(mode='rgb_array')

Render the current state to an RGB array through the play renderer.

Parameters:

mode (str)

Return type:

ndarray

reset(env_ids=None, *, seed=None, options=None)

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

Parameters:
Return type:

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

resolve_play_render_plan(*, play_render_mode, play_steps, output_video)

Resolve high-level playback mode through the concrete backend.

Parameters:
Return type:

PlayRenderPlan

run_playback(*, initialize, step, num_steps, output_video=None, render_spacing=None, render_offset_mode=None, headless=None, record_video=None, frame_state_getter=None, camera_kwargs=None, debug_overlay_getter=None, on_frame=None)

Execute playback through the concrete backend.

Parameters:

  • initialize (Callable[[], Any])

  • step (Callable[[Any], Any])

  • num_steps (int | None)

  • output_video (str | PathLike[str] | None)

  • render_spacing (float | None)

  • render_offset_mode (str | None)

  • headless (bool | None)

  • record_video (bool | None)

  • frame_state_getter (Callable[[], np.ndarray] | None)

  • camera_kwargs (CameraCfg | Mapping[str, Any] | None)

  • debug_overlay_getter (DebugOverlayGetter | None)

  • on_frame (Callable[[int, np.ndarray], np.ndarray | None] | None)

Return type:

str | None

run_playback_mode(*, play_render_mode, play_steps, output_video, initialize, step, render_spacing=None, render_offset_mode=None, frame_state_getter=None, camera_kwargs=None, debug_overlay_getter=None, on_frame=None, on_plan=None)

Resolve configured playback mode and execute it through the backend contract.

Overlay gating follows the resolved plan mode: record playback requires play_capabilities.supports_debug_overlay (backends fail closed otherwise); interactive playback requires supports_interactive_debug_overlay — when the backend lacks it the getter is dropped with a warning so interactive playback still runs without overlays.

camera_kwargs is a record/offline camera profile. Some native interactive viewers (for example the Isaac family) fail closed when they receive one because the user owns the camera in the window; in that case the profile is dropped with a warning and playback retried without it, so interactive playback still runs.

Parameters:
Return type:

str | None

seed(seed=-1)
Parameters:

seed (int)

Return type:

int

set_autoreset(enabled)
Parameters:

enabled (bool)

Return type:

None

set_episode_length_buf(values)

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

set_nan_guard(guard)

Attach the owner-provided diagnostic guard.

Detection stays device-resident. Artifact export occurs only at the guard’s explicit abnormal-diagnostic host boundary.

Parameters:

guard (NanGuard)

Return type:

None

property state: TorchEnvState | None

Current environment state (None before first reset)

step(actions)

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

property step_dt: float
property unwrapped: ManagerBasedRlEnv
update_state(state)

Compute tensor observations, reward, and termination state.

Parameters:

state (TorchEnvState)

Return type:

TorchEnvState

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
tensor_reset_pose_bounds: torch.Tensor | None
tensor_reset_velocity_bounds: torch.Tensor | None
tensor_reset_bounds_key: tuple[Any, ...] | None
event_term_mask_buffer: ResetMaskBuffer