Enable differentiated design band per OMS

Introduce a design_band parameter in ROADM and Transceiver.
- if nothing is defined, use SI band(s)
- if design band is defined in ROADM, use this one for all degrees
- if per degree design band is defined, use this one instead

unsupported case: single band OMS with default multiband design band.
Check that these definitions are consistent with actual amplifiers

Signed-off-by: EstherLerouzic <[email protected]>
Change-Id: Ibea4ce6e72d2b1e96ef8cf4efaf499530d24179c

gnpy/core/elements.py

@@ -20,8 +20,9 @@
 instance as a result.
 """
 
+from copy import deepcopy
 from numpy import abs, array, errstate, ones, interp, mean, pi, polyfit, polyval, sum, sqrt, log10, exp, asarray, full,\
-    squeeze, zeros, append, flip, outer, ndarray
+    squeeze, zeros, outer, ndarray
 from scipy.constants import h, c
 from scipy.interpolate import interp1d
 from collections import namedtuple
@@ -32,7 +33,7 @@
 from gnpy.core.utils import lin2db, db2lin, arrange_frequencies, snr_sum, per_label_average, pretty_summary_print, \
     watt2dbm, psd2powerdbm, calculate_absolute_min_or_zero, nice_column_str
 from gnpy.core.parameters import RoadmParams, FusedParams, FiberParams, PumpParams, EdfaParams, EdfaOperational, \
-    MultiBandParams, RoadmPath, RoadmImpairment, find_band_name, FrequencyBand
+    MultiBandParams, RoadmPath, RoadmImpairment, TransceiverParams, find_band_name, FrequencyBand
 from gnpy.core.science_utils import NliSolver, RamanSolver
 from gnpy.core.info import SpectralInformation, muxed_spectral_information, demuxed_spectral_information
 from gnpy.core.exceptions import NetworkTopologyError, SpectrumError, ParametersError
@@ -81,8 +82,20 @@ def latitude(self):
 
 
 class Transceiver(_Node):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, *args, params=None, **kwargs):
+        if not params:
+            params = {}
+        try:
+            with warnings.catch_warnings(record=True) as caught_warnings:
+                super().__init__(*args, params=TransceiverParams(**params), **kwargs)
+                if caught_warnings:
+                    msg = f'In Transceiver {kwargs["uid"]}: {caught_warnings[0].message}'
+                    _logger.warning(msg)
+        except ParametersError as e:
+            msg = f'Config error in {kwargs["uid"]}: {e}'
+            _logger.critical(msg)
+            raise ParametersError(msg) from e
+
         self.osnr_ase_01nm = None
         self.osnr_ase = None
         self.osnr_nli = None
@@ -97,6 +110,8 @@ def __init__(self, *args, **kwargs):
         self.total_penalty = 0
         self.propagated_labels = [""]
         self.tx_power = None
+        self.design_bands = self.params.design_bands
+        self.per_degree_design_bands = self.params.per_degree_design_bands
 
     def _calc_cd(self, spectral_info):
         """Updates the Transceiver property with the CD of the received channels. CD in ps/nm.
@@ -283,6 +298,8 @@ def __init__(self, *args, params=None, **kwargs):
                                                                                 "to_degree": i["to_degree"],
                                                                                 "impairment_id": i["impairment_id"]}
                                        for i in self.params.per_degree_impairments}
+        self.design_bands = deepcopy(self.params.design_bands)
+        self.per_degree_design_bands = deepcopy(self.params.per_degree_design_bands)
 
     @property
     def to_json(self):
@@ -316,6 +333,11 @@ def to_json(self):
         if self.per_degree_impairments:
             to_json['per_degree_impairments'] = list(self.per_degree_impairments.values())
 
+        if self.params.design_bands is not None:
+            if len(self.params.design_bands) > 1:
+                to_json['params']['design_bands'] = self.params.design_bands
+        if self.params.per_degree_design_bands:
+            to_json['params']['per_degree_design_bands'] = self.params.per_degree_design_bands
         return to_json
 
     def __repr__(self):

gnpy/core/network.py

@@ -11,11 +11,13 @@
 from operator import attrgetter
 from collections import namedtuple
 from logging import getLogger
-from typing import Tuple, List, Optional
+from typing import Tuple, List, Optional, Union
+from networkx import DiGraph
 
 from gnpy.core import elements
 from gnpy.core.exceptions import ConfigurationError, NetworkTopologyError
-from gnpy.core.utils import round2float, convert_length, psd2powerdbm, lin2db, watt2dbm, dbm2watt
+from gnpy.core.utils import round2float, convert_length, psd2powerdbm, lin2db, watt2dbm, dbm2watt, automatic_nch, \
+    find_common_range
 from gnpy.core.info import ReferenceCarrier, create_input_spectral_information
 from gnpy.core.parameters import SimParams, EdfaParams
 from gnpy.core.science_utils import RamanSolver
@@ -354,6 +356,48 @@ def set_amplifier_voa(amp, power_target, power_mode):
         amp.out_voa = voa
 
 
+def get_oms_edge_list(oms_ingress_node: Union[elements.Roadm, elements.Transceiver], network: DiGraph) \
+        -> List[Tuple]:
+    """get the list of OMS edges (node, neighbour next node) starting from its ingress down to its egress
+    oms_ingress_node can be a ROADM or a Transceiver
+    """
+    oms_edges = []
+    node = oms_ingress_node
+    visited_nodes = []
+    # collect the OMS element list (ROADM to ROADM, or Transceiver to ROADM)
+    while not (isinstance(node, elements.Roadm) or isinstance(node, elements.Transceiver)):
+        next_node = get_next_node(node, network)
+        visited_nodes.append(node.uid)
+        if next_node.uid in visited_nodes:
+            raise NetworkTopologyError(f'Loop detected for {type(node).__name__} {node.uid}, '
+                                       + 'please check network topology')
+        oms_edges.append((node, next_node))
+        node = next_node
+
+    return oms_edges
+
+
+def check_oms_single_type(oms_edges: List[Tuple]) -> List[str]:
+    """Verifies that the OMS only contains all single band amplifiers or all multi band amplifiers
+    No mixed OMS are permitted for the time being.
+    returns the amplifiers'type of the OMS
+    """
+    oms_types = {}
+    for node, _ in oms_edges:
+        if isinstance(node, elements.Edfa):
+            oms_types[node.uid] = 'Edfa'
+        elif isinstance(node, elements.Multiband_amplifier):
+            oms_types[node.uid] = 'Multiband_amplifier'
+    # checks that the element in the OMS are consistant (no multi-band mixed with single band)
+    types = set(list(oms_types.values()))
+    if len(types) > 1:
+        msg = 'type_variety Multiband ("Multiband_amplifier") and single band ("Edfa") cannot be mixed;\n' \
+            + f'Multiband amps: {[e for e in oms_types.keys() if oms_types[e] == "Multiband_amplifier"]}\n' \
+            + f'single band amps: {[e for e in oms_types.keys() if oms_types[e] == "Edfa"]}'
+        raise NetworkTopologyError(msg)
+    return list(types)
+
+
 def set_egress_amplifier(network, this_node, equipment, pref_ch_db, pref_total_db, verbose):
     """This node can be a transceiver or a ROADM (same function called in both cases).
     go through each link staring from this_node until next Roadm or Transceiver and
@@ -544,6 +588,80 @@ def set_roadm_per_degree_targets(roadm, network):
                 raise ConfigurationError(roadm.uid, 'needs an equalization target')
 
 
+def set_per_degree_design_band(node: Union[elements.Roadm, elements.Transceiver], network: DiGraph, equipment: dict):
+    """Configures the design bands for each degree of a node based on network and equipment constraints.
+    This function determines the design bands for each degree of a node (either a ROADM or a Transceiver)
+    based on the existing amplifier types and spectral information (SI) constraints. It uses a default
+    design band derived from the SI or ROADM bands if no specific bands are defined by the user.
+    node.params.x contains the values initially defined by user (with x in design_bands,
+    per_degree_design_bands). node.x contains the autodesign values.
+
+    Parameters:
+        node (Node): The node for which design bands are being set.
+        network (Network): The network containing the node and its connections.
+        equipment (dict): A dictionary containing equipment data, including spectral information.
+
+    Raises:
+        NetworkTopologyError: If there is an inconsistency in band definitions or unsupported configurations.
+
+    Notes:
+        - The function prioritizes user-defined bands in `node.params` if available.
+        - It checks for consistency between default bands and amplifier types.
+        - Mixed single-band and multi-band configurations are not supported and will raise an error.
+        - The function ensures that all bands are ordered by their minimum frequency.
+    """
+    next_oms = (n for n in network.successors(node))
+    if len(node.design_bands) == 0:
+        node.design_bands = [{'f_min': si.f_min, 'f_max': si.f_max} for si in equipment['SI'].values()]
+
+    default_is_single_band = len(node.design_bands) == 1
+    for next_node in next_oms:
+        # get all the elements from the OMS and retrieve their amps types and bands
+        oms_edges = get_oms_edge_list(next_node, network)
+        amps_type = check_oms_single_type(oms_edges)
+        oms_is_single_band = "Edfa" in amps_type if len(amps_type) == 1 else None
+        # oms_is_single_band can be True (single band OMS), False (Multiband OMS) or None (undefined: waiting for
+        # autodesign).
+        el_list = [n for n, _ in oms_edges]
+        amp_bands = [n.params.bands for n in el_list if isinstance(n, (elements.Edfa, elements.Multiband_amplifier))
+                     and n.params.bands]
+        # Use node.design_bands constraints if they are consistent with the amps type
+        if oms_is_single_band == default_is_single_band:
+            amp_bands.append(node.design_bands)
+
+        common_range = find_common_range(amp_bands, None, None)
+        # node.per_degree_design_bands has already been populated with node.params.per_degree_design_bands loaded
+        # from the json.
+        # let's complete the dict with the design band of degrees for which there was no definition
+        if next_node.uid not in node.per_degree_design_bands:
+            if common_range:
+                # if degree design band was not defined, then use the common_range computed with the oms amplifiers
+                # already defined
+                node.per_degree_design_bands[next_node.uid] = common_range
+            elif oms_is_single_band is None or (oms_is_single_band == default_is_single_band):
+                # else if no amps are defined (no bands) then use default ROADM bands
+                # use default ROADM bands only if this is consistent with the oms amps type
+                node.per_degree_design_bands[next_node.uid] = node.design_bands
+            else:
+                # unsupported case: single band OMS with default multiband design band
+                raise NetworkTopologyError(f"in {node.uid} degree {next_node.uid}: inconsistent design multiband/"
+                                           + " single band definition on a single band/ multiband OMS")
+        if next_node.uid in node.params.per_degree_design_bands:
+            # order bands per min frequency in params.per_degree_design_bands for those degree that are defined there
+            node.params.per_degree_design_bands[next_node.uid] = \
+                sorted(node.params.per_degree_design_bands[next_node.uid], key=lambda x: x['f_min'])
+        # order the bands per min frequency in .per_degree_design_bands (all degrees must exist there)
+        node.per_degree_design_bands[next_node.uid] = \
+            sorted(node.per_degree_design_bands[next_node.uid], key=lambda x: x['f_min'])
+    # check node.params.per_degree_design_bands keys
+    if node.params.per_degree_design_bands:
+        next_oms_uid = [n.uid for n in network.successors(node)]
+        for degree in node.params.per_degree_design_bands.keys():
+            if degree not in next_oms_uid:
+                raise NetworkTopologyError(f"in {node.uid} degree {degree} does not match any degree"
+                                           + f"{list(node.per_degree_design_bands.keys())}")
+
+
 def set_roadm_input_powers(network, roadm, equipment, pref_ch_db):
     """Set reference powers at ROADM input for a reference channel and based on the adjacent OMS.
     This supposes that there is no dependency on path. For example, the succession:
@@ -935,6 +1053,9 @@ def build_network(network, equipment, pref_ch_db, pref_total_db, set_connector_l
     for roadm in roadms:
         set_roadm_ref_carrier(roadm, equipment)
         set_roadm_per_degree_targets(roadm, network)
+        set_per_degree_design_band(roadm, network, equipment)
+    for transceiver in transceivers:
+        set_per_degree_design_band(transceiver, network, equipment)
     # then set amplifiers gain, delta_p and out_voa on each OMS
     for roadm in roadms + transceivers:
         set_egress_amplifier(network, roadm, equipment, pref_ch_db, pref_total_db, verbose)
@@ -950,6 +1071,9 @@ def design_network(reference_channel, network, equipment, set_connector_losses=T
     print all warnings or not
     """
     pref_ch_db = watt2dbm(reference_channel.power)  # reference channel power
-    pref_total_db = pref_ch_db + lin2db(reference_channel.nb_channel)  # reference total power
+    # reference total power (limited to C band till C+L autodesign is not solved)
+    designed_nb_channel = min(reference_channel.nb_channel,
+                              automatic_nch(191.0e12, 196.2e12, reference_channel.spacing))
+    pref_total_db = pref_ch_db + lin2db(designed_nb_channel)
     build_network(network, equipment, pref_ch_db, pref_total_db, set_connector_losses=set_connector_losses,
                   verbose=verbose)

gnpy/core/parameters.py

@@ -118,6 +118,8 @@ def __init__(self, **kwargs):
         except KeyError as e:
             raise ParametersError(f'ROADM configurations must include {e}. Configuration: {kwargs}')
         self.per_degree_impairments = kwargs.get('per_degree_impairments', [])
+        self.design_bands = kwargs.get('design_bands', [])
+        self.per_degree_design_bands = kwargs.get('per_degree_design_bands', {})
 
     def get_roadm_path_impairments(self, path_impairments_list):
         """Get the ROADM list of profiles for impairments definition
@@ -643,6 +645,12 @@ def update_attr(self, kwargs):
                 setattr(self, k, clean_kwargs.get(k, v))
 
 
+class TransceiverParams:
+    def __init__(self, **params):
+        self.design_bands = params.get('design_bands', [])
+        self.per_degree_design_bands = params.get('per_degree_design_bands', {})
+
+
 @dataclass
 class FrequencyBand:
     """Frequency band

gnpy/tools/json_io.py

@@ -466,6 +466,11 @@ def _equipment_from_json(json_data, filename):
     equipment = _update_dual_stage(equipment)
     equipment = _update_band(equipment)
     _roadm_restrictions_sanity_check(equipment)
+    possible_SI = list(equipment['SI'].keys())
+    if 'default' not in possible_SI:
+        # Use "default" key in the equipment, using the first listed keys
+        equipment['SI']['default'] = equipment['SI'][possible_SI[0]]
+        del equipment['SI'][possible_SI[0]]
     return equipment
 
 
@@ -516,6 +521,8 @@ def network_from_json(json_data, equipment):
         typ = el_config.pop('type')
         variety = el_config.pop('type_variety', 'default')
         cls = _cls_for(typ)
+        if typ == 'Transceiver':
+            temp = el_config.setdefault('params', {})
         if typ == 'Multiband_amplifier':
             if variety in ['default', '']:
                 extra_params = None