src.environment.parallelenv.vectorenvs

Module Contents

Classes

BaseVectorEnv
DummyVectorEnv	Dummy vectorized environment wrapper, implemented in for-loop.
SubprocVectorEnv	Vectorized environment wrapper based on subprocess.
RayVectorEnv	Vectorized environment wrapper based on ray.
RaySubprocVectorEnv	Vectorized environment wrapper based on ray.

API

class src.environment.parallelenv.vectorenvs.BaseVectorEnv(env_fns: List[Callable[[], gym.Env]], worker_fn: Callable[[Callable[[], gym.Env]], src.environment.parallelenv.worker.EnvWorker], wait_num: Optional[int] = None, timeout: Optional[float] = None, norm_obs: bool = False, update_obs_rms: bool = True, num_cpu_per_worker=1, num_gpu_per_worker=0, sub_worker_fn=DummyEnvWorker, no_warning=False)

Bases: gym.Env

Initialization

_assert_is_not_closed() → None

__len__() → int

Return len(self), which is the number of environments.

__getattribute__(key: str) → Any

Switch the attribute getter depending on the key.

Any class who inherits gym.Env will inherit some attributes, like action_space. However, we would like the attribute lookup to go straight into the worker (in fact, this vector env’s action_space is always None).

get_env_attr(key: str, id: Optional[Union[int, List[int], numpy.ndarray]] = None) → List[Any]

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: Optional[Union[int, List[int], numpy.ndarray]] = None) → None

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.

get_env_obj()

_wrap_id(id: Optional[Union[int, List[int], numpy.ndarray]] = None) → Union[List[int], numpy.ndarray]

_assert_id(id: Union[List[int], numpy.ndarray]) → None

_wrap_RSEW_id(id, data=None)

reset(id: Optional[Union[int, List[int], numpy.ndarray]] = None) → numpy.ndarray

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: numpy.ndarray, id: Optional[Union[int, List[int], numpy.ndarray]] = None) → Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]

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.

Returns:

A tuple including four items:

• obs a numpy.ndarray, the agent’s observation of current environments

• rew a numpy.ndarray, the amount of rewards returned after previous actions

• done a numpy.ndarray, whether these episodes have ended, in which case further step() calls will return undefined results

• info 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 the id argument should be a subset of the env_id in the last returned info (initially they are env_ids of all the environments). If action is None, fetch unfinished step() calls instead.

seed(seed: Optional[Union[int, List[int]]] = None) → List[Optional[List[int]]]

Set the seed for all environments.

Accept None, an int (which will extend i 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”.

render(**kwargs: Any) → List[Any]

Render all of the environments.

customized_method(call_method, data=None, id: Optional[Union[int, List[int], numpy.ndarray]] = None)

close() → None

Close all of the environments.

This function will be called only once (if not, it will be called during garbage collected). This way, close of all workers can be assured.

normalize_obs(obs: numpy.ndarray) → numpy.ndarray

Normalize observations by statistics in obs_rms.

class src.environment.parallelenv.vectorenvs.DummyVectorEnv(env_fns: List[Callable[[], gym.Env]], num_cpus=1, num_gpus=0, no_warning=False, **kwargs: Any)

Bases: src.environment.parallelenv.vectorenvs.BaseVectorEnv

Dummy vectorized environment wrapper, implemented in for-loop.

Initialization

class src.environment.parallelenv.vectorenvs.SubprocVectorEnv(env_fns: List[Callable[[], gym.Env]], num_cpus=1, num_gpus=0, no_warning=False, **kwargs: Any)

Bases: src.environment.parallelenv.vectorenvs.BaseVectorEnv

Vectorized environment wrapper based on subprocess.

Initialization

class src.environment.parallelenv.vectorenvs.RayVectorEnv(env_fns: List[Callable[[], gym.Env]], num_cpus=1, num_gpus=0, no_warning=False, temp_ray_dir=None, num_cpu_per_worker=1, num_gpu_per_worker=0, **kwargs: Any)

Bases: src.environment.parallelenv.vectorenvs.BaseVectorEnv

Vectorized environment wrapper based on ray.

Initialization

Parameters:

• num_cpus (int) – the number of cpu cores allocated.

• no_warning (bool) – whether to show warnings

• temp_ray_dir (str) – the absolute dir path to store temporal files generated by ray during execution, default to creating a ray_tmp dir in workdir.

rollout()

class src.environment.parallelenv.vectorenvs.RaySubprocVectorEnv(env_fns: List[Callable[[], gym.Env]], num_cpus, num_gpus=0, no_warning=False, temp_ray_dir=None, num_cpu_per_worker=1, num_gpu_per_worker=0, **kwargs: Any)

Bases: src.environment.parallelenv.vectorenvs.BaseVectorEnv

Vectorized environment wrapper based on ray.

This is a choice to run distributed environments in a cluster.

Initialization