Add client node names to reported metrics

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 <key>`` 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 <metric>[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)

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']}")
 

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)
 

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

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):

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),

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)
 
 

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

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)
 
 

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

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

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
 
 

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)