colosseum.benchmark.run

  1import os
  2import shutil
  3from typing import TYPE_CHECKING, Dict, List, Type, Tuple, Iterable
  4
  5from colosseum import config
  6from colosseum.benchmark.benchmark import ColosseumBenchmark
  7from colosseum.benchmark.utils import (
  8    instantiate_benchmark_folder,
  9    instantiate_agent_configs,
 10)
 11from colosseum.experiment.experiment_instance import ExperimentInstance
 12from colosseum.experiment.experiment_instances import (
 13    get_experiment_instances_from_folder,
 14)
 15from colosseum.utils import ensure_folder
 16
 17if TYPE_CHECKING:
 18    from colosseum.agent.agents.base import BaseAgent
 19
 20
 21def instantiate_agents_and_benchmark(
 22    agents_configs: Dict[Type["BaseAgent"], str],
 23    benchmark: ColosseumBenchmark,
 24    overwrite_previous_experiment: bool = False,
 25    experiment_folder: str = None,
 26) -> str:
 27    """
 28    instantiate the benchmark and the agents configs locally.
 29
 30    Parameters
 31    ----------
 32    agents_configs : Dict[Type["BaseAgent"], str]
 33        The dictionary associates agent classes to their gin config files.
 34    benchmark : ColosseumBenchmark
 35        The benchmark to be instantiated.
 36    overwrite_previous_experiment : bool
 37        If True the destination folder is cleared before instantiating the MDP configs. If False, it raises an error if
 38        it finds a different set of agents configs, MDP configs, or `ExperimentalConfig` in the destination folder.
 39    experiment_folder : str
 40        The folder where to instantiate the benchmark and the agents configs. By default, it is taken from the package
 41        configurations.
 42
 43    Returns
 44    -------
 45    str
 46        The folder where the benchmark and the agents configs have been instantiated.
 47    """
 48
 49    # Avoid mixing episodic and continuous settings
 50    assert all(
 51        agent_class.is_episodic() == list(agents_configs)[0].is_episodic()
 52        for agent_class in agents_configs
 53    )
 54    assert all(
 55        mdp_configs.is_episodic() == list(agents_configs)[0].is_episodic()
 56        for mdp_configs in benchmark.mdps_gin_configs
 57    )
 58
 59    # Set the current experiment folder of the given benchmark
 60    benchmark_folder = (
 61        config.get_experiments_folder()
 62        if experiment_folder is None
 63        else ensure_folder(experiment_folder)
 64    ) + benchmark.name
 65
 66    if overwrite_previous_experiment:
 67        # Remove any previous experiment directory
 68        shutil.rmtree(benchmark_folder, ignore_errors=True)
 69        os.makedirs(benchmark_folder)
 70
 71    # Instantiate the mdp configs
 72    instantiate_benchmark_folder(benchmark, benchmark_folder)
 73
 74    # Instantiate the agent configs
 75    instantiate_agent_configs(agents_configs, benchmark_folder)
 76
 77    return benchmark_folder
 78
 79
 80def instantiate_and_get_exp_instances_from_benchmark(
 81    agents_configs: Dict[Type["BaseAgent"], str],
 82    benchmark: ColosseumBenchmark,
 83    overwrite_previous_experiment: bool = False,
 84    experiment_folder: str = None,
 85) -> List[ExperimentInstance]:
 86    """
 87    instantiate the benchmark and the agents configs locally, and creates the corresponding `ExperimentInstance`.
 88
 89    Parameters
 90    ----------
 91    agents_configs : Dict[Type["BaseAgent"], str]
 92        The dictionary associates agent classes to their gin config files.
 93    benchmark : ColosseumBenchmark
 94        The benchmark to be instantiated.
 95    overwrite_previous_experiment : bool
 96        If True the destination folder is cleared before instantiating the MDP configs. If False, it raises an error if
 97        it finds a different set of agents configs, MDP configs, or `ExperimentalConfig` in the destination folder.
 98    experiment_folder : str
 99        The folder where to instantiate the benchmark and the agents configs. By default, it is taken from the package
100        configurations.
101
102    Returns
103    -------
104    List[ExperimentInstance]
105        The `ExperimentInstance`s corresponding to the benchmark and the agents configs.
106    """
107
108    # Instantiate the local directories for the benchmark
109    benchmark_folder = instantiate_agents_and_benchmark(
110        agents_configs, benchmark, overwrite_previous_experiment, experiment_folder
111    )
112
113    # Create the ExperimentInstance objects
114    return get_experiment_instances_from_folder(benchmark_folder)
115
116
117def instantiate_and_get_exp_instances_from_agents_and_benchmarks(
118    agents_and_benchmarks: Iterable[
119        Tuple[Dict[Type["BaseAgent"], str], ColosseumBenchmark]
120    ],
121    overwrite_previous_experiment: bool = False,
122    experiment_folder: str = None,
123) -> List[ExperimentInstance]:
124    """
125    instantiate the benchmarks_and_agents and the agents configs locally, and creates the corresponding `ExperimentInstance`.
126
127    Parameters
128    ----------
129    agents_and_benchmarks : Iterable[Tuple[Dict[Type["BaseAgent"], str], ColosseumBenchmark]]
130        The agent configs and benchmarks_and_agents to be instantiated.
131    overwrite_previous_experiment : bool
132        If True the destination folder is cleared before instantiating the MDP configs. If False, it raises an error if
133        it finds a different set of agents configs, MDP configs, or `ExperimentalConfig` in the destination folder.
134    experiment_folder : str
135        The folder where to instantiate the benchmark and the agents configs. By default, it is taken from the package
136        configurations.
137
138    Returns
139    -------
140    List[ExperimentInstance]
141        The `ExperimentInstance`s corresponding to the benchmarks_and_agents and the agents configs.
142    """
143
144    experiment_instances = []
145    for agents_configs, benchmark in agents_and_benchmarks:
146        experiment_instances += instantiate_and_get_exp_instances_from_benchmark(
147            agents_configs, benchmark, overwrite_previous_experiment, experiment_folder
148        )
149    return experiment_instances
150
151
152def instantiate_and_get_exp_instances_from_agents_and_benchmarks_for_hyperopt(
153    agents_and_benchmarks: Iterable[
154        Tuple[Dict[Type["BaseAgent"], str], ColosseumBenchmark]
155    ],
156    overwrite_previous_experiment: bool = False,
157) -> List[ExperimentInstance]:
158    """
159    instantiate the benchmarks_and_agents and the agents configs for the parameters optimization locally (by
160    checking the parameters optimization folder in the package configurations), and creates the corresponding
161    `ExperimentInstance`.
162
163    Parameters
164    ----------
165    agents_and_benchmarks : Iterable[Tuple[Dict[Type["BaseAgent"], str], ColosseumBenchmark]]
166        The agent configs and benchmarks_and_agents to be instantiated.
167    overwrite_previous_experiment : bool
168        If True the destination folder is cleared before instantiating the MDP configs. If False, it raises an error if
169        it finds a different set of agents configs, MDP configs, or `ExperimentalConfig` in the destination folder.
170
171    Returns
172    -------
173    List[ExperimentInstance]
174        The `ExperimentInstance`s corresponding to the benchmarks_and_agents and the agents configs.
175    """
176    return instantiate_and_get_exp_instances_from_agents_and_benchmarks(
177        agents_and_benchmarks,
178        overwrite_previous_experiment,
179        config.get_hyperopt_folder(),
180    )