From c9047ea68d32d9efd833070053a32b6113e3c5bb Mon Sep 17 00:00:00 2001 From: faildeny Date: Thu, 29 Oct 2026 10:54:44 +0100 Subject: [PATCH] Fix weighted random forest support --- .github/workflows/tests.yml | 2 +- .../FedCustomAggregator.py | 9 +++- .../models/weighted_random_forest/client.py | 46 ++++++------------- .../models/weighted_random_forest/server.py | 14 +++--- tests/test_federated_training.py | 5 +- 5 files changed, 33 insertions(+), 43 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8d31f9d..0985075 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -4,7 +4,7 @@ on: push: branches: [main] pull_request: - branches: [main, cmd_line] + branches: [main, cmd_line, tests_and_CI] workflow_dispatch: concurrency: diff --git a/flcore/models/weighted_random_forest/FedCustomAggregator.py b/flcore/models/weighted_random_forest/FedCustomAggregator.py index e2b194c..aca9aad 100644 --- a/flcore/models/weighted_random_forest/FedCustomAggregator.py +++ b/flcore/models/weighted_random_forest/FedCustomAggregator.py @@ -136,7 +136,14 @@ def aggregate_fit( #ndarrays_to_parameters necessary to send the message - parameters_aggregated = serialize_RF(aggregation_result) + #each result is a ragged tuple, so wrap it in an object array + packed_result = [] + for res in aggregation_result: + packed = np.empty(len(res), dtype=object) + for k, item in enumerate(res): + packed[k] = item + packed_result.append(packed) + parameters_aggregated = serialize_RF(packed_result) #DropOut Center: initially aggregate all execution times of all clients diff --git a/flcore/models/weighted_random_forest/client.py b/flcore/models/weighted_random_forest/client.py index cd58500..7133bb3 100644 --- a/flcore/models/weighted_random_forest/client.py +++ b/flcore/models/weighted_random_forest/client.py @@ -21,7 +21,7 @@ import flcore.datasets as datasets from flcore.serialization_funs import serialize_RF, deserialize_RF import flcore.models.weighted_random_forest.utils as utils -from flcore.performance import measurements_metrics +from flcore.metrics import calculate_metrics from flwr.common import ( Code, EvaluateIns, @@ -37,14 +37,14 @@ from mlxtend.classifier import EnsembleVoteClassifier #Ensamble in the level of RF trees -def ensambleRFTrees(parameters): +def ensambleRFTrees(parameters, n_classes): list_classifiers = [None] * len(parameters) weights_classifiers = [None] * len(parameters) for i in range(len(parameters)): model = utils.get_model(True) model.estimators_= parameters[i][0][0] - model.n_classes_ =2 + model.n_classes_ = n_classes model.n_outputs_ = 1 model.classes_ = np.array([j for j in range(model.n_classes_)]) list_classifiers[i] = model @@ -88,19 +88,20 @@ def ensambleDecisionTrees(parameters): # Define Flower client class MnistClient(fl.client.Client): def __init__(self, data,config): + self.config = config self.node_name = config["node_name"] n_folds_out=config['num_rounds'] seed=42 # Load data (self.X_train, self.y_train), (self.X_test, self.y_test) = data - self.splits_nested = datasets.split_partitions(n_folds_out,0.2, seed, self.X_train, self.y_train) - self.bal_RF = config['weighted_random_forest']['balanced_rf'] + self.splits_nested = datasets.split_partitions(n_folds_out, config['test_size'], config['seed'], self.X_train, self.y_train, config['task']) + self.bal_RF = str(config['balanced']).lower() == 'true' self.model = utils.get_model(self.bal_RF) # Setting initial parameters, akin to model.compile for keras models utils.set_initial_params_client(self.model,self.X_train, self.y_train) self.ensamble_tree = [] self.weight_ensamble_tree = [] - self.levelOfDetail = config['weighted_random_forest']['levelOfDetail'] + self.levelOfDetail = config['levelOfDetail'] def get_parameters(self, ins: GetParametersIns): # , config type: ignore params = utils.get_model_parameters(self.model) @@ -137,20 +138,9 @@ def fit(self, ins: FitIns): # , parameters, config type: ignore # accuracy,specificity,sensitivity,balanced_accuracy, precision, F1_score = \ # measurements_metrics(self.model,X_val, y_val) # measurements_metrics outputs accuracy, specificity, sensitivity, balanced_accuracy, precision, F1_score - metrics_res = measurements_metrics(self.model, X_val, y_val) - accuracy = metrics_res[0] - specificity = metrics_res[1] - sensitivity = metrics_res[2] - balanced_accuracy = metrics_res[3] - precision = metrics_res[4] - F1_score = metrics_res[5] - print(f"Accuracy client in fit: {accuracy}") - print(f"Sensitivity client in fit: {sensitivity}") - print(f"Specificity client in fit: {specificity}") - print(f"Balanced_accuracy in fit: {balanced_accuracy}") - print(f"precision in fit: {precision}") - print(f"F1_score in fit: {F1_score}") - + metrics = calculate_metrics(y_val, self.model.predict(X_val), self.config) + print(f"Metrics client in fit: {metrics}") + ellapsed_time = (time.time() - start_time) print(f"num_client {self.node_name} has an ellapsed time {ellapsed_time}") @@ -188,7 +178,7 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore #self.ensamble_tree = np.concatenate(((self.ensamble_tree),(list_classifiers))) #self.weight_ensamble_tree = np.concatenate(((self.weight_ensamble_tree),(weights_classifiers))) else: - list_classifiers,weights_classifiers = ensambleRFTrees(parameters) + list_classifiers,weights_classifiers = ensambleRFTrees(parameters, len(self.model.classes_)) #Merge the trees of all clients in each round self.ensamble_tree = (self.ensamble_tree)+(list_classifiers) @@ -205,15 +195,9 @@ def evaluate(self, ins: EvaluateIns): # , parameters, config type: ignore #utils.set_model_params(self.model, parameters) y_pred_prob = eclf.predict_proba(self.X_test) - loss = log_loss(self.y_test, y_pred_prob) - accuracy,specificity,sensitivity,balanced_accuracy, precision, F1_score = \ - measurements_metrics(eclf,self.X_test, self.y_test) - print(f"Accuracy client in evaluate: {accuracy}") - print(f"Sensitivity client in evaluate: {sensitivity}") - print(f"Specificity client in evaluate: {specificity}") - print(f"Balanced_accuracy in evaluate: {balanced_accuracy}") - print(f"precision in evaluate: {precision}") - print(f"F1_score in evaluate: {F1_score}") + loss = log_loss(self.y_test, y_pred_prob, labels=self.model.classes_) + metrics = calculate_metrics(self.y_test, eclf.predict(self.X_test), self.config) + print(f"Metrics client in evaluate: {metrics}") # Serialize to send it to the server #params = get_model_parameters(model) @@ -224,7 +208,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=metrics, ) except Exception as e: from flcore.utils import log_detailed_error diff --git a/flcore/models/weighted_random_forest/server.py b/flcore/models/weighted_random_forest/server.py index 877b871..051a3b1 100644 --- a/flcore/models/weighted_random_forest/server.py +++ b/flcore/models/weighted_random_forest/server.py @@ -32,7 +32,7 @@ def fit_round( server_round: int ) -> Dict: def get_server_and_strategy(config): - bal_RF = config['weighted_random_forest']['balanced_rf'] + bal_RF = str(config['balanced']).lower() == 'true' model = get_model(bal_RF) utils.set_initial_params_server( model) @@ -40,20 +40,20 @@ def get_server_and_strategy(config): #strategy = fl.server.strategy.FedAvg( strategy = FedCustom( #Have running the same number of clients otherwise it does not run the federated - min_available_clients = config['num_clients'], - min_fit_clients = config['num_clients'], - min_evaluate_clients = config['num_clients'], + min_available_clients = config['min_available_clients'], + min_fit_clients = config['min_fit_clients'], + min_evaluate_clients = config['min_evaluate_clients'], #enable evaluate_fn if we have data to evaluate in the server #evaluate_fn = utils_RF.get_evaluate_fn( model ), #no data in server evaluate_metrics_aggregation_fn = utils.evaluate_metrics_aggregation_fn, on_fit_config_fn = fit_round ) #Select normal RF or Balanced RF from config - strategy.bal_RF= config['weighted_random_forest']['balanced_rf'] + strategy.bal_RF= bal_RF strategy.dropout_method = config['dropout_method'] - strategy.percentage_drop = config['dropout']['percentage_drop'] + strategy.percentage_drop = config['dropout_percentage'] strategy.smoothing_method = config['smooth_method'] - strategy.smoothing_strenght = config['smoothWeights']['smoothing_strenght'] + strategy.smoothing_strenght = config['smoothing_strenght'] return None, strategy diff --git a/tests/test_federated_training.py b/tests/test_federated_training.py index dcd02b2..0ad1739 100644 --- a/tests/test_federated_training.py +++ b/tests/test_federated_training.py @@ -93,10 +93,9 @@ def case(model, scenario, client_extra=(), id_suffix="", bug=None): ), case("random_forest", "regression"), ] -wrf_bug = "weighted_random_forest server reads config['weighted_random_forest'], never set by the CLI" CASES += [ - case("weighted_random_forest", "binary", bug=wrf_bug), - case("weighted_random_forest", "multiclass", bug=wrf_bug), + case("weighted_random_forest", "binary"), + case("weighted_random_forest", "multiclass"), ] CASES += [case("xgb", s, bug=XGB_BUG) for s in ("binary", "multiclass", "regression")] CASES += [case("nn", s) for s in ("binary", "multiclass", "regression")]