Utils Module

Overview

At a Glance

Purpose:

Common utility functions used throughout FUSION

Location:

fusion/utils/

Key Files:

config.py, logging_config.py, spectrum.py, network.py, random.py, os.py, data.py

Depends On:

fusion.configs.constants, fusion.configs.errors, numpy

Used By:

Nearly all FUSION modules

The utils module provides shared utility functions organized by domain: configuration handling, logging, spectrum management, network analysis, random number generation, file operations, and data manipulation.

Important

To avoid circular dependencies, import directly from specific modules rather than from the package:

# Preferred
from fusion.utils.logging_config import get_logger
from fusion.utils.spectrum import find_free_slots

# Avoid (limited exports)
from fusion.utils import get_logger

Module Summary

Utils Module Files

File	Purpose	Functions	Category
config.py	Configuration type conversion and CLI override handling	5	Configuration
logging_config.py	Centralized logging setup with rotation and formatting	8 + 1 class	Logging
os.py	Directory creation and project root discovery	2	File I/O
random.py	Reproducible random number generation for simulations	3	RNG
spectrum.py	Spectrum slot/channel finding and overlap detection	8	Spectrum
network.py	Path analysis, congestion, fragmentation, modulation	6	Network
data.py	Dictionary sorting and manipulation	2	Data

Helper Function Reference

This section provides a quick reference for all helper functions in each file. Use this to find the right function for your task.

config.py - Configuration Management

Purpose:

Type conversion, dictionary parsing, CLI argument handling

Lines:

~126

config.py Functions

Function	Signature	Description
str_to_bool	(string: str) -> bool	Convert “true”, “yes”, “1” to boolean
convert_string_to_dict	(value: str) -> str | dict	Parse string dict representation using ast.literal_eval
apply_cli_override	(config_val, cli_val, converter) -> Any	Apply CLI argument override with type conversion
safe_type_convert	(value: str, converter, option: str) -> Any	Convert with error context; raises ConfigTypeConversionError
convert_dict_params_if_needed	(value: Any, option: str) -> Any	Auto-convert dict params based on option name

Usage Example:

from fusion.utils.config import str_to_bool, safe_type_convert

# Boolean conversion
enabled = str_to_bool("true")   # True
enabled = str_to_bool("yes")    # True
enabled = str_to_bool("false")  # False

# Safe type conversion with error context
value = safe_type_convert("100", int, "num_requests")

logging_config.py - Centralized Logging

Purpose:

Standardized logging setup with file rotation and formatting

Lines:

~330

logging_config.py Functions

Function	Signature	Description
setup_logger	(name, level, log_file, ...) -> Logger	Configure logger with console/file handlers and rotation
get_logger	(name: str, level: str) -> Logger	Get existing or create basic logger (convenience)
configure_simulation_logging	(sim_name, erlang, thread_num, ...) -> Logger	Configure simulation-specific logging with naming
log_function_call	(logger: Logger) -> Callable	Decorator to log function calls with args/return
set_global_log_level	(level: str) -> None	Set log level for all FUSION loggers
log_message	(message: str, log_queue) -> None	Log to queue (multi-process) or standard logger

logging_config.py Classes

Class	Purpose	Key Methods
LoggerAdapter	Add contextual info to log messages	__init__(logger, extra), process(msg, kwargs)

Constants:

• DEFAULT_FORMAT: Standard log format with timestamp, name, level, message

• DETAILED_FORMAT: Includes filename and line number for debugging

• LOG_LEVELS: Dict mapping level names to logging constants

Usage Example:

from fusion.utils.logging_config import get_logger, setup_logger

# Simple usage
logger = get_logger(__name__)
logger.info("Starting simulation")

# Advanced with file rotation
logger = setup_logger(
    name="simulation",
    level="DEBUG",
    log_file="sim.log",
    max_bytes=10_485_760,  # 10MB
    backup_count=5
)

# Simulation-specific logging
from fusion.utils.logging_config import configure_simulation_logging
logger = configure_simulation_logging("NSFNet", erlang=300.0, thread_num=1)

os.py - File and Path Operations

Purpose:

Directory creation and project root discovery

Lines:

~63

os.py Functions

Function	Signature	Description
create_directory	(directory_path: str) -> None	Create directory with parent dirs; raises ValueError if None
find_project_root	(start_path: str | None) -> str	Find root by searching for .git or run_sim.py

Usage Example:

from fusion.utils.os import create_directory, find_project_root

# Create nested directories
create_directory("results/experiment_1/data")

# Find project root from anywhere
root = find_project_root()
config_path = os.path.join(root, "configs", "default.ini")

random.py - Random Number Generation

Purpose:

Reproducible RNG for simulations using numpy

Lines:

~79

random.py Functions

Function	Signature	Description
set_random_seed	(seed_value: int) -> None	Set numpy seed for reproducibility
generate_uniform_random_variable	(scale: float | None) -> float | int	Uniform [0,1] or scaled integer [0, scale)
generate_exponential_random_variable	(scale: float) -> float	Exponential distribution via inverse transform

Usage Example:

from fusion.utils.random import (
    set_random_seed,
    generate_uniform_random_variable,
    generate_exponential_random_variable
)

# Reproducible simulation
set_random_seed(42)

# Uniform random for path selection
rand = generate_uniform_random_variable()  # [0, 1)

# Scaled integer for node selection
node = generate_uniform_random_variable(14)  # [0, 14) integer

# Inter-arrival time (Poisson process)
inter_arrival = generate_exponential_random_variable(scale=10.0)

spectrum.py - Spectrum Allocation Utilities

Purpose:

Slot/channel finding, overlap detection, multi-core fiber support

Lines:

~471

spectrum.py Functions

Function	Signature	Description
find_free_slots	(network_spectrum, link_tuple, ...) -> dict	Find unallocated spectral slots per link/core
find_free_channels	(network_spectrum, slots_needed, ...) -> dict	Find contiguous free super-channels
find_taken_channels	(network_spectrum, link_tuple, ...) -> dict	Find occupied super-channels on link
get_channel_overlaps	(free_channels, free_slots) -> dict	Find overlapping channels between adjacent cores
find_common_channels_on_paths	(spectrum, paths, slots, band, core) -> list	Find slots available on ALL paths (for 1+1 protection)
adjacent_core_indices	(core_id: int, cores_per_link: int) -> list	Get adjacent cores for 7/4/13/19-core layouts
edge_set	(path: list, bidirectional: bool) -> set	Normalize path to set of link tuples
get_overlapping_lightpaths	(new_lp, lp_list, ...) -> list	Find lightpaths that overlap with new lightpath

Core Adjacency Diagram:

7-Core Fiber Layout:
      1
    /   \
   6  0  2      Core 0 (center) -> adjacent to 1,2,3,4,5,6
    \   /       Core 1 -> adjacent to 0,2,6
   5  4  3      Core 2 -> adjacent to 0,1,3
                ...etc (hexagonal pattern)

Usage Example:

from fusion.utils.spectrum import (
    find_free_slots,
    find_free_channels,
    adjacent_core_indices
)

# Find free slots on a link
free = find_free_slots(
    network_spectrum=network_state.spectrum,
    link_tuple=(0, 1)
)
# Returns: {"c": {0: [0,1,2,5,6], 1: [0,1,2,3,4], ...}}

# Find contiguous channels for allocation
channels = find_free_channels(
    network_spectrum=network_state.spectrum,
    slots_needed=4,
    link_tuple=(0, 1)
)

# Get adjacent cores for crosstalk analysis
adjacent = adjacent_core_indices(core_id=0, cores_per_link=7)
# Returns: [1, 2, 3, 4, 5, 6]

network.py - Network Path and Congestion Utilities

Purpose:

Path analysis, modulation selection, congestion/fragmentation metrics

Lines:

~269

network.py Functions

Function	Signature	Description
find_path_length	(path: list, topology: nx.Graph) -> float	Sum edge weights along path (km)
find_core_congestion	(core: int, spectrum, path) -> float	Average congestion % on core along path
get_path_modulation	(formats: dict, length: float) -> str | bool	Choose modulation by max reach; False if too long
find_path_congestion	(path, spectrum, band) -> tuple[float, float]	Return (avg_congestion, scaled_capacity)
find_path_fragmentation	(path, spectrum, band) -> float	Compute fragmentation ratio [0,1] along path
average_bandwidth_usage	(bw_dict, departure_time) -> float	Time-weighted average bandwidth utilization

Usage Example:

from fusion.utils.network import (
    find_path_length,
    get_path_modulation,
    find_path_congestion
)

# Calculate path length
path = [0, 3, 5, 8]
length = find_path_length(path, topology)  # e.g., 850.5 km

# Select appropriate modulation
mod = get_path_modulation(
    modulation_formats=engine_props["mod_per_bw"]["50GHz"],
    path_length=length
)  # Returns "QPSK", "16-QAM", "64-QAM", or False

# Check path congestion
congestion, capacity = find_path_congestion(
    path_list=path,
    network_spectrum=spectrum,
    band="c"
)

data.py - Data Structure Manipulation

Purpose:

Dictionary sorting utilities

Lines:

~41

data.py Functions

Function	Signature	Description
sort_dict_keys	(dictionary: dict) -> dict	Sort keys descending (numeric order)
sort_nested_dict_values	(dict, nested_key: str) -> dict	Sort by nested key value ascending

Usage Example:

from fusion.utils.data import sort_dict_keys, sort_nested_dict_values

# Sort modulation formats by bandwidth (descending)
bw_dict = {"100": {...}, "50": {...}, "200": {...}}
sorted_bw = sort_dict_keys(bw_dict)
# Returns: {"200": {...}, "100": {...}, "50": {...}}

# Sort paths by weight
paths = {
    "path1": {"hops": 3, "weight": 0.8},
    "path2": {"hops": 2, "weight": 0.3},
}
sorted_paths = sort_nested_dict_values(paths, "weight")
# Returns paths sorted by weight ascending

Architecture

Module Organization

fusion/utils/
+-- __init__.py          # Limited exports (get_logger, setup_logger)
+-- config.py            # Configuration helpers
+-- logging_config.py    # Logging infrastructure
+-- os.py                # File/path operations
+-- random.py            # RNG for simulations
+-- spectrum.py          # Spectrum allocation
+-- network.py           # Network analysis
+-- data.py              # Data manipulation
+-- tests/               # Unit tests

Dependency Flow

External Dependencies:
+---------------------+
| numpy               | <-- random.py, network.py
| networkx            | <-- network.py
| logging (stdlib)    | <-- logging_config.py
| pathlib (stdlib)    | <-- os.py
| ast (stdlib)        | <-- config.py
+---------------------+

Internal Dependencies:
+---------------------+
| fusion.configs      |
|   .constants        | <-- os.py (PROJECT_ROOT)
|   .errors           | <-- config.py (ConfigTypeConversionError)
+---------------------+

Why Utils Exists Here

The fusion/utils/ module exists to:

1. Avoid circular dependencies: Common functions that multiple modules need (logging, spectrum, network) are centralized here to prevent import cycles between fusion.core, fusion.sim, and fusion.pipelines.

2. Provide shared infrastructure: Logging configuration, RNG seeding, and path utilities are used everywhere and need a single source of truth.

3. Keep domain modules focused: By extracting utility functions, domain modules like fusion.core can focus on simulation logic.

Note

There is also a fusion/sim/utils/ module with simulation-specific utilities. The distinction:

• fusion/utils/: General utilities used across the entire codebase

• fusion/sim/utils/: Simulation orchestration utilities

Common Patterns

Backward Compatibility

Many functions support legacy parameter names:

# Both work - new name preferred
find_free_slots(network_spectrum=state)       # New
find_free_slots(network_spectrum_dict=state)  # Legacy

# Both work - new name preferred
get_path_modulation(modulation_formats=mods)  # New
get_path_modulation(mods_dict=mods)           # Legacy

Error Handling

Functions raise specific exceptions with context:

Error Handling by Module

Module	Exception	Condition
config.py	ConfigTypeConversionError	Type conversion fails
os.py	ValueError	Directory path is None
os.py	RuntimeError	Project root not found
random.py	ValueError	Negative seed or non-positive scale
spectrum.py	ValueError	Missing required parameters
network.py	ValueError	Invalid time progression

Logging Best Practices

# DO: Use get_logger with module name
logger = get_logger(__name__)

# DO: Use appropriate log levels
logger.debug("Detailed debug info")
logger.info("Normal operation info")
logger.warning("Something unexpected")
logger.error("Something failed")

# DON'T: Use print statements
print("Debug info")  # Bad!

# DON'T: Create loggers manually
logger = logging.getLogger(__name__)  # Use get_logger instead

Related Documentation

• Configs Module - Configuration system using config.py

• Core Module - Simulation engine using spectrum/network utilities

• Sim Module - Orchestration layer using logging utilities

• Pipelines Module - Spectrum assignment using spectrum utilities