colosseum.emission_maps.tensor_encoding 

 1from typing import TYPE_CHECKING, Type, Dict, Any
 2
 3import numpy as np
 4
 5from colosseum.emission_maps.base import EmissionMap
 6from colosseum.emission_maps.base import _get_symbol_mapping
 7
 8if TYPE_CHECKING:
 9    from colosseum.mdp import BaseMDP, NODE_TYPE
10    from colosseum.noises.base import Noise
11
12
13class TensorEncoding(EmissionMap):
14    """
15    The `TensorEncoding` emission map creates a non-tabular 3d numpy array representation that one-hot encodes the
16    visual elements in the visual representation of the MDP.
17    """
18
19    @property
20    def is_tabular(self) -> bool:
21        return False
22
23    def __init__(
24        self,
25        mdp: "BaseMDP",
26        noise_class: Type["Noise"],
27        noise_kwargs: Dict[str, Any],
28    ):
29        self._symbol_mapping = None
30
31        super(TensorEncoding, self).__init__(mdp, noise_class, noise_kwargs)
32
33    def node_to_observation(
34        self, node: "NODE_TYPE", in_episode_time: int = None
35    ) -> np.ndarray:
36        if self._symbol_mapping is None:
37            self._symbol_mapping = _get_symbol_mapping(self._mdp)
38
39        grid = self._mdp.get_grid_representation(node, in_episode_time)
40        if self._mdp.is_episodic():
41            grid = grid[2:]
42
43        obs = np.zeros((*grid.shape, len(self._symbol_mapping)), dtype=np.float32)
44        for i in range(grid.shape[0]):
45            for j in range(grid.shape[1]):
46                obs[i, j, self._symbol_mapping[grid[i, j]]] = 1
47
48        if self._mdp.is_episodic():
49            return np.concatenate(
50                (obs, np.zeros((*grid.shape, 1), np.float32) + in_episode_time), axis=-1
51            )
52        return obs
class TensorEncoding(colosseum.emission_maps.base.EmissionMap): 
14class TensorEncoding(EmissionMap):
15    """
16    The `TensorEncoding` emission map creates a non-tabular 3d numpy array representation that one-hot encodes the
17    visual elements in the visual representation of the MDP.
18    """
19
20    @property
21    def is_tabular(self) -> bool:
22        return False
23
24    def __init__(
25        self,
26        mdp: "BaseMDP",
27        noise_class: Type["Noise"],
28        noise_kwargs: Dict[str, Any],
29    ):
30        self._symbol_mapping = None
31
32        super(TensorEncoding, self).__init__(mdp, noise_class, noise_kwargs)
33
34    def node_to_observation(
35        self, node: "NODE_TYPE", in_episode_time: int = None
36    ) -> np.ndarray:
37        if self._symbol_mapping is None:
38            self._symbol_mapping = _get_symbol_mapping(self._mdp)
39
40        grid = self._mdp.get_grid_representation(node, in_episode_time)
41        if self._mdp.is_episodic():
42            grid = grid[2:]
43
44        obs = np.zeros((*grid.shape, len(self._symbol_mapping)), dtype=np.float32)
45        for i in range(grid.shape[0]):
46            for j in range(grid.shape[1]):
47                obs[i, j, self._symbol_mapping[grid[i, j]]] = 1
48
49        if self._mdp.is_episodic():
50            return np.concatenate(
51                (obs, np.zeros((*grid.shape, 1), np.float32) + in_episode_time), axis=-1
52            )
53        return obs

The TensorEncoding emission map creates a non-tabular 3d numpy array representation that one-hot encodes the visual elements in the visual representation of the MDP.

TensorEncoding( mdp: colosseum.mdp.base.BaseMDP, noise_class: Type[colosseum.noises.base.Noise], noise_kwargs: Dict[str, Any]) 
24    def __init__(
25        self,
26        mdp: "BaseMDP",
27        noise_class: Type["Noise"],
28        noise_kwargs: Dict[str, Any],
29    ):
30        self._symbol_mapping = None
31
32        super(TensorEncoding, self).__init__(mdp, noise_class, noise_kwargs)
Parameters
  • mdp (BaseMDP): The tabular MDP.
  • noise_class (Type["Noise"]): The noise that renders the emission map stochastic.
  • noise_kwargs (Dict[str, Any]): The parameters for the noise class.
is_tabular: bool
Returns
  • bool: The boolean for whether the emission map is tabular.
def node_to_observation( self, node: Union[colosseum.mdp.custom_mdp.CustomNode, colosseum.mdp.river_swim.base.RiverSwimNode, colosseum.mdp.deep_sea.base.DeepSeaNode, colosseum.mdp.frozen_lake.base.FrozenLakeNode, colosseum.mdp.simple_grid.base.SimpleGridNode, colosseum.mdp.minigrid_empty.base.MiniGridEmptyNode, colosseum.mdp.minigrid_rooms.base.MiniGridRoomsNode, colosseum.mdp.taxi.base.TaxiNode], in_episode_time: int = None) -> numpy.ndarray: 
34    def node_to_observation(
35        self, node: "NODE_TYPE", in_episode_time: int = None
36    ) -> np.ndarray:
37        if self._symbol_mapping is None:
38            self._symbol_mapping = _get_symbol_mapping(self._mdp)
39
40        grid = self._mdp.get_grid_representation(node, in_episode_time)
41        if self._mdp.is_episodic():
42            grid = grid[2:]
43
44        obs = np.zeros((*grid.shape, len(self._symbol_mapping)), dtype=np.float32)
45        for i in range(grid.shape[0]):
46            for j in range(grid.shape[1]):
47                obs[i, j, self._symbol_mapping[grid[i, j]]] = 1
48
49        if self._mdp.is_episodic():
50            return np.concatenate(
51                (obs, np.zeros((*grid.shape, 1), np.float32) + in_episode_time), axis=-1
52            )
53        return obs
Returns
  • np.ndarray: The non-tabular representation corresponding to the state given in input. Episodic MDPs also requires the current in-episode time step.
Inherited Members
colosseum.emission_maps.base.EmissionMap
shape
all_observations
get_observation