Data Structures

This document covers the key data structures used in the core module: Domain objects (shared simulation concepts), Result objects (orchestrator outputs), and Properties classes (legacy state containers).

Overview

FUSION uses three families of data structures:

Domain Objects (Shared): Core simulation concepts in fusion/domain/. Used by both legacy and orchestrator architectures: NetworkState, Request, Lightpath, SimulationConfig.
Result Objects (v6.0+ Orchestrator): Immutable frozen dataclasses in fusion/domain/results.py. Capture pipeline outputs with validation and factory methods.
Properties Classes (Legacy): Mutable classes in fusion/core/properties.py. Carry state between legacy components.

+------------------------+     +------------------------+
|    RESULT OBJECTS      |     |   PROPERTIES CLASSES   |
|    (Orchestrator)      |     |   (Legacy)             |
+------------------------+     +------------------------+
| - RouteResult          |     | - RoutingProps         |
| - SpectrumResult       |     | - SpectrumProps        |
| - GroomingResult       |     | - SDNProps             |
| - SlicingResult        |     | - SNRProps             |
| - SNRResult            |     | - StatsProps           |
| - AllocationResult     |     | - GroomingProps        |
| - ProtectionResult     |     |                        |
+------------------------+     +------------------------+

+------------------------+
|    DOMAIN OBJECTS      |
|    (Shared)            |
+------------------------+
| - NetworkState         |
| - LinkSpectrum         |
| - Request              |
| - Lightpath            |
| - SimulationConfig     |
+------------------------+

Domain Objects

Domain objects in fusion/domain/ are the shared data structures used by both the legacy and orchestrator architectures. They represent the core simulation concepts.

NetworkState

The single source of truth for network state during simulation.

Location:: fusion/domain/network_state.py
Key Classes:: NetworkState, LinkSpectrum

The NetworkState class provides immutable access to the current network state, including topology, spectrum allocation, and active lightpaths.

NetworkState fields:

@dataclass
class NetworkState:
    topology: nx.Graph                          # Network graph
    link_spectrum: dict[tuple[str, str], LinkSpectrum]  # Per-link spectrum
    lightpaths: dict[int, Lightpath] = field(default_factory=dict)
    active_failures: set[tuple[str, str]] = field(default_factory=set)

Key methods:

Method	Description
`get_link_spectrum(src, dst)`	Get LinkSpectrum for a link
`is_range_free(path, core, start, end)`	Check if spectrum range is free on all path links
`create_lightpath(lightpath)`	Create new lightpath (returns new NetworkState)
`release_lightpath(lightpath_id)`	Release lightpath (returns new NetworkState)
`get_lightpaths_with_capacity(src, dst, bandwidth)`	Find lightpaths that can groom this request
`get_links_on_path(path)`	Get all links for a path
`is_link_failed(src, dst)`	Check if a link is currently failed

Usage:

from fusion.domain import NetworkState

# Create from simulation state
network_state = NetworkState(
    topology=topology,
    link_spectrum=link_spectrum_dict,
    lightpaths=active_lightpaths,
)

# Check spectrum availability
if network_state.is_range_free(path, core=0, start=100, end=108):
    # Allocate...

# Find grooming opportunities
candidates = network_state.get_lightpaths_with_capacity(
    source="0",
    destination="5",
    bandwidth_gbps=100,
)

LinkSpectrum

Per-link spectrum management with slot allocation tracking.

Location:: fusion/domain/network_state.py

@dataclass
class LinkSpectrum:
    num_cores: int
    slots_per_core: int
    spectrum: np.ndarray  # Shape: (num_cores, slots_per_core)

Key methods:

Method	Description
`is_range_free(core, start, end)`	Check if slot range is free on a core
`allocate_range(core, start, end, request_id)`	Mark slots as allocated (returns new LinkSpectrum)
`release_range(core, start, end)`	Mark slots as free (returns new LinkSpectrum)
`get_utilization(core)`	Get utilization ratio for a core
`find_first_free_block(core, num_slots)`	Find first free contiguous block

Usage:

# Check if slots 100-108 are free on core 0
if link_spectrum.is_range_free(core=0, start=100, end=108):
    new_spectrum = link_spectrum.allocate_range(
        core=0, start=100, end=108, request_id=42
    )

Request

Represents a network connection request with lifecycle tracking.

Location:: fusion/domain/request.py
Key Classes:: Request, RequestType, RequestStatus, BlockReason, ProtectionStatus

Request dataclass:

@dataclass
class Request:
    request_id: int
    source: str
    destination: str
    bandwidth_gbps: int
    arrive_time: float
    depart_time: float

    # Lifecycle tracking
    request_type: RequestType = RequestType.ARRIVAL
    status: RequestStatus = RequestStatus.PENDING

    # Allocation results (set after processing)
    allocated_path: tuple[str, ...] | None = None
    allocated_core: int | None = None
    allocated_start_slot: int | None = None
    allocated_end_slot: int | None = None
    allocated_modulation: str | None = None
    lightpath_id: int | None = None

    # Protection (1+1)
    protection_status: ProtectionStatus = ProtectionStatus.UNPROTECTED
    backup_path: tuple[str, ...] | None = None
    backup_core: int | None = None
    backup_start_slot: int | None = None
    backup_end_slot: int | None = None

    # Blocking info
    block_reason: BlockReason | None = None

RequestType enum:

class RequestType(Enum):
    ARRIVAL = "arrival"    # New connection request
    RELEASE = "release"    # Connection teardown

RequestStatus enum (state machine):

class RequestStatus(Enum):
    PENDING = "pending"          # Not yet processed
    ROUTED = "routed"            # Path found
    ALLOCATED = "allocated"      # Spectrum assigned
    ACTIVE = "active"            # Currently using resources
    BLOCKED = "blocked"          # Could not be served
    RELEASED = "released"        # Resources freed
    FAILED = "failed"            # Failed during active state

BlockReason enum:

class BlockReason(Enum):
    CONGESTION = "congestion"       # No spectrum available
    DISTANCE = "distance"           # Path too long for modulation
    XT_THRESHOLD = "xt_threshold"   # Cross-talk too high
    SNR_THRESHOLD = "snr_threshold" # SNR too low
    NO_PATH = "no_path"             # No route exists
    FAILURE = "failure"             # Network failure blocked request

ProtectionStatus enum:

class ProtectionStatus(Enum):
    UNPROTECTED = "unprotected"     # No backup path
    PROTECTED = "protected"          # Backup path established
    SWITCHED = "switched"            # Using backup path
    FAILED = "failed"                # Both paths failed

Key methods:

# Duration calculation
request.holding_time  # depart_time - arrive_time

# State transitions
request.mark_routed(path)
request.mark_allocated(core, start, end, modulation)
request.mark_blocked(reason)
request.mark_released()

Lightpath

Represents an established optical connection with capacity management.

Location:: fusion/domain/lightpath.py

@dataclass
class Lightpath:
    lightpath_id: int
    path: tuple[str, ...]
    core: int
    start_slot: int
    end_slot: int               # Exclusive
    modulation: str
    capacity_gbps: int          # Total capacity
    allocated_gbps: int = 0     # Currently used

    # Protection
    is_protected: bool = False
    backup_path: tuple[str, ...] | None = None
    backup_core: int | None = None
    backup_start_slot: int | None = None
    backup_end_slot: int | None = None

    # Request tracking
    request_ids: set[int] = field(default_factory=set)

Key methods:

Method	Description
`available_capacity`	Property: `capacity_gbps - allocated_gbps`
`num_slots`	Property: `end_slot - start_slot`
`can_accommodate(bandwidth)`	Check if bandwidth fits in available capacity
`allocate_bandwidth(bandwidth, request_id)`	Add bandwidth allocation (returns new Lightpath)
`release_bandwidth(bandwidth, request_id)`	Remove bandwidth allocation (returns new Lightpath)
`uses_link(src, dst)`	Check if lightpath traverses a link

Usage:

from fusion.domain import Lightpath

# Create lightpath
lp = Lightpath(
    lightpath_id=1,
    path=("0", "2", "5"),
    core=0,
    start_slot=100,
    end_slot=108,
    modulation="QPSK",
    capacity_gbps=100,
)

# Check grooming possibility
if lp.can_accommodate(50):
    lp = lp.allocate_bandwidth(50, request_id=42)

# Check available capacity
print(f"Available: {lp.available_capacity} Gbps")

SimulationConfig

Immutable configuration replacing mutable engine_props dict.

Location:: fusion/domain/config.py

The SimulationConfig dataclass provides type-safe, validated configuration for simulations. It replaces the legacy engine_props dictionary approach.

@dataclass(frozen=True)
class SimulationConfig:
    # Simulation control
    max_iters: int = 10
    num_requests: int = 1000
    erlang: float = 300.0
    holding_time: float = 3600.0
    seed: int | None = None

    # Network configuration
    network: str = "NSFNet"
    cores_per_link: int = 7
    c_band: int = 320
    l_band: int = 0
    s_band: int = 0

    # Algorithm selection
    route_method: str = "k_shortest_path"
    allocation_method: str = "first_fit"
    k_paths: int = 3

    # Feature flags
    use_orchestrator: bool = False
    enable_grooming: bool = True
    enable_slicing: bool = False
    enable_protection: bool = False

    # Survivability
    failure_enabled: bool = False
    failure_type: str | None = None

    # Output control
    output_train_data: bool = False
    save_snapshots: bool = False
    print_step: int = 0

Key methods:

# Create from legacy dict
config = SimulationConfig.from_engine_props(engine_props)

# Convert back to dict (for legacy components)
engine_props = config.to_engine_props()

# Access with type safety
if config.enable_protection:
    # Protection is enabled
    ...

Advantages over engine_props dict:

Aspect	Benefit
Type safety	IDE autocompletion, mypy validation
Immutability	Cannot be accidentally modified
Defaults	Clear default values in one place
Validation	`__post_init__` checks constraints
Documentation	Field types and docstrings

Result Objects

Result objects are immutable (frozen=True) dataclasses that capture the output of each pipeline stage. They provide:

Validation via __post_init__
Factory methods (blocked(), success(), etc.)
Conversion from legacy properties (from_routing_props())

AllocationResult

The SINGLE SOURCE OF TRUTH for request allocation.

Location:: fusion/domain/results.py
Immutability:: Frozen dataclass

The success field is the final authority on whether a request was served.

Fields:

@dataclass(frozen=True)
class AllocationResult:
    # Final authority
    success: bool

    # Lightpath tracking
    lightpaths_created: tuple[int, ...] = ()
    lightpaths_groomed: tuple[int, ...] = ()
    total_bandwidth_allocated_gbps: int = 0

    # Feature flags
    is_groomed: bool = False
    is_partially_groomed: bool = False
    is_sliced: bool = False
    is_protected: bool = False

    # Failure info
    block_reason: BlockReason | None = None

    # Per-segment tracking (for sliced allocations)
    bandwidth_allocations: tuple[int, ...] = ()
    modulations: tuple[str, ...] = ()
    cores: tuple[int, ...] = ()
    bands: tuple[str, ...] = ()
    start_slots: tuple[int, ...] = ()
    end_slots: tuple[int, ...] = ()
    snr_values: tuple[float, ...] = ()

    # Nested results (for debugging)
    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

Invariants:

If success=True: at least one lightpath (created or groomed)
If success=True: total_bandwidth_allocated_gbps > 0
If success=False: block_reason is set

Factory methods:

# Blocked request
result = AllocationResult.blocked(BlockReason.CONGESTION)

# New lightpath
result = AllocationResult.success_new_lightpath(
    lightpath_id=42,
    bandwidth_gbps=100,
)

# Groomed request
result = AllocationResult.success_groomed(
    lightpath_ids=[1, 2],
    bandwidth_gbps=100,
)

# Sliced request
result = AllocationResult.success_sliced(
    lightpath_ids=[1, 2, 3],
    bandwidth_gbps=300,
)

RouteResult

Output of the routing pipeline - candidate paths with modulation options.

@dataclass(frozen=True)
class RouteResult:
    # Primary paths
    paths: tuple[tuple[str, ...], ...] = ()
    weights_km: tuple[float, ...] = ()
    modulations: tuple[tuple[str, ...], ...] = ()

    # Backup paths (for 1+1 protection)
    backup_paths: tuple[tuple[str, ...], ...] | None = None
    backup_weights_km: tuple[float, ...] | None = None
    backup_modulations: tuple[tuple[str, ...], ...] | None = None

    # Metadata
    strategy_name: str = ""
    connection_index: int | None = None  # For external SNR lookup

Key properties:

result.is_empty          # True if no paths found
result.num_paths         # Number of candidate paths
result.has_protection    # True if backup paths available
result.best_path         # First (best) path or None
result.best_weight       # Weight of best path

Usage:

result = RouteResult(
    paths=(("0", "2", "5"),),
    weights_km=(100.0,),
    modulations=(("QPSK", "16-QAM"),),
    strategy_name="k_shortest_path",
)

for i in range(result.num_paths):
    path = result.get_path(i)
    mods = result.get_modulations_for_path(i)

SpectrumResult

Output of spectrum assignment - allocated slot range.

@dataclass(frozen=True)
class SpectrumResult:
    is_free: bool                 # Whether allocation succeeded

    # Allocation details (valid only if is_free=True)
    start_slot: int = 0
    end_slot: int = 0             # Exclusive
    core: int = 0
    band: str = "c"
    modulation: str = ""
    slots_needed: int = 0

    # Dynamic LP slicing
    achieved_bandwidth_gbps: int | None = None  # May be less than requested

    # Backup spectrum (for 1+1 protection)
    backup_start_slot: int | None = None
    backup_end_slot: int | None = None
    backup_core: int | None = None
    backup_band: str | None = None

Key properties:

result.num_slots    # Slots allocated (0 if not free)
result.has_backup   # True if backup spectrum allocated

Factory methods:

# No spectrum found
result = SpectrumResult.not_found(slots_needed=8)

# Successful allocation
result = SpectrumResult(
    is_free=True,
    start_slot=100,
    end_slot=108,
    core=0,
    band="c",
    modulation="QPSK",
    slots_needed=8,
)

GroomingResult

Output of grooming pipeline - using existing lightpath capacity.

@dataclass(frozen=True)
class GroomingResult:
    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

Invariants:

fully_groomed and partially_groomed are mutually exclusive
If fully_groomed: remaining_bandwidth_gbps == 0
If groomed: len(lightpaths_used) > 0

Factory methods:

# No grooming possible
result = GroomingResult.no_grooming(bandwidth_gbps=100)

# Fully groomed
result = GroomingResult.full(
    bandwidth_gbps=100,
    lightpath_ids=[1, 2],
)

# Partially groomed
result = GroomingResult.partial(
    bandwidth_groomed=50,
    remaining=50,
    lightpath_ids=[1],
    forced_path=["0", "2", "5"],
)

SlicingResult

Output of slicing pipeline - request split across lightpaths.

@dataclass(frozen=True)
class SlicingResult:
    success: bool = False
    num_slices: int = 0
    slice_bandwidth_gbps: int = 0
    lightpaths_created: tuple[int, ...] = ()
    total_bandwidth_gbps: int = 0

Key property:

result.is_sliced    # True if num_slices > 1

Factory methods:

# Slicing failed
result = SlicingResult.failed()

# Single lightpath (no slicing needed)
result = SlicingResult.single_lightpath(
    bandwidth_gbps=100,
    lightpath_id=1,
)

# Sliced into multiple lightpaths
result = SlicingResult.sliced(
    num_slices=4,
    slice_bandwidth=25,
    lightpath_ids=[1, 2, 3, 4],
)

SNRResult

Output of SNR validation - signal quality check.

@dataclass(frozen=True)
class SNRResult:
    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] = field(default_factory=dict)

Key properties:

result.is_degraded        # Passed but margin < 1 dB
result.has_link_breakdown # Per-link values available

Factory methods:

# SNR passed
result = SNRResult.success(snr_db=18.5, required_snr_db=15.0)

# SNR failed
result = SNRResult.failure(
    snr_db=12.0,
    required_snr_db=15.0,
    reason="SNR below threshold",
)

# SNR validation skipped
result = SNRResult.skipped()

ProtectionResult

Output of protection pipeline - 1+1 path establishment.

@dataclass(frozen=True)
class ProtectionResult:
    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_type: str | None = None

Key properties:

result.is_fully_protected  # Both paths established
result.recovery_duration_ms  # Time to recover (if completed)

Properties Classes (Legacy)

Properties classes are mutable state containers used by legacy components.

RoutingProps

State for path computation.

class RoutingProps:
    def __init__(self):
        # Path results
        self.paths_matrix: list[Any] = []
        self.modulation_formats_matrix: list[list[str]] = []
        self.weights_list: list[float] = []
        self.path_index_list: list[int] = []

        # Backup paths (1+1 protection)
        self.backup_paths_matrix: list[list[int] | None] = []
        self.backup_modulation_formats_matrix: list[list[str]] = []

        # Physical parameters
        self.input_power: float = DEFAULT_INPUT_POWER
        self.frequency_spacing: float = DEFAULT_FREQUENCY_SPACING
        self.span_length: float = DEFAULT_SPAN_LENGTH

SpectrumProps

State for spectrum assignment.

class SpectrumProps:
    def __init__(self):
        # Path and requirements
        self.path_list: list[int] | None = None
        self.backup_path: list[int] | None = None
        self.slots_needed: int | None = None
        self.modulation: str | None = None

        # Core assignment
        self.forced_core: int | None = None
        self.core_number: int | None = None

        # Allocation results
        self.is_free: bool = False
        self.start_slot: int | None = None
        self.end_slot: int | None = None
        self.current_band: str | None = None

        # Grooming
        self.lightpath_id: int | None = None
        self.lightpath_bandwidth: float | None = None

SDNProps

State for SDN controller operations.

class SDNProps:
    def __init__(self):
        # Network state
        self.topology: Any | None = None
        self.network_spectrum_dict: dict | None = None

        # Current request
        self.request_id: int | None = None
        self.source: str | None = None
        self.destination: str | None = None
        self.bandwidth: float | None = None

        # Allocation results
        self.was_routed: bool | None = None
        self.block_reason: str | None = None

        # Per-segment tracking (slicing)
        self.modulation_list: list[str] = []
        self.core_list: list[int] = []
        self.start_slot_list: list[int] = []
        self.end_slot_list: list[int] = []
        # ... many more lists

        # Protection (1+1)
        self.primary_path: list[int] | None = None
        self.backup_path: list[int] | None = None
        self.is_protected: bool = False

SNRProps - Multi-Fiber Mappings

State for SNR calculations, including file mappings for pre-calculated data.

class SNRProps:
    def __init__(self):
        # Physical constants
        self.light_frequency: float = LIGHT_FREQUENCY_CENTER
        self.planck_constant: float = PLANCK_CONSTANT

        # SNR thresholds per modulation
        self.req_snr: dict[str, float] = {
            "BPSK": 3.71,
            "QPSK": 6.72,
            "8-QAM": 10.84,
            "16-QAM": 13.24,
            "32-QAM": 16.16,
            "64-QAM": 19.01,
        }

        # Pre-calculated SNR file mappings
        self.file_mapping_dict = {...}  # See below

Multi-Fiber File Mappings:

The file_mapping_dict maps topology and core configurations to pre-calculated SNR data files:

self.file_mapping_dict = {
    "USbackbone60": {
        # Multi-fiber (single core per fiber)
        "multi_fiber": {
            "mf": "MF-USB6014-MF.npy",
            "gsnr": "GSNR-USB6014-MF.npy",
        },
        # Multi-core fiber: (center_core, total_cores)
        (2, 4): {  # 4-core fiber, center core 2
            "mf": "MF-USB6014-MCF4-C2.npy",
            "gsnr": "GSNR-USB6014-MCF4-C2.npy",
        },
        (6, 7): {  # 7-core fiber, center core 6
            "mf": "MF-USB6014-MCF7-C6.npy",
            "gsnr": "GSNR-USB6014-MCF7-C6.npy",
        },
        (3, 7): {  # 7-core fiber, edge core 3
            "mf": "MF-USB6014-MCF7-C3.npy",
            "gsnr": "GSNR-USB6014-MCF7-C3.npy",
        },
        # ... 13-core, 19-core configurations
    },
    "Spainbackbone30": {
        # Similar structure for Spain topology
    },
}

Understanding the mappings:

"multi_fiber": Standard single-core fibers (one fiber per core)
(center_core, total_cores): Multi-core fiber configuration
"mf": Modulation format selection file
"gsnr": Generalized SNR values file

Example usage:

# Get files for 7-core MCF on USbackbone60
topology = "USbackbone60"
center_core = 6
total_cores = 7

files = snr_props.file_mapping_dict[topology][(center_core, total_cores)]
mf_file = files["mf"]    # "MF-USB6014-MCF7-C6.npy"
gsnr_file = files["gsnr"]  # "GSNR-USB6014-MCF7-C6.npy"

Converting Between Legacy and Result Objects

Adapters convert between legacy properties and result objects:

# Legacy to Result
route_result = RouteResult.from_routing_props(routing_props)
spectrum_result = SpectrumResult.from_spectrum_props(spectrum_props)

# Result to Legacy (for backward compatibility)
allocation_dict = spectrum_result.to_allocation_dict()

Design Principles

Result objects are immutable - Use frozen=True for all dataclasses - Prevents accidental mutation - Enables safe caching and sharing
Validation on creation - __post_init__ validates invariants - Fail fast if data is inconsistent
Factory methods for common cases - blocked(), success(), etc. - Clearer code than constructor calls
Conversion methods for migration - from_*_props() for legacy compatibility - to_*_dict() for backward compatibility

Data Structures

Overview

Domain Objects

NetworkState

LinkSpectrum

Request

Lightpath

SimulationConfig

Result Objects

AllocationResult

RouteResult

SpectrumResult

GroomingResult

SlicingResult

SNRResult

ProtectionResult

Properties Classes (Legacy)

RoutingProps

SpectrumProps

SDNProps

SNRProps - Multi-Fiber Mappings

Converting Between Legacy and Result Objects

Design Principles

See Also