From 2666a4983fe15cca68910a52928ab23851badac4 Mon Sep 17 00:00:00 2001 From: faildeny Date: Thu, 5 Nov 2026 11:26:30 +0100 Subject: [PATCH] Add node name in reported client metrics --- flcore/metrics.py | 31 ++++++++++++------- flcore/models/linear_models/client.py | 4 +-- flcore/models/linear_models/utils.py | 23 ++------------ flcore/models/nn/client.py | 1 + flcore/models/nn/server.py | 10 +++--- flcore/models/random_forest/client.py | 3 ++ flcore/models/random_forest/utils.py | 13 ++------ .../models/weighted_random_forest/client.py | 2 +- flcore/models/weighted_random_forest/utils.py | 13 ++------ flcore/models/xgb/client.py | 1 + flcore/models/xgb/server.py | 7 +++++ server_cmd.py | 7 +++-- tests/fl_runner.py | 7 +++-- 13 files changed, 55 insertions(+), 67 deletions(-) diff --git a/flcore/metrics.py b/flcore/metrics.py index 8523b6b..22f6a03 100644 --- a/flcore/metrics.py +++ b/flcore/metrics.py @@ -112,19 +112,28 @@ def calculate_metrics(y_true, y_pred, config): return metrics +def add_per_client_metrics(metrics, distributed_metrics): + """Add to ``metrics`` one ``per client `` list for every key reported + by the clients (string values, such as ``node_name``, are never averaged). + + Position i of every list belongs to the same client, so + ``per client node_name[i]`` identifies who got ``per client [i]``. + The order may change from round to round, so always read the names of the + same round. + """ + for kn in distributed_metrics[0][1].keys(): + metrics['per client ' + kn] = [res.get(kn) for _, res in distributed_metrics] + metrics['per client n samples'] = [n for n, _ in distributed_metrics] + return metrics + + def metrics_aggregation_fn(distributed_metrics): print(distributed_metrics[0][1].keys()) - keys_names = distributed_metrics[0][1].keys() - keys_names = list(keys_names) + metrics = {} - metrics ={} - - for kn in keys_names: + for kn in distributed_metrics[0][1].keys(): results = [ evaluate_res[kn] for _, evaluate_res in distributed_metrics] - metrics[kn] = np.mean(results) - metrics['per client ' + kn] = results - #print(f"Metric {kn} in aggregation evaluate: {metrics[kn]}\n") - - metrics['per client n samples'] = [res[0] for res in distributed_metrics] + if not any(isinstance(r, (str, bytes)) for r in results): + metrics[kn] = np.mean(results) - return metrics + return add_per_client_metrics(metrics, distributed_metrics) diff --git a/flcore/models/linear_models/client.py b/flcore/models/linear_models/client.py index cc18fb6..891ed18 100644 --- a/flcore/models/linear_models/client.py +++ b/flcore/models/linear_models/client.py @@ -119,6 +119,7 @@ def fit(self, parameters, config): # type: ignore print(f"Training finished for round {self.round}") self.round += 1 + metrics["node_name"] = self.node_name return utils.get_model_parameters(self.model), len(self.X_train), metrics except Exception as e: from flcore.utils import log_detailed_error @@ -162,8 +163,7 @@ def evaluate(self, parameters, config): loss = mean_squared_error(self.y_test, y_pred) metrics["round_time [s]"] = self.round_time - # No tiene sentido agregar el client ID - # metrics["client_id"] = self.node_name + metrics["node_name"] = self.node_name # print(f"Client {self.node_name} Evaluation after aggregated model: {metrics['balanced_accuracy']}") diff --git a/flcore/models/linear_models/utils.py b/flcore/models/linear_models/utils.py index 6661d2b..f79c75f 100644 --- a/flcore/models/linear_models/utils.py +++ b/flcore/models/linear_models/utils.py @@ -179,25 +179,6 @@ def set_initial_params(model,config): #the client using client data and combine #all the metrics of the clients def evaluate_metrics_aggregation_fn(eval_metrics): - print(eval_metrics[0][1].keys()) - keys_names = eval_metrics[0][1].keys() - keys_names = list(keys_names) - - metrics ={} - - for kn in keys_names: - results = [ evaluate_res[kn] for _, evaluate_res in eval_metrics] - metrics[kn] = np.mean(results) - #print(f"Metric {kn} in aggregation evaluate: {metrics[kn]}\n") - - # filename = 'server_results.txt' - # with open( - # filename, - # "a", - # ) as f: - # f.write(f"Accuracy: {metrics['accuracy']} \n") - # f.write(f"Sensitivity: {metrics['sensitivity']} \n") - # f.write(f"Specificity: {metrics['specificity']} \n") - - return metrics + from flcore.metrics import metrics_aggregation_fn + return metrics_aggregation_fn(eval_metrics) diff --git a/flcore/models/nn/client.py b/flcore/models/nn/client.py index a011248..72ebafe 100644 --- a/flcore/models/nn/client.py +++ b/flcore/models/nn/client.py @@ -257,6 +257,7 @@ def evaluate(self, parameters, config): test_loss = total_loss / total dataset_len = self.y_test.shape[0] + metrics["node_name"] = str(self.config.get("node_name")) return float(test_loss), dataset_len, self._clean_metrics(metrics) except Exception as e: from flcore.utils import log_detailed_error diff --git a/flcore/models/nn/server.py b/flcore/models/nn/server.py index f8f8bee..013eeb8 100644 --- a/flcore/models/nn/server.py +++ b/flcore/models/nn/server.py @@ -16,7 +16,7 @@ import flwr as fl import flcore.models.linear_models.utils as utils -from flcore.metrics import metrics_aggregation_fn +from flcore.metrics import metrics_aggregation_fn, add_per_client_metrics from sklearn.metrics import log_loss import joblib from flcore.models.nn.FedCustomAggregator import UncertaintyWeightedFedAvg @@ -33,7 +33,7 @@ def weighted_average(metrics: List[Tuple[int, Metrics]]) -> Metrics: total_examples = sum(num_examples for num_examples, _ in metrics) - metric_keys = metrics[0][1].keys() + metric_keys = [k for k, v in metrics[0][1].items() if not isinstance(v, (str, bytes))] weighted_metrics = {} for key in metric_keys: @@ -42,7 +42,7 @@ def weighted_average(metrics: List[Tuple[int, Metrics]]) -> Metrics: ) weighted_metrics[key] = weighted_sum / total_examples - return weighted_metrics + return add_per_client_metrics(weighted_metrics, metrics) def equal_average(metrics: List[Tuple[int, Metrics]]) -> Metrics: if not metrics: @@ -52,7 +52,7 @@ def equal_average(metrics: List[Tuple[int, Metrics]]) -> Metrics: num_clients = len(metrics) # Asumimos que todas las métricas tienen las mismas keys - metric_keys = metrics[0][1].keys() + metric_keys = [k for k, v in metrics[0][1].items() if not isinstance(v, (str, bytes))] equal_metrics = {} for key in metric_keys: @@ -61,7 +61,7 @@ def equal_average(metrics: List[Tuple[int, Metrics]]) -> Metrics: ) equal_metrics[key] = equal_sum / num_clients - return equal_metrics + return add_per_client_metrics(equal_metrics, metrics) def get_server_and_strategy(config): diff --git a/flcore/models/random_forest/client.py b/flcore/models/random_forest/client.py index ac06ea1..af6f914 100644 --- a/flcore/models/random_forest/client.py +++ b/flcore/models/random_forest/client.py @@ -162,6 +162,7 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore #parameters_updated = serialize_RF(params) # Build and return response status = Status(code=Code.OK, message="Success") + metrics["node_name"] = self.node_name return EvaluateRes( status=status, loss=float(loss), @@ -182,6 +183,7 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore #parameters_updated = serialize_RF(params) # Build and return response status = Status(code=Code.OK, message="Success") + metrics["node_name"] = self.node_name return EvaluateRes( status=status, loss=float(loss), @@ -199,6 +201,7 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore #parameters_updated = serialize_RF(params) # Build and return response status = Status(code=Code.OK, message="Success") + metrics["node_name"] = self.node_name return EvaluateRes( status=status, loss=float(loss), diff --git a/flcore/models/random_forest/utils.py b/flcore/models/random_forest/utils.py index 1d9fc4e..8f2bcda 100644 --- a/flcore/models/random_forest/utils.py +++ b/flcore/models/random_forest/utils.py @@ -94,16 +94,7 @@ def set_initial_params_client(model,X_train, y_train): #the client using client data and combine #all the metrics of the clients def evaluate_metrics_aggregation_fn(eval_metrics): - print(eval_metrics[0][1].keys()) - keys_names = eval_metrics[0][1].keys() - keys_names = list(keys_names) - - metrics ={} - - for kn in keys_names: - results = [ evaluate_res[kn] for _, evaluate_res in eval_metrics] - metrics[kn] = np.mean(results) - - return metrics + from flcore.metrics import metrics_aggregation_fn + return metrics_aggregation_fn(eval_metrics) diff --git a/flcore/models/weighted_random_forest/client.py b/flcore/models/weighted_random_forest/client.py index cd58500..1235120 100644 --- a/flcore/models/weighted_random_forest/client.py +++ b/flcore/models/weighted_random_forest/client.py @@ -224,7 +224,7 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore status=status, loss=float(loss), num_examples=len(self.X_test), - metrics={"accuracy": float(accuracy),"sensitivity":float(sensitivity),"specificity":float(specificity)}, + metrics={"accuracy": float(accuracy),"sensitivity":float(sensitivity),"specificity":float(specificity),"node_name":self.node_name}, ) except Exception as e: from flcore.utils import log_detailed_error diff --git a/flcore/models/weighted_random_forest/utils.py b/flcore/models/weighted_random_forest/utils.py index 026c294..e1d6694 100644 --- a/flcore/models/weighted_random_forest/utils.py +++ b/flcore/models/weighted_random_forest/utils.py @@ -66,16 +66,7 @@ def set_initial_params_client(model: RandomForestClassifier,X_train, y_train): #the client using client data and combine #all the metrics of the clients def evaluate_metrics_aggregation_fn(eval_metrics): - print(eval_metrics[0][1].keys()) - keys_names = eval_metrics[0][1].keys() - keys_names = list(keys_names) - - metrics ={} - - for kn in keys_names: - results = [ evaluate_res[kn] for _, evaluate_res in eval_metrics] - metrics[kn] = np.mean(results) - - return metrics + from flcore.metrics import metrics_aggregation_fn + return metrics_aggregation_fn(eval_metrics) diff --git a/flcore/models/xgb/client.py b/flcore/models/xgb/client.py index 65bcc2d..430ca18 100644 --- a/flcore/models/xgb/client.py +++ b/flcore/models/xgb/client.py @@ -343,6 +343,7 @@ def evaluate( print(f"[Client] Metrics: {metrics}") print(f"[Client] Loss: {loss:.4f}") + metrics["node_name"] = str(self.config.get("node_name")) return loss, num_examples, metrics except Exception as e: from flcore.utils import log_detailed_error diff --git a/flcore/models/xgb/server.py b/flcore/models/xgb/server.py index dbe0327..649265e 100644 --- a/flcore/models/xgb/server.py +++ b/flcore/models/xgb/server.py @@ -285,6 +285,13 @@ def aggregate_evaluate( value * eval_res.num_examples / total_examples ) + # Per client values, identified by node_name + from flcore.metrics import add_per_client_metrics + add_per_client_metrics( + metrics_aggregated, + [(r.num_examples, r.metrics) for _, r in results], + ) + print(f"[Round {server_round}] Eval loss: {avg_loss:.4f}") return avg_loss, metrics_aggregated diff --git a/server_cmd.py b/server_cmd.py index 9b17624..d42f56b 100644 --- a/server_cmd.py +++ b/server_cmd.py @@ -253,11 +253,12 @@ if isinstance(history[logs], dict): for metric in history[logs]: extracted_values = [value for (round, value) in history[logs][metric]] + # Strings (e.g. node_name) are kept as they are, the rest is cast to float + to_plain = lambda value: str(value) if isinstance(value, (str, bytes)) else float(value) if isinstance(extracted_values[0], list): - # Convert list elements to float - extracted_values = [[float(value) for value in sublist] for sublist in extracted_values] + extracted_values = [[to_plain(value) for value in sublist] for sublist in extracted_values] else: - extracted_values = [float(value) for value in extracted_values] + extracted_values = [to_plain(value) for value in extracted_values] history[logs][metric] = extracted_values diff --git a/tests/fl_runner.py b/tests/fl_runner.py index 56a6a60..02520b7 100644 --- a/tests/fl_runner.py +++ b/tests/fl_runner.py @@ -183,10 +183,13 @@ def result(ok, reason, flaky=False): client_files = [open(p, "w") for p in client_logs] try: clients = [] - for f in client_files: + for i, f in enumerate(client_files): clients.append( subprocess.Popen( - client_cmd, cwd=ROOT, stdout=f, stderr=subprocess.STDOUT + [*client_cmd, "--node_name", f"client_{i}"], + cwd=ROOT, + stdout=f, + stderr=subprocess.STDOUT, ) ) processes.extend(clients)