Refactor equipment and add some tests

This fixes error message for wrong trx type,  catches the case of
KeyError when trx_type is not part of the library.

removes power setting from this function: power out of transceiver or
at the input of span is nor defined in equipment Transceiver

Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Change-Id: I15fa7cc772ab5c1a8c7637738eb83c2ddffa1219

gnpy/core/equipment.py

@@ -8,34 +8,39 @@ gnpy.core.equipment
 This module contains functionality for specifying equipment.
 """
 
-from gnpy.core.utils import automatic_nch, db2lin
 from gnpy.core.exceptions import EquipmentConfigError
 
 
 def trx_mode_params(equipment, trx_type_variety='', trx_mode='', error_message=False):
-    """return the trx and SI parameters from eqpt_config for a given type_variety and mode (ie format)"""
+    """return the trx and SI parameters from eqpt_config for a given type_variety and mode (ie format)
+
+    if the type or mode do no match an existing transceiver in the library, then the function
+    raises an error if error_message is True else returns a default mode based on equipment['SI']['default']
+    If trx_mode is None (but type is valid), it returns an undetermined mode whatever the error message:
+    this is a special case for automatic mode selection.
+    """
     trx_params = {}
     default_si_data = equipment['SI']['default']
-
+    # default transponder characteristics
-    try:
+    # mainly used with transmission_main_example.py
-        trxs = equipment['Transceiver']
+    default_trx_params = {
-        # if called from path_requests_run.py, trx_mode is filled with None when not specified by user
+        'f_min': default_si_data.f_min,
-        # if called from transmission_main.py, trx_mode is ''
+        'f_max': default_si_data.f_max,
-        if trx_mode is not None:
+        'baud_rate': default_si_data.baud_rate,
-            mode_params = next(mode for trx in trxs
+        'spacing': default_si_data.spacing,
-                               if trx == trx_type_variety
+        'OSNR': None,
-                               for mode in trxs[trx].mode
+        'penalties': {},
-                               if mode['format'] == trx_mode)
+        'bit_rate': None,
-            trx_params = {**mode_params}
+        'cost': None,
-            # sanity check: spacing baudrate must be smaller than min spacing
+        'roll_off': default_si_data.roll_off,
-            if trx_params['baud_rate'] > trx_params['min_spacing']:
+        'tx_osnr': default_si_data.tx_osnr,
-                raise EquipmentConfigError(f'Inconsistency in equipment library:\n Transponder "{trx_type_variety}"'
+        'min_spacing': None,
-                                           + f' mode "{trx_params["format"]}" has baud rate'
+        'equalization_offset_db': 0
-                                           + f' {trx_params["baud_rate"] * 1e-9:.3f} GHz greater than min_spacing'
+    }
-                                           + f' {trx_params["min_spacing"] * 1e-9:.3f}.')
+    # Undetermined transponder characteristics
-            trx_params['equalization_offset_db'] = trx_params.get('equalization_offset_db', 0)
+    # mainly used with path_request_run.py for the automatic mode computation case
-        else:
+    undetermined_trx_params = {
-            mode_params = {"format": "undetermined",
+        "format": "undetermined",
         "baud_rate": None,
         "OSNR": None,
         "penalties": None,
@@ -44,34 +49,34 @@ def trx_mode_params(equipment, trx_type_variety='', trx_mode='', error_message=F
         "tx_osnr": None,
         "min_spacing": None,
         "cost": None,
-                           "equalization_offset_db": 0}
+        "equalization_offset_db": 0
-            trx_params = {**mode_params}
+    }
-        trx_params['f_min'] = equipment['Transceiver'][trx_type_variety].frequency['min']
-        trx_params['f_max'] = equipment['Transceiver'][trx_type_variety].frequency['max']
-
-        # TODO: novel automatic feature maybe unwanted if spacing is specified
-        # trx_params['spacing'] = _automatic_spacing(trx_params['baud_rate'])
-        # temp = trx_params['spacing']
-        # print(f'spacing {temp}')
-    except StopIteration:
-        if error_message:
-            raise EquipmentConfigError(f'Could not find transponder "{trx_type_variety}" with mode "{trx_mode}" in equipment library')
-        else:
-            # default transponder charcteristics
-            # mainly used with transmission_main_example.py
-            trx_params['f_min'] = default_si_data.f_min
-            trx_params['f_max'] = default_si_data.f_max
-            trx_params['baud_rate'] = default_si_data.baud_rate
-            trx_params['spacing'] = default_si_data.spacing
-            trx_params['OSNR'] = None
-            trx_params['penalties'] = {}
-            trx_params['bit_rate'] = None
-            trx_params['cost'] = None
-            trx_params['roll_off'] = default_si_data.roll_off
-            trx_params['tx_osnr'] = default_si_data.tx_osnr
-            trx_params['min_spacing'] = None
-            trx_params['equalization_offset_db'] = 0
-
-    trx_params['power'] = db2lin(default_si_data.power_dbm) * 1e-3
 
+    trxs = equipment['Transceiver']
+    if trx_type_variety in trxs:
+        modes = {mode['format']: mode for mode in trxs[trx_type_variety].mode}
+        trx_frequencies = {'f_min': trxs[trx_type_variety].frequency['min'],
+                           'f_max': trxs[trx_type_variety].frequency['max']}
+        if trx_mode in modes:
+            # if called from transmission_main.py, trx_mode is ''
+            trx_params = {**modes[trx_mode], **trx_frequencies}
+            if trx_params['baud_rate'] > trx_params['min_spacing']:
+                # sanity check: baudrate must be smaller than min spacing
+                raise EquipmentConfigError(f'Inconsistency in equipment library:\n Transponder "{trx_type_variety}" '
+                                           + f'mode "{trx_params["format"]}" has baud rate '
+                                           + f'{trx_params["baud_rate"] * 1e-9:.2f} GHz greater than min_spacing '
+                                           + f'{trx_params["min_spacing"] * 1e-9:.2f}.')
+            trx_params['equalization_offset_db'] = trx_params.get('equalization_offset_db', 0)
+            return trx_params
+        if trx_mode is None:
+            # if called from path_requests_run.py, trx_mode is filled with None when not specified by user
+            trx_params = {**undetermined_trx_params, **trx_frequencies}
+            return trx_params
+    if trx_type_variety in trxs and error_message:
+        raise EquipmentConfigError(f'Could not find transponder "{trx_type_variety}" with mode "{trx_mode}" '
+                                   + 'in equipment library')
+    if error_message:
+        raise EquipmentConfigError(f'Could not find transponder "{trx_type_variety}" in equipment library')
+
+    trx_params = {**default_trx_params}
     return trx_params

gnpy/tools/cli_examples.py

@@ -193,8 +193,9 @@ def transmission_main_example(args=None):
     params['path_bandwidth'] = 0
     params['effective_freq_slot'] = None
     trx_params = trx_mode_params(equipment)
+    trx_params['power'] = dbm2watt(equipment['SI']['default'].power_dbm)
     if args.power:
-        trx_params['power'] = db2lin(float(args.power)) * 1e-3
+        trx_params['power'] = dbm2watt(float(args.power))
     params.update(trx_params)
     initial_spectrum = None
     params['nb_channel'] = automatic_nch(trx_params['f_min'], trx_params['f_max'], trx_params['spacing'])
@@ -370,7 +371,8 @@ def path_requests_run(args=None):
         'path_bandwidth': 0,
         'effective_freq_slot': None,
         'nb_channel': automatic_nch(equipment['SI']['default'].f_min, equipment['SI']['default'].f_max,
-                                    equipment['SI']['default'].spacing)
+                                    equipment['SI']['default'].spacing),
+        'power': dbm2watt(equipment['SI']['default'].power_dbm)
     }
     trx_params = trx_mode_params(equipment)
     params.update(trx_params)

gnpy/tools/json_io.py

@@ -20,7 +20,7 @@ from gnpy.core.equipment import trx_mode_params
 from gnpy.core.exceptions import ConfigurationError, EquipmentConfigError, NetworkTopologyError, ServiceError
 from gnpy.core.science_utils import estimate_nf_model
 from gnpy.core.info import Carrier
-from gnpy.core.utils import automatic_nch, automatic_fmax, merge_amplifier_restrictions
+from gnpy.core.utils import automatic_nch, automatic_fmax, merge_amplifier_restrictions, dbm2watt
 from gnpy.core.parameters import DEFAULT_RAMAN_COEFFICIENT, EdfaParams
 from gnpy.topology.request import PathRequest, Disjunction, compute_spectrum_slot_vs_bandwidth
 from gnpy.topology.spectrum_assignment import mvalue_to_slots
@@ -584,13 +584,11 @@ def requests_from_json(json_data, equipment):
             msg = f'Equipment Config error in {req["request-id"]}: {e}'
             raise EquipmentConfigError(msg) from e
         params.update(trx_params)
-        # optical power might be set differently in the request. if it is indicated then the
+        params['power'] = req['path-constraints']['te-bandwidth'].get('output-power')
-        # params['power'] is updated
+        # params must not be None, but user can set to None: catch this case
-        try:
+        if params['power'] is None:
-            if req['path-constraints']['te-bandwidth']['output-power']:
+            params['power'] = dbm2watt(equipment['SI']['default'].power_dbm)
-                params['power'] = req['path-constraints']['te-bandwidth']['output-power']
+
-        except KeyError:
-            pass
         # same process for nb-channel
         f_min = params['f_min']
         f_max_from_si = params['f_max']

tests/test_disjunction.py

@@ -258,7 +258,7 @@ def request_set():
         'f_min': 191.1e12,
         'f_max': 196.3e12,
         'nb_channel': None,
-        'power': 0,
+        'power': 0.001,
         'path_bandwidth': 200e9}
 
 

tests/test_logger.py

@@ -181,7 +181,7 @@ def wrong_requests():
             }]
         },
         'expected_msg': 'Equipment Config error in imposed_mode: '
-                        + 'Could not find transponder "test_offset" with mode "mode 3" in equipment library'
+                        + 'Could not find transponder "test_offset" in equipment library'
     })
     data.append({
         'error': ServiceError,

tests/test_roadm_restrictions.py

@@ -254,7 +254,8 @@ def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm, roa
               'format': '',
               'path_bandwidth': 100e9,
               'effective_freq_slot': None,
-              'nb_channel': nb_channel
+              'nb_channel': nb_channel,
+              'power': dbm2watt(power_dbm)
               }
     trx_params = trx_mode_params(equipment)
     params.update(trx_params)
@@ -309,7 +310,8 @@ def create_per_oms_request(network, eqpt, req_power):
         'format': '',
         'path_bandwidth': 100e9,
         'effective_freq_slot': None,
-        'nb_channel': nb_channel
+        'nb_channel': nb_channel,
+        'power': dbm2watt(req_power)
     }
     trx_params = trx_mode_params(eqpt)
     params.update(trx_params)

tests/test_trx_mode_params.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# @Author: Esther Le Rouzic
+# @Date:   2023-09-29
+"""
+@author: esther.lerouzic
+checks all possibilities of this function
+
+"""
+
+from pathlib import Path
+import pytest
+
+from gnpy.core.equipment import trx_mode_params
+from gnpy.core.exceptions import EquipmentConfigError
+from gnpy.tools.json_io import load_equipment, load_json, _equipment_from_json
+
+
+TEST_DIR = Path(__file__).parent
+EQPT_LIBRARY_NAME = TEST_DIR / 'data/eqpt_config.json'
+NETWORK_FILE_NAME = TEST_DIR / 'data/testTopology_expected.json'
+
+
+@pytest.mark.parametrize('trx_type, trx_mode, error_message, no_error, expected_result',
+    [('', '', False, True, "SI"),
+     ('', '', True, False, 'Could not find transponder "" in equipment library'),
+     ('vendorA_trx-type1', '', True, False,
+      'Could not find transponder "vendorA_trx-type1" with mode "" in equipment library'),
+     ('vendorA_trx-type1', '', False, True, "SI"),
+     ('', 'mode 1', True, False, 'Could not find transponder "" in equipment library'),
+     ('', 'mode 1', False, True, "SI"),
+     ('vendorA_trx-type1', 'mode 2', True, True, 'mode 2'),
+     ('vendorA_trx-type1', 'mode 2', False, True, 'mode 2'),
+     ('wrong type', '', True, False, 'Could not find transponder "wrong type" in equipment library'),
+     ('wrong type', '', False, True, 'SI'),
+     ('vendorA_trx-type1', 'wrong mode', True, False,
+      'Could not find transponder "vendorA_trx-type1" with mode "wrong mode" in equipment library'),
+     ('vendorA_trx-type1', 'wrong mode', False, True, 'SI'),
+     ('wrong type', 'wrong mode', True, False, 'Could not find transponder "wrong type" in equipment library'),
+     ('wrong type', 'wrong mode', False, True, 'SI'),
+     ('vendorA_trx-type1', None, True, True, 'None'),
+     ('vendorA_trx-type1', None, False, True, 'None'),
+     (None, None, True, False, 'Could not find transponder "None" in equipment library'),
+     (None, None, False, True, 'SI'),
+     (None, 'mode 2', True, False, 'Could not find transponder "None" in equipment library'),
+     (None, 'mode 2', False, True, 'SI'),
+     ])
+def test_trx_mode_params(trx_type, trx_mode, error_message, no_error, expected_result):
+    """Checks all combinations of trx_type and mode
+    """
+    possible_results = {}
+    possible_results["SI"] = {
+        'OSNR': None,
+        'baud_rate': 32000000000.0,
+        'bit_rate': None,
+        'cost': None,
+        'equalization_offset_db': 0,
+        'f_max': 196100000000000.0,
+        'f_min': 191300000000000.0,
+        'min_spacing': None,
+        'penalties': {},
+        'roll_off': 0.15,
+        'spacing': 50000000000.0,
+        'tx_osnr': 100
+    }
+    possible_results["mode 2"] = {
+        'format': 'mode 2',
+        'baud_rate': 64e9,
+        'OSNR': 15,
+        'bit_rate': 200e9,
+        'roll_off': 0.15,
+        'tx_osnr': 100,
+        'equalization_offset_db': 0,
+        'min_spacing': 75e9,
+        'f_max': 196100000000000.0,
+        'f_min': 191350000000000.0,
+        'penalties': {},
+        'cost': 1
+    }
+    possible_results["None"] = {
+        'format': 'undetermined',
+        'baud_rate': None,
+        'OSNR': None,
+        'bit_rate': None,
+        'roll_off': None,
+        'tx_osnr': None,
+        'equalization_offset_db': 0,
+        'min_spacing': None,
+        'f_max': 196100000000000.0,
+        'f_min': 191350000000000.0,
+        'penalties': None,
+        'cost': None
+    }
+    equipment = load_equipment(EQPT_LIBRARY_NAME)
+    if no_error:
+        trx_params = trx_mode_params(equipment, trx_type, trx_mode, error_message)
+        print(trx_params)
+        assert trx_params == possible_results[expected_result]
+    else:
+        with pytest.raises(EquipmentConfigError, match=expected_result):
+            _ = trx_mode_params(equipment, trx_type, trx_mode, error_message)
+
+
+@pytest.mark.parametrize('baudrate, spacing, error_message',
+    [(60e9, 50e9, 'Inconsistency in equipment library:\n Transponder "vendorB_trx-type1" mode "wrong mode" '
+                  + 'has baud rate 60.00 GHz greater than min_spacing 50.00.'),
+     (32e9, 50, 'Inconsistency in equipment library:\n Transponder "vendorB_trx-type1" mode "wrong mode" '
+                + 'has baud rate 32.00 GHz greater than min_spacing 0.00.')])
+def test_wrong_baudrate_spacing(baudrate, spacing, error_message):
+    """Checks wrong values for baudrate and spacing correctly raise an error
+    """
+    json_data = load_json(EQPT_LIBRARY_NAME)
+    wrong_transceiver = {
+        'type_variety': 'vendorB_trx-type1',
+        'frequency': {
+            'min': 191.35e12,
+            'max': 196.1e12
+        },
+        'mode': [{
+            'format': 'PS_SP64_1',
+            'baud_rate': 32e9,
+            'OSNR': 11,
+            'bit_rate': 100e9,
+            'roll_off': 0.15,
+            'tx_osnr': 100,
+            'min_spacing': 50e9,
+            'cost': 1,
+            'penalties': [{
+                'chromatic_dispersion': 80000,
+                'penalty_value': 0.5
+            }, {
+                'pmd': 120,
+                'penalty_value': 0.5}],
+            'equalization_offset_db': 0
+        }, {
+            'format': 'wrong mode',
+            'baud_rate': baudrate,
+            'OSNR': 11,
+            'bit_rate': 100e9,
+            'roll_off': 0.15,
+            'tx_osnr': 40,
+            'min_spacing': spacing,
+            'cost': 1,
+            'penalties': [{
+                'chromatic_dispersion': 80000,
+                'penalty_value': 0.5
+            }, {
+                'pmd': 120,
+                'penalty_value': 0.5}],
+            'equalization_offset_db': 0}]
+    }
+    json_data['Transceiver'].append(wrong_transceiver)
+    equipment = _equipment_from_json(json_data, EQPT_LIBRARY_NAME)
+
+    with pytest.raises(EquipmentConfigError, match=error_message):
+        _ = trx_mode_params(equipment, 'vendorB_trx-type1', 'wrong mode', error_message=False)