Make choice of aggregator dynamic at session level and additional agg…

…regators (FedOpt) (#498)

* Removed unused base class

* work in progress, fedadam

* Fix race condition in docker-compose template

* Working fedopt, sgd as server side optimizer

* Working fedopt, sgd as server side optimizer

* Simple notebook demonstrating use of API to run an experiment and to plot validation results

* Make it possible to configure the aggregator per session

* code checks

* Set initial model_id in session config

* fedadam working for pytorch

* Rename numpyarrayhelper to numpyhelper

* Updated helper interface with numerics primitives

* PyTorch models now use list of numpy ndarray as format

* kerashelper and pytorchhelper consolidated into one numpyhelper

* Cleaned a bit in examples and added documentation

* removed non working healthcheck

* codechecks

* add back inference entrypoint

* Update integration tests

* codechecks

* Fix imports

* Removed unused arguments to combine_models

* Refactor helper module and update unit tests

* Refactor helper module

* codecheck

* Improve aggrgator interface

* codecheck

* Removed addition to fedn.yaml

* Updated docstrings

* Changed RoundControl to RoundHandler to avoid confusion with the global Controller

* Clean up notebook

* Moved notebook into pytorch example folder

* Added notebooks in torch example folder to gitignore

* Updated docstrings

---------

Co-authored-by: Andreas Hellander <[email protected]>

.github/workflows/integration-tests.yaml

@@ -15,8 +15,8 @@ jobs:
     strategy:
       matrix:
         to_test:
-          - "mnist-keras kerashelper"
-          - "mnist-pytorch pytorchhelper"
+          - "mnist-keras numpyhelper"
+          - "mnist-pytorch numpyhelper"
         python_version: ["3.8", "3.9","3.10"]
         os:
           - ubuntu-20.04

docker-compose.yaml

@@ -119,6 +119,8 @@ services:
       - "/venv/bin/pip install --no-cache-dir -e /app/fedn && /venv/bin/fedn run combiner --init config/settings-combiner.yaml"
     ports:
       - 12080:12080
+    depends_on:
+      - api-server
 
   # Client
   client:
@@ -136,3 +138,6 @@ services:
       - "/venv/bin/pip install --no-cache-dir -e /app/fedn && /venv/bin/fedn run client --init config/settings-client.yaml"
     deploy:
       replicas: 0
+    depends_on:
+      - api-server
+      - combiner

docs/fedn.utils.rst

@@ -33,10 +33,10 @@ fedn.utils.dispatcher module
    :undoc-members:
    :show-inheritance:
 
-fedn.utils.helpers module
+fedn.utils.helpers.helpers module
 -------------------------
 
-.. automodule:: fedn.utils.helpers
+.. automodule:: fedn.utils.helpers.helpers
    :members:
    :undoc-members:
    :show-inheritance:

docs/tutorial.rst

@@ -73,7 +73,7 @@ A *entrypoint.py* example can look like this:
     import fire
     import torch
 
-    from fedn.utils.helpers import get_helper, save_metadata, save_metrics
+    from fedn.utils.helpers.helpers import get_helper, save_metadata, save_metrics
 
     HELPER_MODULE = 'pytorchhelper'
     NUM_CLASSES = 10
@@ -298,7 +298,7 @@ For validations it is a requirement that the output is saved in a valid json for
 
    python entrypoint.py validate in_model_path out_json_path <extra-args>
  
-In the code example we use the helper function :py:meth:`fedn.utils.helpers.save_metrics` to save the validation scores as a json file. 
+In the code example we use the helper function :py:meth:`fedn.utils.helpers.helpers.save_metrics` to save the validation scores as a json file. 
 
 The Dahboard in the FEDn UI will plot any scalar metric in this json file, but you can include any type in the file assuming that it is valid json. These values can then be obtained (by an athorized user) from the MongoDB database or using the :py:mod:`fedn.network.api.client`. 
 

examples/mnist-keras/client/entrypoint

@@ -7,9 +7,11 @@ import fire
 import numpy as np
 import tensorflow as tf
 
-from fedn.utils.helpers import get_helper, save_metadata, save_metrics
+from fedn.utils.helpers.helpers import get_helper, save_metadata, save_metrics
+
+HELPER_MODULE = 'numpyhelper'
+helper = get_helper(HELPER_MODULE)
 
-HELPER_MODULE = 'kerashelper'
 NUM_CLASSES = 10
 
 
@@ -22,7 +24,16 @@ def _get_data_path():
     return f"/var/data/clients/{number}/mnist.npz"
 
 
-def _compile_model(img_rows=28, img_cols=28):
+def compile_model(img_rows=28, img_cols=28):
+    """ Compile the TF model.
+
+    param: img_rows: The number of rows in the image
+    type: img_rows: int
+    param: img_cols: The number of rows in the image
+    type: img_cols: int
+    return: The compiled model
+    type: keras.model.Sequential
+    """
     # Set input shape
     input_shape = (img_rows, img_cols, 1)
 
@@ -36,10 +47,11 @@ def _compile_model(img_rows=28, img_cols=28):
     model.compile(loss=tf.keras.losses.categorical_crossentropy,
                   optimizer=tf.keras.optimizers.Adam(),
                   metrics=['accuracy'])
+
     return model
 
 
-def _load_data(data_path, is_train=True):
+def load_data(data_path, is_train=True):
     # Load data
     if data_path is None:
         data = np.load(_get_data_path())
@@ -63,46 +75,77 @@ def _load_data(data_path, is_train=True):
 
 
 def init_seed(out_path='seed.npz'):
-    weights = _compile_model().get_weights()
-    helper = get_helper(HELPER_MODULE)
+    """ Initialize seed model and save it to file.
+
+    :param out_path: The path to save the seed model to.
+    :type out_path: str
+    """
+    weights = compile_model().get_weights()
     helper.save(weights, out_path)
 
 
 def train(in_model_path, out_model_path, data_path=None, batch_size=32, epochs=1):
+    """ Complete a model update.
+
+    Load model paramters from in_model_path (managed by the FEDn client),
+    perform a model update, and write updated paramters
+    to out_model_path (picked up by the FEDn client).
+
+    :param in_model_path: The path to the input model.
+    :type in_model_path: str
+    :param out_model_path: The path to save the output model to.
+    :type out_model_path: str
+    :param data_path: The path to the data file.
+    :type data_path: str
+    :param batch_size: The batch size to use.
+    :type batch_size: int
+    :param epochs: The number of epochs to train.
+    :type epochs: int
+    """
     # Load data
-    x_train, y_train = _load_data(data_path)
+    x_train, y_train = load_data(data_path)
 
     # Load model
-    model = _compile_model()
-    helper = get_helper(HELPER_MODULE)
+    model = compile_model()
     weights = helper.load(in_model_path)
     model.set_weights(weights)
 
     # Train
     model.fit(x_train, y_train, batch_size=batch_size, epochs=epochs)
 
-    # Save
-    weights = model.get_weights()
-    helper.save(weights, out_model_path)
-
     # Metadata needed for aggregation server side
     metadata = {
+        # num_examples are mandatory
         'num_examples': len(x_train),
         'batch_size': batch_size,
         'epochs': epochs,
     }
 
-    # Save JSON metadata file
+    # Save JSON metadata file (mandatory)
     save_metadata(metadata, out_model_path)
 
+    # Save model update (mandatory)
+    weights = model.get_weights()
+    helper.save(weights, out_model_path)
+
 
 def validate(in_model_path, out_json_path, data_path=None):
+    """ Validate model.
+
+    :param in_model_path: The path to the input model.
+    :type in_model_path: str
+    :param out_json_path: The path to save the output JSON to.
+    :type out_json_path: str
+    :param data_path: The path to the data file.
+    :type data_path: str
+    """
+
     # Load data
-    x_train, y_train = _load_data(data_path)
-    x_test, y_test = _load_data(data_path, is_train=False)
+    x_train, y_train = load_data(data_path)
+    x_test, y_test = load_data(data_path, is_train=False)
 
     # Load model
-    model = _compile_model()
+    model = compile_model()
     helper = get_helper(HELPER_MODULE)
     weights = helper.load(in_model_path)
     model.set_weights(weights)
@@ -127,10 +170,10 @@ def validate(in_model_path, out_json_path, data_path=None):
 
 def infer(in_model_path, out_json_path, data_path=None):
     # Using test data for inference but another dataset could be loaded
-    x_test, _ = _load_data(data_path, is_train=False)
+    x_test, _ = load_data(data_path, is_train=False)
 
     # Load model
-    model = _compile_model()
+    model = compile_model()
     helper = get_helper(HELPER_MODULE)
     weights = helper.load(in_model_path)
     model.set_weights(weights)

examples/mnist-pytorch/.gitignore

@@ -2,5 +2,6 @@ data
 *.npz
 *.tgz
 *.tar.gz
+*.ipynb
 .mnist-pytorch
 client.yaml