internutopia.core.env#

gym_env#

vec_env#

class internutopia.core.vec_env.Env(config: Config)[source]#

Vectorized environments to run multiple independent environments simultaneously with multiple agents.

NOT gymnasium compatible.

Parameters:

config (Config) – The config instance used for simulation management.

property active_task_configs#

Get active task configs with env id as key.

close()[source]#

Close the environment

finished() → bool[source]#

Check if all tasks are finished

get_dt()[source]#

Get dt of simulation environment.

get_observations() → List | Any[source]#

Get observations from Isaac environment

property is_render#

Get render state.

reset(env_ids: List[int] = None) → Tuple[List, List][source]#

Resets the environments specified by the given environment IDs and returns the initial observations and task configs. If no environment IDs are provided, all environments will be reset. If no tasks are running after the reset, a log message is generated, and empty lists are returned.

Parameters:

env_ids (List[int]) – A list of environment IDs to reset. If None, all environments will be reset.

Returns:

A tuple containing two lists: the initial observations and task configs for the reset environments. If no tasks are running, both lists will be empty.

Return type:

Tuple[List, List]

property runner#

The runner property provides access to the internal runner instance.

property simulation_app#

Simulation app instance

step(action: List[Dict | OrderedDict]) → Tuple[List, List, List, List, List][source]#

Perform a single step in the environment using the provided actions.

This method takes a list of actions, validates its structure and length, then delegates the step execution to the internal runner. It computes and returns the observations, rewards, termination status, truncation status, and additional information.

Parameters:

action (List[Union[Dict, OrderedDict]]) – A list of actions to be executed in the environment. Each action is either a dictionary or an ordered dictionary.

Returns:

A tuple containing the following elements:

  • obs (List): The observations resulting from the actions.

  • reward (List): The rewards obtained from the actions.

  • terminated (List): The termination status of the environments.

  • truncated (List): The truncation status of the environments.

  • info (List): Additional information about the step execution.

Return type:

Tuple[List, List, List, List, List]

warm_up(steps: int = 10, render: bool = True, physics: bool = True)[source]#

Warm up the env by running a specified number of steps.

Parameters:

  • steps (int) – The number of warm-up steps to perform. Defaults to 10.

  • render (bool) – Whether to render the scene during warm-up. Defaults to True.

  • physics (bool) – Whether to enable physics during warm-up. Defaults to True.

runner#

class internutopia.core.runner.SimulatorRunner(config: Config, task_config_manager: BaseTaskConfigManager)[source]#
clear_single_task(task_name: str)[source]#

Clear single task with task_name

Parameters:

task_name (str) – Task name to clear.

get_obs() → List[Dict | None][source]#

Get obs :returns: obs from isaac sim. :rtype: List[Dict]

reset(env_ids: List[int] | None = None) → Tuple[List, List][source]#

Resets the environment for the given environment IDs or initializes it if no IDs are provided. This method handles resetting the simulation context, generating new task configs, and finalizing tasks when necessary. It supports partial resets for specific environments and ensures proper handling of task transitions.

Parameters:

env_ids (Optional[List[int]]) – A list of environment IDs to reset. If None, all environments are reset or initialized based on the current state.

Returns:

A tuple containing two lists. The first list contains observations for

the reset environments, and the second list contains the new task configs.

Return type:

Tuple[List, List]

Raises:

  • ValueError – If the provided env_ids are invalid or don’t correspond to any existing tasks.

  • RuntimeError – If the simulation context fails to reset or initialize properly.

Notes

  • If env_ids is None and there are current tasks, all environments are reset.

  • If env_ids is None and there are no current tasks, the simulation context is initialized.

  • Observations are collected only for environments that are reset or initialized.

  • Tasks corresponding to the reset environments are transitioned to new episodes.

step(actions: List[Dict] | None = None, render: bool = True) → Tuple[List[Dict], List[bool], List[float]][source]#

Step function to advance the simulation environment by one time step.

This method processes actions for active tasks, steps the simulation world, collects observations, updates metrics, and determines task terminations. It also handles rendering based on specified intervals.

Parameters:

  • actions (Union[List[Dict], None], optional) – A dictionary mapping task names to another dictionary of robot names and their respective actions. If None, no actions are applied. Defaults to None.

  • render (bool, optional) – Flag indicating whether to render the simulation at this step. True triggers rendering if the render interval is met. Defaults to True.

Returns:

  • obs (Dict): A dictionary containing observations for each task, further divided by robot names and their observation data.

  • terminated_status (Dict[str, bool]): A dictionary mapping task names to boolean values indicating whether the task has terminated.

  • reward (Dict[str, float]): A dictionary that would contain rewards for each task or robot; however, the actual return and computation of rewards is not shown in the provided code snippet.

Return type:

Tuple[Dict, Dict[str, bool], Dict[str, float]]

Raises:

Exception – If an error occurs when applying an action to a robot, the exception is logged and re-raised, providing context about the task, robot, and current tasks state.

Notes

  • The _world.step() method advances the simulation, optionally rendering the environment based on the render flag and the render interval.

  • get_obs() is a method to collect observations from the simulation world, though its implementation details are not shown.

  • Metrics for each task are updated, and upon task completion, results are saved to a JSON file. This includes a flag ‘normally_end’ set to True, which seems to indicate normal termination of the task.

  • The function also manages a mechanism to prevent further action application and metric updates for tasks that have been marked as finished.

stop()[source]#

Stop all current operations and clean up the World

warm_up(steps: int = 10, render: bool = True, physics: bool = True)[source]#

Warm up the simulation by running a specified number of steps.

Parameters:

  • steps (int) – The number of warm-up steps to perform. Defaults to 10.

  • render (bool) – Whether to render the scene during warm-up. Defaults to True.

  • physics (bool) – Whether to enable physics during warm-up. Defaults to True.

Raises:

ValueError – If both render and physics are set to False, or if steps is less than or equal to 0.