openrl.envs.vec_env.utils package

Submodules

openrl.envs.vec_env.utils.numpy_utils module

Numpy utility functions: concatenate space samples and create empty array.

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.space.Space, items: Iterable, out: Union[tuple, dict, numpy.ndarray]) → Union[tuple, dict, numpy.ndarray]

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.multi_binary.MultiBinary, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.multi_discrete.MultiDiscrete, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.discrete.Discrete, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.box.Box, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.tuple.Tuple, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.dict.Dict, items, out)

openrl.envs.vec_env.utils.numpy_utils.concatenate(space: gymnasium.spaces.space.Space, items, out)

Concatenate multiple samples from space into a single object.

Args:

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) numpy array.

Raises:

ValueError: Space is not a valid gym.Space instance

Example:

>>> 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)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.space.Space, n: int = 1, agent_num=1, fn: Callable[[...], numpy.ndarray] = <built-in function zeros>) → Union[tuple, dict, numpy.ndarray]

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.multi_binary.MultiBinary, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.multi_discrete.MultiDiscrete, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.box.Box, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.discrete.Discrete, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.tuple.Tuple, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.dict.Dict, n=1, agent_num=1, fn=<built-in function zeros>)

openrl.envs.vec_env.utils.numpy_utils.create_empty_array(space: gymnasium.spaces.space.Space, n=1, agent_num=1, fn=<built-in function zeros>)

Create an empty (possibly nested) numpy array.

Args:

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) numpy array.

Raises:

ValueError: Space is not a valid gym.Space instance

Example:

>>> 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)
OrderedDict([('position', array([[0., 0., 0.],
       [0., 0., 0.]], dtype=float32)), ('velocity', array([[0., 0.],
       [0., 0.]], dtype=float32))])

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.space.Space, actions) → Iterator

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.discrete.Discrete, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.multi_binary.MultiBinary, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.multi_discrete.MultiDiscrete, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.box.Box, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.tuple.Tuple, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.dict.Dict, actions)

openrl.envs.vec_env.utils.numpy_utils.iterate_action(space: gymnasium.spaces.space.Space, actions)

Iterate over the elements of a batched actions.

Args:

space: Space to which actions belong to. actions: actions to be iterated over.

Returns:

Iterator over the elements in actions.

Raises:

ValueError: Space is not an instance of gym.Space

openrl.envs.vec_env.utils.numpy_utils.single_random_action(space: gymnasium.spaces.space.Space, action_mask: Optional[Union[List[int], numpy.ndarray]] = None) → Union[tuple, dict, numpy.ndarray]

openrl.envs.vec_env.utils.numpy_utils.single_random_action(space: gymnasium.spaces.discrete.Discrete, action_mask: Optional[Union[List[int], numpy.ndarray]] = None)

openrl.envs.vec_env.utils.numpy_utils.single_random_action(space: gymnasium.spaces.box.Box, action_mask: Optional[Union[List[int], numpy.ndarray]] = None)

openrl.envs.vec_env.utils.share_memory module

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.space.Space, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>) → Union[dict, tuple, multiprocessing.context.BaseContext.Array]

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.multi_binary.MultiBinary, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.multi_discrete.MultiDiscrete, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.discrete.Discrete, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.box.Box, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.tuple.Tuple, n: int = 1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

openrl.envs.vec_env.utils.share_memory.create_shared_memory(space: gymnasium.spaces.dict.Dict, n=1, agent_num: int = 1, ctx=<module 'multiprocessing' from '/home/docs/.asdf/installs/python/3.11.6/lib/python3.11/multiprocessing/__init__.py'>)

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

This eventually contains the observations from the vectorized environment.

Args:

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). agent_num: Number of agents in each environment. ctx: The multiprocess module

Returns:

shared_memory for the shared object across processes.

Raises:

CustomSpaceError: Space is not a valid gymnasium.Space instance

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.space.Space, shared_memory: Union[dict, tuple, multiprocessing.context.BaseContext.Array], n: int = 1, agent_num: int = 1) → Union[dict, tuple, numpy.ndarray]

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.multi_binary.MultiBinary, shared_memory, n: int = 1, agent_num: int = 1)

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.multi_discrete.MultiDiscrete, shared_memory, n: int = 1, agent_num: int = 1)

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.discrete.Discrete, shared_memory, n: int = 1, agent_num: int = 1)

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.box.Box, shared_memory, n: int = 1, agent_num: int = 1)

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.tuple.Tuple, shared_memory, n: int = 1, agent_num: int = 1)

openrl.envs.vec_env.utils.share_memory.read_from_shared_memory(space: gymnasium.spaces.dict.Dict, shared_memory, n: int = 1, agent_num: 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.

Args:

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) numpy array.

Raises:

CustomSpaceError: Space is not a valid gymnasium.Space instance

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.space.Space, agent_num: int, index: int, value: numpy.ndarray, shared_memory: Union[dict, tuple, multiprocessing.context.BaseContext.Array])

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.multi_binary.MultiBinary, agent_num, index, value, shared_memory)

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.multi_discrete.MultiDiscrete, agent_num, index, value, shared_memory)

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.discrete.Discrete, agent_num, index, value, shared_memory)

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.box.Box, agent_num, index, value, shared_memory)

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.tuple.Tuple, agent_num, index, values, shared_memory)

openrl.envs.vec_env.utils.share_memory.write_to_shared_memory(space: gymnasium.spaces.dict.Dict, agent_num, index, values, shared_memory)

Write the observation of a single environment into shared memory.

Args:

space: Observation space of a single environment in the vectorized environment. agent_num: Number of agents in the 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

openrl.envs.vec_env.utils.util module

openrl.envs.vec_env.utils.util.prepare_action_masks(info: Optional[List[Dict[str, Any]]] = None, agent_num: int = 1, as_batch: bool = True) → Optional[numpy.ndarray]

openrl.envs.vec_env.utils.util.tile_images(img_nhwc: Sequence[numpy.ndarray]) → numpy.ndarray

Tile N images into one big PxQ image (P,Q) are chosen to be as close as possible, and if N is square, then P=Q.

Parameters

img_nhwc – list or array of images, ndim=4 once turned into array. img nhwc n = batch index, h = height, w = width, c = channel

Returns

img_HWc, ndim=3

Module contents