Skip to content

lit_sb3.vec_env

source package lit_sb3.vec_env

Classes

  • CloudpickleWrapper — Uses cloudpickle to serialize contents (otherwise multiprocessing tries to use pickle).

  • DummyVecEnv — Creates a simple vectorized wrapper for multiple environments, calling each environment in sequence on the.

  • StackedObservations — Frame stacking wrapper for data.

  • SubprocVecEnv — Creates a multiprocess vectorized wrapper for multiple environments, distributing each environment to its own.

  • VecCheckNan — NaN and inf checking wrapper for vectorized environment, will raise a warning by default,.

  • VecEnv — An abstract asynchronous, vectorized environment.

  • VecEnvWrapper — Vectorized environment base class.

  • VecExtractDictObs — A vectorized wrapper for extracting dictionary observations.

  • VecFrameStack — Frame stacking wrapper for vectorized environment. Designed for image observations.

  • VecMonitor — A vectorized monitor wrapper for vectorized Gym environments,.

  • VecNormalize — A moving average, normalizing wrapper for vectorized environment.

  • VecTransposeImage — Re-order channels, from HxWxC to CxHxW.

  • VecVideoRecorder — Wraps a VecEnv or VecEnvWrapper object to record rendered image as mp4 video.

Functions

source class CloudpickleWrapper(var: Any)

Uses cloudpickle to serialize contents (otherwise multiprocessing tries to use pickle).

Parameters

  • var : Any — the variable you wish to wrap for pickling with cloudpickle

source class DummyVecEnv(env_fns: list[Callable[[], gym.Env]])

Bases : VecEnv

Creates a simple vectorized wrapper for multiple environments, calling each environment in sequence on the.

current.

Python process. This is useful for computationally simple environment such as Cartpole-v1, as the overhead of multiprocess or multithread outweighs the environment computation time. This can also be used for RL methods that require a vectorized environment, but that you want a single environments to train with.

Parameters

  • env_fns : list[Callable[[], gym.Env]] — a list of functions that return environments to vectorize

Raises

  • ValueError — If the same environment instance is passed as the output of two or more different env_fn.

Methods

source method DummyVecEnv.step_async(actions: np.ndarray) → None

source method DummyVecEnv.step_wait() → VecEnvStepReturn

source method DummyVecEnv.reset() → VecEnvObs

source method DummyVecEnv.close() → None

source method DummyVecEnv.get_images() → Sequence[np.ndarray | None]

source method DummyVecEnv.render(mode: str | None = None) → np.ndarray | None

Gym environment rendering. If there are multiple environments then.

they are tiled together in one image via BaseVecEnv.render().

Parameters

  • mode : str | None — The rendering type.

source method DummyVecEnv.get_attr(attr_name: str, indices: VecEnvIndices = None) → list[Any]

Return attribute from vectorized environment (see base class).

source method DummyVecEnv.set_attr(attr_name: str, value: Any, indices: VecEnvIndices = None) → None

Set attribute inside vectorized environments (see base class).

source method DummyVecEnv.env_method(method_name: str, *method_args, indices: VecEnvIndices = None, **method_kwargs) → list[Any]

Call instance methods of vectorized environments.

source method DummyVecEnv.env_is_wrapped(wrapper_class: type[gym.Wrapper], indices: VecEnvIndices = None) → list[bool]

Check if worker environments are wrapped with a given wrapper.

source class StackedObservations[TObs: (np.ndarray, dict[str, np.ndarray])](num_envs: int, n_stack: int, observation_space: spaces.Box | spaces.Dict, channels_order: str | Mapping[str, str | None] | None = None)

Frame stacking wrapper for data.

Dimension to stack over is either first (channels-first) or last (channels-last), which is detected automatically using common.preprocessing.is_image_space_channels_first if observation is an image space.

Parameters

  • num_envs : int — Number of environments

  • n_stack : int — Number of frames to stack

  • observation_space : spaces.Box | spaces.Dict — Environment observation space

  • channels_order : str | Mapping[str, str | None] | None — If "first", stack on first image dimension. If "last", stack on last dimension. If None, automatically detect channel to stack over in case of image observation or default to "last". For Dict space, channels_order can also be a dictionary.

Methods

  • compute_stacking — Calculates the parameters in order to stack observations.

  • reset — Reset the stacked_obs, add the reset observation to the stack, and return the stack.

  • update — Add the observations to the stack and use the dones to update the infos.

source staticmethod StackedObservations.compute_stacking(n_stack: int, observation_space: spaces.Box, channels_order: str | None = None) → tuple[bool, int, tuple[int, ...], int]

Calculates the parameters in order to stack observations.

Parameters

  • n_stack : int — Number of observations to stack

  • observation_space : spaces.Box — Observation space

  • channels_order : str | None — Order of the channels

Returns

  • tuple[bool, int, tuple[int, ...], int] — Tuple of channels_first, stack_dimension, stackedobs, repeat_axis

source method StackedObservations.reset(observation: TObs) → TObs

Reset the stacked_obs, add the reset observation to the stack, and return the stack.

Parameters

  • observation : TObs — Reset observation

Returns

  • TObs — The stacked reset observation

source method StackedObservations.update(observations: TObs, dones: np.ndarray, infos: list[dict[str, Any]]) → tuple[TObs, list[dict[str, Any]]]

Add the observations to the stack and use the dones to update the infos.

Parameters

  • observations : TObs — Observations

  • dones : np.ndarray — Dones

  • infos : list[dict[str, Any]] — Infos

Returns

  • tuple[TObs, list[dict[str, Any]]] — Tuple of the stacked observations and the updated infos

source class SubprocVecEnv(env_fns: list[Callable[[], gym.Env]], start_method: str | None = None)

Bases : VecEnv

Creates a multiprocess vectorized wrapper for multiple environments, distributing each environment to its own.

process, allowing significant speed up when the environment is computationally complex.

For performance reasons, if your environment is not IO bound, the number of environments should not exceed the number of logical cores on your CPU.

.. warning::

Only 'forkserver' and 'spawn' start methods are thread-safe,
which is important when TensorFlow sessions or other non thread-safe
libraries are used in the parent (see issue #217). However, compared to
'fork' they incur a small start-up cost and have restrictions on
global variables. With those methods, users must wrap the code in an
``if __name__ == "__main__":`` block.
For more information, see the multiprocessing documentation.

Parameters

  • env_fns : list[Callable[[], gym.Env]] — Environments to run in subprocesses

  • start_method : str | None — method used to start the subprocesses. Must be one of the methods returned by multiprocessing.get_all_start_methods(). Defaults to 'forkserver' on available platforms, and 'spawn' otherwise.

Methods

source method SubprocVecEnv.step_async(actions: np.ndarray) → None

source method SubprocVecEnv.step_wait() → VecEnvStepReturn

source method SubprocVecEnv.reset() → VecEnvObs

source method SubprocVecEnv.close() → None

source method SubprocVecEnv.get_images() → Sequence[np.ndarray | None]

source method SubprocVecEnv.has_attr(attr_name: str) → bool

Check if an attribute exists for a vectorized environment. (see base class).

source method SubprocVecEnv.get_attr(attr_name: str, indices: VecEnvIndices = None) → list[Any]

Return attribute from vectorized environment (see base class).

source method SubprocVecEnv.set_attr(attr_name: str, value: Any, indices: VecEnvIndices = None) → None

Set attribute inside vectorized environments (see base class).

source method SubprocVecEnv.env_method(method_name: str, *method_args, indices: VecEnvIndices = None, **method_kwargs) → list[Any]

Call instance methods of vectorized environments.

source method SubprocVecEnv.env_is_wrapped(wrapper_class: type[gym.Wrapper], indices: VecEnvIndices = None) → list[bool]

Check if worker environments are wrapped with a given wrapper.

source class VecCheckNan(venv: VecEnv, raise_exception: bool = False, warn_once: bool = True, check_inf: bool = True)

Bases : VecEnvWrapper

NaN and inf checking wrapper for vectorized environment, will raise a warning by default,.

allowing you to know from what the NaN of inf originated from.

Parameters

  • venv : VecEnv — the vectorized environment to wrap

  • raise_exception : bool — Whether to raise a ValueError, instead of a UserWarning

  • warn_once : bool — Whether to only warn once.

  • check_inf : bool — Whether to check for +inf or -inf as well

Methods

source method VecCheckNan.step_async(actions: np.ndarray) → None

source method VecCheckNan.step_wait() → VecEnvStepReturn

source method VecCheckNan.reset() → VecEnvObs

source method VecCheckNan.check_array_value(name: str, value: np.ndarray) → list[tuple[str, str]]

Check for inf and NaN for a single numpy array.

Parameters

  • name : str — Name of the value being check

  • value : np.ndarray — Value (numpy array) to check

Returns

  • list[tuple[str, str]] — A list of issues found.

source class VecEnv(num_envs: int, observation_space: spaces.Space, action_space: spaces.Space)

Bases : ABC

An abstract asynchronous, vectorized environment.

Parameters

  • num_envs : int — Number of environments

  • observation_space : spaces.Space — Observation space

  • action_space : spaces.Space — Action space

Methods

  • reset — Reset all the environments and return an array of.

  • step_async — Tell all the environments to start taking a step.

  • step_wait — Wait for the step taken with step_async().

  • close — Clean up the environment's resources.

  • has_attr — Check if an attribute exists for a vectorized environment.

  • get_attr — Return attribute from vectorized environment.

  • set_attr — Set attribute inside vectorized environments.

  • env_method — Call instance methods of vectorized environments.

  • env_is_wrapped — Check if environments are wrapped with a given wrapper.

  • step — Step the environments with the given action.

  • get_images — Return RGB images from each environment when available.

  • render — Gym environment rendering.

  • seed — Sets the random seeds for all environments, based on a given seed.

  • set_options — Set environment options for all environments.

  • getattr_depth_check — Check if an attribute reference is being hidden in a recursive call to getattr.

source method VecEnv.reset() → VecEnvObs

Reset all the environments and return an array of.

observations, or a tuple of observation arrays.

If step_async is still doing work, that work will be cancelled and step_wait() should not be called until step_async() is invoked again.

Returns

  • VecEnvObs — observation

Raises

  • NotImplementedError

source method VecEnv.step_async(actions: np.ndarray) → None

Tell all the environments to start taking a step.

with the given actions. Call step_wait() to get the results of the step.

You should not call this if a step_async run is already pending.

Raises

  • NotImplementedError

source method VecEnv.step_wait() → VecEnvStepReturn

Wait for the step taken with step_async().

Returns

  • VecEnvStepReturn — observation, reward, done, information

Raises

  • NotImplementedError

source method VecEnv.close() → None

Clean up the environment's resources.

Raises

  • NotImplementedError

source method VecEnv.has_attr(attr_name: str) → bool

Check if an attribute exists for a vectorized environment.

Parameters

  • attr_name : str — The name of the attribute to check

Returns

  • bool — True if 'attr_name' exists in all environments

source method VecEnv.get_attr(attr_name: str, indices: VecEnvIndices = None) → list[Any]

Return attribute from vectorized environment.

Parameters

  • attr_name : str — The name of the attribute whose value to return

  • indices : VecEnvIndices — Indices of envs to get attribute from

Returns

  • list[Any] — List of values of 'attr_name' in all environments

Raises

  • NotImplementedError

source method VecEnv.set_attr(attr_name: str, value: Any, indices: VecEnvIndices = None) → None

Set attribute inside vectorized environments.

Parameters

  • attr_name : str — The name of attribute to assign new value

  • value : Any — Value to assign to attr_name

  • indices : VecEnvIndices — Indices of envs to assign value

Raises

  • NotImplementedError

source method VecEnv.env_method(method_name: str, *method_args, indices: VecEnvIndices = None, **method_kwargs) → list[Any]

Call instance methods of vectorized environments.

Parameters

  • method_name : str — The name of the environment method to invoke.

  • indices : VecEnvIndices — Indices of envs whose method to call

  • method_args — Any positional arguments to provide in the call

  • method_kwargs — Any keyword arguments to provide in the call

Returns

  • list[Any] — List of items returned by the environment's method call

Raises

  • NotImplementedError

source method VecEnv.env_is_wrapped(wrapper_class: type[gym.Wrapper], indices: VecEnvIndices = None) → list[bool]

Check if environments are wrapped with a given wrapper.

Parameters

  • method_name — The name of the environment method to invoke.

  • indices : VecEnvIndices — Indices of envs whose method to call

  • method_args — Any positional arguments to provide in the call

  • method_kwargs — Any keyword arguments to provide in the call

  • wrapper_class : type[gym.Wrapper] — The wrapper class parameter.

Returns

  • list[bool] — True if the env is wrapped, False otherwise, for each env queried.

Raises

  • NotImplementedError

source method VecEnv.step(actions: np.ndarray) → VecEnvStepReturn

Step the environments with the given action.

Parameters

  • actions : np.ndarray — the action

Returns

  • VecEnvStepReturn — observation, reward, done, information

source method VecEnv.get_images() → Sequence[np.ndarray | None]

Return RGB images from each environment when available.

Raises

  • NotImplementedError

source method VecEnv.render(mode: str | None = None) → np.ndarray | None

Gym environment rendering.

Parameters

  • mode : str | None — the rendering type

source method VecEnv.seed(seed: int | None = None) → Sequence[None | int]

Sets the random seeds for all environments, based on a given seed.

Each individual environment will still get its own seed, by incrementing the given seed. WARNING: since gym 0.26, those seeds will only be passed to the environment at the next reset.

Parameters

  • seed : int | None — The random seed. May be None for completely random seeding.

Returns

  • Sequence[None | int] — Returns a list containing the seeds for each individual env. Note that all list elements may be None, if the env does not return anything when being seeded.

source method VecEnv.set_options(options: list[dict] | dict | None = None) → None

Set environment options for all environments.

If a dict is passed instead of a list, the same options will be used for all environments. WARNING: Those options will only be passed to the environment at the next reset.

Parameters

  • options : list[dict] | dict | None — A dictionary of environment options to pass to each environment at the next reset.

source property VecEnv.unwrapped: VecEnv

source method VecEnv.getattr_depth_check(name: str, already_found: bool) → str | None

Check if an attribute reference is being hidden in a recursive call to getattr.

Parameters

  • name : str — name of attribute to check for

  • already_found : bool — whether this attribute has already been found in a wrapper

Returns

  • str | None — name of module whose attribute is being shadowed, if any.

source class VecEnvWrapper(venv: VecEnv, observation_space: spaces.Space | None = None, action_space: spaces.Space | None = None)

Bases : VecEnv

Vectorized environment base class.

Parameters

  • venv : VecEnv — the vectorized environment to wrap

  • observation_space : spaces.Space | None — the observation space (can be None to load from venv)

  • action_space : spaces.Space | None — the action space (can be None to load from venv)

Methods

source method VecEnvWrapper.step_async(actions: np.ndarray) → None

source method VecEnvWrapper.reset() → VecEnvObs

source method VecEnvWrapper.step_wait() → VecEnvStepReturn

source method VecEnvWrapper.seed(seed: int | None = None) → Sequence[None | int]

source method VecEnvWrapper.set_options(options: list[dict] | dict | None = None) → None

source method VecEnvWrapper.close() → None

source method VecEnvWrapper.render(mode: str | None = None) → np.ndarray | None

source method VecEnvWrapper.get_images() → Sequence[np.ndarray | None]

source method VecEnvWrapper.has_attr(attr_name: str) → bool

source method VecEnvWrapper.get_attr(attr_name: str, indices: VecEnvIndices = None) → list[Any]

source method VecEnvWrapper.set_attr(attr_name: str, value: Any, indices: VecEnvIndices = None) → None

source method VecEnvWrapper.env_method(method_name: str, *method_args, indices: VecEnvIndices = None, **method_kwargs) → list[Any]

source method VecEnvWrapper.env_is_wrapped(wrapper_class: type[gym.Wrapper], indices: VecEnvIndices = None) → list[bool]

source method VecEnvWrapper.getattr_recursive(name: str) → Any

Recursively check wrappers to find attribute.

Parameters

  • name : str — name of attribute to look for

Returns

  • Any — attribute

source method VecEnvWrapper.getattr_depth_check(name: str, already_found: bool) → str | None

See base class.

Returns

  • str | None — name of module whose attribute is being shadowed, if any.

source class VecExtractDictObs(venv: VecEnv, key: str)

Bases : VecEnvWrapper

A vectorized wrapper for extracting dictionary observations.

Parameters

  • venv : VecEnv — The vectorized environment

  • key : str — The key of the dictionary observation

Methods

source method VecExtractDictObs.reset() → np.ndarray

source method VecExtractDictObs.step_wait() → VecEnvStepReturn

source class VecFrameStack(venv: VecEnv, n_stack: int, channels_order: str | Mapping[str, str] | None = None)

Bases : VecEnvWrapper

Frame stacking wrapper for vectorized environment. Designed for image observations.

Parameters

  • venv : VecEnv — Vectorized environment to wrap

  • n_stack : int — Number of frames to stack

  • channels_order : str | Mapping[str, str] | None — If "first", stack on first image dimension. If "last", stack on last dimension. If None, automatically detect channel to stack over in case of image observation or default to "last" (default). Alternatively channels_order can be a dictionary which can be used with environments with Dict observation spaces

Methods

source method VecFrameStack.step_wait() → tuple[np.ndarray | dict[str, np.ndarray], np.ndarray, np.ndarray, list[dict[str, Any]]]

source method VecFrameStack.reset() → np.ndarray | dict[str, np.ndarray]

Reset all environments.

source class VecMonitor(venv: VecEnv, filename: str | None = None, info_keywords: tuple[str, ...] = ())

Bases : VecEnvWrapper

A vectorized monitor wrapper for vectorized Gym environments,.

it is used to record the episode reward, length, time and other data.

Some environments like openai/procgen <https://github.com/openai/procgen>_ or gym3 <https://github.com/openai/gym3>_ directly initialize the vectorized environments, without giving us a chance to use the Monitor wrapper. So this class simply does the job of the Monitor wrapper on a vectorized level.

Parameters

  • venv : VecEnv — The vectorized environment

  • filename : str | None — the location to save a log file, can be None for no log

  • info_keywords : tuple[str, ...] — extra information to log, from the information return of env.step()

Methods

source method VecMonitor.reset() → VecEnvObs

source method VecMonitor.step_wait() → VecEnvStepReturn

source method VecMonitor.close() → None

source class VecNormalize(venv: VecEnv, training: bool = True, norm_obs: bool = True, norm_reward: bool = True, clip_obs: float = 10.0, clip_reward: float = 10.0, gamma: float = 0.99, epsilon: float = 1e-08, norm_obs_keys: list[str] | None = None)

Bases : VecEnvWrapper

A moving average, normalizing wrapper for vectorized environment.

has support for saving/loading moving average,.

Parameters

  • venv : VecEnv — the vectorized environment to wrap

  • training : bool — Whether to update or not the moving average

  • norm_obs : bool — Whether to normalize observation or not (default: True)

  • norm_reward : bool — Whether to normalize rewards or not (default: True)

  • clip_obs : float — Max absolute value for observation

  • clip_reward : float — Max value absolute for discounted reward

  • gamma : float — discount factor

  • epsilon : float — To avoid division by zero

  • norm_obs_keys : list[str] | None — Which keys from observation dict to normalize. If not specified, all keys will be normalized.

Methods

  • set_venv — Sets the vector environment to wrap to venv.

  • step_wait — Apply sequence of actions to sequence of environments.

  • normalize_obs — Normalize observations using this VecNormalize's observations statistics.

  • normalize_reward — Normalize rewards using this VecNormalize's rewards statistics.

  • unnormalize_obs

  • unnormalize_reward

  • get_original_obs — Returns an unnormalized version of the observations from the most recent.

  • get_original_reward — Returns an unnormalized version of the rewards from the most recent step.

  • reset — Reset all environments.

  • load — Loads a saved VecNormalize object.

  • save — Save current VecNormalize object with.

source method VecNormalize.set_venv(venv: VecEnv) → None

Sets the vector environment to wrap to venv.

Also sets attributes derived from this such as num_env.

Parameters

Raises

  • ValueError

source method VecNormalize.step_wait() → VecEnvStepReturn

Apply sequence of actions to sequence of environments.

actions -> (observations, rewards, dones).

where dones is a boolean vector indicating whether each element is new.

source method VecNormalize.normalize_obs(obs: np.ndarray | dict[str, np.ndarray]) → np.ndarray | dict[str, np.ndarray]

Normalize observations using this VecNormalize's observations statistics.

Calling this method does not update statistics.

source method VecNormalize.normalize_reward(reward: np.ndarray) → np.ndarray

Normalize rewards using this VecNormalize's rewards statistics.

Calling this method does not update statistics.

source method VecNormalize.unnormalize_obs(obs: np.ndarray | dict[str, np.ndarray]) → np.ndarray | dict[str, np.ndarray]

source method VecNormalize.unnormalize_reward(reward: np.ndarray) → np.ndarray

source method VecNormalize.get_original_obs() → np.ndarray | dict[str, np.ndarray]

Returns an unnormalized version of the observations from the most recent.

step or reset.

source method VecNormalize.get_original_reward() → np.ndarray

Returns an unnormalized version of the rewards from the most recent step.

source method VecNormalize.reset() → np.ndarray | dict[str, np.ndarray]

Reset all environments.

Returns

  • np.ndarray | dict[str, np.ndarray] — first observation of the episode

source staticmethod VecNormalize.load(load_path: str | Path, venv: VecEnv) → VecNormalize

Loads a saved VecNormalize object.

Parameters

  • load_path : str | Path — the path to load from.

  • venv : VecEnv — the VecEnv to wrap.

source method VecNormalize.save(save_path: str | Path) → None

Save current VecNormalize object with.

all running statistics and settings (e.g. clip_obs).

Parameters

  • save_path : str | Path — The path to save to

source class VecTransposeImage(venv: VecEnv, skip: bool = False)

Bases : VecEnvWrapper

Re-order channels, from HxWxC to CxHxW.

It is required for PyTorch convolution layers.

Parameters

  • venv : VecEnv

  • skip : bool — Skip this wrapper if needed as we rely on heuristic to apply it or not, which may result in unwanted behavior, see GH issue #671.

Methods

source staticmethod VecTransposeImage.transpose_space(observation_space: spaces.Box, key: str = '') → spaces.Box

Transpose an observation space (re-order channels).

Parameters

  • observation_space : spaces.Box

  • key : str — In case of dictionary space, the key of the observation space.

  • observation_space : spaces.Box — The observation space parameter.

source staticmethod VecTransposeImage.transpose_image(image: np.ndarray) → np.ndarray

Transpose an image or batch of images (re-order channels).

Parameters

  • image : np.ndarray

  • image : np.ndarray — The image parameter.

source method VecTransposeImage.transpose_observations(observations: np.ndarray | dict) → np.ndarray | dict

Transpose (if needed) and return new observations.

Parameters

  • observations : np.ndarray | dict

  • observations : np.ndarray | dict — The observations parameter.

Returns

  • np.ndarray | dict — Transposed observations

source method VecTransposeImage.step_wait() → VecEnvStepReturn

source method VecTransposeImage.reset() → np.ndarray | dict

Reset all environments.

source method VecTransposeImage.close() → None

source class VecVideoRecorder(venv: VecEnv, video_folder: str | Path, record_video_trigger: Callable[[int], bool], video_length: int = 200, name_prefix: str = 'rl-video')

Bases : VecEnvWrapper

Wraps a VecEnv or VecEnvWrapper object to record rendered image as mp4 video.

It requires ffmpeg or avconv to be installed on the machine.

Note: for now it only allows to record one video and all videos must have at least two frames.

The video recorder code was adapted from Gymnasium v1.0.

Parameters

  • venv : VecEnv

  • video_folder : str | Path — Where to save videos

  • record_video_trigger : Callable[[int], bool] — Function that defines when to start recording. The function takes the current number of step, and returns whether we should start recording or not.

  • video_length : int — Length of recorded videos

  • name_prefix : str — Prefix to the video name

Methods

source method VecVideoRecorder.reset() → VecEnvObs

source method VecVideoRecorder.step_wait() → VecEnvStepReturn

source method VecVideoRecorder.close() → None

Closes the wrapper then the video recorder.

source is_vecenv_wrapped(env: VecEnv, vec_wrapper_class: type[VecEnvWrapper]) → bool

Check if an environment is already wrapped in a given VecEnvWrapper.

Parameters

  • env : VecEnv — The VecEnv that is going to be checked

  • vec_wrapper_class : type[VecEnvWrapper] — The desired VecEnvWrapper class.

Returns

  • bool — True if the VecEnv is wrapped with the desired wrapper, False otherwise

source sync_envs_normalization(env: VecEnv, eval_env: VecEnv) → None

Synchronize the normalization statistics of an eval environment and train environment.

when they are both wrapped in a VecNormalize wrapper.

Parameters

  • env : VecEnv — Training env

  • eval_env : VecEnv — Environment used for evaluation.

source unwrap_vec_normalize(env: VecEnv) → VecNormalize | None

Retrieve a VecNormalize object by recursively searching.

Parameters

  • env : VecEnv — The VecEnv that is going to be unwrapped

Returns

  • VecNormalize | None — The VecNormalize object if the VecEnv is wrapped with VecNormalize, None otherwise

source unwrap_vec_wrapper[VecEnvWrapperT: VecEnvWrapper](env: VecEnv, vec_wrapper_class: type[VecEnvWrapperT]) → VecEnvWrapperT | None

Retrieve a VecEnvWrapper object by recursively searching.

Parameters

  • env : VecEnv — The VecEnv that is going to be unwrapped

  • vec_wrapper_class : type[VecEnvWrapperT] — The desired VecEnvWrapper class.

Returns

  • VecEnvWrapperT | None — The VecEnvWrapper object if the VecEnv is wrapped with the desired wrapper, None otherwise