colosseum.hardness.measures.utils

 1import os
 2from glob import glob
 3from typing import TYPE_CHECKING, Any, Dict, Type, Union
 4
 5import yaml
 6
 7from colosseum.utils import cleaner
 8
 9if TYPE_CHECKING:
10    from colosseum.mdp import BaseMDP
11
12
13def find_hardness_report_file(
14    mdp: "BaseMDP", hardness_reports_folder="hardness_reports"
15) -> Union[str, None]:
16    """
17    tries to find a previously calculated hardness report for an MDP instance.
18
19    Parameters
20    ----------
21    mdp : MDP
22        the MDP instance for which it will be looking for the hardness report.
23    hardness_reports_folder : str
24        the folder in which the hardness reports are stored
25    Returns
26    -------
27    Union[str, None]
28        The file path for the cache file or None if no file has been found.
29    """
30    seed_report = glob(f"{hardness_reports_folder}{os.sep}{type(mdp).__name__}_*.yml")
31    for existing_report in seed_report:
32        if type(mdp).__name__ not in existing_report:
33            continue
34        with open(existing_report, "r") as f:
35            report = yaml.load(f, yaml.Loader)
36        same_mdp = True
37        for k, v in report["MDP parameters"].items():
38            if not same_mdp:
39                break
40            same_mdp = cleaner(getattr(mdp, k)) == v
41        for k, v in report["MDP graph metrics"].items():
42            if not same_mdp:
43                break
44            same_mdp = mdp.graph_metrics[k] == v
45        if same_mdp:
46            return existing_report
47    return None
48
49
50def get_average_measure_values(
51    mdp_class: Type["BaseMDP"],
52    mdp_kwargs: Dict[str, Any],
53    n_seeds: int = None,
54    measure_names_max_values: Dict[str, float] = None,
55) -> Union[None, Dict[str, float]]:
56    """
57    computes the values of the measures given in inputs averaged across several seed if necessary. Note that this is
58    only necessary when the MDP structure changes for different seeds.
59
60    Parameters
61    ----------
62    mdp_class : Type["BaseMDP"]
63        The MDP class for which the measures will be computed.
64    mdp_kwargs : Dict[str, Any]
65        The parameters that are used to instantiate the MDP.
66    n_seeds : int, optional
67        The number of seeds for the average. By default, it is set to five when necessary.
68    measure_names_max_values : Dict[str, float], optional
69        The dictionary whose keys are the names of the measure of hardness and the values are the maximum value that
70        we allow them to be. By default, it is set to 'dict(diameter=200, value_norm=5.0)'.
71
72    Returns
73    -------
74    Union[None, Dict[str, float]]
75        The values of the measures. If any of the measure produces a value that is higher than the limit, the function
76        returns None.
77    """
78    if mdp_class.does_seed_change_MDP_structure():
79        n_seeds = 5 if n_seeds is None else n_seeds
80        assert n_seeds > 1
81    else:
82        n_seeds = 1 if n_seeds is None else n_seeds
83        assert n_seeds == 1
84
85    if measure_names_max_values is None:
86        measure_names_max_values = dict(diameter=200, value_norm=5.0)
87    average_values = dict(
88        zip(measure_names_max_values.keys(), [0] * len(measure_names_max_values))
89    )
90
91    for seed in range(n_seeds):
92        mdp = mdp_class(seed=seed, **mdp_kwargs)
93        for m_name, m_max_value in measure_names_max_values.items():
94            m_value = mdp.get_measure_from_name(m_name)
95            if m_value is None or m_value > m_max_value:
96                return None
97            average_values[m_name] += m_value / n_seeds
98    return average_values