env_wrapper

EnvironmentWrapper

Bases: Wrapper, Registerable

Base class for all environment wrappers in OmniGibson. In general, reset(), step(), and observation_spec() should be overwritten

Parameters:

Name	Type	Description	Default
env	OmniGibsonEnv	The environment to wrap.	required

Source code in OmniGibson/omnigibson/envs/env_wrapper.py

class EnvironmentWrapper(Wrapper, Registerable):
    """
    Base class for all environment wrappers in OmniGibson. In general, reset(), step(), and observation_spec() should
    be overwritten

    Args:
        env (OmniGibsonEnv): The environment to wrap.
    """

    def __init__(self, env):
        self.env = env

        # Run super
        super().__init__(obj=env)

    def step(self, action, n_render_iterations=1):
        """
        By default, run the normal environment step() function

        Args:
            action (th.tensor): action to take in environment
            n_render_iterations (int): Number of rendering iterations to use before returning observations
        Returns:
            4-tuple:
                - (dict) observations from the environment
                - (float) reward from the environment
                - (bool) whether the current episode is terminated
                - (bool) whether the current episode is truncated
                - (dict) misc information
        """
        return self.env.step(action, n_render_iterations=n_render_iterations)

    def reset(self, **kwargs):
        """
        By default, run the normal environment reset() function. Keyword arguments (e.g. ``env_indices``
        for vectorized per-env reset, ``seed``, ``options``) are forwarded to the wrapped environment so
        the wrapper stays transparent to reset.

        Returns:
            dict: Environment observation space after reset occurs
        """
        return self.env.reset(**kwargs)

    def observation_spec(self):
        """
        By default, grabs the normal environment observation_spec

        Returns:
            dict: Observations from the environment
        """
        return self.env.observation_spec()

    @classproperty
    def _do_not_register_classes(cls):
        # Don't register this class since it's an abstract template
        classes = super()._do_not_register_classes
        classes.add("EnvironmentWrapper")
        return classes

    @classproperty
    def _cls_registry(cls):
        # Global robot registry
        global REGISTERED_ENV_WRAPPERS
        return REGISTERED_ENV_WRAPPERS

observation_spec()

By default, grabs the normal environment observation_spec

Returns:

Type	Description
dict	Observations from the environment

Source code in OmniGibson/omnigibson/envs/env_wrapper.py

def observation_spec(self):
    """
    By default, grabs the normal environment observation_spec

    Returns:
        dict: Observations from the environment
    """
    return self.env.observation_spec()

reset(**kwargs)

By default, run the normal environment reset() function. Keyword arguments (e.g. env_indices for vectorized per-env reset, seed, options) are forwarded to the wrapped environment so the wrapper stays transparent to reset.

Returns:

Type	Description
dict	Environment observation space after reset occurs

Source code in OmniGibson/omnigibson/envs/env_wrapper.py

def reset(self, **kwargs):
    """
    By default, run the normal environment reset() function. Keyword arguments (e.g. ``env_indices``
    for vectorized per-env reset, ``seed``, ``options``) are forwarded to the wrapped environment so
    the wrapper stays transparent to reset.

    Returns:
        dict: Environment observation space after reset occurs
    """
    return self.env.reset(**kwargs)

step(action, n_render_iterations=1)

By default, run the normal environment step() function

Parameters:

Name	Type	Description	Default
action	tensor	action to take in environment	required
n_render_iterations	int	Number of rendering iterations to use before returning observations	1

Returns: 4-tuple: - (dict) observations from the environment - (float) reward from the environment - (bool) whether the current episode is terminated - (bool) whether the current episode is truncated - (dict) misc information

Source code in OmniGibson/omnigibson/envs/env_wrapper.py

def step(self, action, n_render_iterations=1):
    """
    By default, run the normal environment step() function

    Args:
        action (th.tensor): action to take in environment
        n_render_iterations (int): Number of rendering iterations to use before returning observations
    Returns:
        4-tuple:
            - (dict) observations from the environment
            - (float) reward from the environment
            - (bool) whether the current episode is terminated
            - (bool) whether the current episode is truncated
            - (dict) misc information
    """
    return self.env.step(action, n_render_iterations=n_render_iterations)

create_wrapper(env, wrapper_cfg=None)

Wraps environment @env with wrapper defined @wrapper_cfg

Parameters:

Name	Type	Description	Default
env	Environment	environment to wrap	required
wrapper_cfg	None or dict	Specified, configuration to wrap environment with If not specified, will default to env.wrapper_config	None

Source code in OmniGibson/omnigibson/envs/env_wrapper.py

def create_wrapper(env, wrapper_cfg=None):
    """
    Wraps environment @env with wrapper defined @wrapper_cfg

    Args:
        env (og.Environment): environment to wrap
        wrapper_cfg (None or dict): Specified, configuration to wrap environment with
            If not specified, will default to env.wrapper_config
    """
    wrapper_cfg = deepcopy(env.wrapper_config if wrapper_cfg is None else wrapper_cfg)
    wrapper_type = wrapper_cfg.pop("type")
    wrapper_cfg["env"] = env

    return create_class_from_registry_and_config(
        cls_name=wrapper_type,
        cls_registry=REGISTERED_ENV_WRAPPERS,
        cfg=wrapper_cfg,
        cls_type_descriptor="wrapper",
    )