venv_wrappers#
Source code: tianshou/env/venv_wrappers.py
- class VectorEnvWrapper(venv: BaseVectorEnv)[source]#
Bases:
BaseVectorEnv
Base class for vectorized environments wrapper.
- get_env_attr(key: str, id: int | list[int] | ndarray | None = None) list[Any] [source]#
Get an attribute from the underlying environments.
If id is an int, retrieve the attribute denoted by key from the environment underlying the worker at index id. The result is returned as a list with one element. Otherwise, retrieve the attribute for all workers at indices id and return a list that is ordered correspondingly to id.
- Parameters:
key (str) – The key of the desired attribute.
id – Indice(s) of the desired worker(s). Default to None for all env_id.
- Return list:
The list of environment attributes.
- set_env_attr(key: str, value: Any, id: int | list[int] | ndarray | None = None) None [source]#
Set an attribute in the underlying environments.
If id is an int, set the attribute denoted by key from the environment underlying the worker at index id to value. Otherwise, set the attribute for all workers at indices id.
- Parameters:
key (str) – The key of the desired attribute.
value (Any) – The new value of the attribute.
id – Indice(s) of the desired worker(s). Default to None for all env_id.
- reset(env_id: int | list[int] | ndarray | None = None, **kwargs: Any) tuple[ndarray, ndarray] [source]#
Reset the state of some envs and return initial observations.
If id is None, reset the state of all the environments and return initial observations, otherwise reset the specific environments with the given id, either an int or a list.
- step(action: ndarray | Tensor | None, id: int | list[int] | ndarray | None = None) tuple[ndarray, ndarray, ndarray, ndarray, ndarray] [source]#
Run one timestep of some environments’ dynamics.
If id is None, run one timestep of all the environments` dynamics; otherwise run one timestep for some environments with given id, either an int or a list. When the end of episode is reached, you are responsible for calling reset(id) to reset this environment`s state.
Accept a batch of action and return a tuple (batch_obs, batch_rew, batch_done, batch_info) in numpy format.
- Parameters:
action (numpy.ndarray) – a batch of action provided by the agent. If the venv is async, the action can be None, which will result in all arrays in the returned tuple being empty.
- Returns:
A tuple consisting of either:
obs
a numpy.ndarray, the agent’s observation of current environmentsrew
a numpy.ndarray, the amount of rewards returned after previous actionsterminated
a numpy.ndarray, whether these episodes have been terminatedtruncated
a numpy.ndarray, whether these episodes have been truncatedinfo
a numpy.ndarray, contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)
For the async simulation:
Provide the given action to the environments. The action sequence should correspond to the
id
argument, and theid
argument should be a subset of theenv_id
in the last returnedinfo
(initially they are env_ids of all the environments). If action is None, fetch unfinished step() calls instead.
- seed(seed: int | list[int] | None = None) list[list[int] | None] [source]#
Set the seed for all environments.
Accept
None
, an int (which will extendi
to[i, i + 1, i + 2, ...]
) or a list.- Returns:
The list of seeds used in this env’s random number generators. The first value in the list should be the “main” seed, or the value which a reproducer pass to “seed”.
- class VectorEnvNormObs(venv: BaseVectorEnv, update_obs_rms: bool = True)[source]#
Bases:
VectorEnvWrapper
An observation normalization wrapper for vectorized environments.
- Parameters:
update_obs_rms – whether to update obs_rms. Default to True.
- reset(env_id: int | list[int] | ndarray | None = None, **kwargs: Any) tuple[ndarray, ndarray] [source]#
Reset the state of some envs and return initial observations.
If id is None, reset the state of all the environments and return initial observations, otherwise reset the specific environments with the given id, either an int or a list.
- step(action: ndarray | Tensor | None, id: int | list[int] | ndarray | None = None) tuple[ndarray, ndarray, ndarray, ndarray, ndarray] [source]#
Run one timestep of some environments’ dynamics.
If id is None, run one timestep of all the environments` dynamics; otherwise run one timestep for some environments with given id, either an int or a list. When the end of episode is reached, you are responsible for calling reset(id) to reset this environment`s state.
Accept a batch of action and return a tuple (batch_obs, batch_rew, batch_done, batch_info) in numpy format.
- Parameters:
action (numpy.ndarray) – a batch of action provided by the agent. If the venv is async, the action can be None, which will result in all arrays in the returned tuple being empty.
- Returns:
A tuple consisting of either:
obs
a numpy.ndarray, the agent’s observation of current environmentsrew
a numpy.ndarray, the amount of rewards returned after previous actionsterminated
a numpy.ndarray, whether these episodes have been terminatedtruncated
a numpy.ndarray, whether these episodes have been truncatedinfo
a numpy.ndarray, contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)
For the async simulation:
Provide the given action to the environments. The action sequence should correspond to the
id
argument, and theid
argument should be a subset of theenv_id
in the last returnedinfo
(initially they are env_ids of all the environments). If action is None, fetch unfinished step() calls instead.
- set_obs_rms(obs_rms: RunningMeanStd) None [source]#
Set with given observation running mean/std.
- get_obs_rms() RunningMeanStd [source]#
Return observation running mean/std.