[Feature/events] Improve event system (#36)

* 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 event system

* change location of topology image

* remove upper table in dashboard

* improve readability, explanations, and fix some issues

* Rename GPS directory

---------

Co-authored-by: enriquetomasmb <[email protected]>
Co-authored-by: Enrique Tomás <[email protected]>

Author: AlejandroAvilesSerrano

nebula/addons/attacks/communications/communicationattack.py

@@ -1,22 +1,24 @@
 import logging
+import random
 import types
 from abc import abstractmethod
-import random
 
 from nebula.addons.attacks.attacks import Attack
 
 
 class CommunicationAttack(Attack):
-    def __init__(self, engine,
-                 target_class,
-                 target_method,
-                 round_start_attack,
-                 round_stop_attack,
-                 attack_interval,
-                 decorator_args=None,
-                 selectivity_percentage: int = 100,
-                 selection_interval: int = None
-                 ):
+    def __init__(
+        self,
+        engine,
+        target_class,
+        target_method,
+        round_start_attack,
+        round_stop_attack,
+        attack_interval,
+        decorator_args=None,
+        selectivity_percentage: int = 100,
+        selection_interval: int = None,
+    ):
         super().__init__()
         self.engine = engine
         self.target_class = target_class
@@ -41,7 +43,7 @@ def decorator(self, *args):
 
     async def select_targets(self):
         if self.selectivity_percentage != 100:
-            if self.selection_interval:      
+            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)
@@ -55,10 +57,10 @@ async def select_targets(self):
         else:
             logging.info("All neighbors selected as targets")
             self.targets = await self.engine.cm.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)
@@ -80,7 +82,9 @@ async def attack(self):
         elif self.engine.round == self.round_stop_attack:
             logging.info(f"[{self.__class__.__name__}] Stoping attack")
             await self._restore_original_behaviour()
-        elif (self.engine.round == self.round_start_attack) or ((self.engine.round - self.round_start_attack) % self.attack_interval == 0):
+        elif (self.engine.round == self.round_start_attack) or (
+            (self.engine.round - self.round_start_attack) % self.attack_interval == 0
+        ):
             await self.select_targets()
             logging.info(f"[{self.__class__.__name__}] Performing attack")
             await self._inject_malicious_behaviour()

nebula/addons/attacks/communications/delayerattack.py

@@ -56,9 +56,9 @@ def decorator(self, delay: int):
         def decorator(func):
             @wraps(func)
             async def wrapper(*args, **kwargs):
-                if len(args) > 1:  
+                if len(args) > 1:
                     dest_addr = args[1]
-                    if dest_addr in self.targets:  
+                    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

nebula/addons/attacks/communications/floodingattack.py

@@ -1,18 +1,18 @@
 import logging
 from functools import wraps
- 
+
 from nebula.addons.attacks.communications.communicationattack import CommunicationAttack
- 
- 
+
+
 class FloodingAttack(CommunicationAttack):
     """
     Implements an attack that delays the execution of a target method by a specified amount of time.
     """
- 
+
     def __init__(self, engine, attack_params: dict):
         """
         Initializes the DelayerAttack with the engine and attack parameters.
- 
+
         Args:
             engine: The engine managing the attack context.
             attack_params (dict): Parameters for the attack, including the delay duration.
@@ -28,9 +28,9 @@ def __init__(self, engine, attack_params: dict):
             raise ValueError(f"Missing required attack parameter: {e}")
         except ValueError:
             raise ValueError("Invalid value in attack_params. Ensure all values are integers.")
-        
+
         self.verbose = False
- 
+
         super().__init__(
             engine,
             engine._cm,
@@ -42,33 +42,35 @@ def __init__(self, engine, attack_params: dict):
             self.target_percentage,
             self.selection_interval,
         )
- 
+
     def decorator(self, flooding_factor: int):
         """
         Decorator that adds a delay to the execution of the original method.
- 
+
         Args:
             flooding_factor (int): The number of times to repeat the function execution.
- 
+
         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:  
+                if len(args) > 1:
                     dest_addr = args[1]
-                    if dest_addr in self.targets:  
+                    if dest_addr in self.targets:
                         logging.info(f"[FloodingAttack] Flooding message to {dest_addr} by {flooding_factor} times")
                         for i in range(flooding_factor):
                             if self.verbose:
-                                logging.info(f"[FloodingAttack] Sending duplicate {i+1}/{flooding_factor} to {dest_addr}")
+                                logging.info(
+                                    f"[FloodingAttack] Sending duplicate {i + 1}/{flooding_factor} to {dest_addr}"
+                                )
                             _, *new_args = args  # Exclude self argument
                             await func(*new_args, **kwargs)
                 _, *new_args = args  # Exclude self argument
                 return await func(*new_args)
- 
+
             return wrapper
- 
-        return decorator
+
+        return decorator

nebula/addons/gps/gpsmodule.py

@@ -0,0 +1,38 @@
+from abc import ABC, abstractmethod
+
+
+class GPSModule(ABC):
+    @abstractmethod
+    async def start(self):
+        pass
+
+    @abstractmethod
+    async def stop(self):
+        pass
+
+    @abstractmethod
+    async def is_running(self):
+        pass
+
+    @abstractmethod
+    async def calculate_distance(self, self_lat, self_long, other_lat, other_long):
+        pass
+
+
+class GPSModuleException(Exception):
+    pass
+
+
+def factory_gpsmodule(gps_module, config, addr, update_interval: float = 5.0, verbose=False) -> GPSModule:
+    from nebula.addons.gps.nebulagps import NebulaGPS
+
+    GPS_SERVICES = {
+        "nebula": NebulaGPS,
+    }
+
+    gps_module = GPS_SERVICES.get(gps_module, NebulaGPS)
+
+    if gps_module:
+        return gps_module(config, addr, update_interval, verbose)
+    else:
+        raise GPSModuleException(f"GPS Module {gps_module} not found")

nebula/addons/gps/nebulagps.py

@@ -0,0 +1,106 @@
+import asyncio
+import logging
+import socket
+
+from geopy import distance
+
+from nebula.addons.gps.gpsmodule import GPSModule
+from nebula.core.eventmanager import EventManager
+from nebula.core.nebulaevents import GPSEvent
+from nebula.core.utils.locker import Locker
+
+
+class NebulaGPS(GPSModule):
+    BROADCAST_IP = "255.255.255.255"  # Broadcast IP
+    BROADCAST_PORT = 50001  # Port used for GPS
+    INTERFACE = "eth2"  # Interface to avoid network conditions
+
+    def __init__(self, config, addr, update_interval: float = 5.0, verbose=False):
+        self._config = config
+        self._addr = addr
+        self.update_interval = update_interval  # Frequency
+        self.running = False
+        self._node_locations = {}  # Dictionary for storing node locations
+        self._broadcast_socket = None
+        self._nodes_location_lock = Locker("nodes_location_lock", async_lock=True)
+        self._verbose = verbose
+
+    async def start(self):
+        """Inicia el servicio de GPS, enviando y recibiendo ubicaciones."""
+        logging.info("Starting NebulaGPS service...")
+        self.running = True
+
+        # Create broadcast socket
+        self._broadcast_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self._broadcast_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+
+        # Bind socket on eth2 to also receive data
+        self._broadcast_socket.bind(("", self.BROADCAST_PORT))
+
+        # Start sending and receiving tasks
+        asyncio.create_task(self._send_location_loop())
+        asyncio.create_task(self._receive_location_loop())
+        asyncio.create_task(self._notify_geolocs())
+
+    async def stop(self):
+        """Stops the GPS service."""
+        logging.info("Stopping NebulaGPS service...")
+        self.running = False
+        if self._broadcast_socket:
+            self._broadcast_socket.close()
+            self._broadcast_socket = None
+
+    async def is_running(self):
+        return self.running
+
+    async def get_geoloc(self):
+        latitude = self._config.participant["mobility_args"]["latitude"]
+        longitude = self._config.participant["mobility_args"]["longitude"]
+        return (latitude, longitude)
+
+    async def calculate_distance(self, self_lat, self_long, other_lat, other_long):
+        distance_m = distance.distance((self_lat, self_long), (other_lat, other_long)).m
+        return distance_m
+
+    async def _send_location_loop(self):
+        """Send the geolocation periodically by broadcast."""
+        while self.running:
+            latitude, longitude = await self.get_geoloc()  # Obtener ubicación actual
+            message = f"GPS-UPDATE {self._addr} {latitude} {longitude}"
+            self._broadcast_socket.sendto(message.encode(), (self.BROADCAST_IP, self.BROADCAST_PORT))
+            if self._verbose:
+                logging.info(f"Sent GPS location: ({latitude}, {longitude})")
+            await asyncio.sleep(self.update_interval)
+
+    async def _receive_location_loop(self):
+        """Escucha y almacena geolocalizaciones de otros nodos."""
+        while self.running:
+            try:
+                data, addr = await asyncio.get_running_loop().run_in_executor(
+                    None, self._broadcast_socket.recvfrom, 1024
+                )
+                message = data.decode().strip()
+                if message.startswith("GPS-UPDATE"):
+                    _, sender_addr, lat, lon = message.split()
+                    if sender_addr != self._addr:
+                        async with self._nodes_location_lock:
+                            self._node_locations[sender_addr] = (float(lat), float(lon))
+                    if self._verbose:
+                        logging.info(f"Received GPS from {addr[0]}: {lat}, {lon}")
+            except Exception as e:
+                logging.error(f"Error receiving GPS update: {e}")
+
+    async def _notify_geolocs(self):
+        while True:
+            await asyncio.sleep(self.update_interval)
+            await self._nodes_location_lock.acquire_async()
+            geolocs: dict = self._node_locations.copy()
+            await self._nodes_location_lock.release_async()
+            if geolocs:
+                distances = {}
+                self_lat, self_long = await self.get_geoloc()
+                for addr, (lat, long) in geolocs.items():
+                    dist = await self.calculate_distance(self_lat, self_long, lat, long)
+                    distances[addr] = (dist, (lat, long))
+                gpsevent = GPSEvent(distances)
+                asyncio.create_task(EventManager.get_instance().publish_addonevent(gpsevent))