diff --git a/gnpy/core/equipment.py b/gnpy/core/equipment.py index 50bb20b85..d1b7ab11d 100644 --- a/gnpy/core/equipment.py +++ b/gnpy/core/equipment.py @@ -8,70 +8,76 @@ This module contains functionality for specifying equipment. """ -from gnpy.core.utils import automatic_nch, db2lin +from gnpy.core.utils import dbm2watt 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 + # mainly used with transmission_main_example.py + default_trx_params = { + 'f_min': default_si_data.f_min, + 'f_max': default_si_data.f_max, + 'baud_rate': default_si_data.baud_rate, + 'spacing': default_si_data.spacing, + 'OSNR': None, + 'penalties': {}, + 'bit_rate': None, + 'cost': None, + 'roll_off': default_si_data.roll_off, + 'tx_osnr': default_si_data.tx_osnr, + 'min_spacing': None, + 'equalization_offset_db': 0 + } + # Undetermined transponder characteristics + # mainly used with path_request_run.py for the automatic mode computation case + undetermined_trx_params = { + "format": "undetermined", + "baud_rate": None, + "OSNR": None, + "penalties": None, + "bit_rate": None, + "roll_off": None, + "tx_osnr": None, + "min_spacing": None, + "cost": None, + "equalization_offset_db": 0 + } - try: - trxs = equipment['Transceiver'] - # if called from path_requests_run.py, trx_mode is filled with None when not specified by user - # if called from transmission_main.py, trx_mode is '' - if trx_mode is not None: - mode_params = next(mode for trx in trxs - if trx == trx_type_variety - for mode in trxs[trx].mode - if mode['format'] == trx_mode) - trx_params = {**mode_params} - # sanity check: spacing baudrate must be smaller than min spacing + 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']: - 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:.3f} GHz greater than min_spacing' - + f' {trx_params["min_spacing"] * 1e-9:.3f}.') + # 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) - else: - mode_params = {"format": "undetermined", - "baud_rate": None, - "OSNR": None, - "penalties": None, - "bit_rate": None, - "roll_off": None, - "tx_osnr": None, - "min_spacing": None, - "cost": None, - "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 + 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 diff --git a/gnpy/tools/cli_examples.py b/gnpy/tools/cli_examples.py index 2ae7c18f4..b0d632565 100644 --- a/gnpy/tools/cli_examples.py +++ b/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) diff --git a/gnpy/tools/json_io.py b/gnpy/tools/json_io.py index 834705582..8854ae692 100644 --- a/gnpy/tools/json_io.py +++ b/gnpy/tools/json_io.py @@ -20,7 +20,7 @@ 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'] is updated - try: - if req['path-constraints']['te-bandwidth']['output-power']: - params['power'] = req['path-constraints']['te-bandwidth']['output-power'] - except KeyError: - pass + params['power'] = req['path-constraints']['te-bandwidth'].get('output-power') + # params must not be None, but user can set to None: catch this case + if params['power'] is None: + params['power'] = dbm2watt(equipment['SI']['default'].power_dbm) + # same process for nb-channel f_min = params['f_min'] f_max_from_si = params['f_max'] diff --git a/tests/test_disjunction.py b/tests/test_disjunction.py index 40f055b52..a48ebb681 100644 --- a/tests/test_disjunction.py +++ b/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} diff --git a/tests/test_logger.py b/tests/test_logger.py index 509d9c965..0e671c30e 100644 --- a/tests/test_logger.py +++ b/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, diff --git a/tests/test_roadm_restrictions.py b/tests/test_roadm_restrictions.py index ebe76fa56..0a2568913 100644 --- a/tests/test_roadm_restrictions.py +++ b/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) diff --git a/tests/test_trx_mode_params.py b/tests/test_trx_mode_params.py new file mode 100644 index 000000000..e319cdb91 --- /dev/null +++ b/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)