Introduce multi band amps

Introduce a new multi-band element that contains a list of Edfa element:
- reads multiple amps out of the element config.
- deduces frequency band from the amp in the list.

no autodesign yet: multi-band amps must have type_variety.

- checks that type variety of individual EDFAs is consistent with multiband
type variety
- demux and mux spectrum when propagate in multiband
- don't add a preamp or booster if a multiband amp is already defined.

The print of channel number is removed from equipment, since the channel number
may now depend on the path's amplifiers. This changes invocation results layout.

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

gnpy/core/elements.py

@@ -25,16 +25,16 @@
 from scipy.constants import h, c
 from scipy.interpolate import interp1d
 from collections import namedtuple
-from typing import Union
+from typing import Union, List
 from logging import getLogger
 import warnings
 
 from gnpy.core.utils import lin2db, db2lin, arrange_frequencies, snr_sum, per_label_average, pretty_summary_print, \
-    watt2dbm, psd2powerdbm, calculate_absolute_min_or_zero
+    watt2dbm, psd2powerdbm, calculate_absolute_min_or_zero, nice_column_str
 from gnpy.core.parameters import RoadmParams, FusedParams, FiberParams, PumpParams, EdfaParams, EdfaOperational, \
-    RoadmPath, RoadmImpairment
+    MultiBandParams, RoadmPath, RoadmImpairment, find_band_name, FrequencyBand
 from gnpy.core.science_utils import NliSolver, RamanSolver
-from gnpy.core.info import SpectralInformation, demuxed_spectral_information
+from gnpy.core.info import SpectralInformation, muxed_spectral_information, demuxed_spectral_information
 from gnpy.core.exceptions import NetworkTopologyError, SpectrumError, ParametersError
 
 
@@ -1211,3 +1211,127 @@ def __call__(self, spectral_info):
             self.propagate(spectral_info)
             return spectral_info
         raise ValueError(f'Amp {self.uid} Defined propagation band does not match amplifiers band.')
+
+
+class Multiband_amplifier(_Node):
+    """Represents a multiband amplifier that manages multiple amplifiers across different frequency bands.
+
+    This class allows for the initialization and management of amplifiers, each associated with a specific
+    frequency band. It provides methods for signal propagation through the amplifiers and for exporting
+    to JSON format.
+
+    Parameters:
+    -----------
+    amplifiers : list of dict
+        A list of dictionaries, each containing parameters for setting an individual amplifier.
+    params : dict
+        A dictionary of parameters for the multiband amplifier, which must include necessary configuration
+        settings.
+    *args, **kwargs :
+        Additional positional and keyword arguments passed to the parent class `_Node`.
+
+    Attributes:
+    -----------
+    variety_list : list
+        A list of varieties associated with the amplifier.
+    amplifiers : dict
+        A dictionary mapping band names to their corresponding amplifier instances.
+
+    Methods:
+    --------
+    __call__(spectral_info):
+        Propagates the input spectral information through each amplifier and returns the multiplexed spectrum.
+
+    to_json:
+        Converts the amplifier's state to a JSON-compatible dictionary.
+
+    __repr__():
+        Returns a string representation of the multiband amplifier instance.
+
+    __str__():
+        Returns a formatted string representation of the multiband amplifier and its amplifiers.
+
+    Raises:
+    -------
+    ParametersError:
+        If there are conflicting amplifier definitions for the same frequency band during initialization.
+
+    ValueError:
+        If the input spectral information does not match any defined amplifier bands during propagation.
+    """
+    # separate the top level type_variety from kwargs to avoid having multiple type_varieties on each element processing
+    def __init__(self, *args, amplifiers: List[dict], params: dict, **kwargs):
+        self.variety_list = kwargs.pop('variety_list', None)
+        try:
+            super().__init__(params=MultiBandParams(**params), **kwargs)
+        except ParametersError as e:
+            raise ParametersError(f'{kwargs["uid"]}: {e}')
+        self.amplifiers = {}
+        if 'type_variety' in kwargs:
+            kwargs.pop('type_variety')
+        self.passive = False
+        for amp_dict in amplifiers:
+            # amplifiers dict uses default names as key to represent the band
+            amp = Edfa(**amp_dict, **kwargs)
+            band = next(b for b in amp.params.bands)
+            band_name = find_band_name(FrequencyBand(f_min=band["f_min"], f_max=band["f_max"]))
+            if band_name not in self.amplifiers.keys() and band not in self.params.bands:
+                self.params.bands.append(band)
+                self.amplifiers[band_name] = amp
+            elif band_name not in self.amplifiers.keys() and band in self.params.bands:
+                self.amplifiers[band_name] = amp
+            else:
+                raise ParametersError(f'{kwargs["uid"]}: has more than one amp defined for the same band')
+
+    def __call__(self, spectral_info: SpectralInformation):
+        """propagates in each amp and returns the muxed spectrum
+        """
+        out_si = []
+        for _, amp in self.amplifiers.items():
+            si = demuxed_spectral_information(spectral_info, amp.params.bands[0])
+            # if spectral_info frequencies are outside amp band, si is None
+            if si:
+                si = amp(si)
+                out_si.append(si)
+        if not out_si:
+            raise ValueError('Defined propagation band does not match amplifiers band.')
+        return muxed_spectral_information(out_si)
+
+    @property
+    def to_json(self):
+        return {'uid': self.uid,
+                'type': type(self).__name__,
+                'type_variety': self.type_variety,
+                'amplifiers': [{
+                    'type_variety': amp.type_variety,
+                    'operational': {
+                        'gain_target': round(amp.effective_gain, 6),
+                        'delta_p': amp.delta_p,
+                        'tilt_target': amp.tilt_target,
+                        'out_voa': amp.out_voa
+                    }} for amp in self.amplifiers.values()
+                ],
+                'metadata': {
+                    'location': self.metadata['location']._asdict()
+                }
+                }
+
+    def __repr__(self):
+        return (f'{type(self).__name__}(uid={self.uid!r}, '
+                f'type_variety={self.type_variety!r}, ')
+
+    def __str__(self):
+        amp_str = [f'{type(self).__name__} {self.uid}',
+                   f'  type_variety:           {self.type_variety}']
+        multi_str_data = []
+        max_width = 0
+        for amp in self.amplifiers.values():
+            lines = amp.__str__().split('\n')
+            # start at index 1 to remove uid from each amp list of strings
+            # records only if amp is used ie si has frequencies in amp) otherwise there is no other string than the uid
+            if len(lines) > 1:
+                max_width = max(max_width, max([len(line) for line in lines[1:]]))
+                multi_str_data.append(lines[1:])
+        # multi_str_data contains lines with each amp str, instead we want to print per column: transpose the string
+        transposed_data = list(map(list, zip(*multi_str_data)))
+        return '\n'.join(amp_str) + '\n' + nice_column_str(data=transposed_data, max_length=max_width + 2, padding=3)

gnpy/core/equipment.py

@@ -7,8 +7,11 @@
 
 This module contains functionality for specifying equipment.
 """
+from collections import defaultdict
+from functools import reduce
+from typing import List
 
-from gnpy.core.exceptions import EquipmentConfigError
+from gnpy.core.exceptions import EquipmentConfigError, ConfigurationError
 
 
 def trx_mode_params(equipment, trx_type_variety='', trx_mode='', error_message=False):
@@ -80,3 +83,50 @@ def trx_mode_params(equipment, trx_type_variety='', trx_mode='', error_message=F
 
     trx_params = {**default_trx_params}
     return trx_params
+
+
+def find_type_variety(amps: List[str], equipment: dict) -> str:
+    """Returns the multiband type_variety associated with a list of single band type_varieties
+    Args:
+        amps (List[str]): A list of single band type_varieties.
+        equipment (dict): A dictionary containing equipment information.
+
+    Returns:
+        str: an amplifier type variety
+    """
+    listes = find_type_varieties(amps, equipment)
+
+    _found_type = list(reduce(lambda x, y: set(x) & set(y), listes))
+    # Given a list of single band amplifiers, find the multiband amplifier whose multi_band group
+    # matches. For example, if amps list contains ["a1_LBAND", "a2_CBAND"], with a1.multi_band = [a1_LBAND, a1_CBAND]
+    # and a2.multi_band = [a1_LBAND, a2_CBAND], then:
+    # possible_type_varieties = {"a1_LBAND": ["a1", "a2"], "a2_CBAND": ["a2"]}
+    # listes = [["a1", "a2"],  ["a2"]]
+    # and _found_type = [a2]
+    if not _found_type:
+        msg = f'{amps} amps do not belong to the same amp type {listes}'
+        raise ConfigurationError(msg)
+    return _found_type[0]
+
+
+def find_type_varieties(amps: List[str], equipment: dict) -> List[List[str]]:
+    """Returns the multiband list of type_varieties associated with a list of single band type_varieties
+    Args:
+        amps (List[str]): A list of single band type_varieties.
+        equipment (dict): A dictionary containing equipment information.
+
+    Returns:
+        List[List[str]]: A list of lists containing the multiband type_varieties
+                         associated with each single band type_variety.
+    """
+    possible_type_varieties = defaultdict(list)
+    for amp_name, amp in equipment['Edfa'].items():
+        if amp.multi_band is not None:
+            for elem in amp.multi_band:
+                # possible_type_varieties stores the list of multiband amp names that list this elem as
+                # a possible amplifier of the multiband group. For example, if "std_medium_gain_multiband"
+                # and "std_medium_gain_multiband_new" contain "std_medium_gain_C" in their "multi_band" list, then:
+                # possible_type_varieties["std_medium_gain_C"] =
+                # ["std_medium_gain_multiband", "std_medium_gain_multiband_new"]
+                possible_type_varieties[elem].append(amp_name)
+    return [possible_type_varieties[a] for a in amps]

gnpy/core/network.py

@@ -422,7 +422,7 @@ def set_egress_amplifier(network, this_node, equipment, pref_ch_db, pref_total_d
                         restrictions = next_node.restrictions['preamp_variety_list']
                     else:
                         restrictions = None
-                    edfa_eqpt = {n: a for n, a in equipment['Edfa'].items()}
+                    edfa_eqpt = {n: a for n, a in equipment['Edfa'].items() if a.type_def != 'multi_band'}
                     edfa_variety, power_reduction = \
                         select_edfa(raman_allowed, gain_target, power_target, edfa_eqpt,
                                     node.uid,
@@ -480,6 +480,35 @@ def set_egress_amplifier(network, this_node, equipment, pref_ch_db, pref_total_d
                     node.target_pch_out_dbm = None
             elif isinstance(node, elements.RamanFiber):
                 _ = span_loss(network, node, equipment, input_power=pref_ch_db + dp)
+            if isinstance(node, elements.Multiband_amplifier):
+                for amp in node.amplifiers.values():
+                    node_loss = span_loss(network, prev_node, equipment)
+                    voa = amp.out_voa if amp.out_voa else 0
+                    if amp.delta_p is None:
+                        dp = target_power(network, next_node, equipment) + voa
+                    else:
+                        dp = amp.delta_p
+                    if amp.effective_gain is None or power_mode:
+                        gain_target = node_loss + dp - prev_dp + prev_voa
+                    else:  # gain mode with effective_gain
+                        gain_target = amp.effective_gain
+                        dp = prev_dp - node_loss - prev_voa + gain_target
+
+                    power_target = pref_total_db + dp
+                    amp.delta_p = dp if power_mode else None
+                    amp.effective_gain = gain_target
+                    set_amplifier_voa(amp, power_target, power_mode)
+                    amp._delta_p = amp.delta_p if power_mode else dp
+                    # target_pch_out_dbm records target power for design: If user defines one, then this is displayed,
+                    # else display the one computed during design
+                    if amp.delta_p is not None and amp.operational.delta_p is not None:
+                        # use the user defined target
+                        amp.target_pch_out_dbm = round(amp.operational.delta_p + pref_ch_db, 2)
+                    elif amp.delta_p is not None:
+                        # use the design target if no target were set
+                        amp.target_pch_out_dbm = round(amp.delta_p + pref_ch_db, 2)
+                    elif amp.delta_p is None:
+                        amp.target_pch_out_dbm = None
             prev_dp = dp
             prev_voa = voa
             prev_node = node
@@ -549,6 +578,10 @@ def set_roadm_input_powers(network, roadm, equipment, pref_ch_db):
                 node.get_per_degree_ref_power(degree=previous_node.uid) - loss
         elif isinstance(node, elements.Transceiver):
             roadm.ref_pch_in_dbm[element.uid] = pref_ch_db - loss
+        elif isinstance(node, elements.Multiband_amplifier):
+            # use the worst (min) value among amps
+            roadm.ref_pch_in_dbm[element.uid] = min([pref_ch_db + amp._delta_p - amp.out_voa - loss
+                                                     for amp in node.amplifiers.values()])
     # check if target power can be met
     temp = []
     if roadm.per_degree_pch_out_dbm:
@@ -598,6 +631,9 @@ def set_fiber_input_power(network, fiber, equipment, pref_ch_db):
         fiber.ref_pch_in_dbm = pref_ch_db + node._delta_p - node.out_voa - loss
     elif isinstance(node, elements.Transceiver):
         fiber.ref_pch_in_dbm = pref_ch_db - loss
+    elif isinstance(node, elements.Multiband_amplifier):
+        # use the worst (min) value among amps
+        fiber.ref_pch_in_dbm = min([pref_ch_db + amp._delta_p - amp.out_voa - loss for amp in node.amplifiers.values()])
 
 
 def set_roadm_internal_paths(roadm, network):
@@ -658,7 +694,8 @@ def set_roadm_internal_paths(roadm, network):
 
 def add_roadm_booster(network, roadm):
     next_nodes = [n for n in network.successors(roadm)
-                  if not isinstance(n, (elements.Transceiver, elements.Fused, elements.Edfa))]
+                  if not isinstance(n, (elements.Transceiver, elements.Fused, elements.Edfa,
+                                        elements.Multiband_amplifier))]
     # no amplification for fused spans or TRX
     for next_node in next_nodes:
         network.remove_edge(roadm, next_node)
@@ -684,7 +721,8 @@ def add_roadm_booster(network, roadm):
 
 def add_roadm_preamp(network, roadm):
     prev_nodes = [n for n in network.predecessors(roadm)
-                  if not isinstance(n, (elements.Transceiver, elements.Fused, elements.Edfa))]
+                  if not isinstance(n, (elements.Transceiver, elements.Fused, elements.Edfa,
+                                        elements.Multiband_amplifier))]
     # no amplification for fused spans or TRX
     for prev_node in prev_nodes:
         network.remove_edge(prev_node, roadm)

gnpy/core/parameters.py

@@ -8,7 +8,8 @@
 This module contains all parameters to configure standard network elements.
 """
 from collections import namedtuple
-
+from copy import deepcopy
+from dataclasses import dataclass
 from scipy.constants import c, pi
 from numpy import asarray, array, exp, sqrt, log, outer, ones, squeeze, append, flip, linspace, full
 
@@ -593,7 +594,7 @@ def __init__(self, **params):
 
     def update_params(self, kwargs):
         for k, v in kwargs.items():
-            setattr(self, k, self.update_params(**v) if isinstance(v, dict) else v)
+            setattr(self, k, v)
 
 
 class EdfaOperational:
@@ -616,3 +617,56 @@ def __repr__(self):
         return (f'{type(self).__name__}('
                 f'gain_target={self.gain_target!r}, '
                 f'tilt_target={self.tilt_target!r})')
+
+
+class MultiBandParams:
+    default_values = {
+        'bands': [],
+        'type_variety': '',
+        'type_def': None,
+        'allowed_for_design': False
+    }
+
+    def __init__(self, **params):
+        try:
+            self.update_attr(params)
+        except KeyError as e:
+            raise ParametersError(f'Multiband configurations json must include {e}. Configuration: {params}')
+
+    def update_attr(self, kwargs):
+        clean_kwargs = {k: v for k, v in kwargs.items() if v != ''}
+        for k, v in self.default_values.items():
+            # use deepcopy to avoid sharing same object amongst all instance when v is a list or a dict!
+            if isinstance(v, (list, dict)):
+                setattr(self, k, clean_kwargs.get(k, deepcopy(v)))
+            else:
+                setattr(self, k, clean_kwargs.get(k, v))
+
+
+@dataclass
+class FrequencyBand:
+    """Frequency band
+    """
+    f_min: float
+    f_max: float
+
+
+DEFAULT_BANDS_DEFINITION = {
+    "LBAND": FrequencyBand(f_min=187e12, f_max=189e12),
+    "CBAND": FrequencyBand(f_min=191.3e12, f_max=196.0e12)
+}
+# use this definition to index amplifiers'element of a multiband amplifier.
+# this is not the design band
+
+
+def find_band_name(band: FrequencyBand) -> str:
+    """return the default band name (CBAND, LBAND, ...) that corresponds to the band frequency range
+    Use the band center frequency: if center frequency is inside the band then returns CBAND.
+    This is to flexibly encompass all kind of bands definitions.
+    returns the first matching band name.
+    """
+    for band_name, frequency_range in DEFAULT_BANDS_DEFINITION.items():
+        center_frequency = (band.f_min + band.f_max) / 2
+        if center_frequency >= frequency_range.f_min and center_frequency <= frequency_range.f_max:
+            return band_name
+    return 'unknown_band'