diff --git a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Portland/Portland_config.yaml b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Portland/Portland_config.yaml old mode 100755 new mode 100644 index a37e5010bd..ef0cc77e01 --- a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Portland/Portland_config.yaml +++ b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Portland/Portland_config.yaml @@ -1,6 +1,7 @@ settings: director_host: localhost director_port: 50050 + review_experiment: True Portland: private_attributes: private_attributes.portland_attrs \ No newline at end of file diff --git a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Seattle/Seattle_config.yaml b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Seattle/Seattle_config.yaml old mode 100755 new mode 100644 index 07a819115c..d4c9cc20cf --- a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Seattle/Seattle_config.yaml +++ b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Seattle/Seattle_config.yaml @@ -1,6 +1,7 @@ settings: director_host: localhost director_port: 50050 + review_experiment: True Seattle: private_attributes: private_attributes.seattle_attrs \ No newline at end of file diff --git a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/director/director_config.yaml b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/director/director_config.yaml old mode 100755 new mode 100644 index 021cfc59c9..951dcd0db0 --- a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/director/director_config.yaml +++ b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/director/director_config.yaml @@ -1,4 +1,5 @@ settings: listen_host: localhost listen_port: 50050 - envoy_health_check_period: 5 # in seconds \ No newline at end of file + envoy_health_check_period: 5 # in seconds + review_experiment: True \ No newline at end of file diff --git a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/workspace/101_MNIST_FederatedRuntime.ipynb b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/workspace/101_MNIST_FederatedRuntime.ipynb index 00daa8095e..4dc406e20c 100644 --- a/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/workspace/101_MNIST_FederatedRuntime.ipynb +++ b/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/workspace/101_MNIST_FederatedRuntime.ipynb @@ -50,7 +50,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 1, "id": "d79eacbd", "metadata": {}, "outputs": [], @@ -77,10 +77,225 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 2, "id": "f7475cba", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Collecting git+https://github.com/securefederatedai/openfl.git\n", + " Cloning https://github.com/securefederatedai/openfl.git to /tmp/pip-req-build-smr5rg8s\n", + " Running command git clone --filter=blob:none --quiet https://github.com/securefederatedai/openfl.git /tmp/pip-req-build-smr5rg8s\n", + " Resolved https://github.com/securefederatedai/openfl.git to commit 6e03dfe70fca5d54913fed3290fef200dd2cfb64\n", + " Installing build dependencies ... \u001b[?25ldone\n", + "\u001b[?25h Getting requirements to build wheel ... \u001b[?25ldone\n", + "\u001b[?25h Installing backend dependencies ... \u001b[?25ldone\n", + "\u001b[?25h Preparing metadata (pyproject.toml) ... \u001b[?25ldone\n", + "\u001b[?25hRequirement already satisfied: click in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (8.1.8)\n", + "Requirement already satisfied: psutil in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (7.0.0)\n", + "Requirement already satisfied: pyyaml in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (6.0.2)\n", + "Requirement already satisfied: rich in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (13.9.4)\n", + "Requirement already satisfied: dynaconf in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (3.2.10)\n", + "Requirement already satisfied: tqdm in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (4.67.1)\n", + "Requirement already satisfied: numpy in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (2.2.3)\n", + "Requirement already satisfied: requests>=2.32.0 in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (2.32.3)\n", + "Requirement already satisfied: cloudpickle in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (3.1.1)\n", + "Requirement already satisfied: cryptography in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (44.0.2)\n", + "Requirement already satisfied: pandas in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (2.2.3)\n", + "Requirement already satisfied: scikit-learn in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (1.6.1)\n", + "Requirement already satisfied: flatten_json in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (0.1.14)\n", + "Requirement already satisfied: tensorboardX in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (2.6.2.2)\n", + "Requirement already satisfied: protobuf<6.0.0,>=4.22 in /home/neeraj/.local/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (5.29.3)\n", + "Requirement already satisfied: grpcio<1.66.0,>=1.56.2 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from openfl==1.9.0.dev0) (1.65.5)\n", + "Requirement already satisfied: charset-normalizer<4,>=2 in /home/neeraj/.local/lib/python3.10/site-packages (from requests>=2.32.0->openfl==1.9.0.dev0) (3.4.1)\n", + "Requirement already satisfied: idna<4,>=2.5 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests>=2.32.0->openfl==1.9.0.dev0) (3.10)\n", + "Requirement already satisfied: urllib3<3,>=1.21.1 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests>=2.32.0->openfl==1.9.0.dev0) (2.4.0)\n", + "Requirement already satisfied: certifi>=2017.4.17 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests>=2.32.0->openfl==1.9.0.dev0) (2025.1.31)\n", + "Requirement already satisfied: cffi>=1.12 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from cryptography->openfl==1.9.0.dev0) (1.17.1)\n", + "Requirement already satisfied: six in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from flatten_json->openfl==1.9.0.dev0) (1.17.0)\n", + "Requirement already satisfied: python-dateutil>=2.8.2 in /home/neeraj/.local/lib/python3.10/site-packages (from pandas->openfl==1.9.0.dev0) (2.9.0.post0)\n", + "Requirement already satisfied: pytz>=2020.1 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from pandas->openfl==1.9.0.dev0) (2025.2)\n", + "Requirement already satisfied: tzdata>=2022.7 in /home/neeraj/.local/lib/python3.10/site-packages (from pandas->openfl==1.9.0.dev0) (2025.1)\n", + "Requirement already satisfied: markdown-it-py>=2.2.0 in /home/neeraj/.local/lib/python3.10/site-packages (from rich->openfl==1.9.0.dev0) (3.0.0)\n", + "Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /home/neeraj/.local/lib/python3.10/site-packages (from rich->openfl==1.9.0.dev0) (2.19.1)\n", + "Requirement already satisfied: typing-extensions<5.0,>=4.0.0 in /home/neeraj/.local/lib/python3.10/site-packages (from rich->openfl==1.9.0.dev0) (4.12.2)\n", + "Requirement already satisfied: scipy>=1.6.0 in /home/neeraj/.local/lib/python3.10/site-packages (from scikit-learn->openfl==1.9.0.dev0) (1.15.2)\n", + "Requirement already satisfied: joblib>=1.2.0 in /home/neeraj/.local/lib/python3.10/site-packages (from scikit-learn->openfl==1.9.0.dev0) (1.4.2)\n", + "Requirement already satisfied: threadpoolctl>=3.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from scikit-learn->openfl==1.9.0.dev0) (3.5.0)\n", + "Requirement already satisfied: packaging in /home/neeraj/.local/lib/python3.10/site-packages (from tensorboardX->openfl==1.9.0.dev0) (24.2)\n", + "Requirement already satisfied: pycparser in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from cffi>=1.12->cryptography->openfl==1.9.0.dev0) (2.22)\n", + "Requirement already satisfied: mdurl~=0.1 in /home/neeraj/.local/lib/python3.10/site-packages (from markdown-it-py>=2.2.0->rich->openfl==1.9.0.dev0) (0.1.2)\n", + "Requirement already satisfied: chardet in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 1)) (5.2.0)\n", + "Requirement already satisfied: charset-normalizer in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 2)) (3.4.1)\n", + "Requirement already satisfied: dill==0.3.6 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 3)) (0.3.6)\n", + "Requirement already satisfied: matplotlib>=2.0.0 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 4)) (3.10.1)\n", + "Requirement already satisfied: metaflow==2.7.15 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 5)) (2.7.15)\n", + "Requirement already satisfied: nbdev==2.3.37 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 6)) (2.3.37)\n", + "Requirement already satisfied: nbformat==5.10.4 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 7)) (5.10.4)\n", + "Requirement already satisfied: ray==2.43.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 8)) (2.43.0)\n", + "Requirement already satisfied: tabulate==0.9.0 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 9)) (0.9.0)\n", + "Requirement already satisfied: torch==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 10)) (2.3.1)\n", + "Requirement already satisfied: torchvision==0.18.1 in /home/neeraj/.local/lib/python3.10/site-packages (from -r ../../../workflow_interface_requirements.txt (line 11)) (0.18.1)\n", + "Requirement already satisfied: requests in /home/neeraj/.local/lib/python3.10/site-packages (from metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (2.32.3)\n", + "Requirement already satisfied: boto3 in /home/neeraj/.local/lib/python3.10/site-packages (from metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (1.37.15)\n", + "Requirement already satisfied: pylint in /home/neeraj/.local/lib/python3.10/site-packages (from metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (3.3.5)\n", + "Requirement already satisfied: packaging in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (24.2)\n", + "Requirement already satisfied: fastcore>=1.8.0 in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (1.8.0)\n", + "Requirement already satisfied: execnb>=0.1.12 in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.1.12)\n", + "Requirement already satisfied: astunparse in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (1.6.3)\n", + "Requirement already satisfied: ghapi>=1.0.3 in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (1.0.6)\n", + "Requirement already satisfied: watchdog in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (6.0.0)\n", + "Requirement already satisfied: asttokens in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (3.0.0)\n", + "Requirement already satisfied: setuptools in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (75.8.0)\n", + "Requirement already satisfied: plum-dispatch in /home/neeraj/.local/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (2.5.7)\n", + "Requirement already satisfied: PyYAML in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (6.0.2)\n", + "Requirement already satisfied: fastjsonschema>=2.15 in /home/neeraj/.local/lib/python3.10/site-packages (from nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (2.21.1)\n", + "Requirement already satisfied: jsonschema>=2.6 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (4.23.0)\n", + "Requirement already satisfied: jupyter-core!=5.0.*,>=4.12 in /home/neeraj/.local/lib/python3.10/site-packages (from nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (5.7.2)\n", + "Requirement already satisfied: traitlets>=5.1 in /home/neeraj/.local/lib/python3.10/site-packages (from nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (5.14.3)\n", + "Requirement already satisfied: click>=7.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (8.1.8)\n", + "Requirement already satisfied: filelock in /home/neeraj/.local/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (3.18.0)\n", + "Requirement already satisfied: msgpack<2.0.0,>=1.0.0 in /home/neeraj/.local/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (1.1.0)\n", + "Requirement already satisfied: protobuf!=3.19.5,>=3.15.3 in /home/neeraj/.local/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (5.29.3)\n", + "Requirement already satisfied: aiosignal in /home/neeraj/.local/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (1.3.2)\n", + "Requirement already satisfied: frozenlist in /home/neeraj/.local/lib/python3.10/site-packages (from ray==2.43.0->-r ../../../workflow_interface_requirements.txt (line 8)) (1.5.0)\n", + "Requirement already satisfied: typing-extensions>=4.8.0 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (4.12.2)\n", + "Requirement already satisfied: sympy in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (1.13.3)\n", + "Requirement already satisfied: networkx in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (3.4.2)\n", + "Requirement already satisfied: jinja2 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (3.1.6)\n", + "Requirement already satisfied: fsspec in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (2025.3.0)\n", + "Requirement already satisfied: nvidia-cuda-nvrtc-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-runtime-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-cupti-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.105)\n", + "Requirement already satisfied: nvidia-cudnn-cu12==8.9.2.26 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (8.9.2.26)\n", + "Requirement already satisfied: nvidia-cublas-cu12==12.1.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.3.1)\n", + "Requirement already satisfied: nvidia-cufft-cu12==11.0.2.54 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (11.0.2.54)\n", + "Requirement already satisfied: nvidia-curand-cu12==10.3.2.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (10.3.2.106)\n", + "Requirement already satisfied: nvidia-cusolver-cu12==11.4.5.107 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (11.4.5.107)\n", + "Requirement already satisfied: nvidia-cusparse-cu12==12.1.0.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.0.106)\n", + "Requirement already satisfied: nvidia-nccl-cu12==2.20.5 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (2.20.5)\n", + "Requirement already satisfied: nvidia-nvtx-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.1.105)\n", + "Requirement already satisfied: triton==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (2.3.1)\n", + "Requirement already satisfied: numpy in /home/neeraj/.local/lib/python3.10/site-packages (from torchvision==0.18.1->-r ../../../workflow_interface_requirements.txt (line 11)) (2.2.3)\n", + "Requirement already satisfied: pillow!=8.3.*,>=5.3.0 in /home/neeraj/.local/lib/python3.10/site-packages (from torchvision==0.18.1->-r ../../../workflow_interface_requirements.txt (line 11)) (11.1.0)\n", + "Requirement already satisfied: nvidia-nvjitlink-cu12 in /home/neeraj/.local/lib/python3.10/site-packages (from nvidia-cusolver-cu12==11.4.5.107->torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (12.8.93)\n", + "Requirement already satisfied: contourpy>=1.0.1 in /home/neeraj/.local/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (1.3.1)\n", + "Requirement already satisfied: cycler>=0.10 in /home/neeraj/.local/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (0.12.1)\n", + "Requirement already satisfied: fonttools>=4.22.0 in /home/neeraj/.local/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (4.56.0)\n", + "Requirement already satisfied: kiwisolver>=1.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (1.4.8)\n", + "Requirement already satisfied: pyparsing>=2.3.1 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (3.2.3)\n", + "Requirement already satisfied: python-dateutil>=2.7 in /home/neeraj/.local/lib/python3.10/site-packages (from matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (2.9.0.post0)\n", + "Requirement already satisfied: ipython in /home/neeraj/.local/lib/python3.10/site-packages (from execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (8.34.0)\n", + "Requirement already satisfied: attrs>=22.2.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jsonschema>=2.6->nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (25.3.0)\n", + "Requirement already satisfied: jsonschema-specifications>=2023.03.6 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jsonschema>=2.6->nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (2024.10.1)\n", + "Requirement already satisfied: referencing>=0.28.4 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jsonschema>=2.6->nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (0.36.2)\n", + "Requirement already satisfied: rpds-py>=0.7.1 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jsonschema>=2.6->nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (0.24.0)\n", + "Requirement already satisfied: platformdirs>=2.5 in /home/neeraj/.local/lib/python3.10/site-packages (from jupyter-core!=5.0.*,>=4.12->nbformat==5.10.4->-r ../../../workflow_interface_requirements.txt (line 7)) (4.3.6)\n", + "Requirement already satisfied: six>=1.5 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from python-dateutil>=2.7->matplotlib>=2.0.0->-r ../../../workflow_interface_requirements.txt (line 4)) (1.17.0)\n", + "Requirement already satisfied: wheel<1.0,>=0.23.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from astunparse->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.45.1)\n", + "Requirement already satisfied: botocore<1.38.0,>=1.37.15 in /home/neeraj/.local/lib/python3.10/site-packages (from boto3->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (1.37.15)\n", + "Requirement already satisfied: jmespath<2.0.0,>=0.7.1 in /home/neeraj/.local/lib/python3.10/site-packages (from boto3->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (1.0.1)\n", + "Requirement already satisfied: s3transfer<0.12.0,>=0.11.0 in /home/neeraj/.local/lib/python3.10/site-packages (from boto3->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (0.11.4)\n", + "Requirement already satisfied: MarkupSafe>=2.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jinja2->torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (3.0.2)\n", + "Requirement already satisfied: beartype>=0.16.2 in /home/neeraj/.local/lib/python3.10/site-packages (from plum-dispatch->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.20.0)\n", + "Requirement already satisfied: rich>=10.0 in /home/neeraj/.local/lib/python3.10/site-packages (from plum-dispatch->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (13.9.4)\n", + "Requirement already satisfied: astroid<=3.4.0-dev0,>=3.3.8 in /home/neeraj/.local/lib/python3.10/site-packages (from pylint->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (3.3.9)\n", + "Requirement already satisfied: isort!=5.13.0,<7,>=4.2.5 in /home/neeraj/.local/lib/python3.10/site-packages (from pylint->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (6.0.1)\n", + "Requirement already satisfied: mccabe<0.8,>=0.6 in /home/neeraj/.local/lib/python3.10/site-packages (from pylint->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (0.7.0)\n", + "Requirement already satisfied: tomli>=1.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from pylint->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (2.2.1)\n", + "Requirement already satisfied: tomlkit>=0.10.1 in /home/neeraj/.local/lib/python3.10/site-packages (from pylint->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (0.13.2)\n", + "Requirement already satisfied: idna<4,>=2.5 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (3.10)\n", + "Requirement already satisfied: urllib3<3,>=1.21.1 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (2.4.0)\n", + "Requirement already satisfied: certifi>=2017.4.17 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from requests->metaflow==2.7.15->-r ../../../workflow_interface_requirements.txt (line 5)) (2025.1.31)\n", + "Requirement already satisfied: mpmath<1.4,>=1.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from sympy->torch==2.3.1->-r ../../../workflow_interface_requirements.txt (line 10)) (1.3.0)\n", + "Requirement already satisfied: markdown-it-py>=2.2.0 in /home/neeraj/.local/lib/python3.10/site-packages (from rich>=10.0->plum-dispatch->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (3.0.0)\n", + "Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /home/neeraj/.local/lib/python3.10/site-packages (from rich>=10.0->plum-dispatch->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (2.19.1)\n", + "Requirement already satisfied: decorator in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (5.2.1)\n", + "Requirement already satisfied: exceptiongroup in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (1.2.2)\n", + "Requirement already satisfied: jedi>=0.16 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.19.2)\n", + "Requirement already satisfied: matplotlib-inline in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.1.7)\n", + "Requirement already satisfied: pexpect>4.3 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (4.9.0)\n", + "Requirement already satisfied: prompt_toolkit<3.1.0,>=3.0.41 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (3.0.50)\n", + "Requirement already satisfied: stack_data in /home/neeraj/.local/lib/python3.10/site-packages (from ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.6.3)\n", + "Requirement already satisfied: parso<0.9.0,>=0.8.4 in /home/neeraj/.local/lib/python3.10/site-packages (from jedi>=0.16->ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.8.4)\n", + "Requirement already satisfied: mdurl~=0.1 in /home/neeraj/.local/lib/python3.10/site-packages (from markdown-it-py>=2.2.0->rich>=10.0->plum-dispatch->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.1.2)\n", + "Requirement already satisfied: ptyprocess>=0.5 in /home/neeraj/.local/lib/python3.10/site-packages (from pexpect>4.3->ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.7.0)\n", + "Requirement already satisfied: wcwidth in /home/neeraj/.local/lib/python3.10/site-packages (from prompt_toolkit<3.1.0,>=3.0.41->ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.2.13)\n", + "Requirement already satisfied: executing>=1.2.0 in /home/neeraj/.local/lib/python3.10/site-packages (from stack_data->ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (2.2.0)\n", + "Requirement already satisfied: pure-eval in /home/neeraj/.local/lib/python3.10/site-packages (from stack_data->ipython->execnb>=0.1.12->nbdev==2.3.37->-r ../../../workflow_interface_requirements.txt (line 6)) (0.2.3)\n", + "Requirement already satisfied: torch==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (2.3.1)\n", + "Requirement already satisfied: filelock in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (3.18.0)\n", + "Requirement already satisfied: typing-extensions>=4.8.0 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (4.12.2)\n", + "Requirement already satisfied: sympy in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (1.13.3)\n", + "Requirement already satisfied: networkx in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (3.4.2)\n", + "Requirement already satisfied: jinja2 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from torch==2.3.1) (3.1.6)\n", + "Requirement already satisfied: fsspec in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (2025.3.0)\n", + "Requirement already satisfied: nvidia-cuda-nvrtc-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-runtime-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-cupti-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cudnn-cu12==8.9.2.26 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (8.9.2.26)\n", + "Requirement already satisfied: nvidia-cublas-cu12==12.1.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.3.1)\n", + "Requirement already satisfied: nvidia-cufft-cu12==11.0.2.54 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (11.0.2.54)\n", + "Requirement already satisfied: nvidia-curand-cu12==10.3.2.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (10.3.2.106)\n", + "Requirement already satisfied: nvidia-cusolver-cu12==11.4.5.107 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (11.4.5.107)\n", + "Requirement already satisfied: nvidia-cusparse-cu12==12.1.0.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.0.106)\n", + "Requirement already satisfied: nvidia-nccl-cu12==2.20.5 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (2.20.5)\n", + "Requirement already satisfied: nvidia-nvtx-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (12.1.105)\n", + "Requirement already satisfied: triton==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1) (2.3.1)\n", + "Requirement already satisfied: nvidia-nvjitlink-cu12 in /home/neeraj/.local/lib/python3.10/site-packages (from nvidia-cusolver-cu12==11.4.5.107->torch==2.3.1) (12.8.93)\n", + "Requirement already satisfied: MarkupSafe>=2.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jinja2->torch==2.3.1) (3.0.2)\n", + "Requirement already satisfied: mpmath<1.4,>=1.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from sympy->torch==2.3.1) (1.3.0)\n", + "Requirement already satisfied: torchvision==0.18.1 in /home/neeraj/.local/lib/python3.10/site-packages (0.18.1)\n", + "Requirement already satisfied: numpy in /home/neeraj/.local/lib/python3.10/site-packages (from torchvision==0.18.1) (2.2.3)\n", + "Requirement already satisfied: torch==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torchvision==0.18.1) (2.3.1)\n", + "Requirement already satisfied: pillow!=8.3.*,>=5.3.0 in /home/neeraj/.local/lib/python3.10/site-packages (from torchvision==0.18.1) (11.1.0)\n", + "Requirement already satisfied: filelock in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (3.18.0)\n", + "Requirement already satisfied: typing-extensions>=4.8.0 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (4.12.2)\n", + "Requirement already satisfied: sympy in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (1.13.3)\n", + "Requirement already satisfied: networkx in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (3.4.2)\n", + "Requirement already satisfied: jinja2 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (3.1.6)\n", + "Requirement already satisfied: fsspec in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (2025.3.0)\n", + "Requirement already satisfied: nvidia-cuda-nvrtc-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-runtime-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cuda-cupti-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.105)\n", + "Requirement already satisfied: nvidia-cudnn-cu12==8.9.2.26 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (8.9.2.26)\n", + "Requirement already satisfied: nvidia-cublas-cu12==12.1.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.3.1)\n", + "Requirement already satisfied: nvidia-cufft-cu12==11.0.2.54 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (11.0.2.54)\n", + "Requirement already satisfied: nvidia-curand-cu12==10.3.2.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (10.3.2.106)\n", + "Requirement already satisfied: nvidia-cusolver-cu12==11.4.5.107 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (11.4.5.107)\n", + "Requirement already satisfied: nvidia-cusparse-cu12==12.1.0.106 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.0.106)\n", + "Requirement already satisfied: nvidia-nccl-cu12==2.20.5 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (2.20.5)\n", + "Requirement already satisfied: nvidia-nvtx-cu12==12.1.105 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (12.1.105)\n", + "Requirement already satisfied: triton==2.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from torch==2.3.1->torchvision==0.18.1) (2.3.1)\n", + "Requirement already satisfied: nvidia-nvjitlink-cu12 in /home/neeraj/.local/lib/python3.10/site-packages (from nvidia-cusolver-cu12==11.4.5.107->torch==2.3.1->torchvision==0.18.1) (12.8.93)\n", + "Requirement already satisfied: MarkupSafe>=2.0 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from jinja2->torch==2.3.1->torchvision==0.18.1) (3.0.2)\n", + "Requirement already satisfied: mpmath<1.4,>=1.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from sympy->torch==2.3.1->torchvision==0.18.1) (1.3.0)\n", + "Requirement already satisfied: ipywidgets in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (8.1.6)\n", + "Requirement already satisfied: comm>=0.1.3 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from ipywidgets) (0.2.2)\n", + "Requirement already satisfied: ipython>=6.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from ipywidgets) (8.34.0)\n", + "Requirement already satisfied: traitlets>=4.3.1 in /home/neeraj/.local/lib/python3.10/site-packages (from ipywidgets) (5.14.3)\n", + "Requirement already satisfied: widgetsnbextension~=4.0.14 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from ipywidgets) (4.0.14)\n", + "Requirement already satisfied: jupyterlab_widgets~=3.0.14 in /home/neeraj/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages (from ipywidgets) (3.0.14)\n", + "Requirement already satisfied: decorator in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (5.2.1)\n", + "Requirement already satisfied: exceptiongroup in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (1.2.2)\n", + "Requirement already satisfied: jedi>=0.16 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (0.19.2)\n", + "Requirement already satisfied: matplotlib-inline in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (0.1.7)\n", + "Requirement already satisfied: pexpect>4.3 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (4.9.0)\n", + "Requirement already satisfied: prompt_toolkit<3.1.0,>=3.0.41 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (3.0.50)\n", + "Requirement already satisfied: pygments>=2.4.0 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (2.19.1)\n", + "Requirement already satisfied: stack_data in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (0.6.3)\n", + "Requirement already satisfied: typing_extensions>=4.6 in /home/neeraj/.local/lib/python3.10/site-packages (from ipython>=6.1.0->ipywidgets) (4.12.2)\n", + "Requirement already satisfied: parso<0.9.0,>=0.8.4 in /home/neeraj/.local/lib/python3.10/site-packages (from jedi>=0.16->ipython>=6.1.0->ipywidgets) (0.8.4)\n", + "Requirement already satisfied: ptyprocess>=0.5 in /home/neeraj/.local/lib/python3.10/site-packages (from pexpect>4.3->ipython>=6.1.0->ipywidgets) (0.7.0)\n", + "Requirement already satisfied: wcwidth in /home/neeraj/.local/lib/python3.10/site-packages (from prompt_toolkit<3.1.0,>=3.0.41->ipython>=6.1.0->ipywidgets) (0.2.13)\n", + "Requirement already satisfied: executing>=1.2.0 in /home/neeraj/.local/lib/python3.10/site-packages (from stack_data->ipython>=6.1.0->ipywidgets) (2.2.0)\n", + "Requirement already satisfied: asttokens>=2.1.0 in /home/neeraj/.local/lib/python3.10/site-packages (from stack_data->ipython>=6.1.0->ipywidgets) (3.0.0)\n", + "Requirement already satisfied: pure-eval in /home/neeraj/.local/lib/python3.10/site-packages (from stack_data->ipython>=6.1.0->ipywidgets) (0.2.3)\n" + ] + } + ], "source": [ "#| export\n", "\n", @@ -112,7 +327,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 3, "id": "9bd8ac2d", "metadata": {}, "outputs": [], @@ -211,10 +426,19 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 4, "id": "9836c542", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stderr", + "output_type": "stream", + "text": [ + "/home/neeraj/.local/lib/python3.10/site-packages/torch/cuda/__init__.py:619: UserWarning: Can't initialize NVML\n", + " warnings.warn(\"Can't initialize NVML\")\n" + ] + } + ], "source": [ "#| export\n", "\n", @@ -268,7 +492,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 5, "id": "89cf4866", "metadata": {}, "outputs": [], @@ -304,10 +528,23 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 6, "id": "52c4a752", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Aggregator step \"start\" registered\n", + "Collaborator step \"aggregated_model_validation\" registered\n", + "Collaborator step \"train\" registered\n", + "Collaborator step \"local_model_validation\" registered\n", + "Aggregator step \"join\" registered\n", + "Aggregator step \"end\" registered\n" + ] + } + ], "source": [ "# | export\n", "\n", @@ -451,10 +688,18 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 7, "id": "bffcc141", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Local runtime collaborators = ['Portland', 'Seattle']\n" + ] + } + ], "source": [ "# | export\n", "\n", @@ -511,7 +756,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 8, "id": "e5f10d5d", "metadata": {}, "outputs": [], @@ -521,8 +766,8 @@ "model = None\n", "optimizer = None\n", "flflow = FederatedFlow_TorchMNIST(model, optimizer, learning_rate, momentum, rounds=2, checkpoint=True)\n", - "flflow.runtime = local_runtime\n", - "flflow.run()" + "#flflow.runtime = local_runtime\n", + "#flflow.run()" ] }, { @@ -535,17 +780,25 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 9, "id": "a5d77540", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "********** Simulation results **********\n" + ] + } + ], "source": [ "from tabulate import tabulate \n", "\n", "headers = [\"Rounds\", \"Agg Model Validation Score\", \"Local Train loss\", \"Local Model Validation score\"]\n", "print('********** Simulation results **********')\n", - "simulation_results = flflow.results\n", - "print(tabulate(simulation_results, headers=headers, tablefmt=\"outline\"))\n" + "#simulation_results = flflow.results\n", + "#print(tabulate(simulation_results, headers=headers, tablefmt=\"outline\"))\n" ] }, { @@ -591,7 +844,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 10, "id": "1715a373", "metadata": {}, "outputs": [], @@ -622,10 +875,36 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 11, "id": "1f1be87f", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Status of Envoys connected to Federation at: 2025-04-22 17:31:16\n", + "\n", + "+----------+----------+---------------------+----------------------+-------------------+\n", + "| Name | Online | Last Updated | Experiment Running | Experiment Name |\n", + "+==========+==========+=====================+======================+===================+\n", + "| Portland | Yes | 2025-04-22 17:31:11 | No | None |\n", + "+----------+----------+---------------------+----------------------+-------------------+\n", + "| Seattle | Yes | 2025-04-22 17:31:15 | No | None |\n", + "+----------+----------+---------------------+----------------------+-------------------+\n" + ] + }, + { + "data": { + "text/plain": [ + "['Portland', 'Seattle']" + ] + }, + "execution_count": 11, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ "federated_runtime.get_envoys()" ] @@ -644,10 +923,59 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 12, "id": "c6d19819", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Aggregator step \"start\" registered\n", + "Collaborator step \"aggregated_model_validation\" registered\n", + "Collaborator step \"train\" registered\n", + "Collaborator step \"local_model_validation\" registered\n", + "Aggregator step \"join\" registered\n", + "Aggregator step \"end\" registered\n", + "Local runtime collaborators = ['Portland', 'Seattle']\n", + "\n", + "New experimental workspace directory structure:\n", + "generated_workspace\n", + "├── .workspace\n", + "├── requirements.txt\n", + "├── src\n", + "│ ├── __init__.py\n", + "│ ├── experiment.py\n", + "│ └── __pycache__\n", + "└── plan\n", + " ├── plan.yaml\n", + " └── defaults\n", + "\n", + "3 directories, 6 files\n", + "Archive created at /home/neeraj/rpc_openfl/openfl_redesign/openfl/openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/workspace/experiment.zip\n", + "\u001b[91m ❌Experiment 'FederatedFlow_TorchMNIST' submission was rejected by the Director\u001b[0m\n", + "\u001b[91m❌ Experiment 'FederatedFlow_TorchMNIST' was rejected by the Director.\u001b[0m\n", + "\u001b[91m❌ Flow execution stopped: FederatedRuntime: Experiment FederatedFlow_TorchMNIST failed to run due to error: Experiment 'FederatedFlow_TorchMNIST' submission was rejected. Stopping execution.\u001b[0m\n" + ] + }, + { + "ename": "Exception", + "evalue": "FederatedRuntime: Experiment FederatedFlow_TorchMNIST failed to run due to error: Experiment 'FederatedFlow_TorchMNIST' submission was rejected. Stopping execution.", + "output_type": "error", + "traceback": [ + "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m", + "\u001b[0;31mException\u001b[0m Traceback (most recent call last)", + "File \u001b[0;32m~/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages/openfl/experimental/workflow/interface/fl_spec.py:198\u001b[0m, in \u001b[0;36mFLSpec._run_federated\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m 197\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;130;01m\\033\u001b[39;00m\u001b[38;5;124m[91m❌ Experiment \u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mexp_name\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m was rejected by the Director.\u001b[39m\u001b[38;5;130;01m\\033\u001b[39;00m\u001b[38;5;124m[0m\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m--> 198\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mExperiment \u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mexp_name\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m submission was rejected. Stopping execution.\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m 199\u001b[0m \u001b[38;5;66;03m# Experiment was submitted successfully\u001b[39;00m\n", + "\u001b[0;31mException\u001b[0m: Experiment 'FederatedFlow_TorchMNIST' submission was rejected. Stopping execution.", + "\nDuring handling of the above exception, another exception occurred:\n", + "\u001b[0;31mException\u001b[0m Traceback (most recent call last)", + "Cell \u001b[0;32mIn[12], line 3\u001b[0m\n\u001b[1;32m 1\u001b[0m flflow\u001b[38;5;241m.\u001b[39mresults \u001b[38;5;241m=\u001b[39m [] \u001b[38;5;66;03m# clear results from previous run\u001b[39;00m\n\u001b[1;32m 2\u001b[0m flflow\u001b[38;5;241m.\u001b[39mruntime \u001b[38;5;241m=\u001b[39m federated_runtime\n\u001b[0;32m----> 3\u001b[0m \u001b[43mflflow\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrun\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n", + "File \u001b[0;32m~/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages/openfl/experimental/workflow/interface/fl_spec.py:134\u001b[0m, in \u001b[0;36mFLSpec.run\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m 132\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_run_local()\n\u001b[1;32m 133\u001b[0m \u001b[38;5;28;01melif\u001b[39;00m \u001b[38;5;28mstr\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_runtime) \u001b[38;5;241m==\u001b[39m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mFederatedRuntime\u001b[39m\u001b[38;5;124m\"\u001b[39m:\n\u001b[0;32m--> 134\u001b[0m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_run_federated\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m 135\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m 136\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mRuntime not implemented\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n", + "File \u001b[0;32m~/miniconda3/envs/opfl_plan_agree/lib/python3.10/site-packages/openfl/experimental/workflow/interface/fl_spec.py:209\u001b[0m, in \u001b[0;36mFLSpec._run_federated\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m 207\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_update_from_flspec_obj(flspec_obj)\n\u001b[1;32m 208\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[0;32m--> 209\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m(\n\u001b[1;32m 210\u001b[0m \u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mFederatedRuntime: Experiment \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mexp_name\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m failed to run due to error: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00me\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m 211\u001b[0m )\n", + "\u001b[0;31mException\u001b[0m: FederatedRuntime: Experiment FederatedFlow_TorchMNIST failed to run due to error: Experiment 'FederatedFlow_TorchMNIST' submission was rejected. Stopping execution." + ] + } + ], "source": [ "flflow.results = [] # clear results from previous run\n", "flflow.runtime = federated_runtime\n", @@ -671,7 +999,7 @@ "source": [ "headers = [\"Rounds\", \"Agg Model Validation Score\", \"Local Train loss\", \"Local Model Validation score\"]\n", "print('********** Simulation results **********')\n", - "print(tabulate(simulation_results, headers=headers, tablefmt=\"outline\"))\n", + "#print(tabulate(simulation_results, headers=headers, tablefmt=\"outline\"))\n", "\n", "print('********** Federation results **********')\n", "federation_results = flflow.results\n", @@ -681,7 +1009,7 @@ ], "metadata": { "kernelspec": { - "display_name": "fed_run", + "display_name": "opfl_plan_agree", "language": "python", "name": "python3" }, @@ -695,7 +1023,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.10.15" + "version": "3.10.16" } }, "nbformat": 4, diff --git a/openfl/experimental/workflow/component/aggregator/aggregator.py b/openfl/experimental/workflow/component/aggregator/aggregator.py index 717436f8a0..8e75271420 100644 --- a/openfl/experimental/workflow/component/aggregator/aggregator.py +++ b/openfl/experimental/workflow/component/aggregator/aggregator.py @@ -220,6 +220,10 @@ async def run_flow(self) -> FLSpec: logger.info(f"Tasks will not be sent to {k}") while not self.collaborator_task_results.is_set(): + # Check if its time to quite + if self.time_to_quit: + logger.info("Time to quit detected. Exiting wait loop") + break len_sel_collabs = len(self.selected_collaborators) len_connected_collabs = len(self.connected_collaborators) if len_connected_collabs < len_sel_collabs: @@ -243,13 +247,16 @@ async def run_flow(self) -> FLSpec: + " collaborators to send results..." ) await asyncio.sleep(Aggregator._get_sleep_time()) + + if self.time_to_quit: + logger.info("Time to quit detected after waiting loop . Stopping aggregator") + break self.collaborator_task_results.clear() f_name = self.next_step if hasattr(self, "instance_snapshot"): self.flow.restore_instance_snapshot(self.flow, list(self.instance_snapshot)) delattr(self, "instance_snapshot") - return self.flow def call_checkpoint( @@ -309,25 +316,32 @@ def get_tasks(self, collaborator_name: str) -> Tuple: f"Aggregator GetTasks function reached from collaborator {collaborator_name}..." ) + + # Check if time_to_quit is set by aggregator . + # If set, send quit job to collaborator and return + if self.time_to_quit: + return self._send_shutdown_signal(collaborator_name) + # If queue of requesting collaborator is empty while self.__collaborator_tasks_queue[collaborator_name].qsize() == 0: # If it is time to then inform the collaborator if self.time_to_quit: - logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...") - self.quit_job_sent_to.append(collaborator_name) - # FIXME: 0, and "" instead of None is just for protobuf compatibility. - # Cleaner solution? - return ( - 0, - "", - None, - Aggregator._get_sleep_time(), - self.time_to_quit, - ) - + return self._send_shutdown_signal(collaborator_name) + # If not time to quit then sleep for 10 seconds time.sleep(Aggregator._get_sleep_time()) + ''' + # Check if quit job is sent to collaborator + if collaborator_name in self.quit_job_sent_to: + logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...") + return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit) + # Check if quit job is sent to all collaborators + if self.all_quit_jobs_sent(): + logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...") + return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit) + ''' + # Get collaborator step, and clone for requesting collaborator next_step, clone = self.__collaborator_tasks_queue[collaborator_name].get() @@ -343,6 +357,14 @@ def get_tasks(self, collaborator_name: str) -> Tuple: 0, self.time_to_quit, ) + + def _send_shutdown_signal(self, collaborator_name: str) -> Tuple: + """Send a shutdown signal to the collaborator.""" + if collaborator_name not in self.quit_job_sent_to: + logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...") + self.quit_job_sent_to.append(collaborator_name) + return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit) + def do_task(self, f_name: str) -> Any: """Execute aggregator steps until transition. @@ -514,7 +536,55 @@ def valid_collaborator_cn_and_id( def all_quit_jobs_sent(self) -> bool: """Assert all quit jobs are sent to collaborators.""" + #logger.info(f"self.quit_job_sent_to {self.quit_job_sent_to} ,self.authorized_cols {self.authorized_cols}") # Added for debugging + # Check if all quit jobs are sent to collaborators return set(self.quit_job_sent_to) == set(self.authorized_cols) + + ''' + def stop(self, failed_collaborator: str = None) -> None: + """Stop the aggregator and all collaborators. + + Args: + failed_collaborator (str, optional): Name of the failed + collaborator. Defaults to None. + """ + if failed_collaborator is not None: + logger.info(f"Stopping aggregator due to failure of {failed_collaborator}.") + else: + logger.info("Stopping aggregator.") + self.time_to_quit = True + self.collaborator_task_results.set() + # Stop all collaborators + for collab in self.authorized_cols: + if collab not in self.quit_job_sent_to: + logger.info(f"Sending quit job to {collab}.") + self.quit_job_sent_to.append(collab) + ''' + def stop(self, failed_collaborator: str = None) -> None: + """Stop the aggregator , failed collaborator and send quit jobs to all collaborators. + + Args: + failed_collaborator (str, optional): Name of the failed + collaborator. Defaults to None. + """ + logger.info("Force stopping the aggregator execution.") + self.time_to_quit = True # Signal All collaborators to quit + + if failed_collaborator: + logger.info(f"Stopping aggregator triggred by {failed_collaborator}.") + if failed_collaborator not in self.quit_job_sent_to: + logger.info(f"Sending quit job to {failed_collaborator}.") + self.quit_job_sent_to.append(failed_collaborator) + + # Send quit jobs to all collaborators. + #for collaborator_name in self.authorized_cols: + #if collaborator_name not in self.quit_job_sent_to: + #logger.info(f"Sending quit job to collaborator '{collaborator_name}'.") + #self.quit_job_sent_to.append(collaborator_name) + + + + the_dragon = """ diff --git a/openfl/experimental/workflow/component/collaborator/collaborator.py b/openfl/experimental/workflow/component/collaborator/collaborator.py index d33250afe4..226d194087 100644 --- a/openfl/experimental/workflow/component/collaborator/collaborator.py +++ b/openfl/experimental/workflow/component/collaborator/collaborator.py @@ -138,6 +138,8 @@ def run(self) -> None: while True: next_step, clone, sleep_time, time_to_quit = self.get_tasks() if time_to_quit: + self.logger.info(f"Collaborator {self.name} received Quit signal. ⛔Shutting down.....") + self.logger.info("Received end of federation signal. Exiting...") break elif sleep_time > 0: time.sleep(sleep_time) diff --git a/openfl/experimental/workflow/component/director/director.py b/openfl/experimental/workflow/component/director/director.py index 07b44f7a18..e1a9718452 100644 --- a/openfl/experimental/workflow/component/director/director.py +++ b/openfl/experimental/workflow/component/director/director.py @@ -18,6 +18,7 @@ ExperimentsRegistry, ) from openfl.experimental.workflow.transport.grpc.exceptions import EnvoyNotFoundError +from openfl.experimental.workflow.component.director.experiment import Status logger = logging.getLogger(__name__) @@ -45,6 +46,7 @@ class Director: envoy_health_check_period (int): The period for health check of envoys in seconds. authorized_cols (list): A list of authorized envoys + review_callback (Optional[Callable]): A callback function for reviewing experiments. """ def __init__( @@ -57,6 +59,7 @@ def __init__( director_config: Optional[Path] = None, envoy_health_check_period: int = 60, install_requirements: bool = True, + review_callback = None, # Add review_callback parameter ) -> None: """Initialize a Director object. @@ -74,6 +77,7 @@ def __init__( in seconds. install_requirements (bool, optional): A flag indicating if the requirements should be installed. Defaults to True. + review_callback (Optional[Callable]): A callback function for reviewing experiments. """ self.tls = tls self.root_certificate = root_certificate @@ -82,7 +86,7 @@ def __init__( self.director_config = director_config self.install_requirements = install_requirements self._flow_status = asyncio.Queue() - + self.review_callback = review_callback # Store the review_callback self.experiments_registry = ExperimentsRegistry() self.col_exp = {} self.col_exp_queues = defaultdict(asyncio.Queue) @@ -172,16 +176,16 @@ async def set_new_experiment( collaborator_names: Iterable[str], experiment_archive_path: Path, ) -> bool: - """Set new experiment. + """Set new experiment and optionally review experiment . Args: - experiment_name (str): String id for experiment. - sender_name (str): The name of the sender. - collaborator_names (Iterable[str]): Names of collaborators. - experiment_archive_path (Path): Path of the experiment. + experiment_name (str): Identifier for the new experiment. + sender_name (str): Initiator of the experiment. + collaborator_names (Iterable[str]): Participating collaborators. + experiment_archive_path (Path): Path to the experiment archive. Returns: - bool : Boolean returned if the experiment register was successful. + bool: True if the experiment is accepted and registered; False otherwise. """ experiment = Experiment( name=experiment_name, @@ -190,10 +194,18 @@ async def set_new_experiment( users=[sender_name], sender=sender_name, ) - + # Check if review callback is enabled + if self.review_callback: + review_approved = await experiment.review_experiment(self.review_callback) + if not review_approved: + logger.warning(f"Experiment '{experiment_name}' was rejected❌ by the Director Admin during review.") + return False # Experiment rejected + + # Add the experiment to the registry self.authorized_cols = collaborator_names self.experiments_registry.add(experiment) - return True + logger.info(f"Experiment '{experiment_name}' was approved✅ and added to the registry.") + return True # Experiment approved async def stream_experiment_stdout( self, experiment_name: str, caller: str @@ -234,6 +246,7 @@ async def stream_experiment_stdout( # Yield none if the queue is empty but the experiment is still running. yield None + def get_experiment_data(self, experiment_name: str) -> Path: """Get experiment data. @@ -315,3 +328,30 @@ def update_envoy_status( ) return self.envoy_health_check_period + + + def set_experiment_failed(self, *, experiment_name: str, collaborator_name: str): + """ Handle experiment Failure triggred by a collaborator. + + Args: + experiment_name (str): String id for experiment. + collaborator_name (str): Name of the collaborator. + """ + if experiment_name not in self.experiments_registry: + logger.warning(f"Experiment '{experiment_name}' not found in registry.") + return + experiment = self.experiments_registry[experiment_name] + aggregator = experiment.aggregator + + if aggregator: + logger.info(f"Stopping aggregator for experiment '{experiment_name}' due to failure by collaborator '{collaborator_name}'.") + aggregator.stop(failed_collaborator=collaborator_name) + + experiment.status = Status.FAILED + logger.info(f"Experiment '{experiment_name}' marked as FAILED.") + + + + + + diff --git a/openfl/experimental/workflow/component/director/experiment.py b/openfl/experimental/workflow/component/director/experiment.py index c85d6b90d9..64902f9208 100644 --- a/openfl/experimental/workflow/component/director/experiment.py +++ b/openfl/experimental/workflow/component/director/experiment.py @@ -9,12 +9,12 @@ from contextlib import asynccontextmanager from enum import Enum, auto from pathlib import Path -from typing import Any, Iterable, List, Optional, Tuple, Union +from typing import Any, Iterable, List, Optional, Tuple, Union, Callable from openfl.experimental.workflow.federated import Plan from openfl.experimental.workflow.transport import AggregatorGRPCServer from openfl.utilities.workspace import ExperimentWorkspace - +from openfl.experimental.workflow.component.aggregator import Aggregator logger = logging.getLogger(__name__) @@ -130,6 +130,7 @@ async def start( _, self.updated_flow = await asyncio.gather( self._run_aggregator_grpc_server( aggregator_grpc_server, + self._wait_for_all_quit_jobs, ), self.aggregator.run_flow(), ) @@ -142,6 +143,40 @@ async def start( return self.status == Status.FINISHED, self.updated_flow + async def review_experiment(self, review_plan_callback: Callable) -> bool: + """Get plan approval in console. + Asynchronously review the experiment plan using the provided callback. + The review runs in a separate thread to avoid blocking the server. + If rejected, the experiment is marked as REJECTED and its archive is deleted. + + Args: + review_plan_callback (Callable): A callable that takes (name, plan_path) and returns a bool. + + Returns: + bool: True if approved, False otherwise. + + """ + logger.debug("Experiment Review starts") + # Extract the workspace for review (without installing requirements) + with ExperimentWorkspace( + self.name, + self.archive_path, + install_requirements=False, + remove_archive=False + ): + loop = asyncio.get_event_loop() + # Call for a review in a separate thread (server is not blocked) + review_approve = await loop.run_in_executor( + None, + review_plan_callback, + self.name, self.plan_path + ) + if not review_approve: + self.status = Status.REJECTED + self.archive_path.unlink(missing_ok=True) + return False + return True + def _create_aggregator_grpc_server( self, *, @@ -183,12 +218,14 @@ def _create_aggregator_grpc_server( @staticmethod async def _run_aggregator_grpc_server( aggregator_grpc_server: AggregatorGRPCServer, + wait_for_all_quit_jobs: Callable[[Aggregator], None], ) -> None: - """Run aggregator. + """" + Run the aggregator gRPC server. Args: - aggregator_grpc_server (AggregatorGRPCServer): The aggregator gRPC - server to run. + aggregator_grpc_server (AggregatorGRPCServer): The aggregator gRPC server to run. + wait_for_all_quit_jobs (Callable): A function to wait for all quit jobs to be sent. """ logger.info("🧿 Starting the Aggregator Service.") grpc_server = aggregator_grpc_server.get_server() @@ -196,15 +233,29 @@ async def _run_aggregator_grpc_server( logger.info("Starting Aggregator gRPC Server") try: - while not aggregator_grpc_server.aggregator.all_quit_jobs_sent(): - # Awaiting quit job sent to collaborators - await asyncio.sleep(10) + # Wait for all quit jobs to be sent + await asyncio.wait_for( + wait_for_all_quit_jobs(aggregator_grpc_server.aggregator), + timeout=None, # No timeout, but can be added if needed + ) logger.debug("Aggregator sent quit jobs calls to all collaborators") + except asyncio.TimeoutError: + logger.error("Timeout while waiting for all quit jobs to be sent.") except KeyboardInterrupt: logger.info("Keyboard interrupt received. Stopping the server.") finally: grpc_server.stop(0) + logger.info("Aggregator gRPC Server stopped.") + async def _wait_for_all_quit_jobs(self, aggregator: Aggregator) -> None: + """ + Wait until all quit jobs are sent. + Args: + aggregator (Aggregator): The aggregator instance. + """ + while not aggregator.all_quit_jobs_sent(): + await asyncio.sleep(10) + logger.debug("Waiting for all quit jobs to be sent.") class ExperimentsRegistry: """ExperimentsList class.""" diff --git a/openfl/experimental/workflow/component/envoy/envoy.py b/openfl/experimental/workflow/component/envoy/envoy.py index fd9a87379c..f1aaac0312 100644 --- a/openfl/experimental/workflow/component/envoy/envoy.py +++ b/openfl/experimental/workflow/component/envoy/envoy.py @@ -9,7 +9,7 @@ import uuid from concurrent.futures import ThreadPoolExecutor from pathlib import Path -from typing import Optional, Union +from typing import Optional, Union, Callable from openfl.experimental.workflow.federated import Plan from openfl.experimental.workflow.transport.grpc.director_client import EnvoyDirectorClient @@ -56,6 +56,7 @@ def __init__( certificate: Optional[Union[Path, str]] = None, tls: bool = True, install_requirements: bool = True, + review_callback: Union[None,Callable] = None, ) -> None: """Initialize a envoy object. @@ -87,6 +88,7 @@ def __init__( # experiment workspace provided by the director self.plan = "plan/plan.yaml" self._health_check_future = None + self.review_callback = review_callback def _create_envoy_dir_client( self, director_host: str, director_port: int @@ -150,6 +152,23 @@ def _run(self) -> None: data_file_path=data_file_path, install_requirements=self.install_requirements, ): + if self.review_callback: + # envoy to review the experiment before running + logger.info("🧿 Reviewing the experiment plan before running...") + # If review_plan_callback is provided, call it with the plan config path and the data file path + if not self.review_callback('plan', 'plan/plan.yaml'): + self._envoy_dir_client.send_experiment_review_result( + experiment_name=experiment_name, + error_description='Experiment is rejected' + f' by Envoy "{self.name}"', + ) + logger.warning(f'⚠️ Experiment "{experiment_name}" was rejected by Envoy "{self.name}".') + continue + logger.debug(f'Experiment "{experiment_name}" is accepted by Envoy "{self.name}".') + + # Run the experiment + logger.info("🚀 Starting the experiment...") + # Set the experiment running flag to True to indicate that experiment is running self.is_experiment_running = True self._run_collaborator() except Exception as exc: diff --git a/openfl/experimental/workflow/interface/cli/cli_helper.py b/openfl/experimental/workflow/interface/cli/cli_helper.py index e45b3cdd09..7d5799fb1a 100644 --- a/openfl/experimental/workflow/interface/cli/cli_helper.py +++ b/openfl/experimental/workflow/interface/cli/cli_helper.py @@ -6,12 +6,13 @@ import os import re +import time from itertools import islice from os import environ from pathlib import Path from sys import argv -from click import echo, style +from click import echo, style, confirm, open_file from yaml import FullLoader, load FX = argv[0] @@ -140,3 +141,34 @@ def replace_line_in_file(line, line_num_to_replace, filename): else: f.write(i) f.truncate() + +def review_plan_callback(file_name: str, file_path) -> bool: + """ + CLI-based review callback used by the Director or Envoy to approve a plan file. + + Args: + file_name (str): Display name of the file to be reviewed. + file_path (Union[str, Path]): Path to the file containing the plan. + + Returns: + bool: True if the user approves the file; False otherwise. + """ + DISPLAY_DELAY_SECONDS = 3 + + echo(style(f"🧿 Please review the contents of 📂 {file_name} before proceeding...", fg="green", bold=True)) + time.sleep(DISPLAY_DELAY_SECONDS) + + try: + with open_file(file_path, "r") as f: + echo(f.read()) + except Exception as e: + echo(style(f"⚠️ Failed to read file: {e}", fg="red", bold=True)) + return False + + # Ask for user confirmation to accept the file + if confirm(style(f"Do you want to accept the 📂 {file_name}❔", fg="green", bold=True)): + echo(style(f"{file_name} accepted!", fg="green", bold=True)) + return True + else: + echo(style(f"{file_name} rejected!", fg="red", bold=True)) + return False # Return False on rejection \ No newline at end of file diff --git a/openfl/experimental/workflow/interface/cli/director.py b/openfl/experimental/workflow/interface/cli/director.py index b7e202715e..9b383f9536 100644 --- a/openfl/experimental/workflow/interface/cli/director.py +++ b/openfl/experimental/workflow/interface/cli/director.py @@ -15,6 +15,7 @@ from openfl.experimental.workflow.transport import DirectorGRPCServer from openfl.utilities import merge_configs from openfl.utilities.path_check import is_directory_traversal +from openfl.experimental.workflow.interface.cli.cli_helper import review_plan_callback logger = logging.getLogger(__name__) @@ -104,6 +105,7 @@ def start(director_config_path, tls, root_certificate, private_key, certificate) gte=1, lte=24 * 60 * 60, ), + Validator("settings.review_experiment", default=False), ], ) @@ -116,6 +118,11 @@ def start(director_config_path, tls, root_certificate, private_key, certificate) if config.certificate: config.certificate = Path(config.certificate).absolute() + # Check if review_experiment is enabled in the configuration + review_experiment = config.settings.review_experiment + # Set the review callback if review_experiment is True + review_callback = review_plan_callback if review_experiment else None + director_server = DirectorGRPCServer( director_cls=Director, tls=tls, @@ -127,5 +134,6 @@ def start(director_config_path, tls, root_certificate, private_key, certificate) envoy_health_check_period=config.settings.envoy_health_check_period, install_requirements=config.settings.install_requirements, director_config=director_config_path, + review_callback=review_callback, ) director_server.start() diff --git a/openfl/experimental/workflow/interface/cli/envoy.py b/openfl/experimental/workflow/interface/cli/envoy.py index 1c16ec9990..4acf72d4c0 100644 --- a/openfl/experimental/workflow/interface/cli/envoy.py +++ b/openfl/experimental/workflow/interface/cli/envoy.py @@ -14,6 +14,7 @@ from openfl.experimental.workflow.component.envoy import Envoy from openfl.utilities import is_fqdn, merge_configs from openfl.utilities.path_check import is_directory_traversal +from openfl.experimental.workflow.interface.cli.cli_helper import review_plan_callback logger = logging.getLogger(__name__) @@ -126,6 +127,7 @@ def start_( }, ), Validator("params.install_requirements", default=True), + Validator("settings.review_experiment", default=False), ], ) @@ -136,6 +138,8 @@ def start_( else: install_requirements = False + + if config.root_certificate: config.root_certificate = Path(config.root_certificate).absolute() if config.private_key: @@ -143,6 +147,12 @@ def start_( if config.certificate: config.certificate = Path(config.certificate).absolute() + # Check if review_experiment is enabled in the settings + review_experiment = config.settings.review_experiment + + # Set the review callback if review_experiment is True + review_callback = review_plan_callback if review_experiment else None + envoy = Envoy( envoy_name=envoy_name, director_host=config.settings.director_host, @@ -153,6 +163,7 @@ def start_( certificate=config.certificate, tls=tls, install_requirements=install_requirements, + review_callback=review_callback, ) envoy.start() diff --git a/openfl/experimental/workflow/interface/fl_spec.py b/openfl/experimental/workflow/interface/fl_spec.py index 3e8365458b..a5238ac73d 100644 --- a/openfl/experimental/workflow/interface/fl_spec.py +++ b/openfl/experimental/workflow/interface/fl_spec.py @@ -126,13 +126,19 @@ def runtime(self, runtime: Type[Runtime]) -> None: def run(self) -> None: """Starts the execution of the flow.""" - # Submit flow to Runtime - if str(self._runtime) == "LocalRuntime": - self._run_local() - elif str(self._runtime) == "FederatedRuntime": - self._run_federated() - else: - raise Exception("Runtime not implemented") + try: + # Submit flow to Runtime + if str(self._runtime) == "LocalRuntime": + self._run_local() + elif str(self._runtime) == "FederatedRuntime": + self._run_federated() + else: + raise Exception("Runtime not implemented") + except Exception as e: + # Stop execution and print error message + print(f"\033[91m❌ Flow execution stopped: {e}\033[0m") + raise + def _run_local(self) -> None: """Executes the flow using LocalRuntime.""" @@ -186,7 +192,12 @@ def _run_federated(self) -> None: try: # Prepare workspace and submit it for the FederatedRuntime archive_path, exp_name = self.runtime.prepare_workspace_archive() - self.runtime.submit_experiment(archive_path, exp_name) + submission_result = self.runtime.submit_experiment(archive_path, exp_name) + if not submission_result: + print(f"\033[91m❌ Experiment '{exp_name}' was rejected by the Director.\033[0m") + raise Exception(f"Experiment '{exp_name}' submission was rejected. Stopping execution.") + # Experiment was submitted successfully + print(f"\033[92m✅ Experiment '{exp_name}' approved and running.\033[0m") # Stream the experiment's stdout if the checkpoint is enabled if self._checkpoint: self.runtime.stream_experiment_stdout(exp_name) diff --git a/openfl/experimental/workflow/protocols/director.proto b/openfl/experimental/workflow/protocols/director.proto index 0e4a16e548..14497ceb8d 100644 --- a/openfl/experimental/workflow/protocols/director.proto +++ b/openfl/experimental/workflow/protocols/director.proto @@ -14,6 +14,7 @@ service Director { rpc WaitExperiment(WaitExperimentRequest) returns (WaitExperimentResponse) {} rpc GetExperimentData(GetExperimentDataRequest) returns (stream ExperimentData) {} rpc UpdateEnvoyStatus(UpdateEnvoyStatusRequest) returns (UpdateEnvoyStatusResponse) {} + rpc SendExperimentReviewResult (SendExperimentReviewResultRequest) returns (SendExperimentReviewResultResponse) {} //Runtime RPCs rpc SetNewExperiment(stream ExperimentInfo) returns (SetNewExperimentResponse) {} @@ -58,6 +59,15 @@ message UpdateEnvoyStatusResponse { google.protobuf.Duration health_check_period = 1; } +message SendExperimentReviewResultRequest { + string experiment_name = 1; + string collaborator_name = 2; + uint32 error_code = 3; + string error_description = 4; +} + +message SendExperimentReviewResultResponse {} + message ExperimentInfo { string name = 1; repeated string collaborator_names = 2; @@ -65,7 +75,7 @@ message ExperimentInfo { } message SetNewExperimentResponse { - bool status = 1; + bool status = 1; // True if approved, False if rejected } message EnvoyInfo { diff --git a/openfl/experimental/workflow/runtime/federated_runtime.py b/openfl/experimental/workflow/runtime/federated_runtime.py index 6420b6d8dc..51cf5a11d5 100644 --- a/openfl/experimental/workflow/runtime/federated_runtime.py +++ b/openfl/experimental/workflow/runtime/federated_runtime.py @@ -154,13 +154,16 @@ def prepare_workspace_archive(self) -> Tuple[Path, str]: ) return archive_path, exp_name - def submit_experiment(self, archive_path, exp_name) -> None: + def submit_experiment(self, archive_path, exp_name) -> bool: """ Submits experiment archive to the director Args: archive_path (str): Archive file path containing the workspace. exp_name (str): The name of the experiment to be submitted. + + Returns: + bool: True if experiment was accepted or submitted succesfully , False if rejected or not submitted to director """ try: response = self._runtime_dir_client.set_new_experiment( @@ -169,12 +172,14 @@ def submit_experiment(self, archive_path, exp_name) -> None: self.experiment_submitted = response.status if self.experiment_submitted: - print( - f"\033[92mExperiment {exp_name} was successfully " - "submitted to the director!\033[0m" - ) + print(f"\033[92m ✅Experiment '{exp_name}' was successfully submitted!\033[0m") + return True else: - print(f"\033[91mFailed to submit experiment '{exp_name}' to the director.\033[0m") + #print(f"\033[91m ❌Experiment '{exp_name}' submission was rejected by the Director\033[0m") + return False + #raise RuntimeError( + #f"Failed to submit experiment '{exp_name}' to the director. Rejected by director." + #) finally: self.remove_workspace_archive(archive_path) diff --git a/openfl/experimental/workflow/transport/grpc/director_client.py b/openfl/experimental/workflow/transport/grpc/director_client.py index 189ab00937..755cdee2c0 100644 --- a/openfl/experimental/workflow/transport/grpc/director_client.py +++ b/openfl/experimental/workflow/transport/grpc/director_client.py @@ -172,6 +172,23 @@ def send_health_check( health_check_period = response.health_check_period.seconds return health_check_period + + + def send_experiment_review_result( + self, + experiment_name: str, + error_code: int = 1, + error_description: str = '' + ) -> None: + + """Set the experiment failed.""" + request = director_pb2.SendExperimentReviewResultRequest( + experiment_name=experiment_name, + collaborator_name=self.envoy_name, + error_code=error_code, + error_description=error_description + ) + self.stub.SendExperimentReviewResult(request) class RuntimeDirectorClient: diff --git a/openfl/experimental/workflow/transport/grpc/director_server.py b/openfl/experimental/workflow/transport/grpc/director_server.py index 608d2a7743..1ef9715331 100644 --- a/openfl/experimental/workflow/transport/grpc/director_server.py +++ b/openfl/experimental/workflow/transport/grpc/director_server.py @@ -55,6 +55,7 @@ def __init__( listen_port: int = 50051, envoy_health_check_period: int = 0, director_config: Optional[Path] = None, + review_callback=None, # Add review_callback parameter **kwargs, ) -> None: """ @@ -78,6 +79,7 @@ def __init__( listen_port (int, optional): The port to listen on. Defaults to 50051. director_config (Optional[Path]): Path to director_config file + review_callback (Optional[Callable]): Callback function for reviewing experiment plans. **kwargs: Additional keyword arguments. """ super().__init__() @@ -93,6 +95,7 @@ def __init__( certificate=self.certificate, envoy_health_check_period=envoy_health_check_period, director_config=director_config, + review_callback=review_callback, # Pass review_callback to Director **kwargs, ) @@ -311,7 +314,8 @@ async def SetNewExperiment(self, stream, context) -> director_pb2.SetNewExperime experiment_archive_path=data_file_path, ) - logger.info("Experiment %s registered", request.name) + #logger.info("Experiment %s registered", request.name) + logger.info(f"Experiment '{request.name}' registration status: {'Approved' if is_accepted else 'Rejected'}") return director_pb2.SetNewExperimentResponse(status=is_accepted) async def GetFlowState(self, request, context) -> director_pb2.GetFlowStateResponse: @@ -351,3 +355,23 @@ async def GetExperimentStdout( await asyncio.sleep(1) continue yield director_pb2.GetExperimentStdoutResponse(**stdout_dict) + + async def SendExperimentReviewResult(self, request, context): + """Set the experiment failed.""" + response = director_pb2.SendExperimentReviewResultResponse() + if self.get_caller(context) != CLIENT_ID_DEFAULT: + return response + logger.info( + '[Experiment Rejection] Collaborator "%s" rejected experiment "%s" ' + 'with Reason: (%s). ' + 'Initiating experiment termination.', + request.collaborator_name, + request.experiment_name, + request.error_description + ) + self.director.set_experiment_failed( + experiment_name=request.experiment_name, + collaborator_name=request.collaborator_name + ) + + return response