Petting Zoo Vector Base Class

Base class for vectorized pettingzoo environments.

Parameters

class agilerl.vector.pz_vec_env.PettingZooVecEnv(num_envs: int, observation_spaces: dict[str, Space], action_spaces: dict[str, Space], possible_agents: list[str], *, metadata: dict[str, Any] | None = None, render_mode: str | None = None)

An abstract asynchronous, vectorized environment.

References:

https://github.com/openai/baselines/tree/master/baselines/common/vec_env https://github.com/Farama-Foundation/Gymnasium/blob/main/gymnasium/vector/vector_env.py

Parameters:

• num_envs (int) – Number of environments to vectorize

• observation_spaces (dict[str, gymnasium.spaces.Space]) – Dictionary of observation spaces

• action_spaces (dict[str, gymnasium.spaces.Space]) – Dictionary of action spaces keyed by agent

• possible_agents (list[str]) – List of possible agents

property action_space: Callable[[str], Space]

Return callable to get an agent’s action space.

close(**kwargs: Any) → None

Clean up the environments’ resources.

close_extras(*, timeout: float | None = None, terminate: bool = False, **kwargs: Any) → None

Clean up the extra resources e.g. beyond what’s in this base class.

Subclasses may override with compatible signatures (same params or wider).

property observation_space: Callable[[str], Space]

Return callable to get an agent’s observation space.

render() → ndarray[tuple[int, ...], dtype[_ScalarType_co]] | str | list | tuple[Any, ...] | None

Return the rendered frames from the parallel environments.

reset(*, seed: int | None = None, options: dict[str, Any] | None = None) → tuple[dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]] | dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]] | tuple[ndarray[tuple[int, ...], dtype[_ScalarType_co]], ...]], Mapping[str, Mapping[str, object]]]

Reset all the environments and return two dictionaries of batched observations and infos.

Parameters:

• seed (None | int, optional) – Random seed, defaults to None

• options (dict[str, Any]) – Options dictionary

Returns:

Tuple of (observations, infos)

Return type:

PzResetReturn

property single_action_space: Callable[[str], Space]

Return callable to get an agent’s single action space.

property single_observation_space: Callable[[str], Space]

Return callable to get an agent’s single observation space.

step(actions: Mapping[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]] | Tensor], *args: Any, **kwargs: Any) → tuple[dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]] | dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]] | tuple[ndarray[tuple[int, ...], dtype[_ScalarType_co]], ...]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], Mapping[str, Mapping[str, object]]]

Take an action for each parallel environment.

Parameters:

actions (Mapping[str, ArrayOrTensor]) – Dictionary of vectorized actions for each agent. Values may be arrays or tensors.

Returns:

Tuple of observations, rewards, terminated, truncated, infos

Return type:

PzStepReturn

step_async(actions: list[dict[str, int | float | ndarray[tuple[int, ...], dtype[_ScalarType_co]] | Tensor]]) → 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.

Parameters:

actions – List of dictionaries of length num_envs, each sub dictionary contains

actions for each agent in a given environment :type actions: list[dict[str, int | float | npt.NDArray]]

step_wait(timeout: float | None = None) → tuple[dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]] | dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]] | tuple[ndarray[tuple[int, ...], dtype[_ScalarType_co]], ...]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], dict[str, ndarray[tuple[int, ...], dtype[_ScalarType_co]]], Mapping[str, Mapping[str, object]]]

Wait for the step taken with step_async().

Parameters:

timeout (float | None, optional) – Number of seconds before the call times out. If None, never times out.

property unwrapped: PettingZooVecEnv

Return the base environment.