[Feature] Include realistic mobility and frontend-controller adaptation

nebula/addons/attacks/communications/communicationattack.py

@@ -1,12 +1,26 @@
 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):
+    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,
@@ -46,21 +60,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 = 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,7 +33,7 @@ def __init__(self, engine, attack_params: dict):
 
         super().__init__(
             engine,
-            engine._cm,
+            CommunicationsManager.get_instance(),
             "send_model",
             round_start,
             round_stop,

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,

nebula/addons/mobility.py

@@ -3,19 +3,16 @@
 import math
 import random
 import time
-from typing import TYPE_CHECKING
-
-from nebula.addons.functions import print_msg_box
+from functools import cached_property
+from nebula.core.network.communications import CommunicationsManager
 from nebula.core.eventmanager import EventManager
-from nebula.core.nebulaevents import GPSEvent
+from nebula.core.nebulaevents import GPSEvent, ChangeLocationEvent
 from nebula.core.utils.locker import Locker
-
-if TYPE_CHECKING:
-    from nebula.core.network.communications import CommunicationsManager
+from nebula.addons.functions import print_msg_box
 
 
 class Mobility:
-    def __init__(self, config, cm: "CommunicationsManager", verbose=False):
+    def __init__(self, config, verbose=False):
         """
         Initializes the mobility module with specified configuration and communication manager.
 
@@ -52,7 +49,6 @@
         """
         logging.info("Starting mobility module...")
         self.config = config
-        self.cm = cm
         self.grace_time = self.config.participant["mobility_args"]["grace_time_mobility"]
         self.period = self.config.participant["mobility_args"]["change_geo_interval"]
         self.mobility = self.config.participant["mobility_args"]["mobility"]
@@ -61,9 +57,9 @@
         self.scheme_mobility = self.config.participant["mobility_args"]["scheme_mobility"]
         self.round_frequency = int(self.config.participant["mobility_args"]["round_frequency"])
         # Protocol to change connections based on distance
-        self.max_distance_with_direct_connections = 300  # meters
-        self.max_movement_random_strategy = 100  # meters
-        self.max_movement_nearest_strategy = 100  # meters
+        self.max_distance_with_direct_connections = 150  # meters
+        self.max_movement_random_strategy = 50  # meters
+        self.max_movement_nearest_strategy = 50  # meters
         self.max_initiate_approximation = self.max_distance_with_direct_connections * 1.2
         # Logging box with mobility information
         mobility_msg = f"Mobility: {self.mobility}\nMobility type: {self.mobility_type}\nRadius federation: {self.radius_federation}\nScheme mobility: {self.scheme_mobility}\nEach {self.round_frequency} rounds"
@@ -72,6 +68,10 @@
         self._nodes_distances_lock = Locker("nodes_distances_lock", async_lock=True)
         self._verbose = verbose
 
+    @cached_property
+    def cm(self):
+        return CommunicationsManager.get_instance()
+
     @property
     def round(self):
         """
@@ -101,10 +101,11 @@
                            `run_mobility` operation.
         """
         await EventManager.get_instance().subscribe_addonevent(GPSEvent, self.update_nodes_distances)
+        await EventManager.get_instance().subscribe_addonevent(GPSEvent, self.update_nodes_distances)
         task = asyncio.create_task(self.run_mobility())
         return task
 
-    async def update_nodes_distances(self, gpsevent: GPSEvent):
+    async def update_nodes_distances(self, gpsevent : GPSEvent):
         distances = await gpsevent.get_event_data()
         async with self._nodes_distances_lock:
             self._nodes_distances = dict(distances)
@@ -135,7 +136,7 @@
         """
         if not self.mobility:
             return
-        # await asyncio.sleep(self.grace_time)
+        #await asyncio.sleep(self.grace_time)
         while True:
             await self.change_geo_location()
             await asyncio.sleep(self.period)
@@ -161,8 +162,7 @@
             - The calculated radius is converted from meters to degrees based on an approximate
               conversion factor (1 degree is approximately 111 kilometers).
         """
-        if self._verbose:
-            logging.info("📍  Changing geo location randomly")
+        if self._verbose: logging.info("📍  Changing geo location randomly")
         # radius_in_degrees = self.radius_federation / 111000
         max_radius_in_degrees = self.max_movement_random_strategy / 111000
         radius = random.uniform(0, max_radius_in_degrees)  # noqa: S311
@@ -198,7 +198,7 @@
               coordinates to determine the direction of movement.
             - The conversion from meters to degrees is based on approximate geographical conversion factors.
         """
-        logging.info("📍  Changing geo location towards the nearest neighbor")
+        if self._verbose: logging.info("📍  Changing geo location towards the nearest neighbor")
         scale_factor = min(1, self.max_movement_nearest_strategy / distance)
         # Calcular el ángulo hacia el vecino
         angle = math.atan2(neighbor_longitude - longitude, neighbor_latitude - latitude)
@@ -243,8 +243,9 @@
 
         self.config.participant["mobility_args"]["latitude"] = latitude
         self.config.participant["mobility_args"]["longitude"] = longitude
-        if self._verbose:
-            logging.info(f"📍  New geo location: {latitude}, {longitude}")
+        if self._verbose: logging.info(f"📍  New geo location: {latitude}, {longitude}")
+        cle = ChangeLocationEvent(latitude, longitude)
+        asyncio.create_task(EventManager.get_instance().publish_addonevent(cle))
 
     async def change_geo_location(self):
         """
@@ -281,11 +282,11 @@
 
                 selected_neighbor = result[0] if result else None
                 if selected_neighbor:
-                    # logging.info(f"📍  Selected neighbor: {selected_neighbor}")
+                    #logging.info(f"📍  Selected neighbor: {selected_neighbor}")
                     addr, dist, (lat, long) = selected_neighbor
                     if dist > self.max_initiate_approximation:
                         # If the distance is too big, we move towards the neighbor
-                        logging.info(f"Moving towards nearest neighbor: {addr}")
+                        if self._verbose: logging.info(f"Moving towards nearest neighbor: {addr}")
                         await self.change_geo_location_nearest_neighbor_strategy(
                             dist,
                             latitude,

nebula/addons/networksimulation/nebulanetworksimulator.py

@@ -1,16 +1,14 @@
 import asyncio
 import logging
 import subprocess
-from typing import TYPE_CHECKING
+from functools import cached_property
 
 from nebula.addons.networksimulation.networksimulator import NetworkSimulator
 from nebula.core.eventmanager import EventManager
 from nebula.core.nebulaevents import GPSEvent
+from nebula.core.network.communications import CommunicationsManager
 from nebula.core.utils.locker import Locker
 
-if TYPE_CHECKING:
-    from nebula.core.network.communications import CommunicationsManager
-
 
 class NebulaNS(NetworkSimulator):
     NETWORK_CONDITIONS = {
@@ -21,8 +19,7 @@ class NebulaNS(NetworkSimulator):
     }
     IP_MULTICAST = "239.255.255.250"
 
-    def __init__(self, communication_manager: "CommunicationsManager", changing_interval, interface, verbose=False):
-        self._cm = communication_manager
+    def __init__(self, changing_interval, interface, verbose=False):
         self._refresh_interval = changing_interval
         self._node_interface = interface
         self._verbose = verbose
@@ -31,9 +28,16 @@ def __init__(self, communication_manager: "CommunicationsManager", changing_inte
         self._current_network_conditions = {}
         self._running = False
 
+    @cached_property
+    def cm(self):
+        return CommunicationsManager.get_instance()
+
     async def start(self):
         logging.info("🌐  Nebula Network Simulator starting...")
         self._running = True
+        grace_time = self.cm.config.participant["mobility_args"]["grace_time_mobility"]
+        # if self._verbose: logging.info(f"Waiting {grace_time}s to start applying network conditions based on distances between devices")
+        # await asyncio.sleep(grace_time)
         await EventManager.get_instance().subscribe_addonevent(
             GPSEvent, self._change_network_conditions_based_on_distances
         )
@@ -213,7 +217,7 @@ def extract_number(value):
 
             match = re.match(r"([\d.]+)", value)
             if not match:
-                raise ValueError(f"Invalid format: {value}")
+                raise ValueError(f"Formato inválido: {value}")
             return float(match.group(1))
 
         if self._verbose:
@@ -224,11 +228,11 @@ def extract_number(value):
 
         thresholds = sorted(th.keys())
 
-        # If the distance is less than the first threshold, return the best condition
+        # Si la distancia es menor que el primer umbral, devolver la mejor condición
         if distance < thresholds[0]:
             conditions = {"bandwidth": th[thresholds[0]]["bandwidth"], "delay": th[thresholds[0]]["delay"]}
 
-        # Find the section in which the distance is located.
+        # Encontrar el tramo en el que se encuentra la distancia
         for i in range(len(thresholds) - 1):
             lower_bound = thresholds[i]
             upper_bound = thresholds[i + 1]
@@ -241,7 +245,7 @@ def extract_number(value):
                 lower_cond = th[lower_bound]
                 upper_cond = th[upper_bound]
 
-                # Extract numerical values and units
+                # Extraer valores numéricos y unidades
                 lower_bandwidth_value = extract_number(lower_cond["bandwidth"])
                 upper_bandwidth_value = extract_number(upper_cond["bandwidth"])
                 lower_bandwidth_unit = lower_cond["bandwidth"].replace(str(lower_bandwidth_value), "")
@@ -251,22 +255,22 @@ def extract_number(value):
                 upper_delay_value = extract_number(upper_cond["delay"])
                 delay_unit = lower_cond["delay"].replace(str(lower_delay_value), "")
 
-                # Calculate progress in the leg (0 to 1)
+                # Calcular el progreso en el tramo (0 a 1)
                 progress = (distance - lower_bound) / (upper_bound - lower_bound)
                 if self._verbose:
                     logging.info(f"Progress between the bounds: {progress}")
 
-                # Linear interpolation of values
+                # Interpolación lineal de valores
                 bandwidth_value = lower_bandwidth_value - progress * (lower_bandwidth_value - upper_bandwidth_value)
                 delay_value = lower_delay_value + progress * (upper_delay_value - lower_delay_value)
 
-                # Reconstruct values with original units
+                # Reconstruir valores con unidades originales
                 bandwidth = f"{round(bandwidth_value, 2)}{lower_bandwidth_unit}"
                 delay = f"{round(delay_value, 2)}{delay_unit}"
 
                 conditions = {"bandwidth": bandwidth, "delay": delay}
 
-        # If the distance is infinite, return the last value
+        # Si la distancia es infinita, devolver el último valor
         if not conditions:
             conditions = {"bandwidth": th[float("inf")]["bandwidth"], "delay": th[float("inf")]["delay"]}
         if self._verbose:

nebula/addons/networksimulation/networksimulator.py

@@ -27,9 +27,7 @@ class NetworkSimulatorException(Exception):
     pass
 
 
-def factory_network_simulator(
-    net_sim, communication_manager, changing_interval, interface, verbose
-) -> NetworkSimulator:
+def factory_network_simulator(net_sim, changing_interval, interface, verbose) -> NetworkSimulator:
     from nebula.addons.networksimulation.nebulanetworksimulator import NebulaNS
 
     SIMULATION_SERVICES = {
@@ -39,6 +37,6 @@ def factory_network_simulator(
     net_serv = SIMULATION_SERVICES.get(net_sim, NebulaNS)
 
     if net_serv:
-        return net_serv(communication_manager, changing_interval, interface, verbose)
+        return net_serv(changing_interval, interface, verbose)
     else:
         raise NetworkSimulatorException(f"Network Simulator {net_sim} not found")

nebula/addons/reporter.py

@@ -10,11 +10,11 @@
 import psutil
 
 if TYPE_CHECKING:
-    from nebula.core.network.communications import CommunicationsManager
+    pass
 
 
 class Reporter:
-    def __init__(self, config, trainer, cm: "CommunicationsManager"):
+    def __init__(self, config, trainer):
         """
         Initializes the reporter module for sending periodic updates to a dashboard controller.
 
@@ -48,9 +48,9 @@ def __init__(self, config, trainer, cm: "CommunicationsManager"):
             - Initializes both current and accumulated metrics for traffic monitoring.
         """
         logging.info("Starting reporter module")
+        self._cm = None
         self.config = config
         self.trainer = trainer
-        self.cm = cm
         self.frequency = self.config.participant["reporter_args"]["report_frequency"]
         self.grace_time = self.config.participant["reporter_args"]["grace_time_reporter"]
         self.data_queue = asyncio.Queue()
@@ -68,6 +68,16 @@ def __init__(self, config, trainer, cm: "CommunicationsManager"):
         self.acc_packets_sent = 0
         self.acc_packets_recv = 0
 
+    @property
+    def cm(self):
+        if not self._cm:
+            from nebula.core.network.communications import CommunicationsManager
+
+            self._cm = CommunicationsManager.get_instance()
+            return self._cm
+        else:
+            return self._cm
+
     async def enqueue_data(self, name, value):
         """
         Asynchronously enqueues data for reporting.