- implement a new ModeBasedAssigner class extending RandomGroupedAssi…

…gner

- update defaults/assigner.yaml to use ModeBasedAssigner with default mode to train_and_validate
- update fedeval sample workspace to use default assigner, tasks and aggregator
- use of federated-evaluation/aggregator.yaml for FedEval specific workspace example to use round_number as 1
- removed assigner and tasks yaml from defaults/federated-evaluation, superseded by default assigner/tasks
- add happy flow tests for ModeBasedAssigner
- TODO revert workflow change post validation on workflow run in draft mode
Signed-off-by: Shailesh Pant <[email protected]>

.github/workflows/pytest_coverage.yml

@@ -18,7 +18,7 @@ env:
 
 jobs:
   build:
-    if: github.event.pull_request.draft == false
+    if: github.event.pull_request.draft == true
     runs-on: ubuntu-latest
     timeout-minutes: 15
 

openfl-workspace/torch_cnn_mnist_fed_eval/plan/plan.yaml

@@ -34,10 +34,12 @@ network :
   defaults : plan/defaults/network.yaml
 
 assigner :
-  defaults : plan/defaults/federated-evaluation/assigner.yaml
-
+  defaults : plan/defaults/assigner.yaml
+  settings :
+    mode : evaluate
+
 tasks :
-  defaults : plan/defaults/federated-evaluation/tasks_torch.yaml
+  defaults : plan/defaults/tasks_torch.yaml
 
 compression_pipeline :
   defaults : plan/defaults/compression_pipeline.yaml

openfl-workspace/workspace/plan/defaults/assigner.yaml

@@ -1,4 +1,4 @@
-template : openfl.component.RandomGroupedAssigner
+template : openfl.component.ModeBasedAssigner
 settings :
   task_groups  :
     - name       : train_and_validate
@@ -7,3 +7,8 @@ settings :
         - aggregated_model_validation
         - train
         - locally_tuned_model_validation
+    - name       : evaluate
+      percentage : 1.0
+      tasks      :
+        - aggregated_model_validation
+  mode: train_and_validate

openfl/component/__init__.py

@@ -5,6 +5,7 @@
 from openfl.component.aggregator.aggregator import Aggregator
 from openfl.component.assigner.assigner import Assigner
 from openfl.component.assigner.random_grouped_assigner import RandomGroupedAssigner
+from openfl.component.assigner.mode_based_assigner import ModeBasedAssigner
 from openfl.component.assigner.static_grouped_assigner import StaticGroupedAssigner
 from openfl.component.collaborator.collaborator import Collaborator
 from openfl.component.straggler_handling_functions.cutoff_time_based_straggler_handling import (

openfl/component/assigner/__init__.py

@@ -5,3 +5,4 @@
 from openfl.component.assigner.assigner import Assigner
 from openfl.component.assigner.random_grouped_assigner import RandomGroupedAssigner
 from openfl.component.assigner.static_grouped_assigner import StaticGroupedAssigner
+from openfl.component.assigner.mode_based_assigner import ModeBasedAssigner

openfl/component/assigner/mode_based_assigner.py

@@ -0,0 +1,62 @@
+# Copyright 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+
+"""Mode based assigner module."""
+
+from openfl.component.assigner.random_grouped_assigner import RandomGroupedAssigner
+
+
+class ModeBasedAssigner(RandomGroupedAssigner):
+    r"""The task assigner maintains a list of tasks.
+
+    Also it decides the policy for which collaborator should run those tasks
+    There may be many types of policies implemented, but a natural place to
+    start is with a:
+
+        - ModeBasedAssigner :
+            Given a set of task groups and mode it filters the task groups based
+            on mode. It futher enforces checks for specific modes.
+            Post filtering it deletgates the task assignment to RandomGroupedAssigner.
+
+    Attributes:
+        task_groups* (list of object): Task groups to assign.
+        mode* (str): Mode to determine task assignments.
+
+    .. note::
+        \* - Plan setting.
+    """
+
+    def __init__(self, task_groups, mode, **kwargs):
+        """Initializes the ModeBasedAssigner.
+
+        Args:
+            task_groups (list of object): Task groups to assign.
+            mode (str): Mode to determine task assignments.
+            **kwargs: Additional keyword arguments.
+        """
+        self.task_groups = task_groups
+        self.mode = mode
+        super().__init__(task_groups=task_groups, **kwargs)
+
+    def define_task_assignments(self):
+        """Define task assignments for each round and collaborator.
+
+        This method filters tasks for the
+        collaborators for each round based on the mode.
+                
+        Args:
+            None
+
+        Returns:
+            None
+        """
+        self.task_groups = [
+            group for group in self.task_groups
+            if group["name"] == self.mode
+            ]
+        if self.mode == "evaluate" :
+            assert (
+                self.rounds == 1
+                ), "Number of rounds should be 1 for evaluate mode"
+        super().define_task_assignments()

tests/openfl/component/assigner/test_mode_based_assigner.py

@@ -0,0 +1,114 @@
+import pytest
+from openfl.component.assigner.mode_based_assigner import ModeBasedAssigner
+
+@pytest.fixture
+def sample_tasks():
+    return ["task1", "task2", "task3", "task4"]
+
+@pytest.fixture
+def sample_authorized_cols():
+    return ["col1", "col2", "col3"]
+
+@pytest.fixture
+def sample_task_groups():
+    return [
+        {"name": "train", "tasks": ["task1", "task2"], "percentage": 1.0},
+        {"name": "evaluate", "tasks": ["task3"], "percentage": 1.0},
+        {"name": "random-mode", "tasks": ["task4"], "percentage": 1.0}
+    ]
+
+def test_init_with_valid_mode(sample_tasks, sample_authorized_cols):
+    """Test initialization with valid mode and task groups."""
+    task_groups = [{"name": "train", "tasks": ["task1"], "percentage": 1.0}]
+    assigner = ModeBasedAssigner(
+        task_groups=task_groups,
+        mode="train",
+        tasks=sample_tasks,
+        authorized_cols=sample_authorized_cols,
+        rounds_to_train=3
+    )
+    assert assigner.mode == "train"
+    assert assigner.task_groups == task_groups
+
+def test_define_task_assignments_train_mode(sample_task_groups, sample_tasks, sample_authorized_cols):
+    """Test task assignments filtering for train mode."""
+    assigner = ModeBasedAssigner(
+        task_groups=sample_task_groups,
+        mode="train",
+        tasks=sample_tasks,
+        authorized_cols=sample_authorized_cols,
+        rounds_to_train=3
+    )
+    assigner.define_task_assignments()
+    assert len(assigner.task_groups) == 1
+    assert assigner.task_groups[0]["name"] == "train"
+
+def test_define_task_assignments_evaluate_mode(sample_task_groups, sample_tasks, sample_authorized_cols):
+    """Test task assignments filtering for evaluate mode with rounds=1."""
+    assigner = ModeBasedAssigner(
+        task_groups=sample_task_groups,
+        mode="evaluate",
+        tasks=sample_tasks,
+        authorized_cols=sample_authorized_cols,
+        rounds_to_train=1
+    )
+    assigner.define_task_assignments()
+    assert len(assigner.task_groups) == 1
+    assert assigner.task_groups[0]["name"] == "evaluate"
+
+# def test_evaluate_mode_with_invalid_rounds(sample_task_groups, sample_tasks, sample_authorized_cols):
+#     """Test that evaluate mode raises error when rounds > 1."""
+#     assigner = ModeBasedAssigner(
+#         task_groups=sample_task_groups,
+#         mode="evaluate",
+#         tasks=sample_tasks,
+#         authorized_cols=sample_authorized_cols,
+#         rounds_to_train=2
+#     )
+#     with pytest.raises(AssertionError):
+#         assigner.define_task_assignments()
+
+# def test_empty_task_groups_after_filtering(sample_tasks, sample_authorized_cols):
+#     """Test behavior when no task groups match the specified mode."""
+#     task_groups = [{"name": "train", "tasks": ["task1"], "percentage": 1}]
+#     assigner = ModeBasedAssigner(
+#         task_groups=task_groups,
+#         mode="non_existent_mode",
+#         tasks=sample_tasks,
+#         authorized_cols=sample_authorized_cols,
+#         rounds_to_train=1
+#     )
+#     with pytest.raises(AssertionError):
+#         assigner.define_task_assignments()
+
+def test_multiple_matching_tg(sample_tasks, sample_authorized_cols):
+    """Test behavior when multiple task groups match the mode."""
+    task_groups = [
+        {"name": "train", "tasks": ["task1"], "percentage": 1.0},
+        {"name": "train", "tasks": ["task2"], "percentage": 1.0}
+    ]
+    assigner = ModeBasedAssigner(
+        task_groups=task_groups,
+        mode="train",
+        tasks=sample_tasks,
+        authorized_cols=sample_authorized_cols,
+        rounds_to_train=1
+    )
+    with pytest.raises(AssertionError, match="Task group percentages must sum to 100%"):
+        assigner.define_task_assignments()
+
+# def test_multiple_matching_tg_percentage_error(sample_tasks, sample_authorized_cols):
+#     """Test behavior when multiple task groups match the mode."""
+#     task_groups = [
+#         {"name": "train", "tasks": ["task1"], "percentage": 1},
+#         {"name": "train", "tasks": ["task2"], "percentage": 1}
+#     ]
+#     assigner = ModeBasedAssigner(
+#         task_groups=task_groups,
+#         mode="train",
+#         tasks=sample_tasks,
+#         authorized_cols=sample_authorized_cols,
+#         rounds_to_train=1
+#     )
+#     with pytest.raises(AssertionError):
+#         assigner.define_task_assignments()