colosseum.emission_maps.tabular
1from typing import TYPE_CHECKING 2 3import numpy as np 4 5from colosseum.emission_maps.base import EmissionMap 6 7if TYPE_CHECKING: 8 from colosseum.mdp import BaseMDP, NODE_TYPE 9 10 11class Tabular(EmissionMap): 12 """ 13 The `Tabular` emission map is just the tabular mapping. 14 """ 15 16 @property 17 def is_tabular(self) -> bool: 18 return True 19 20 def __init__(self, mdp: "BaseMDP"): 21 super(Tabular, self).__init__(mdp, None, None) 22 self._mdp = mdp 23 self._cached_obs = dict() 24 25 def node_to_observation( 26 self, node: "NODE_TYPE", in_episode_time: int = None 27 ) -> np.ndarray: 28 raise NotImplementedError() 29 30 def get_observation( 31 self, state: "NODE_TYPE", in_episode_time: int = None 32 ) -> np.ndarray: 33 raise NotImplementedError()
12class Tabular(EmissionMap): 13 """ 14 The `Tabular` emission map is just the tabular mapping. 15 """ 16 17 @property 18 def is_tabular(self) -> bool: 19 return True 20 21 def __init__(self, mdp: "BaseMDP"): 22 super(Tabular, self).__init__(mdp, None, None) 23 self._mdp = mdp 24 self._cached_obs = dict() 25 26 def node_to_observation( 27 self, node: "NODE_TYPE", in_episode_time: int = None 28 ) -> np.ndarray: 29 raise NotImplementedError() 30 31 def get_observation( 32 self, state: "NODE_TYPE", in_episode_time: int = None 33 ) -> np.ndarray: 34 raise NotImplementedError()
The Tabular
emission map is just the tabular mapping.
Tabular(mdp: colosseum.mdp.base.BaseMDP)
21 def __init__(self, mdp: "BaseMDP"): 22 super(Tabular, self).__init__(mdp, None, None) 23 self._mdp = mdp 24 self._cached_obs = dict()
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.
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:
26 def node_to_observation( 27 self, node: "NODE_TYPE", in_episode_time: int = None 28 ) -> np.ndarray: 29 raise NotImplementedError()
Returns
- np.ndarray: The non-tabular representation corresponding to the state given in input. Episodic MDPs also requires the current in-episode time step.
def
get_observation( self, state: 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:
31 def get_observation( 32 self, state: "NODE_TYPE", in_episode_time: int = None 33 ) -> np.ndarray: 34 raise NotImplementedError()
computes the observation numpy array corresponding to the state in input.
Parameters
- state (NODE_TYPE): The state for which we are computing the observation.
- in_episode_time (int): The in episode time. It is ignored in the continuous setting, and, by default, it is set to None.
Returns
- np.ndarray: The observation.