refactor(modules): apply comprehensive coding standards to all modules

fusion/core/TODO.md

@@ -76,6 +76,17 @@ This file tracks known issues and future improvements for the FUSION core module
   2. Add comprehensive testing for best-fit allocation
   3. Validate allocation algorithms across different scenarios
 
+### Cross-talk Aware Allocation Limitation
+- **Issue**: Cross-talk aware allocation only works for 7-core configurations
+- **File**: `fusion/core/spectrum_assignment.py:216`
+- **Description**: The handle_crosstalk_aware_allocation method has hardcoded limitations for seven-core fiber configurations
+- **Impact**: Limited support for different core configurations (4, 13, 19 cores)
+- **Next Steps**:
+  1. Analyze code dependencies on seven-core assumption
+  2. Generalize method to support all core configurations  
+  3. Test with different core counts
+  4. Update algorithm to dynamically determine core adjacency patterns
+
 ### SNR Measurement System Updates
 - **Issue**: SNR measurement system needs variable renaming and feature additions
 - **Files**: `fusion/core/snr_measurements.py:448`, `fusion/core/snr_measurements.py:451`, `fusion/core/snr_measurements.py:454`

fusion/core/sdn_controller.py

@@ -4,7 +4,7 @@
 import numpy as np
 
 # Removed unused imports: sort_dict_keys, get_path_mod, find_path_len
-from fusion.modules.ml.train_utils import get_ml_obs
+from fusion.modules.ml import get_ml_obs
 from fusion.core.properties import SDNProps
 from fusion.core.routing import Routing
 from fusion.core.spectrum_assignment import SpectrumAssignment
@@ -238,8 +238,8 @@ def _get_ml_prediction(self, ml_model: Optional[Any], request_dict: Dict[str, An
         :rtype: float
         """
         if ml_model is not None:
-            input_df = get_ml_obs(req_dict=request_dict, engine_props=self.engine_props,
-                                  sdn_props=self.sdn_props)
+            input_df = get_ml_obs(request_dict=request_dict, engine_properties=self.engine_props,
+                                  sdn_properties=self.sdn_props)
             return ml_model.predict(input_df)[0]
         return -1.0
 

fusion/core/simulation.py

@@ -14,7 +14,7 @@
 from fusion.core.metrics import SimStats
 from fusion.core.persistence import StatsPersistence
 from fusion.core.ml_metrics import MLMetricsCollector
-from fusion.modules.ml.train_utils import load_model
+from fusion.modules.ml import load_model
 from fusion.sim.utils import log_message
 from fusion.reporting.simulation_reporter import SimulationReporter
 from fusion.utils.logging_config import get_logger
@@ -277,7 +277,7 @@ def init_iter(self, iteration: int, seed: int = None, print_flag: bool = True, t
                         self.engine_props['erlang'], self.engine_props['thread_num'])
 
             if self.engine_props['deploy_model']:
-                self.ml_model = load_model(engine_props=self.engine_props)
+                self.ml_model = load_model(engine_properties=self.engine_props)
 
         # You can pass a list of seeds, a constant seed, or default to iteration number
         if seed is None:

fusion/interfaces/factory.py

@@ -2,7 +2,8 @@
 Factory classes for creating algorithm instances using interfaces.
 """
 
-from typing import Dict, Any
+from typing import Any, Dict
+
 from fusion.interfaces.router import AbstractRoutingAlgorithm
 from fusion.interfaces.spectrum import AbstractSpectrumAssigner
 from fusion.interfaces.snr import AbstractSNRMeasurer
@@ -262,7 +263,7 @@ def _process_snr_and_allocation(self, result: Dict[str, Any], path: Any,
 
         # Get modulation format for SNR threshold
         modulation = request.modulation if hasattr(request, 'modulation') else 'QPSK'
-        topology = self.engine_props.get('topology', self.sdn_props.topology)
+        topology: Any = self.engine_props.get('topology', self.sdn_props.topology)
         path_length = sum(topology[path[i]][path[i + 1]]['length']
                           for i in range(len(path) - 1))
 

fusion/interfaces/router.py

@@ -4,7 +4,7 @@
 # pylint: disable=duplicate-code
 
 from abc import ABC, abstractmethod
-from typing import List, Dict, Any, Optional
+from typing import Any, Dict, List, Optional
 
 
 class AbstractRoutingAlgorithm(ABC):

fusion/interfaces/snr.py

@@ -4,7 +4,7 @@
 # pylint: disable=duplicate-code
 
 from abc import ABC, abstractmethod
-from typing import List, Dict, Any
+from typing import Any, Dict, List
 
 
 class AbstractSNRMeasurer(ABC):

fusion/interfaces/spectrum.py

@@ -4,7 +4,7 @@
 # pylint: disable=duplicate-code
 
 from abc import ABC, abstractmethod
-from typing import List, Dict, Any, Optional
+from typing import Any, Dict, List, Optional
 
 
 class AbstractSpectrumAssigner(ABC):

fusion/io/exporter.py

@@ -5,11 +5,11 @@
 and export destinations.
 """
 
-from pathlib import Path
-from typing import Dict, Any, Union, List
-import json
 import csv
+import json
 from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Any, Dict, List, Union
 
 
 class BaseExporter(ABC):

fusion/io/generate.py

@@ -1,17 +1,18 @@
-from pathlib import Path
-
-import math
 import json
-
-
-def create_pt(cores_per_link: int, network_spectrum_dict: dict):
-    """
-    Generates information relevant to the physical topology of the network.
-
-    :param cores_per_link: The number of cores in each fiber's link.
-    :param network_spectrum_dict: The network spectrum database.
-    :return: Physical layer information topology of the network.
-    :rtype: dict
+import math
+from pathlib import Path
+from typing import Dict, Optional
+
+
+def create_pt(cores_per_link: int, network_spectrum_dict: Dict[tuple, float]) -> Dict[str, Dict]:
+    """Generate information relevant to the physical topology of the network.
+
+    :param cores_per_link: The number of cores in each fiber's link
+    :type cores_per_link: int
+    :param network_spectrum_dict: The network spectrum database mapping node pairs to lengths
+    :type network_spectrum_dict: Dict[tuple, float]
+    :return: Physical layer information topology of the network
+    :rtype: Dict[str, Dict]
     """
     fiber_props_dict = {
         'attenuation': 0.2 / 4.343 * 1e-3,
@@ -48,14 +49,17 @@ def create_pt(cores_per_link: int, network_spectrum_dict: dict):
     return topology_dict
 
 
-def create_bw_info(mod_assumption: str, mod_assumptions_path: str = None) -> dict:
-    """
-    Determines reach and slots needed for each bandwidth and modulation format.
-
-    :param mod_assumption: Controls which assumptions to be used.
-    :param mod_assumptions_path: Path to modulation assumptions file.
-    :return: The number of spectral slots needed for each bandwidth and modulation format pair.
-    :rtype: dict
+def create_bw_info(mod_assumption: str, mod_assumptions_path: Optional[str] = None) -> Dict[str, Dict]:
+    """Determine reach and slots needed for each bandwidth and modulation format.
+
+    :param mod_assumption: Controls which assumptions to be used
+    :type mod_assumption: str
+    :param mod_assumptions_path: Path to modulation assumptions file
+    :type mod_assumptions_path: Optional[str]
+    :return: The number of spectral slots needed for each bandwidth and modulation format pair
+    :rtype: Dict[str, Dict]
+    :raises FileNotFoundError: If modulation assumptions file is not found
+    :raises NotImplementedError: If unknown modulation assumption is provided
     """
     # Set default path if none provided
     if not mod_assumptions_path or mod_assumptions_path == "None":
@@ -69,15 +73,15 @@ def create_bw_info(mod_assumption: str, mod_assumptions_path: str = None) -> dic
         mod_assumptions_path = project_root / mod_assumptions_path
 
     try:
-        with mod_assumptions_path.open("r", encoding="utf-8") as mod_fp:
-            mod_formats_obj = json.load(mod_fp)
+        with mod_assumptions_path.open("r", encoding="utf-8") as modulation_file:
+            modulation_formats_dict = json.load(modulation_file)
 
-        if mod_assumption in mod_formats_obj:
-            return mod_formats_obj[mod_assumption]
+        if mod_assumption in modulation_formats_dict:
+            return modulation_formats_dict[mod_assumption]
 
-    except json.JSONDecodeError as json_err:
-        raise FileExistsError(f"[ERROR] Could not parse JSON: {json_err.doc}") from json_err
-    except FileNotFoundError as file_err:
-        raise FileNotFoundError(f"[ERROR] File not found: {mod_assumptions_path}") from file_err
+    except json.JSONDecodeError as json_error:
+        raise FileExistsError(f"Could not parse JSON: {json_error.doc}") from json_error
+    except FileNotFoundError as file_error:
+        raise FileNotFoundError(f"File not found: {mod_assumptions_path}") from file_error
 
-    raise NotImplementedError(f"[ERROR] Unknown modulation assumption '{mod_assumption}'")
+    raise NotImplementedError(f"Unknown modulation assumption '{mod_assumption}'")

fusion/io/structure.py

@@ -1,11 +1,20 @@
 from pathlib import Path
-
-
-def assign_link_lengths(network_fp: Path, node_pairs_dict: dict, constant_weight: bool = False):
-    """
-    Assign length to each link in a given topology.
+from typing import Dict, List, Tuple
+
+
+def assign_link_lengths(network_fp: Path, node_pairs_dict: Dict[str, str], constant_weight: bool = False) -> Dict[Tuple[str, str], float]:
+    """Assign length to each link in a given topology.
+
+    :param network_fp: Path to the network topology file
+    :type network_fp: Path
+    :param node_pairs_dict: Dictionary mapping node names to alternative names
+    :type node_pairs_dict: Dict[str, str]
+    :param constant_weight: If True, assign constant weight of 1.0 to all links
+    :type constant_weight: bool
+    :return: Dictionary mapping node pairs to link lengths
+    :rtype: Dict[Tuple[str, str], float]
     """
-    response_dict = {}
+    link_lengths_dict = {}
     with network_fp.open('r', encoding='utf-8') as file_obj:
         for line in file_obj:
             src, dest, link_len_str = line.strip().split('\t')
@@ -16,27 +25,43 @@ def assign_link_lengths(network_fp: Path, node_pairs_dict: dict, constant_weight
                 node_pairs_dict.get(dest, dest)
             )
 
-            if src_dest_tuple not in response_dict:
-                response_dict[src_dest_tuple] = link_len
+            if src_dest_tuple not in link_lengths_dict:
+                link_lengths_dict[src_dest_tuple] = link_len
 
-    return response_dict
+    return link_lengths_dict
 
 
-def assign_core_nodes(core_nodes_fp: Path):
+def assign_core_nodes(core_nodes_fp: Path) -> List[str]:
+    """Determine which nodes are core nodes in the network.
+
+    :param core_nodes_fp: Path to the core nodes file
+    :type core_nodes_fp: Path
+    :return: List of core node identifiers
+    :rtype: List[str]
     """
-    Determine which nodes are core nodes in the network.
-    """
-    response_list = []
+    core_nodes_list = []
     with core_nodes_fp.open('r', encoding='utf-8') as file_obj:
         for line in file_obj:
-            response_list.append(line.strip().split('\t')[0])
-    return response_list
+            core_nodes_list.append(line.strip().split('\t')[0])
+    return core_nodes_list
 
 
-def create_network(net_name: str, base_fp: str = None, const_weight: bool = False, is_only_core_node: bool = False):
-    """
-    Builds a physical network structure from topology files.
+def create_network(net_name: str, base_fp: str = None, const_weight: bool = False, is_only_core_node: bool = False) -> Tuple[Dict[Tuple[str, str], float], List[str]]:
+    """Build a physical network structure from topology files.
+
     Resolves all paths safely using project-root-relative logic.
+
+    :param net_name: Name of the network topology to load
+    :type net_name: str
+    :param base_fp: Base file path for topology files
+    :type base_fp: str
+    :param const_weight: If True, assign constant weight to all links
+    :type const_weight: bool
+    :param is_only_core_node: If True, only load core nodes
+    :type is_only_core_node: bool
+    :return: Tuple of (network dictionary, core nodes list)
+    :rtype: Tuple[Dict[Tuple[str, str], float], List[str]]
+    :raises NotImplementedError: If unknown network name is provided
     """
     core_nodes_list = []