MEDfl.LearningManager package

Submodules

MEDfl.LearningManager.client module

class MEDfl.LearningManager.client.FlowerClient(cid: str, local_model: Model, trainloader: DataLoader, valloader: DataLoader, diff_priv: bool = True)

Bases: NumPyClient

FlowerClient class for creating MEDfl clients.

cid

Client ID.

Type:

str

local_model

Local model of the federated learning network.

Type:

Model

trainloader

DataLoader for training data.

Type:

DataLoader

valloader

DataLoader for validation data.

Type:

DataLoader

diff_priv

Flag indicating whether to use differential privacy.

Type:

bool

__init__(cid: str, local_model: Model, trainloader: DataLoader, valloader: DataLoader, diff_priv: bool = True)

Initializes the FlowerClient instance.

Parameters:

• cid (str) – Client ID.

• local_model (Model) – Local model of the federated learning network.

• trainloader (DataLoader) – DataLoader for training data.

• valloader (DataLoader) – DataLoader for validation data.

• diff_priv (bool) – Flag indicating whether to use differential privacy.

evaluate(parameters, config)

Evaluates the local model on the validation data and returns the loss and accuracy.

Parameters:

• parameters – Parameters received from the server.

• config – Configuration information.

Returns:

Loss, number of validation examples, and accuracy information.

Return type:

Tuple

fit(parameters, config)

Fits the local model to the received parameters using federated learning.

Parameters:

• parameters – Parameters received from the server.

• config – Configuration information.

Returns:

Parameters of the local model, number of training examples, and privacy information.

Return type:

Tuple

get_parameters(config)

Returns the current parameters of the local model.

Parameters:

config – Configuration information.

Returns:

Parameters of the local model.

Return type:

Numpy array

validate()

Validates cid, local_model, trainloader, valloader.

MEDfl.LearningManager.dynamicModal module

class MEDfl.LearningManager.dynamicModal.DynamicModel

Bases: object

DynamicModel class for creating various types of neural network models.

static create_binary_classifier(input_dim, hidden_dims, output_dim, activation='relu', dropout_rate=0.0, batch_norm=False, use_gpu=False)

Creates a binary classifier neural network model with customizable architecture.

Parameters:

• input_dim (int) – Dimension of the input data.

• hidden_dims (List[int]) – List of dimensions for hidden layers.

• output_dim (int) – Dimension of the output (number of classes).

• activation (str, optional) – Activation function for hidden layers. Default is ‘relu’.

• dropout_rate (float, optional) – Dropout rate for regularization. Default is 0.0 (no dropout).

• batch_norm (bool, optional) – Whether to apply batch normalization. Default is False.

• use_gpu (bool, optional) – Whether to use GPU acceleration. Default is False.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_convolutional_neural_network(input_channels, output_channels, kernel_size, use_gpu=False)

Creates a convolutional neural network (CNN) model.

Parameters:

• input_channels (int) – Number of input channels.

• output_channels (int) – Number of output channels.

• kernel_size (int) – Size of the convolutional kernel.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_linear_regressor(input_dim, output_dim, use_gpu=False)

Creates a linear regressor neural network model.

Parameters:

• input_dim (int) – Dimension of the input data.

• output_dim (int) – Dimension of the output.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_logistic_regressor(input_dim, use_gpu=False)

Creates a logistic regressor neural network model.

Parameters:

input_dim (int) – Dimension of the input data.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_lstm_network(input_size, hidden_size, use_gpu=False)

Creates a Long Short-Term Memory (LSTM) network model.

Parameters:

• input_size (int) – Size of the input layer.

• hidden_size (int) – Size of the hidden layer.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

create_model(model_type: str, params_dict={}) → Module

Create a specific type of model dynamically based on the given parameters.

Parameters:

• model_type (str) – Type of the model to create (‘Binary Classifier’, ‘Multiclass Classifier’, ‘Linear Regressor’, ‘Logistic Regressor’, ‘SVM’, ‘Neural Network Classifier’, ‘Convolutional Neural Network’, ‘Recurrent Neural Network’, ‘LSTM Network’, ‘Autoencoder’).

• params_dict (dict) – Dictionary containing parameters for model creation.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_multiclass_classifier(input_dim, hidden_dims, output_dim, activation='relu', dropout_rate=0.0, batch_norm=False, use_gpu=False)

Creates a multiclass classifier neural network model with customizable architecture.

Parameters:

• input_dim (int) – Dimension of the input data.

• hidden_dims (List[int]) – List of dimensions for hidden layers.

• output_dim (int) – Dimension of the output (number of classes).

• activation (str, optional) – Activation function for hidden layers. Default is ‘relu’.

• dropout_rate (float, optional) – Dropout rate for regularization. Default is 0.0 (no dropout).

• batch_norm (bool, optional) – Whether to apply batch normalization. Default is False.

• use_gpu (bool, optional) – Whether to use GPU acceleration. Default is False.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

static create_recurrent_neural_network(input_size, hidden_size, use_gpu=False)

Creates a recurrent neural network (RNN) model.

Parameters:

• input_size (int) – Size of the input.

• hidden_size (int) – Size of the hidden layer.

Returns:

Created PyTorch model.

Return type:

torch.nn.Module

MEDfl.LearningManager.federated_dataset module

class MEDfl.LearningManager.federated_dataset.FederatedDataset(name: str, train_nodes: list, test_nodes: list, trainloaders: list, valloaders: list, testloaders: list)

Bases: object

__init__(name: str, train_nodes: list, test_nodes: list, trainloaders: list, valloaders: list, testloaders: list)

Represents a Federated Dataset.

Parameters:

• name – Name of the Federated Dataset.

• train_nodes – List of train nodes.

• test_nodes – List of test nodes.

• trainloaders – List of train data loaders.

• valloaders – List of validation data loaders.

• testloaders – List of test data loaders.

create(FLsetupId: int)

Create a new Federated Dataset in the database.

Parameters:

FLsetupId – The FLsetup ID associated with the Federated Dataset.

update(FLpipeId: int, FedId: int)

Update the FLpipe ID associated with the Federated Dataset in the database.

Parameters:

• FLpipeId – The new FLpipe ID to be updated.

• FedId – The Federated Dataset ID.

MEDfl.LearningManager.flpipeline module

class MEDfl.LearningManager.flpipeline.FLpipeline(name: str, description: str, server: FlowerServer)

Bases: object

FLpipeline class for managing Federated Learning pipelines.

name

The name of the FLpipeline.

Type:

str

description

A description of the FLpipeline.

Type:

str

server

The FlowerServer object associated with the FLpipeline.

Type:

FlowerServer

__init__(self, name

str, description: str, server: FlowerServer) -> None: Initialize FLpipeline with the specified name, description, and server.

auto_test(test_frac=1) → List[dict]

Automatically test the FLpipeline on all nodes with the specified test_frac.

Parameters:

test_frac (float, optional) – The fraction of the test data to use. Default is 1.

Returns:

A list of dictionaries containing the node names and the classification reports.

Return type:

List[dict]

create(result: str) → None

Create a new FLpipeline entry in the database with the given result.

Parameters:

result (str) – The result string to store in the database.

delete() → None

Delete the FLpipeline entry from the database based on its name.

Note: This is a placeholder method and needs to be implemented based on your specific database setup.

test_by_node(node_name: str, test_frac=1) → dict

Test the FLpipeline by node with the specified test_frac.

Parameters:

• node_name (str) – The name of the node to test.

• test_frac (float, optional) – The fraction of the test data to use. Default is 1.

Returns:

A dictionary containing the node name and the classification report.

Return type:

dict

validate() → None

Validate the name, description, and server attributes. :raises TypeError: If the name is not a string, the description is not a string, or the server is not a FlowerServer object.

MEDfl.LearningManager.flpipeline.create_query(name, description, creation_date, result)

MEDfl.LearningManager.model module

class MEDfl.LearningManager.model.Model(model: Module, optimizer: Optimizer, criterion: Callable)

Bases: object

Model class for training and testing PyTorch neural networks.

model

PyTorch neural network.

Type:

torch.nn.Module

optimizer

PyTorch optimizer.

Type:

torch.optim.Optimizer

criterion

Loss function.

Type:

Callable

__init__(model: Module, optimizer: Optimizer, criterion: Callable) → None

Initialize Model class with the specified model, optimizer, and criterion.

Parameters:

• model (torch.nn.Module) – PyTorch neural network.

• optimizer (torch.optim.Optimizer) – PyTorch optimizer.

• criterion (Callable) – Loss function.

evaluate(val_loader, device=device(type='cpu')) → Tuple[float, float]

Evaluate the model on the given validation data.

Parameters:

• val_loader – The data loader for validation data.

• device – The device on which to perform the evaluation. Default is ‘cpu’.

Returns:

The evaluation loss and accuracy.

Return type:

Tuple[float, float]

get_parameters() → List[ndarray]

Get the parameters of the model as a list of NumPy arrays.

Returns:

The parameters of the model as a list of NumPy arrays.

Return type:

List[np.ndarray]

static load_model(model_path: str)

Loads a PyTorch model from a file.

Parameters:

model_path (str) – Path to the model file to be loaded.

Returns:

Loaded PyTorch model.

Return type:

torch.nn.Module

static save_model(model, model_name: str)

Saves a PyTorch model to a file.

Parameters:

• model (torch.nn.Module) – PyTorch model to be saved.

• model_name (str) – Name of the model file.

Raises:

Exception – If there is an issue during the saving process.

Returns:

None

set_parameters(parameters: List[ndarray]) → None

Set the parameters of the model from a list of NumPy arrays.

Parameters:

parameters (List[np.ndarray]) – The parameters to be set.

train(train_loader, epoch, device, privacy_engine, diff_priv=False) → float

Train the model on the given train_loader for one epoch.

Parameters:

• train_loader – The data loader for training data.

• epoch (int) – The current epoch number.

• device – The device on which to perform the training.

• privacy_engine – The privacy engine used for differential privacy (if enabled).

• diff_priv (bool, optional) – Whether differential privacy is used. Default is False.

Returns:

The value of epsilon used in differential privacy.

Return type:

float

validate() → None

Validate model and optimizer.

MEDfl.LearningManager.plot module

class MEDfl.LearningManager.plot.AccuracyLossPlotter(results_dict)

Bases: object

A utility class for plotting accuracy and loss metrics based on experiment results.

Parameters:

results_dict (dict) – Dictionary containing experiment results organized by parameters and metrics.

results_dict

Dictionary containing experiment results organized by parameters and metrics.

Type:

dict

parameters

List of unique parameters in the experiment results.

Type:

list

metrics

List of unique metrics in the experiment results.

Type:

list

iterations

Range of iterations (rounds or epochs) in the experiment.

Type:

range

__init__(results_dict)

Initialize the AccuracyLossPlotter with experiment results.

Parameters:

results_dict (dict) – Dictionary containing experiment results organized by parameters and metrics.

plot_accuracy_loss()

Plot accuracy and loss metrics for different parameters.

static plot_classification_report(pipeline_name: str)

Plot a comparison of classification report metrics between nodes.

Parameters:

pipeline_name (str) – Name of the pipeline.

Returns:

None

static plot_confusion_Matrix_by_node(node_name: str, pipeline_name: str)

Plot a confusion matrix for a specific node in the pipeline.

Parameters:

• node_name (str) – Name of the node.

• pipeline_name (str) – Name of the pipeline.

Returns:

None

static plot_global_confusion_matrix(pipeline_name: str)

Plot a global confusion matrix based on pipeline results.

Parameters:

pipeline_name (str) – Name of the pipeline.

Returns:

None

MEDfl.LearningManager.plot.results_dict = {('LR: 0.001, Optimizer: Adam', 'accuracy'): [0.85, 0.89, 0.92, 0.94, Ellipsis], ('LR: 0.001, Optimizer: Adam', 'loss'): [0.2, 0.15, 0.1, 0.08, Ellipsis], ('LR: 0.01, Optimizer: SGD', 'accuracy'): [0.88, 0.91, 0.93, 0.95, Ellipsis], ('LR: 0.01, Optimizer: SGD', 'loss'): [0.18, 0.13, 0.09, 0.07, Ellipsis], ('LR: 0.1, Optimizer: Adam', 'accuracy'): [0.82, 0.87, 0.91, 0.93, Ellipsis], ('LR: 0.1, Optimizer: Adam', 'loss'): [0.25, 0.2, 0.15, 0.12, Ellipsis]}

server should have:

#len = num of rounds

self.accuracies self.losses

Client should have

# len = num of epochs self.accuracies self.losses self.epsilons self.deltas

#common things : LR,SGD, Aggregation

MEDfl.LearningManager.server module

class MEDfl.LearningManager.server.FlowerServer(global_model: Model, strategy: Strategy, num_rounds: int, num_clients: int, fed_dataset: FederatedDataset, diff_privacy: bool = False, client_resources: Optional[Dict[str, float]] = {'num_cpus': 1, 'num_gpus': 0.0})

Bases: object

A class representing the central server for Federated Learning using Flower.

global_model

The global model that will be federated among clients.

Type:

Model

strategy

The strategy used for federated learning, specifying communication and aggregation methods.

Type:

Strategy

num_rounds

The number of federated learning rounds to perform.

Type:

int

num_clients

The number of clients participating in the federated learning process.

Type:

int

fed_dataset

The federated dataset used for training and evaluation.

Type:

FederatedDataset

diff_priv

Whether differential privacy is used during the federated learning process.

Type:

bool

accuracies

A list to store the accuracy of the global model during each round.

Type:

List[float]

losses

A list to store the loss of the global model during each round.

Type:

List[float]

flower_clients

A list to store the FlowerClient objects representing individual clients.

Type:

List[FlowerClient]

__init__(global_model: Model, strategy: Strategy, num_rounds: int, num_clients: int, fed_dataset: FederatedDataset, diff_privacy: bool = False, client_resources: Optional[Dict[str, float]] = {'num_cpus': 1, 'num_gpus': 0.0}) → None

Initialize a FlowerServer object with the specified parameters.

Parameters:

• global_model (Model) – The global model that will be federated among clients.

• strategy (Strategy) – The strategy used for federated learning, specifying communication and aggregation methods.

• num_rounds (int) – The number of federated learning rounds to perform.

• num_clients (int) – The number of clients participating in the federated learning process.

• fed_dataset (FederatedDataset) – The federated dataset used for training and evaluation.

• diff_privacy (bool, optional) – Whether differential privacy is used during the federated learning process. Default is False.

client_fn(cid) → FlowerClient

Return a FlowerClient object for a specific client ID.

Parameters:

cid – The client ID.

Returns:

A FlowerClient object representing the individual client.

Return type:

FlowerClient

evaluate(server_round: int, parameters: list[numpy.ndarray[Any, numpy.dtype[Any]]], config: Dict[str, Union[bool, bytes, float, int, str]]) → Optional[Tuple[float, Dict[str, Union[bool, bytes, float, int, str]]]]

Evaluate the global model on the validation dataset and update the accuracies and losses.

Parameters:

• server_round (int) – The current round of the federated learning process.

• parameters (fl.common.NDArrays) – The global model parameters.

• config (Dict[str, fl.common.Scalar]) – Configuration dictionary.

Returns:

The evaluation loss and accuracy.

Return type:

Optional[Tuple[float, Dict[str, fl.common.Scalar]]]

run() → None

Run the federated learning process using Flower simulation.

Returns:

The history of the accuracies and losses during the training of each node

Return type:

History

validate() → None

Validate global_model, strategy, num_clients, num_rounds, fed_dataset, diff_privacy

MEDfl.LearningManager.strategy module

class MEDfl.LearningManager.strategy.Strategy(name: str = 'FedAvg', fraction_fit: float = 1.0, fraction_evaluate: float = 1.0, min_fit_clients: int = 2, min_evaluate_clients: int = 2, min_available_clients: int = 2, initial_parameters=[], evaluation_methode='centralized')

Bases: object

A class representing a strategy for Federated Learning.

name

The name of the strategy. Default is “FedAvg”.

Type:

str

fraction_fit

Fraction of clients to use for training during each round. Default is 1.0.

Type:

float

fraction_evaluate

Fraction of clients to use for evaluation during each round. Default is 1.0.

Type:

float

min_fit_clients

Minimum number of clients to use for training during each round. Default is 2.

Type:

int

min_evaluate_clients

Minimum number of clients to use for evaluation during each round. Default is 2.

Type:

int

min_available_clients

Minimum number of available clients required to start a round. Default is 2.

Type:

int

initial_parameters

The initial parameters of the server model

Type:

Optional[]

Methods:

__init__(name: str = 'FedAvg', fraction_fit: float = 1.0, fraction_evaluate: float = 1.0, min_fit_clients: int = 2, min_evaluate_clients: int = 2, min_available_clients: int = 2, initial_parameters=[], evaluation_methode='centralized') → None

Initialize a Strategy object with the specified parameters.

Parameters:

• name (str) – The name of the strategy. Default is “FedAvg”.

• fraction_fit (float) – Fraction of clients to use for training during each round. Default is 1.0.

• fraction_evaluate (float) – Fraction of clients to use for evaluation during each round. Default is 1.0.

• min_fit_clients (int) – Minimum number of clients to use for training during each round. Default is 2.

• min_evaluate_clients (int) – Minimum number of clients to use for evaluation during each round. Default is 2.

• min_available_clients (int) – Minimum number of available clients required to start a round. Default is 2.

• initial_parameters (Optional[]) – The initial parametres of the server model

• evaluation_methode ("centralized" | "distributed") –

create_strategy()

get_strategy_by_name()

optuna_fed_optimization(direction: str, hpo_rate: int, params_config)

MEDfl.LearningManager.utils module

MEDfl.LearningManager.utils.create_MEDfl_db()

MEDfl.LearningManager.utils.custom_classification_report(y_true, y_pred_prob)

Compute custom classification report metrics including accuracy, sensitivity, specificity, precision, NPV, F1-score, false positive rate, and true positive rate.

Parameters:

• y_true (array-like) – True labels.

• y_pred (array-like) – Predicted labels.

Returns:

A dictionary containing custom classification report metrics.

Return type:

dict

MEDfl.LearningManager.utils.empty_db()

Empty the database by deleting records from multiple tables and resetting auto-increment counters.

Returns:

None

MEDfl.LearningManager.utils.get_node_confusion_matrix(pipeline_id, node_name)

Get the confusion matrix for a specific node in a pipeline based on test results.

Parameters:

• pipeline_id (int) – ID of the pipeline.

• node_name (str) – Name of the node.

Returns:

A dictionary representing the confusion matrix for the specified node.

Return type:

dict

MEDfl.LearningManager.utils.get_pipeline_confusion_matrix(pipeline_id)

Get the global confusion matrix for a pipeline based on test results.

Parameters:

pipeline_id (int) – ID of the pipeline.

Returns:

A dictionary representing the global confusion matrix.

Return type:

dict

MEDfl.LearningManager.utils.get_pipeline_from_name(name)

Get the pipeline ID from its name in the database.

Parameters:

name (str) – Name of the pipeline.

Returns:

ID of the pipeline.

Return type:

int

MEDfl.LearningManager.utils.get_pipeline_result(pipeline_id)

Get the test results for a pipeline.

Parameters:

pipeline_id (int) – ID of the pipeline.

Returns:

DataFrame containing test results for the specified pipeline.

Return type:

pandas.DataFrame

MEDfl.LearningManager.utils.load_db_config()

Read a dictionary from an environment variable.

MEDfl.LearningManager.utils.load_db_config_dep()

MEDfl.LearningManager.utils.set_db_config(config_path)

MEDfl.LearningManager.utils.set_db_config_dep(config_path)

MEDfl.LearningManager.utils.test(model, test_loader, device=device(type='cpu'))

Evaluate a model using a test loader and return a custom classification report.

Parameters:

• model (torch.nn.Module) – PyTorch model to evaluate.

• test_loader (torch.utils.data.DataLoader) – DataLoader for the test dataset.

• device (torch.device, optional) – Device for model evaluation. Default is “cpu”.

Returns:

A dictionary containing custom classification report metrics.

Return type:

dict

Module contents