Feature/SK-1688 | Rework Client

examples/importer-client/.gitignore

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

examples/importer-client/README.rst

@@ -0,0 +1,47 @@
+FEDn Project: Importer Client
+-----------------------------
+
+This is an example FEDn Project on how to design a client that imports client training code rather than running it in a separate process. 
+This enables the user to have access to the grpc channel to send information to thecontroller during training. 
+
+   **Note: We recommend that all new users start by taking the Quickstart Tutorial: https://fedn.readthedocs.io/en/stable/quickstart.html** 
+
+Prerequisites
+-------------
+
+-  `Python >=3.9, <=3.12 <https://www.python.org/downloads>`__
+-  `A project in FEDn Studio  <https://fedn.scaleoutsystems.com/signup>`__   
+
+Creating the compute package and seed model
+-------------------------------------------
+
+Install fedn: 
+
+.. code-block::
+
+   pip install fedn
+
+Clone this repository, then locate into this directory:
+
+.. code-block::
+
+   git clone https://github.com/scaleoutsystems/fedn.git
+   cd fedn/examples/importer-client
+
+Create the compute package:
+
+.. code-block::
+
+   fedn package create --path client
+
+This creates a file 'package.tgz' in the project folder.
+
+Running the project on FEDn
+----------------------------
+
+.. code-block::
+
+   fedn client start --importer --init client.yaml
+
+
+To learn how to set up your FEDn Studio project and connect clients, take the quickstart tutorial: https://fedn.readthedocs.io/en/stable/quickstart.html. 

examples/importer-client/client/build.py

@@ -0,0 +1,7 @@
+def main():
+    print("Hello World!")
+    # Do the build stuff usually creating the seed.npz
+
+
+if __name__ == "__main__":
+    main()

examples/importer-client/client/fedn.yaml

@@ -0,0 +1,7 @@
+# Remove the python_env tag below to handle the environment manually
+python_env: python_env.yaml 
+
+entry_points:
+  build: 
+    command: python build.py
+  startup: startup.py

examples/importer-client/client/python_env.yaml

@@ -0,0 +1,14 @@
+name: .mnist-pytorch
+build_dependencies:
+  - pip
+  - setuptools
+  - wheel
+dependencies:
+  - fedn
+  - torch==2.4.1; (sys_platform == "darwin" and platform_machine == "arm64") or (sys_platform == "win32" or sys_platform == "win64" or sys_platform == "linux")
+  # PyTorch macOS x86 builds deprecation
+  - torch==2.2.2; sys_platform == "darwin" and platform_machine == "x86_64"
+  - torchvision==0.19.1; (sys_platform == "darwin" and platform_machine == "arm64") or (sys_platform == "win32" or sys_platform == "win64" or sys_platform == "linux")
+  - torchvision==0.17.2; sys_platform == "darwin" and platform_machine == "x86_64"
+  - numpy==2.0.2; (sys_platform == "darwin" and platform_machine == "arm64") or (sys_platform == "win32" or sys_platform == "win64" or sys_platform == "linux")
+  - numpy==1.26.4; (sys_platform == "darwin" and platform_machine == "x86_64")

examples/importer-client/client/startup.py

@@ -0,0 +1,38 @@
+from fedn.network.clients.fedn_client import FednClient
+
+
+def startup(client: FednClient):
+    MyClient(client)
+
+
+class MyClient:
+    def __init__(self, client: FednClient):
+        self.client = client
+        client.set_train_callback(self.train)
+        client.set_validate_callback(self.validate)
+        client.set_predict_callback(self.predict)
+
+    def train(self, model_params, settings):
+        """Train the model with the given parameters and settings."""
+        # Implement training logic here
+        print("Training with model parameters:", model_params)
+        iterations = 100
+        for i in iterations:
+            # Do training
+            if i % 10 == 0:
+                self.client.log_metric({"training_loss": 0.1, "training_accuracy": 0.9})
+            # Regularly check if the task has been aborted
+            self.client.check_task_abort()  # Throws an exception if the task has been aborted
+        return model_params, {"training_metadata": {"num_examples": 1}}
+
+    def validate(self, model_params):
+        """Validate the model with the given parameters."""
+        # Implement validation logic here
+        print("Validating with model parameters:", model_params)
+        return {"validation_accuracy": 0.95}
+
+    def predict(self, model_params, data):
+        """Make predictions with the model using the given parameters and data."""
+        # Implement prediction logic here
+        print("Predicting with model parameters:", model_params, "and data:", data)
+        return {"predictions": [1, 0, 1]}  # Example predictions

examples/mnist-pytorch/.gitignore

@@ -3,4 +3,4 @@ data
 *.tgz
 *.tar.gz
 .mnist-pytorch
-client.yaml
+client*.yaml

examples/pytorch-keyworddetection-api/data.py

@@ -55,6 +55,11 @@ def __len__(self) -> int:
         return self._end_idx - self._start_idx
 
 
+def sc_collate_fn(batch: tuple[int, str, torch.Tensor, torch.Tensor]) -> tuple[torch.Tensor, torch.Tensor]:
+    ys, _, spectrogram, _ = zip(*batch)
+    return torch.tensor(ys, dtype=torch.long), torch.stack(spectrogram)
+
+
 class FedSCDataset(Dataset):
     """Dataset for the Federated Speech Commands dataset."""
 
@@ -241,11 +246,7 @@ def get_stats(self) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
         return torch.tensor(label_mean), spectrogram_mean[:, None], spectrogram_std[:, None]
 
     def get_collate_fn(self) -> callable:
-        def collate_fn(batch: tuple[int, str, torch.Tensor, torch.Tensor]) -> tuple[torch.Tensor, torch.Tensor]:
-            ys, _, spectrogram, _ = zip(*batch)
-            return torch.tensor(ys, dtype=torch.long), torch.stack(spectrogram)
-
-        return collate_fn
+        return sc_collate_fn
 
     def _get_spectogram_transform(self, n_mels: int, hop_length: int, sr: int, data_augmentation: bool = False) -> torch.nn.Sequential:
         if data_augmentation:
@@ -266,7 +267,9 @@ def get_dataloaders(
 ) -> tuple[DataLoader, DataLoader, DataLoader]:
     """Get the dataloaders for the training, validation, and testing datasets."""
     dataset_train = FedSCDataset(path, keywords, "training", dataset_split_idx, dataset_total_splits, data_augmentation=True)
-    dataloader_train = DataLoader(dataset=dataset_train, batch_size=batchsize_train, collate_fn=dataset_train.get_collate_fn(), shuffle=True, drop_last=True)
+    dataloader_train = DataLoader(
+        dataset=dataset_train, batch_size=batchsize_train, num_workers=2, collate_fn=dataset_train.get_collate_fn(), shuffle=True, drop_last=True
+    )
 
     dataset_valid = FedSCDataset(path, keywords, "validation", dataset_split_idx, dataset_total_splits)
     dataloader_valid = DataLoader(dataset=dataset_valid, batch_size=batchsize_valid, collate_fn=dataset_valid.get_collate_fn(), shuffle=False, drop_last=False)

examples/pytorch-keyworddetection-api/fedn_api.py

@@ -40,7 +40,8 @@ def main():
 
     if args.upload_seed:
         init_seedmodel()
-        api_client.set_active_model("seed.npz")
+        response = api_client.set_active_model("seed.npz")
+        print(response)
     elif args.start_session:
         # Depending on the computer hosting the clients this round_timeout might need to increase
         response = api_client.start_session(name="Training", round_timeout=1200)

fedn/__main__.py

@@ -0,0 +1,4 @@
+from fedn.cli import main
+
+if __name__ == "__main__":
+    main()

fedn/cli/client_cmd.py

@@ -10,8 +10,8 @@
 from fedn.cli.shared import CONTROLLER_DEFAULTS, STUDIO_DEFAULTS, apply_config, get_context, get_response, print_response
 from fedn.common.exceptions import InvalidClientConfig
 from fedn.common.log_config import set_log_level_from_string
-from fedn.network.clients.client_v2 import Client as ClientV2
-from fedn.network.clients.client_v2 import ClientOptions
+from fedn.network.clients.dispatcher_client import ClientOptions, DispatcherClient
+from fedn.network.clients.importer_client import ImporterClient
 
 home_dir = os.path.expanduser("~")
 
@@ -200,6 +200,8 @@ def _complement_client_params(config: dict) -> None:
 @click.option("-tr", "--trainer", required=False, default=None)
 @click.option("-hp", "--helper_type", required=False, default=None)
 @click.option("-in", "--init", required=False, default=None, help="Set to a filename to (re)init client from file state.")
+@click.option("--importer", is_flag=True, help="Use the importer client instead of the dispatcher client.")
+@click.option("--manual_env", is_flag=True, help="Use the manual environment for the client. This will not use the managed environment.")
 @click.pass_context
 def client_start_v2_cmd(
     ctx,
@@ -217,6 +219,8 @@ def client_start_v2_cmd(
     trainer: bool,
     helper_type: str,
     init: str,
+    importer: bool,
+    manual_env: bool = False,
 ):
     """Start client."""
     package = "local" if local_package else "remote"
@@ -329,15 +333,30 @@ def client_start_v2_cmd(
         preferred_combiner=config["preferred_combiner"],
         id=config["client_id"],
     )
-    client = ClientV2(
-        api_url=config["api_url"],
-        api_port=config["api_port"],
-        client_obj=client_options,
-        combiner_host=config["combiner"],
-        combiner_port=config["combiner_port"],
-        token=config["token"],
-        package_checksum=config["package_checksum"],
-        helper_type=config["helper_type"],
-    )
+    if importer:
+        click.echo("Using ImporterClient")
+        client = ImporterClient(
+            api_url=config["api_url"],
+            api_port=config["api_port"],
+            client_obj=client_options,
+            combiner_host=config["combiner"],
+            combiner_port=config["combiner_port"],
+            token=config["token"],
+            package_checksum=config["package_checksum"],
+            helper_type=config["helper_type"],
+            manual_env=manual_env,
+        )
+    else:
+        click.echo("Using DispatcherClient")
+        client = DispatcherClient(
+            api_url=config["api_url"],
+            api_port=config["api_port"],
+            client_obj=client_options,
+            combiner_host=config["combiner"],
+            combiner_port=config["combiner_port"],
+            token=config["token"],
+            package_checksum=config["package_checksum"],
+            helper_type=config["helper_type"],
+        )
 
     client.start()

fedn/cli/run_cmd.py

@@ -11,7 +11,8 @@
 from fedn.common.log_config import logger
 from fedn.network.storage.dbconnection import DatabaseConnection
 from fedn.network.storage.s3.repository import Repository
-from fedn.utils.dispatcher import Dispatcher, _read_yaml_file
+from fedn.utils.dispatcher import Dispatcher
+from fedn.utils.yaml import read_yaml_file
 
 
 def get_statestore_config_from_file(init):
@@ -46,8 +47,8 @@ def check_yaml_exists(path):
     return yaml_file
 
 
-def delete_virtual_environment(dispatcher):
-    if dispatcher.python_env_path:
+def delete_virtual_environment(dispatcher: Dispatcher):
+    if dispatcher.python_env_path and os.path.exists(dispatcher.python_env_path):
         logger.info(f"Removing virtualenv {dispatcher.python_env_path}")
         shutil.rmtree(dispatcher.python_env_path)
     else:
@@ -77,14 +78,14 @@ def validate_cmd(ctx, path, input, output, keep_venv):
     path = os.path.abspath(path)
     yaml_file = check_yaml_exists(path)
 
-    config = _read_yaml_file(yaml_file)
+    config = read_yaml_file(yaml_file)
     # Check that validate is defined in fedn.yaml under entry_points
     if "validate" not in config["entry_points"]:
         logger.error("No validate command defined in fedn.yaml")
         exit(-1)
 
     dispatcher = Dispatcher(config, path)
-    _ = dispatcher._get_or_create_python_env()
+    _ = dispatcher.get_or_create_python_env()
     dispatcher.run_cmd("validate {} {}".format(input, output))
     if not keep_venv:
         delete_virtual_environment(dispatcher)
@@ -106,14 +107,14 @@ def train_cmd(ctx, path, input, output, keep_venv):
     path = os.path.abspath(path)
     yaml_file = check_yaml_exists(path)
 
-    config = _read_yaml_file(yaml_file)
+    config = read_yaml_file(yaml_file)
     # Check that train is defined in fedn.yaml under entry_points
     if "train" not in config["entry_points"]:
         logger.error("No train command defined in fedn.yaml")
         exit(-1)
 
     dispatcher = Dispatcher(config, path)
-    _ = dispatcher._get_or_create_python_env()
+    _ = dispatcher.get_or_create_python_env()
     dispatcher.run_cmd("train {} {}".format(input, output))
     if not keep_venv:
         delete_virtual_environment(dispatcher)
@@ -133,13 +134,13 @@ def startup_cmd(ctx, path, keep_venv):
     path = os.path.abspath(path)
     yaml_file = check_yaml_exists(path)
 
-    config = _read_yaml_file(yaml_file)
+    config = read_yaml_file(yaml_file)
     # Check that startup is defined in fedn.yaml under entry_points
     if "startup" not in config["entry_points"]:
         logger.error("No startup command defined in fedn.yaml")
         exit(-1)
     dispatcher = Dispatcher(config, path)
-    _ = dispatcher._get_or_create_python_env()
+    _ = dispatcher.get_or_create_python_env()
     dispatcher.run_cmd("startup")
     if not keep_venv:
         delete_virtual_environment(dispatcher)
@@ -159,14 +160,14 @@ def build_cmd(ctx, path, keep_venv):
     path = os.path.abspath(path)
     yaml_file = check_yaml_exists(path)
 
-    config = _read_yaml_file(yaml_file)
+    config = read_yaml_file(yaml_file)
     # Check that build is defined in fedn.yaml under entry_points
     if "build" not in config["entry_points"]:
         logger.error("No build command defined in fedn.yaml")
         exit(-1)
 
     dispatcher = Dispatcher(config, path)
-    _ = dispatcher._get_or_create_python_env()
+    _ = dispatcher.get_or_create_python_env()
     dispatcher.run_cmd("build")
     if not keep_venv:
         delete_virtual_environment(dispatcher)

fedn/common/config.py

@@ -18,6 +18,7 @@
 
 FEDN_PACKAGE_EXTRACT_DIR = os.environ.get("FEDN_PACKAGE_EXTRACT_DIR", "package")
 FEDN_COMPUTE_PACKAGE_DIR = os.environ.get("FEDN_COMPUTE_PACKAGE_DIR", "/app/client/package/")
+FEDN_ARCHIVE_DIR = os.environ.get("FEDN_ARCHIVE_DIR", ".fedn")
 
 FEDN_OBJECT_STORAGE_TYPE = os.environ.get("FEDN_OBJECT_STORAGE_TYPE", "BOTO3").upper()
 FEDN_OBJECT_MODEL_BUCKET = os.environ.get("FEDN_OBJECT_MODEL_BUCKET", "fedn-model")

fedn/network/clients/connect.py

@@ -17,12 +17,7 @@
     FEDN_CUSTOM_URL_PREFIX,
 )
 from fedn.common.log_config import logger
-
-# Constants for HTTP status codes
-HTTP_STATUS_OK = 200
-HTTP_STATUS_NO_CONTENT = 204
-HTTP_STATUS_BAD_REQUEST = 400
-HTTP_STATUS_UNAUTHORIZED = 401
+from fedn.network.clients.http_status_codes import HTTP_STATUS_BAD_REQUEST, HTTP_STATUS_NO_CONTENT, HTTP_STATUS_OK, HTTP_STATUS_UNAUTHORIZED
 
 # Default timeout for requests
 REQUEST_TIMEOUT = 10  # seconds