Feature/SK-707 | Flower client example

.github/workflows/code-checks.yaml

@@ -43,6 +43,7 @@ jobs:
           --exclude-dir='.mnist-pytorch' 
           --exclude-dir='.mnist-keras' 
           --exclude-dir='docs'
+          --exclude-dir='flower-client'
           --exclude='tests.py'
           '^[ \t]+(import|from) ' -I . 
 

.vscode/settings.json

@@ -1,7 +1,7 @@
 {
     "editor.formatOnSave": true,
     "editor.codeActionsOnSave": {
-        "source.organizeImports": true
+        "source.organizeImports": "explicit"
     },
     "python.linting.enabled": true,
     "python.linting.flake8Enabled": true,

examples/flower-client/.dockerignore

@@ -0,0 +1,4 @@
+data
+seed.npz
+*.tgz
+*.tar.gz

examples/flower-client/.gitignore

@@ -0,0 +1,6 @@
+data
+*.npz
+*.tgz
+*.tar.gz
+.flower-example
+client.yaml

examples/flower-client/Dockerfile

@@ -0,0 +1,10 @@
+FROM ghcr.io/scaleoutsystems/fedn/fedn:0.8.0
+
+COPY requirements.txt /app/config/requirements.txt
+
+# Install requirements
+RUN python -m venv /venv \
+    && /venv/bin/pip install --upgrade pip \
+    && /venv/bin/pip install --no-cache-dir -r /app/config/requirements.txt \
+    # Clean up
+    && rm -r /app/config/requirements.txt

examples/flower-client/README.rst

@@ -0,0 +1,74 @@
+Using Flower clients in FEDn
+-------------
+
+This example shows how to run a Flower 'ClientApp' using the FEDn server-side infrastructure.
+
+See `flwr_client.py` and `flwr_task.py` for the Flower client code. The FEDn compute package is complemented
+with an adapter for the Flower client, `client_app_adapter.py`.
+
+
+Running the example
+-----------
+
+See `https://fedn.readthedocs.io/en/stable/quickstart.html` for a general introduction to FEDn. This example follows the same structure
+as the pytorch quickstart example. To build the compute package and seed model: 
+
+.. code-block::
+
+   bin/init_venv.sh
+
+.. code-block::
+
+   bin/build.sh
+
+Build a docker image containing the project dependencies including flower (this might take a long time):
+
+.. code-block::
+
+   docker build -t flower-client .
+
+In a separate terminal, navigate to this folder, then start a client and inject the `CLIENT_NUMBER` 
+dependency, for example for client1:
+
+If you are running FEDn locally using the provided docker-compose template:
+=====
+
+Use the FEDn API Client to initalize FEDn with the compute package and seed model: 
+
+.. code-block::
+
+   python init_fedn.py
+
+Create a file 'client.yaml' with the following content: 
+
+.. code-block::
+
+   network_id: fedn-network
+   discover_host: api-server
+   discover_port: 8092
+
+The start the client
+
+.. code-block::
+
+   docker run \
+   -v $PWD/client.yaml:/app/client.yaml \
+   --network=fedn_default \
+   -e CLIENT_NUMBER=0 \
+   flower-client run client -in client.yaml --name client1
+
+
+If you are using a FEDn Studio project:
+=====
+
+- Register a client in Studio and obtain the corresponding 'client.yaml' 
+
+Then start the client: 
+
+.. code-block::
+
+   docker run \
+   -v $PWD/client.yaml:/app/client.yaml \
+   -e CLIENT_NUMBER=0 \
+   flower-client run client -in client.yaml --secure=True --force-ssl
+

examples/flower-client/bin/build.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+set -e
+
+# Init seed
+client/entrypoint init_seed
+
+# Make compute package
+tar -czvf package.tgz client

examples/flower-client/bin/init_venv.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+# Init venv
+python3.9 -m venv .flower-example
+
+# Pip deps
+.flower-example/bin/pip install --upgrade pip
+.flower-example/bin/pip install -e ../../fedn
+.flower-example/bin/pip install -r requirements.txt

examples/flower-client/client/client_app_adapter.py

@@ -0,0 +1,84 @@
+from typing import Tuple
+
+from flwr.client import ClientApp
+from flwr.common import (Context, EvaluateIns, FitIns, Message, MessageType,
+                         Metadata, NDArrays, ndarrays_to_parameters,
+                         parameters_to_ndarrays)
+from flwr.common.recordset_compat import (evaluateins_to_recordset,
+                                          fitins_to_recordset,
+                                          recordset_to_evaluateres,
+                                          recordset_to_fitres)
+from flwr_task import Net, get_weights
+
+
+class FlwrClientAppAdapter:
+    """Flwr ClientApp wrapper."""
+
+    def __init__(self, app: ClientApp) -> None:
+        self.app = app
+
+    def init_parameters(self):
+        return get_weights(net=Net())
+
+    def train(self, parameters: NDArrays, partition_id: int):
+        # Construct a train message for the ClientApp with given parameters
+        message, context = self._construct_message(
+            MessageType.TRAIN, parameters, partition_id
+        )
+        # Call client app with train message
+        client_return_message = self.app(message, context)
+        # Parse return message
+        params, num_examples = self._parse_train_message(client_return_message)
+        return params, num_examples
+
+    def evaluate(self, parameters: NDArrays, partition_id: int):
+        # Construct an evaluate message for the ClientApp with given parameters
+        message, context = self._construct_message(
+            MessageType.EVALUATE, parameters, partition_id
+        )
+        # Call client app with evaluate message
+        client_return_message = self.app(message, context)
+        # Parse return message
+        loss, accuracy = self._parse_evaluate_message(client_return_message)
+        return loss, accuracy
+
+    def _parse_train_message(self, message: Message) -> Tuple[NDArrays, int]:
+        fitres = recordset_to_fitres(message.content, keep_input=False)
+        params, num_examples = (
+            parameters_to_ndarrays(fitres.parameters),
+            fitres.num_examples,
+        )
+        return params, num_examples
+
+    def _parse_evaluate_message(self, message: Message) -> Tuple[float, float]:
+        evaluateres = recordset_to_evaluateres(message.content)
+        return evaluateres.loss, evaluateres.metrics.get("accuracy", -1)
+
+    def _construct_message(
+        self,
+        message_type: MessageType,
+        parameters: NDArrays,
+        partition_id: int,
+    ) -> Tuple[Message, Context]:
+        parameters = ndarrays_to_parameters(parameters)
+        if message_type == MessageType.TRAIN:
+            fit_ins: FitIns = FitIns(parameters=parameters, config={})
+            recordset = fitins_to_recordset(fitins=fit_ins, keep_input=False)
+        if message_type == MessageType.EVALUATE:
+            ev_ins: EvaluateIns = EvaluateIns(parameters=parameters, config={})
+            recordset = evaluateins_to_recordset(evaluateins=ev_ins, keep_input=False)
+
+        metadata = Metadata(
+            run_id=0,
+            message_id="",
+            src_node_id=0,
+            dst_node_id=0,
+            reply_to_message="",
+            group_id="",
+            ttl="",
+            message_type=message_type,
+            partition_id=partition_id,
+        )
+        context = Context(recordset)
+        message = Message(metadata=metadata, content=recordset)
+        return message, context

examples/flower-client/client/entrypoint

@@ -0,0 +1,109 @@
+#!./.flower-example/bin/python
+import os
+import sys
+
+import fire
+from client_app_adapter import FlwrClientAppAdapter
+from flwr_client import app
+
+from fedn.utils.helpers.helpers import get_helper, save_metadata, save_metrics
+
+HELPER_MODULE = "numpyhelper"
+helper = get_helper(HELPER_MODULE)
+
+flwr_adapter = FlwrClientAppAdapter(app)
+
+
+def _get_node_id():
+    """Get client number from environment variable. """
+
+    number = os.environ.get("CLIENT_NUMBER", "0")
+    return int(number)
+
+
+def save_parameters(out_path, parameters_np):
+    """Save model paramters to file.
+
+    :param model: The model to serialize.
+    :type model: torch.nn.Module
+    :param out_path: The path to save to.
+    :type out_path: str
+    """
+    helper.save(parameters_np, out_path)
+
+
+def init_seed(out_path="seed.npz"):
+    """Initialize seed model and save it to file.
+
+    :param out_path: The path to save the seed model to.
+    :type out_path: str
+    """
+    # Init and save
+    parameters_np = flwr_adapter.init_parameters()
+    save_parameters(out_path, parameters_np)
+
+
+def train(in_model_path, out_model_path):
+    """Complete a model update.
+
+    Load model paramters from in_model_path (managed by the FEDn client),
+    perform a model update through the flower client, 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
+    """
+    parameters_np = helper.load(in_model_path)
+
+    # Train on flower client
+    params, num_examples = flwr_adapter.train(
+        parameters=parameters_np, partition_id=_get_node_id()
+    )
+
+    # Metadata needed for aggregation server side
+    metadata = {
+        # num_examples are mandatory
+        "num_examples": num_examples,
+    }
+
+    # Save JSON metadata file (mandatory)
+    save_metadata(metadata, out_model_path)
+
+    # Save model update (mandatory)
+    save_parameters(out_model_path, params)
+
+
+def validate(in_model_path, out_json_path, data_path=None):
+    """Validate model on the clients test dataset.
+
+    :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
+    """
+    parameters_np = helper.load(in_model_path)
+
+    loss, accuracy = flwr_adapter.evaluate(parameters_np, partition_id=_get_node_id())
+
+    # JSON schema
+    report = {
+        "test_loss": loss,
+        "test_accuracy": accuracy,
+    }
+    print(f"Loss: {loss}, accuracy: {accuracy}")
+    # Save JSON
+    save_metrics(report, out_json_path)
+
+
+if __name__ == "__main__":
+    fire.Fire(
+        {
+            "init_seed": init_seed,
+            "train": train,
+            "validate": validate,
+        }
+    )

examples/flower-client/client/fedn.yaml

@@ -0,0 +1,5 @@
+entry_points:
+  train:
+    command: /venv/bin/python entrypoint train $ENTRYPOINT_OPTS
+  validate:
+    command: /venv/bin/python entrypoint validate $ENTRYPOINT_OPTS

examples/flower-client/client/flwr_client.py

@@ -0,0 +1,37 @@
+"""Flower client code using the ClientApp abstraction. Code from https://github.com/adap/flower/tree/main/examples/app-pytorch."""
+
+from flwr.client import ClientApp, NumPyClient
+from flwr_task import (DEVICE, Net, get_weights, load_data, set_weights, test,
+                       train)
+
+
+# Define FlowerClient and client_fn
+class FlowerClient(NumPyClient):
+    def __init__(self, cid) -> None:
+        super().__init__()
+        print(f"STARTED CLIENT WITH CID {cid}")
+        self.net = Net().to(DEVICE)
+        self.trainloader, self.testloader = load_data(
+            partition_id=int(cid), num_clients=10
+        )
+
+    def fit(self, parameters, config):
+        set_weights(self.net, parameters)
+        train(self.net, self.trainloader, epochs=3)
+        return get_weights(self.net), len(self.trainloader.dataset), {}
+
+    def evaluate(self, parameters, config):
+        set_weights(self.net, parameters)
+        loss, accuracy = test(self.net, self.testloader)
+        return loss, len(self.testloader.dataset), {"accuracy": accuracy}
+
+
+def client_fn(cid: str):
+    """Create and return an instance of Flower `Client`."""
+    return FlowerClient(cid).to_client()
+
+
+# Flower ClientApp
+app = ClientApp(
+    client_fn=client_fn,
+)