实用函数

向量化空间

gymnasium.vector.utils.batch_space(space: Space[Any], n: int = 1) Space[Any][source]
gymnasium.vector.utils.batch_space(space: Box, n: int = 1)
gymnasium.vector.utils.batch_space(space: Discrete, n: int = 1)
gymnasium.vector.utils.batch_space(space: MultiDiscrete, n: int = 1)
gymnasium.vector.utils.batch_space(space: MultiBinary, n: int = 1)
gymnasium.vector.utils.batch_space(space: Tuple, n: int = 1)
gymnasium.vector.utils.batch_space(space: Dict, n: int = 1)
gymnasium.vector.utils.batch_space(space: Graph | Text | Sequence | OneOf, n: int = 1)
gymnasium.vector.utils.batch_space(space: Graph | Text | Sequence | OneOf, n: int = 1)
gymnasium.vector.utils.batch_space(space: Graph | Text | Sequence | OneOf, n: int = 1)
gymnasium.vector.utils.batch_space(space: Graph | Text | Sequence | OneOf, n: int = 1)
gymnasium.vector.utils.batch_space(space: Graph | Text | Sequence | OneOf, n: int = 1)

批处理大小为 n 且为神经网络优化的空间。

参数:
  • space – 空间(例如,向量化环境中单个环境的观察空间)。

  • n – 要批处理的空间数量(例如,向量化环境中环境的数量)。

返回:

大小为 `n` 的批处理空间。

引发:

ValueError – 无法批处理没有注册函数的空间。

示例

>>> from gymnasium.spaces import Box, Dict
>>> import numpy as np
>>> 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(0.0, 1.0, (5, 3), float32), 'velocity': Box(0.0, 1.0, (5, 2), float32))
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: tuple[Any, ...] | dict[str, Any] | ndarray) tuple[Any, ...] | dict[str, Any] | ndarray[source]
gymnasium.vector.utils.concatenate(space: Box | Discrete | MultiDiscrete | MultiBinary, items: Iterable, out: ndarray) ndarray
gymnasium.vector.utils.concatenate(space: Box | Discrete | MultiDiscrete | MultiBinary, items: Iterable, out: ndarray) ndarray
gymnasium.vector.utils.concatenate(space: Box | Discrete | MultiDiscrete | MultiBinary, items: Iterable, out: ndarray) ndarray
gymnasium.vector.utils.concatenate(space: Box | Discrete | MultiDiscrete | MultiBinary, items: Iterable, out: ndarray) ndarray
gymnasium.vector.utils.concatenate(space: Tuple, items: Iterable, out: tuple[Any, ...]) tuple[Any, ...]
gymnasium.vector.utils.concatenate(space: Dict, items: Iterable, out: dict[str, Any]) dict[str, Any]
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: None) tuple[Any, ...]
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: None) tuple[Any, ...]
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: None) tuple[Any, ...]
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: None) tuple[Any, ...]
gymnasium.vector.utils.concatenate(space: Space, items: Iterable, out: None) tuple[Any, ...]

将来自空间的多个样本连接成单个对象。

参数:
  • space – 每个项的空间(例如,向量化环境中的 single_action_space

  • items – 要连接的样本(例如,所有样本都应是 space 的一个元素)。

  • out – 输出对象(例如,由 create_empty_array 生成)

返回:

输出对象,可以是相同的 out 对象。

引发:

ValueError – Space 不是有效的 gymnasium.Space 实例

示例

>>> from gymnasium.spaces import Box
>>> import numpy as np
>>> space = Box(low=0, high=1, shape=(3,), seed=42, dtype=np.float32)
>>> out = np.zeros((2, 3), dtype=np.float32)
>>> items = [space.sample() for _ in range(2)]
>>> concatenate(space, items, out)
array([[0.77395606, 0.43887845, 0.85859793],
       [0.697368  , 0.09417735, 0.97562236]], dtype=float32)
gymnasium.vector.utils.iterate(space: Space[T_cov], items: T_cov) Iterator[source]
gymnasium.vector.utils.iterate(space: Discrete, items: Iterable)
gymnasium.vector.utils.iterate(space: Box | MultiDiscrete | MultiBinary, items: ndarray)
gymnasium.vector.utils.iterate(space: Box | MultiDiscrete | MultiBinary, items: ndarray)
gymnasium.vector.utils.iterate(space: Box | MultiDiscrete | MultiBinary, items: ndarray)
gymnasium.vector.utils.iterate(space: Tuple, items: tuple[Any, ...])
gymnasium.vector.utils.iterate(space: Dict, items: dict[str, Any])

遍历(批处理)空间的元素。

参数:
  • space – (批处理)空间(例如,向量化环境中的 action_spaceobservation_space)。

  • items – 要遍历的批处理样本(例如,来自空间的样本)。

示例

>>> from gymnasium.spaces import Box, Dict
>>> import numpy as np
>>> space = Dict({
... 'position': Box(low=0, high=1, shape=(2, 3), seed=42, dtype=np.float32),
... 'velocity': Box(low=0, high=1, shape=(2, 2), seed=42, dtype=np.float32)})
>>> items = space.sample()
>>> it = iterate(space, items)
>>> next(it)
{'position': array([0.77395606, 0.43887845, 0.85859793], dtype=float32), 'velocity': array([0.77395606, 0.43887845], dtype=float32)}
>>> next(it)
{'position': array([0.697368  , 0.09417735, 0.97562236], dtype=float32), 'velocity': array([0.85859793, 0.697368  ], dtype=float32)}
>>> next(it)
Traceback (most recent call last):
    ...
StopIteration
gymnasium.vector.utils.create_empty_array(space: Space, n: int = 1, fn: Callable = np.zeros) tuple[Any, ...] | dict[str, Any] | ndarray[source]
gymnasium.vector.utils.create_empty_array(space: Box, n: int = 1, fn=np.zeros) ndarray
gymnasium.vector.utils.create_empty_array(space: Box, n: int = 1, fn=np.zeros) ndarray
gymnasium.vector.utils.create_empty_array(space: Box, n: int = 1, fn=np.zeros) ndarray
gymnasium.vector.utils.create_empty_array(space: Box, n: int = 1, fn=np.zeros) ndarray
gymnasium.vector.utils.create_empty_array(space: Tuple, n: int = 1, fn=np.zeros) tuple[Any, ...]
gymnasium.vector.utils.create_empty_array(space: Dict, n: int = 1, fn=np.zeros) dict[str, Any]
gymnasium.vector.utils.create_empty_array(space: Graph, n: int = 1, fn=np.zeros) tuple[GraphInstance, ...]
gymnasium.vector.utils.create_empty_array(space: Text, n: int = 1, fn=np.zeros) tuple[str, ...]
gymnasium.vector.utils.create_empty_array(space: Sequence, n: int = 1, fn=np.zeros) tuple[Any, ...]
gymnasium.vector.utils.create_empty_array(space: OneOf, n: int = 1, fn=np.zeros)
gymnasium.vector.utils.create_empty_array(space: ~gymnasium.spaces.space.Space, n=1, fn=<built-in function zeros>)

创建一个空的(可能嵌套且通常基于 numpy 的)数组,与 concatenate(..., out=array) 结合使用。

在大多数情况下,数组将包含在批处理空间中,但这不作保证。

参数:
  • space – 向量化环境中单个环境的观察空间。

  • n – 向量化环境中的环境数量。如果为 None,则从 space 创建一个空样本。

  • fn – 创建空 numpy 数组时要应用的函数。此类函数的示例有 np.emptynp.zeros

返回:

输出对象。此对象是(可能嵌套的)

引发:

ValueError – Space 不是有效的 gymnasium.Space 实例

示例

>>> from gymnasium.spaces import Box, Dict
>>> import numpy as np
>>> 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)
{'position': array([[0., 0., 0.],
       [0., 0., 0.]], dtype=float32), 'velocity': array([[0., 0.],
       [0., 0.]], dtype=float32)}

空间的共享内存

gymnasium.vector.utils.create_shared_memory(space: Space[Any], n: int = 1, ctx=mp) dict[str, Any] | tuple[Any, ...] | SynchronizedArray[source]
gymnasium.vector.utils.create_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Tuple, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Dict, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Text, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: OneOf, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Graph | Sequence, n: int = 1, ctx=mp)
gymnasium.vector.utils.create_shared_memory(space: Graph | Sequence, n: int = 1, ctx=mp)

创建共享内存对象,以便在进程间共享。

这最终包含来自向量化环境的观察结果。

参数:
  • space – 向量化环境中单个环境的观察空间。

  • n – 向量化环境中的环境数量(即进程数量)。

  • ctx – 多进程模块

返回:

用于跨进程共享对象的 shared_memory

引发:

CustomSpaceError – Space 不是有效的 gymnasium.Space 实例

gymnasium.vector.utils.read_from_shared_memory(space: Space, shared_memory: dict | tuple | SynchronizedArray, n: int = 1) dict[str, Any] | tuple[Any, ...] | ndarray[source]
gymnasium.vector.utils.read_from_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, shared_memory, n: int = 1)
gymnasium.vector.utils.read_from_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, shared_memory, n: int = 1)
gymnasium.vector.utils.read_from_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, shared_memory, n: int = 1)
gymnasium.vector.utils.read_from_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, 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)
gymnasium.vector.utils.read_from_shared_memory(space: Text, shared_memory, n: int = 1) tuple[str, ...]
gymnasium.vector.utils.read_from_shared_memory(space: OneOf, shared_memory, n: int = 1) tuple[Any, ...]

从共享内存读取批处理观察结果作为 numpy 数组。

注意:

read_from_shared_memory 返回的 numpy 数组对象共享 shared_memory 的内存。对 shared_memory 的任何更改都会转发到 observations,反之亦然。为避免任何副作用,请使用 np.copy

参数:
  • space – 向量化环境中单个环境的观察空间。

  • shared_memory – 跨进程共享的对象。这包含来自向量化环境的观察结果。此对象使用 create_shared_memory 创建。

  • n – 向量化环境中的环境数量(即进程数量)。

返回:

批处理观察结果作为(可能嵌套的)

引发:

CustomSpaceError – Space 不是有效的 gymnasium.Space 实例

gymnasium.vector.utils.write_to_shared_memory(space: Space, index: int, value: ndarray, shared_memory: dict | tuple | SynchronizedArray)[source]
gymnasium.vector.utils.write_to_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, index: int, value, shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, index: int, value, shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, index: int, value, shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Box | Discrete | MultiDiscrete | MultiBinary, index: int, value, shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Tuple, index: int, values: tuple[Any, ...], shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Dict, index: int, values: dict[str, Any], shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: Text, index: int, values: str, shared_memory)
gymnasium.vector.utils.write_to_shared_memory(space: OneOf, index: int, values: tuple[int, Any], shared_memory)

将单个环境的观察结果写入共享内存。

参数:
  • space – 向量化环境中单个环境的观察空间。

  • index – 环境的索引(必须在 [0, num_envs) 范围内)。

  • value – 要写入共享内存的单个环境的观察结果。

  • shared_memory – 跨进程共享的对象。这包含来自向量化环境的观察结果。此对象使用 create_shared_memory 创建。

引发:

CustomSpaceError – Space 不是有效的 gymnasium.Space 实例

杂项

gymnasium.vector.utils.CloudpickleWrapper(fn: Callable[[], Env])[source]

使用 cloudpickle 对结果进行序列化和反序列化的封装器。

gymnasium.vector.utils.clear_mpi_env_vars()[source]

清除 MPI 环境变量。

from mpi4py import MPI 默认会调用 MPI_Init。如果子进程具有 MPI 环境变量,MPI 会认为子进程是一个 MPI 进程,就像父进程一样,并可能导致挂起等不良行为。

此上下文管理器是一种临时清除这些环境变量的取巧方法,例如当我们启动多进程时。

产出:

上下文管理器的产出