Environment Interface#

This section provides the key APIs of the environment in the RLinf framework, specifically designed for embodied intelligence scenarios. As an example, we demonstrate the env setup using LIBERO as the background. For any other simulator, the process is analogous.

EnvWorker#

class rlinf.workers.env.env_worker.EnvWorker#
__init__(cfg)#

Initialize the Worker with the given parent address and world size.

Only non-Ray workers should provide parent_address, world_size and rank. For example, when a Worker is created via multiprocessing by another Worker, the parent address, world size and rank should be provided.

Parameters:

  • parent_address (Optional[WorkerAddress]) – The address of the parent worker. This is used to set up the WorkerAddress for this worker.

  • world_size (Optional[int]) – The total number of workers in the group. If not provided, it will be set to the environment variable WORLD_SIZE.

  • rank (Optional[int]) – The rank of this worker in the group. If not provided, it will be set to the environment variable RANK.

  • cfg (DictConfig)

set_global_step(global_step)#

Set the trajectory step used to correlate distributed events.

Parameters:

global_step (int)

Return type:

None

env_interact_step(chunk_actions, stage_id)#

Execute a chunk and build environment output and LeRobot payload.

obs_list in the payload stays post-action. The caller snapshots the observation from before this chunk for LeRobot action pairing.

Parameters:

  • chunk_actions (Tensor)

  • stage_id (int)

Return type:

tuple[EnvOutput, dict[str, Any], dict[str, Any]]

env_evaluate_step(raw_actions, stage_id)#

This function is used to evaluate the environment.

Parameters:

  • raw_actions (Tensor)

  • stage_id (int)

Return type:

tuple[EnvOutput, dict[str, Any]]

prefetch_train_bootstrap(rollout_channel)#

Prepare and send the first env batch for the next training rollout.

Parameters:

rollout_channel (Channel)

Return type:

None

Environment#

class rlinf.envs.sim.libero.libero_env.LiberoEnv#

Batched LIBERO simulation with task scheduling and episode-local history.

__init__(cfg, num_envs, seed_offset, total_num_processes, worker_info)#
Parameters:

  • cfg (DictConfig)

  • num_envs (int)

  • seed_offset (int)

  • total_num_processes (int)

  • worker_info (WorkerInfo)

Return type:

None

get_env_fns()#

Return simulator constructors for the current task assignments.

Return type:

list[Callable[[], Env]]

get_env_fn_params(env_idx=None)#

Build simulator arguments and update prompts for selected environments.

Parameters:

env_idx (list[int] | ndarray | None)

Return type:

list[dict[str, Any]]

update_reset_state_ids()#

Assign one reset state per group, shared by its environments.

Return type:

None

get_reset_state_ids_all()#

Distribute reset states across environment processes.

Return type:

ndarray

reset(env_idx=None, reset_state_ids=None)#

Reset selected environments and return observations for the full batch.

Parameters:

  • env_idx (list[int] | ndarray | None)

  • reset_state_ids (list[int] | ndarray | None)

Return type:

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

get_camera_meta(camera_name='agentview', height=256, width=256)#

Fetch camera intrinsics/extrinsics and depth planes.

Returns camera calibration from worker 0’s robosuite sim: intrinsic matrix, cam-to-world transform, and depth near/far. The agentview camera is fixed in the world, so this is constant per episode.

Parameters:

  • camera_name (str)

  • height (int)

  • width (int)

Return type:

dict[str, Any]

render_camera(camera_name='agentview', height=1024, width=1024, depth=False)#

Render an arbitrary camera at the requested resolution.

Returns:

The rendered RGB image, or a (rgb, depth) tuple when depth is True.

Parameters:

  • camera_name (str)

  • height (int)

  • width (int)

  • depth (bool)

Return type:

ndarray | tuple[ndarray, ndarray]

step(actions=None, auto_reset=True, _skip_obs_wrap=False, *, _include_history=True)#

Step the environment with the given actions.

Parameters:

  • actions (ndarray | Tensor | None)

  • auto_reset (bool)

  • _skip_obs_wrap (bool)

  • _include_history (bool)

Return type:

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

chunk_step(chunk_actions)#

Execute actions shaped [num_envs, chunk_steps, action_dim].

Parameters:

chunk_actions (ndarray | Tensor)

Return type:

tuple[list[dict[str, Any] | None], Tensor, Tensor, Tensor, list[dict[str, Any]]]