tianshou.env

VectorEnv

BaseVectorEnv

class tianshou.env.BaseVectorEnv(env_fns: List[Callable[[], Union[Env, Env, PettingZooEnv]]], worker_fn: Callable[[Callable[[], Env]], EnvWorker], wait_num: Optional[int] = None, timeout: Optional[float] = None)

Bases: object

Base class for vectorized environments.

Usage:

env_num = 8
envs = DummyVectorEnv([lambda: gym.make(task) for _ in range(env_num)])
assert len(envs) == env_num

It accepts a list of environment generators. In other words, an environment generator efn of a specific task means that efn() returns the environment of the given task, for example, gym.make(task).

All of the VectorEnv must inherit BaseVectorEnv. Here are some other usages:

envs.seed(2)  # which is equal to the next line
envs.seed([2, 3, 4, 5, 6, 7, 8, 9])  # set specific seed for each env
obs = envs.reset()  # reset all environments
obs = envs.reset([0, 5, 7])  # reset 3 specific environments
obs, rew, done, info = envs.step([1] * 8)  # step synchronously
envs.render()  # render all environments
envs.close()  # close all environments

Warning

If you use your own environment, please make sure the seed method is set up properly, e.g.,

def seed(self, seed):
    np.random.seed(seed)

Otherwise, the outputs of these envs may be the same with each other.

Parameters

• env_fns – a list of callable envs, env_fns[i]() generates the i-th env.

• worker_fn – a callable worker, worker_fn(env_fns[i]) generates a worker which contains the i-th env.

• wait_num (int) – use in asynchronous simulation if the time cost of env.step varies with time and synchronously waiting for all environments to finish a step is time-wasting. In that case, we can return when wait_num environments finish a step and keep on simulation in these environments. If None, asynchronous simulation is disabled; else, 1 <= wait_num <= env_num.

• timeout (float) – use in asynchronous simulation same as above, in each vectorized step it only deal with those environments spending time within timeout seconds.

__len__() → int

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

get_env_attr(key: str, id: Optional[Union[int, List[int], 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], 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.

reset(id: Optional[Union[int, List[int], ndarray]] = None, **kwargs: Any) → Tuple[ndarray, Union[dict, List[dict]]]

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

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

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

• terminated a numpy.ndarray, whether these episodes have been terminated

• truncated a numpy.ndarray, whether these episodes have been truncated

• 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.

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.

DummyVectorEnv

class tianshou.env.DummyVectorEnv(env_fns: List[Callable[[], Union[Env, Env, PettingZooEnv]]], **kwargs: Any)

Bases: BaseVectorEnv

Dummy vectorized environment wrapper, implemented in for-loop.

See also

Please refer to BaseVectorEnv for other APIs’ usage.

SubprocVectorEnv

class tianshou.env.SubprocVectorEnv(env_fns: List[Callable[[], Union[Env, Env, PettingZooEnv]]], **kwargs: Any)

Bases: BaseVectorEnv

Vectorized environment wrapper based on subprocess.

See also

Please refer to BaseVectorEnv for other APIs’ usage.

ShmemVectorEnv

class tianshou.env.ShmemVectorEnv(env_fns: List[Callable[[], Union[Env, Env, PettingZooEnv]]], **kwargs: Any)

Bases: BaseVectorEnv

Optimized SubprocVectorEnv with shared buffers to exchange observations.

ShmemVectorEnv has exactly the same API as SubprocVectorEnv.

See also

Please refer to BaseVectorEnv for other APIs’ usage.

RayVectorEnv

class tianshou.env.RayVectorEnv(env_fns: List[Callable[[], Union[Env, Env, PettingZooEnv]]], **kwargs: Any)

Bases: BaseVectorEnv

Vectorized environment wrapper based on ray.

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

See also

Please refer to BaseVectorEnv for other APIs’ usage.

Wrapper

ContinuousToDiscrete

class tianshou.env.ContinuousToDiscrete(env: Env, action_per_dim: Union[int, List[int]])

Bases: ActionWrapper

Gym environment wrapper to take discrete action in a continuous environment.

Parameters

• env (gym.Env) – gym environment with continuous action space.

• action_per_dim (int) – number of discrete actions in each dimension of the action space.

action(act: ndarray) → ndarray

Returns a modified action before env.step() is called.

Args:

action: The original step() actions

Returns:

The modified actions

VectorEnvWrapper

class tianshou.env.VectorEnvWrapper(venv: BaseVectorEnv)

Bases: BaseVectorEnv

Base class for vectorized environments wrapper.

__len__() → int

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

get_env_attr(key: str, id: Optional[Union[int, List[int], 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], 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.

reset(id: Optional[Union[int, List[int], ndarray]] = None, **kwargs: Any) → Tuple[ndarray, Union[dict, List[dict]]]

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

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

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

• terminated a numpy.ndarray, whether these episodes have been terminated

• truncated a numpy.ndarray, whether these episodes have been truncated

• 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.

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.

VectorEnvNormObs

class tianshou.env.VectorEnvNormObs(venv: BaseVectorEnv, update_obs_rms: bool = True)

Bases: VectorEnvWrapper

An observation normalization wrapper for vectorized environments.

Parameters

update_obs_rms (bool) – whether to update obs_rms. Default to True.

reset(id: Optional[Union[int, List[int], ndarray]] = None, **kwargs: Any) → Tuple[ndarray, Union[dict, List[dict]]]

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

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

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

• terminated a numpy.ndarray, whether these episodes have been terminated

• truncated a numpy.ndarray, whether these episodes have been truncated

• 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.

set_obs_rms(obs_rms: RunningMeanStd) → None

Set with given observation running mean/std.

get_obs_rms() → RunningMeanStd

Return observation running mean/std.

Worker

EnvWorker

class tianshou.env.worker.EnvWorker(env_fn: Callable[[], Env])

Bases: ABC

An abstract worker for an environment.

abstract get_env_attr(key: str) → Any

abstract set_env_attr(key: str, value: Any) → None

send(action: Optional[ndarray]) → None

Send action signal to low-level worker.

When action is None, it indicates sending “reset” signal; otherwise it indicates “step” signal. The paired return value from “recv” function is determined by such kind of different signal.

recv() → Union[Tuple[ndarray, ndarray, ndarray, ndarray, ndarray], Tuple[ndarray, dict]]

Receive result from low-level worker.

If the last “send” function sends a NULL action, it only returns a single observation; otherwise it returns a tuple of (obs, rew, done, info) or (obs, rew, terminated, truncated, info), based on whether the environment is using the old step API or the new one.

abstract reset(**kwargs: Any) → Tuple[ndarray, dict]

step(action: ndarray) → Tuple[ndarray, ndarray, ndarray, ndarray, ndarray]

Perform one timestep of the environment’s dynamic.

“send” and “recv” are coupled in sync simulation, so users only call “step” function. But they can be called separately in async simulation, i.e. someone calls “send” first, and calls “recv” later.

static wait(workers: List[EnvWorker], wait_num: int, timeout: Optional[float] = None) → List[EnvWorker]

Given a list of workers, return those ready ones.

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

abstract render(**kwargs: Any) → Any

Render the environment.

abstract close_env() → None

close() → None

DummyEnvWorker

class tianshou.env.worker.DummyEnvWorker(env_fn: Callable[[], Env])

Bases: EnvWorker

Dummy worker used in sequential vector environments.

get_env_attr(key: str) → Any

set_env_attr(key: str, value: Any) → None

reset(**kwargs: Any) → Tuple[ndarray, dict]

static wait(workers: List[DummyEnvWorker], wait_num: int, timeout: Optional[float] = None) → List[DummyEnvWorker]

Given a list of workers, return those ready ones.

send(action: Optional[ndarray], **kwargs: Any) → None

Send action signal to low-level worker.

When action is None, it indicates sending “reset” signal; otherwise it indicates “step” signal. The paired return value from “recv” function is determined by such kind of different signal.

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

render(**kwargs: Any) → Any

Render the environment.

close_env() → None

SubprocEnvWorker

class tianshou.env.worker.SubprocEnvWorker(env_fn: Callable[[], Env], share_memory: bool = False)

Bases: EnvWorker

Subprocess worker used in SubprocVectorEnv and ShmemVectorEnv.

get_env_attr(key: str) → Any

set_env_attr(key: str, value: Any) → None

static wait(workers: List[SubprocEnvWorker], wait_num: int, timeout: Optional[float] = None) → List[SubprocEnvWorker]

Given a list of workers, return those ready ones.

send(action: Optional[ndarray], **kwargs: Any) → None

Send action signal to low-level worker.

When action is None, it indicates sending “reset” signal; otherwise it indicates “step” signal. The paired return value from “recv” function is determined by such kind of different signal.

recv() → Union[Tuple[ndarray, ndarray, ndarray, ndarray, ndarray], Tuple[ndarray, dict]]

Receive result from low-level worker.

If the last “send” function sends a NULL action, it only returns a single observation; otherwise it returns a tuple of (obs, rew, done, info) or (obs, rew, terminated, truncated, info), based on whether the environment is using the old step API or the new one.

reset(**kwargs: Any) → Tuple[ndarray, dict]

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

render(**kwargs: Any) → Any

Render the environment.

close_env() → None

RayEnvWorker

class tianshou.env.worker.RayEnvWorker(env_fn: Callable[[], Env])

Bases: EnvWorker

Ray worker used in RayVectorEnv.

get_env_attr(key: str) → Any

set_env_attr(key: str, value: Any) → None

reset(**kwargs: Any) → Any

static wait(workers: List[RayEnvWorker], wait_num: int, timeout: Optional[float] = None) → List[RayEnvWorker]

Given a list of workers, return those ready ones.

send(action: Optional[ndarray], **kwargs: Any) → None

Send action signal to low-level worker.

When action is None, it indicates sending “reset” signal; otherwise it indicates “step” signal. The paired return value from “recv” function is determined by such kind of different signal.

recv() → Tuple[ndarray, ndarray, ndarray, ndarray, ndarray]

Receive result from low-level worker.

If the last “send” function sends a NULL action, it only returns a single observation; otherwise it returns a tuple of (obs, rew, done, info) or (obs, rew, terminated, truncated, info), based on whether the environment is using the old step API or the new one.

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

render(**kwargs: Any) → Any

Render the environment.

close_env() → None

Utils

PettingZooEnv

class tianshou.env.PettingZooEnv(env: BaseWrapper)

Bases: AECEnv, ABC

The interface for petting zoo environments.

Multi-agent environments must be wrapped as PettingZooEnv. Here is the usage:

env = PettingZooEnv(...)
# obs is a dict containing obs, agent_id, and mask
obs = env.reset()
action = policy(obs)
obs, rew, trunc, term, info = env.step(action)
env.close()

The available action’s mask is set to True, otherwise it is set to False. Further usage can be found at Multi-Agent Reinforcement Learning.

agents: List[AgentID]

rewards: Dict[AgentID, float]

reset(*args: Any, **kwargs: Any) → Tuple[dict, dict]

Resets the environment to a starting state.

step(action: Any) → Tuple[Dict, List[int], bool, bool, Dict]

Accepts and executes the action of the current agent_selection in the environment.

Automatically switches control to the next agent.

close() → None

Closes any resources that should be released.

Closes the rendering window, subprocesses, network connections, or any other resources that should be released.

seed(seed: Optional[Any] = None) → None

Reseeds the environment (making the resulting environment deterministic).

render() → Any

Renders the environment as specified by self.render_mode.

Render mode can be human to display a window. Other render modes in the default environments are ‘rgb_array’ which returns a numpy array and is supported by all environments outside of classic, and ‘ansi’ which returns the strings printed (specific to classic environments).

metadata: Dict[str, Any]

possible_agents: List[AgentID]

observation_spaces: Dict[AgentID, gymnasium.spaces.Space]

action_spaces: Dict[AgentID, gymnasium.spaces.Space]

terminations: Dict[AgentID, bool]

truncations: Dict[AgentID, bool]

infos: Dict[AgentID, Dict[str, Any]]

agent_selection: AgentID