flex.pool package

Submodules

flex.pool.aggregators module

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.

flex.pool.aggregators.fed_avg(aggregated_weights_as_list: list)[source]: Function that implements the FedAvg aggregation method

Args:

aggregated_weights_as_list (list): List which contains all the weights to aggregate

Returns:

tensor array: An array with the aggregated weights

Example of use assuming you are using a client-server architecture:

from flex.pool.primitive_functions import fed_avg

aggregator = flex_pool.aggregators server = flex_pool.servers aggregator.map(server, fed_avg)

Example of use using the FlexPool without separating server and aggregator, and following a client-server architecture.

from flex.pool.primitive_functions import fed_avg

flex_pool.aggregators.map(flex_pool.servers, fed_avg)

flex.pool.aggregators.fed_avg_f(aggregated_weights_as_list: list)[source]

flex.pool.aggregators.flatten(xs)[source]

flex.pool.aggregators.set_tensorly_backend(aggregated_weights_as_list: list, supported_modules: list | None = None)[source]

flex.pool.aggregators.weighted_fed_avg(aggregated_weights_as_list: list, ponderation: list)[source]: Function that implements the weighted FedAvg aggregation method.

Args:

aggregated_weights_as_list (list): List which contains all the weights to aggregate ponderation (list): weights assigned to each client

Returns:

tensor array: An array with the aggregated weights

Example of use assuming you are using a client-server architecture:

from flex.pool.primitive_functions import weighted_fed_avg

aggregator = flex_pool.aggregators server = flex_pool.servers dummy_poderation = [1.]*len(flex_pool.clients) aggregator.map(server, weighted_fed_avg, ponderation=dummy_poderation)

Example of use using the FlexPool without separating server and aggregator, and following a client-server architecture.

from flex.pool.primitive_functions import weighted_fed_avg dummy_poderation = [1.]*len(flex_pool.clients) flex_pool.aggregators.map(flex_pool.servers, weighted_fed_avg, ponderation=dummy_poderation)

flex.pool.aggregators.weighted_fed_avg_f(aggregated_weights_as_list: list, ponderation: list)[source]

flex.pool.decorators module

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.

flex.pool.decorators.ERROR_MSG_MIN_ARG_GENERATOR(f, min_args)[source]

flex.pool.decorators.aggregate_weights(func)[source]

flex.pool.decorators.collect_clients_weights(func)[source]

flex.pool.decorators.deploy_server_model(func)[source]

flex.pool.decorators.evaluate_server_model(func)[source]

flex.pool.decorators.init_server_model(func)[source]

flex.pool.decorators.set_aggregated_weights(func)[source]

flex.pool.pool module

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.

class flex.pool.pool.FlexPool(flex_data: FedDataset, flex_actors: FlexActors, flex_models: dict[Hashable, FlexModel] | None = None)[source]

Bases: object

Class that orchest the training phase of a federated learning experiment. The FlexPool class is responsible for orchestating the nodes to train a federated model. This class represents a pool of actors and is in charge of checking the communications between them during the process of training a federated model.

Note: At the moment this class only supports Horizontal Federated Learning, but in the future it will cover Vertical Federated Learning and Transfer Learning, so users can simulate all the experiments correctly.

- flex_data (FedDataset)

Type:: The federated dataset prepared to be used.

- flex_actors (FlexActors)

Type:: Actors with its roles.

- flex_models (defaultdict)

Type:: A dictionary containing the each actor id,

and initialized to None. The model to train by each actor will be initialized

using the map function following the communication constraints.

We offer two class methods to create two architectures, client-server architecture and p2p architecture. In the client-server architecture, every id from the FedDataset is assigned to be a client, and we create a third-party actor, supposed to be neutral, to orchestate the training. Meanwhile, in the p2p architecture, each id from the FedDataset will be assigned to be client, server and aggregator. In both cases, the method will create the actors so the user will only have to apply the map function to train the model.

If the user wants to use a different architecture, she will need to create the actors by using the FlexActors class. For example, we let the user create a client-server architecture with multiple aggregators, to carry out the aggregation.

property actor_ids

property aggregators: Property to get all the aggregator available in a pool.

Returns:

FlexPool: Pool containing all the aggregators from a pool

classmethod check_compatibility(src_pool, dst_pool)[source]

Method to check the compatibility between two different pools. This method is used by the map function to check if the function to apply from the source pool to the destination pool can be done.

Parameters:

src_pool (FlexPool) – Source pool. Pool that will send the message.
dst_pool (FlexPool) – Destination pool. Pool that will recieve the message.

Returns:

True if pools are compatible. False in other case.

Return type:

bool

classmethod client_server_pool(fed_dataset: FedDataset, init_func: Callable, server_id: str = 'server', **kwargs)[source]

Method to create a client-server architeture for a FlexDataset given. This functions is used when you have a FlexDataset and you want to start the learning phase following a traditional client-server architecture.

This method will assing to each id from the FlexDataset the client-role, and will create a new actor that will be the server-aggregator that will orchestrate the learning phase.

Args:

fed_dataset (FedDataset): Federated dataset used to train a model.

Returns:

FlexPool: A FlexPool with the assigned roles for a client-server architecture.

property clients: Property to get all the clients available in a pool.

Returns:

FlexPool: Pool containing all the clients from a pool

map(func: Callable, dst_pool: FlexPool | None = None, **kwargs)[source]: Method used to send messages from one pool to another. The pool using this method is the source pool, and it will send a message, apply a function, to the destination pool. If no destination pool is provided, then the function is applied to the source pool. The pool sends a message in order to complete a round in the Federated Learning (FL) paradigm, so, some examples of the messages that will be used by the pools are: - send_model: Aggregators send the model to the server when the aggregation is done. - aggregation_step: Clients send the model to the aggregator so it can apply the aggregation mechanism given. - deploy_model: Server sends the global model to the clients once the weights has been aggregated. - init_model: Server sends the model to train during the learning phase, so the clients can initialize it. This is a particular case from the deploy_model case.

Args:

func (Callable): If dst_pool is None, then message is sent to the source (self). In this situation the function func is called for each actor in the pool, providing actor’s data and actor’s model as arguments in addition to *args and **kwargs. If dst_pool is not None, the message is sent from the source pool (self) to the destination pool (dst_pool). The function func is called for each actor in the pool, providing the model of the current actor in the source pool and all the models of the actors in the destination pool.

dst_pool (FlexPool): Pool that will recieve the message from the source pool (self), it can be None.

Raises:

ValueError: This method raises and error if the pools aren’t allowed to comunicate

Returns:

List[Any]: A list of the result of applying the function (func) from the source pool (self) to the destination pool (dst_pool). If dst_pool is None, then the results come from the source pool. The length of the returned values equals the number of actors in the source pool.

classmethod p2p_pool(fed_dataset: FedDataset, init_func: Callable, **kwargs)[source]

Method to create a peer-to-peer (p2p) architecture for a FlexDataset given. This method is used when you have a FlexDataset and you want to start the learning phase following a p2p architecture.

This method will assing all roles (client-aggregator-server) to every id from the FlexDataset, so each participant in the learning phase can act as client, aggregator and server.

Args:

fed_dataset (FedDataset): Federated dataset used to train a model.

Returns:

FlexPool: A FlexPool with the assigned roles for a p2p architecture.

select(criteria: int | Callable, *criteria_args, **criteria_kwargs)[source]

Function that returns a subset of a FlexPool meeting a certain criteria. If criteria is an integer, a subset of the available nodes of size criteria is randomly sampled. If criteria is a function, then we select those nodes where the function returns True values. Note that, the function must have at least two arguments, a node id and the roles associated to such node id. The actor_id is a string, and the actor_role is a FlexRole object.

Note: This function doesn’t send a copy of the original pool, it sends a reference.: Changes made on the returned pool affect the original pool.

Args:

criteria (int, Callable): if a function is provided, then it must return True/False values for each pair of node_id, node_roles. Additional arguments required for the function are passed in criteria_args and criteria_kwargs. Otherwise, criteria is interpreted as number of nodes to randomly sample from the pool. criteria_args: additional args required for the criteria function. Otherwise ignored. criteria_kwargs: additional keyword args required for the criteria function. Otherwise ignored.

Returns:

FlexPool: a pool that contains the nodes that meet the criteria.

property servers: Property to get all the servers available in a pool.

Returns:

FlexPool: Pool containing all the servers from a pool

validate()[source]: Function that checks whether the object is correct or not.

flex.pool.primitives_pt module

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.

flex.pool.primitives_pt.check_ignored_weights_pt(name, ignore_weights=None)[source]: Checks wether name contains any of the words in ignore_weights.

Args:

name (str): name to check ignore_weights (list, optional): A list of str. Defaults to None.

Returns:

bool: True if any og the elements of list ignore_weights is present in name, otherwise False.

flex.pool.primitives_pt.collect_client_diff_weights_pt(client_flex_model, *args, **kwargs)[source]

Function that collect the weights for a PyTorch model. Particularly,: it collects the difference between the model before and after training, that is, what the model has learnt in its local training step. Also note that the weights of the model before training are assume to be stored using previous_model as key.

This function returns the weights of the model.

Args:

client_flex_model (FlexModel): A client’s FlexModel ignore_weights (list): the name of the weights not to collect, by default, those containind the words num_batches_tracked are not collected, as they only make sense in the local model

Returns:

List: List with the weights of the client’s model

Example of use assuming you are using a client-server architecture:

from flex.pool import collect_client_diff_weights_pt

clients = flex_pool.clients aggregator = flex_pool.aggregators

clients.map(collect_client_diff_weights_pt, aggregator)

Example of using the FlexPool without separating clients and aggregator, and following a client-server architecture.

from flex.pool import collect_client_diff_weights_pt

flex_pool.clients.map(collect_client_diff_weights_pt, flex_pool.aggregators)

flex.pool.primitives_pt.collect_clients_weights_pt(client_flex_model, *args, **kwargs)[source]

Function that collect the weights for a PyTorch model.

This function returns all the weights of the model.

Args:

client_flex_model (FlexModel): A client’s FlexModel ignore_weights (list): the name of the weights not to collect, by default, those containind the words num_batches_tracked are not collected, as they only make sense in the local model

Returns:

List: List with all the weights of the client’s model

Example of use assuming you are using a client-server architecture:

from flex.pool import collect_weights_pt

clients = flex_pool.clients aggregator = flex_pool.aggregators

clients.map(collect_weights_pt, aggregator)

Example of using the FlexPool without separating clients and aggregator, and following a client-server architecture.

from flex.pool import collect_weights_pt

flex_pool.clients.map(collect_weights_pt, flex_pool.aggregators)

flex.pool.primitives_pt.deploy_server_model_pt(server_flex_model, *args, **kwargs)[source]: Creates a copy of the server_flex_model and it is set to client nodes using the decorator @deploy_server_model.

Args:

server_flex_model (FlexModel): object storing information needed to run a Pytorch model

flex.pool.primitives_pt.set_aggregated_diff_weights_pt(server_flex_model, aggregated_diff_weights, *args, **kwargs)[source]: Function to add the aggregated weights to the server.

Args:

server_flex_model (FlexModel): The server’s FlexModel aggregated_diff_weights (np.array): Aggregated weights

Example of use assuming you are using a client-server architecture:

from flex.pool import set_aggregated_diff_weights_pt

aggregator = flex_pool.aggregators

aggregator.map(set_aggregated_diff_weights_pt)

Example of using the FlexPool without separating clients and aggregator, and following a client-server architecture.

from flex.pool import set_aggregated_diff_weights_pt

flex_pool.aggregators.map(set_aggregated_diff_weights_pt)

flex.pool.primitives_pt.set_aggregated_weights_pt(server_flex_model, aggregated_weights, *args, **kwargs)[source]: Function that replaces the weights of the server with the aggregated weights of the aggregator.

Args:

server_flex_model (FlexModel): The server’s FlexModel aggregated_weights (np.array): Aggregated weights

Example of use assuming you are using a client-server architecture:

from flex.pool import set_aggregated_weights_pt

aggregator = flex_pool.aggregators

aggregator.map(set_aggregated_weights_pt)

Example of using the FlexPool without separating clients and aggregator, and following a client-server architecture.

from flex.pool import set_aggregated_weights_pt

flex_pool.aggregators.map(set_aggregated_weights_pt)

flex.pool.primitives_tf module

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.

flex.pool.primitives_tf.collect_clients_weights_tf(client_flex_model, *args, **kwargs)[source]

Function that collect the weights for a TensorFlow model.

This function returns all the weights of the model.

Args:

client_flex_model (FlexModel): A client’s FlexModel

Returns:

np.array: An array with all the weights of the client’s model

Example of use assuming you are using a client-server architecture:

from flex.pool import collect_weights_tf

clients = flex_pool.clients aggregator = flex_pool.aggregators

clients.map(collect_weights_tf, aggregator)

Example of using the FlexPool without separating clients and aggregator, and following a client-server architecture.

from flex.pool import collect_weights_tf

flex_pool.clients.map(collect_weights_tf, flex_pool.aggregators)

flex.pool.primitives_tf.deploy_server_model_tf(server_flex_model, *args, **kwargs)[source]

Function to deploy a TensorFlow model from the server to a client.

The function will make a deepcopy for a TensorFlow model, as it needs a special method of copying. Also, it compiles the model for being able to train the model.

This function uses the decorator @deploy_server_model to deploy the server_flex_model to the all the clients, so we only need to create the steps for 1 client.

Args:

server_flex_model (FlexModel): Server FlexModel

Returns:

FlexModel: The client’s FlexModel

flex.pool.primitives_tf.evaluate_model_tf(flex_model, test_data)[source]: Function that evaluate the global model on the test data.

Args:

flex_model (FlexModel): server’s FlexModel test_data (Dataset): Test inputs.

Returns:

Evaluations by the model on the test data.

flex.pool.primitives_tf.init_server_model_tf(model=None, optimizer=None, loss=None, metrics=None, *args, **kwargs)[source]

Function that initialize a model in the server side for the TensorFlow framework.

This function acts as a message handler, that will initialize the model at the server side in a client-server architecture.

Args:

model (tf.keras.Model): A tf.keras.model initialized. optimizer (tf.keras.optimizers, optional): Optimizer for the model. Defaults to None. loss (tf.keras.losses, optional): _description_. Defaults to None. metrics (tf.keras.metrics, optional): _description_. Defaults to None.

Raises:

ValueError: If the model is not compiled and any of the optimizer, loss or metrics is not provided, then it will raise an error because we can’t initialize the model.

Returns:

FlexModel: A FlexModel that will be assigned to the server.

flex.pool.primitives_tf.set_aggregated_weights_tf(server_flex_model, aggregated_weights, *args, **kwargs)[source]: Function that replaces the weights of the server with the aggregated weights of the aggregator.

Args:

server_flex_model (FlexModel): The server’s FlexModel aggregated_weights (np.array): An array with the aggregated weights of the models.

flex.pool.primitives_tf.train_tf(client_flex_model, client_data, *args, **kwargs)[source]: Function of general purpose to train a TensorFlow model using FLEXible.

Args:

client_flex_model (FlexModel): client’s FlexModel client_data (FedDataset): client’s FedDataset

Example of use assuming you are using a client-server architecture:

from flex.pool import train_tf

clients = flex_pool.clients

clients.map(train_tf)

Example of using the FlexPool without separating clients and following a client-server architecture.

from flex.pool import train_tf

flex_pool.clients.map(train_tf)

Module contents

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.