實用工具函數¶

向量化空間¶

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)

針對神經網路優化的尺寸為 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] | np.ndarray) → tuple[Any, ...] | dict[str, Any] | np.ndarray[source]¶

gymnasium.vector.utils.concatenate(space: Box | Discrete | MultiDiscrete | MultiBinary, items: Iterable, out: np.ndarray) → np.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, ...]

將來自空間的多個樣本串聯成單一物件。

參數:

space – 每個項目的空間（例如，來自向量化環境的 single_action_space）
items – 要串聯的樣本（例如，所有樣本都應為 space 的元素）。
out – 輸出物件（例如，從 create_empty_array 產生）

返回:

輸出物件可以是相同的物件 `out`。

引發:

ValueError – 空間不是有效的 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: np.ndarray)

gymnasium.vector.utils.iterate(space: Tuple, items: tuple[Any, ...])

gymnasium.vector.utils.iterate(space: Dict, items: dict[str, Any])

迭代（批次）空間的元素。

參數:

space – （批次）空間（例如，來自向量化環境的 action_space 或 observation_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] | np.ndarray[source]¶

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.empty 或 np.zeros。

返回:

輸出物件。此物件為（可能是巢狀的）

引發:

ValueError – 空間不是有效的 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, ...] | mp.Array[source]¶

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)

建立一個共享記憶體物件，以便在多個程序之間共享。

這最終會包含來自向量化環境的觀測。

參數:

space – 向量化環境中單個環境的觀測空間。
n – 向量化環境中環境的數量（即程序的數量）。
ctx – 多程序模組

返回:

跨程序的共享物件的 shared_memory。

引發:

CustomSpaceError – 空間不是有效的 gymnasium.Space 實例

gymnasium.vector.utils.read_from_shared_memory(space: Space, shared_memory: dict | tuple | mp.Array, n: int = 1) → dict[str, Any] | tuple[Any, ...] | np.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: 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 陣列傳回。

..notes:: 由 read_from_shared_memory 傳回的 NumPy 陣列物件會共享 shared_memory 的記憶體。對 shared_memory 所做的任何變更都會轉發到 observations，反之亦然。為了避免任何副作用，請使用 np.copy。

參數:

space – 向量化環境中單個環境的觀測空間。
shared_memory – 跨程序的共享物件。這包含來自向量化環境的觀測值。此物件是使用 create_shared_memory 建立的。
n – 向量化環境中環境的數量（即程序的數量）。

返回:

觀測值批次，以（可能是巢狀的）

引發:

CustomSpaceError – 空間不是有效的 gymnasium.Space 實例

gymnasium.vector.utils.write_to_shared_memory(space: Space, index: int, value: np.ndarray, shared_memory: dict[str, Any] | tuple[Any, ...] | mp.Array)[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: 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 – 空間不是有效的 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 程序，就像父程序一樣，並執行錯誤的操作，例如掛起。

此上下文管理器是一種臨時清除這些環境變數的權宜之計，例如當我們啟動多程序處理程序時。

Yields:: 為上下文管理器產生 (Yields)