Spaces Vector Utils#

gymnasium.vector.utils.batch_space(space: Space, n: int = 1) → Space#

gymnasium.vector.utils.batch_space(space: Box, n=1)

gymnasium.vector.utils.batch_space(space: Discrete, n=1)

gymnasium.vector.utils.batch_space(space: MultiDiscrete, n=1)

gymnasium.vector.utils.batch_space(space: MultiBinary, n=1)

gymnasium.vector.utils.batch_space(space: Tuple, n=1)

gymnasium.vector.utils.batch_space(space: Dict, n=1)

gymnasium.vector.utils.batch_space(space: Space, n=1)

Create a (batched) space, containing multiple copies of a single space.

Example:

>>> from gymnasium.spaces import Box, Dict
>>> space = Dict({
...     'position': Box(low=0, high=1, shape=(3,), dtype=np.float32),
...     'velocity': Box(low=0, high=1, shape=(2,), dtype=np.float32)
... })
>>> batch_space(space, n=5)
Dict(position:Box(5, 3), velocity:Box(5, 2))

Parameters:

space – Space (e.g. the observation space) for a single environment in the vectorized environment.
n – Number of environments in the vectorized environment.

Returns:

Space (e.g. the observation space)

Raises:

ValueError – Cannot batch space that is not a valid gym.Space instance

gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: Union[tuple, dict, ndarray]) → Union[tuple, dict, ndarray]#

gymnasium.vector.utils.concatenate(space: MultiBinary, items, out)

gymnasium.vector.utils.concatenate(space: MultiDiscrete, items, out)

gymnasium.vector.utils.concatenate(space: Discrete, items, out)

gymnasium.vector.utils.concatenate(space: Box, items, out)

gymnasium.vector.utils.concatenate(space: Tuple, items, out)

gymnasium.vector.utils.concatenate(space: Dict, items, out)

gymnasium.vector.utils.concatenate(space: Space, items, out)

Concatenate multiple samples from space into a single object.

Example:

>>> from gymnasium.spaces import Box
>>> space = Box(low=0, high=1, shape=(3,), dtype=np.float32)
>>> out = np.zeros((2, 3), dtype=np.float32)
>>> items = [space.sample() for _ in range(2)]
>>> concatenate(space, items, out)
array([[0.6348213 , 0.28607962, 0.60760117],
       [0.87383074, 0.192658  , 0.2148103 ]], dtype=float32)

Parameters:

space – Observation space of a single environment in the vectorized environment.
items – Samples to be concatenated.
out – The output object. This object is a (possibly nested) numpy array.

Returns:

The output object. This object is a (possibly nested)

Raises:

ValueError – Space is not a valid gym.Space instance

gymnasium.vector.utils.iterate(space: Space, items) → Iterator#

gymnasium.vector.utils.iterate(space: Discrete, items)

gymnasium.vector.utils.iterate(space: MultiBinary, items)

gymnasium.vector.utils.iterate(space: MultiDiscrete, items)

gymnasium.vector.utils.iterate(space: Box, items)

gymnasium.vector.utils.iterate(space: Tuple, items)

gymnasium.vector.utils.iterate(space: Dict, items)

gymnasium.vector.utils.iterate(space: Space, items)

Iterate over the elements of a (batched) space.

Example:

>>> from gymnasium.spaces import Box, Dict
>>> space = Dict({
... 'position': Box(low=0, high=1, shape=(2, 3), dtype=np.float32),
... 'velocity': Box(low=0, high=1, shape=(2, 2), dtype=np.float32)})
>>> items = space.sample()
>>> it = iterate(space, items)
>>> next(it)
{'position': array([-0.99644893, -0.08304597, -0.7238421 ], dtype=float32),
'velocity': array([0.35848552, 0.1533453 ], dtype=float32)}
>>> next(it)
{'position': array([-0.67958736, -0.49076623,  0.38661423], dtype=float32),
'velocity': array([0.7975036 , 0.93317133], dtype=float32)}
>>> next(it)
StopIteration

Parameters:

space – Space to which items belong to.
items – Items to be iterated over.

Returns:

Iterator over the elements in `items`.

Raises:

ValueError – Space is not an instance of gym.Space

Shared Memory Utils#

gymnasium.vector.utils.create_empty_array(space: Space, n: int = 1, fn: callable = np.zeros) → Union[tuple, dict, ndarray]#

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.multi_binary.MultiBinary, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.multi_discrete.MultiDiscrete, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.discrete.Discrete, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.box.Box, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.tuple.Tuple, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.dict.Dict, n=1, fn=<built-in function zeros>)

gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.space.Space, n=1, fn=<built-in function zeros>)

Create an empty (possibly nested) numpy array.

Example:

>>> from gymnasium.spaces import Box, Dict
>>> space = Dict({
... 'position': Box(low=0, high=1, shape=(3,), dtype=np.float32),
... 'velocity': Box(low=0, high=1, shape=(2,), dtype=np.float32)})
>>> create_empty_array(space, n=2, fn=np.zeros)
OrderedDict([('position', array([[0., 0., 0.],
                                 [0., 0., 0.]], dtype=float32)),
             ('velocity', array([[0., 0.],
                                 [0., 0.]], dtype=float32))])

Parameters:

space – Observation space of a single environment in the vectorized environment.
n – Number of environments in the vectorized environment. If None, creates an empty sample from space.
fn – Function to apply when creating the empty numpy array. Examples of such functions are np.empty or np.zeros.

Returns:

The output object. This object is a (possibly nested)

Raises:

ValueError – Space is not a valid gym.Space instance

gymnasium.vector.utils.create_shared_memory(space: Space, n: int = 1, ctx=mp) → Union[dict, tuple, Array]#

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.multi_binary.MultiBinary, n: int = 1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.multi_discrete.MultiDiscrete, n: int = 1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.discrete.Discrete, n: int = 1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.box.Box, n: int = 1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.tuple.Tuple, n: int = 1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

gymnasium.vector.utils.create_shared_memory(space: ~gymnasium.spaces.dict.Dict, n=1, ctx=<module 'multiprocessing' from '/opt/hostedtoolcache/Python/3.9.15/x64/lib/python3.9/multiprocessing/__init__.py'>)

Create a shared memory object, to be shared across processes.

This eventually contains the observations from the vectorized environment.

Parameters:

space – Observation space of a single environment in the vectorized environment.
n – Number of environments in the vectorized environment (i.e. the number of processes).
ctx – The multiprocess module

Returns:

shared_memory for the shared object across processes.

Raises:

CustomSpaceError – Space is not a valid gymnasium.Space instance

gymnasium.vector.utils.read_from_shared_memory(space: Space, shared_memory: Union[dict, tuple, Array], n: int = 1) → Union[dict, tuple, ndarray]#

gymnasium.vector.utils.read_from_shared_memory(space: MultiBinary, shared_memory, n: int = 1)

gymnasium.vector.utils.read_from_shared_memory(space: MultiDiscrete, shared_memory, n: int = 1)

gymnasium.vector.utils.read_from_shared_memory(space: Discrete, shared_memory, n: int = 1)

gymnasium.vector.utils.read_from_shared_memory(space: Box, shared_memory, n: int = 1)

gymnasium.vector.utils.read_from_shared_memory(space: Tuple, shared_memory, n: int = 1)

gymnasium.vector.utils.read_from_shared_memory(space: Dict, shared_memory, n: int = 1)

Read the batch of observations from shared memory as a numpy array.

..notes::: The numpy array objects returned by read_from_shared_memory shares the memory of shared_memory. Any changes to shared_memory are forwarded to observations, and vice-versa. To avoid any side-effect, use np.copy.

Parameters:

space – Observation space of a single environment in the vectorized environment.
shared_memory – Shared object across processes. This contains the observations from the vectorized environment. This object is created with create_shared_memory.
n – Number of environments in the vectorized environment (i.e. the number of processes).

Returns:

Batch of observations as a (possibly nested)

Raises:

CustomSpaceError – Space is not a valid gymnasium.Space instance

gymnasium.vector.utils.write_to_shared_memory(space: Space, index: int, value: ndarray, shared_memory: Union[dict, tuple, Array])#

gymnasium.vector.utils.write_to_shared_memory(space: MultiBinary, index, value, shared_memory)

gymnasium.vector.utils.write_to_shared_memory(space: MultiDiscrete, index, value, shared_memory)

gymnasium.vector.utils.write_to_shared_memory(space: Discrete, index, value, shared_memory)

gymnasium.vector.utils.write_to_shared_memory(space: Box, index, value, shared_memory)

gymnasium.vector.utils.write_to_shared_memory(space: Tuple, index, values, shared_memory)

gymnasium.vector.utils.write_to_shared_memory(space: Dict, index, values, shared_memory)

Write the observation of a single environment into shared memory.

Parameters:

space – Observation space of a single environment in the vectorized environment.
index – Index of the environment (must be in [0, num_envs)).
value – Observation of the single environment to write to shared memory.
shared_memory – Shared object across processes. This contains the observations from the vectorized environment. This object is created with create_shared_memory.

Raises:

CustomSpaceError – Space is not a valid gymnasium.Space instance