colosseum.agent.agents.random

  1import abc
  2from typing import Any, Dict, TYPE_CHECKING
  3
  4import numpy as np
  5from ray import tune
  6
  7from colosseum.agent.actors import RandomActor
  8from colosseum.agent.agents.base import BaseAgent
  9from colosseum.utils.acme.specs import DiscreteArray, MDPSpec
 10
 11if TYPE_CHECKING:
 12    from colosseum.emission_maps import EmissionMap
 13
 14
 15class RandomAgent(BaseAgent, abc.ABC):
 16    """
 17    The `RandomAgent` implements a uniformly randomly acting agent.
 18    """
 19
 20    @staticmethod
 21    def is_emission_map_accepted(emission_map: "EmissionMap") -> bool:
 22        return True
 23
 24    @staticmethod
 25    def get_hyperparameters_search_spaces() -> Dict[str, tune.sample.Domain]:
 26        return {}
 27
 28    @staticmethod
 29    def produce_gin_file_from_parameters(parameters: Dict[str, Any], index: int = 0):
 30        return ""
 31
 32    @property
 33    def current_optimal_stochastic_policy(self) -> np.ndarray:
 34        return self._policy
 35
 36    def __init__(self, seed: int, mdp_specs: MDPSpec):
 37        """
 38        Parameters
 39        ----------
 40        seed : int
 41            The random seed.
 42        mdp_specs : MDPSpec
 43            The full specification of the MDP.
 44        """
 45        super(RandomAgent, self).__init__(
 46            seed,
 47            mdp_specs,
 48            mdp_model=None,
 49            actor=RandomActor(seed, mdp_specs),
 50            optimization_horizon=0,
 51        )
 52
 53        if type(self._mdp_spec.observations) == DiscreteArray:
 54            if type(self._mdp_spec.actions) == DiscreteArray:
 55                n_s = self._mdp_spec.observations.num_values
 56                n_a = self._mdp_spec.actions.num_values
 57
 58                self._policy = (
 59                    np.ones(
 60                        (n_s, n_a)
 61                        if self._time_horizon == np.inf
 62                        else (self._time_horizon, n_s, n_a)
 63                    )
 64                    / n_a
 65                )
 66        else:
 67            raise NotImplementedError(
 68                "The RandomAgent is implemented for discrete MDP only."
 69            )
 70
 71    def episode_end_update(self):
 72        pass
 73
 74    def before_start_interacting(self):
 75        pass
 76
 77
 78class RandomAgentEpisodic(RandomAgent):
 79    @staticmethod
 80    def get_agent_instance_from_parameters(
 81        seed: int,
 82        optimization_horizon: int,
 83        mdp_specs: MDPSpec,
 84        parameters: Dict[str, Any],
 85    ) -> "BaseAgent":
 86        return RandomAgentEpisodic(seed, mdp_specs)
 87
 88    @staticmethod
 89    def is_episodic() -> bool:
 90        return True
 91
 92
 93class RandomAgentContinuous(RandomAgent):
 94    @staticmethod
 95    def get_agent_instance_from_parameters(
 96        seed: int,
 97        optimization_horizon: int,
 98        mdp_specs: MDPSpec,
 99        parameters: Dict[str, Any],
100    ) -> "BaseAgent":
101        return RandomAgentContinuous(seed, mdp_specs)
102
103    @staticmethod
104    def is_episodic() -> bool:
105        return False