colosseum.emission_maps.image_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 ImageEncoding(EmissionMap):
14    """
15    The `ImageEncoding` emission map creates a non-tabular matrix representation that encodes the visual representation
16    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(ImageEncoding, 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.array(
44            list(map(np.vectorize(lambda x: self._symbol_mapping[x]), grid))
45        ).astype(np.float32)
46        if self._mdp.is_episodic():
47            x = in_episode_time + np.zeros(obs.shape[1])
48            return np.vstack((x, obs))
49        return obs
class ImageEncoding(colosseum.emission_maps.base.EmissionMap): 
14class ImageEncoding(EmissionMap):
15    """
16    The `ImageEncoding` emission map creates a non-tabular matrix representation that encodes the visual representation
17    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(ImageEncoding, 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.array(
45            list(map(np.vectorize(lambda x: self._symbol_mapping[x]), grid))
46        ).astype(np.float32)
47        if self._mdp.is_episodic():
48            x = in_episode_time + np.zeros(obs.shape[1])
49            return np.vstack((x, obs))
50        return obs

The ImageEncoding emission map creates a non-tabular matrix representation that encodes the visual representation of the MDP.

ImageEncoding( 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(ImageEncoding, 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.array(
45            list(map(np.vectorize(lambda x: self._symbol_mapping[x]), grid))
46        ).astype(np.float32)
47        if self._mdp.is_episodic():
48            x = in_episode_time + np.zeros(obs.shape[1])
49            return np.vstack((x, obs))
50        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