API Reference

This section provides auto-generated API documentation from source code docstrings. Use this reference for looking up specific classes, functions, and their parameters.

For conceptual documentation, tutorials, and examples, see the Developer Guide.

Tip

Use your browser’s search (Ctrl+F / Cmd+F) or the documentation search box to quickly find specific classes or functions.

---

Configuration System

fusion.configs

Configuration loading, validation, and management.

Centralized configuration management for FUSION simulator.

class fusion.configs.config.SimulationConfig

Bases: object

Data class for simulation configuration.

Contains all configuration sections needed for FUSION simulation. Each attribute represents a configuration section with its parameters.

general: dict[str, Any]

topology: dict[str, Any]

spectrum: dict[str, Any]

snr: dict[str, Any]

rl: dict[str, Any]

ml: dict[str, Any]

file: dict[str, Any]

__init__(general, topology, spectrum, snr, rl, ml, file)

Parameters:

• general (dict[str, Any]) –

• topology (dict[str, Any]) –

• spectrum (dict[str, Any]) –

• snr (dict[str, Any]) –

• rl (dict[str, Any]) –

• ml (dict[str, Any]) –

• file (dict[str, Any]) –

Return type:

None

class fusion.configs.config.ConfigManager

Bases: object

Centralized configuration manager with schema validation.

Manages loading, validation, and access to FUSION configuration from various file formats (INI, JSON, YAML). Supports schema validation and CLI argument merging.

__init__(config_path=None, schema_dir=None)

Initialize configuration manager.

Parameters:

• config_path (Optional[str]) – Path to configuration file to load on initialization

• schema_dir (Optional[str]) – Directory containing schema files for validation

load_config(path)

Load and validate configuration from file.

Automatically detects file format based on extension and loads the configuration. Validates against schema if available.

Parameters:

path (str) – Path to configuration file (INI, JSON, or YAML)

Returns:

Validated configuration object

Return type:

SimulationConfig

Raises:

• ConfigFileNotFoundError – If configuration file doesn’t exist

• ConfigParseError – If file format is unsupported or parsing fails

• ConfigError – If configuration validation fails

Example

>>> manager = ConfigManager()
>>> config = manager.load_config('config.ini')
>>> print(config.general['holding_time'])
10

get_config()

Get the loaded configuration object.

Returns:

Loaded configuration or None if not loaded

Return type:

Optional[SimulationConfig]

get_module_config(module_name)

Get configuration for a specific module.

Returns the configuration section(s) relevant to a specific module. Some modules like ‘routing’ may combine multiple sections.

Parameters:

module_name (str) – Name of the module (e.g., ‘routing’, ‘spectrum’, ‘snr’)

Returns:

Module-specific configuration dictionary

Return type:

Dict[str, Any]

Example

>>> config = manager.get_module_config('routing')
>>> print(config['k_paths'])  # From general settings
3

save_config(path, format_type='ini')

Save current configuration to file.

Saves the current configuration in the specified format.

Parameters:

• path (str) – Output file path

• format_type (str) – Output format (‘ini’, ‘json’, ‘yaml’), defaults to ‘ini’

Raises:

• ValueError – If no configuration is loaded

• ConfigError – If format is unsupported or save fails

Return type:

None

update_config(section, key, value)

Update a specific configuration value.

Updates a single configuration value and recreates the configuration object to reflect the change.

Parameters:

• section (str) – Configuration section name

• key (str) – Configuration key within the section

• value (Any) – New value to set

Return type:

None

Example

>>> manager.update_config('general_settings', 'holding_time', 20)

merge_cli_args(args)

Merge CLI arguments into configuration.

CLI arguments take precedence over existing configuration values.

Parameters:

args (Dict[str, Any]) – Dictionary of CLI arguments to merge

Raises:

ConfigTypeConversionError – If CLI argument mapping fails

Return type:

None

Example

>>> cli_args = {'holding_time': 15, 'network': 'nsfnet'}
>>> manager.merge_cli_args(cli_args)

Schema validation for FUSION configuration files.

exception fusion.configs.validate.ValidationError

Bases: Exception

Custom exception for configuration validation errors.

class fusion.configs.validate.SchemaValidator

Bases: object

Schema validator for configuration files.

__init__(schema_dir=None)

Initialize schema validator.

Parameters:

schema_dir (str | None) – Directory containing schema files

validate(config, schema_name='main')

Validate configuration against schema.

Parameters:

• config (dict[str, Any]) – Configuration dictionary to validate

• schema_name (str) – Name of schema to use for validation, defaults to ‘main’

Raises:

ValidationError – If configuration is invalid

Return type:

None

get_default_config(schema_name='main')

Generate default configuration from schema.

Parameters:

schema_name (str) – Name of schema to use, defaults to ‘main’

Returns:

Default configuration dictionary

Return type:

dict[str, Any]

fusion.configs.validate.validate_survivability_config(config)

Validate survivability-specific configuration.

Parameters:

config (dict[str, Any]) – Configuration dictionary

Raises:

ValidationError – If validation fails

Return type:

None

Example

>>> config = load_config('survivability_experiment.ini')
>>> validate_survivability_config(config)

Configuration registry and factory for FUSION simulator.

class fusion.configs.registry.ConfigRegistry

Bases: object

Registry for managing configuration templates and presets.

This class provides a centralized registry for configuration templates, predefined profiles, and utilities for creating custom configurations. It supports template loading, validation, and profile-based configuration.

__init__(templates_dir=None, schemas_dir=None)

Initialize configuration registry.

Parameters:

• templates_dir (Optional[str]) – Directory containing configuration templates, defaults to ‘templates’ subdirectory

• schemas_dir (Optional[str]) – Directory containing schema files, defaults to ‘schemas’ subdirectory

list_templates()

Get list of available configuration templates.

Returns:

List of available template names

Return type:

List[str]

get_template_path(template_name)

Get path to a specific template.

Parameters:

template_name (str) – Name of the template to find

Returns:

Path to template file, or None if not found

Return type:

Optional[str]

load_template(template_name)

Load a configuration template.

Parameters:

template_name (str) – Name of the template to load

Returns:

ConfigManager instance with loaded template

Return type:

ConfigManager

Raises:

ConfigError – If template is not found

Example

>>> registry = ConfigRegistry()
>>> config = registry.load_template('default')

create_custom_config(base_template='default', overrides=None)

Create a custom configuration based on a template with overrides.

Parameters:

• base_template (str) – Name of base template to use, defaults to ‘default’

• overrides (Optional[Dict[str, Any]]) – Dictionary of configuration overrides in ‘section.key’ or ‘key’ format

Returns:

ConfigManager instance with custom configuration

Return type:

ConfigManager

Raises:

ConfigError – If base template is not found

Example

>>> registry = ConfigRegistry()
>>> config = registry.create_custom_config(
...     'default',
...     {'general_settings.max_iters': 5}
... )

validate_config(config_path)

Validate a configuration file.

Parameters:

config_path (str) – Path to configuration file to validate

Returns:

List of validation errors (empty if valid)

Return type:

List[str]

get_config_profiles()

Get predefined configuration profiles for different use cases.

Returns:

Dictionary mapping profile names to their configurations

Return type:

Dict[str, Dict[str, Any]]

create_profile_config(profile_name, additional_overrides=None)

Create configuration based on a predefined profile.

Parameters:

• profile_name (str) – Name of the profile to use

• additional_overrides (Optional[Dict[str, Any]]) – Additional overrides to apply on top of profile

Returns:

ConfigManager instance with profile configuration

Return type:

ConfigManager

Raises:

ConfigError – If profile is not found

Example

>>> registry = ConfigRegistry()
>>> config = registry.create_profile_config(
...     'development',
...     {'num_requests': 100}
... )

export_config_template(config_manager, template_name, description='')

Export a configuration as a new template.

Parameters:

• config_manager (ConfigManager) – ConfigManager instance to export

• template_name (str) – Name for the new template file

• description (str) – Optional description for the template

Returns:

Path to the exported template file

Return type:

str

CLI argument to configuration mapping for FUSION simulator.

This module handles mapping between CLI arguments and configuration structure.

CORE PARAMETERS (used by both engines):

general_settings, topology_settings, spectrum_settings, snr_settings, file_settings - These are fundamental simulation parameters used by both the legacy engine and the new orchestrator.

LEGACY PARAMETERS (to be phased out):

rl_settings, ml_settings - These use the old RL/ML integration approach. Will be replaced by policy_settings in the orchestrator.

ORCHESTRATOR PARAMETERS (v6.0+):

policy_settings, heuristic_settings, protection_settings, failure_settings, routing_settings, reporting_settings - New parameters for the orchestrator-based simulation engine supporting survivability experiments and policy-based routing.

See schema.py for the corresponding type definitions.

class fusion.configs.cli_to_config.CLIToConfigMapper

Bases: object

Maps CLI arguments to configuration structure.

This class provides utilities to map command-line arguments to the hierarchical configuration structure used by FUSION. It maintains a mapping dictionary that associates CLI argument names with their corresponding configuration sections and keys.

__init__()

Initialize CLI to config mapper.

Sets up the internal mapping dictionary that defines how CLI arguments correspond to configuration sections and keys.

Return type:

None

map_args_to_config(args)

Map CLI arguments to configuration structure.

Takes a flat dictionary of CLI arguments and transforms it into a hierarchical configuration dictionary organized by sections.

Parameters:

args (Dict[str, Any]) – Dictionary of CLI arguments where keys are argument names and values are the argument values

Returns:

Configuration dictionary organized by sections where each section contains its respective configuration parameters

Return type:

Dict[str, Dict[str, Any]]

Example

>>> mapper = CLIToConfigMapper()
>>> cli_args = {'holding_time': 10, 'network': 'nsfnet'}
>>> config = mapper.map_args_to_config(cli_args)
>>> print(config)
{'general_settings': {'holding_time': 10},
 'topology_settings': {'network': 'nsfnet'}}

map_namespace_to_config(args)

Map argparse Namespace to configuration structure.

Convenience method that converts an argparse.Namespace object to a configuration dictionary.

Parameters:

args (argparse.Namespace) – argparse.Namespace object containing parsed CLI arguments

Returns:

Configuration dictionary organized by sections

Return type:

Dict[str, Dict[str, Any]]

Raises:

AttributeError – If args is not a valid Namespace object

get_cli_override_config(cli_args, base_config)

Get configuration with CLI arguments overriding base config.

Merges CLI arguments with a base configuration, where CLI arguments take precedence over base configuration values.

Parameters:

• cli_args (Dict[str, Any]) – CLI arguments dictionary to override base config

• base_config (Dict[str, Any]) – Base configuration dictionary to be overridden

Returns:

Merged configuration with CLI overrides applied

Return type:

Dict[str, Any]

Example

>>> base = {'general_settings': {'holding_time': 5}}
>>> cli = {'holding_time': 10}
>>> merged = mapper.get_cli_override_config(cli, base)
>>> print(merged['general_settings']['holding_time'])
10

get_reverse_mapping()

Get reverse mapping from config path to CLI argument name.

Creates a reverse lookup dictionary that maps configuration paths (in the format ‘section.key’) to their corresponding CLI argument names.

Returns:

Dictionary mapping config paths to CLI argument names

Return type:

Dict[str, str]

Example

>>> mapper = CLIToConfigMapper()
>>> reverse_map = mapper.get_reverse_mapping()
>>> print(reverse_map['general_settings.holding_time'])
'holding_time'

Configuration schema definitions for FUSION simulator.

This module defines the required and optional configuration options for the FUSION simulator, including their expected types and conversion functions. These schemas are used for validation and type conversion during configuration loading.

Schema Organization

CORE PARAMETERS (SIM_REQUIRED_OPTIONS_DICT):

Required parameters used by both legacy engine and orchestrator. Includes: general_settings, topology_settings, spectrum_settings, snr_settings, file_settings, ml_settings

OPTIONAL PARAMETERS (OPTIONAL_OPTIONS_DICT):

Optional parameters organized by usage:

• Core optional: general_settings, topology_settings, spectrum_settings, file_settings

• LEGACY (to be phased out): rl_settings, ml_settings

• ORCHESTRATOR (v6.0+): policy_settings, heuristic_settings, protection_settings, failure_settings, routing_settings, reporting_settings, dataset_logging_settings, offline_rl_settings, recovery_timing_settings

See cli_to_config.py for the corresponding CLI argument mappings.

Configuration-related exception classes for the FUSION CLI.

exception fusion.configs.errors.ConfigError

Bases: Exception

Base exception for configuration errors.

All configuration-related exceptions inherit from this class, allowing for broad exception handling when needed.

exception fusion.configs.errors.ConfigFileNotFoundError

Bases: ConfigError

Raised when config file cannot be found.

This exception is raised when attempting to load a configuration file that doesn’t exist at the specified path.

exception fusion.configs.errors.ConfigParseError

Bases: ConfigError

Raised when config file cannot be parsed.

This exception is raised when a configuration file exists but contains invalid syntax or cannot be parsed in the expected format (INI, JSON, YAML).

exception fusion.configs.errors.MissingRequiredOptionError

Bases: ConfigError

Raised when a required option is missing from config.

This exception is raised during configuration validation when a required parameter is not present in the configuration.

exception fusion.configs.errors.ConfigTypeConversionError

Bases: ConfigError

Raised when a config value cannot be converted to expected type.

This exception is raised when a configuration value cannot be converted to its expected type (e.g., string to int conversion fails).

Configuration constants for the FUSION CLI.

---

Command Line Interface

fusion.cli

CLI entry points and argument parsing.

CLI entry point for running simulations.

Follows architecture best practice: entry points should have no logic.

fusion.cli.run_sim.main(stop_flag=None)

Entry point for running simulations from the command line.

Delegates all logic to appropriate modules following the clean architecture pattern where entry points contain minimal logic and delegate to pipelines.

Parameters:

stop_flag (Any) – Optional multiprocessing event for stopping simulation

Returns:

Exit code (0 for success, 1 for error or interruption)

Return type:

int

Raises:

KeyboardInterrupt – When user interrupts the simulation

CLI entry point for training FUSION agents.

This module provides the command-line interface for training reinforcement learning (RL) and supervised learning (SL) agents. It supports multiple training algorithms and provides detailed error handling with helpful guidance for common issues.

fusion.cli.run_train.main()

Train FUSION agents using RL or SL algorithms.

Parses command line arguments and delegates training execution to the appropriate training pipeline module. Supports both reinforcement learning (RL) and supervised learning (SL) workflows with proper error handling and user feedback.

Returns:

Exit code (0 for success, 1 for error or interruption)

Return type:

int

Raises:

SystemExit – On argument parsing errors (handled by argparse)

Configuration setup module for the FUSION simulator CLI.

This module handles loading, parsing, and validating INI configuration files for the FUSION optical network simulator. It provides:

• Configuration file path resolution and validation

• INI file parsing with type conversion

• CLI argument overrides for config values

• Multi-process configuration section support (s1, s2, etc.)

• Backward compatibility with legacy flat config structure

Key Functions: - load_config(): Main entry point for loading configuration files - setup_config_from_cli(): Wrapper for CLI-based config loading - normalize_config_path(): Resolves relative/absolute config paths

Key Classes: - ConfigManager: High-level interface for configuration access

WARNING: If is_training is set to True, the confidence interval (CI) for blocking probability will be ignored during simulation.

fusion.cli.config_setup.normalize_config_path(config_path)

Normalize the config file path.

Parameters:

config_path (str) – Path to config file (relative or absolute)

Returns:

Absolute path to config file

Return type:

str

Raises:

ConfigFileNotFoundError – If config file does not exist

fusion.cli.config_setup.setup_config_from_cli(args)

Set up configuration from command line input.

Parameters:

args (Any) – Parsed command line arguments

Returns:

Configuration dictionary

Return type:

Dict[str, Any]

fusion.cli.config_setup.load_config(config_path, args_dict=None)

Load an existing config from a config file.

This function handles the complete configuration loading process: 1. Resolves and validates the config file path 2. Reads and parses the INI configuration file 3. Validates the configuration structure and required sections 4. Processes both required and optional configuration options 5. Applies type conversions and CLI argument overrides 6. Handles multi-process configuration sections (s1, s2, etc.)

TODO (v6.1.0): Remove returning empty dict on error - raise exceptions instead. Use setup_config_from_cli for better error handling.

Parameters:

• config_path (Optional[str]) – Path to configuration file

• args_dict (Optional[Dict[str, Any]]) – Optional CLI arguments dictionary for overrides

Returns:

Configuration dictionary with process-based structure (s1, s2, etc.)

Return type:

Dict[str, Any]

fusion.cli.config_setup.load_and_validate_config(args)

Load and validate configuration from CLI arguments.

Parameters:

args (Any) – Parsed command line arguments

Returns:

Validated configuration dictionary

Return type:

Dict[str, Any]

class fusion.cli.config_setup.ConfigManager

Bases: object

Centralized configuration management for FUSION simulator.

Provides a unified interface for accessing configuration from both INI files and command-line arguments, with proper validation and error handling. Supports multi-process configuration sections (s1, s2, etc.).

__init__(config_dict, args)

Initialize ConfigManager with configuration dictionary and arguments.

Parameters:

• config_dict (Dict[str, Any]) – Parsed configuration dictionary

• args (Any) – Command line arguments

Return type:

None

classmethod from_args(args)

Load arguments from command line input.

Parameters:

args (Any) – Parsed command line arguments

Returns:

ConfigManager instance

Return type:

ConfigManager

Raises:

ConfigError – If configuration loading fails

classmethod from_file(config_path, args_dict=None)

Create ConfigManager from config file path.

Parameters:

• config_path (str) – Path to configuration file

• args_dict (Optional[Dict[str, Any]]) – Optional dictionary of arguments to override config

Returns:

ConfigManager instance

Return type:

ConfigManager

as_dict()

Get config as dict.

Returns:

Configuration dictionary

Return type:

Dict[str, Any]

get(thread='s1')

Return a single config thread.

Parameters:

thread (str) – Thread identifier, defaults to ‘s1’

Returns:

Configuration for specified thread

Return type:

Dict[str, Any]

get_value(key, thread='s1', default=None)

Get a specific configuration value.

Parameters:

• key (str) – Configuration key

• thread (str) – Thread identifier, defaults to ‘s1’

• default (Any) – Default value if key not found, defaults to None

Returns:

Configuration value or default

Return type:

Any

has_thread(thread)

Check if a thread exists in configuration.

Parameters:

thread (str) – Thread identifier

Returns:

True if thread exists, False otherwise

Return type:

bool

get_threads()

Get list of all configured threads.

Returns:

List of thread identifiers

Return type:

List[str]

get_args()

Get args.

Returns:

Command line arguments

Return type:

Any

Main CLI argument parser using the centralized registry system.

exception fusion.cli.main_parser.GUINotSupportedError

Bases: Exception

Raised when attempting to use the unsupported GUI module.

fusion.cli.main_parser.build_main_argument_parser()

Build the main CLI argument parser with all subcommands configured.

Creates the primary argument parser that handles all CLI interactions for the FUSION simulator, including subcommands for training and simulation execution.

Returns:

Fully configured argument parser with subcommands

Return type:

ArgumentParser

fusion.cli.main_parser.create_training_argument_parser()

Create and parse arguments for training simulations.

Builds a specialized argument parser configured for reinforcement learning (RL) and supervised learning (SL) training workflows. Includes all necessary argument groups for comprehensive training configuration.

Returns:

Parsed command line arguments for training operations

Return type:

Namespace

Raises:

SystemExit – If required arguments are missing or invalid

fusion.cli.main_parser.create_gui_argument_parser()

Create and parse arguments for GUI-based simulator interface.

Raises:

GUINotSupportedError – Always raised - GUI is not currently supported

Return type:

Namespace

fusion.cli.main_parser.build_parser()

Legacy function name for building main argument parser.

Returns:

Configured main parser

Return type:

ArgumentParser

Deprecated:

Use build_main_argument_parser() instead

fusion.cli.main_parser.get_train_args()

Legacy function name for creating training argument parser.

Returns:

Parsed arguments for training

Return type:

Namespace

Deprecated:

Use create_training_argument_parser() instead

fusion.cli.main_parser.get_gui_args()

Legacy function name for creating GUI argument parser.

Raises:

GUINotSupportedError – Always raised - GUI is not currently supported

Deprecated:

Use create_gui_argument_parser() instead

Return type:

Namespace

---

Core Simulation

fusion.core

Core simulation engine and orchestration.

Simulation engine module for running optical network simulations.

This module provides the main simulation engine functionality for running optical network simulations with support for ML/RL models and various metrics.

fusion.core.simulation.seed_request_generation(seed)

Seed random number generators used for request generation.

This function seeds ONLY NumPy, which is used for generating traffic patterns (arrivals, departures, bandwidth selection, etc.). This allows request generation to vary per iteration while keeping RL/ML models deterministic across iterations.

Parameters:

seed (int) – Random seed (integer)

Return type:

None

fusion.core.simulation.seed_rl_components(seed)

Seed random number generators used for RL/ML components.

This function seeds Python’s random module and PyTorch (CPU and CUDA), which are used by RL agents and ML models. This allows RL components to remain deterministic even when request generation seeds vary.

Also sets PyTorch to deterministic mode to prevent non-deterministic operations in neural network computations.

Parameters:

seed (int) – Random seed (integer)

Return type:

None

fusion.core.simulation.seed_all_rngs(seed)

Seed all random number generators for reproducibility.

Seeds: - Python’s built-in random module - NumPy’s random state - PyTorch’s random state (CPU and CUDA)

Also sets PyTorch to deterministic mode to prevent non-deterministic operations.

Note

For finer control, use seed_request_generation() and seed_rl_components() separately. This allows different seeding strategies for traffic generation vs RL/ML models.

Parameters:

seed (int) – Random seed (integer)

Return type:

None

fusion.core.simulation.generate_seed_from_time()

Generate a seed from current time.

Used when no seed is explicitly provided.

Returns:

Seed value

Return type:

int

fusion.core.simulation.validate_seed(seed)

Validate and normalize seed value.

Ensures seed is in valid range for all RNGs.

Parameters:

seed (int) – Seed value

Returns:

Validated seed

Return type:

int

Raises:

ValueError – If seed is out of valid range

class fusion.core.simulation.SimulationEngine

Bases: object

Controls a single simulation.

This class manages the execution of a single optical network simulation, including topology creation, request processing, and statistics collection.

Parameters:

engine_props (dict[str, Any]) – Engine configuration properties

__init__(engine_props)

Parameters:

engine_props (dict[str, Any]) –

Return type:

None

update_arrival_params(current_time)

Update parameters for a request after attempted allocation.

Parameters:

current_time (tuple[int, float]) – The current simulated time as (iteration, time) tuple

Return type:

None

handle_arrival(current_time, force_route_matrix=None, force_core=None, force_slicing=False, forced_index=None, force_mod_format=None)

Update the SDN controller to handle an arrival request.

In orchestrator mode, delegates to _handle_arrival_orchestrator. In legacy mode, uses SDNController.

Parameters:

• current_time (float) – The arrival time of the request

• force_route_matrix (Optional[List[Any]]) – Passes forced routes to the SDN controller

• force_core (Optional[int]) – Force a certain core for allocation

• force_slicing (bool) – Forces slicing in the SDN controller

• forced_index (Optional[int]) – Forces an index in the SDN controller

• force_mod_format (Optional[str]) – Forces a modulation format

Return type:

None

handle_release(current_time)

Update the SDN controller to handle the release of a request.

In orchestrator mode (v5), delegates to _handle_release_orchestrator. In legacy mode, uses SDNController.

Parameters:

current_time (float) – The arrival time of the request

Return type:

None

create_topology()

Create the physical topology of the simulation.

In orchestrator mode, also initializes:

• SimulationConfig from engine_props

• NetworkState with topology

• SDNOrchestrator with pipelines

Return type:

None

generate_requests(seed)

Call the request generator to generate requests.

Parameters:

seed (int) – The seed to use for the random generation

Return type:

None

handle_request(current_time, request_number)

Carry out arrival or departure functions for a given request.

Parameters:

• current_time (float) – The current simulated time

• request_number (int) – The request number

Return type:

None

reset()

Reset simulation state for new iteration.

Clears all tracking dictionaries and counters to prepare for a fresh simulation run.

In orchestrator mode, also resets NetworkState.

Return type:

None

end_iter(iteration, print_flag=True, base_file_path=None)

Update iteration statistics.

Parameters:

• iteration (int) – The current iteration

• print_flag (bool) – Whether to print or not

• base_file_path (Optional[str]) – The base file path to save output statistics

Returns:

Whether confidence interval has been reached

Return type:

bool

init_iter(iteration, seed=None, print_flag=True, trial=None)

Initialize an iteration.

Seeds all random number generators (Python random, NumPy, PyTorch) to ensure reproducible results across iterations.

Parameters:

• iteration (int) – The current iteration number

• seed (Optional[int]) – The seed to use for the random generation

• print_flag (bool) – Flag to determine printing iter info

• trial (Optional[int]) – The trial number

Return type:

None

run(seed=None)

Run the simulation by creating the topology and processing requests.

This method creates the topology, processes requests, and sends iteration-based updates to the parent’s queue.

Parameters:

seed (Optional[int]) – Optional seed for random generation

Returns:

Number of completed iteration units

Return type:

int

property num_requests: int

Total number of arrival requests in current episode.

Returns:

Number of arrival requests (excludes release events)

Return type:

int

property network_state: NetworkState | None

Access to the v5 NetworkState object.

Returns:

NetworkState if orchestrator mode enabled, None otherwise

Return type:

NetworkState | None

get_next_request()

Get the next unprocessed arrival request for RL.

This method is used by UnifiedSimEnv to get requests one at a time for step-by-step RL decision making. Requires orchestrator mode.

Returns:

Next Request object, or None if all arrivals processed

Return type:

Request | None

process_releases_until(time)

Process all release events due before given time.

This method is used by UnifiedSimEnv to process lightpath releases between RL decision points. Requires orchestrator mode.

Parameters:

time (float) – Process all releases with depart_time <= time

Return type:

None

record_allocation_result(request, result)

Record allocation result for RL statistics.

This method is used by UnifiedSimEnv to record the result of each RL decision and advance to the next request. Updates statistics counters, schedules release event if successful, and advances index.

Parameters:

• request (Request) – The Request that was processed

• result (AllocationResult) – AllocationResult from orchestrator

Return type:

None

reset_rl_state()

Reset RL-specific state for new episode.

Called by UnifiedSimEnv.reset() to prepare for a new episode.

Return type:

None

SDN Orchestrator for FUSION simulation.

This module provides SDNOrchestrator, a thin coordination layer that routes requests through pipelines without implementing algorithm logic.

RULES (enforced by code review): - No algorithm logic (K-shortest-path, first-fit, SNR calculation) - No direct numpy access - No hardcoded slicing/grooming logic - Receives NetworkState per call, never stores it

class fusion.core.orchestrator.SDNOrchestrator

Bases: object

Thin coordination layer for request handling.

The orchestrator sequences pipeline calls and combines their results. It does NOT implement algorithm logic - all computation is delegated to pipelines.

Does NOT store network_state - receives per call only.

Variables:

• config – Simulation configuration

• routing – Pipeline for finding candidate routes

• spectrum – Pipeline for spectrum assignment

• grooming – Pipeline for traffic grooming (optional)

• snr – Pipeline for SNR validation (optional)

• slicing – Pipeline for request slicing (optional)

__init__(config, pipelines, policy=None, rl_adapter=None, protection_pipeline=None)

Initialize orchestrator with config and pipelines.

Parameters:

• config (SimulationConfig) – Simulation configuration

• pipelines (PipelineSet) – Container with all pipeline implementations

• policy (ControlPolicy | None) – Optional control policy for path selection

• rl_adapter (RLSimulationAdapter | None) – Optional RL adapter for building options

• protection_pipeline (ProtectionPipeline | None) – Optional protection pipeline

Return type:

None

config: SimulationConfig

routing: RoutingPipeline

spectrum: SpectrumPipeline

grooming: GroomingPipeline | None

snr: SNRPipeline | None

slicing: SlicingPipeline | None

current_iteration: int

handle_arrival(request, network_state, forced_path=None, forced_modulation=None, forced_path_index=None)

Handle request arrival by coordinating pipelines.

Parameters:

• request (Request) – The incoming request to process

• network_state (NetworkState) – Current network state (passed per call)

• forced_path (list[str] | None) – Optional forced path from external source

• forced_modulation (str | None) – Optional forced modulation format (for RL)

• forced_path_index (int | None) – Optional path index for stats tracking (for RL)

Returns:

AllocationResult with success/failure and details

Return type:

AllocationResult

handle_release(request, network_state)

Handle request release.

Releases bandwidth from all lightpaths associated with the request. If a lightpath has no remaining allocations, it is fully released.

Parameters:

• request (Request) –

• network_state (NetworkState) –

Return type:

None

property policy: ControlPolicy | None

Get the current control policy (None if not set).

property rl_adapter: RLSimulationAdapter | None

Get the RL adapter (None if not set).

property protection_pipeline: ProtectionPipeline | None

Get the protection pipeline (None if not set).

has_policy()

Check if a policy is configured.

Return type:

bool

handle_arrival_with_policy(request, network_state)

Handle request arrival using the configured policy.

This method provides policy-driven path selection while maintaining backward compatibility with handle_arrival(). It builds PathOption list via RL adapter, calls policy.select_action() to get selected path, validates action, and routes through standard allocation with forced path.

If no policy is configured, delegates to handle_arrival().

Protection pipeline is only used when protection_enabled is True in config, request.protection_required is True, and protection_pipeline is set.

Parameters:

• request (Request) – The incoming request to process

• network_state (NetworkState) – Current network state (passed per call)

Returns:

AllocationResult with success/failure and details

Return type:

AllocationResult

Software-defined network controller module for managing network requests.

This module provides the main SDN controller functionality for routing and spectrum allocation in software-defined optical networks.

class fusion.core.sdn_controller.SDNController

Bases: object

Software-defined network controller for managing network requests.

This class provides functionality for routing, spectrum allocation, and resource management in software-defined optical networks. It handles request allocation, release, and various slicing strategies.

Parameters:

engine_props (dict[str, Any]) – Engine configuration properties

__init__(engine_props)

Parameters:

engine_props (dict[str, Any]) –

Return type:

None

release(lightpath_id=None, slicing_flag=False, skip_validation=False)

Remove a previously allocated request from the network.

Parameters:

• lightpath_id (int | None) – Specific lightpath ID to release (for grooming)

• slicing_flag (bool) – If True, only release spectrum, not transponders

• skip_validation (bool) –

Return type:

None

allocate()

Allocate spectrum resources for a network request.

Assigns spectrum slots to the current request across all links in the path, including guard bands if configured. For 1+1 protected requests, allocates on both primary and backup paths.

Raises:

• BufferError – If attempting to allocate already taken spectrum

• ValueError – If no spectrum is detected during allocation

Return type:

None

handle_event(request_dict, request_type, force_slicing=False, force_route_matrix=None, forced_index=None, force_core=None, ml_model=None, force_mod_format=None, forced_band=None)

Handle any event that occurs in the simulation.

Controls the main flow of request processing including routing, spectrum allocation, and various slicing strategies.

Parameters:

• request_dict (dict[str, Any]) – Request dictionary containing request parameters

• request_type (str) – Type of request (‘arrival’ or ‘release’)

• force_slicing (bool) – Whether to force light path segment slicing

• force_route_matrix (list[Any] | None) – Optional forced routing matrix

• forced_index (int | None) – Optional forced start index for spectrum allocation

• force_mod_format (str | None) – Optional forced modulation format

• force_core (int | None) – Optional forced core number

• ml_model (Any | None) – Optional machine learning model for predictions

• forced_band (str | None) – Optional forced spectral band

Return type:

None

Pipeline factory for FUSION simulation.

This module provides PipelineFactory for creating pipelines based on SimulationConfig, and PipelineSet for holding pipeline references.

Design Principles:

• Factory is stateless (static/class methods only)

• Lazy imports to avoid circular dependencies

• Config-driven pipeline selection

• Optional pipelines return None when disabled

Usage:

>>> config = SimulationConfig.from_engine_props(engine_props)
>>> pipelines = PipelineFactory.create_pipeline_set(config)
>>> orchestrator = PipelineFactory.create_orchestrator(config)

class fusion.core.pipeline_factory.PipelineSet

Bases: object

Container for all pipelines used by SDNOrchestrator.

This dataclass holds references to pipeline implementations that will be called by the orchestrator during request handling.

Required pipelines (routing, spectrum) are always present. Optional pipelines (grooming, snr, slicing) may be None if disabled.

Variables:

• routing – RoutingPipeline implementation (required)

• spectrum – SpectrumPipeline implementation (required)

• grooming – GroomingPipeline or None if disabled

• snr – SNRPipeline or None if disabled

• slicing – SlicingPipeline or None if disabled

routing: RoutingPipeline

spectrum: SpectrumPipeline

grooming: GroomingPipeline | None

snr: SNRPipeline | None

slicing: SlicingPipeline | None

property has_grooming: bool

Check if grooming is available.

property has_snr: bool

Check if SNR validation is available.

property has_slicing: bool

Check if slicing is available.

__init__(routing, spectrum, grooming=None, snr=None, slicing=None)

Parameters:

• routing (RoutingPipeline) –

• spectrum (SpectrumPipeline) –

• grooming (GroomingPipeline | None) –

• snr (SNRPipeline | None) –

• slicing (SlicingPipeline | None) –

Return type:

None

class fusion.core.pipeline_factory.PipelineFactory

Bases: object

Factory for creating pipelines based on SimulationConfig.

This factory uses lazy imports to avoid circular dependencies and selects between legacy adapters and new pipeline implementations based on configuration values. All methods are static or class methods (stateless factory pattern).

static create_routing(config)

Create routing pipeline based on config.

Returns ProtectedRoutingPipeline for 1+1 protection, otherwise RoutingAdapter.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

RoutingPipeline implementation

Return type:

RoutingPipeline

static create_spectrum(config)

Create spectrum pipeline based on config.

Currently always returns SpectrumAdapter (wraps legacy).

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

SpectrumPipeline implementation

Return type:

SpectrumPipeline

static create_grooming(config)

Create grooming pipeline if enabled.

Returns GroomingAdapter if grooming_enabled, otherwise None.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

GroomingPipeline if enabled, None otherwise

Return type:

GroomingPipeline | None

static create_snr(config)

Create SNR pipeline if enabled.

Returns SNRAdapter if snr_enabled, otherwise None.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

SNRPipeline if enabled, None otherwise

Return type:

SNRPipeline | None

static create_slicing(config)

Create slicing pipeline if enabled.

Returns StandardSlicingPipeline if slicing_enabled, otherwise None.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

SlicingPipeline if enabled, None otherwise

Return type:

SlicingPipeline | None

classmethod create_pipeline_set(config)

Create complete pipeline set from configuration.

Creates all pipelines based on configuration settings. Required pipelines (routing, spectrum) are always created. Optional pipelines may be None.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

PipelineSet with all pipelines configured

Return type:

PipelineSet

classmethod create_orchestrator(config)

Create orchestrator with configured pipelines.

Convenience method that creates both pipeline set and orchestrator. The orchestrator is the main entry point for request handling.

Parameters:

config (SimulationConfig) – Simulation configuration

Returns:

SDNOrchestrator instance with configured pipelines

Return type:

SDNOrchestrator

Request generation module for FUSION simulations.

This module handles the generation of arrival and departure requests for network simulations. It creates requests based on specified distributions and traffic patterns.

fusion.core.request.generate_simulation_requests(seed, engine_properties)

Generate requests for a single simulation run.

Creates both arrival and departure events for network requests based on the specified traffic parameters and distributions. Each request has a unique ID and is placed at specific simulation times.

Parameters:

• seed (int) – Random seed for reproducible generation

• engine_properties (Dict[str, Any]) – Simulation engine configuration containing: - is_only_core_node: Whether all nodes can originate traffic - topology_info: Network topology information - core_nodes: List of core nodes (if not all nodes are core) - arrival_rate: Rate parameter for exponential arrivals - holding_time: Mean request duration - num_requests: Total number of requests to generate - request_distribution: Bandwidth distribution percentages - mod_per_bw: Modulation formats per bandwidth

Returns:

Dictionary mapping (request_id, time) tuples to request events

Return type:

Dict[tuple[int, float], Dict[str, Any]]

Raises:

ValueError – If no nodes available or distribution invalid

Example

>>> engine_props = {
...     "is_only_core_node": True,
...     "topology_info": {"nodes": {"A": {}, "B": {}, "C": {}}},
...     "arrival_rate": 1.0,
...     "holding_time": 10.0,
...     "num_requests": 100,
...     "request_distribution": {"50GHz": 0.5, "100GHz": 0.5},
...     "mod_per_bw": {"50GHz": {...}, "100GHz": {...}}
... }
>>> requests = generate_simulation_requests(42, engine_props)

fusion.core.request.validate_request_distribution(request_distribution, number_of_requests)

Validate that request distribution can be properly allocated.

Checks if the request distribution percentages can be evenly distributed across the specified number of requests without remainder.

Parameters:

• request_distribution (Dict[str, float]) – Bandwidth distribution percentages

• number_of_requests (int) – Total number of requests to generate

Returns:

True if distribution is valid, False otherwise

Return type:

bool

Example

>>> distribution = {"50GHz": 0.5, "100GHz": 0.5}
>>> validate_request_distribution(distribution, 100)
True

class fusion.core.metrics.SimStats

Bases: object

The SimStats class finds and stores all relevant statistics in simulations.

__init__(engine_props, sim_info, stats_props=None)

Parameters:

• engine_props (dict[str, Any]) –

• sim_info (str) –

• stats_props (StatsProps | None) –

update_snapshot(network_spectrum_dict, request_number, path_list=None)

Stub for snapshot collection - not currently functional.

This method is a placeholder that maintains API compatibility but does not collect snapshot data. Full implementation planned for v6.1.

Parameters:

• network_spectrum_dict (dict[tuple[Any, Any], dict[str, Any]]) – The current network spectrum database.

• request_number (int) – The current request number.

• path_list (list[tuple[int, int]] | None) – The desired path to find the occupied slots on.

Returns:

None

Return type:

None

init_iter_stats()

Initializes data structures used in other methods of this class.

Returns:

None

Return type:

None

calculate_blocking_statistics()

Gets the current blocking probability.

Returns:

None

Return type:

None

iter_update(req_data, sdn_data, network_spectrum_dict)

Continuously updates the statistical data for each request allocated/blocked in the current iteration.

Parameters:

• req_data (dict[str, Any]) – Holds data relevant to the current request.

• sdn_data (SDNProps) – Hold the response data from the sdn controller.

• network_spectrum_dict (dict[tuple[Any, Any], dict[str, Any]]) –

Returns:

None

Return type:

None

update_utilization_dict(utilization_dict)

Update lightpath bandwidth utilization statistics.

Called after lightpath release to track bandwidth utilization metrics per bandwidth/band/core combination.

Parameters:

utilization_dict (dict[int, dict[str, Any]]) – Dictionary mapping lightpath_id to utilization info with keys: ‘bit_rate’, ‘band’, ‘core’, ‘utilization’

Return type:

None

finalize_iteration_statistics()

Updates relevant stats after an iteration has finished.

Returns:

None

Return type:

None

calculate_confidence_interval()

Get the confidence interval for every iteration so far.

Returns:

Whether the simulations should end for this erlang.

Return type:

bool

get_blocking_statistics()

Get all blocking-related statistics for persistence.

Returns:

Dictionary containing blocking statistics

Return type:

dict

end_iter_update()

Backward compatibility wrapper for finalize_iteration_statistics.

Return type:

None

save_stats(base_fp='data')

Backward compatibility method for saving statistics.

Parameters:

base_fp (str) – Base file path for saving

Return type:

None

record_recovery_event(failure_time, recovery_time, affected_requests, recovery_type)

Record a recovery event.

Parameters:

• failure_time (float) – Time of failure occurrence

• recovery_time (float) – Time when recovery completed

• affected_requests (int) – Number of affected requests

• recovery_type (str) – Type of recovery mechanism

Return type:

None

Example

>>> stats.record_recovery_event(
...     failure_time=100.0,
...     recovery_time=100.05,  # 50ms later
...     affected_requests=5,
...     recovery_type='protection'
... )

get_recovery_stats()

Get recovery time statistics.

Computes mean, P95, and max recovery times across all recovery events.

Returns:

Dict with recovery statistics

Return type:

dict[str, float]

Example

>>> stats = get_recovery_stats()
>>> print(stats)
{
    'mean_ms': 52.3,
    'p95_ms': 98.5,
    'max_ms': 105.2,
    'count': 12
}

compute_failure_window_bp(failure_time, arrival_times, blocked_requests)

Compute blocking probability within failure window.

Measures BP in the window [failure_time, failure_time + window_size] where window_size is specified in number of arrivals.

Parameters:

• failure_time (float) – Failure occurrence time

• arrival_times (list[float]) – List of all request arrival times

• blocked_requests (list[int]) – List of blocked request indices

Returns:

BP within failure window

Return type:

float

Example

>>> bp = stats.compute_failure_window_bp(
...     failure_time=100.0,
...     arrival_times=all_arrivals,
...     blocked_requests=blocked_ids
... )
>>> print(f"Failure window BP: {bp:.4f}")
0.0823

get_failure_window_stats()

Get failure window BP statistics.

Returns:

Dict with mean and P95 failure window BP

Return type:

dict[str, float]

get_recovery_csv_row()

Export recovery statistics as CSV row.

Returns:

Dict with all recovery metric values

Return type:

dict[str, Any]

compute_fragmentation_proxy(path, network_spectrum_dict)

Compute fragmentation proxy for a path.

Fragmentation = 1 - (largest_contiguous_block / total_free_slots)

Higher values indicate more fragmentation.

Parameters:

• path (list[int]) – Path node list

• network_spectrum_dict (dict[tuple[Any, Any], dict[str, Any]]) – Spectrum state

Returns:

Fragmentation score [0, 1]

Return type:

float

Example

>>> frag = stats.compute_fragmentation_proxy(path, spectrum_dict)
>>> print(f"Fragmentation: {frag:.3f}")
0.347

record_fragmentation(path, network_spectrum_dict)

Record fragmentation score for a path.

Parameters:

• path (list[int]) – Path node list

• network_spectrum_dict (dict[tuple[Any, Any], dict[str, Any]]) – Spectrum state

Return type:

None

record_decision_time(decision_time_ms)

Record policy decision time.

Parameters:

decision_time_ms (float) – Decision time in milliseconds

Return type:

None

get_fragmentation_stats()

Get fragmentation statistics.

Returns:

Dict with mean and P95 fragmentation scores

Return type:

dict[str, float]

get_decision_time_stats()

Get decision time statistics.

Returns:

Dict with mean and P95 decision times

Return type:

dict[str, float]

to_csv_row()

Export all statistics as CSV row.

Includes standard metrics plus survivability-specific metrics.

Returns:

Dict with all metric values

Return type:

dict[str, Any]

record_arrival(request, result, network_state, was_rollback=False)

Record statistics from orchestrator allocation result.

This method provides the same functionality as iter_update but works with domain objects instead of legacy sdn_props.

Parameters:

• request (Any) – The Request that was processed

• result (Any) – The AllocationResult from orchestrator

• network_state (Any) – Current NetworkState for spectrum data

• was_rollback (bool) – True if this was a rollback (skip utilization tracking)

Return type:

None

record_grooming_rollback(request, rolled_back_lightpath_ids, network_state)

Record that grooming was rolled back - adjust utilization stats.

Called when partial grooming succeeds but new lightpath allocation fails, requiring rollback of the groomed bandwidth.

Parameters:

• request (Any) – The Request being processed

• rolled_back_lightpath_ids (list[int]) – IDs of lightpaths that had grooming rolled back

• network_state (Any) – Current NetworkState

Return type:

None

---

Domain Models

fusion.domain

Domain objects and data structures.

SimulationConfig - Immutable simulation configuration.

This module defines the SimulationConfig frozen dataclass that replaces the legacy engine_props dictionary with a typed, immutable structure.

class fusion.domain.config.SimulationConfig

Bases: object

Immutable simulation configuration.

This dataclass captures all simulation parameters in a typed, frozen structure. Once created, the configuration cannot be modified.

network_name

Network topology identifier (e.g., “USbackbone60”).

Type:

str

cores_per_link

Number of cores per fiber link (MCF support).

Type:

int

band_list

Available frequency bands as immutable tuple.

Type:

tuple[str, …]

band_slots

Slot count per band (e.g., {“c”: 320, “l”: 320}).

Type:

dict[str, int]

guard_slots

Guard band slots between allocations.

Type:

int

span_length

Default span length in km.

Type:

float

max_link_length

Maximum link length constraint (None = no limit).

Type:

float | None

max_span

Maximum spans per link (None = no limit).

Type:

int | None

max_transponders

Maximum transponders per node (None = no limit).

Type:

int | None

single_core

Force single core allocation.

Type:

bool

topology_info

Physical topology info for SNR calculations.

Type:

dict[str, Any]

num_requests

Total requests to simulate.

Type:

int

erlang

Traffic intensity (arrival_rate * holding_time).

Type:

float

holding_time

Mean request duration.

Type:

float

route_method

Routing algorithm name.

Type:

str

k_paths

Number of candidate paths to compute.

Type:

int

allocation_method

Spectrum allocation strategy.

Type:

str

grooming_enabled

Enable traffic grooming.

Type:

bool

grooming_type

Grooming algorithm type (None if disabled).

Type:

str | None

slicing_enabled

Enable lightpath slicing.

Type:

bool

max_slices

Maximum slices per request.

Type:

int

snr_enabled

Enable SNR validation.

Type:

bool

snr_type

SNR calculation method or None.

Type:

str | None

snr_recheck

Re-validate SNR after allocation.

Type:

bool

recheck_adjacent_cores

Include adjacent cores in SNR recheck.

Type:

bool

recheck_crossband

Include crossband in SNR recheck.

Type:

bool

can_partially_serve

Allow partial bandwidth fulfillment.

Type:

bool

fixed_grid

True for fixed grid, False for flexi-grid.

Type:

bool

spectrum_priority

Band selection priority (“BSC”, “CSB”, or None).

Type:

str | None

multi_fiber

True for multi-fiber, False for multi-core fiber.

Type:

bool

dynamic_lps

True for dynamic lightpath slicing mode.

Type:

bool

protection_switchover_ms

Time to switch to backup path (ms).

Type:

float

restoration_latency_ms

Time to restore after failure (ms).

Type:

float

bw_per_slot

Bandwidth per slot in GHz.

Type:

float

input_power

Optical input power in Watts.

Type:

float

frequency_spacing

Channel spacing in Hz.

Type:

float

light_frequency

Center light frequency in Hz.

Type:

float

planck_constant

Planck’s constant in J*s.

Type:

float

noise_spectral_density

Noise spectral density.

Type:

float

mci_worst

Worst-case mutual coupling interference.

Type:

float

nsp_per_band

Noise spontaneous parameter per band.

Type:

dict[str, float]

request_bit_rate

Default request bit rate in Gb/s.

Type:

float

request_snr

Default requested SNR in dB.

Type:

float

snr_thresholds

SNR thresholds per modulation format.

Type:

dict[str, float]

phi

SNR phi parameter per modulation format.

Type:

dict[str, float]

egn_model

Use EGN model for SNR calculation.

Type:

bool

xt_type

Crosstalk type.

Type:

str | None

beta

SNR beta parameter.

Type:

float

theta

SNR theta parameter.

Type:

float

bi_directional

Bi-directional SNR calculation.

Type:

bool

xt_noise

Include crosstalk noise.

Type:

bool

requested_xt

Requested crosstalk per format.

Type:

dict[str, float]

modulation_formats

Available modulation format definitions.

Type:

dict[str, Any]

mod_per_bw

Modulation formats available per bandwidth.

Type:

dict[str, Any]

mod_format_map

Mapping of format ID to format name.

Type:

dict[int, str]

bandwidth_map

Mapping of format name to bandwidth capacity.

Type:

dict[str, int]

pre_calc_mod_selection

Use pre-calculated modulation selection.

Type:

bool

Example

>>> config = SimulationConfig.from_engine_props(engine_props)
>>> config.k_paths
3
>>> config.grooming_enabled
True
>>> legacy = config.to_engine_props()

network_name: str

cores_per_link: int

band_list: tuple[str, ...]

band_slots: dict[str, int]

guard_slots: int

num_requests: int

erlang: float

holding_time: float

route_method: str

k_paths: int

allocation_method: str

span_length: float = 100.0

max_link_length: float | None = None

max_span: int | None = None

max_transponders: int | None = None

single_core: bool = False

topology_info: dict[str, Any]

grooming_enabled: bool = False

grooming_type: str | None = None

slicing_enabled: bool = False

max_slices: int = 1

snr_enabled: bool = False

snr_type: str | None = None

snr_recheck: bool = False

recheck_adjacent_cores: bool = True

recheck_crossband: bool = True

can_partially_serve: bool = False

fixed_grid: bool = True

spectrum_priority: str | None = None

multi_fiber: bool = False

dynamic_lps: bool = False

protection_switchover_ms: float = 50.0

restoration_latency_ms: float = 100.0

bw_per_slot: float = 12.5

input_power: float = 0.001

frequency_spacing: float = 12500000000.0

light_frequency: float = 193410000000000.0

planck_constant: float = 6.62607004e-34

noise_spectral_density: float = 1.8

mci_worst: float = 6.334975555658596e-27

nsp_per_band: dict[str, float]

request_bit_rate: float = 12.5

request_snr: float = 8.5

snr_thresholds: dict[str, float]

phi: dict[str, float]

egn_model: bool = False

xt_type: str | None = None

beta: float = 0.5

theta: float = 0.0

bi_directional: bool = True

xt_noise: bool = False

requested_xt: dict[str, float]

modulation_formats: dict[str, Any]

mod_per_bw: dict[str, Any]

mod_format_map: dict[int, str]

bandwidth_map: dict[str, int]

pre_calc_mod_selection: bool = False

property total_slots: int

Total spectrum slots across all bands.

property arrival_rate: float

Computed arrival rate from erlang and holding_time.

property is_multiband: bool

True if using multiple frequency bands.

property is_multicore: bool

True if using multiple cores per fiber.

property protection_enabled: bool

True if protection routing is enabled.

classmethod from_engine_props(engine_props)

Create SimulationConfig from legacy engine_props dictionary.

Parameters:

engine_props (dict[str, Any]) – Legacy configuration dictionary.

Returns:

New SimulationConfig instance.

Return type:

SimulationConfig

Raises:

• KeyError – If required fields are missing.

• ValueError – If field values are invalid.

to_engine_props()

Convert to legacy engine_props dictionary format.

Returns:

Dictionary compatible with legacy engine_props consumers.

Return type:

dict[str, Any]

__init__(network_name, cores_per_link, band_list, band_slots, guard_slots, num_requests, erlang, holding_time, route_method, k_paths, allocation_method, span_length=100.0, max_link_length=None, max_span=None, max_transponders=None, single_core=False, topology_info=<factory>, grooming_enabled=False, grooming_type=None, slicing_enabled=False, max_slices=1, snr_enabled=False, snr_type=None, snr_recheck=False, recheck_adjacent_cores=True, recheck_crossband=True, can_partially_serve=False, fixed_grid=True, spectrum_priority=None, multi_fiber=False, dynamic_lps=False, protection_switchover_ms=50.0, restoration_latency_ms=100.0, bw_per_slot=12.5, input_power=0.001, frequency_spacing=12500000000.0, light_frequency=193410000000000.0, planck_constant=6.62607004e-34, noise_spectral_density=1.8, mci_worst=6.334975555658596e-27, nsp_per_band=<factory>, request_bit_rate=12.5, request_snr=8.5, snr_thresholds=<factory>, phi=<factory>, egn_model=False, xt_type=None, beta=0.5, theta=0.0, bi_directional=True, xt_noise=False, requested_xt=<factory>, modulation_formats=<factory>, mod_per_bw=<factory>, mod_format_map=<factory>, bandwidth_map=<factory>, pre_calc_mod_selection=False)

Parameters:

• network_name (str) –

• cores_per_link (int) –

• band_list (tuple[str, ...]) –

• band_slots (dict[str, int]) –

• guard_slots (int) –

• num_requests (int) –

• erlang (float) –

• holding_time (float) –

• route_method (str) –

• k_paths (int) –

• allocation_method (str) –

• span_length (float) –

• max_link_length (float | None) –

• max_span (int | None) –

• max_transponders (int | None) –

• single_core (bool) –

• topology_info (dict[str, Any]) –

• grooming_enabled (bool) –

• grooming_type (str | None) –

• slicing_enabled (bool) –

• max_slices (int) –

• snr_enabled (bool) –

• snr_type (str | None) –

• snr_recheck (bool) –

• recheck_adjacent_cores (bool) –

• recheck_crossband (bool) –

• can_partially_serve (bool) –

• fixed_grid (bool) –

• spectrum_priority (str | None) –

• multi_fiber (bool) –

• dynamic_lps (bool) –

• protection_switchover_ms (float) –

• restoration_latency_ms (float) –

• bw_per_slot (float) –

• input_power (float) –

• frequency_spacing (float) –

• light_frequency (float) –

• planck_constant (float) –

• noise_spectral_density (float) –

• mci_worst (float) –

• nsp_per_band (dict[str, float]) –

• request_bit_rate (float) –

• request_snr (float) –

• snr_thresholds (dict[str, float]) –

• phi (dict[str, float]) –

• egn_model (bool) –

• xt_type (str | None) –

• beta (float) –

• theta (float) –

• bi_directional (bool) –

• xt_noise (bool) –

• requested_xt (dict[str, float]) –

• modulation_formats (dict[str, Any]) –

• mod_per_bw (dict[str, Any]) –

• mod_format_map (dict[int, str]) –

• bandwidth_map (dict[str, int]) –

• pre_calc_mod_selection (bool) –

Return type:

None

Request domain model with lifecycle tracking.

This module defines: - RequestType: Enum for request event types (arrival/release) - RequestStatus: Enum for request lifecycle states - BlockReason: Enum for allocation failure reasons - Request: Mutable dataclass for network service requests

All classes support legacy conversion via from_legacy_dict/to_legacy_dict methods.

class fusion.domain.request.RequestType

Bases: Enum

Type of request event in the simulation.

ARRIVAL indicates a request arriving that needs resource allocation. RELEASE indicates a request departing where resources should be freed.

ARRIVAL = 'arrival'

RELEASE = 'release'

classmethod from_legacy(value)

Convert legacy string to enum.

Parameters:

value (str) – Legacy string “arrival” or “release”.

Returns:

Corresponding RequestType enum value.

Return type:

RequestType

to_legacy()

Convert enum to legacy string.

Returns:

String value compatible with legacy code.

Return type:

str

class fusion.domain.request.RequestStatus

Bases: Enum

Request lifecycle status.

Initial state is PENDING. Processing states are ROUTING, SPECTRUM_SEARCH, SNR_CHECK. Terminal success states are ALLOCATED, GROOMED, PARTIALLY_GROOMED. Terminal failure state is BLOCKED. Release states are RELEASING and RELEASED.

State Machine:

PENDING --> ROUTING --> SPECTRUM_SEARCH --> SNR_CHECK
                |              |               |
                v              v               v
            BLOCKED        BLOCKED         BLOCKED
                |              |               |
                +--------------+---------------+
                               |
                               v
            +------------------+------------------+
            |                  |                  |
            v                  v                  v
        ALLOCATED          GROOMED      PARTIALLY_GROOMED
            |                  |                  |
            +------------------+------------------+
                               |
                               v
                           RELEASING
                               |
                               v
                           RELEASED

PENDING = 1

Request created, not yet processed.

ROUTING = 2

Finding candidate paths.

SPECTRUM_SEARCH = 3

Searching for available spectrum.

SNR_CHECK = 4

Validating signal-to-noise ratio.

ALLOCATED = 5

Successfully allocated new resources.

GROOMED = 6

Successfully groomed onto existing lightpath.

PARTIALLY_GROOMED = 7

Partially groomed, rest allocated on new lightpath.

BLOCKED = 8

Could not allocate resources (see block_reason).

RELEASING = 9

Release in progress.

RELEASED = 10

Resources freed (terminal).

is_terminal()

Check if this is a terminal state.

Return type:

bool

is_success()

Check if this represents successful allocation.

Return type:

bool

is_processing()

Check if request is currently being processed.

Return type:

bool

can_transition_to(target)

Check if transition to target state is valid.

Parameters:

target (RequestStatus) – The state to transition to.

Returns:

True if the transition is valid, False otherwise.

Return type:

bool

class fusion.domain.request.ProtectionStatus

Bases: Enum

Protection state for 1+1 protected connections.

States include UNPROTECTED (normal request), ESTABLISHING (setting up paths), ACTIVE_PRIMARY (using primary), ACTIVE_BACKUP (using backup after switchover), SWITCHOVER_IN_PROGRESS, PRIMARY_FAILED (on backup), BACKUP_FAILED (on primary), and BOTH_FAILED (connection lost).

UNPROTECTED = 1

No protection (normal request).

ESTABLISHING = 2

Setting up protection paths.

ACTIVE_PRIMARY = 3

Protected, using primary path.

ACTIVE_BACKUP = 4

Protected, using backup path (after switchover).

SWITCHOVER_IN_PROGRESS = 5

Switching between paths.

PRIMARY_FAILED = 6

Primary path failed, on backup.

BACKUP_FAILED = 7

Backup path failed, on primary.

BOTH_FAILED = 8

Both paths failed (connection lost).

is_active()

Check if connection is active (either path).

Return type:

bool

is_degraded()

Check if protection is degraded (one path failed).

Return type:

bool

is_failed()

Check if connection is failed.

Return type:

bool

is_protected()

Check if request has protection enabled.

Return type:

bool

classmethod from_legacy(is_protected, active_path)

Convert legacy protection fields to enum.

Parameters:

• is_protected (bool) – Legacy is_protected flag.

• active_path (str | None) – Legacy active_path string (“primary” or “backup”).

Returns:

Corresponding ProtectionStatus enum value.

Return type:

ProtectionStatus

to_legacy_active_path()

Convert to legacy active_path string.

Returns:

“primary” or “backup” based on current state.

Return type:

str

class fusion.domain.request.BlockReason

Bases: Enum

Reasons for request blocking.

Path-related reasons are NO_PATH and DISTANCE. Spectrum-related is CONGESTION. Quality-related are SNR_THRESHOLD and XT_THRESHOLD. Feature-specific are GROOMING_FAIL, SLICING_FAIL, and PROTECTION_FAIL. Failure-related are LINK_FAILURE, NODE_FAILURE, and FAILURE. Resource limits are TRANSPONDER_LIMIT and MAX_SEGMENTS.

NO_PATH = 'no_path'

No path exists between source and destination.

DISTANCE = 'distance'

Path too long for any modulation format.

CONGESTION = 'congestion'

No spectrum available on any path.

SNR_THRESHOLD = 'snr_fail'

SNR below required threshold.

SNR_RECHECK_FAIL = 'snr_recheck_fail'

SNR recheck failed for existing lightpaths after new allocation (congestion).

XT_THRESHOLD = 'xt_threshold'

Crosstalk exceeds allowed threshold.

GROOMING_FAIL = 'grooming_fail'

Grooming attempt failed.

SLICING_FAIL = 'slicing_fail'

Slicing attempt failed.

PROTECTION_FAIL = 'protection_fail'

Cannot establish protection path.

LINK_FAILURE = 'link_failure'

Link failed during processing.

NODE_FAILURE = 'node_failure'

Node failed during processing.

FAILURE = 'failure'

Generic failure (legacy).

TRANSPONDER_LIMIT = 'transponder_limit'

Max transponders reached.

MAX_SEGMENTS = 'max_segments'

Max slicing segments reached.

classmethod from_legacy_string(reason)

Convert legacy block reason string to enum.

Parameters:

reason (str | None) – Legacy string like “distance”, “congestion”, etc.

Returns:

Corresponding BlockReason or None if reason is None/empty.

Return type:

BlockReason | None

to_legacy_string()

Convert to legacy block reason string.

Returns:

String value compatible with legacy code.

Return type:

str

is_path_related()

Check if blocking is path-related.

Return type:

bool

is_spectrum_related()

Check if blocking is spectrum-related.

Return type:

bool

is_quality_related()

Check if blocking is quality-related.

Return type:

bool

is_feature_related()

Check if blocking is feature-related.

Return type:

bool

is_failure_related()

Check if blocking is failure-related.

Return type:

bool

is_resource_limit()

Check if blocking is due to resource limits.

Return type:

bool

class fusion.domain.request.Request

Bases: object

Network service request with lifecycle tracking.

A Request represents a bandwidth demand between two network nodes. It tracks the full lifecycle from arrival through allocation to release.

request_id

Unique identifier (immutable after creation).

Type:

int

source

Source node ID (immutable after creation).

Type:

str

destination

Destination node ID (immutable after creation).

Type:

str

bandwidth_gbps

Requested bandwidth in Gbps (immutable after creation).

Type:

int

arrive_time

Simulation time of arrival (immutable after creation).

Type:

float

depart_time

Simulation time of departure (immutable after creation).

Type:

float

modulation_formats

Available modulation formats for this request.

Type:

dict[str, dict[str, Any]]

status

Current lifecycle state (PENDING -> ROUTED/BLOCKED -> RELEASED).

Type:

fusion.domain.request.RequestStatus

lightpath_ids

IDs of lightpaths serving this request.

Type:

list[int]

block_reason

Reason for blocking (if status == BLOCKED).

Type:

fusion.domain.request.BlockReason | None

is_groomed

Fully served by existing lightpath capacity.

Type:

bool

is_partially_groomed

Partially served by existing lightpath.

Type:

bool

is_sliced

Split across multiple lightpaths.

Type:

bool

protection_status

Protection state for 1+1 protected connections.

Type:

fusion.domain.request.ProtectionStatus

primary_path

Primary path for protected requests.

Type:

list[str] | None

backup_path

Backup path for protected requests.

Type:

list[str] | None

last_switchover_time

Time of last protection switchover.

Type:

float | None

Example

>>> request = Request(
...     request_id=42,
...     source="0",
...     destination="5",
...     bandwidth_gbps=100,
...     arrive_time=12.345,
...     depart_time=17.890,
... )
>>> request.status
<RequestStatus.PENDING: 1>
>>> request.is_arrival
True

request_id: int

source: str

destination: str

bandwidth_gbps: int

arrive_time: float

depart_time: float

modulation_formats: dict[str, dict[str, Any]]

status: RequestStatus = 1

lightpath_ids: list[int]

block_reason: BlockReason | None = None

__init__(request_id, source, destination, bandwidth_gbps, arrive_time, depart_time, modulation_formats=<factory>, status=RequestStatus.PENDING, lightpath_ids=<factory>, block_reason=None, is_groomed=False, is_partially_groomed=False, is_sliced=False, protection_status=ProtectionStatus.UNPROTECTED, primary_path=None, backup_path=None, last_switchover_time=None)

Parameters:

• request_id (int) –

• source (str) –

• destination (str) –

• bandwidth_gbps (int) –

• arrive_time (float) –

• depart_time (float) –

• modulation_formats (dict[str, dict[str, Any]]) –

• status (RequestStatus) –

• lightpath_ids (list[int]) –

• block_reason (BlockReason | None) –

• is_groomed (bool) –

• is_partially_groomed (bool) –

• is_sliced (bool) –

• protection_status (ProtectionStatus) –

• primary_path (list[str] | None) –

• backup_path (list[str] | None) –

• last_switchover_time (float | None) –

Return type:

None

is_groomed: bool = False

is_partially_groomed: bool = False

is_sliced: bool = False

protection_status: ProtectionStatus = 1

primary_path: list[str] | None = None

backup_path: list[str] | None = None

last_switchover_time: float | None = None

property is_arrival: bool

True if request is pending processing (arrival event).

property is_successful: bool

True if request was successfully allocated/groomed.

property is_blocked: bool

True if request was blocked.

property is_released: bool

True if request has been released.

property is_processing: bool

True if request is currently being processed.

property is_terminal: bool

True if request is in a terminal state.

property endpoint_key: tuple[str, str]

Canonical endpoint pair for lightpath matching.

Returns:

Sorted tuple to ensure (A, B) == (B, A) for endpoint matching.

Return type:

tuple[str, str]

property holding_time: float

Duration of the request in simulation time units.

property num_lightpaths: int

Number of lightpaths serving this request.

property is_protected: bool

True if request has protection enabled.

property active_path: str

Current active path (‘primary’ or ‘backup’) for protected requests.

property is_protection_degraded: bool

True if protection is degraded (one path failed).

property is_protection_failed: bool

True if both protection paths have failed.

mark_allocated(lightpath_ids)

Mark request as successfully allocated to new lightpath(s).

Parameters:

lightpath_ids (list[int]) – IDs of lightpaths allocated for this request.

Raises:

ValueError – If transition is invalid or no lightpaths provided.

Return type:

None

mark_groomed(lightpath_ids)

Mark request as successfully groomed onto existing lightpath(s).

Parameters:

lightpath_ids (list[int]) – IDs of lightpaths used for grooming.

Raises:

ValueError – If transition is invalid or no lightpaths provided.

Return type:

None

mark_partially_groomed(lightpath_ids)

Mark request as partially groomed (some traffic on existing LP, rest new).

Parameters:

lightpath_ids (list[int]) – IDs of all lightpaths (existing and new).

Raises:

ValueError – If transition is invalid or no lightpaths provided.

Return type:

None

mark_routed(lightpath_ids)

Mark request as successfully routed (alias for mark_allocated).

This method provides backwards compatibility with code that uses the simplified PENDING -> ROUTED -> RELEASED flow.

Parameters:

lightpath_ids (list[int]) – IDs of lightpaths allocated for this request.

Raises:

ValueError – If transition is invalid or no lightpaths provided.

Return type:

None

mark_blocked(reason)

Mark request as blocked.

Parameters:

reason (BlockReason) – Reason for blocking.

Raises:

ValueError – If transition is invalid.

Return type:

None

mark_releasing()

Mark request as releasing (resources being freed).

Raises:

ValueError – If transition is invalid.

Return type:

None

mark_released()

Mark request as released (resources freed).

If currently in a success state, transitions through RELEASING first.

Raises:

ValueError – If transition is invalid.

Return type:

None

set_status(status)

Set status with transition validation.

Use this for processing state transitions (ROUTING, SPECTRUM_SEARCH, etc.).

Parameters:

status (RequestStatus) – The target status.

Raises:

ValueError – If transition is invalid.

Return type:

None

classmethod from_legacy_dict(time_key, request_dict, request_id=None)

Create Request from legacy request dictionary.

The legacy format stores requests in a dict indexed by (request_id, time). Each entry contains fields like “req_id”, “source”, “bandwidth”, etc.

Parameters:

• time_key (tuple[int, float]) – Tuple of (request_id, time) used as dict key.

• request_dict (dict[str, Any]) – Legacy request dictionary with fields req_id (int), source (str), destination (str), arrive (float), depart (float), bandwidth (str like “50Gbps”), mod_formats (dict, optional), request_type (str “arrival”/”release”, optional).

• request_id (int | None) – Override request_id (default: from time_key[0] or req_id).

Returns:

New Request instance in PENDING state.

Return type:

Request

Example

>>> legacy = {
...     "req_id": 42,
...     "source": "0",
...     "destination": "5",
...     "arrive": 12.345,
...     "depart": 17.890,
...     "bandwidth": "100Gbps",
... }
>>> request = Request.from_legacy_dict((42, 12.345), legacy)
>>> request.request_id
42
>>> request.bandwidth_gbps
100

to_legacy_dict()

Convert to legacy request dictionary format.

This enables interoperability with legacy code that expects request dictionaries during the migration period.

Returns:

Dictionary compatible with legacy request consumers with keys req_id (int), source (str), destination (str), arrive (float), depart (float), bandwidth (str like “100Gbps”), mod_formats (dict), request_type (str “arrival”/”release”).

Return type:

dict[str, Any]

Example

>>> request = Request(
...     request_id=42,
...     source="0",
...     destination="5",
...     bandwidth_gbps=100,
...     arrive_time=12.345,
...     depart_time=17.890,
... )
>>> legacy = request.to_legacy_dict()
>>> legacy["bandwidth"]
'100Gbps'

to_legacy_time_key()

Generate legacy dictionary key for this request.

Returns:

Tuple of (request_id, time) where time depends on status: PENDING/ROUTED/BLOCKED uses arrive_time, RELEASED uses depart_time.

Return type:

tuple[int, float]

Lightpath domain model with capacity management.

This module defines the Lightpath dataclass representing an allocated optical path with spectrum assignment and capacity tracking.

class fusion.domain.lightpath.Lightpath

Bases: object

Allocated optical path with capacity management.

A Lightpath represents an established connection through the network with allocated spectrum resources. It tracks capacity usage for traffic grooming and supports 1+1 protection.

lightpath_id

Unique identifier.

Type:

int

path

Ordered list of node IDs from source to destination.

Type:

list[str]

start_slot

First allocated slot index (inclusive).

Type:

int

end_slot

Last allocated slot index (exclusive).

Type:

int

core

Core number for MCF (0-indexed).

Type:

int

band

Frequency band (“c”, “l”, “s”).

Type:

str

modulation

Modulation format (“QPSK”, “16-QAM”, etc.).

Type:

str

total_bandwidth_gbps

Maximum capacity.

Type:

int

remaining_bandwidth_gbps

Available for new requests.

Type:

int

path_weight_km

Physical path length in km.

Type:

float

request_allocations

Maps request_id to allocated bandwidth.

Type:

dict[int, int]

time_bw_usage

Timestamp to utilization mapping for tracking.

Type:

dict[float, float]

snr_db

Signal-to-noise ratio (optional).

Type:

float | None

xt_cost

Crosstalk cost (optional).

Type:

float | None

is_degraded

True if quality has degraded.

Type:

bool

connection_index

External routing index for pre-calculated SNR lookup.

Type:

int | None

backup_path

Disjoint backup path (optional).

Type:

list[str] | None

backup_start_slot

Backup spectrum start slot.

Type:

int | None

backup_end_slot

Backup spectrum end slot.

Type:

int | None

backup_core

Backup core number.

Type:

int | None

backup_band

Backup frequency band.

Type:

str | None

is_protected

Has backup path.

Type:

bool

active_path

Currently active path (“primary” or “backup”).

Type:

str

protection_lp_id

For working LPs, ID of backup LP.

Type:

int | None

working_lp_id

For backup LPs, ID of working LP.

Type:

int | None

Example

>>> lp = Lightpath(
...     lightpath_id=1,
...     path=["0", "2", "5"],
...     start_slot=10,
...     end_slot=18,
...     core=0,
...     band="c",
...     modulation="QPSK",
...     total_bandwidth_gbps=100,
...     remaining_bandwidth_gbps=100,
... )
>>> lp.allocate_bandwidth(42, 50)
True
>>> lp.utilization
0.5

lightpath_id: int

path: list[str]

start_slot: int

end_slot: int

core: int

band: str

modulation: str

total_bandwidth_gbps: int

remaining_bandwidth_gbps: int

path_weight_km: float = 0.0

request_allocations: dict[int, int]

time_bw_usage: dict[float, float]

snr_db: float | None = None

xt_cost: float | None = None

is_degraded: bool = False

connection_index: int | None = None

backup_path: list[str] | None = None

backup_start_slot: int | None = None

backup_end_slot: int | None = None

backup_core: int | None = None

backup_band: str | None = None

is_protected: bool = False

active_path: str = 'primary'

protection_lp_id: int | None = None

working_lp_id: int | None = None

property source: str

Source node ID.

property destination: str

Destination node ID.

property endpoint_key: tuple[str, str]

Canonical endpoint pair for lightpath matching.

Returns:

Sorted tuple to ensure bidirectional matching.

Return type:

tuple[str, str]

property num_slots: int

Number of spectrum slots used.

property num_hops: int

Number of links in the path.

property utilization: float

Bandwidth utilization ratio.

Returns:

Float between 0.0 and 1.0 representing capacity usage.

Return type:

float

property has_capacity: bool

True if lightpath can accept more traffic.

property num_requests: int

Number of requests using this lightpath.

property is_empty: bool

True if no requests are using this lightpath.

can_accommodate(bandwidth_gbps)

Check if lightpath can accommodate the requested bandwidth.

Parameters:

bandwidth_gbps (int) – Bandwidth to check.

Returns:

True if sufficient capacity available.

Return type:

bool

allocate_bandwidth(request_id, bandwidth_gbps, timestamp=None)

Allocate bandwidth to a request.

Parameters:

• request_id (int) – ID of the request.

• bandwidth_gbps (int) – Bandwidth to allocate.

• timestamp (float | None) – Optional arrival time for utilization tracking.

Returns:

True if allocation successful, False if insufficient capacity.

Return type:

bool

Raises:

ValueError – If request_id already has an allocation or bandwidth <= 0.

release_bandwidth(request_id, timestamp=None)

Release bandwidth from a request.

Parameters:

• request_id (int) – ID of the request to release.

• timestamp (float | None) – Optional departure time for utilization tracking.

Returns:

Amount of bandwidth released.

Return type:

int

Raises:

KeyError – If request_id has no allocation.

get_allocation(request_id)

Get bandwidth allocated to a request.

Parameters:

request_id (int) – ID of the request.

Returns:

Allocated bandwidth or None if no allocation.

Return type:

int | None

switch_to_backup()

Switch to backup path (for failure recovery).

Returns:

True if switch successful, False if not protected.

Return type:

bool

Raises:

ValueError – If already on backup path.

switch_to_primary()

Switch back to primary path (after recovery).

Returns:

True if switch successful, False if not protected.

Return type:

bool

Raises:

ValueError – If already on primary path.

property current_path: list[str]

Get currently active path.

classmethod from_legacy_dict(lightpath_id, lp_info)

Create Lightpath from legacy dictionary.

The legacy format stores lightpath info in nested dictionaries indexed by endpoint pair and lightpath ID.

Parameters:

• lightpath_id (int) – Unique lightpath identifier.

• lp_info (dict[str, Any]) – Legacy lightpath info dictionary with fields: path (list[str]): Node IDs in path, start_slot (int): First slot index, end_slot (int): Last slot index (exclusive), core (int): Core number, band (str): Frequency band, mod_format (str): Modulation format, lightpath_bandwidth (float): Total bandwidth, remaining_bandwidth (float): Available bandwidth, requests_dict (dict): Request allocations (optional), snr_cost (float): SNR value (optional), xt_cost (float): Crosstalk (optional), path_weight (float): Path length (optional), is_degraded (bool): Degradation flag (optional), backup_path (list[str]): Backup path (optional), is_protected (bool): Protection flag (optional), active_path (str): Active path (optional).

Returns:

New Lightpath instance.

Return type:

Lightpath

Example

>>> lp_info = {
...     "path": ["0", "2", "5"],
...     "start_slot": 10,
...     "end_slot": 18,
...     "core": 0,
...     "band": "c",
...     "mod_format": "QPSK",
...     "lightpath_bandwidth": 100.0,
...     "remaining_bandwidth": 50.0,
...     "requests_dict": {42: 50.0},
... }
>>> lp = Lightpath.from_legacy_dict(1, lp_info)
>>> lp.lightpath_id
1

to_legacy_dict()

Convert to legacy dictionary format.

This enables interoperability with legacy code that expects lightpath info dictionaries during the migration period.

Returns:

Dictionary compatible with legacy lightpath consumers with keys: path (list[str]), start_slot (int), end_slot (int), core (int), band (str), mod_format (str), lightpath_bandwidth (float), remaining_bandwidth (float), requests_dict (dict[int, float]), snr_cost (float | None), xt_cost (float | None), path_weight (float), is_degraded (bool), backup_path (list[str] | None), is_protected (bool), active_path (str).

Return type:

dict[str, Any]

Example

>>> lp = Lightpath(lightpath_id=1, path=["0", "5"], ...)
>>> legacy = lp.to_legacy_dict()
>>> legacy["mod_format"]
'QPSK'

to_legacy_key()

Generate legacy dictionary key (endpoint pair).

Returns:

Sorted tuple of (source, destination) for dict indexing.

Return type:

tuple[str, str]

__init__(lightpath_id, path, start_slot, end_slot, core, band, modulation, total_bandwidth_gbps, remaining_bandwidth_gbps, path_weight_km=0.0, request_allocations=<factory>, time_bw_usage=<factory>, snr_db=None, xt_cost=None, is_degraded=False, connection_index=None, backup_path=None, backup_start_slot=None, backup_end_slot=None, backup_core=None, backup_band=None, is_protected=False, active_path='primary', protection_lp_id=None, working_lp_id=None)

Parameters:

• lightpath_id (int) –

• path (list[str]) –

• start_slot (int) –

• end_slot (int) –

• core (int) –

• band (str) –

• modulation (str) –

• total_bandwidth_gbps (int) –

• remaining_bandwidth_gbps (int) –

• path_weight_km (float) –

• request_allocations (dict[int, int]) –

• time_bw_usage (dict[float, float]) –

• snr_db (float | None) –

• xt_cost (float | None) –

• is_degraded (bool) –

• connection_index (int | None) –

• backup_path (list[str] | None) –

• backup_start_slot (int | None) –

• backup_end_slot (int | None) –

• backup_core (int | None) –

• backup_band (str | None) –

• is_protected (bool) –

• active_path (str) –

• protection_lp_id (int | None) –

• working_lp_id (int | None) –

Return type:

None

Network state management for FUSION simulation.

This module provides: - LinkSpectrum: Per-link spectrum state management - NetworkState: Single source of truth for network state during simulation

Design Principles:

• NetworkState is a state container, NOT a routing algorithm

• One NetworkState instance per simulation

• Passed by reference to pipelines (not stored)

• All state mutations through explicit methods

• Read methods are query-only (no side effects)

Spectrum Array Values:

• 0: Slot is free

• Positive int: Slot occupied by lightpath with that ID

• Negative int: Guard band slot for lightpath with abs(value) ID

class fusion.domain.network_state.LinkSpectrum

Bases: object

Per-link spectrum management.

Encapsulates the spectrum allocation state for a single network link, supporting multi-band and multi-core configurations.

Spectrum array values: 0 = free, positive int = lightpath ID, negative int = guard band for lightpath with abs(value) ID.

link

The link identifier as (source, destination) tuple.

Type:

tuple[str, str]

cores_matrix

Spectrum arrays per band, shape (cores, slots) each.

Type:

dict[str, numpy.ndarray[Any, numpy.dtype[numpy.int64]]]

usage_count

Number of active spectrum allocations on this link.

Type:

int

throughput

Cumulative bandwidth served through this link (Gbps).

Type:

float

link_num

Link index for reference.

Type:

int

length_km

Physical link length in kilometers.

Type:

float

Example

>>> spectrum = LinkSpectrum.from_config(("A", "B"), config, link_num=0)
>>> spectrum.is_range_free(0, 10, core=0, band="c")
True
>>> spectrum.allocate_range(0, 10, core=0, band="c", lightpath_id=1)
>>> spectrum.is_range_free(0, 10, core=0, band="c")
False

link: tuple[str, str]

cores_matrix: dict[str, ndarray[Any, dtype[int64]]]

usage_count: int = 0

throughput: float = 0.0

link_num: int = 0

length_km: float = 0.0

FREE_SLOT: ClassVar[int] = 0

is_range_free(start_slot, end_slot, core, band)

Check if a spectrum range is completely free.

Parameters:

• start_slot (int) – First slot index (inclusive).

• end_slot (int) – Last slot index (exclusive).

• core (int) – Core index (0 to cores_per_link - 1).

• band (str) – Band identifier (“c”, “l”, “s”).

Returns:

True if all slots in range are 0 (free), False otherwise.

Return type:

bool

Raises:

• KeyError – If band is not in cores_matrix.

• IndexError – If core or slot indices are out of bounds.

allocate_range(start_slot, end_slot, core, band, lightpath_id, guard_slots=0)

Allocate a spectrum range to a lightpath.

Marks data slots with positive lightpath_id and guard slots with negative. This follows the legacy convention where guard bands use negative IDs. Increments usage_count on successful allocation.

Parameters:

• start_slot (int) – First slot index (inclusive).

• end_slot (int) – Last slot index (exclusive), includes guard slots.

• core (int) – Core index.

• band (str) – Band identifier.

• lightpath_id (int) – Positive integer ID for the lightpath.

• guard_slots (int) – Number of guard band slots at end of allocation.

Raises:

• ValueError – If lightpath_id <= 0 or range is not free.

• KeyError – If band is not in cores_matrix.

• IndexError – If indices are out of bounds.

Return type:

None

release_range(start_slot, end_slot, core, band)

Release a spectrum range (set to free).

Sets spectrum slots to 0 and decrements usage_count (clamped to >= 0).

Parameters:

• start_slot (int) – First slot index (inclusive).

• end_slot (int) – Last slot index (exclusive).

• core (int) – Core index.

• band (str) – Band identifier.

Raises:

• KeyError – If band is not in cores_matrix.

• IndexError – If indices are out of bounds.

Return type:

None

release_by_lightpath_id(lightpath_id, band, core)

Release all slots allocated to a specific lightpath.

Finds and clears both data slots (positive ID) and guard slots (negative ID). Decrements usage_count on successful release.

Parameters:

• lightpath_id (int) – The lightpath ID to release.

• band (str) – Band identifier.

• core (int) – Core index.

Returns:

Tuple of (start_slot, end_slot) if found, None if not found.

Return type:

tuple[int, int] | None

get_spectrum_array(band)

Get spectrum array for a band.

Returns actual array reference. Callers should NOT modify directly. Use allocate_range() and release_range() instead.

Parameters:

band (str) – Band identifier.

Returns:

numpy array of shape (cores, slots), values are lightpath IDs or 0.

Return type:

npt.NDArray[np.int64]

get_slot_count(band)

Get number of slots in a band.

Parameters:

band (str) –

Return type:

int

get_core_count()

Get number of cores (same across all bands).

Return type:

int

get_bands()

Get list of available bands.

Return type:

list[str]

get_free_slot_count(band, core)

Count free slots on a specific core and band.

Parameters:

• band (str) – Band identifier.

• core (int) – Core index.

Returns:

Number of free (value == 0) slots.

Return type:

int

get_fragmentation_ratio(band, core)

Calculate fragmentation ratio for a core/band.

Fragmentation = 1 - (largest_contiguous_free / total_free). Value of 0 means no fragmentation, 1 means maximum fragmentation.

Parameters:

• band (str) – Band identifier.

• core (int) – Core index.

Returns:

Fragmentation ratio between 0.0 and 1.0.

Return type:

float

classmethod from_config(link, config, link_num=0, length_km=0.0)

Create LinkSpectrum initialized from SimulationConfig.

Parameters:

• link (tuple[str, str]) – Link identifier tuple.

• config (SimulationConfig) – Simulation configuration with band/core specs.

• link_num (int) – Link index.

• length_km (float) – Physical link length.

Returns:

New LinkSpectrum with zero-initialized spectrum arrays matching config.band_list and config.cores_per_link.

Return type:

LinkSpectrum

to_legacy_dict()

Convert to legacy network_spectrum_dict entry format.

Returns:

Dictionary compatible with legacy sdn_props.network_spectrum_dict[link].

Return type:

dict[str, Any]

__init__(link, cores_matrix=<factory>, usage_count=0, throughput=0.0, link_num=0, length_km=0.0)

Parameters:

• link (tuple[str, str]) –

• cores_matrix (dict[str, ndarray[Any, dtype[int64]]]) –

• usage_count (int) –

• throughput (float) –

• link_num (int) –

• length_km (float) –

Return type:

None

class fusion.domain.network_state.NetworkState

Bases: object

Single source of truth for network state during simulation.

Owns network topology (read-only after init), per-link spectrum state (LinkSpectrum objects), active lightpaths (by ID), and lightpath ID counter.

One instance per simulation, passed by reference to pipelines. All state mutations through explicit methods. Read methods are query-only (no side effects). This is a state container, NOT a routing algorithm.

Note

Request queue, request status tracking, statistics, and per-request temporary lists are managed elsewhere (SimulationEngine, SimStats, SDNController).

Example

>>> state = NetworkState(topology, config)
>>> if state.is_spectrum_available(path, 0, 10, core=0, band="c"):
...     # Spectrum is free, can allocate
...     pass

__init__(topology, config)

Initialize NetworkState from topology and configuration.

Topology is stored as reference (not copied). Callers must not modify topology after passing to NetworkState.

Parameters:

• topology (nx.Graph) – NetworkX graph representing network structure. Edge attributes are preserved (length, etc.).

• config (SimulationConfig) – Immutable simulation configuration.

Raises:

ValueError – If topology is empty (no edges) or config band_list is empty.

Return type:

None

property topology: Graph

Network topology graph (read-only reference).

property config: SimulationConfig

Simulation configuration (immutable).

property lightpath_count: int

Number of active lightpaths.

property link_count: int

Number of links in topology (edges, not bidirectional pairs).

property next_lightpath_id: int

Next lightpath ID that will be assigned (read-only).

property node_count: int

Number of nodes in topology.

get_lightpath(lightpath_id)

Retrieve a lightpath by ID.

Returns direct reference. Callers may read Lightpath state but should use NetworkState methods to modify it.

Parameters:

lightpath_id (int) – Unique lightpath identifier.

Returns:

Lightpath object if found, None otherwise.

Return type:

Lightpath | None

get_lightpaths_between(source, destination)

Get all lightpaths between two endpoints.

Handles both (src, dst) and (dst, src) automatically.

Parameters:

• source (str) – Source node ID.

• destination (str) – Destination node ID.

Returns:

List of lightpaths connecting these endpoints (either direction). Empty list if none found.

Return type:

list[Lightpath]

get_lightpaths_with_capacity(source, destination, min_bandwidth_gbps=1)

Get lightpaths between endpoints with available capacity.

Useful for grooming: find lightpaths that can accept more traffic.

Parameters:

• source (str) – Source node ID.

• destination (str) – Destination node ID.

• min_bandwidth_gbps (int) – Minimum remaining bandwidth required.

Returns:

List of lightpaths with at least min_bandwidth_gbps available. Sorted by remaining bandwidth descending (best candidates first).

Return type:

list[Lightpath]

get_lightpaths_on_link(link)

Get all lightpaths that traverse a specific link.

Useful for failure impact analysis.

Parameters:

link (tuple[str, str]) – (source, destination) tuple.

Returns:

List of lightpaths that include this link in their path.

Return type:

list[Lightpath]

iter_lightpaths()

Iterate over all active lightpaths.

Safe to iterate during simulation. Do not add/remove lightpaths during iteration.

Yields:

Each active Lightpath in arbitrary order.

Return type:

Iterator[Lightpath]

has_lightpath(lightpath_id)

Check if a lightpath exists.

Parameters:

lightpath_id (int) –

Return type:

bool

get_link_spectrum(link)

Get LinkSpectrum for a specific link.

Both (u, v) and (v, u) return the same LinkSpectrum object.

Parameters:

link (tuple[str, str]) – (source, destination) tuple.

Returns:

LinkSpectrum object for that link.

Return type:

LinkSpectrum

Raises:

KeyError – If link does not exist in topology.

is_spectrum_available(path, start_slot, end_slot, core, band)

Check if spectrum range is free along entire path.

Parameters:

• path (list[str]) – Ordered list of node IDs forming the path.

• start_slot (int) – First slot index (inclusive).

• end_slot (int) – Last slot index (exclusive).

• core (int) – Core index.

• band (str) – Band identifier.

Returns:

True if all slots in range are free on all links in path.

Return type:

bool

Raises:

• ValueError – If path has fewer than 2 nodes.

• KeyError – If any link in path doesn’t exist.

find_first_fit(path, slots_needed, core, band)

Find first available slot range along path (first-fit algorithm).

This is a read-only query method. It does NOT allocate. More sophisticated algorithms should be in pipelines.

Parameters:

• path (list[str]) – Ordered list of node IDs.

• slots_needed (int) – Number of contiguous slots required.

• core (int) – Core index to search.

• band (str) – Band identifier.

Returns:

Start slot index if found, None if no fit available.

Return type:

int | None

find_last_fit(path, slots_needed, core, band)

Find last available slot range along path (last-fit algorithm).

This is a read-only query method. It does NOT allocate.

Parameters:

• path (list[str]) – Ordered list of node IDs.

• slots_needed (int) – Number of contiguous slots required.

• core (int) – Core index to search.

• band (str) – Band identifier.

Returns:

Start slot index if found, None if no fit available.

Return type:

int | None

get_spectrum_utilization(band=None)

Calculate overall spectrum utilization.

Parameters:

band (str | None) – Specific band to check, or None for all bands.

Returns:

Utilization ratio between 0.0 and 1.0.

Return type:

float

get_neighbors(node)

Get neighboring nodes.

Parameters:

node (str) –

Return type:

list[str]

has_link(link)

Check if a link exists in the topology.

Parameters:

link (tuple[str, str]) –

Return type:

bool

get_link_length(link)

Get physical length of a link in km.

Parameters:

link (tuple[str, str]) – (source, destination) tuple.

Returns:

Link length in kilometers.

Return type:

float

Raises:

KeyError – If link does not exist.

get_link_utilization(link)

Get spectrum utilization for a link as a fraction (0.0 to 1.0).

Calculates the ratio of occupied slots to total slots across all bands and cores. A slot is considered occupied if its value is non-zero (positive for lightpath data, negative for guard bands).

Parameters:

link (tuple[str, str]) – (source, destination) tuple.

Returns:

Utilization fraction between 0.0 (empty) and 1.0 (full).

Return type:

float

Raises:

KeyError – If link does not exist.

reset()

Reset network state for new RL episode.

Clears: - All spectrum allocations (sets arrays to zero) - Active lightpaths (clears registry) - Resets lightpath ID counter

Preserves: - Topology structure - Configuration - Link spectrum objects (just cleared, not recreated)

This method is used by UnifiedSimEnv.reset() to prepare for a new episode without needing to recreate the entire NetworkState object.

Example

>>> state = NetworkState(topology, config)
>>> # ... run episode ...
>>> state.reset()  # Clear for new episode

Return type:

None

create_lightpath(path, start_slot, end_slot, core, band, modulation, bandwidth_gbps, path_weight_km, guard_slots=0, *, backup_path=None, backup_start_slot=None, backup_end_slot=None, backup_core=None, backup_band=None, snr_db=None, xt_cost=None, connection_index=None, arrive_time=None)

Create and register a new lightpath, allocating spectrum.

Parameters:

• path (list[str]) – Ordered list of node IDs (minimum 2 nodes).

• start_slot (int) – First spectrum slot (inclusive).

• end_slot (int) – Last spectrum slot (exclusive, includes guard slots).

• core (int) – Core index for allocation.

• band (str) – Band identifier (“c”, “l”, “s”).

• modulation (str) – Modulation format name.

• bandwidth_gbps (int) – Total lightpath capacity.

• path_weight_km (float) – Path length/weight in kilometers.

• guard_slots (int) – Number of guard band slots at end (default 0).

• backup_path (list[str] | None) – Optional backup path for 1+1 protection.

• backup_start_slot (int | None) – Backup path start slot.

• backup_end_slot (int | None) – Backup path end slot.

• backup_core (int | None) – Backup path core index.

• backup_band (str | None) – Backup path band.

• snr_db (float | None) – Measured SNR value (optional).

• xt_cost (float | None) – Crosstalk cost (optional).

• connection_index (int | None) – External routing index for SNR lookup (optional).

• arrive_time (float | None) – Request arrival time for utilization tracking (optional).

Returns:

Newly created Lightpath with assigned lightpath_id.

Return type:

Lightpath

Raises:

• ValueError – If parameters are invalid or spectrum unavailable.

• KeyError – If any link in path doesn’t exist in topology.

release_lightpath(lightpath_id)

Release a lightpath and free its spectrum.

Parameters:

lightpath_id (int) – ID of lightpath to release.

Returns:

True if lightpath was found and released, False if not found.

Return type:

bool

allocate_request_bandwidth(lightpath_id, request_id, bandwidth_gbps, timestamp=None)

Allocate bandwidth on existing lightpath to a request.

Used by grooming to share lightpath capacity among requests.

Parameters:

• lightpath_id (int) – ID of the lightpath.

• request_id (int) – ID of the request.

• bandwidth_gbps (int) – Bandwidth to allocate.

• timestamp (float | None) – Optional arrival time for utilization tracking.

Raises:

• KeyError – If lightpath not found.

• ValueError – If insufficient remaining bandwidth or request already allocated.

Return type:

None

release_request_bandwidth(lightpath_id, request_id, timestamp=None)

Release bandwidth allocated to a request.

Parameters:

• lightpath_id (int) – ID of the lightpath.

• request_id (int) – ID of the request to release.

• timestamp (float | None) – Optional departure time for utilization tracking.

Returns:

Amount of bandwidth released.

Return type:

int

Raises:

KeyError – If lightpath or request not found.

property network_spectrum_dict: dict[tuple[str, str], dict[str, Any]]

Returns spectrum state in sdn_props format.

Warning

This property is a TEMPORARY MIGRATION SHIM.

Both (u,v) and (v,u) return the SAME dict object. cores_matrix arrays are direct references (not copies). Edge attributes from topology are included.

Returns:

Dictionary mapping link tuples to link state dicts. Format matches sdn_props.network_spectrum_dict exactly.

Return type:

dict[tuple[str, str], dict[str, Any]]

Type:

Legacy compatibility

property lightpath_status_dict: dict[tuple[str, str], dict[int, dict[str, Any]]]

Returns lightpath state in sdn_props format.

Warning

This property is a TEMPORARY MIGRATION SHIM.

Keys are SORTED tuples (“A”, “C”) not (“C”, “A”). Bandwidth values are float (legacy format). time_bw_usage is empty dict by default. Rebuilds dict on each access (not cached).

Returns:

Dictionary mapping sorted endpoint tuples to lightpath dicts. Format matches sdn_props.lightpath_status_dict exactly.

Return type:

dict[tuple[str, str], dict[int, dict[str, Any]]]

Type:

Legacy compatibility

Result objects for FUSION pipeline stages.

This module defines frozen dataclasses for all pipeline outputs: - RouteResult: Routing pipeline output (candidate paths) - SpectrumResult: Spectrum assignment output - GroomingResult: Grooming pipeline output - SlicingResult: Slicing pipeline output - SNRResult: SNR validation output - AllocationResult: Final orchestrator output (SINGLE SOURCE OF TRUTH)

All result objects are immutable (frozen dataclasses) to ensure consistency throughout the processing pipeline.

class fusion.domain.results.RouteResult

Bases: object

Result of routing pipeline - candidate paths with modulation options.

Contains ordered list of candidate paths from best to worst, each with associated path weight and valid modulation formats. For 1+1 protection, includes disjoint backup paths.

paths

List of candidate paths (each path is tuple of node IDs).

Type:

tuple[tuple[str, …], …]

weights_km

Path lengths/weights in kilometers.

Type:

tuple[float, …]

modulations

Valid modulation formats per path.

Type:

tuple[tuple[str, …], …]

backup_paths

Disjoint backup paths for 1+1 protection (optional).

Type:

tuple[tuple[str, …], …] | None

backup_weights_km

Backup path lengths (optional).

Type:

tuple[float, …] | None

backup_modulations

Modulations for backup paths (optional).

Type:

tuple[tuple[str, …], …] | None

strategy_name

Name of routing algorithm used.

Type:

str

metadata

Algorithm-specific metadata.

Type:

dict[str, Any]

connection_index

External routing index for pre-calculated SNR lookup.

Type:

int | None

Example

>>> result = RouteResult(
...     paths=(("0", "2", "5"),),
...     weights_km=(100.0,),
...     modulations=(("QPSK", "16-QAM"),),
...     strategy_name="k_shortest_path",
... )
>>> result.is_empty
False
>>> result.best_path
('0', '2', '5')

paths: tuple[tuple[str, ...], ...] = ()

weights_km: tuple[float, ...] = ()

modulations: tuple[tuple[str, ...], ...] = ()

backup_paths: tuple[tuple[str, ...], ...] | None = None

backup_weights_km: tuple[float, ...] | None = None

backup_modulations: tuple[tuple[str, ...], ...] | None = None

strategy_name: str = ''

metadata: dict[str, Any]

connection_index: int | None = None

property is_empty: bool

True if no paths found (routing failed).

property num_paths: int

Number of candidate paths.

property has_protection: bool

True if backup paths are available.

property best_path: tuple[str, ...] | None

Best (first) candidate path or None if empty.

property best_weight: float | None

Weight of best path or None if empty.

get_path(index)

Get path at index.

Parameters:

index (int) –

Return type:

tuple[str, …]

get_modulations_for_path(index)

Get valid modulations for path at index.

Parameters:

index (int) –

Return type:

tuple[str, …]

classmethod empty(strategy_name='')

Create empty result (no paths found).

Parameters:

strategy_name (str) –

Return type:

RouteResult

classmethod from_routing_props(routing_props)

Create from legacy RoutingProps.

Parameters:

routing_props (Any) – Legacy RoutingProps object.

Returns:

New RouteResult instance.

Return type:

RouteResult

__init__(paths=(), weights_km=(), modulations=(), backup_paths=None, backup_weights_km=None, backup_modulations=None, strategy_name='', metadata=<factory>, connection_index=None)

Parameters:

• paths (tuple[tuple[str, ...], ...]) –

• weights_km (tuple[float, ...]) –

• modulations (tuple[tuple[str, ...], ...]) –

• backup_paths (tuple[tuple[str, ...], ...] | None) –

• backup_weights_km (tuple[float, ...] | None) –

• backup_modulations (tuple[tuple[str, ...], ...] | None) –

• strategy_name (str) –

• metadata (dict[str, Any]) –

• connection_index (int | None) –

Return type:

None

class fusion.domain.results.SpectrumResult

Bases: object

Result of spectrum assignment pipeline.

Indicates whether contiguous spectrum was found and provides the allocation details (slot range, core, band, modulation).

is_free

Whether spectrum assignment was successful.

Type:

bool

start_slot

First slot index (valid only if is_free=True).

Type:

int

end_slot

Last slot index exclusive (valid only if is_free=True).

Type:

int

core

Core number for MCF (valid only if is_free=True).

Type:

int

band

Frequency band (valid only if is_free=True).

Type:

str

modulation

Selected modulation format.

Type:

str

slots_needed

Number of slots required (including guard band).

Type:

int

achieved_bandwidth_gbps

Achieved bandwidth in dynamic_lps mode.

Type:

int | None

snr_db

SNR value calculated during spectrum assignment.

Type:

float | None

backup_start_slot

Backup spectrum start for 1+1 protection.

Type:

int | None

backup_end_slot

Backup spectrum end for 1+1 protection.

Type:

int | None

backup_core

Backup core number for 1+1 protection.

Type:

int | None

backup_band

Backup frequency band for 1+1 protection.

Type:

str | None

Example

>>> result = SpectrumResult(
...     is_free=True,
...     start_slot=100,
...     end_slot=108,
...     core=0,
...     band="c",
...     modulation="QPSK",
...     slots_needed=8,
... )
>>> result.num_slots
8

is_free: bool

start_slot: int = 0

end_slot: int = 0

core: int = 0

band: str = 'c'

modulation: str = ''

slots_needed: int = 0

achieved_bandwidth_gbps: int | None = None

snr_db: float | None = None

backup_start_slot: int | None = None

backup_end_slot: int | None = None

backup_core: int | None = None

backup_band: str | None = None

property num_slots: int

Number of slots allocated (0 if not free).

property has_backup: bool

True if backup spectrum is allocated.

classmethod not_found(slots_needed=0)

Create result for failed spectrum assignment.

Parameters:

slots_needed (int) –

Return type:

SpectrumResult

classmethod from_spectrum_props(spectrum_props)

Create from legacy SpectrumProps or dict.

Parameters:

spectrum_props (Any) – Legacy SpectrumProps object or dict.

Returns:

New SpectrumResult instance.

Return type:

SpectrumResult

to_allocation_dict()

Convert to allocation dictionary for legacy compatibility.

Returns:

Dictionary with allocation details.

Return type:

dict[str, Any]

__init__(is_free, start_slot=0, end_slot=0, core=0, band='c', modulation='', slots_needed=0, achieved_bandwidth_gbps=None, snr_db=None, backup_start_slot=None, backup_end_slot=None, backup_core=None, backup_band=None)

Parameters:

• is_free (bool) –

• start_slot (int) –

• end_slot (int) –

• core (int) –

• band (str) –

• modulation (str) –

• slots_needed (int) –

• achieved_bandwidth_gbps (int | None) –

• snr_db (float | None) –

• backup_start_slot (int | None) –

• backup_end_slot (int | None) –

• backup_core (int | None) –

• backup_band (str | None) –

Return type:

None

class fusion.domain.results.GroomingResult

Bases: object

Result of grooming pipeline - using existing lightpath capacity.

Indicates whether a request can be (fully or partially) served by existing lightpaths without creating new ones.

fully_groomed

Entire request fits in existing lightpaths.

Type:

bool

partially_groomed

Some bandwidth groomed, rest needs new LP.

Type:

bool

bandwidth_groomed_gbps

Amount successfully groomed.

Type:

int

remaining_bandwidth_gbps

Amount still needing new lightpath.

Type:

int

lightpaths_used

IDs of lightpaths used for grooming.

Type:

tuple[int, …]

forced_path

Required path for new lightpath (if partial).

Type:

tuple[str, …] | None

snr_list

SNR values from grooming attempts (for legacy compatibility).

Type:

tuple[float, …]

modulation_list

Modulation formats from grooming attempts.

Type:

tuple[str, …]

Example

>>> result = GroomingResult.full(100, [1, 2])
>>> result.fully_groomed
True
>>> result.needs_new_lightpath
False

fully_groomed: bool = False

partially_groomed: bool = False

bandwidth_groomed_gbps: int = 0

remaining_bandwidth_gbps: int = 0

lightpaths_used: tuple[int, ...] = ()

forced_path: tuple[str, ...] | None = None

snr_list: tuple[float, ...] = ()

modulation_list: tuple[str, ...] = ()

property was_groomed: bool

True if any grooming occurred.

property needs_new_lightpath: bool

True if a new lightpath is still needed.

classmethod no_grooming(bandwidth_gbps)

Create result when no grooming is possible.

Parameters:

bandwidth_gbps (int) –

Return type:

GroomingResult

classmethod full(bandwidth_gbps, lightpath_ids, snr_list=None, modulation_list=None)

Create result for fully groomed request.

Parameters:

• bandwidth_gbps (int) –

• lightpath_ids (list[int]) –

• snr_list (list[float] | None) –

• modulation_list (list[str] | None) –

Return type:

GroomingResult

classmethod partial(bandwidth_groomed, remaining, lightpath_ids, forced_path=None, snr_list=None, modulation_list=None)

Create result for partially groomed request.

Parameters:

• bandwidth_groomed (int) –

• remaining (int) –

• lightpath_ids (list[int]) –

• forced_path (list[str] | None) –

• snr_list (list[float] | None) –

• modulation_list (list[str] | None) –

Return type:

GroomingResult

__init__(fully_groomed=False, partially_groomed=False, bandwidth_groomed_gbps=0, remaining_bandwidth_gbps=0, lightpaths_used=(), forced_path=None, snr_list=(), modulation_list=())

Parameters:

• fully_groomed (bool) –

• partially_groomed (bool) –

• bandwidth_groomed_gbps (int) –

• remaining_bandwidth_gbps (int) –

• lightpaths_used (tuple[int, ...]) –

• forced_path (tuple[str, ...] | None) –

• snr_list (tuple[float, ...]) –

• modulation_list (tuple[str, ...]) –

Return type:

None

class fusion.domain.results.SlicingResult

Bases: object

Result of slicing pipeline - splitting request across lightpaths.

When a request is too large for a single lightpath (modulation limitations), it may be split into multiple smaller slices.

success

Whether slicing succeeded.

Type:

bool

num_slices

Number of slices created.

Type:

int

slice_bandwidth_gbps

Bandwidth per slice.

Type:

int

lightpaths_created

IDs of created lightpaths.

Type:

tuple[int, …]

total_bandwidth_gbps

Total bandwidth across all slices.

Type:

int

failed_attempt_snr_values

SNR values from failed attempts (legacy).

Type:

tuple[float, …]

Example

>>> result = SlicingResult.sliced(
...     num_slices=4,
...     slice_bandwidth=25,
...     lightpath_ids=[1, 2, 3, 4],
... )
>>> result.is_sliced
True
>>> result.total_bandwidth_gbps
100

success: bool = False

num_slices: int = 0

slice_bandwidth_gbps: int = 0

lightpaths_created: tuple[int, ...] = ()

total_bandwidth_gbps: int = 0

failed_attempt_snr_values: tuple[float, ...] = ()

property is_sliced: bool

True if request was sliced (multiple lightpaths).

classmethod failed()

Create result for failed slicing.

Return type:

SlicingResult

classmethod single_lightpath(bandwidth_gbps, lightpath_id)

Create result for single lightpath (no slicing needed).

Parameters:

• bandwidth_gbps (int) –

• lightpath_id (int) –

Return type:

SlicingResult

classmethod sliced(num_slices, slice_bandwidth, lightpath_ids)

Create result for sliced request.

Parameters:

• num_slices (int) –

• slice_bandwidth (int) –

• lightpath_ids (list[int]) –

Return type:

SlicingResult

__init__(success=False, num_slices=0, slice_bandwidth_gbps=0, lightpaths_created=(), total_bandwidth_gbps=0, failed_attempt_snr_values=())

Parameters:

• success (bool) –

• num_slices (int) –

• slice_bandwidth_gbps (int) –

• lightpaths_created (tuple[int, ...]) –

• total_bandwidth_gbps (int) –

• failed_attempt_snr_values (tuple[float, ...]) –

Return type:

None

class fusion.domain.results.SNRResult

Bases: object

Result of SNR validation pipeline.

Indicates whether the signal-to-noise ratio meets the threshold required for the selected modulation format.

passed

Whether SNR meets threshold (primary success indicator).

Type:

bool

snr_db

Calculated SNR value in dB.

Type:

float

required_snr_db

Threshold required for modulation format.

Type:

float

margin_db

SNR margin (snr_db - required_snr_db).

Type:

float

failure_reason

Explanation if validation failed.

Type:

str | None

link_snr_values

Per-link SNR breakdown for debugging.

Type:

dict[tuple[str, str], float]

Example

>>> result = SNRResult.success(snr_db=18.5, required_snr_db=15.0)
>>> result.passed
True
>>> result.margin_db
3.5

passed: bool

snr_db: float = 0.0

required_snr_db: float = 0.0

margin_db: float = 0.0

failure_reason: str | None = None

link_snr_values: dict[tuple[str, str], float]

property is_degraded: bool

True if SNR passed but with low margin (< 1 dB).

property has_link_breakdown: bool

True if per-link SNR values are available.

classmethod success(snr_db, required_snr_db, link_snr_values=None)

Create successful SNR result.

Parameters:

• snr_db (float) –

• required_snr_db (float) –

• link_snr_values (dict[tuple[str, str], float] | None) –

Return type:

SNRResult

classmethod failure(snr_db, required_snr_db, reason='SNR below threshold', link_snr_values=None)

Create failed SNR result.

Parameters:

• snr_db (float) –

• required_snr_db (float) –

• reason (str) –

• link_snr_values (dict[tuple[str, str], float] | None) –

Return type:

SNRResult

classmethod skipped()

Create result for when SNR validation is disabled.

Return type:

SNRResult

get_weakest_link()

Get the link with lowest SNR.

Returns:

Tuple of (link, snr_db) or None if no link breakdown.

Return type:

tuple[tuple[str, str], float] | None

__init__(passed, snr_db=0.0, required_snr_db=0.0, margin_db=0.0, failure_reason=None, link_snr_values=<factory>)

Parameters:

• passed (bool) –

• snr_db (float) –

• required_snr_db (float) –

• margin_db (float) –

• failure_reason (str | None) –

• link_snr_values (dict[tuple[str, str], float]) –

Return type:

None

class fusion.domain.results.SNRRecheckResult

Bases: object

Result of SNR recheck after new lightpath allocation.

When a new lightpath is allocated, it may cause interference with existing lightpaths. This result captures the outcome of rechecking SNR for affected lightpaths.

all_pass

True if all affected lightpaths still meet SNR threshold.

Type:

bool

degraded_lightpath_ids

IDs of lightpaths now below threshold.

Type:

tuple[int, …]

violations

Mapping of lightpath_id to SNR shortfall in dB.

Type:

dict[int, float]

checked_count

Number of lightpaths that were checked.

Type:

int

Example

>>> result = SNRRecheckResult.success()
>>> result.all_pass
True
>>> result.num_degraded
0

all_pass: bool

degraded_lightpath_ids: tuple[int, ...] = ()

violations: dict[int, float]

checked_count: int = 0

property num_degraded: int

Number of lightpaths that are now degraded.

property has_violations: bool

True if any lightpaths are degraded.

get_worst_violation()

Get lightpath with largest SNR shortfall.

Returns:

Tuple of (lightpath_id, shortfall_db) or None if no violations.

Return type:

tuple[int, float] | None

classmethod success(checked_count=0)

Create result when all affected lightpaths still pass.

Parameters:

checked_count (int) –

Return type:

SNRRecheckResult

classmethod degraded(degraded_ids, violations, checked_count=0)

Create result when some lightpaths are now degraded.

Parameters:

• degraded_ids (list[int]) –

• violations (dict[int, float]) –

• checked_count (int) –

Return type:

SNRRecheckResult

__init__(all_pass, degraded_lightpath_ids=(), violations=<factory>, checked_count=0)

Parameters:

• all_pass (bool) –

• degraded_lightpath_ids (tuple[int, ...]) –

• violations (dict[int, float]) –

• checked_count (int) –

Return type:

None

class fusion.domain.results.ProtectionResult

Bases: object

Result of protection path establishment or switchover.

Used for 1+1 protection scenarios to track the establishment of primary and backup paths, and any switchover events.

primary_established

Whether primary path was established.

Type:

bool

backup_established

Whether backup path was established.

Type:

bool

primary_spectrum

Spectrum allocation for primary path.

Type:

fusion.domain.results.SpectrumResult | None

backup_spectrum

Spectrum allocation for backup path.

Type:

fusion.domain.results.SpectrumResult | None

switchover_triggered

Whether a switchover event occurred.

Type:

bool

switchover_success

Whether switchover completed successfully.

Type:

bool

switchover_time_ms

Time taken for switchover in milliseconds.

Type:

float | None

failure_type

Type of failure that triggered switchover.

Type:

str | None

recovery_start

Simulation time when recovery started.

Type:

float | None

recovery_end

Simulation time when recovery completed.

Type:

float | None

recovery_type

Type of recovery (“protection” or “restoration”).

Type:

str | None

Example

>>> result = ProtectionResult.established(
...     primary_spectrum=spectrum1,
...     backup_spectrum=spectrum2,
... )
>>> result.is_fully_protected
True

primary_established: bool = False

backup_established: bool = False

primary_spectrum: SpectrumResult | None = None

backup_spectrum: SpectrumResult | None = None

switchover_triggered: bool = False

switchover_success: bool = False

switchover_time_ms: float | None = None

failure_type: str | None = None

recovery_start: float | None = None

recovery_end: float | None = None

recovery_type: str | None = None

property is_fully_protected: bool

True if both primary and backup paths are established.

property recovery_duration_ms: float | None

Duration of recovery in milliseconds, if completed.

classmethod established(primary_spectrum, backup_spectrum=None)

Create result for successfully established protection.

Parameters:

• primary_spectrum (SpectrumResult) –

• backup_spectrum (SpectrumResult | None) –

Return type:

ProtectionResult

classmethod primary_only(primary_spectrum)

Create result when only primary path established (backup failed).

Parameters:

primary_spectrum (SpectrumResult) –

Return type:

ProtectionResult

classmethod failed()

Create result for failed protection establishment.

Return type:

ProtectionResult

classmethod switchover(success, switchover_time_ms, failure_type, recovery_type='protection')

Create result for a switchover event.

Parameters:

• success (bool) –

• switchover_time_ms (float) –

• failure_type (str) –

• recovery_type (str) –

Return type:

ProtectionResult

__init__(primary_established=False, backup_established=False, primary_spectrum=None, backup_spectrum=None, switchover_triggered=False, switchover_success=False, switchover_time_ms=None, failure_type=None, recovery_start=None, recovery_end=None, recovery_type=None)

Parameters:

• primary_established (bool) –

• backup_established (bool) –

• primary_spectrum (SpectrumResult | None) –

• backup_spectrum (SpectrumResult | None) –

• switchover_triggered (bool) –

• switchover_success (bool) –

• switchover_time_ms (float | None) –

• failure_type (str | None) –

• recovery_start (float | None) –

• recovery_end (float | None) –

• recovery_type (str | None) –

Return type:

None

class fusion.domain.results.AllocationResult

Bases: object

Final result of request allocation - SINGLE SOURCE OF TRUTH.

This is the authoritative result returned by the orchestrator. The success field is the final word on whether a request was served (fully or partially).

success

FINAL AUTHORITY - True if request was served.

Type:

bool

lightpaths_created

IDs of newly created lightpaths.

Type:

tuple[int, …]

lightpaths_groomed

IDs of existing lightpaths used.

Type:

tuple[int, …]

total_bandwidth_allocated_gbps

Total bandwidth allocated.

Type:

int

is_groomed

Request used existing lightpath capacity.

Type:

bool

is_partially_groomed

Mix of existing and new lightpath.

Type:

bool

is_sliced

Request split across multiple lightpaths.

Type:

bool

is_protected

Request has 1+1 protection.

Type:

bool

block_reason

BlockReason enum if success=False.

Type:

BlockReason | None

bandwidth_allocations

Bandwidth allocated per segment.

Type:

tuple[int, …]

modulations

Modulation format per segment.

Type:

tuple[str, …]

cores

Core number per segment.

Type:

tuple[int, …]

bands

Frequency band per segment.

Type:

tuple[str, …]

start_slots

Start slot per segment.

Type:

tuple[int, …]

end_slots

End slot per segment.

Type:

tuple[int, …]

xt_costs

Crosstalk cost per segment (from crosstalk_list).

Type:

tuple[float, …]

xt_values

Crosstalk values per segment (from xt_list).

Type:

tuple[float, …]

snr_values

SNR value per segment.

Type:

tuple[float, …]

lightpath_bandwidths

Total bandwidth per lightpath.

Type:

tuple[int, …]

failed_attempt_snr_values

SNR values from failed allocation attempts.

Type:

tuple[float, …]

path_index

Which k-path was selected (0, 1, 2…).

Type:

int

route_result

Routing pipeline output.

Type:

RouteResult | None

spectrum_result

Spectrum pipeline output.

Type:

SpectrumResult | None

grooming_result

Grooming pipeline output.

Type:

GroomingResult | None

slicing_result

Slicing pipeline output.

Type:

SlicingResult | None

snr_result

SNR validation output.

Type:

SNRResult | None

protection_result

Protection establishment output.

Type:

ProtectionResult | None

Example

>>> result = AllocationResult.success_new_lightpath(
...     lightpath_id=42,
...     bandwidth_gbps=100,
... )
>>> result.success
True
>>> result.num_lightpaths
1

success: bool

lightpaths_created: tuple[int, ...] = ()

lightpaths_groomed: tuple[int, ...] = ()

total_bandwidth_allocated_gbps: int = 0

is_groomed: bool = False

is_partially_groomed: bool = False

is_sliced: bool = False

is_protected: bool = False

block_reason: BlockReason | None = None

bandwidth_allocations: tuple[int, ...] = ()

modulations: tuple[str, ...] = ()

cores: tuple[int, ...] = ()

bands: tuple[str, ...] = ()

start_slots: tuple[int, ...] = ()

end_slots: tuple[int, ...] = ()

xt_costs: tuple[float, ...] = ()

xt_values: tuple[float, ...] = ()

snr_values: tuple[float, ...] = ()

lightpath_bandwidths: tuple[int, ...] = ()

failed_attempt_snr_values: tuple[float, ...] = ()

path_index: int = 0

route_result: RouteResult | None = None

spectrum_result: SpectrumResult | None = None

grooming_result: GroomingResult | None = None

slicing_result: SlicingResult | None = None

snr_result: SNRResult | None = None

protection_result: ProtectionResult | None = None

property all_lightpath_ids: tuple[int, ...]

All lightpath IDs (created + groomed).

property num_lightpaths: int

Total number of lightpaths used.

property used_grooming: bool

True if any grooming was used.

classmethod blocked(reason, route_result=None, spectrum_result=None, snr_result=None)

Create result for blocked request.

Parameters:

• reason (BlockReason) –

• route_result (RouteResult | None) –

• spectrum_result (SpectrumResult | None) –

• snr_result (SNRResult | None) –

Return type:

AllocationResult

classmethod success_new_lightpath(lightpath_id, bandwidth_gbps, route_result=None, spectrum_result=None, snr_result=None, is_protected=False)

Create result for successful allocation with new lightpath.

Parameters:

• lightpath_id (int) –

• bandwidth_gbps (int) –

• route_result (RouteResult | None) –

• spectrum_result (SpectrumResult | None) –

• snr_result (SNRResult | None) –

• is_protected (bool) –

Return type:

AllocationResult

classmethod success_groomed(lightpath_ids, bandwidth_gbps, grooming_result=None)

Create result for fully groomed request.

Parameters:

• lightpath_ids (list[int]) –

• bandwidth_gbps (int) –

• grooming_result (GroomingResult | None) –

Return type:

AllocationResult

classmethod success_partial_groom(groomed_ids, new_lightpath_id, total_bandwidth, grooming_result=None, route_result=None, spectrum_result=None, snr_result=None)

Create result for partially groomed request.

Parameters:

• groomed_ids (list[int]) –

• new_lightpath_id (int) –

• total_bandwidth (int) –

• grooming_result (GroomingResult | None) –

• route_result (RouteResult | None) –

• spectrum_result (SpectrumResult | None) –

• snr_result (SNRResult | None) –

Return type:

AllocationResult

classmethod success_sliced(lightpath_ids, bandwidth_gbps, slicing_result=None, route_result=None)

Create result for sliced request.

Parameters:

• lightpath_ids (list[int]) –

• bandwidth_gbps (int) –

• slicing_result (SlicingResult | None) –

• route_result (RouteResult | None) –

Return type:

AllocationResult

__init__(success, lightpaths_created=(), lightpaths_groomed=(), total_bandwidth_allocated_gbps=0, is_groomed=False, is_partially_groomed=False, is_sliced=False, is_protected=False, block_reason=None, bandwidth_allocations=(), modulations=(), cores=(), bands=(), start_slots=(), end_slots=(), xt_costs=(), xt_values=(), snr_values=(), lightpath_bandwidths=(), failed_attempt_snr_values=(), path_index=0, route_result=None, spectrum_result=None, grooming_result=None, slicing_result=None, snr_result=None, protection_result=None)

Parameters:

• success (bool) –

• lightpaths_created (tuple[int, ...]) –

• lightpaths_groomed (tuple[int, ...]) –

• total_bandwidth_allocated_gbps (int) –

• is_groomed (bool) –

• is_partially_groomed (bool) –

• is_sliced (bool) –

• is_protected (bool) –

• block_reason (BlockReason | None) –

• bandwidth_allocations (tuple[int, ...]) –

• modulations (tuple[str, ...]) –

• cores (tuple[int, ...]) –

• bands (tuple[str, ...]) –

• start_slots (tuple[int, ...]) –

• end_slots (tuple[int, ...]) –

• xt_costs (tuple[float, ...]) –

• xt_values (tuple[float, ...]) –

• snr_values (tuple[float, ...]) –

• lightpath_bandwidths (tuple[int, ...]) –

• failed_attempt_snr_values (tuple[float, ...]) –

• path_index (int) –

• route_result (RouteResult | None) –

• spectrum_result (SpectrumResult | None) –

• grooming_result (GroomingResult | None) –

• slicing_result (SlicingResult | None) –

• snr_result (SNRResult | None) –

• protection_result (ProtectionResult | None) –

Return type:

None

---

Interfaces

fusion.interfaces

Abstract base classes and protocols.

Interfaces module for FUSION simulator.

This module contains: 1. Abstract base classes that define contracts for legacy pluggable components 2. Protocol classes for type-safe pipeline interfaces 3. Control policy protocol for unified path selection

Legacy Interfaces (Abstract Base Classes):

• AbstractRoutingAlgorithm

• AbstractSpectrumAssigner

• AbstractSNRMeasurer

• AgentInterface

Pipeline Protocols (typing.Protocol):

• RoutingPipeline

• SpectrumPipeline

• GroomingPipeline

• SNRPipeline

• SlicingPipeline

Control Policy Protocol:

• ControlPolicy

class fusion.interfaces.AbstractRoutingAlgorithm

Bases: ABC

Base class for all routing algorithms in FUSION.

This interface defines the contract that all routing algorithms must follow to ensure compatibility with the FUSION simulation framework.

Most routing algorithms store results in route_props (paths_matrix, modulation_formats_matrix, weights_list). Special algorithms like OnePlusOneProtection may use alternative storage mechanisms.

__init__(engine_props, sdn_props)

Initialize the routing algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration and properties

• sdn_props (SDNProps) – Object containing SDN controller properties and network state

abstract property algorithm_name: str

Return the name of the routing algorithm.

Returns:

String identifier for this routing algorithm

Return type:

str

abstract property supported_topologies: list[str]

Return list of supported topology types.

Returns:

List of topology names this algorithm supports (e.g., [‘NSFNet’, ‘USBackbone’])

Return type:

List[str]

abstract validate_environment(topology)

Validate that the routing algorithm can work with the given environment.

Parameters:

topology (Any) – NetworkX graph representing the network topology

Returns:

True if the algorithm can route in this environment, False otherwise

Return type:

bool

abstract route(source, destination, request)

Find a route from source to destination for the given request.

Results are stored in the algorithm’s route_props attribute (paths_matrix, modulation_formats_matrix, weights_list). This method does not return a value; consumers should access route_props.paths_matrix to retrieve computed paths.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• request (Any) – Request object containing traffic demand details

Return type:

None

abstract get_paths(source, destination, k=1)

Get k shortest paths between source and destination.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• k (int) – Number of paths to return

Returns:

List of k paths, where each path is a list of nodes

Return type:

List[List[Any]]

abstract update_weights(topology)

Update edge weights based on current network state.

Parameters:

topology (Any) – NetworkX graph to update weights for

Return type:

None

abstract get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics

Return type:

Dict[str, Any]

reset()

Reset the routing algorithm state.

This method can be overridden by subclasses that maintain state.

Return type:

None

class fusion.interfaces.AbstractSpectrumAssigner

Bases: ABC

Base class for all spectrum assignment algorithms in FUSION.

This interface defines the contract that all spectrum assignment algorithms must follow to ensure compatibility with the FUSION simulation framework.

__init__(engine_props, sdn_props, route_props)

Initialize the spectrum assignment algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration and properties

• sdn_props (SDNProps) – Object containing SDN controller properties and network state

• route_props (object) – Object containing routing properties and selected path

abstract property algorithm_name: str

Return the name of the spectrum assignment algorithm.

Returns:

String identifier for this spectrum assignment algorithm

Return type:

str

abstract property supports_multiband: bool

Indicate whether this algorithm supports multi-band assignment.

Returns:

True if the algorithm supports multi-band, False otherwise

Return type:

bool

abstract assign(path, request)

Assign spectrum resources along the given path for the request.

Parameters:

• path (List[Any]) – List of nodes representing the path

• request (Any) – Request object containing traffic demand and spectrum requirements

Returns:

Dictionary containing spectrum assignment details (start_slot, end_slot, core_num, band, is_free), or None if assignment fails.

Return type:

Optional[Dict[str, Any]]

abstract check_spectrum_availability(path, start_slot, end_slot, core_num, band)

Check if spectrum slots are available along the entire path.

Parameters:

• path (List[Any]) – List of nodes representing the path

• start_slot (int) – Starting slot index

• end_slot (int) – Ending slot index (inclusive)

• core_num (int) – Core number to check

• band (str) – Band identifier

Returns:

True if all slots are available, False otherwise

Return type:

bool

abstract allocate_spectrum(path, start_slot, end_slot, core_num, band, request_id)

Allocate spectrum resources along the path.

Parameters:

• path (List[Any]) – List of nodes representing the path

• start_slot (int) – Starting slot index

• end_slot (int) – Ending slot index (inclusive)

• core_num (int) – Core number to allocate

• band (str) – Band identifier

• request_id (Any) – Unique identifier for the request

Returns:

True if allocation successful, False otherwise

Return type:

bool

abstract deallocate_spectrum(path, start_slot, end_slot, core_num, band)

Deallocate spectrum resources along the path.

Parameters:

• path (List[Any]) – List of nodes representing the path

• start_slot (int) – Starting slot index

• end_slot (int) – Ending slot index (inclusive)

• core_num (int) – Core number to deallocate

• band (str) – Band identifier

Returns:

True if deallocation successful, False otherwise

Return type:

bool

abstract get_fragmentation_metric(path)

Calculate fragmentation metric for the given path.

Parameters:

path (List[Any]) – List of nodes representing the path

Returns:

Fragmentation value (0.0 = no fragmentation, 1.0 = maximum fragmentation)

Return type:

float

abstract get_metrics()

Get spectrum assignment algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics

Return type:

Dict[str, Any]

reset()

Reset the spectrum assignment algorithm state.

This method can be overridden by subclasses that maintain state.

Return type:

None

class fusion.interfaces.AbstractSNRMeasurer

Bases: ABC

Base class for all SNR measurement algorithms in FUSION.

This interface defines the contract that all SNR measurement algorithms must follow to ensure compatibility with the FUSION simulation framework.

__init__(engine_props, sdn_props, spectrum_props, route_props)

Initialize the SNR measurement algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration and properties

• sdn_props (object) – Object containing SDN controller properties and network state

• spectrum_props (object) – Object containing spectrum assignment properties

• route_props (object) – Object containing routing properties and selected path

abstract property algorithm_name: str

Return the name of the SNR measurement algorithm.

Returns:

String identifier for this SNR measurement algorithm

Return type:

str

abstract property supports_multicore: bool

Indicate whether this algorithm supports multi-core fiber measurements.

Returns:

True if the algorithm supports multi-core, False otherwise

Return type:

bool

abstract calculate_snr(path, spectrum_info)

Calculate the SNR for a given path and spectrum assignment.

Parameters:

• path (List[Any]) – List of nodes representing the path

• spectrum_info (Dict[str, Any]) – Dictionary containing spectrum assignment details: - ‘start_slot’: Starting slot index - ‘end_slot’: Ending slot index - ‘core_num’: Core number (for multi-core fibers) - ‘band’: Band identifier - ‘modulation’: Modulation format

Returns:

SNR value in dB

Return type:

float

abstract calculate_link_snr(source, destination, spectrum_info)

Calculate the SNR for a single link.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• spectrum_info (Dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

SNR value in dB for the link

Return type:

float

abstract calculate_crosstalk(path, core_num, spectrum_info)

Calculate crosstalk noise for the given path and core.

Parameters:

• path (List[Any]) – List of nodes representing the path

• core_num (int) – Core number to calculate crosstalk for

• spectrum_info (Dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

Crosstalk noise power in linear units

Return type:

float

abstract calculate_nonlinear_noise(path, spectrum_info)

Calculate nonlinear noise components.

Parameters:

• path (List[Any]) – List of nodes representing the path

• spectrum_info (Dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

Dictionary containing nonlinear noise components: - ‘sci’: Self-channel interference - ‘xci’: Cross-channel interference - ‘xpm’: Cross-phase modulation - ‘fwm’: Four-wave mixing (if applicable)

Return type:

Dict[str, float]

abstract get_required_snr_threshold(modulation, reach)

Get the required SNR threshold for a given modulation format and reach.

Parameters:

• modulation (str) – Modulation format identifier

• reach (float) – Transmission reach in km

Returns:

Required SNR threshold in dB

Return type:

float

abstract is_snr_acceptable(calculated_snr, required_snr, margin=0.0)

Check if calculated SNR meets the requirement with optional margin.

Parameters:

• calculated_snr (float) – Calculated SNR value in dB

• required_snr (float) – Required SNR threshold in dB

• margin (float) – Additional margin in dB (default: 0.0)

Returns:

True if SNR is acceptable, False otherwise

Return type:

bool

abstract update_link_state(source, destination, spectrum_info)

Update link state based on new spectrum allocation.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• spectrum_info (Dict[str, Any]) – Dictionary containing spectrum assignment details

Return type:

None

abstract get_metrics()

Get SNR measurement algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics

Return type:

Dict[str, Any]

reset()

Reset the SNR measurement algorithm state.

This method can be overridden by subclasses that maintain state.

Return type:

None

class fusion.interfaces.AgentInterface

Bases: ABC

Base interface for all reinforcement learning agents in FUSION.

This interface defines the contract that all RL agents must follow to ensure compatibility with the FUSION simulation framework.

abstract property algorithm_name: str

Return the name of the RL algorithm.

Returns:

String identifier for this RL algorithm

Return type:

str

abstract property action_space_type: str

Return the type of action space.

Returns:

‘discrete’ or ‘continuous’

Return type:

str

abstract property observation_space_shape: tuple[int, ...]

Return the shape of the observation space.

Returns:

Tuple describing observation dimensions

Return type:

Tuple[int, …]

abstract act(observation, _deterministic=False)

Select an action based on the current observation.

Parameters:

• observation (Any) – Current environment observation

• deterministic (bool) – If True, select action deterministically (no exploration)

• _deterministic (bool) –

Returns:

Action to take (int for discrete, array for continuous)

Return type:

Union[int, Any]

abstract train(env, _total_timesteps, **kwargs)

Train the agent on the given environment.

Parameters:

• env (Any) – Training environment (e.g., Gym environment)

• total_timesteps (int) – Total number of timesteps to train for

• kwargs (dict) – Additional training parameters

• _total_timesteps (int) –

Returns:

Dictionary containing training metrics and results

Return type:

Dict[str, Any]

abstract learn_from_experience(observation, action, reward, _next_observation, done)

Learn from a single experience tuple.

Parameters:

• observation (Any) – Current observation

• action (Union[int, Any]) – Action taken

• reward (float) – Reward received

• next_observation (Any) – Resulting observation

• done (bool) – Whether episode terminated

• _next_observation (Any) –

Returns:

Optional dictionary containing learning metrics (e.g., loss values)

Return type:

Optional[Dict[str, float]]

abstract save(path)

Save the agent’s model/parameters to disk.

Parameters:

path (str) – Path where to save the model

Return type:

None

abstract load(path)

Load the agent’s model/parameters from disk.

Parameters:

path (str) – Path from where to load the model

Return type:

None

abstract get_reward(state, action, _next_state, info)

Calculate reward for a state-action-next_state transition.

Parameters:

• state (Dict[str, Any]) – Current state information

• action (Union[int, Any]) – Action taken

• next_state (Dict[str, Any]) – Resulting state information

• info (Dict[str, Any]) – Additional information from environment

• _next_state (dict[str, Any]) –

Returns:

Calculated reward value

Return type:

float

abstract update_exploration_params(_timestep, _total_timesteps)

Update exploration parameters based on training progress.

Parameters:

• timestep (int) – Current training timestep

• total_timesteps (int) – Total training timesteps

• _timestep (int) –

• _total_timesteps (int) –

Return type:

None

abstract get_config()

Get agent configuration parameters.

Returns:

Dictionary containing agent configuration

Return type:

Dict[str, Any]

abstract set_config(config)

Set agent configuration parameters.

Parameters:

config (Dict[str, Any]) – Dictionary containing agent configuration

Return type:

None

abstract get_metrics()

Get agent performance metrics.

Returns:

Dictionary containing agent-specific metrics

Return type:

Dict[str, Any]

reset()

Reset the agent’s internal state.

This method can be overridden by subclasses that maintain episode state.

Return type:

None

on_episode_start()

Called at the beginning of each episode.

This method can be overridden by subclasses for episode initialization.

Return type:

None

on_episode_end()

Called at the end of each episode.

This method can be overridden by subclasses for episode cleanup.

Return type:

None

class fusion.interfaces.AlgorithmFactory

Bases: object

Factory for creating algorithm instances using the interface system.

static create_routing_algorithm(name, engine_props, sdn_props)

Create a routing algorithm instance.

Parameters:

• name (str) – Name of the routing algorithm

• engine_props (dict) – Engine configuration properties

• sdn_props (SDNProps) – SDN controller properties

Returns:

Configured routing algorithm instance

Return type:

AbstractRoutingAlgorithm

Raises:

ValueError – If algorithm name is not found

static create_spectrum_algorithm(name, engine_props, sdn_props, route_props)

Create a spectrum assignment algorithm instance.

Parameters:

• name (str) – Name of the spectrum algorithm

• engine_props (dict) – Engine configuration properties

• sdn_props (SDNProps) – SDN controller properties

• route_props (object) – Routing properties

Returns:

Configured spectrum assignment algorithm instance

Return type:

AbstractSpectrumAssigner

Raises:

ValueError – If algorithm name is not found

static create_snr_algorithm(name, engine_props, sdn_props, spectrum_props, route_props)

Create an SNR measurement algorithm instance.

Parameters:

• name (str) – Name of the SNR algorithm

• engine_props (dict) – Engine configuration properties

• sdn_props (object) – SDN controller properties

• spectrum_props (object) – Spectrum assignment properties

• route_props (object) – Routing properties

Returns:

Configured SNR measurement algorithm instance

Return type:

AbstractSNRMeasurer

Raises:

ValueError – If algorithm name is not found

class fusion.interfaces.SimulationPipeline

Bases: object

Complete simulation pipeline using interface-based algorithms.

__init__(config)

Initialize the simulation pipeline.

Parameters:

config (Dict[str, Any]) – Configuration dictionary containing algorithm selections and parameters

process_request(source, destination, request)

Process a single network request through the complete pipeline.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• request (Any) – Request object containing traffic demand details

Returns:

Dictionary containing processing results

Return type:

Dict[str, Any]

get_algorithm_info()

Get information about all algorithms in the pipeline.

Return type:

dict[str, dict[str, Any]]

reset_all_algorithms()

Reset state for all algorithms.

Return type:

None

fusion.interfaces.create_simulation_pipeline(config)

Create a complete simulation pipeline from configuration.

Parameters:

config (Dict[str, Any]) – Configuration dictionary

Returns:

Configured SimulationPipeline instance

Return type:

SimulationPipeline

class fusion.interfaces.RoutingPipeline

Bases: Protocol

Protocol for route finding algorithms.

Implementations find candidate routes between source and destination nodes, returning paths with their weights and valid modulation formats.

Design Notes:

• Implementations should NOT store NetworkState as instance attribute

• Receive NetworkState as method parameter for each call

• Return RouteResult without modifying NetworkState

• Configuration accessed via network_state.config

Common Implementations:

• K-shortest paths routing

• Congestion-aware routing

• Fragmentation-aware routing

• NLI-aware routing (non-linear interference)

Example:

class KShortestPathRouter:
    def find_routes(
        self,
        source: str,
        destination: str,
        bandwidth_gbps: int,
        network_state: NetworkState,
        *,
        forced_path: list[str] | None = None,
    ) -> RouteResult:
        if forced_path is not None:
            return RouteResult(paths=(tuple(forced_path),), ...)
        k = network_state.config.k_paths
        paths = find_k_shortest(network_state.topology, source, destination, k)
        return RouteResult(paths=paths, ...)

find_routes(source, destination, bandwidth_gbps, network_state, *, forced_path=None)

Find candidate routes between source and destination.

Parameters:

• source (str) – Source node identifier

• destination (str) – Destination node identifier

• bandwidth_gbps (int) – Required bandwidth (used for modulation selection)

• network_state (NetworkState) – Current network state (topology, config)

• forced_path (list[str] | None) – If provided, use this path instead of searching (typically from partial grooming)

Returns:

RouteResult containing paths, weights_km, modulations, and strategy_name. Returns empty RouteResult if no routes found.

Return type:

RouteResult

Note

• Number of paths limited by network_state.config.k_paths

• Modulation formats filtered by reach based on path weight

• This is a pure query method with no side effects

__init__(*args, **kwargs)

class fusion.interfaces.SpectrumPipeline

Bases: Protocol

Protocol for spectrum assignment algorithms.

Implementations find available spectrum slots along a path for a given bandwidth and modulation format.

Design Notes:

• Implementations should NOT store NetworkState as instance attribute

• Return SpectrumResult without modifying NetworkState

• Actual allocation done by NetworkState.create_lightpath()

Common Implementations:

• First-fit: Allocate lowest available slot range

• Best-fit: Allocate smallest sufficient gap

• Last-fit: Allocate highest available slot range

Example:

class FirstFitSpectrum:
    def find_spectrum(
        self,
        path: list[str],
        modulation: str,
        bandwidth_gbps: int,
        network_state: NetworkState,
    ) -> SpectrumResult:
        slots_needed = calculate_slots(bandwidth_gbps, modulation)
        for band in network_state.config.band_list:
            for core in range(network_state.config.cores_per_link):
                start = find_first_free(path, slots_needed, core, band)
                if start is not None:
                    return SpectrumResult(is_free=True, start_slot=start, ...)
        return SpectrumResult.not_found(slots_needed)

find_spectrum(path, modulation, bandwidth_gbps, network_state, *, connection_index=None, path_index=0, use_dynamic_slicing=False, snr_bandwidth=None, request_id=None, slice_bandwidth=None, excluded_modulations=None)

Find available spectrum along a path.

Parameters:

• path (list[str]) – Ordered list of node IDs forming the route

• modulation (str | list[str]) – Modulation format name (e.g., “QPSK”) or list of valid modulations to try

• bandwidth_gbps (int) – Required bandwidth in Gbps

• network_state (NetworkState) – Current network state

• connection_index (int | None) – External routing index for pre-calculated SNR lookup

• path_index (int) – Index of which k-path is being tried (0, 1, 2…)

• use_dynamic_slicing (bool) – If True, use dynamic slicing to find spectrum

• snr_bandwidth (int | None) – Override bandwidth for SNR calculation

• request_id (int | None) – Request ID for tracking/logging

• slice_bandwidth (int | None) – Bandwidth per slice when slicing is active

• excluded_modulations (set[str] | None) – Modulations to exclude from dynamic slicing

Returns:

SpectrumResult with is_free, start_slot, end_slot, core, band, modulation, and slots_needed fields

Return type:

SpectrumResult

Note

This is a pure query method - does NOT allocate spectrum. Caller must use NetworkState.create_lightpath() to allocate.

find_protected_spectrum(primary_path, backup_path, modulation, bandwidth_gbps, network_state)

Find spectrum for both primary and backup paths (1+1 protection).

Parameters:

• primary_path (list[str]) – Primary route node sequence

• backup_path (list[str]) – Backup route node sequence (should be disjoint)

• modulation (str) – Modulation format name

• bandwidth_gbps (int) – Required bandwidth

• network_state (NetworkState) – Current network state

Returns:

SpectrumResult with primary allocation in main fields and backup allocation in backup_* fields. Returns is_free=False if either path lacks spectrum.

Return type:

SpectrumResult

Note

This is a pure query method - does NOT allocate spectrum. Both paths must have free spectrum for success.

__init__(*args, **kwargs)

class fusion.interfaces.GroomingPipeline

Bases: Protocol

Protocol for traffic grooming algorithms.

Grooming attempts to pack multiple requests onto existing lightpaths that have available capacity, reducing new lightpath establishment.

Design Notes:

• MAY have side effects (modifies lightpath bandwidth allocations)

• Must support rollback for failed allocations

• Returns GroomingResult indicating success/partial/failure

Grooming Strategies:

• Full grooming: Entire request fits on existing lightpath(s)

• Partial grooming: Some bandwidth groomed, rest needs new lightpath

• No grooming: No suitable lightpaths found

Example:

class SimpleGrooming:
    def try_groom(
        self,
        request: Request,
        network_state: NetworkState,
    ) -> GroomingResult:
        candidates = network_state.get_lightpaths_between(
            request.source, request.destination
        )
        for lp in candidates:
            if lp.can_accommodate(request.bandwidth_gbps):
                lp.allocate_bandwidth(request.request_id, request.bandwidth_gbps)
                return GroomingResult.full(request.bandwidth_gbps, [lp.lightpath_id])
        return GroomingResult.no_grooming(request.bandwidth_gbps)

try_groom(request, network_state)

Attempt to groom request onto existing lightpaths.

Parameters:

• request (Request) – The request to groom (source, destination, bandwidth)

• network_state (NetworkState) – Current network state with active lightpaths

Returns:

GroomingResult with fully_groomed, partially_groomed, bandwidth_groomed_gbps, remaining_bandwidth_gbps, lightpaths_used, and forced_path fields

Return type:

GroomingResult

Note

If grooming succeeds (full or partial), modifies lightpath bandwidth allocations via Lightpath.allocate_bandwidth().

rollback_groom(request, lightpath_ids, network_state)

Rollback grooming allocations (e.g., after downstream failure).

Parameters:

• request (Request) – The request that was groomed

• lightpath_ids (list[int]) – Lightpath IDs to rollback

• network_state (NetworkState) – Current network state

Return type:

None

Note

Releases bandwidth from specified lightpaths via Lightpath.release_bandwidth(). Safe to call even if request wasn’t groomed on all lightpaths.

__init__(*args, **kwargs)

class fusion.interfaces.SNRPipeline

Bases: Protocol

Protocol for signal-to-noise ratio validation.

Validates that lightpaths meet SNR requirements for their modulation format, considering interference from other channels.

Design Notes:

• validate() is a pure query method (no side effects)

• recheck_affected() checks existing lightpaths after new allocation

• Returns SNRResult/SNRRecheckResult with pass/fail and measurements

SNR Considerations:

• ASE noise from amplifiers

• Non-linear interference (NLI) from other channels

• Crosstalk in multi-core fibers (MCF)

• Modulation-dependent thresholds

Example:

class GNModelSNR:
    def validate(
        self,
        lightpath: Lightpath,
        network_state: NetworkState,
    ) -> SNRResult:
        required_snr = get_threshold(lightpath.modulation)
        snr_db = calculate_gn_model_snr(lightpath, network_state)
        if snr_db >= required_snr:
            return SNRResult.success(snr_db, required_snr)
        return SNRResult.failure(snr_db, required_snr, "SNR below threshold")

validate(lightpath, network_state)

Validate SNR for a lightpath.

Parameters:

• lightpath (Lightpath) – The lightpath to validate (may be newly created or existing lightpath being rechecked)

• network_state (NetworkState) – Current network state with spectrum allocations

Returns:

SNRResult with passed, snr_db, required_snr_db, margin_db, failure_reason, and link_snr_values fields

Return type:

SNRResult

Note

This is a pure query method. Uses lightpath.modulation to determine threshold and considers interference from all other active lightpaths.

recheck_affected(new_lightpath_id, network_state, *, _affected_range_slots=5, slicing_flag=False)

Recheck SNR of existing lightpaths after new allocation.

Some existing lightpaths may be degraded by the new allocation’s interference. This method identifies them.

Parameters:

• new_lightpath_id (int) – ID of newly created lightpath

• network_state (NetworkState) – Current network state

• affected_range_slots (int) – Consider lightpaths within this many slots of the new allocation (default: 5)

• slicing_flag (bool) – If True, indicates this is a slicing context

• _affected_range_slots (int) –

Returns:

SNRRecheckResult with all_pass, degraded_lightpath_ids, violations, and checked_count fields

Return type:

SNRRecheckResult

Note

This is a pure query method. Only checks lightpaths on same links and in nearby spectrum. Used to trigger rollback if existing lightpaths are degraded.

__init__(*args, **kwargs)

class fusion.interfaces.SlicingPipeline

Bases: Protocol

Protocol for request slicing algorithms.

Slicing divides a large request into multiple smaller lightpaths when a single lightpath cannot accommodate the full bandwidth.

Design Notes:

• try_slice() is a pure query (checks feasibility)

• rollback_slices() removes created lightpaths on failure

• Used when normal allocation fails due to spectrum fragmentation

• Limited by config.max_slices

Slicing Strategies:

• Static slicing: Fixed slice size (e.g., 50 Gbps per slice)

• Dynamic slicing: Adaptive based on availability

Example:

class DynamicSlicing:
    def try_slice(
        self,
        request: Request,
        path: list[str],
        modulation: str,
        bandwidth_gbps: int,
        network_state: NetworkState,
        *,
        max_slices: int | None = None,
    ) -> SlicingResult:
        limit = max_slices or network_state.config.max_slices
        for num_slices in range(2, limit + 1):
            slice_bw = bandwidth_gbps // num_slices
            if can_allocate_slices(path, slice_bw, num_slices, network_state):
                return SlicingResult.sliced(num_slices, slice_bw, ...)
        return SlicingResult.failed()

try_slice(request, path, modulation, bandwidth_gbps, network_state, *, max_slices=None, spectrum_pipeline=None, snr_pipeline=None, connection_index=None, path_index=0, snr_accumulator=None, path_weight=None)

Attempt to slice request into multiple smaller allocations.

Parameters:

• request (Request) – The request being processed

• path (list[str]) – Route to use for slicing

• modulation (str) – Modulation format for slices

• bandwidth_gbps (int) – Total bandwidth to slice

• network_state (NetworkState) – Current network state

• max_slices (int | None) – Override config.max_slices

• spectrum_pipeline (SpectrumPipeline | None) – Pipeline for finding spectrum per slice

• snr_pipeline (SNRPipeline | None) – Pipeline for validating each slice

• connection_index (int | None) – External routing index for pre-calculated SNR lookup

• path_index (int) – Index of which k-path is being tried (0, 1, 2…)

• snr_accumulator (list[float] | None) – List to accumulate SNR values from failed attempts

• path_weight (float | None) – Path weight in km (from routing, for metrics tracking)

Returns:

SlicingResult with success, num_slices, slice_bandwidth_gbps, lightpaths_created, and total_bandwidth_gbps fields

Return type:

SlicingResult

Note

Pure query method by default. When spectrum_pipeline/snr_pipeline provided, may allocate slices.

rollback_slices(lightpath_ids, network_state)

Rollback all slices on failure.

Called when partial slice allocation fails and all created slices must be released.

Parameters:

• lightpath_ids (list[int]) – IDs of lightpaths to release

• network_state (NetworkState) – Network state to modify

Return type:

None

Note

Releases/deletes specified lightpaths from network_state. Safe to call with empty list.

__init__(*args, **kwargs)

class fusion.interfaces.ControlPolicy

Bases: Protocol

Protocol for control policies that select actions for resource allocation.

This protocol defines the interface for all path selection strategies in the FUSION simulation framework. Implementations include:

• Heuristic policies: Rule-based selection (first-fit, shortest-path)

• RL policies: Reinforcement learning agents (PPO, DQN, etc.)

• Supervised/unsupervised policies: Pre-trained neural networks or classifiers

• Composite policies: FallbackPolicy, TiebreakingPolicy

All policies must:

1. Respect feasibility: Only select paths where PathOption.is_feasible is True

2. Return valid indices: Return 0 to len(options)-1, or -1 for no valid action

3. Never mutate state: NetworkState must remain unchanged during select_action

4. Provide a name: Return descriptive name via get_name() for logging

Example

>>> policy = FirstFeasiblePolicy()
>>> action = policy.select_action(request, options, network_state)
>>> if action >= 0:
...     result = orchestrator.apply_action(action, request, options)
...     policy.update(request, action, result.reward)
...     logger.info(f"Policy {policy.get_name()} selected action {action}")
>>> else:
...     # No feasible path - request blocked
...     handle_blocking(request)

select_action(request, options, network_state)

Select an action (path index) for the given request.

This method is the core decision-making interface. It receives the current request, available path options (with feasibility information), and read-only network state. It returns the index of the selected path.

Parameters:

• request (Request) – The incoming request to serve. Contains source, destination, bandwidth requirements, and timing information.

• options (list[PathOption]) – List of available path options, each with path_index, path, weight_km, is_feasible, congestion, slots_needed, modulation. For protected paths, also includes backup_path, backup_feasible.

• network_state (NetworkState) – Current state of the network. This is read-only; policies must not modify network state.

Returns:

Path index (0 to len(options)-1) for the selected path, or -1 if no valid action exists.

Return type:

int

Note

• Policies MUST only return indices where options[i].is_feasible is True

• For protected paths, check options[i].both_paths_feasible for full protection

• Returning an infeasible index is undefined behavior (orchestrator may reject)

update(request, action, reward)

Update policy based on experience.

Called after an action is executed and the reward is computed. This enables online learning for RL policies. Heuristic and pre-trained supervised/unsupervised policies typically implement this as a no-op.

Parameters:

• request (Request) – The request that was served

• action (int) – The action (path index) that was taken

• reward (float) – The reward received. Typically positive for successful allocation, negative for blocking.

Return type:

None

Note

• Heuristic policies should implement this as pass

• RL policies may update internal state, replay buffers, etc.

• Supervised/unsupervised policies (pre-trained) typically implement as pass

• This method should not raise exceptions

get_name()

Return the policy name for logging and metrics.

This method enables meaningful logging messages and metrics tracking. Names should be descriptive and include relevant configuration.

Returns:

Human-readable policy name (e.g., “FirstFeasiblePolicy”, “RLPolicy(PPO)”, “SupervisedPolicy(pytorch)”)

Return type:

str

__init__(*args, **kwargs)

fusion.interfaces.PolicyAction

alias of int

---

Policies

fusion.policies

Routing and allocation policies.

PolicyFactory for instantiating control policies.

This module provides a factory to create policy instances based on configuration, supporting: - Heuristic policies (first_feasible, shortest, least_congested, random, load_balanced) - ML policies (via MLControlPolicy) - RL policies (via RLPolicy.from_file)

class fusion.policies.policy_factory.PolicyConfig

Bases: object

Configuration for policy instantiation.

Variables:

• policy_type (str) – Type of policy (“heuristic”, “ml”, “rl”).

• policy_name (str) – Name of specific policy variant.

• model_path (str | None) – Path to model file (for ml/rl types).

• fallback_policy (str) – Fallback policy name if primary fails.

• k_paths (int) – Number of candidate paths (for rl).

• seed (int | None) – Random seed (for random policies).

• alpha (float) – Balance parameter (for load_balanced policy).

• algorithm (str) – RL algorithm name (for rl type, e.g., “PPO”, “MaskablePPO”).

policy_type: str = 'heuristic'

policy_name: str = 'first_feasible'

model_path: str | None = None

fallback_policy: str = 'first_feasible'

k_paths: int = 3

seed: int | None = None

alpha: float = 0.5

algorithm: str = 'PPO'

__init__(policy_type='heuristic', policy_name='first_feasible', model_path=None, fallback_policy='first_feasible', k_paths=3, seed=None, alpha=0.5, algorithm='PPO')

Parameters:

• policy_type (str) –

• policy_name (str) –

• model_path (str | None) –

• fallback_policy (str) –

• k_paths (int) –

• seed (int | None) –

• alpha (float) –

• algorithm (str) –

Return type:

None

class fusion.policies.policy_factory.PolicyFactory

Bases: object

Factory for creating control policy instances.

This factory instantiates policies based on configuration, supporting multiple policy types:

1. Heuristic policies: Built-in rule-based policies

2. ML policies: Pre-trained PyTorch/sklearn/ONNX models

3. RL policies: Pre-trained Stable-Baselines3 models

Default policy is FirstFeasiblePolicy when not specified.

Example:

>>> config = PolicyConfig(policy_type="heuristic", policy_name="shortest")
>>> policy = PolicyFactory.create(config)
>>> action = policy.select_action(request, options, network_state)

static create(config=None)

Create a policy instance from configuration.

Parameters:

config (PolicyConfig | None) – Policy configuration. If None, creates FirstFeasiblePolicy.

Returns:

ControlPolicy instance.

Return type:

ControlPolicy

Raises:

ValueError – If policy type/name is unknown or model_path is missing.

static from_dict(config_dict)

Create policy from dictionary configuration.

Convenience method for creating policies from config files.

Parameters:

config_dict (dict[str, Any]) – Dictionary with policy configuration.

Returns:

ControlPolicy instance.

Return type:

ControlPolicy

static get_default_policy()

Get the default policy (FirstFeasiblePolicy).

Returns:

FirstFeasiblePolicy instance.

Return type:

ControlPolicy

Heuristic policies for path selection.

This module provides deterministic, rule-based path selection strategies implementing the ControlPolicy protocol. Available policies:

• FirstFeasiblePolicy: Select first feasible path (K-shortest first fit)

• ShortestFeasiblePolicy: Select shortest feasible path by distance

• LeastCongestedPolicy: Select least congested feasible path

• RandomFeasiblePolicy: Randomly select among feasible paths

• LoadBalancedPolicy: Balance path length and congestion

All policies implement the ControlPolicy protocol but do not learn from experience (update() is a no-op).

Example

>>> from fusion.policies.heuristic_policy import ShortestFeasiblePolicy
>>> policy = ShortestFeasiblePolicy()
>>> action = policy.select_action(request, options, network_state)

Note

Heuristic policies are typically used as: - Default selection strategies - Baselines for RL/ML comparison - Fallback when ML models fail

class fusion.policies.heuristic_policy.HeuristicPolicy

Bases: ABC

Abstract base class for heuristic path selection policies.

Heuristic policies are deterministic (except RandomFeasiblePolicy), rule-based strategies that select paths without learning. They serve as:

1. Default selection strategies

2. Baselines for RL/ML comparison

3. Fallback policies when ML models fail

All subclasses must implement select_action(). The update() method is a no-op since heuristics don’t learn.

This class implements the ControlPolicy protocol.

abstract select_action(request, options, network_state)

Select an action (path index) based on heuristic rule.

Parameters:

• request (Request) – The incoming request (may be used for context).

• options (list[PathOption]) – List of available PathOptions.

• network_state (NetworkState) – Current network state (read-only).

Returns:

Selected path index (0 to len(options)-1), or -1 if none valid.

Return type:

int

update(request, action, reward)

Update policy based on experience.

Heuristic policies do not learn, so this is a no-op. This is intentionally not abstract - subclasses inherit this no-op.

Parameters:

• request (Request) – The request that was served (ignored).

• action (int) – The action taken (ignored).

• reward (float) – The reward received (ignored).

Return type:

None

get_name()

Return the policy name for logging.

Returns:

Policy class name.

Return type:

str

class fusion.policies.heuristic_policy.FirstFeasiblePolicy

Bases: HeuristicPolicy

Select the first feasible path in index order.

This is the simplest heuristic. It iterates through options and returns the first path where is_feasible=True. Equivalent to “K-Shortest Path First Fit” when paths are pre-sorted by length.

Selection Logic:

1. Iterate through options in index order

2. Return first option where is_feasible=True

3. Return -1 if no feasible option exists

Time Complexity: O(n) worst case, O(1) best case Space Complexity: O(1)

select_action(request, options, network_state)

Select the first feasible path.

Parameters:

• request (Request) –

• options (list[PathOption]) –

• network_state (NetworkState) –

Return type:

int

class fusion.policies.heuristic_policy.ShortestFeasiblePolicy

Bases: HeuristicPolicy

Select the shortest feasible path by distance.

Finds all feasible paths and selects the one with minimum weight_km (path length in kilometers).

Selection Logic:

1. Filter to feasible options

2. Find option with minimum weight_km

3. Return its path_index, or -1 if none feasible

Tie Breaking: First occurrence when tied on weight_km.

Time Complexity: O(n) Space Complexity: O(n) for feasible list

select_action(request, options, network_state)

Select the shortest feasible path by weight_km.

Parameters:

• request (Request) –

• options (list[PathOption]) –

• network_state (NetworkState) –

Return type:

int

class fusion.policies.heuristic_policy.LeastCongestedPolicy

Bases: HeuristicPolicy

Select the least congested feasible path.

Prioritizes paths with lower congestion values to distribute load across the network and reduce fragmentation.

Selection Logic:

1. Filter to feasible options

2. Find option with minimum congestion (0.0 to 1.0)

3. Return its path_index, or -1 if none feasible

Tie Breaking: First occurrence when tied on congestion.

Time Complexity: O(n) Space Complexity: O(n) for feasible list

select_action(request, options, network_state)

Select the least congested feasible path.

Parameters:

• request (Request) –

• options (list[PathOption]) –

• network_state (NetworkState) –

Return type:

int

class fusion.policies.heuristic_policy.RandomFeasiblePolicy

Bases: HeuristicPolicy

Randomly select among feasible paths.

Uniformly samples from all feasible paths. Useful for:

• Exploration during training

• Baseline comparison (random performance)

• Load distribution across multiple paths

Uses numpy’s random number generator with optional seed.

Selection Logic:

1. Filter to feasible options

2. Uniformly sample one option

3. Return its path_index, or -1 if none feasible

Time Complexity: O(n) Space Complexity: O(n) for feasible list

Variables:

• _rng (numpy.random.Generator) – Numpy random generator.

• _seed (int | None) – Original seed for reset.

__init__(seed=None)

Initialize with optional random seed.

Parameters:

seed (int | None) – Random seed for reproducibility. If None, uses system entropy (non-reproducible).

Return type:

None

select_action(request, options, network_state)

Randomly select a feasible path.

Parameters:

• request (Request) –

• options (list[PathOption]) –

• network_state (NetworkState) –

Return type:

int

reset_rng(seed=None)

Reset the random number generator.

Parameters:

seed (int | None) – New seed. If None, uses the original seed.

Return type:

None

get_name()

Return policy name including seed.

Returns:

Policy name with seed if set.

Return type:

str

class fusion.policies.heuristic_policy.LoadBalancedPolicy

Bases: HeuristicPolicy

Select path balancing length and congestion.

Combines path length (weight_km) and congestion into a weighted score:

score = alpha * normalized_length + (1 - alpha) * congestion

Where normalized_length = weight_km / max_weight_km among feasible paths.

Alpha Parameter:

• 0.0: Pure congestion-based (same as LeastCongestedPolicy)

• 0.5: Equal weight to length and congestion (default)

• 1.0: Pure length-based (same as ShortestFeasiblePolicy)

Selection Logic:

1. Filter to feasible options

2. Normalize weight_km to [0, 1] range

3. Compute weighted score for each option

4. Return option with minimum score

Tie Breaking: First occurrence when tied on score.

Time Complexity: O(n) Space Complexity: O(n) for feasible list

Variables:

_alpha (float) – Weight for path length (0.0 to 1.0).

__init__(alpha=0.5)

Initialize with load balancing weight.

Parameters:

alpha (float) – Weight for path length (0.0 to 1.0). 0.0 = prioritize low congestion, 1.0 = prioritize short length, 0.5 = equal balance (default).

Raises:

ValueError – If alpha is not in [0, 1] range.

Return type:

None

property alpha: float

Current alpha value.

Returns:

The alpha weight parameter.

Return type:

float

select_action(request, options, network_state)

Select path with minimum weighted score.

Parameters:

• request (Request) –

• options (list[PathOption]) –

• network_state (NetworkState) –

Return type:

int

get_name()

Return policy name including alpha.

Returns:

Policy name with alpha parameter.

Return type:

str

RL Policy wrapper for Stable-Baselines3 models.

This module provides the RLPolicy class that wraps pre-trained SB3 models to implement the ControlPolicy protocol, enabling unified policy handling in the SDNOrchestrator.

class fusion.policies.rl_policy.RLPolicy

Bases: object

Wrapper enabling SB3 models to implement ControlPolicy.

This adapter bridges the existing RL infrastructure with the ControlPolicy protocol. It handles:

1. Observation building: Converts (request, options, state) to RL observation

2. Action masking: Enforces feasibility constraints during prediction

3. Action conversion: Converts SB3 action to path index

The wrapped model is pre-trained and does not learn online. Use UnifiedSimEnv and SB3’s learn() for online training.

Variables:

• model (BaseAlgorithm) – Pre-trained SB3 model (PPO, DQN, A2C, etc.).

• k_paths (int) – Number of path options (for observation space).

Example:

>>> from stable_baselines3 import PPO
>>> model = PPO.load("trained_model.zip")
>>> policy = RLPolicy(model)
>>> action = policy.select_action(request, options, network_state)

__init__(model, adapter=None, k_paths=5)

Initialize RLPolicy with a trained SB3 model.

Parameters:

• model (BaseAlgorithm) – Pre-trained SB3 model with predict() method.

• adapter (RLSimulationAdapter | None) – Optional adapter for observation building. If None, uses internal observation construction.

• k_paths (int) – Expected number of path options (for obs space size).

Raises:

ValueError – If model does not have predict() method.

Return type:

None

select_action(request, options, network_state)

Select action using the trained SB3 model.

Builds an observation from the inputs, generates an action mask from feasibility flags, and uses the model to predict an action.

Parameters:

• request (Request) – The incoming request to serve.

• options (list[PathOption]) – Available path options with feasibility information.

• network_state (NetworkState) – Current state of the network.

Returns:

Path index (0 to len(options)-1), or -1 if no valid action.

Return type:

int

Note

The model must support action masking. For models trained with sb3-contrib’s MaskablePPO, the action_masks parameter is used. For standard models, masking is applied post-prediction.

update(request, action, reward)

Update policy based on experience.

RLPolicy wraps pre-trained models that do not learn online. This method is a no-op to satisfy the ControlPolicy protocol.

For online RL training, use UnifiedSimEnv with SB3’s learn() method.

Parameters:

• request (Request) – The request that was served (ignored).

• action (int) – The action taken (ignored).

• reward (float) – The reward received (ignored).

Return type:

None

get_name()

Return policy name for logging and metrics.

Returns:

String identifying this policy and underlying model.

Return type:

str

set_adapter(adapter)

Set the RL simulation adapter for observation building.

Parameters:

adapter (RLSimulationAdapter) – RL simulation adapter for observation building.

Return type:

None

classmethod from_file(model_path, algorithm='PPO', **kwargs)

Load RLPolicy from a saved model file.

Parameters:

• model_path (str) – Path to saved model (e.g., “model.zip”).

• algorithm (str) – SB3 algorithm name (“PPO”, “DQN”, “A2C”, “MaskablePPO”, etc.).

• kwargs (Any) – Additional arguments passed to RLPolicy.__init__.

Returns:

RLPolicy wrapping the loaded model.

Return type:

RLPolicy

Raises:

ValueError – If algorithm is unknown/not installed.

Example:

>>> policy = RLPolicy.from_file("trained_ppo.zip", algorithm="PPO")

fusion.policies.rl_policy.RLControlPolicy

alias of RLPolicy

ML Control Policy for path selection using pre-trained models.

This module provides MLControlPolicy, which loads pre-trained ML models (PyTorch, sklearn, ONNX) and uses them for path selection inference.

MLControlPolicy implements the ControlPolicy protocol with: - Multi-framework support via file extension detection - Robust fallback to heuristic policies on errors - Action masking for feasibility constraints - Feature engineering matching RL observation space

MLControlPolicy is deployment-only: no online training, update() is no-op.

Example

>>> from fusion.policies.ml_policy import MLControlPolicy
>>> policy = MLControlPolicy("model.pt", fallback_type="first_feasible")
>>> action = policy.select_action(request, options, network_state)

class fusion.policies.ml_policy.FeatureBuilder

Bases: object

Build feature vectors for ML model inference.

Creates fixed-size feature vectors from Request and PathOption inputs, with padding for variable numbers of paths.

The feature layout matches the RL training observation space, ensuring model compatibility.

Variables:

• k_paths (int) – Expected number of paths (for padding).

• features_per_path (int) – Number of features extracted per path.

Example:

>>> builder = FeatureBuilder(k_paths=5)
>>> features = builder.build(request, options, network_state)
>>> features.shape
(21,)  # 1 + 5*4

FEATURES_PER_PATH = 4

MAX_BANDWIDTH_GBPS = 1000.0

MAX_WEIGHT_KM = 10000.0

MAX_SLOTS = 100.0

__init__(k_paths=5)

Initialize feature builder.

Parameters:

k_paths (int) – Expected number of path options.

Return type:

None

property feature_size: int

Total size of feature vector.

Returns:

Size of the feature vector.

Return type:

int

build(request, options, network_state)

Build feature vector from inputs.

Parameters:

• request (Request) – The request being processed.

• options (list[PathOption]) – Available path options.

• network_state (NetworkState) – Current network state (for future extensions).

Returns:

Feature vector of shape (feature_size,).

Return type:

np.ndarray

class fusion.policies.ml_policy.ModelWrapper

Bases: Protocol

Protocol for model wrappers providing predict interface.

predict(features)

Predict action scores from features.

Parameters:

features (np.ndarray) – Feature array of shape (feature_size,) or (batch, feature_size).

Returns:

Scores/logits for each action of shape (k_paths,) or (batch, k_paths).

Return type:

np.ndarray

__init__(*args, **kwargs)

class fusion.policies.ml_policy.TorchModelWrapper

Bases: object

Wrapper for PyTorch models.

__init__(model, device='cpu')

Initialize torch model wrapper.

Parameters:

• model (Any) – PyTorch nn.Module.

• device (str) – Device to run inference on.

Return type:

None

predict(features)

Run inference through PyTorch model.

Parameters:

features (np.ndarray) – Input feature array.

Returns:

Model output scores.

Return type:

np.ndarray

class fusion.policies.ml_policy.SklearnModelWrapper

Bases: object

Wrapper for sklearn models.

__init__(model)

Initialize sklearn model wrapper.

Parameters:

model (Any) – sklearn model with predict_proba or predict method.

Return type:

None

predict(features)

Run inference through sklearn model.

Parameters:

features (np.ndarray) – Input feature array.

Returns:

Model output scores or predictions.

Return type:

np.ndarray

class fusion.policies.ml_policy.OnnxModelWrapper

Bases: object

Wrapper for ONNX models.

__init__(session)

Initialize ONNX model wrapper.

Parameters:

session (Any) – onnxruntime InferenceSession.

Return type:

None

predict(features)

Run inference through ONNX model.

Parameters:

features (np.ndarray) – Input feature array.

Returns:

Model output scores.

Return type:

np.ndarray

class fusion.policies.ml_policy.CallableModelWrapper

Bases: object

Wrapper for callable models (functions or objects with __call__).

__init__(model)

Initialize callable model wrapper.

Parameters:

model (Callable[[np.ndarray], np.ndarray]) – Any callable that takes features and returns scores.

Return type:

None

predict(features)

Run inference through callable.

Parameters:

features (np.ndarray) – Input feature array.

Returns:

Model output scores.

Return type:

np.ndarray

fusion.policies.ml_policy.load_model(model_path, device='cpu')

Load a model based on file extension.

Supported formats:

• .pt, .pth: PyTorch models

• .joblib, .pkl: sklearn models (via joblib)

• .onnx: ONNX models

Parameters:

• model_path (str) – Path to model file.

• device (str) – Device for PyTorch models (“cpu”, “cuda”, “mps”).

Returns:

ModelWrapper with predict() method.

Return type:

ModelWrapper

Raises:

• FileNotFoundError – If model file doesn’t exist.

• ValueError – If file extension not supported.

• ImportError – If required framework not installed.

class fusion.policies.ml_policy.MLControlPolicy

Bases: object

ML-based control policy for path selection.

Loads pre-trained ML models and uses them for deterministic inference. Implements robust fallback to heuristic policies when model fails.

This is a deployment-only policy: update() is a no-op.

Variables:

• model (ModelWrapper) – Wrapped model with predict() interface.

• fallback (HeuristicPolicy) – Fallback heuristic policy.

• feature_builder (FeatureBuilder) – Feature vector constructor.

Example:

>>> policy = MLControlPolicy("model.pt", fallback_type="first_feasible")
>>> action = policy.select_action(request, options, network_state)
>>> print(policy.get_stats())  # View fallback statistics

__init__(model_path=None, model=None, device='cpu', k_paths=5, fallback_policy=None, fallback_type='first_feasible')

Initialize ML control policy.

Parameters:

• model_path (str | None) – Path to model file. Mutually exclusive with model.

• model (ModelWrapper | None) – Pre-loaded model wrapper. Mutually exclusive with model_path.

• device (str) – Device for PyTorch models (“cpu”, “cuda”, “mps”).

• k_paths (int) – Expected number of path options (for feature builder).

• fallback_policy (Any | None) – Explicit fallback policy instance.

• fallback_type (str) – Fallback type if policy not provided: “first_feasible” (default), “shortest_feasible”, “least_congested”, “random”.

Raises:

• ValueError – If neither or both model_path and model provided.

• FileNotFoundError – If model file doesn’t exist.

• ImportError – If required framework not installed.

Return type:

None

property fallback: ControlPolicy

Current fallback policy.

Returns:

The fallback policy instance.

Return type:

ControlPolicy

property fallback_rate: float

Percentage of calls that used fallback.

Returns:

Fallback rate as a fraction (0.0 to 1.0).

Return type:

float

get_stats()

Get fallback statistics.

Returns:

Dictionary with total_calls, fallback_calls, fallback_rate, and error_types.

Return type:

dict[str, Any]

reset_stats()

Reset fallback statistics.

Return type:

None

select_action(request, options, network_state)

Select an action using ML model with fallback.

Flow:

1. Build features from inputs

2. Run model inference

3. Apply action masking (infeasible -> -inf)

4. Select argmax action

5. Validate action is feasible

6. On any error/invalid action: use fallback

Parameters:

• request (Request) – The request to serve.

• options (list[PathOption]) – Available path options.

• network_state (NetworkState) – Current network state.

Returns:

Path index (0 to len(options)-1), or -1 if no valid action.

Return type:

int

update(request, action, reward)

Update policy based on experience.

MLControlPolicy is deployment-only and does not learn online. This method is a no-op.

Parameters:

• request (Request) – The request that was served (ignored).

• action (int) – The action taken (ignored).

• reward (float) – The reward received (ignored).

Return type:

None

get_name()

Return the policy name for logging.

Returns:

Policy name with model path.

Return type:

str

---

Pipelines

fusion.pipelines

Processing pipelines for routing, spectrum, and SNR.

Pipeline implementations for FUSION simulation.

This package contains pipeline implementations that can be selected by PipelineFactory based on SimulationConfig.

Current implementations:

• ProtectedRoutingPipeline: 1+1 protection routing

• StandardSlicingPipeline: Request slicing across multiple lightpaths

Routing Strategies:

• KShortestPathStrategy: Basic k-shortest paths routing

• LoadBalancedStrategy: Routing considering link utilization

• ProtectionAwareStrategy: Routing for disjoint path pairs

Protection Pipeline:

• DisjointPathFinder: Link-disjoint and node-disjoint path algorithms

• ProtectionPipeline: 1+1 dedicated protection allocation

class fusion.pipelines.RoutingStrategy

Bases: Protocol

Protocol for routing strategies.

Defines the interface that all routing strategy implementations must follow. Strategies are responsible for selecting candidate routes between source and destination nodes.

Design Notes:

• Strategies are stateless (configuration passed to __init__)

• select_routes() is a pure function (no side effects)

• Returns RouteResult from fusion.domain.results

Example

>>> strategy = KShortestPathStrategy(k=3)
>>> result = strategy.select_routes(
...     "A", "Z", 100, network_state
... )
>>> for path in result.paths:
...     print(path)

property name: str

Strategy name for logging and configuration.

select_routes(source, destination, bandwidth_gbps, network_state, constraints=None)

Select candidate routes for the request.

Parameters:

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

• bandwidth_gbps (int) – Required bandwidth (for modulation selection).

• network_state (NetworkState) – Current network state.

• constraints (RouteConstraints | None) – Optional routing constraints.

Returns:

RouteResult with ordered candidate routes (best first).

Return type:

RouteResult

__init__(*args, **kwargs)

class fusion.pipelines.RouteConstraints

Bases: object

Constraints for route selection.

Used to specify requirements and exclusions when finding routes, particularly for protection path computation where the backup must avoid links/nodes used by the primary path.

Variables:

• max_hops (int | None) – Maximum number of hops allowed (None = no limit).

• max_length_km (float | None) – Maximum path length in km (None = no limit).

• min_bandwidth_gbps (int | None) – Minimum available bandwidth required.

• exclude_links (frozenset[tuple[str, str]]) – Links to avoid (for disjoint paths).

• exclude_nodes (frozenset[str]) – Nodes to avoid (for node-disjoint paths).

• required_modulation (str | None) – Force specific modulation format.

• protection_mode (bool) – True if finding protection/backup path.

Example

>>> # Find backup path avoiding primary path links
>>> constraints = RouteConstraints(
...     exclude_links={("A", "B"), ("B", "C")},
...     protection_mode=True,
... )

max_hops: int | None = None

max_length_km: float | None = None

min_bandwidth_gbps: int | None = None

exclude_links: frozenset[tuple[str, str]]

exclude_nodes: frozenset[str]

required_modulation: str | None = None

protection_mode: bool = False

with_exclusions(links=None, nodes=None)

Create new constraints with additional exclusions.

Parameters:

• links (set[tuple[str, str]] | None) – Additional links to exclude.

• nodes (set[str] | None) – Additional nodes to exclude.

Returns:

New RouteConstraints with merged exclusions.

Return type:

RouteConstraints

__init__(max_hops=None, max_length_km=None, min_bandwidth_gbps=None, exclude_links=<factory>, exclude_nodes=<factory>, required_modulation=None, protection_mode=False)

Parameters:

• max_hops (int | None) –

• max_length_km (float | None) –

• min_bandwidth_gbps (int | None) –

• exclude_links (frozenset[tuple[str, str]]) –

• exclude_nodes (frozenset[str]) –

• required_modulation (str | None) –

• protection_mode (bool) –

Return type:

None

class fusion.pipelines.KShortestPathStrategy

Bases: object

K-shortest paths routing strategy.

Finds the k shortest paths between source and destination using path weight (typically distance in km). This is the default routing strategy.

Variables:

• k (int) – Number of candidate paths to find.

• _name (str) – Strategy identifier.

Example

>>> strategy = KShortestPathStrategy(k=3)
>>> result = strategy.select_routes("A", "Z", 100, network_state)

__init__(k=3)

Initialize k-shortest path strategy.

Parameters:

k (int) – Number of candidate paths to compute.

Return type:

None

property name: str

Strategy name.

select_routes(source, destination, bandwidth_gbps, network_state, constraints=None)

Select k-shortest paths between source and destination.

Parameters:

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

• bandwidth_gbps (int) – Required bandwidth.

• network_state (NetworkState) – Current network state.

• constraints (RouteConstraints | None) – Optional routing constraints.

Returns:

RouteResult with up to k candidate paths.

Return type:

RouteResult

class fusion.pipelines.LoadBalancedStrategy

Bases: object

Load-balanced routing considering link utilization.

Extends k-shortest paths by scoring routes based on both path length and link utilization, preferring less congested paths.

Variables:

• k (int) – Number of candidate paths to consider.

• utilization_weight (float) – Weight for utilization in scoring (0-1).

• _name (str) – Strategy identifier.

Scoring:

score = (1 - util_weight) * (1/length) + util_weight * (1 - utilization)
Higher score = better path

Example

>>> strategy = LoadBalancedStrategy(k=5, utilization_weight=0.5)
>>> result = strategy.select_routes("A", "Z", 100, network_state)

__init__(k=3, utilization_weight=0.5)

Initialize load-balanced strategy.

Parameters:

• k (int) – Number of candidate paths to consider.

• utilization_weight (float) – Weight for utilization (0=length only, 1=util only).

Return type:

None

property name: str

Strategy name.

select_routes(source, destination, bandwidth_gbps, network_state, constraints=None)

Select routes balancing path length and utilization.

Parameters:

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

• bandwidth_gbps (int) – Required bandwidth.

• network_state (NetworkState) – Current network state.

• constraints (RouteConstraints | None) – Optional routing constraints.

Returns:

RouteResult with routes sorted by combined score.

Return type:

RouteResult

class fusion.pipelines.ProtectionAwareStrategy

Bases: object

Strategy for finding disjoint working/protection paths.

Used for 1+1 protection scenarios where the backup path must be disjoint from the primary path (no shared links or nodes).

Variables:

• node_disjoint (bool) – If True, paths must be node-disjoint (stricter).

• _name (str) – Strategy identifier.

Usage:

1. First call with protection_mode=False to get working path

2. Second call with protection_mode=True and exclude_links set to working path’s links

Example

>>> strategy = ProtectionAwareStrategy(node_disjoint=True)
>>> # Get working path
>>> working = strategy.select_routes("A", "Z", 100, state)
>>> # Get protection path avoiding working path
>>> constraints = RouteConstraints(
...     exclude_links=get_path_links(working.best_path),
...     protection_mode=True,
... )
>>> backup = strategy.select_routes("A", "Z", 100, state, constraints)

__init__(node_disjoint=False)

Initialize protection-aware strategy.

Parameters:

node_disjoint (bool) – If True, require node-disjoint paths.

Return type:

None

property name: str

Strategy name.

select_routes(source, destination, bandwidth_gbps, network_state, constraints=None)

Select routes for protection scenarios.

When protection_mode=True in constraints, finds paths avoiding the excluded links/nodes (typically the working path).

Parameters:

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

• bandwidth_gbps (int) – Required bandwidth.

• network_state (NetworkState) – Current network state.

• constraints (RouteConstraints | None) – Constraints including exclusions for protection.

Returns:

RouteResult with candidate paths.

Return type:

RouteResult

find_disjoint_pair(source, destination, bandwidth_gbps, network_state)

Find a pair of disjoint paths (working + protection).

Convenience method that finds both paths in one call.

Parameters:

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

• bandwidth_gbps (int) – Required bandwidth.

• network_state (NetworkState) – Current network state.

Returns:

Tuple of (working_result, protection_result).

Return type:

tuple[RouteResult, RouteResult]

class fusion.pipelines.DisjointnessType

Bases: Enum

Type of path disjointness.

LINK = 'link'

NODE = 'node'

class fusion.pipelines.DisjointPathFinder

Bases: object

Finds disjoint path pairs for 1+1 protection.

Supports two disjointness modes:

• LINK: Paths share no common edges (may share intermediate nodes)

• NODE: Paths share no common intermediate nodes (stronger guarantee)

For link-disjoint paths, wraps the existing OnePlusOneProtection algorithm (Suurballe’s via NetworkX edge_disjoint_paths). For node-disjoint paths, uses node removal in residual graph.

Variables:

disjointness (DisjointnessType) – Type of disjointness (LINK or NODE).

Example

>>> finder = DisjointPathFinder(DisjointnessType.LINK)
>>> paths = finder.find_disjoint_pair(topology, "A", "D")
>>> if paths:
...     primary, backup = paths

__init__(disjointness=DisjointnessType.LINK)

Initialize DisjointPathFinder.

Parameters:

disjointness (DisjointnessType) – Type of disjointness to enforce.

Return type:

None

find_disjoint_pair(topology, source, destination)

Find a disjoint path pair between source and destination.

Parameters:

• topology (nx.Graph) – Network topology graph.

• source (str) – Source node identifier.

• destination (str) – Destination node identifier.

Returns:

Tuple of (primary_path, backup_path) or None if not possible.

Return type:

tuple[list[str], list[str]] | None

find_all_disjoint_paths(topology, source, destination, max_paths=10)

Find all disjoint paths between source and destination.

Parameters:

• topology (nx.Graph) – Network topology graph.

• source (str) – Source node.

• destination (str) – Destination node.

• max_paths (int) – Maximum paths to return.

Returns:

List of disjoint paths.

Return type:

list[list[str]]

are_link_disjoint(path1, path2)

Check if two paths are link-disjoint.

Parameters:

• path1 (list[str]) – First path as list of node IDs.

• path2 (list[str]) – Second path as list of node IDs.

Returns:

True if paths share no common edges (in either direction).

Return type:

bool

are_node_disjoint(path1, path2)

Check if two paths are node-disjoint (except endpoints).

Parameters:

• path1 (list[str]) – First path as list of node IDs.

• path2 (list[str]) – Second path as list of node IDs.

Returns:

True if paths share no common intermediate nodes.

Return type:

bool

class fusion.pipelines.ProtectionPipeline

Bases: object

Pipeline for 1+1 dedicated path protection.

Provides methods to:

• Find disjoint path pairs (link or node disjoint)

• Allocate the SAME spectrum on both primary and backup paths

• Integrate with NetworkState for lightpath creation

1+1 Dedicated Protection:

• Both primary and backup paths are pre-provisioned

• SAME spectrum slots are allocated on BOTH paths

• Traffic is transmitted on both paths simultaneously

• Receiver monitors primary and switches to backup on failure

• Fast switchover (typically < 50ms)

Variables:

• disjoint_finder (DisjointPathFinder) – DisjointPathFinder for path computation.

• switchover_latency_ms (float) – Protection switchover latency (default 50ms).

Example

>>> pipeline = ProtectionPipeline(DisjointnessType.LINK)
>>> result = pipeline.allocate_protected(
...     primary_path=["A", "B", "C"],
...     backup_path=["A", "D", "C"],
...     slots_needed=8,
...     network_state=state,
...     core=0,
...     band="c",
... )
>>> if result.success:
...     print(f"Allocated slots {result.start_slot}-{result.end_slot}")

__init__(disjointness=DisjointnessType.LINK, switchover_latency_ms=50.0)

Initialize ProtectionPipeline.

Parameters:

• disjointness (DisjointnessType) – Type of path disjointness (LINK or NODE).

• switchover_latency_ms (float) – Switchover latency in milliseconds.

Return type:

None

property disjointness: DisjointnessType

Current disjointness mode.

find_protected_paths(topology, source, destination)

Find a disjoint path pair for protection.

Parameters:

• topology (nx.Graph) – Network topology graph.

• source (str) – Source node ID.

• destination (str) – Destination node ID.

Returns:

Tuple of (primary_path, backup_path) or None if not possible.

Return type:

tuple[list[str], list[str]] | None

allocate_protected(primary_path, backup_path, slots_needed, network_state, core=0, band='c')

Allocate spectrum on both primary and backup paths.

For 1+1 dedicated protection, allocates the SAME spectrum slots on both paths to enable fast switchover.

Parameters:

• primary_path (list[str]) – Primary path node sequence.

• backup_path (list[str]) – Backup path node sequence.

• slots_needed (int) – Number of spectrum slots needed.

• network_state (NetworkStateLike) – Current network state.

• core (int) – Core index (default 0).

• band (str) – Band identifier (default “c”).

Returns:

ProtectedAllocationResult with spectrum assignment or failure reason.

Return type:

ProtectedAllocationResult

Algorithm:

1. Get spectrum availability on both paths (bitwise arrays)

2. Compute intersection (common free blocks)

3. Find first-fit contiguous block satisfying slots_needed

4. Return allocation details

create_protected_lightpath(network_state, primary_path, backup_path, start_slot, end_slot, core, band, modulation, bandwidth_gbps, path_weight_km, guard_slots=0)

Create a protected lightpath with both primary and backup paths.

Uses NetworkState.create_lightpath() which already supports protection. Allocates spectrum on both paths with the same slot range.

Parameters:

• network_state (NetworkState) – Network state to create lightpath in.

• primary_path (list[str]) – Primary path node sequence.

• backup_path (list[str]) – Backup path node sequence.

• start_slot (int) – Starting spectrum slot.

• end_slot (int) – Ending spectrum slot (exclusive).

• core (int) – Core index.

• band (str) – Band identifier.

• modulation (str) – Modulation format.

• bandwidth_gbps (int) – Bandwidth in Gbps.

• path_weight_km (float) – Primary path weight in km.

• guard_slots (int) – Number of guard slots (default 0).

Returns:

Created Lightpath with protection fields set.

Return type:

Lightpath

Raises:

ValueError – If spectrum not available on either path.

verify_disjointness(path1, path2)

Verify that two paths meet the disjointness requirement.

Parameters:

• path1 (list[str]) – First path.

• path2 (list[str]) – Second path.

Returns:

True if paths are disjoint according to current mode.

Return type:

bool

get_switchover_latency()

Get the protection switchover latency in milliseconds.

Return type:

float

class fusion.pipelines.ProtectedAllocationResult

Bases: object

Result of protected spectrum allocation.

Variables:

• success (bool) – Whether allocation succeeded.

• start_slot (int) – Starting spectrum slot (same on both paths).

• end_slot (int) – Ending spectrum slot (exclusive).

• failure_reason (str | None) – Reason for failure if success=False.

success: bool

start_slot: int = -1

end_slot: int = -1

failure_reason: str | None = None

classmethod no_disjoint_paths()

Create result for no disjoint paths found.

Return type:

ProtectedAllocationResult

classmethod no_common_spectrum()

Create result for no common spectrum available.

Return type:

ProtectedAllocationResult

classmethod allocated(start_slot, end_slot)

Create result for successful allocation.

Parameters:

• start_slot (int) –

• end_slot (int) –

Return type:

ProtectedAllocationResult

__init__(success, start_slot=-1, end_slot=-1, failure_reason=None)

Parameters:

• success (bool) –

• start_slot (int) –

• end_slot (int) –

• failure_reason (str | None) –

Return type:

None

---

Modules

fusion.modules

Algorithm implementations for routing, spectrum assignment, and SNR.

fusion.modules.routing

FUSION Routing Module.

This package contains various routing algorithm implementations for optical network path selection, including:

• K-Shortest Path routing

• Congestion-aware routing

• Fragmentation-aware routing

• Least congested link routing

• Non-linear impairment (NLI) aware routing

• Cross-talk (XT) aware routing

All algorithms implement the AbstractRoutingAlgorithm interface and can be accessed through the RoutingRegistry for dynamic algorithm selection.

class fusion.modules.routing.RoutingRegistry

Bases: object

Registry for managing routing algorithm implementations.

__init__()

Initialize the routing registry.

Return type:

None

register(name, algorithm_class)

Register a routing algorithm.

Parameters:

• name (str) – Unique name for the algorithm.

• algorithm_class (Any) – Class that implements AbstractRoutingAlgorithm.

Raises:

• TypeError – If algorithm_class doesn’t implement AbstractRoutingAlgorithm.

• ValueError – If name is already registered.

Return type:

None

get(name)

Get a routing algorithm class by name.

Parameters:

name (str) – Name of the algorithm.

Returns:

Algorithm class that implements AbstractRoutingAlgorithm.

Return type:

Any

Raises:

KeyError – If algorithm is not found.

create(name, engine_props, sdn_props)

Create an instance of a routing algorithm.

Parameters:

• name (str) – Name of the algorithm.

• engine_props (dict) – Engine configuration properties.

• sdn_props (object) – SDN controller properties.

Returns:

Configured routing algorithm instance.

Return type:

Any

list_algorithms()

List all registered algorithm names.

Returns:

List of algorithm names.

Return type:

list[str]

get_algorithm_info(name)

Get information about a specific algorithm.

Parameters:

name (str) – Name of the algorithm.

Returns:

Dictionary with algorithm information including name, class, module, supported topologies, and description.

Return type:

dict[str, str]

validate_algorithm(name, topology)

Validate that an algorithm can work with the given topology.

Parameters:

• name (str) – Name of the algorithm.

• topology (Any) – Network topology to validate against.

Returns:

True if algorithm supports the topology.

Return type:

Any

fusion.modules.routing.create_algorithm(name, engine_props, sdn_props)

Create an algorithm instance from the global registry.

Parameters:

• name (str) – Name of the algorithm.

• engine_props (dict) – Engine configuration properties.

• sdn_props (object) – SDN controller properties.

Returns:

Configured routing algorithm instance.

Return type:

Any

fusion.modules.routing.get_algorithm(name)

Get an algorithm class from the global registry.

Parameters:

name (str) – Name of the algorithm.

Returns:

Algorithm class that implements AbstractRoutingAlgorithm.

Return type:

Any

Raises:

KeyError – If algorithm is not found.

fusion.modules.routing.list_routing_algorithms()

List all available routing algorithms.

Returns:

List of registered algorithm names.

Return type:

list[str]

fusion.modules.routing.get_routing_algorithm_info(name)

Get information about a routing algorithm.

Parameters:

name (str) – Name of the algorithm.

Returns:

Dictionary with algorithm information including name, class, module, supported topologies, and description.

Return type:

dict[str, str]

class fusion.modules.routing.KShortestPath

Bases: AbstractRoutingAlgorithm

K-Shortest Path routing algorithm.

This algorithm finds the k shortest paths between source and destination based on hop count or other specified weights.

__init__(engine_props, sdn_props)

Initialize K-Shortest Path routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘k_shortest_path’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (nx.Graph) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find a route from source to destination for the given request.

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k shortest paths between source and destination.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths, where each path is a list of nodes.

Return type:

list[list[Any]]

update_weights(topology)

Update edge weights based on current network state.

This implementation doesn’t dynamically update weights. Subclasses can override this method for adaptive routing.

Parameters:

topology (nx.Graph) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average hop count, k value, and weight metric.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.CongestionAwareRouting

Bases: AbstractRoutingAlgorithm

Congestion-aware routing algorithm.

This algorithm finds paths by considering network congestion levels, selecting the path with the least congested link.

__init__(engine_props, sdn_props)

Initialize congestion-aware routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘congestion_aware’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (Any) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find a route from source to destination using congestion-aware k-shortest.

For the first k shortest-length candidate paths we compute:

score = alpha * mean_path_congestion + (1 - alpha) * (path_len / max_len)

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k paths ordered by congestion level.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths ordered by congestion (least congested first).

Return type:

list[list[Any]]

update_weights(topology)

Update congestion weights based on current network state.

Parameters:

topology (Any) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average congestion, and total congestion considered.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.LeastCongestedRouting

Bases: AbstractRoutingAlgorithm

Least congested link routing algorithm.

This algorithm finds paths with minimum hops ± 1 and selects the one with the least congested bottleneck link (not the mean congestion).

__init__(engine_props, sdn_props)

Initialize least congested routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘least_congested’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (Any) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find the least congested path in the network.

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k paths ordered by congestion level.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths ordered by congestion (least congested first).

Return type:

list[list[Any]]

update_weights(topology)

Update congestion weights based on current network state.

For least congested routing, we don’t pre-compute weights since we evaluate congestion dynamically.

Parameters:

topology (Any) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average congestion, and total congestion considered.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.FragmentationAwareRouting

Bases: AbstractRoutingAlgorithm

Fragmentation-aware routing algorithm.

This algorithm finds paths by considering spectrum fragmentation levels, selecting the path with the least fragmentation.

__init__(engine_props, sdn_props)

Initialize fragmentation-aware routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘fragmentation_aware’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (Any) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find a route from source to destination considering fragmentation.

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k paths ordered by fragmentation level.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths ordered by fragmentation (least fragmented first).

Return type:

list[list[Any]]

update_weights(topology)

Update fragmentation weights based on current spectrum state.

Parameters:

topology (Any) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average fragmentation, and total fragmentation considered.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.NLIAwareRouting

Bases: AbstractRoutingAlgorithm

NLI-aware routing algorithm.

This algorithm finds paths by considering non-linear impairments, selecting the path with the least amount of NLI.

__init__(engine_props, sdn_props)

Initialize NLI-aware routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘nli_aware’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (Any) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find a route from source to destination considering NLI.

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k paths ordered by NLI level.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths ordered by NLI (least NLI first).

Return type:

list[list[Any]]

update_weights(topology)

Update NLI weights based on current network state.

Parameters:

topology (Any) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average NLI, and total NLI considered.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.XTAwareRouting

Bases: AbstractRoutingAlgorithm

Cross-talk aware routing algorithm.

This algorithm finds paths by considering intra-core crosstalk interference, selecting the path with the least amount of cross-talk.

__init__(engine_props, sdn_props)

Initialize XT-aware routing algorithm.

Parameters:

• engine_props (dict[str, Any]) – Dictionary containing engine configuration.

• sdn_props (Any) – Object containing SDN controller properties.

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘xt_aware’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names including NSFNet, USBackbone60, Pan-European, and Generic.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

Parameters:

topology (Any) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find a route from source to destination considering cross-talk.

Results are stored in route_props (paths_matrix, modulation_formats_matrix, weights_list). Consumers should access route_props.paths_matrix for paths.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• request (Any) – Request object containing traffic demand details.

Return type:

None

get_paths(source, destination, k=1)

Get k paths ordered by cross-talk level.

Parameters:

• source (Any) – Source node identifier.

• destination (Any) – Destination node identifier.

• k (int) – Number of paths to return.

Returns:

List of k paths ordered by cross-talk (least XT first).

Return type:

list[list[Any]]

update_weights(topology)

Update cross-talk weights based on current spectrum state.

Parameters:

topology (Any) – NetworkX graph to update weights for.

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics including algorithm name, paths computed, average XT, total XT considered, and XT calculation type.

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Return type:

None

class fusion.modules.routing.OnePlusOneProtection

Bases: AbstractRoutingAlgorithm

Traditional 1+1 disjoint protection routing for optical networks.

Implements standard 1+1 protection as used in optical networking: - Each request gets ONE pair of link-disjoint paths (primary + backup) - Uses max-flow algorithm (Suurballe’s) for optimal disjoint pair finding - Spectrum allocated on BOTH paths simultaneously with same slots - Traffic transmitted on both paths, receiver uses primary signal - Fast switchover on failure (protection switchover latency: default 50ms)

Traditional Flow: 1. Find ONE optimal disjoint pair using max-flow algorithm 2. Validate BOTH paths are feasible (modulation format constraints) 3. Allocate spectrum on BOTH paths (shared protection spectrum) 4. Transmit on both, receiver monitors primary, switches to backup on failure

Key features: - Max-flow based optimal disjoint path computation - Link-disjoint path validation (no shared links) - Modulation format feasibility check for both paths - Simultaneous dual allocation (shared spectrum) - Automatic failure detection and switchover - Optional revert-to-primary after repair

Parameters:

• engine_props (dict[str, Any]) – Engine configuration

• sdn_props (SDNProps) – SDN controller properties

__init__(engine_props, sdn_props)

Initialize 1+1 protection router.

Parameters:

• engine_props (dict[str, Any]) – Engine configuration

• sdn_props (SDNProps) – SDN properties

Return type:

None

property algorithm_name: str

Get the name of the routing algorithm.

Returns:

The algorithm name ‘1plus1_protection’.

Return type:

str

property supported_topologies: list[str]

Get the list of supported topology types.

Returns:

List of supported topology names.

Return type:

list[str]

validate_environment(topology)

Validate that the routing algorithm can work with the given topology.

1+1 protection requires a connected graph with sufficient edge connectivity for disjoint paths.

Parameters:

topology (nx.Graph) – NetworkX graph representing the network topology.

Returns:

True if the algorithm can route in this environment.

Return type:

bool

route(source, destination, request)

Find ONE link-disjoint path pair for traditional 1+1 protection.

Traditional 1+1 protection flow: 1. Use max-flow algorithm to find optimal disjoint pair 2. Validate BOTH paths are feasible (modulation format check) 3. Pass the single pair to SDN controller 4. SDN controller allocates spectrum on BOTH paths simultaneously 5. Traffic transmitted on both paths, receiver uses primary (fast switchover)

Parameters:

• source (Any) – Source node ID

• destination (Any) – Destination node ID

• request (Any) – Request details (optional)

Return type:

None

Example

>>> router = OnePlusOneProtection(props, sdn_props)
>>> router.route(0, 5)
>>> # route_props contains ONE path pair
>>> # SDN controller will allocate on BOTH primary and backup paths

find_all_disjoint_paths(source, destination)

Find all link-disjoint paths using max-flow algorithm.

Uses NetworkX’s edge_disjoint_paths which implements Suurballe’s algorithm to find all edge-disjoint paths between source and destination.

Parameters:

• source (Any) – Source node

• destination (Any) – Destination node

Returns:

List of all disjoint paths found (empty if none exist)

Return type:

list[list[int]]

Example

>>> all_paths = router.find_all_disjoint_paths(0, 5)
>>> # Returns: [[0,1,5], [0,2,5], [0,3,4,5], ...]

find_disjoint_paths(source, destination)

Find link-disjoint primary and backup paths.

Uses NetworkX’s edge_disjoint_paths function for optimal disjoint path finding. Returns the first two paths found.

Parameters:

• source (Any) – Source node

• destination (Any) – Destination node

Returns:

(primary_path, backup_path) or (None, None)

Return type:

tuple[list[int] | None, list[int] | None]

Example

>>> primary, backup = router.find_disjoint_paths(0, 5)
>>> # Verify disjointness
>>> primary_links = set(zip(primary[:-1], primary[1:]))
>>> backup_links = set(zip(backup[:-1], backup[1:]))
>>> assert primary_links.isdisjoint(backup_links)

find_disjoint_paths_k_shortest(source, destination, k=10)

Find disjoint paths using K-shortest paths (alternative method).

This is an alternative to edge_disjoint_paths. It finds the shortest path as primary, then finds the shortest path on a graph with primary links removed.

Note: This method is retained for potential future use or comparison, but find_disjoint_paths() should be preferred as it uses a max-flow algorithm which is more robust.

Parameters:

• source (Any) – Source node

• destination (Any) – Destination node

• k (int) – Max paths to consider

Returns:

(primary, backup) paths

Return type:

tuple[list[int] | None, list[int] | None]

select_best_path_pair(path_pairs)

Select the best path pair from multiple options.

Currently selects based on shortest combined path length (primary + backup). This minimizes total resource usage while maintaining protection.

Parameters:

path_pairs (list[tuple[list[int], list[int]]]) – List of (primary, backup) path tuples

Returns:

Best (primary, backup) path pair

Return type:

tuple[list[int], list[int]]

Example

>>> pairs = [([0,1,2], [0,3,2]), ([0,4,5,2], [0,3,2])]
>>> best = router.select_best_path_pair(pairs)
>>> print(best)
([0, 1, 2], [0, 3, 2])  # Shortest combined length

handle_failure(current_time, affected_requests)

Handle failure by switching protected requests to backup.

Parameters:

• current_time (float) – Current simulation time

• affected_requests (list[dict[str, Any]]) – Requests on failed links

Returns:

Recovery actions performed

Return type:

list[dict[str, Any]]

Example

>>> actions = router.handle_failure(100.0, affected_requests)
>>> for action in actions:
...     print(f"Request {action['request_id']}: "
...           f"switched in {action['recovery_time_ms']}ms")

get_paths(source, destination, k=2)

Get k shortest paths between source and destination.

For 1+1 protection, this returns the primary and backup paths.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• k (int) – Number of paths to return (default 2 for 1+1)

Returns:

List of k paths, where each path is a list of nodes

Return type:

list[list[Any]]

update_weights(topology)

Update edge weights based on current network state.

For 1+1 protection, weights are typically uniform (hop count).

Parameters:

topology (nx.Graph) – NetworkX graph to update weights for

Return type:

None

get_metrics()

Get routing algorithm performance metrics.

Returns:

Dictionary containing algorithm-specific metrics

Return type:

dict[str, Any]

reset()

Reset the routing algorithm state.

Clears metrics and counters.

Return type:

None

fusion.modules.spectrum

FUSION Spectrum Assignment Module.

This package contains various spectrum assignment algorithm implementations for optical network resource allocation, including:

• First Fit spectrum assignment

• Best Fit spectrum assignment (minimizes fragmentation)

• Last Fit spectrum assignment

• Light path slicing management for segmented requests

• Multi-band spectrum assignment support

All algorithms implement the AbstractSpectrumAssigner interface and can be accessed through the SpectrumRegistry for dynamic algorithm selection.

class fusion.modules.spectrum.SpectrumRegistry

Bases: object

Registry for managing spectrum assignment algorithm implementations.

__init__()

Initialize the spectrum assignment registry.

Return type:

None

register(name, algorithm_class)

Register a spectrum assignment algorithm.

Parameters:

• name (str) – Unique name for the algorithm

• algorithm_class (type[AbstractSpectrumAssigner]) – Class that implements AbstractSpectrumAssigner

Raises:

• TypeError – If algorithm_class doesn’t implement AbstractSpectrumAssigner

• ValueError – If name is already registered

Return type:

None

get(name)

Get a spectrum assignment algorithm class by name.

Parameters:

name (str) – Name of the algorithm

Returns:

Algorithm class that implements AbstractSpectrumAssigner

Raises:

KeyError – If algorithm is not found

Return type:

type[AbstractSpectrumAssigner]

create(name, engine_props, sdn_props, route_props)

Create an instance of a spectrum assignment algorithm.

Parameters:

• name (str) – Name of the algorithm

• engine_props (dict) – Engine configuration properties

• sdn_props (SDNProps) – SDN controller properties

• route_props (object) – Routing properties

Returns:

Configured spectrum assignment algorithm instance

Return type:

AbstractSpectrumAssigner

list_algorithms()

List all registered algorithm names.

Returns:

List of algorithm names

Return type:

list[str]

get_algorithm_info(name)

Get information about a specific algorithm.

Parameters:

name (str) – Name of the algorithm

Returns:

Dictionary with algorithm information

Return type:

dict[str, Any]

get_multiband_algorithms()

Get list of algorithms that support multi-band assignment.

Returns:

List of algorithm names that support multi-band

Return type:

list[str]

fusion.modules.spectrum.create_spectrum_algorithm(name, engine_props, sdn_props, route_props)

Create a spectrum algorithm instance from the global registry.

Parameters:

• name (str) –

• engine_props (dict) –

• sdn_props (SDNProps) –

• route_props (object) –

Return type:

AbstractSpectrumAssigner

fusion.modules.spectrum.get_spectrum_algorithm(name)

Get a spectrum algorithm class from the global registry.

Parameters:

name (str) –

Return type:

type[AbstractSpectrumAssigner]

fusion.modules.spectrum.list_spectrum_algorithms()

List all available spectrum assignment algorithms.

Return type:

list[str]

fusion.modules.spectrum.get_spectrum_algorithm_info(name)

Get information about a spectrum assignment algorithm.

Parameters:

name (str) –

Return type:

dict[str, Any]

fusion.modules.spectrum.get_multiband_spectrum_algorithms()

Get algorithms that support multi-band assignment.

Return type:

list[str]

class fusion.modules.spectrum.FirstFitSpectrum

Bases: AbstractSpectrumAssigner

First Fit spectrum assignment algorithm.

This algorithm assigns spectrum by finding the first available contiguous set of slots that can accommodate the request.

__init__(engine_props, sdn_props, route_props)

Initialize First Fit spectrum assignment algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration

• sdn_props (SDNProps) – Object containing SDN controller properties

• route_props (object) – Object containing routing properties

property algorithm_name: str

Return the name of the spectrum assignment algorithm.

property supports_multiband: bool

Indicate whether this algorithm supports multi-band assignment.

assign(path, request)

Assign spectrum resources along the given path for the request.

Parameters:

• path (list[Any]) – List of nodes representing the path

• request (Any) – Request object containing traffic demand and spectrum requirements

Returns:

Dictionary containing spectrum assignment details or None if assignment fails

Raises:

ValueError – If path is empty or request is invalid

Return type:

dict[str, Any] | None

check_spectrum_availability(path, start_slot, end_slot, core_num, band)

Check if spectrum slots are available along the entire path.

Parameters:

• path (list[Any]) – List of nodes representing the path

• start_slot (int) – Starting slot index

• end_slot (int) – Ending slot index

• core_num (int) – Core number to check

• band (str) – Frequency band to check

Returns:

True if all slots are available along the path, False otherwise

Return type:

bool

allocate_spectrum(path, start_slot, end_slot, core_num, band, request_id)

Allocate spectrum resources along the path.

Parameters:

• path (list[Any]) – List of nodes representing the path

• start_slot (int) – Starting slot index to allocate

• end_slot (int) – Ending slot index to allocate

• core_num (int) – Core number to allocate on

• band (str) – Frequency band to allocate in

• request_id (Any) – Unique identifier for the request

Returns:

True if allocation successful, False otherwise

Return type:

bool

deallocate_spectrum(path, start_slot, end_slot, core_num, band)

Deallocate spectrum resources along the path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

Return type:

bool

get_fragmentation_metric(path)

Calculate fragmentation metric for the given path.

Parameters:

path (list[Any]) –

Return type:

float

get_metrics()

Get spectrum assignment algorithm performance metrics.

Return type:

dict[str, Any]

reset()

Reset the spectrum assignment algorithm state.

Return type:

None

class fusion.modules.spectrum.BestFitSpectrum

Bases: AbstractSpectrumAssigner

Best Fit spectrum assignment algorithm.

This algorithm assigns spectrum by finding the smallest available contiguous set of slots that can accommodate the request, minimizing fragmentation.

__init__(engine_props, sdn_props, route_props)

Initialize Best Fit spectrum assignment algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration

• sdn_props (SDNProps) – Object containing SDN controller properties

• route_props (object) – Object containing routing properties

property algorithm_name: str

Return the name of the spectrum assignment algorithm.

property supports_multiband: bool

Indicate whether this algorithm supports multi-band assignment.

assign(path, request)

Assign spectrum resources along the given path for the request.

Parameters:

• path (list[Any]) – List of nodes representing the path

• request (Any) – Request object containing traffic demand and spectrum requirements

Returns:

Dictionary containing spectrum assignment details or None if assignment fails

Return type:

dict[str, Any] | None

check_spectrum_availability(path, start_slot, end_slot, core_num, band)

Check if spectrum slots are available along the entire path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

Return type:

bool

allocate_spectrum(path, start_slot, end_slot, core_num, band, request_id)

Allocate spectrum resources along the path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

• request_id (Any) –

Return type:

bool

deallocate_spectrum(path, start_slot, end_slot, core_num, band)

Deallocate spectrum resources along the path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

Return type:

bool

get_fragmentation_metric(path)

Calculate fragmentation metric for the given path.

Parameters:

path (list[Any]) –

Return type:

float

get_metrics()

Get spectrum assignment algorithm performance metrics.

Return type:

dict[str, Any]

reset()

Reset the spectrum assignment algorithm state.

Return type:

None

class fusion.modules.spectrum.LastFitSpectrum

Bases: AbstractSpectrumAssigner

Last Fit spectrum assignment algorithm.

This algorithm assigns spectrum by finding the last (highest index) available contiguous set of slots that can accommodate the request.

__init__(engine_props, sdn_props, route_props)

Initialize Last Fit spectrum assignment algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration

• sdn_props (SDNProps) – Object containing SDN controller properties

• route_props (object) – Object containing routing properties

property algorithm_name: str

Return the name of the spectrum assignment algorithm.

property supports_multiband: bool

Indicate whether this algorithm supports multi-band assignment.

assign(path, request)

Assign spectrum resources along the given path for the request.

Parameters:

• path (list[Any]) – List of nodes representing the path

• request (Any) – Request object containing traffic demand and spectrum requirements

Returns:

Dictionary with spectrum assignment details or None if fails

Return type:

dict[str, Any] | None

check_spectrum_availability(path, start_slot, end_slot, core_num, band)

Check if spectrum slots are available along the entire path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

Return type:

bool

allocate_spectrum(path, start_slot, end_slot, core_num, band, request_id)

Allocate spectrum resources along the path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

• request_id (Any) –

Return type:

bool

deallocate_spectrum(path, start_slot, end_slot, core_num, band)

Deallocate spectrum resources along the path.

Parameters:

• path (list[Any]) –

• start_slot (int) –

• end_slot (int) –

• core_num (int) –

• band (str) –

Return type:

bool

get_fragmentation_metric(path)

Calculate fragmentation metric for the given path.

Parameters:

path (list[Any]) –

Return type:

float

get_metrics()

Get spectrum assignment algorithm performance metrics.

Return type:

dict[str, Any]

reset()

Reset the spectrum assignment algorithm state.

Return type:

None

class fusion.modules.spectrum.LightPathSlicingManager

Bases: object

Manages light path segment slicing for optical network requests.

This class handles the allocation of network requests using segment slicing strategies including static and dynamic slicing approaches.

Parameters:

• engine_props (Dict[str, Any]) – Engine configuration properties

• sdn_props (Any) – SDN controller properties

• spectrum_obj (Any) – Spectrum assignment object

__init__(engine_props, sdn_props, spectrum_obj)

Parameters:

• engine_props (dict[str, Any]) –

• sdn_props (Any) –

• spectrum_obj (Any) –

Return type:

None

allocate_slicing(num_segments, mod_format, path_list, bandwidth)

Allocate network request using segment slicing.

Parameters:

• num_segments (int) – Number of segments to allocate

• mod_format (str) – Modulation format to use

• path_list (List[Any]) – List of nodes in the routing path

• bandwidth (str) – Bandwidth requirement for each segment

Return type:

Generator[tuple[str, str | None], None, None]

handle_static_slicing(path_list, forced_segments)

Handle static segment slicing for a network request.

Parameters:

• path_list (List[Any]) – List of nodes in the routing path

• forced_segments (int) – Number of segments to force (-1 for auto)

Returns:

True if slicing was successful, False otherwise

Return type:

bool

handle_static_slicing_direct(path_list, forced_segments, sdn_controller)

Handle static slicing using original logic with direct method calls.

Parameters:

• path_list (List[Any]) – List of nodes in the routing path

• forced_segments (int) – Number of segments to force (-1 for auto)

• sdn_controller (Any) – Reference to the SDN controller

Returns:

True if slicing was successful

Return type:

bool

handle_dynamic_slicing_direct(path_list, path_index, forced_segments, sdn_controller)

Handle dynamic slicing using original logic with direct method calls.

Parameters:

• path_list (List[Any]) – List of nodes in the routing path

• path_index (int) – Index of the current path being processed

• forced_segments (int) – Number of forced segments (unused)

• sdn_controller (Any) – Reference to the SDN controller

Returns:

True if slicing was successful

Return type:

bool

allocate_slicing_direct(num_segments, mod_format, path_list, bandwidth, sdn_controller)

Direct implementation of allocate slicing logic.

Parameters:

• num_segments (int) – Number of segments to allocate

• mod_format (str) – Modulation format to use

• path_list (List[Any]) – List of nodes in the routing path

• bandwidth (str) – Bandwidth requirement for each segment

• sdn_controller (Any) – Reference to the SDN controller

Return type:

None

handle_dynamic_slicing(path_list, path_index, forced_segments)

Handle dynamic slicing for a network request.

Attempts to allocate bandwidth using dynamic slicing when traditional allocation fails due to fragmentation.

Parameters:

• path_list (List[Any]) – List of nodes in the routing path

• path_index (int) – Index of the current path being processed

• forced_segments (int) – Number of segments to force (unused)

Return type:

Generator[tuple[str, str | int | None], None, None]

class fusion.modules.spectrum.SpectrumHelpers

Bases: object

Helper utilities for spectrum assignment algorithms.

This class provides utility methods for spectrum assignment operations including spectrum availability checking, band management, and slot allocation.

Parameters:

• engine_props – Dictionary containing engine configuration

• sdn_props – Object containing SDN controller properties

• spectrum_props – Object containing spectrum assignment properties

__init__(engine_props, sdn_props, spectrum_props)

Initialize spectrum helper utilities.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration

• sdn_props (Any) – Object containing SDN controller properties

• spectrum_props (SpectrumProps) – Object containing spectrum assignment properties

Return type:

None

check_other_links()

Check spectrum availability on remaining links in the path.

This method validates that the spectrum slots identified as free on the first link are also available on all subsequent links in the path.

Updates:

spectrum_props.is_free: Set to False if any link has conflicting

spectrum

Return type:

None

check_super_channels(open_slots_matrix, flag)

Find available super-channel for current request allocation.

Parameters:

• open_slots_matrix (list) – Matrix of available super-channel indexes

• flag (str) – Allocation flag for forced index checking

Returns:

True if request can be allocated, False otherwise

Return type:

bool

find_link_inters()

Find slots and channels with potential intersections.

Returns:

Dictionary with free slots, channels and intersections

Return type:

dict

find_best_core()

Finds the core with the least amount of overlapping super channels.

Returns:

The core with the least amount of overlapping channels.

Return type:

int

fusion.modules.snr

FUSION SNR Measurement Module.

This package contains various SNR (Signal-to-Noise Ratio) measurement algorithm implementations for optical network quality assessment, including:

• Standard SNR measurement with linear and nonlinear noise modeling

• Multi-core fiber support with cross-talk calculations

• Amplified spontaneous emission (ASE) noise calculations

• Self-channel interference (SCI) and cross-channel interference (XCI) modeling

All algorithms implement the AbstractSNRMeasurer interface and can be accessed through the SNRRegistry for dynamic algorithm selection.

class fusion.modules.snr.SNRRegistry

Bases: object

Registry for managing SNR measurement algorithm implementations.

__init__()

Initialize the SNR measurement registry.

Return type:

None

register(name, algorithm_class)

Register an SNR measurement algorithm.

Parameters:

• name (str) – Unique name for the algorithm

• algorithm_class (type[AbstractSNRMeasurer]) – Class that implements AbstractSNRMeasurer

Raises:

• TypeError – If algorithm_class doesn’t implement AbstractSNRMeasurer

• ValueError – If name is already registered

Return type:

None

get(name)

Get an SNR measurement algorithm class by name.

Parameters:

name (str) – Name of the algorithm

Returns:

Algorithm class that implements AbstractSNRMeasurer

Raises:

KeyError – If algorithm is not found

Return type:

type[AbstractSNRMeasurer]

create(name, engine_props, sdn_props, spectrum_props, route_props)

Create an instance of an SNR measurement algorithm.

Parameters:

• name (str) – Name of the algorithm

• engine_props (dict) – Engine configuration properties

• sdn_props (object) – SDN controller properties

• spectrum_props (object) – Spectrum assignment properties

• route_props (object) – Routing properties

Returns:

Configured SNR measurement algorithm instance

Return type:

AbstractSNRMeasurer

list_algorithms()

List all registered algorithm names.

Returns:

List of algorithm names

Return type:

list[str]

get_algorithm_info(name)

Get information about a specific algorithm.

Parameters:

name (str) – Name of the algorithm

Returns:

Dictionary with algorithm information

Return type:

dict[str, Any]

get_multicore_algorithms()

Get list of algorithms that support multi-core fiber measurements.

Returns:

List of algorithm names that support multi-core

Return type:

list[str]

fusion.modules.snr.create_snr_algorithm(name, engine_props, sdn_props, spectrum_props, route_props)

Create an SNR algorithm instance from the global registry.

Parameters:

• name (str) –

• engine_props (dict) –

• sdn_props (object) –

• spectrum_props (object) –

• route_props (object) –

Return type:

AbstractSNRMeasurer

fusion.modules.snr.get_snr_algorithm(name)

Get an SNR algorithm class from the global registry.

Parameters:

name (str) –

Return type:

type[AbstractSNRMeasurer]

fusion.modules.snr.list_snr_algorithms()

List all available SNR measurement algorithms.

Return type:

list[str]

fusion.modules.snr.get_snr_algorithm_info(name)

Get information about an SNR measurement algorithm.

Parameters:

name (str) –

Return type:

dict[str, Any]

fusion.modules.snr.get_multicore_snr_algorithms()

Get algorithms that support multi-core fiber measurements.

Return type:

list[str]

class fusion.modules.snr.StandardSNRMeasurer

Bases: AbstractSNRMeasurer

Standard SNR measurement algorithm.

This algorithm calculates signal-to-noise ratio considering linear noise (ASE), nonlinear impairments (SCI, XCI), and optionally cross-talk in multi-core fibers.

__init__(engine_props, sdn_props, spectrum_props, route_props)

Initialize SNR measurement algorithm.

Parameters:

• engine_props (dict) – Dictionary containing engine configuration

• sdn_props (object) – Object containing SDN controller properties

• spectrum_props (object) – Object containing spectrum assignment properties

• route_props (object) – Object containing routing properties

property algorithm_name: str

Return the name of the SNR measurement algorithm.

property supports_multicore: bool

Indicate whether this algorithm supports multi-core fiber measurements.

calculate_snr(path, spectrum_info)

Calculate the SNR for a given path and spectrum assignment.

Parameters:

• path (list[Any]) – List of nodes representing the path

• spectrum_info (dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

SNR value in dB

Raises:

ValueError – If path is empty or spectrum_info is invalid

Return type:

float

calculate_link_snr(source, destination, spectrum_info)

Calculate the SNR for a single link.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• spectrum_info (dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

SNR value in dB for the link

Return type:

float

calculate_nonlinear_noise(path, spectrum_info)

Calculate nonlinear noise components.

Parameters:

• path (list[Any]) – List of nodes representing the path

• spectrum_info (dict[str, Any]) – Dictionary containing spectrum assignment details

Returns:

Dictionary containing nonlinear noise components

Return type:

dict[str, float]

calculate_crosstalk(path, core_num, spectrum_info)

Calculate crosstalk noise for the given path and core.

Parameters:

• path (list[Any]) –

• core_num (int) –

• spectrum_info (dict[str, Any]) –

Return type:

float

get_required_snr_threshold(modulation, reach)

Get the required SNR threshold for a given modulation format and reach.

Parameters:

• modulation (str) – Modulation format (e.g., ‘QPSK’, ‘16QAM’, ‘64QAM’)

• reach (float) – Transmission reach in kilometers

Returns:

Required SNR threshold in dB

Return type:

float

is_snr_acceptable(calculated_snr, required_snr, margin=0.0)

Check if calculated SNR meets the requirement with optional margin.

Parameters:

• calculated_snr (float) –

• required_snr (float) –

• margin (float) –

Return type:

bool

update_link_state(source, destination, spectrum_info)

Update link state based on new spectrum allocation.

Parameters:

• source (Any) – Source node identifier

• destination (Any) – Destination node identifier

• spectrum_info (dict[str, Any]) – Dictionary containing spectrum assignment details

Return type:

None

Note

This implementation doesn’t maintain dynamic state. Subclasses can override for adaptive algorithms.

get_metrics()

Get SNR measurement algorithm performance metrics.

Return type:

dict[str, Any]

reset()

Reset the SNR measurement algorithm state.

Return type:

None

fusion.modules.failures

FUSION Failures Module

Warning

This module is currently in beta status. It has not been fully tested or used in production simulations yet. Use with caution and expect potential API changes.

Provides failure injection and tracking for network survivability testing.

Supports: - F1: Link failures - F2: Node failures - F3: SRLG (Shared Risk Link Group) failures - F4: Geographic failures (hop-radius based)

Example usage:

>>> from fusion.modules.failures import FailureManager
>>> manager = FailureManager(engine_props, topology)
>>> event = manager.inject_failure(
...     'geo',
...     t_fail=100.0,
...     t_repair=200.0,
...     center_node=5,
...     hop_radius=2
... )
>>> is_feasible = manager.is_path_feasible([0, 1, 2])

class fusion.modules.failures.FailureManager

Bases: object

Manages failure injection and tracking for network simulations.

Tracks active failures, maintains failure history, and provides path feasibility checking based on current network state.

Parameters:

• engine_props (dict[str, Any]) – Engine configuration properties

• topology (nx.Graph) – Network topology graph

__init__(engine_props, topology)

Initialize FailureManager.

Parameters:

• engine_props (dict[str, Any]) – Engine configuration

• topology (nx.Graph) – Network topology

Return type:

None

inject_failure(failure_type, t_fail, t_repair, **kwargs)

Inject a failure event into the network.

Parameters:

• failure_type (str) – Type of failure (link, node, srlg, geo)

• t_fail (float) – Failure occurrence time

• t_repair (float) – Repair completion time

• kwargs (Any) – Additional failure-specific parameters

Returns:

Failure event details

Return type:

dict[str, Any]

Raises:

• FailureConfigError – If failure configuration is invalid

• InvalidFailureTypeError – If failure type is unknown

Example

>>> manager = FailureManager(props, topology)
>>> event = manager.inject_failure(
...     'link',
...     t_fail=10.0,
...     t_repair=20.0,
...     link_id=(0, 1)
... )
>>> print(event['failed_links'])
[(0, 1)]

activate_failures(current_time)

Activate all failures scheduled for activation at current_time.

Parameters:

current_time (float) – Current simulation time

Returns:

List of newly activated link tuples

Return type:

list[tuple[Any, Any]]

Example

>>> manager = FailureManager(props, topology)
>>> manager.inject_failure(
...     'link', t_fail=10.0, t_repair=20.0, link_id=(0, 1)
... )
>>> activated = manager.activate_failures(10.0)
>>> print(activated)
[(0, 1)]
>>> print(manager.active_failures)
{(0, 1)}

is_path_feasible(path)

Check if path is feasible given active failures.

A path is infeasible if any of its links are currently failed.

Parameters:

path (list[int]) – List of node IDs forming the path

Returns:

True if path has no failed links, False otherwise

Return type:

bool

Example

>>> manager = FailureManager(props, topology)
>>> manager.active_failures = {(0, 1)}
>>> manager.is_path_feasible([0, 1, 2])
False
>>> manager.is_path_feasible([0, 3, 2])
True

get_affected_links()

Get list of currently failed links.

Returns:

List of failed link tuples

Return type:

list[tuple[Any, Any]]

repair_failures(current_time)

Repair all failures scheduled for repair at current_time.

Parameters:

current_time (float) – Current simulation time

Returns:

List of repaired link tuples

Return type:

list[tuple[Any, Any]]

Example

>>> manager = FailureManager(props, topology)
>>> manager.inject_failure(
...     'link', t_fail=10.0, t_repair=20.0, link_id=(0, 1)
... )
>>> repaired = manager.repair_failures(20.0)
>>> print(repaired)
[(0, 1)]
>>> print(manager.active_failures)
set()

get_failure_count()

Get number of currently active failures.

Returns:

Number of failed links

Return type:

int

clear_all_failures()

Clear all active and scheduled failures (for testing or reset).

This removes all active failures and clears both failure and repair schedules.

Return type:

None

fusion.modules.failures.fail_link(topology, link_id, t_fail, t_repair)

Fail a single link (F1).

Parameters:

• topology (nx.Graph) – Network topology

• link_id (tuple[Any, Any]) – Link tuple (src, dst)

• t_fail (float) – Failure time

• t_repair (float) – Repair time

Returns:

Failure event details

Return type:

dict[str, Any]

Raises:

FailureConfigError – If link does not exist

Example

>>> event = fail_link(G, link_id=(0, 1), t_fail=10.0, t_repair=20.0)
>>> print(event['failed_links'])
[(0, 1)]

fusion.modules.failures.fail_node(topology, node_id, t_fail, t_repair)

Fail a node and all adjacent links (F2).

Parameters:

• topology (nx.Graph) – Network topology

• node_id (Any) – Node ID to fail

• t_fail (float) – Failure time

• t_repair (float) – Repair time

Returns:

Failure event details

Return type:

dict[str, Any]

Raises:

FailureConfigError – If node does not exist

Example

>>> event = fail_node(G, node_id=1, t_fail=10.0, t_repair=20.0)
>>> print(len(event['failed_links']))
3  # Node 1 has 3 adjacent links

fusion.modules.failures.fail_srlg(topology, srlg_links, t_fail, t_repair)

Fail all links in a Shared Risk Link Group (F3).

Parameters:

• topology (nx.Graph) – Network topology

• srlg_links (list[tuple[Any, Any]]) – List of link tuples in SRLG

• t_fail (float) – Failure time

• t_repair (float) – Repair time

Returns:

Failure event details

Return type:

dict[str, Any]

Raises:

FailureConfigError – If SRLG list is empty or contains invalid links

Example

>>> srlg = [(0, 1), (2, 3), (4, 5)]
>>> event = fail_srlg(G, srlg_links=srlg, t_fail=10.0, t_repair=20.0)
>>> print(len(event['failed_links']))
3

fusion.modules.failures.fail_geo(topology, center_node, hop_radius, t_fail, t_repair)

Fail all links within hop_radius of center_node (F4).

Uses NetworkX shortest path to determine hop distance.

Parameters:

• topology (nx.Graph) – Network topology

• center_node (Any) – Center node of disaster

• hop_radius (int) – Hop radius for failure region

• t_fail (float) – Failure time

• t_repair (float) – Repair time

Returns:

Failure event details

Return type:

dict[str, Any]

Raises:

FailureConfigError – If center_node invalid or radius non-positive

Example

>>> event = fail_geo(G, center_node=5, hop_radius=2, t_fail=10.0, t_repair=20.0)
>>> print(event['meta']['affected_nodes'])
[5, 4, 6, 3, 7, 2, 8]

fusion.modules.failures.get_failure_handler(failure_type)

Get failure handler function by type.

Parameters:

failure_type (str) – Failure type (link, node, srlg, geo)

Returns:

Failure handler function

Return type:

Callable

Raises:

InvalidFailureTypeError – If failure type is unknown

Example

>>> handler = get_failure_handler('link')
>>> event = handler(topology, link_id=(0, 1), t_fail=10.0, t_repair=20.0)

fusion.modules.failures.register_failure_type(name, handler)

Register a custom failure type handler.

Allows extending the failure module with custom failure types.

Parameters:

• name (str) – Failure type name

• handler (Callable) – Failure handler function

Return type:

None

exception fusion.modules.failures.FailureError

Bases: Exception

Base exception for failure-related errors.

exception fusion.modules.failures.FailureConfigError

Bases: FailureError

Raised when failure configuration is invalid.

exception fusion.modules.failures.FailureNotFoundError

Bases: FailureError

Raised when a requested failure cannot be found.

exception fusion.modules.failures.InvalidFailureTypeError

Bases: FailureError

Raised when an unknown failure type is requested.

fusion.modules.ml

Machine Learning module for FUSION.

This module provides machine learning utilities for supervised/unsupervised learning in network optimization.

Note

Status: Beta - This module is actively used by the legacy simulation path. Orchestrator integration is planned for a future version.

The module provides utilities for:

• Feature engineering from network states

• Data preprocessing and transformation

• Model persistence (save/load)

• Model evaluation metrics

• Visualization

Current Integration:

• Legacy path (use_orchestrator=False): Uses get_ml_obs() and load_model()

• Orchestrator path: Not yet integrated (planned for v6.x)

Enable with deploy_model=True in engine_props.

Relationship with MLControlPolicy:

This module (fusion/modules/ml/) and fusion/policies/ml_policy.py are currently separate implementations:

• This module: utilities for legacy path (works with engine_props, sdn_props)

• MLControlPolicy: path selection for orchestrator (works with Request, NetworkState)

Both have their own feature extraction and model loading. The plan for v6.x is to unify these so MLControlPolicy uses this module’s utilities, eliminating duplication.

Note: This module provides ML utilities, not ML algorithms. You bring your own models (sklearn, tensorflow, pytorch, etc.) and use these utilities around them.

fusion.modules.ml.plot_data_distributions(simulation_dict, input_dataframe, erlang)

Plot data distributions for machine learning simulation runs.

Creates pie charts and histograms for key features in the dataset.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• input_dataframe (pd.DataFrame) – DataFrame containing simulation data

• erlang (float) – Traffic volume value

Returns:

None

Return type:

None

Example

>>> sim_dict = {'train_file_path': 'experiment_001'}
>>> data = pd.DataFrame({'bandwidth': [50, 100, 200]})
>>> plot_data_distributions(sim_dict, data, 1000.0)

fusion.modules.ml.plot_feature_importance(simulation_dict, model, feature_names, erlang, test_features, test_labels)

Plot feature importance for a trained model.

Supports tree-based models, linear models, and uses permutation importance for models without built-in importance scores.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• model (Any) – Trained machine learning model

• feature_names (List[str]) – List of feature names

• erlang (float) – Traffic volume value

• test_features (np.ndarray) – Test feature matrix

• test_labels (np.ndarray) – Test labels

Returns:

None

Return type:

None

Example

>>> features = ['path_length', 'bandwidth', 'congestion']
>>> plot_feature_importance(sim_dict, model, features, 1000.0, X_test, y_test)

fusion.modules.ml.plot_confusion_matrix(simulation_dict, test_labels, predictions, erlang, algorithm)

Plot confusion matrix and calculate classification metrics.

Creates a heatmap visualization of the confusion matrix and returns accuracy, precision, recall, and F1 scores.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• test_labels (np.ndarray) – True labels

• predictions (np.ndarray) – Model predictions

• erlang (str) – Traffic volume value as string

• algorithm (str) – Name of the algorithm used

Returns:

Dictionary of classification metrics

Return type:

Dict[str, float]

Example

>>> metrics = plot_confusion_matrix(
...     sim_dict, y_test, y_pred, "1000", "RandomForest"
... )
>>> print(f"Accuracy: {metrics['accuracy']:.2%}")

fusion.modules.ml.plot_2d_clusters(pca_dataframe, output_path=None)

Plot 2D visualization of clusters using PCA-reduced data.

Parameters:

• pca_dataframe (pd.DataFrame) – DataFrame with PCA components and predicted labels

• output_path (str) – Optional path to save the plot

Returns:

None

Return type:

None

Example

>>> df_pca = pd.DataFrame({
...     'PC1': [...], 'PC2': [...], 'predicted_label': [...]
... })
>>> plot_2d_clusters(df_pca, 'output/clusters_2d.png')

fusion.modules.ml.plot_3d_clusters(pca_dataframe, output_path=None)

Plot 3D visualization of clusters using PCA-reduced data.

Parameters:

• pca_dataframe (pd.DataFrame) – DataFrame with PCA components and predicted labels

• output_path (str) – Optional path to save the plot

Returns:

None

Return type:

None

Example

>>> df_pca = pd.DataFrame({
...     'PC1': [...], 'PC2': [...], 'PC3': [...], 'predicted_label': [...]
... })
>>> plot_3d_clusters(df_pca, 'output/clusters_3d.png')

fusion.modules.ml.process_training_data(simulation_dict, input_dataframe, erlang)

Process raw data for machine learning model training.

Performs one-hot encoding and type conversions as needed.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• input_dataframe (pd.DataFrame) – Raw input DataFrame

• erlang (float) – Traffic volume value

Returns:

Processed DataFrame ready for training

Return type:

pd.DataFrame

Example

>>> sim_dict = {'train_file_path': 'experiment_001'}
>>> raw_data = pd.DataFrame({'bandwidth': [50, 100], 'path_length': [10, 20]})
>>> processed = process_training_data(sim_dict, raw_data, 1000.0)

fusion.modules.ml.balance_training_data(input_dataframe, balance_per_slice, erlang, simulation_dict)

Balance training data to ensure representative sampling.

Can balance equally across all classes or use weighted sampling.

Parameters:

• input_dataframe (pd.DataFrame) – Input DataFrame to balance

• balance_per_slice (bool) – If True, balance equally across segments

• erlang (float) – Traffic volume value

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

Returns:

Balanced DataFrame

Return type:

pd.DataFrame

Example

>>> data = pd.DataFrame({'num_segments': [1, 1, 2, 2, 4, 4, 8, 8]})
>>> balanced = balance_training_data(data, True, 1000.0, sim_dict)

fusion.modules.ml.prepare_prediction_features(raw_features, engine_properties, sdn_properties)

Prepare features for model prediction from raw request data.

Transforms raw features into the format expected by trained models.

Parameters:

• raw_features (Dict[str, Any]) – Dictionary of raw feature values

• engine_properties (Dict[str, Any]) – Engine configuration properties

• sdn_properties (object) – SDN controller properties object

Returns:

DataFrame with properly formatted features

Return type:

pd.DataFrame

Example

>>> features = {'bandwidth': 100, 'path_length': 15, 'congestion': 0.3}
>>> prepared = prepare_prediction_features(features, engine_props, sdn_props)

fusion.modules.ml.normalize_features(features, normalization_type='standard')

Normalize features using specified method.

Parameters:

• features (pd.DataFrame) – Feature DataFrame to normalize

• normalization_type (str) – Type of normalization (‘standard’ or ‘minmax’)

Returns:

Tuple of (normalized features, scaler object)

Return type:

Tuple[pd.DataFrame, Any]

Raises:

ValueError – If normalization type not supported

Example

>>> X = pd.DataFrame({'feature1': [1, 2, 3], 'feature2': [10, 20, 30]})
>>> X_norm, scaler = normalize_features(X, 'standard')

fusion.modules.ml.split_features_labels(dataframe, target_column)

Split DataFrame into features and labels.

Parameters:

• dataframe (pd.DataFrame) – Complete DataFrame with features and target

• target_column (str) – Name of the target column

Returns:

Tuple of (features, labels)

Return type:

Tuple[pd.DataFrame, pd.Series]

Raises:

KeyError – If target column not found

Example

>>> data = pd.DataFrame({'feature1': [1, 2], 'target': [0, 1]})
>>> X, y = split_features_labels(data, 'target')

fusion.modules.ml.extract_ml_features(request_dict, engine_properties, sdn_properties)

Extract machine learning features from a network request.

Creates features including path length, congestion metrics, and bandwidth requirements formatted for ML model consumption.

Parameters:

• request_dict (Dict[str, Any]) – Dictionary containing request information

• engine_properties (Dict[str, Any]) – Engine configuration properties

• sdn_properties (object) – SDN controller properties object

Returns:

DataFrame with extracted features

Return type:

pd.DataFrame

Example

>>> request = {'bandwidth': 100, 'mod_formats': {'QPSK': {'max_length': 2000}}}
>>> features = extract_ml_features(request, engine_props, sdn_props)
>>> print(features.columns.tolist())
['path_length', 'longest_reach', 'ave_cong', 'old_bandwidth_50', ...]

fusion.modules.ml.create_interaction_features(features, interactions=None)

Create interaction features between existing features.

Parameters:

• features (pd.DataFrame) – Original features DataFrame

• interactions (List[Tuple[str, str]]) – List of tuples specifying interactions

Returns:

DataFrame with interaction features added

Return type:

pd.DataFrame

Example

>>> features = pd.DataFrame({'A': [1, 2], 'B': [3, 4]})
>>> interactions = [('A', 'B')]
>>> new_features = create_interaction_features(features, interactions)
>>> print('A_x_B' in new_features.columns)
True

fusion.modules.ml.create_polynomial_features(features, degree=2, include_bias=False)

Create polynomial features up to specified degree.

Parameters:

• features (pd.DataFrame) – Original features DataFrame

• degree (int) – Maximum polynomial degree

• include_bias (bool) – Whether to include bias term

Returns:

DataFrame with polynomial features

Return type:

pd.DataFrame

Example

>>> features = pd.DataFrame({'x': [1, 2, 3]})
>>> poly_features = create_polynomial_features(features, degree=2)

fusion.modules.ml.engineer_network_features(request_dict, network_state)

Engineer advanced network-specific features.

Creates features that capture network topology and state characteristics relevant for routing and spectrum assignment decisions.

Parameters:

• request_dict (Dict[str, Any]) – Request information

• network_state (Dict[str, Any]) – Current network state information

Returns:

Dictionary of engineered features

Return type:

Dict[str, float]

Example

>>> request = {'source': 'A', 'destination': 'B', 'bandwidth': 100}
>>> state = {'link_utilization': {...}, 'active_paths': [...]}
>>> features = engineer_network_features(request, state)

fusion.modules.ml.save_model(simulation_dict, model, algorithm, erlang, metadata=None)

Save a trained machine learning model with metadata.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• model (Any) – Trained model object

• algorithm (str) – Name of the algorithm

• erlang (str) – Traffic volume value as string

• metadata (Dict[str, Any]) – Optional metadata to save with model

Returns:

Path where model was saved

Return type:

str

Example

>>> from sklearn.ensemble import RandomForestClassifier
>>> model = RandomForestClassifier()
>>> # ... train model ...
>>> path = save_model(sim_dict, model, "random_forest", "1000")
>>> print(f"Model saved to: {path}")

fusion.modules.ml.load_model(engine_properties)

Load a trained machine learning model.

Parameters:

engine_properties (Dict[str, Any]) – Properties from engine including model path info

Returns:

Loaded model object

Return type:

Any

Raises:

FileNotFoundError – If model file doesn’t exist

Example

>>> engine_props = {
...     'ml_model': 'random_forest',
...     'train_file_path': 'experiment_001',
...     'erlang': 1000.0
... }
>>> model = load_model(engine_props)

fusion.modules.ml.load_model_with_metadata(engine_properties)

Load a model along with its metadata.

Parameters:

engine_properties (Dict[str, Any]) – Properties from engine

Returns:

Tuple of (model, metadata)

Return type:

Tuple[Any, Dict[str, Any]]

Example

>>> model, metadata = load_model_with_metadata(engine_props)
>>> print(f"Model trained on: {metadata.get('saved_at')}")

fusion.modules.ml.save_model_ensemble(models, simulation_dict, ensemble_name, erlang)

Save an ensemble of models.

Parameters:

• models (List[Any]) – List of model objects

• simulation_dict (Dict[str, Any]) – Simulation parameters

• ensemble_name (str) – Name for the ensemble

• erlang (str) – Traffic volume

Returns:

Path where ensemble was saved

Return type:

str

Example

>>> models = [model1, model2, model3]
>>> path = save_model_ensemble(models, sim_dict, "voting_ensemble", "1000")

fusion.modules.ml.export_model_for_deployment(model, export_path, model_format='onnx', input_sample=None)

Export model to deployment-friendly format.

Parameters:

• model (Any) – Trained model

• export_path (str) – Path for exported model

• model_format (str) – Export format (‘onnx’, ‘pmml’, ‘pickle’)

• input_sample (np.ndarray) – Sample input for ONNX conversion

Returns:

Path to exported model

Return type:

str

Raises:

ValueError – If format not supported

Example

>>> sample = np.array([[1, 2, 3, 4]])
>>> export_path = export_model_for_deployment(
...     model, "model.onnx", "onnx", sample
... )

fusion.modules.ml.check_model_compatibility(model_path, expected_features)

Check if saved model is compatible with current feature set.

Parameters:

• model_path (str) – Path to model file

• expected_features (List[str]) – List of expected feature names

Returns:

Compatibility report

Return type:

Dict[str, Any]

Example

>>> features = ['path_length', 'bandwidth', 'congestion']
>>> report = check_model_compatibility("model.joblib", features)
>>> print(f"Compatible: {report['compatible']}")

fusion.modules.ml.evaluate_classifier(true_labels, predictions, class_names=None)

Comprehensive evaluation of classification model performance.

Parameters:

• true_labels (np.ndarray) – Ground truth labels

• predictions (np.ndarray) – Model predictions

• class_names (List[str]) – Optional names for classes

Returns:

Dictionary containing various metrics

Return type:

Dict[str, Any]

Example

>>> y_true = np.array([0, 1, 0, 1, 1])
>>> y_pred = np.array([0, 1, 1, 1, 0])
>>> metrics = evaluate_classifier(y_true, y_pred)
>>> print(f"Accuracy: {metrics['accuracy']:.2%}")

fusion.modules.ml.evaluate_regressor(true_values, predictions)

Comprehensive evaluation of regression model performance.

Parameters:

• true_values (np.ndarray) – Ground truth values

• predictions (np.ndarray) – Model predictions

Returns:

Dictionary containing regression metrics

Return type:

Dict[str, float]

Example

>>> y_true = np.array([1.0, 2.0, 3.0, 4.0])
>>> y_pred = np.array([1.1, 2.2, 2.9, 3.8])
>>> metrics = evaluate_regressor(y_true, y_pred)
>>> print(f"MSE: {metrics['mse']:.4f}")

fusion.modules.ml.cross_validate_model(model, features, labels, cv_folds=5, scoring_metrics=None)

Perform cross-validation and return detailed results.

Parameters:

• model (Any) – Model to evaluate

• features (pd.DataFrame) – Feature matrix

• labels (pd.Series) – Target labels

• cv_folds (int) – Number of cross-validation folds

• scoring_metrics (List[str]) – List of scoring metrics to use

Returns:

Cross-validation results

Return type:

Dict[str, Any]

Example

>>> from sklearn.ensemble import RandomForestClassifier
>>> model = RandomForestClassifier()
>>> results = cross_validate_model(model, X, y, cv_folds=5)
>>> print(f"Mean accuracy: {results['accuracy']['mean']:.4f}")

fusion.modules.ml.evaluate_model_stability(model, features, labels, n_iterations=10, test_size=0.3)

Evaluate model stability across multiple train/test splits.

Parameters:

• model (Any) – Model to evaluate

• features (pd.DataFrame) – Feature matrix

• labels (pd.Series) – Target labels

• n_iterations (int) – Number of random splits to test

• test_size (float) – Proportion of data for testing

Returns:

Stability analysis results

Return type:

Dict[str, Any]

Example

>>> stability = evaluate_model_stability(model, X, y, n_iterations=20)
>>> print(f"Accuracy variance: {stability['accuracy']['variance']:.6f}")

fusion.modules.ml.compare_models(models, features, labels, test_size=0.3, random_state=42)

Compare multiple models on the same dataset.

Parameters:

• models (Dict[str, Any]) – Dictionary of model_name: model pairs

• features (pd.DataFrame) – Feature matrix

• labels (pd.Series) – Target labels

• test_size (float) – Test set proportion

• random_state (int) – Random seed

Returns:

DataFrame with comparison results

Return type:

pd.DataFrame

Example

>>> models = {
...     'RF': RandomForestClassifier(),
...     'SVM': SVC(),
...     'LR': LogisticRegression()
... }
>>> comparison = compare_models(models, X, y)
>>> print(comparison.sort_values('f1_score', ascending=False))

fusion.modules.ml.analyze_prediction_errors(true_labels, predictions, features=None)

Analyze prediction errors to identify patterns.

Parameters:

• true_labels (np.ndarray) – Ground truth labels

• predictions (np.ndarray) – Model predictions

• features (pd.DataFrame) – Optional features for error analysis

Returns:

Error analysis results

Return type:

Dict[str, Any]

Example

>>> analysis = analyze_prediction_errors(y_true, y_pred, X_test)
>>> print(f"Most confused pair: {analysis['most_confused_pair']}")

fusion.modules.ml.plot_data(simulation_dict, input_dataframe, erlang)

Plot data distributions for machine learning simulation runs.

Creates pie charts and histograms for key features in the dataset.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• input_dataframe (pd.DataFrame) – DataFrame containing simulation data

• erlang (float) – Traffic volume value

Returns:

None

Return type:

None

Example

>>> sim_dict = {'train_file_path': 'experiment_001'}
>>> data = pd.DataFrame({'bandwidth': [50, 100, 200]})
>>> plot_data_distributions(sim_dict, data, 1000.0)

fusion.modules.ml.plot_confusion(simulation_dict, test_labels, predictions, erlang, algorithm)

Plot confusion matrix and calculate classification metrics.

Creates a heatmap visualization of the confusion matrix and returns accuracy, precision, recall, and F1 scores.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• test_labels (np.ndarray) – True labels

• predictions (np.ndarray) – Model predictions

• erlang (str) – Traffic volume value as string

• algorithm (str) – Name of the algorithm used

Returns:

Dictionary of classification metrics

Return type:

Dict[str, float]

Example

>>> metrics = plot_confusion_matrix(
...     sim_dict, y_test, y_pred, "1000", "RandomForest"
... )
>>> print(f"Accuracy: {metrics['accuracy']:.2%}")

fusion.modules.ml.get_ml_obs(request_dict, engine_properties, sdn_properties)

Extract machine learning features from a network request.

Creates features including path length, congestion metrics, and bandwidth requirements formatted for ML model consumption.

Parameters:

• request_dict (Dict[str, Any]) – Dictionary containing request information

• engine_properties (Dict[str, Any]) – Engine configuration properties

• sdn_properties (object) – SDN controller properties object

Returns:

DataFrame with extracted features

Return type:

pd.DataFrame

Example

>>> request = {'bandwidth': 100, 'mod_formats': {'QPSK': {'max_length': 2000}}}
>>> features = extract_ml_features(request, engine_props, sdn_props)
>>> print(features.columns.tolist())
['path_length', 'longest_reach', 'ave_cong', 'old_bandwidth_50', ...]

fusion.modules.ml.process_data(simulation_dict, input_dataframe, erlang)

Process raw data for machine learning model training.

Performs one-hot encoding and type conversions as needed.

Parameters:

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

• input_dataframe (pd.DataFrame) – Raw input DataFrame

• erlang (float) – Traffic volume value

Returns:

Processed DataFrame ready for training

Return type:

pd.DataFrame

Example

>>> sim_dict = {'train_file_path': 'experiment_001'}
>>> raw_data = pd.DataFrame({'bandwidth': [50, 100], 'path_length': [10, 20]})
>>> processed = process_training_data(sim_dict, raw_data, 1000.0)

fusion.modules.ml.even_process_data(input_dataframe, balance_per_slice, erlang, simulation_dict)

Balance training data to ensure representative sampling.

Can balance equally across all classes or use weighted sampling.

Parameters:

• input_dataframe (pd.DataFrame) – Input DataFrame to balance

• balance_per_slice (bool) – If True, balance equally across segments

• erlang (float) – Traffic volume value

• simulation_dict (Dict[str, Any]) – Dictionary containing simulation parameters

Returns:

Balanced DataFrame

Return type:

pd.DataFrame

Example

>>> data = pd.DataFrame({'num_segments': [1, 1, 2, 2, 4, 4, 8, 8]})
>>> balanced = balance_training_data(data, True, 1000.0, sim_dict)

---

Analysis

fusion.analysis

Statistics and results analysis.

Analysis modules for FUSION network simulations.

This package provides utilities for analyzing network topology, performance metrics, and simulation results.

class fusion.analysis.NetworkAnalyzer

Bases: object

Analyzes network topology and usage patterns.

This class provides utilities for examining network characteristics, link utilization, and traffic patterns.

__init__()

Initialize the network analyzer.

Return type:

None

static get_link_usage_summary(network_spectrum)

Generate a summary of link usage across the network.

Records usage for each directional link separately to maintain bidirectional link statistics.

Parameters:

network_spectrum (dict) – Network spectrum database

Returns:

Dictionary mapping link identifiers to usage statistics

Return type:

dict[str, dict[str, Any]]

static analyze_network_congestion(network_spectrum, specific_paths=None)

Analyze network congestion levels.

Parameters:

• network_spectrum (dict) – Network spectrum database

• specific_paths (list | None) – Optional specific paths to analyze

Returns:

Congestion analysis results

Return type:

dict[str, Any]

static get_network_utilization_stats(network_spectrum)

Calculate network-wide utilization statistics.

Parameters:

network_spectrum (dict) – Network spectrum database

Returns:

Dictionary of utilization statistics

Return type:

dict[str, float]

static identify_bottleneck_links(network_spectrum, threshold=0.8)

Identify links that are above a utilization threshold.

Parameters:

• network_spectrum (dict) – Network spectrum database

• threshold (float) – Utilization threshold (0.0 to 1.0)

Returns:

List of bottleneck link identifiers

Return type:

list

---

Reporting

fusion.reporting

Results export and dataset logging.

Reporting module for FUSION simulation output.

This module provides utilities for formatting and reporting simulation results, separating presentation concerns from data collection.

class fusion.reporting.SimulationReporter

Bases: object

Handle reporting and output formatting for simulation statistics.

This class is responsible for presenting simulation results in various formats, including console output, log files, and structured reports.

__init__(logger=None, verbose=True)

Initialize the simulation reporter.

Parameters:

• logger (logging.Logger | None) – Logger instance to use (creates one if not provided)

• verbose (bool) – Whether to output detailed information

report_iteration_stats(iteration, max_iterations, erlang, blocking_list, print_flag=True)

Report statistics for a completed iteration.

This method replaces the print_iter_stats method from SimStats, providing proper logging and formatting.

Parameters:

• iteration (int) – Current iteration number (0-based)

• max_iterations (int) – Total number of iterations

• erlang (float) – Erlang value for this simulation

• blocking_list (list[float]) – List of blocking probabilities from all iterations

• print_flag (bool) – Whether to output the statistics

Return type:

None

report_simulation_start(simulation_info_dict)

Report the start of a simulation run.

Parameters:

simulation_info_dict (dict[str, Any]) – Dictionary containing simulation parameters

Return type:

None

report_simulation_complete(erlang, iterations_completed, confidence_interval=None)

Report the completion of a simulation run.

Parameters:

• erlang (float) – Erlang value that was simulated

• iterations_completed (int) – Number of iterations completed

• confidence_interval (float | None) – Optional confidence interval achieved

Return type:

None

report_blocking_statistics(blocked_requests, total_requests, bit_rate_blocked, bit_rate_total, blocking_reasons_dict)

Report detailed blocking statistics.

Parameters:

• blocked_requests (int) – Number of blocked requests

• total_requests (int) – Total number of requests

• bit_rate_blocked (float) – Blocked bit rate

• bit_rate_total (float) – Total bit rate requested

• blocking_reasons_dict (dict[str, float]) – Dictionary of blocking reasons with percentages

Return type:

None

report_save_location(save_path)

Report where simulation results are being saved.

Parameters:

save_path (str) – Path where results are saved

Return type:

None

report_error(error_message, exception=None)

Report an error during simulation.

Parameters:

• error_message (str) – Error message to log

• exception (Exception | None) – Optional exception object

Return type:

None

report_warning(warning_message)

Report a warning during simulation.

Parameters:

warning_message (str) – Warning message to log

Return type:

None

create_summary_report(statistics_dict)

Create a formatted summary report from simulation statistics.

Parameters:

statistics_dict (dict[str, Any]) – Dictionary containing all simulation statistics

Returns:

Formatted report as a string

Return type:

str

class fusion.reporting.DatasetLogger

Bases: object

Log offline RL training data to JSON Lines format.

Logs tuples (s, a, r, s’, action_mask, meta) for offline training. Each line is a valid JSON object representing one transition.

Parameters:

• output_path (str) – Path to output JSONL file

• engine_props (dict[str, Any]) – Engine configuration

Example

>>> logger = DatasetLogger('datasets/offline_data.jsonl', engine_props)
>>> logger.log_transition(state, action, reward, next_state, mask, meta)
>>> logger.close()

__init__(output_path, engine_props)

Initialize dataset logger.

Parameters:

• output_path (str) – Output file path

• engine_props (dict[str, Any]) – Engine configuration

Return type:

None

log_transition(state, action, reward, next_state, action_mask, meta)

Log a single transition.

Parameters:

• state (dict[str, Any]) – Current state dict

• action (int) – Selected action (path index)

• reward (float) – Reward (1.0 for accept, -1.0 for block)

• next_state (dict[str, Any] | None) – Next state dict (or None)

• action_mask (list[bool]) – Feasibility mask

• meta (dict[str, Any]) – Metadata (decision_time_ms, bp_window_tag, etc.)

Return type:

None

Example

>>> logger.log_transition(
...     state={'src': 0, 'dst': 5, ...},
...     action=2,
...     reward=1.0,
...     next_state=None,
...     action_mask=[False, True, True, False],
...     meta={'decision_time_ms': 0.35, 'bp_window_tag': 'pre'}
... )

close()

Close the output file.

Example

>>> logger.close()

Return type:

None

fusion.reporting.aggregate_seed_results(results, metric_keys=None)

Aggregate results across multiple seeds.

Computes mean, std, and 95% confidence intervals for specified metrics.

Parameters:

• results (list[dict[str, Any]]) – List of result dicts (one per seed)

• metric_keys (list[str] | None) – Metrics to aggregate (if None, aggregates all numeric metrics)

Returns:

Aggregated stats with mean, std, ci95

Return type:

dict[str, dict[str, float]]

Example

>>> results = [
...     {'bp_overall': 0.05, 'recovery_time_mean_ms': 52.3, 'seed': 42},
...     {'bp_overall': 0.06, 'recovery_time_mean_ms': 51.8, 'seed': 43}
... ]
>>> metrics = ['bp_overall', 'recovery_time_mean_ms']
>>> agg = aggregate_seed_results(results, metrics)
>>> print(agg['bp_overall'])
{'mean': 0.055, 'std': 0.007, 'ci95_lower': 0.041, 'n': 2}

fusion.reporting.create_comparison_table(baseline_results, rl_results, metrics)

Create comparison table for baseline vs RL.

Parameters:

• baseline_results (list[dict[str, Any]]) – Baseline results (one per seed)

• rl_results (list[dict[str, Any]]) – RL policy results (one per seed)

• metrics (list[str]) – Metrics to compare

Returns:

Comparison dictionary

Return type:

dict[str, dict[str, Any]]

Example

>>> baseline = [{'bp_overall': 0.10, 'seed': 42}]
>>> rl = [{'bp_overall': 0.08, 'seed': 42}]
>>> comp = create_comparison_table(baseline, rl, ['bp_overall'])
>>> print(comp['bp_overall']['improvement_pct'])
25.0

fusion.reporting.format_comparison_for_display(comparison)

Format comparison table for console display.

Parameters:

comparison (dict[str, dict[str, Any]]) – Comparison dict from create_comparison_table

Returns:

Formatted table string

Return type:

str

Example

>>> comp = create_comparison_table(baseline, rl, ['bp_overall'])
>>> print(format_comparison_for_display(comp))
Metric                | Baseline           | RL                 | Improvement
-------------------------------------------------------------------------------
bp_overall            | 0.1050 ± 0.0071    | 0.0850 ± 0.0071    | +19.05%

fusion.reporting.export_results_to_csv(results, output_path)

Export results to CSV file.

Parameters:

• results (list[dict[str, Any]]) – List of result dicts (each representing one simulation run)

• output_path (str) – Output CSV path

Return type:

None

Example

>>> results = [stats.to_csv_row() for stats in all_stats]
>>> export_results_to_csv(results, 'results/survivability.csv')

fusion.reporting.export_aggregated_results(aggregated, output_path, metadata=None)

Export aggregated results (mean, std, CI95) to CSV.

Parameters:

• aggregated (dict[str, dict[str, float]]) – Aggregated results from aggregate_seed_results

• output_path (str) – Output CSV path

• metadata (dict[str, Any] | None) – Optional metadata to include (topology, policy, etc.)

Return type:

None

Example

>>> agg = aggregate_seed_results(results, ['bp_overall'])
>>> export_aggregated_results(agg, 'results/summary.csv', {'policy': 'bc'})

fusion.reporting.export_comparison_table(comparison, output_path)

Export comparison table (baseline vs RL) to CSV.

Parameters:

• comparison (dict[str, dict[str, Any]]) – Comparison dict from create_comparison_table

• output_path (str) – Output CSV path

Return type:

None

Example

>>> comp = create_comparison_table(baseline, rl, ['bp_overall'])
>>> export_comparison_table(comp, 'results/comparison.csv')

fusion.reporting.append_result_to_csv(result, output_path)

Append a single result to CSV file.

Useful for incremental logging during multi-seed experiments.

Parameters:

• result (dict[str, Any]) – Single result dict

• output_path (str) – Output CSV path

Return type:

None

Example

>>> for seed in range(10):
...     stats = run_simulation(seed)
...     append_result_to_csv(stats.to_csv_row(), 'results/live.csv')

class fusion.reporting.GroomingStatistics

Bases: object

Statistics specific to traffic grooming operations.

__init__()

Initialize grooming statistics.

Return type:

None

update_grooming_outcome(was_groomed, was_partially_groomed, bandwidth, new_lightpaths)

Update statistics for a request allocation attempt.

Parameters:

• was_groomed (bool) – Request was fully groomed

• was_partially_groomed (bool) – Request was partially groomed

• bandwidth (float) – Request bandwidth

• new_lightpaths (int) – Number of new lightpaths created

Return type:

None

update_lightpath_release(_lightpath_id, utilization, _lifetime)

Update statistics when a lightpath is released.

Parameters:

• _lightpath_id (int) – ID of released lightpath

• utilization (float) – Average utilization percentage

• _lifetime (float) – Lightpath lifetime in seconds

Return type:

None

calculate_grooming_rate()

Calculate the overall grooming success rate.

Returns:

Percentage of requests that were groomed (fully or partially)

Return type:

float

calculate_bandwidth_savings()

Calculate bandwidth savings from grooming.

Returns:

Percentage of bandwidth that was groomed vs new lightpaths

Return type:

float

get_average_lightpath_utilization()

Get average utilization across all released lightpaths.

Returns:

Average utilization percentage

Return type:

float

to_dict()

Convert statistics to dictionary for serialization.

Returns:

Dictionary of all statistics

Return type:

dict[str, Any]

class fusion.reporting.SimulationStatistics

Bases: object

Comprehensive statistics tracking for network simulations.

__init__(engine_props)

Initialize statistics collector.

Parameters:

engine_props (dict[str, Any]) – Engine configuration properties

Return type:

None

fusion.reporting.generate_grooming_report(stats)

Generate human-readable grooming statistics report.

Parameters:

stats (GroomingStatistics) – Grooming statistics object

Returns:

Formatted report string

Return type:

str

fusion.reporting.export_grooming_stats_csv(stats, filepath)

Export grooming statistics to CSV file.

Parameters:

• stats (GroomingStatistics) – Grooming statistics object

• filepath (str) – Output CSV file path

Return type:

None

---

I/O Operations

fusion.io

File input/output and topology loading.

Data I/O module for FUSION simulator.

This module provides functionality for: - Data generation (generate.py) - Data structuring (structure.py) - Data export (exporter.py)

class fusion.io.SimulationDataExporter

Bases: object

Main data export interface for simulation data.

__init__()

Return type:

None

export_topology(topology_data, output_path, output_format='json')

Export network topology data.

Parameters:

• topology_data (dict) –

• output_path (str | Path) –

• output_format (str) –

Return type:

None

export_results(results_data, output_path, output_format='json')

Export simulation results data.

Parameters:

• results_data (dict) –

• output_path (str | Path) –

• output_format (str) –

Return type:

None

export_metrics(metrics_data, output_path, output_format='csv')

Export simulation metrics data.

Parameters:

• metrics_data (dict | list[dict]) –

• output_path (str | Path) –

• output_format (str) –

Return type:

None

class fusion.io.ExporterRegistry

Bases: object

Registry for data exporters.

__init__()

Return type:

None

register_exporter(format_name, exporter)

Register a new exporter.

Parameters:

• format_name (str) –

• exporter (BaseExporter) –

Return type:

None

get_exporter(format_name)

Get an exporter by format name.

Parameters:

format_name (str) –

Return type:

BaseExporter

property supported_formats: list[str]

Get list of supported export formats.

fusion.io.create_pt(cores_per_link, network_spectrum_dict)

Generate information relevant to the physical topology of the network.

Parameters:

• cores_per_link (int) – The number of cores in each fiber’s link

• network_spectrum_dict (Dict[tuple, float]) – The network spectrum database mapping node pairs to lengths

Returns:

Physical layer information topology of the network

Return type:

Dict[str, Dict]

fusion.io.create_bw_info(mod_assumption, mod_assumptions_path=None)

Determine reach and slots needed for each bandwidth and modulation format.

Parameters:

• mod_assumption (str) – Controls which assumptions to be used

• mod_assumptions_path (Optional[str]) – Path to modulation assumptions file

Returns:

The number of spectral slots needed for each bandwidth and modulation format pair

Return type:

Dict[str, Dict]

Raises:

• FileNotFoundError – If modulation assumptions file is not found

• NotImplementedError – If unknown modulation assumption is provided

fusion.io.create_network(net_name, base_fp=None, const_weight=False, is_only_core_node=False)

Build a physical network structure from topology files.

Resolves all paths safely using project-root-relative logic.

Parameters:

• net_name (str) – Name of the network topology to load

• base_fp (str) – Base file path for topology files

• const_weight (bool) – If True, assign constant weight to all links

• is_only_core_node (bool) – If True, only load core nodes

Returns:

Tuple of (network dictionary, core nodes list)

Return type:

Tuple[Dict[Tuple[str, str], float], List[str]]

Raises:

NotImplementedError – If unknown network name is provided

fusion.io.assign_link_lengths(network_fp, node_pairs_dict, constant_weight=False)

Assign length to each link in a given topology.

Parameters:

• network_fp (Path) – Path to the network topology file

• node_pairs_dict (Dict[str, str]) – Dictionary mapping node names to alternative names

• constant_weight (bool) – If True, assign constant weight of 1.0 to all links

Returns:

Dictionary mapping node pairs to link lengths

Return type:

Dict[Tuple[str, str], float]

fusion.io.assign_core_nodes(core_nodes_fp)

Determine which nodes are core nodes in the network.

Parameters:

core_nodes_fp (Path) – Path to the core nodes file

Returns:

List of core node identifiers

Return type:

List[str]

---

Utilities

fusion.utils

Helper functions and utilities.

Utility modules for FUSION.

This package provides common utilities used across the FUSION codebase. Import directly from specific modules to avoid circular dependencies.

Example

from fusion.utils.logging_config import get_logger from fusion.utils.network import find_path_length

fusion.utils.setup_logger(name, level='INFO', log_file=None, log_dir=None, console=True, file_mode='a', max_bytes=10485760, backup_count=5, format_string=None)

Set up a standardized logger for FUSION modules.

Creates a logger with optional file and console handlers. File handlers use rotation to prevent unbounded growth.

Parameters:

• name (str) – Logger name (typically __name__ of the calling module)

• level (str) – Logging level as string (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• log_file (str | None) – Optional log file name (created in log_dir)

• log_dir (str | None) – Directory for log files (defaults to logs/ in project root)

• console (bool) – Whether to output to console

• file_mode (str) – File open mode (‘a’ for append, ‘w’ for overwrite)

• max_bytes (int) – Maximum size of log file before rotation

• backup_count (int) – Number of backup files to keep

• format_string (str | None) – Custom format string (uses DEFAULT_FORMAT if None)

Returns:

Configured logger instance

Return type:

Logger

fusion.utils.get_logger(name, level='INFO')

Get an existing logger or create a basic one.

This is a convenience function for modules that need a logger but don’t require special configuration.

Parameters:

• name (str) – Logger name (typically __name__)

• level (str) – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

Returns:

Logger instance

Return type:

Logger