Skip to content

Documentation for Metric Module

TrustMetricManager

Manager class to help store the output directory and handle calls from the FL framework.

Source code in nebula/addons/trustworthiness/metric.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class TrustMetricManager:
    """
    Manager class to help store the output directory and handle calls from the FL framework.
    """

    def __init__(self, scenario_start_time, federation, participant=None):
        if federation == "DFL" or federation == "SDFL":
            self.federation_prefix = "dfl"
            self.factsheet_file_nm = f"factsheet_participant_{participant}.json"
            self.eval_metrics_file_nm = "eval_metrics_dfl.json"
            self.nebula_trust_results_nm = f"nebula_trust_results_{participant}.json"
            self.scenario_start_time = scenario_start_time
        else:
            self.federation_prefix = "cfl"
            self.factsheet_file_nm = "factsheet.json"
            self.eval_metrics_file_nm = "eval_metrics_cfl.json"
            self.nebula_trust_results_nm = "nebula_trust_results.json"
            self.scenario_start_time = scenario_start_time

    def evaluate(self, experiment_name, weights, use_weights=False):
        """
        Evaluates the trustworthiness score.

        Args:
            scenario (object): The scenario in whith the trustworthiness will be calculated.
            weights (dict): The desired weghts of the pillars.
            use_weights (bool): True to turn on the weights in the metric config file, default to False.
        """
        # Get scenario name
        scenario_name = experiment_name
        factsheet_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.factsheet_file_nm)
        results_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.nebula_trust_results_nm)

        if not os.path.exists(factsheet_file):
            logger.error(f"{factsheet_file} is missing! Please check documentation.")
            return

        with open(factsheet_file, "r") as f:
            factsheet = json.load(f)

        metrics_cfg_file = get_eval_metrics_file(self.federation_prefix, factsheet, self.eval_metrics_file_nm)

        if not os.path.exists(metrics_cfg_file):
            logger.error(f"{metrics_cfg_file} is missing! Please check documentation.")
            return

        with open(metrics_cfg_file, "r") as m:
            metrics_cfg = json.load(m)
            metrics = metrics_cfg.items()
            input_docs = {"factsheet": factsheet}

            result_json = {"trust_score": 0, "pillars": []}
            final_score = 0
            result_print = []
            for key, value in metrics:
                pillar = TrustPillar(key, value, input_docs, use_weights, user_weights=weights)
                score, result = pillar.evaluate()
                weight = weights.get(key) / 100
                final_score += weight * score
                result_print.append([key, score])
                result_json["pillars"].append(result)
            final_score = round(final_score, 2)
            result_json["trust_score"] = final_score
            write_results_json(results_file, result_json)

            graphics = Graphics(self.scenario_start_time, scenario_name)
            graphics.graphics()

    def evaluate_participant(self, experiment_name, weights, participant_id, use_weights=False):
        """
        Evaluates the trustworthiness score.

        Args:
            scenario (object): The scenario in whith the trustworthiness will be calculated.
            weights (dict): The desired weghts of the pillars.
            use_weights (bool): True to turn on the weights in the metric config file, default to False.
        """
        # Get scenario name
        scenario_name = experiment_name
        factsheet_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.factsheet_file_nm)
        results_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.nebula_trust_results_nm)

        if not os.path.exists(factsheet_file):
            logger.error(f"{factsheet_file} is missing! Please check documentation.")
            return

        with open(factsheet_file, "r") as f:
            factsheet = json.load(f)

        metrics_cfg_file = get_eval_metrics_file(self.federation_prefix, factsheet, self.eval_metrics_file_nm)

        if not os.path.exists(metrics_cfg_file):
            logger.error(f"{metrics_cfg_file} is missing! Please check documentation.")
            return

        with open(metrics_cfg_file, "r") as m:
            raw_metrics_cfg: str = m.read()
            raw_metrics_cfg = raw_metrics_cfg.replace("factsheet", f"factsheet_participant_{participant_id}")
            metrics_cfg = json.loads(raw_metrics_cfg)

            metrics = metrics_cfg.items()
            input_docs = {f"factsheet_participant_{participant_id}": factsheet}

            result_json = {"trust_score": 0, "pillars": []}
            final_score = 0
            result_print = []
            for key, value in metrics:
                pillar = TrustPillar(key, value, input_docs, use_weights, user_weights=weights)
                score, result = pillar.evaluate()
                weight = weights.get(key) / 100
                final_score += weight * score
                result_print.append([key, score])
                result_json["pillars"].append(result)
            final_score = round(final_score, 2)
            result_json["trust_score"] = final_score
            write_results_json(results_file, result_json)

            graphics = Graphics(self.scenario_start_time, scenario_name, participant_id)
            graphics.graphics_dfl(participant_id)

evaluate(experiment_name, weights, use_weights=False)

Evaluates the trustworthiness score.

Parameters:

Name Type Description Default
scenario object

The scenario in whith the trustworthiness will be calculated.

 required
weights dict

The desired weghts of the pillars.

 required
use_weights bool

True to turn on the weights in the metric config file, default to False.

 False
Source code in nebula/addons/trustworthiness/metric.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def evaluate(self, experiment_name, weights, use_weights=False):
    """
    Evaluates the trustworthiness score.

    Args:
        scenario (object): The scenario in whith the trustworthiness will be calculated.
        weights (dict): The desired weghts of the pillars.
        use_weights (bool): True to turn on the weights in the metric config file, default to False.
    """
    # Get scenario name
    scenario_name = experiment_name
    factsheet_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.factsheet_file_nm)
    results_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.nebula_trust_results_nm)

    if not os.path.exists(factsheet_file):
        logger.error(f"{factsheet_file} is missing! Please check documentation.")
        return

    with open(factsheet_file, "r") as f:
        factsheet = json.load(f)

    metrics_cfg_file = get_eval_metrics_file(self.federation_prefix, factsheet, self.eval_metrics_file_nm)

    if not os.path.exists(metrics_cfg_file):
        logger.error(f"{metrics_cfg_file} is missing! Please check documentation.")
        return

    with open(metrics_cfg_file, "r") as m:
        metrics_cfg = json.load(m)
        metrics = metrics_cfg.items()
        input_docs = {"factsheet": factsheet}

        result_json = {"trust_score": 0, "pillars": []}
        final_score = 0
        result_print = []
        for key, value in metrics:
            pillar = TrustPillar(key, value, input_docs, use_weights, user_weights=weights)
            score, result = pillar.evaluate()
            weight = weights.get(key) / 100
            final_score += weight * score
            result_print.append([key, score])
            result_json["pillars"].append(result)
        final_score = round(final_score, 2)
        result_json["trust_score"] = final_score
        write_results_json(results_file, result_json)

        graphics = Graphics(self.scenario_start_time, scenario_name)
        graphics.graphics()

evaluate_participant(experiment_name, weights, participant_id, use_weights=False)

Evaluates the trustworthiness score.

Parameters:

Name Type Description Default
scenario object

The scenario in whith the trustworthiness will be calculated.

 required
weights dict

The desired weghts of the pillars.

 required
use_weights bool

True to turn on the weights in the metric config file, default to False.

 False
Source code in nebula/addons/trustworthiness/metric.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def evaluate_participant(self, experiment_name, weights, participant_id, use_weights=False):
    """
    Evaluates the trustworthiness score.

    Args:
        scenario (object): The scenario in whith the trustworthiness will be calculated.
        weights (dict): The desired weghts of the pillars.
        use_weights (bool): True to turn on the weights in the metric config file, default to False.
    """
    # Get scenario name
    scenario_name = experiment_name
    factsheet_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.factsheet_file_nm)
    results_file = os.path.join(os.environ.get('NEBULA_LOGS_DIR'), scenario_name, "trustworthiness", self.nebula_trust_results_nm)

    if not os.path.exists(factsheet_file):
        logger.error(f"{factsheet_file} is missing! Please check documentation.")
        return

    with open(factsheet_file, "r") as f:
        factsheet = json.load(f)

    metrics_cfg_file = get_eval_metrics_file(self.federation_prefix, factsheet, self.eval_metrics_file_nm)

    if not os.path.exists(metrics_cfg_file):
        logger.error(f"{metrics_cfg_file} is missing! Please check documentation.")
        return

    with open(metrics_cfg_file, "r") as m:
        raw_metrics_cfg: str = m.read()
        raw_metrics_cfg = raw_metrics_cfg.replace("factsheet", f"factsheet_participant_{participant_id}")
        metrics_cfg = json.loads(raw_metrics_cfg)

        metrics = metrics_cfg.items()
        input_docs = {f"factsheet_participant_{participant_id}": factsheet}

        result_json = {"trust_score": 0, "pillars": []}
        final_score = 0
        result_print = []
        for key, value in metrics:
            pillar = TrustPillar(key, value, input_docs, use_weights, user_weights=weights)
            score, result = pillar.evaluate()
            weight = weights.get(key) / 100
            final_score += weight * score
            result_print.append([key, score])
            result_json["pillars"].append(result)
        final_score = round(final_score, 2)
        result_json["trust_score"] = final_score
        write_results_json(results_file, result_json)

        graphics = Graphics(self.scenario_start_time, scenario_name, participant_id)
        graphics.graphics_dfl(participant_id)