"""
Spectrum assignment algorithms registry for FUSION.
This module provides a centralized registry for all spectrum assignment algorithm
implementations that follow the AbstractSpectrumAssigner interface.
"""
from typing import Any, cast
from fusion.core.properties import SDNProps
from fusion.interfaces.spectrum import AbstractSpectrumAssigner
from .best_fit import BestFitSpectrum
# Import all spectrum assignment algorithm implementations
from .first_fit import FirstFitSpectrum
from .last_fit import LastFitSpectrum
[docs]
class SpectrumRegistry:
"""Registry for managing spectrum assignment algorithm implementations."""
[docs]
def __init__(self) -> None:
"""Initialize the spectrum assignment registry."""
self._algorithms: dict[str, type[AbstractSpectrumAssigner]] = {}
self._register_default_algorithms()
def _register_default_algorithms(self) -> None:
"""Register all default spectrum assignment algorithms."""
algorithms = [FirstFitSpectrum, BestFitSpectrum, LastFitSpectrum]
for algorithm_class in algorithms:
# Create temporary instance to get algorithm name
temp_sdn_props = SDNProps() # Create temporary instance
temp_instance = algorithm_class({}, temp_sdn_props, None) # type: ignore[abstract]
self.register(
temp_instance.algorithm_name,
cast(type[AbstractSpectrumAssigner], algorithm_class),
)
[docs]
def register(self, name: str, algorithm_class: type[AbstractSpectrumAssigner]) -> None:
"""
Register a spectrum assignment algorithm.
:param name: Unique name for the algorithm
:param algorithm_class: Class that implements AbstractSpectrumAssigner
:raises TypeError: If algorithm_class doesn't implement AbstractSpectrumAssigner
:raises ValueError: If name is already registered
"""
if not issubclass(algorithm_class, AbstractSpectrumAssigner):
raise TypeError(f"{algorithm_class.__name__} must implement AbstractSpectrumAssigner")
if name in self._algorithms:
raise ValueError(f"Spectrum algorithm '{name}' is already registered")
self._algorithms[name] = algorithm_class
[docs]
def get(self, name: str) -> type[AbstractSpectrumAssigner]:
"""
Get a spectrum assignment algorithm class by name.
:param name: Name of the algorithm
:return: Algorithm class that implements AbstractSpectrumAssigner
:raises KeyError: If algorithm is not found
"""
if name not in self._algorithms:
raise KeyError(f"Spectrum assignment algorithm '{name}' not found. Available algorithms: {list(self._algorithms.keys())}")
return self._algorithms[name]
[docs]
def create(self, name: str, engine_props: dict, sdn_props: SDNProps, route_props: object) -> AbstractSpectrumAssigner:
"""
Create an instance of a spectrum assignment algorithm.
:param name: Name of the algorithm
:param engine_props: Engine configuration properties
:param sdn_props: SDN controller properties
:param route_props: Routing properties
:return: Configured spectrum assignment algorithm instance
"""
algorithm_class = self.get(name)
return algorithm_class(engine_props, sdn_props, route_props)
[docs]
def list_algorithms(self) -> list[str]:
"""
List all registered algorithm names.
:return: List of algorithm names
"""
return list(self._algorithms.keys())
[docs]
def get_algorithm_info(self, name: str) -> dict[str, Any]:
"""
Get information about a specific algorithm.
:param name: Name of the algorithm
:return: Dictionary with algorithm information
"""
algorithm_class = self.get(name)
# Create temporary instance to get properties
temp_instance = algorithm_class({}, None, None) # type: ignore[abstract, arg-type]
return {
"name": name,
"class": algorithm_class.__name__,
"module": algorithm_class.__module__,
"supports_multiband": temp_instance.supports_multiband,
"description": algorithm_class.__doc__.strip() if algorithm_class.__doc__ else "No description",
}
[docs]
def get_multiband_algorithms(self) -> list[str]:
"""
Get list of algorithms that support multi-band assignment.
:return: List of algorithm names that support multi-band
"""
multiband_algos = []
for name, algorithm_class in self._algorithms.items():
temp_instance = algorithm_class({}, None, None) # type: ignore[abstract, arg-type]
if temp_instance.supports_multiband:
multiband_algos.append(name)
return multiband_algos
# Global registry instance
_registry = SpectrumRegistry()
# Convenience functions for global registry access
def register_spectrum_algorithm(name: str, algorithm_class: type[AbstractSpectrumAssigner]) -> None:
"""Register a spectrum algorithm in the global registry."""
_registry.register(name, algorithm_class)
[docs]
def get_spectrum_algorithm(name: str) -> type[AbstractSpectrumAssigner]:
"""Get a spectrum algorithm class from the global registry."""
return _registry.get(name)
[docs]
def create_spectrum_algorithm(name: str, engine_props: dict, sdn_props: SDNProps, route_props: object) -> AbstractSpectrumAssigner:
"""Create a spectrum algorithm instance from the global registry."""
return _registry.create(name, engine_props, sdn_props, route_props)
[docs]
def list_spectrum_algorithms() -> list[str]:
"""List all available spectrum assignment algorithms."""
return _registry.list_algorithms()
[docs]
def get_spectrum_algorithm_info(name: str) -> dict[str, Any]:
"""Get information about a spectrum assignment algorithm."""
return _registry.get_algorithm_info(name)
[docs]
def get_multiband_spectrum_algorithms() -> list[str]:
"""Get algorithms that support multi-band assignment."""
return _registry.get_multiband_algorithms()
# Dictionary for backward compatibility
SPECTRUM_ALGORITHMS = {
"first_fit": FirstFitSpectrum,
"best_fit": BestFitSpectrum,
"last_fit": LastFitSpectrum,
}