Remove dead code

Signed-off-by: Shah, Karan <karan.shah@intel.com>

Author: MasterSkepticista

openfl/component/aggregator/aggregator.py

@@ -16,7 +16,7 @@
 from openfl.databases import PersistentTensorDB, TensorDB
 from openfl.interface.aggregation_functions import SecureWeightedAverage, WeightedAverage
 from openfl.pipelines import NoCompressionPipeline, TensorCodec
-from openfl.protocols import base_pb2, utils
+from openfl.protocols import utils
 from openfl.protocols.base_pb2 import NamedTensor
 from openfl.utilities import TaskResultKey, TensorKey, change_tags
 
@@ -83,39 +83,13 @@ def __init__(
         single_col_cert_common_name=None,
         compression_pipeline=None,
         db_store_rounds=1,
-        initial_tensor_dict=None,
         log_memory_usage=False,
         write_logs=False,
         callbacks: Optional[List] = [],
         persist_checkpoint=True,
         persistent_db_path=None,
         secure_aggregation=False,
     ):
-        """Initializes the Aggregator.
-
-        Args:
-            aggregator_uuid (int): Aggregation ID.
-            federation_uuid (str): Federation ID.
-            authorized_cols (list of str): The list of IDs of enrolled
-                collaborators.
-            init_state_path (str): The location of the initial weight file.
-            best_state_path (str): The file location to store the weight of
-                the best model.
-            last_state_path (str): The file location to store the latest
-                weight.
-            assigner: Assigner object.
-            straggler_handling_policy (optional): Straggler handling policy.
-            rounds_to_train (int, optional): Number of rounds to train.
-                Defaults to 256.
-            single_col_cert_common_name (str, optional): Common name for single
-                collaborator certificate. Defaults to None.
-            compression_pipeline (optional): Compression pipeline. Defaults to
-                NoCompressionPipeline.
-            db_store_rounds (int, optional): Rounds to store in TensorDB.
-                Defaults to 1.
-            initial_tensor_dict (dict, optional): Initial tensor dictionary.
-            callbacks: List of callbacks to be used during the experiment.
-        """
         self.round_number = 0
         self.next_model_round_number = 0
 
@@ -205,20 +179,10 @@ def __init__(
             last_state_path=self.last_state_path,
         )
 
-        if initial_tensor_dict:
-            self._load_initial_tensors_from_dict(initial_tensor_dict)
-            self.model = utils.construct_model_proto(
-                tensor_dict=initial_tensor_dict,
-                round_number=0,
-                tensor_pipe=self.compression_pipeline,
-            )
-        else:
-            if self.connector:
-                # The model definition will be handled by the respective framework
-                self.model = {}
-            else:
-                self.model: base_pb2.ModelProto = utils.load_proto(self.init_state_path)
-                self._load_initial_tensors()  # keys are TensorKeys
+        self.model = {}
+        if not self.connector:
+            self.model = utils.load_proto(self.init_state_path)
+            self._load_initial_tensors()  # keys are TensorKeys
 
         self._secure_aggregation_enabled = secure_aggregation
         if self._secure_aggregation_enabled:
@@ -337,23 +301,6 @@ def _load_initial_tensors(self):
         self.tensor_db.cache_tensor(tensor_key_dict)
         logger.debug("This is the initial tensor_db: %s", self.tensor_db)
 
-    def _load_initial_tensors_from_dict(self, tensor_dict):
-        """Load all of the tensors required to begin federated learning.
-
-        Required tensors are: \
-            1. Initial model.
-
-        Returns:
-            None
-        """
-        tensor_key_dict = {
-            TensorKey(k, self.uuid, self.round_number, False, ("model",)): v
-            for k, v in tensor_dict.items()
-        }
-        # all initial model tensors are loaded here
-        self.tensor_db.cache_tensor(tensor_key_dict)
-        logger.debug("This is the initial tensor_db: %s", self.tensor_db)
-
     def _save_model(self, round_number, file_path):
         """Save the best or latest model.
 
@@ -485,10 +432,7 @@ def get_tasks(self, collaborator_name):
 
         # first, if it is time to quit, inform the collaborator
         if self._time_to_quit():
-            logger.info(
-                "Sending signal to collaborator %s to shutdown...",
-                collaborator_name,
-            )
+            logger.info("Sending signal to collaborator %s to shutdown...", collaborator_name)
             self.quit_job_sent_to.append(collaborator_name)
 
             tasks = None
@@ -853,7 +797,6 @@ def _compute_validation_related_task_metrics(self, task_name) -> dict:
         )
         # Leave out straggler for the round even if they've partially
         # completed given tasks
-        collaborators_for_task = []
         collaborators_for_task = [
             c for c in all_collaborators_for_task if c in self.collaborators_done
         ]
@@ -925,11 +868,7 @@ def _compute_validation_related_task_metrics(self, task_name) -> dict:
                                 f"{agg_results:f}"
                             )
                             self._save_model(round_number, self.best_state_path)
-                        else:
-                            logger.info(
-                                f"Round {round_number}: best score observed {agg_results:f} "
-                                "(model not saved in evaluation mode)"
-                            )
+
             if "trained" in tags:
                 self._prepare_trained(tensor_name, origin, round_number, report, agg_results)
 

openfl/federated/plan/plan.py

@@ -378,16 +378,11 @@ def get_tasks(self):
             tasks[task]["aggregation_type"] = aggregation_type
         return tasks
 
-    def get_aggregator(self, tensor_dict=None):
+    def get_aggregator(self):
         """Get federation aggregator.
 
         This method retrieves the federation aggregator. If the aggregator
-        does not exist, it is built using the configuration settings and the
-        provided tensor dictionary.
-
-        Args:
-            tensor_dict (dict, optional): The initial tensor dictionary to use
-                when building the aggregator. Defaults to None.
+        does not exist, it is built using the configuration settings.
 
         Returns:
             self.aggregator_ (Aggregator): The federation aggregator.
@@ -415,7 +410,7 @@ def get_aggregator(self, tensor_dict=None):
         # TODO: Load callbacks from plan.
 
         if self.aggregator_ is None:
-            self.aggregator_ = Plan.build(**defaults, initial_tensor_dict=tensor_dict)
+            self.aggregator_ = Plan.build(**defaults)
 
         return self.aggregator_