[Feature] Include realistic mobility and frontend-controller adaptation (#41)

* Servicio de conexión UDP

Implementación de servicio cliente/servidor para que dispositivos ajenos a la federación entablen comunicación

* fix_filename

Changed file name: nebulamulticasting from nebulaconnection

* fix_nebulamulticasting

Additions:
-communication
  - init_ecs
  - stop_ecs
Remove:
-nebulamulticasting
  -stop condition on nebula_server

* feat_mobility_module

Additions:
-neighbormanagement folder
-mobility messages
-proto files updated
-functionalities on comunnications and engine to new messages
- get_loss on nebula model
- model_info on propagator

* fix_function_arguments

* fix_compiler_errors

type notation on factory methods

* feat_timer_generator

Additions:
-TimerGenerator class to generate timer used for waiting updates from nodes
-Weight_modifier for updates received if NodeManager is up

* fix_timer_integrated

Additions:
- TimerGenerator functions and references to be used
- Updated learning cycle to use timer if mobility is up

* fix_lock_release_excep

check lock is acquire

* fix_waiting_logic

Changed the way the node wait and update its state of waiting.

* fixed_some_mobility_configurations

* update_nebula_p2b.py

* feat_mobility_upgrade

.External_connection_service working
-Messages and callbacks integrated

* feat_late_starting_trainning

-late nodes start trainning
-topology structures set apropiately

* feat_LateNodes_train

Later creation nodes now can be integrated into the trainning process

* fix_fast_push

fast push to sinchronize network integrated

* feat_weights_modifiers

Weight modifiers applied to late connected nodes

* fix_slow_push_strategy

slow push strategy working properly
fixing metrics from late connection nodes

* fix_metric_delay

Metric delay done for late creation nodes

* feat_fast_push

fast push integrated
weight strategy integrated
fix modified weights

* fix_fast_reboot

fast reboot when device arrives fully integrated

* feat_upgrading_network_robustness

mechanisms itnegrated:
.-reconnect_to_federation
.-upgrade_connection_robustness

* fix_ecs_run_shutdown

* fix_info_points

* feat_nebulamulticasting_on_off

-fixed learnign rate error
-fixed push sync error
-fix sync errors

* fix_errors_reestructuring

* fix_restructure_errors

fixed all restructure errors found
service working

* feat_connection_optimizator

Connection optimizator to clear inactive connections

* feat_network_optimization

networkoptimizer as a controller
connectionoptimizer to clean inactive connections
timergenerator to generate dynamic timeouts for aggregation

* fix_minor_errors

* fix_remove_weight_error

* fix_solving_distributions

* feat_additional_data_dist_png

additional nodes now show their data distributions

* fix_additional_nodes_ip

* ft_test_setup

* fix_no_coinciding_samples

* fix_update

* feat_defaultMH

default model handler integrated
refactor fastreboot

* fix_error_defaultHM

* fix_wrong_payload

* fix_keyerror_np

* fix_general_errors

* fix_reestructure_loop

* fix_slow_push_issue

* fix:concurrency_issue

* fix_get_neighbors_np

* daily_update

* change_scenario_config

* fix_momentum

* fix_com_error

* updt_momentum

* feat_momentum

momemtum logic implemented

* feat_momemtum_penalty

* updt_momemtum_penalty_ext

* update momemtum

* update_messages_refactor

* fix_msg_errors

* fix_messages_factory

* fix_error

* fix_factory_message_action

* fix_message_factory

feat message_template

* feat_refactor_messages

* fix_clean_code

* fix_refactor_communciations

* fix_handle_model_error

await was required

* feat_message_events

* fix_error_msg

* fix_event_error

* fix_errors

* update_momemtum

* updt

* fix_momemtum_config

* optimization_sinc

* opt_sinc++

* fix_momentum

* fix_disconnect_error

* fix_disconnection_node

* fix_TCP_temporary_port

* fix_notself_agg

* fix_tcp_ports

* feat_node_disconnection

* feat_blacklist

* fix_resinc_error

* fix_resinc_after_disc

* fix_resinc_Node

* feat_target_attacks

* feature select changing targets

* feat standar mobility strategies

* feataure update storage

* fix update storage errors

* fix_no_round_mechs

* fix_error

* Feature update handlers interface

* feature dfl no rounds

* fix updates handling and ecs service

* fix_propagator_error

* opt_test_mobility

* refactor situational awareness

* feature situational awareness module functionalities

* feature nebula discover service asynchronous

* optimization code

* feature beacon service

* fix fully integrated beacon

* feature geolocalization in beacon

* fix daily update

* feature nebula gps service

* fix mobility errors

* feature updating mobility module

* fix mobility errors

* fix missing await

* fix generate network conditions

* fix mobility low threshold error

* feature network simulator

* feature integrated nebula network simulator

* feature SA submodules

* fix additional node network conditions

* fix network conditions fist attemp

* feature training policy interface

* feature event system for addon functionalities

* feat aggregation event

* feature event system integrated

* feature QDS tp - update received event

* feature round start event

* feature speed oriented selection

* feature CFL implementation for udpate storage

* feature beacon received event

* feature sos sa strategy

* opt sat sos

* feature sat hts

* fix error offers accepted after stopped lt process

* opt engine

* fix evaluation before aggregation

* fix additional participants datasets

* opt nebuladataset factory

* saving wating scenarios in the database

* feature hybrid datasets

* feat datasets 'n' splitted

* feature split dataset IID subsets

* wip hybrid datasets

* fix dirichlet subset generated

* feature unbalanced IID hybrid datasets

* feature cifar10 hybrid data partitioning

* fix mnist error

* fix dflupdatehandler & qds

* updt

* opt space

* feature nebula plugin loader

* refactor(controller): migrate user modifications from front-end to controller

Moved the logic for user modifications from the front-end to the controller to enhance separation of concerns and simplify maintenance.

* feature SA command

* fix communciation manager importatition

* feature connectivity commands

* refactor previous created endpoints and scenarios endpoints added

* feature suggestionbuffer

* fix suggestionbuffer

* daile update

* fix endpoints for scenarios

* feature integrated suggestions system

* fix error interface

* opt suggestion buffer

* feat integrated more events

* refactor communication manager property

* fix early updates received before starting learning

* feature training policies as SA Agents

* fix owner missing SACommands

* morning update

* removed unused parameters

* databases removed of frontend docker container

* fix monitor page and node related endpoints

* fix samodule mediate function

* fix sa module arbitatrion

* feature close to integrate node forgiveness

* feature system monitoring + forget nodes

* feature static arbitatrion policy

* feature behavior reputation

* feature reputation messages

* feature dinamically loading sa components

* refactor situational awareness module

* refactor SA module

* feature consistency reputation

* feature advanced consistency metrics

* fix merge errors

* refactor soem stuff

* refactor private methods

* refactor pluging loader

* refactor ring topology np

* feature sa frontend

* fix ring error

* fix random topology cs

* refactor intro message

* feature sar components dinamically loaded

* fix ring error

* fix not simultaneos ring extra participants

* pre integration

* sa commentaries

* optimization  ring error

* fix strict ring topology

* feature undirect connections inactivity

* feature connetion priority

* daily update

* controller and frontend splitted

* upgrade collaborative rep

* refactor controller folder created

* upgrade collaborative

* feature distance neighbor policy

* node endpoints created in the controller

* fix distance errors

* fix await erros

* feature distance candidate selector

* fix mobility longitude

* default mobility

* fix mob

* additional merge changes

* remove bad dir

* added docstrings for the frontend

* fix same owner commands conflict

* improve node dynamism

* fix filtered discovers

* feature ttl for corfimations

* fix rep messages

* fix rep setup

* rep proto

* remove unused files

* add/remove comments

* added docstrings for the controller

* update message definition

* update format, style and comments

* remove situational awareness from participant.json.example

* fix dynamism on node distance

* fix strict_topology

* fix delay error

* fix flooding

* fix processes deployment

* fix delayer attack

---------

Co-authored-by: Alejandro.A.S <jandrosambasil@gmail.com>
Co-authored-by: FerTV <fernando.torres.vega@gmail.com>

Author: 3 people

README.md

@@ -6,7 +6,7 @@
   <h1 align="center">NEBULA: A Platform for Decentralized Federated Learning</h1>
 
   <p align="center">
-    <a href="https://nebula-dfl.com">nebula-dfl.com</a> | 
+    <a href="https://nebula-dfl.com">nebula-dfl.com</a> |
 	<a href="https://nebula-dfl.eu">nebula-dfl.eu</a> |
 	<a href="https://federeratedlearning.inf.um.es">federatedlearning.inf.um.es</a>
   </p>

analysis/README.md

@@ -7,7 +7,7 @@
   <h1 align="center">NEBULA: A Platform for Decentralized Federated Learning</h1>
 
   <p align="center">
-    <a href="https://nebula-dfl.com">nebula-dfl.com</a> | 
+    <a href="https://nebula-dfl.com">nebula-dfl.com</a> |
 	<a href="https://nebula-dfl.eu">nebula-dfl.eu</a> |
 	<a href="https://federeratedlearning.inf.um.es">federatedlearning.inf.um.es</a>
   </p>

app/main.py

@@ -4,8 +4,8 @@
 
 sys.path.append(os.path.join(os.path.dirname(__file__), ".."))  # Parent directory where is the NEBULA module
 import nebula
-from nebula.controller import Controller
-from nebula.scenarios import ScenarioManagement
+from nebula.controller.controller import Controller
+from nebula.controller.scenarios import ScenarioManagement
 
 argparser = argparse.ArgumentParser(description="Controller of NEBULA platform", add_help=False)
 
@@ -54,8 +54,6 @@
     help="Statistics port (default: 8080)",
 )
 
-argparser.add_argument("-t", "--test", dest="test", action="store_true", default=False, help="Run tests")
-
 argparser.add_argument(
     "-st",
     "--stop",

docs/_prebuilt/developerguide.md

@@ -829,8 +829,8 @@ To add a new message to the application, follow these steps:
         FEDERATION_READY = 3;
       }
       Action action = 1;
-      repeated string arguments = 2;  
-      int32 round = 3;                
+      repeated string arguments = 2;
+      int32 round = 3;
     }
     ```
 
@@ -905,4 +905,4 @@ Note that **EventType** is the class that represents the event (not a specific i
 
 When the event is published, all subscribed listeners for that event type will be triggered. As mentioned, there are three different **publish** functions, each tied to a specific type of event.
 
-Finally, to **create a new event**, go to the file **/core/nebulaevents.py**. Depending on the type of event you wish to implement, create a class that extends one of the three native event types. After doing this, the usage of your new event is transparent to the rest of the system, and you can use the functions described above without any issues.
+Finally, to **create a new event**, go to the file **/core/nebulaevents.py**. Depending on the type of event you wish to implement, create a class that extends one of the three native event types. After doing this, the usage of your new event is transparent to the rest of the system, and you can use the functions described above without any issues.

nebula/addons/attacks/communications/communicationattack.py

@@ -1,9 +1,11 @@
 import logging
 import random
+import random
 import types
 from abc import abstractmethod
 
 from nebula.addons.attacks.attacks import Attack
+from nebula.core.network.communications import CommunicationsManager
 
 
 class CommunicationAttack(Attack):
@@ -46,21 +48,23 @@ async def select_targets(self):
             if self.selection_interval:
                 if self.last_selection_round % self.selection_interval == 0:
                     logging.info("Recalculating targets...")
-                    all_nodes = await self.engine.cm.get_addrs_current_connections(only_direct=True)
+                    all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
                     num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
                     self.targets = set(random.sample(list(all_nodes), num_targets))
             elif not self.targets:
                 logging.info("Calculating targets...")
-                all_nodes = await self.engine.cm.get_addrs_current_connections(only_direct=True)
+                all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
                 num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
                 self.targets = set(random.sample(list(all_nodes), num_targets))
         else:
             logging.info("All neighbors selected as targets")
-            self.targets = await self.engine.cm.get_addrs_current_connections(only_direct=True)
+            self.targets = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
 
         logging.info(f"Selected {self.selectivity_percentage}% targets from neighbors: {self.targets}")
         self.last_selection_round += 1
 
+        self.last_selection_round += 1
+
     async def _inject_malicious_behaviour(self):
         """Inject malicious behavior into the target method."""
         decorated_method = self.decorator(self.decorator_args)(self.original_method)

nebula/addons/attacks/communications/delayerattack.py

@@ -3,6 +3,7 @@
 from functools import wraps
 
 from nebula.addons.attacks.communications.communicationattack import CommunicationAttack
+from nebula.core.network.communications import CommunicationsManager
 
 
 class DelayerAttack(CommunicationAttack):
@@ -32,8 +33,8 @@ def __init__(self, engine, attack_params: dict):
 
         super().__init__(
             engine,
-            engine._cm,
-            "send_model",
+            CommunicationsManager.get_instance(),
+            "send_message",
             round_start,
             round_stop,
             attack_interval,
@@ -43,27 +44,27 @@ def __init__(self, engine, attack_params: dict):
         )
 
     def decorator(self, delay: int):
-        """
-        Decorator that adds a delay to the execution of the original method.
+            """
+            Decorator that adds a delay to the execution of the original method.
 
-        Args:
-            delay (int): The time in seconds to delay the method execution.
+            Args:
+                delay (int): The time in seconds to delay the method execution.
 
-        Returns:
-            function: A decorator function that wraps the target method with the delay logic.
-        """
+            Returns:
+                function: A decorator function that wraps the target method with the delay logic.
+            """
 
-        def decorator(func):
-            @wraps(func)
-            async def wrapper(*args, **kwargs):
-                if len(args) > 1:
-                    dest_addr = args[1]
-                    if dest_addr in self.targets:
-                        logging.info(f"[DelayerAttack] Delaying model propagation to {dest_addr} by {delay} seconds")
-                        await asyncio.sleep(delay)
-                _, *new_args = args  # Exclude self argument
-                return await func(*new_args)
+            def decorator(func):
+                @wraps(func)
+                async def wrapper(*args, **kwargs):
+                    if len(args) == 4 and args[3] == "model":
+                        dest_addr = args[1]
+                        if dest_addr in self.targets:
+                            logging.info(f"[DelayerAttack] Delaying model propagation to {dest_addr} by {delay} seconds")
+                            await asyncio.sleep(delay)
+                    _, *new_args = args  # Exclude self argument
+                    return await func(*new_args)
 
-            return wrapper
+                return wrapper
 
-        return decorator
+            return decorator

nebula/addons/attacks/communications/floodingattack.py

@@ -1,9 +1,8 @@
-import asyncio
 import logging
 from functools import wraps
-import time
 
 from nebula.addons.attacks.communications.communicationattack import CommunicationAttack
+from nebula.core.network.communications import CommunicationsManager
 
 
 class FloodingAttack(CommunicationAttack):
@@ -35,8 +34,8 @@ def __init__(self, engine, attack_params: dict):
 
         super().__init__(
             engine,
-            engine._cm,
-            "send_model",
+            CommunicationsManager.get_instance(),
+            "send_message",
             round_start,
             round_stop,
             attack_interval,
@@ -59,7 +58,7 @@ def decorator(self, flooding_factor: int):
         def decorator(func):
             @wraps(func)
             async def wrapper(*args, **kwargs):
-                if len(args) > 1:
+                if len(args) == 4 and args[3] == "model":
                     dest_addr = args[1]
                     if dest_addr in self.targets:
                         logging.info(f"[FloodingAttack] Flooding message to {dest_addr} by {flooding_factor} times")
@@ -68,13 +67,11 @@ async def wrapper(*args, **kwargs):
                                 logging.info(
                                     f"[FloodingAttack] Sending duplicate {i + 1}/{flooding_factor} to {dest_addr}"
                                 )
-                            _, dest_addr, _, serialized_model, weight = args  # Exclude self argument
-                            new_args = [dest_addr, i, serialized_model, weight]
+                            _, *new_args = args  # Exclude self argument
                             await func(*new_args, **kwargs)
-                _, dest_addr, _, serialized_model, weight = args  # Exclude self argument
-                new_args = [dest_addr, i, serialized_model, weight]
+                _, *new_args = args 
                 return await func(*new_args)
-
+            
             return wrapper
 
         return decorator

nebula/addons/attacks/dataset/datasetattack.py

@@ -35,9 +35,11 @@ async def attack(self):
         """
         if self.engine.round not in range(self.round_start_attack, self.round_stop_attack + 1):
             pass
-        elif  self.engine.round == self.round_stop_attack:
+        elif self.engine.round == self.round_stop_attack:
             logging.info(f"[{self.__class__.__name__}] Stopping attack")
-        elif self.engine.round >= self.round_start_attack and ((self.engine.round - self.round_start_attack) % self.attack_interval == 0):
+        elif self.engine.round >= self.round_start_attack and (
+            (self.engine.round - self.round_start_attack) % self.attack_interval == 0
+        ):
             logging.info(f"[{self.__class__.__name__}] Performing attack")
             self.engine.trainer.datamodule.train_set = self.get_malicious_dataset()
 

nebula/addons/attacks/dataset/labelflipping.py

@@ -10,6 +10,7 @@
 import copy
 import logging
 import random
+
 import numpy as np
 
 from nebula.addons.attacks.dataset.datasetattack import DatasetAttack

nebula/addons/attacks/model/gllneuroninversion.py

@@ -34,7 +34,7 @@ def __init__(self, engine, attack_params):
             raise ValueError(f"Missing required attack parameter: {e}")
         except ValueError:
             raise ValueError("Invalid value in attack_params. Ensure all values are integers.")
-        
+
         super().__init__(engine, round_start, round_stop, attack_interval)
 
     def model_attack(self, received_weights):