use explicit file arguments for additional configs

Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Change-Id: I133bb6a2d21d573cf819e1d92b1912dfa87dbfa4
This commit is contained in:
EstherLerouzic
2024-12-11 19:54:31 +01:00
parent 8717156712
commit 4fda8c6002
16 changed files with 161 additions and 112 deletions

View File

@@ -44,14 +44,17 @@ def show_example_data_dir():
print(f'{_examples_dir}/') print(f'{_examples_dir}/')
def load_common_data(equipment_filename: Path, extra_equipment_filenames: List[Path], topology_filename: Path, def load_common_data(equipment_filename: Path, extra_equipment_filenames: List[Path], extra_config_filenames: List[Path],
simulation_filename: Path, save_raw_network_filename: Path): topology_filename: Path, simulation_filename: Path, save_raw_network_filename: Path):
"""Load common configuration from JSON files, merging additional equipment if provided.""" """Load common configuration from JSON files, merging additional equipment if provided."""
try: try:
equipment = load_equipment(equipment_filename) extra_configs = {}
if extra_config_filenames:
extra_configs = {f.name: f for f in extra_config_filenames}
equipment = load_equipment(equipment_filename, extra_configs)
if extra_equipment_filenames: if extra_equipment_filenames:
merge_equipment(equipment, extra_equipment_filenames) merge_equipment(equipment, extra_equipment_filenames, extra_configs)
network = load_network(topology_filename, equipment) network = load_network(topology_filename, equipment)
if save_raw_network_filename is not None: if save_raw_network_filename is not None:
save_network(network, save_raw_network_filename) save_network(network, save_raw_network_filename)
@@ -110,6 +113,12 @@ def _add_common_options(parser: argparse.ArgumentParser, network_default: Path):
parser.add_argument('-x', '--extra-equipment', nargs='+', type=Path, parser.add_argument('-x', '--extra-equipment', nargs='+', type=Path,
metavar=_help_fname_json, default=None, metavar=_help_fname_json, default=None,
help='List of additional equipment files to complement the main equipment file.') help='List of additional equipment files to complement the main equipment file.')
# Option for additional config files
parser.add_argument('-xc', '--extra-config', nargs='+', type=Path,
metavar=_help_fname_json, default=[_examples_dir / "std_medium_gain_advanced_config.json",
_examples_dir / "Juniper-BoosterHG.json"],
help='List of additional config files as referenced in equipment files with '
'"advanced_config_from_json" or "default_config_from_json"')
def transmission_main_example(args=None): def transmission_main_example(args=None):
@@ -133,8 +142,8 @@ def transmission_main_example(args=None):
args = parser.parse_args(args if args is not None else sys.argv[1:]) args = parser.parse_args(args if args is not None else sys.argv[1:])
_setup_logging(args) _setup_logging(args)
(equipment, network) = load_common_data(args.equipment, args.extra_equipment, args.topology, args.sim_params, (equipment, network) = load_common_data(args.equipment, args.extra_equipment, args.extra_config, args.topology,
args.save_network_before_autodesign) args.sim_params, args.save_network_before_autodesign)
if args.plot: if args.plot:
plot_baseline(network) plot_baseline(network)
@@ -322,7 +331,7 @@ def path_requests_run(args=None):
_logger.info(f'Computing path requests {args.service_filename.name} into JSON format') _logger.info(f'Computing path requests {args.service_filename.name} into JSON format')
(equipment, network) = \ (equipment, network) = \
load_common_data(args.equipment, args.extra_equipment, args.topology, args.sim_params, load_common_data(args.equipment, args.extra_equipment, args.extra_config, args.topology, args.sim_params,
args.save_network_before_autodesign) args.save_network_before_autodesign)
# Build the network once using the default power defined in SI in eqpt config # Build the network once using the default power defined in SI in eqpt config

View File

@@ -212,7 +212,7 @@ class Amp(_JsonThing):
self.update_attr(self.default_values, kwargs, 'Amp') self.update_attr(self.default_values, kwargs, 'Amp')
@classmethod @classmethod
def from_json(cls, filename, **kwargs): def from_json(cls, extra_configs, **kwargs):
""" """
""" """
# default EDFA DGT and ripples are defined in parameters DEFAULT_EDFA_CONFIG. copy these values when # default EDFA DGT and ripples are defined in parameters DEFAULT_EDFA_CONFIG. copy these values when
@@ -227,7 +227,7 @@ class Amp(_JsonThing):
if type_def == 'fixed_gain': if type_def == 'fixed_gain':
if 'default_config_from_json' in kwargs: if 'default_config_from_json' in kwargs:
# use user defined default instead of DEFAULT_EDFA_CONFIG # use user defined default instead of DEFAULT_EDFA_CONFIG
config = load_json(Path(filename).parent / kwargs.pop('default_config_from_json')) config = load_json(extra_configs[kwargs.pop('default_config_from_json')])
try: try:
nf0 = kwargs.pop('nf0') nf0 = kwargs.pop('nf0')
except KeyError as exc: # nf0 is expected for a fixed gain amp except KeyError as exc: # nf0 is expected for a fixed gain amp
@@ -241,11 +241,11 @@ class Amp(_JsonThing):
nf_def = Model_fg(nf0) nf_def = Model_fg(nf0)
elif type_def == 'advanced_model': elif type_def == 'advanced_model':
# use the user file name define in library instead of default config # use the user file name define in library instead of default config
config = load_json(Path(filename).parent / kwargs.pop('advanced_config_from_json')) config = load_json(extra_configs[kwargs.pop('advanced_config_from_json')])
elif type_def == 'variable_gain': elif type_def == 'variable_gain':
if 'default_config_from_json' in kwargs: if 'default_config_from_json' in kwargs:
# use user defined default instead of DEFAULT_EDFA_CONFIG # use user defined default instead of DEFAULT_EDFA_CONFIG
config = load_json(Path(filename).parent / kwargs.pop('default_config_from_json')) config = load_json(extra_configs[kwargs.pop('default_config_from_json')])
gain_min, gain_max = kwargs['gain_min'], kwargs['gain_flatmax'] gain_min, gain_max = kwargs['gain_min'], kwargs['gain_flatmax']
try: # nf_min and nf_max are expected for a variable gain amp try: # nf_min and nf_max are expected for a variable gain amp
nf_min = kwargs.pop('nf_min') nf_min = kwargs.pop('nf_min')
@@ -368,14 +368,14 @@ def _spectrum_from_json(json_data: dict):
return spectrum return spectrum
def merge_equipment(equipment: dict, additional_filenames: List[Path]): def merge_equipment(equipment: dict, additional_filenames: List[Path], extra_configs: Dict[str, Path]):
"""Merge additional equipment libraries into the base equipment dictionary. """Merge additional equipment libraries into the base equipment dictionary.
Typical case is the use of third party transceivers which are not part of a the supplier library. Typical case is the use of third party transceivers which are not part of a the supplier library.
raise warnings if the same reference is used on two different libraries raise warnings if the same reference is used on two different libraries
""" """
for filename in additional_filenames: for filename in additional_filenames:
extra_eqpt = load_equipment(filename) extra_eqpt = load_equipment(filename, extra_configs)
# populate with default eqpt to streamline loading # populate with default eqpt to streamline loading
for eqpt_type, extra_items in extra_eqpt.items(): for eqpt_type, extra_items in extra_eqpt.items():
for type_variety, item in extra_items.items(): for type_variety, item in extra_items.items():
@@ -386,11 +386,11 @@ def merge_equipment(equipment: dict, additional_filenames: List[Path]):
_logger.warning(msg) _logger.warning(msg)
def load_equipment(filename: Path) -> dict: def load_equipment(filename: Path, extra_configs: Dict[str, Path]) -> dict:
"""Load equipment, returns equipment dict """Load equipment, returns equipment dict
""" """
json_data = load_json(filename) json_data = load_json(filename)
return _equipment_from_json(json_data, filename) return _equipment_from_json(json_data, extra_configs)
def load_initial_spectrum(filename: Path) -> dict: def load_initial_spectrum(filename: Path) -> dict:
@@ -494,7 +494,7 @@ def _si_sanity_check(equipment):
del equipment['SI'][possible_SI[0]] del equipment['SI'][possible_SI[0]]
def _equipment_from_json(json_data: dict, filename: Path) -> dict: def _equipment_from_json(json_data: dict, extra_configs: Dict[str, Path]) -> dict:
"""build global dictionnary eqpt_library that stores all eqpt characteristics: """build global dictionnary eqpt_library that stores all eqpt characteristics:
edfa type type_variety, fiber type_variety edfa type type_variety, fiber type_variety
from the eqpt_config.json (filename parameter) from the eqpt_config.json (filename parameter)
@@ -509,7 +509,7 @@ def _equipment_from_json(json_data: dict, filename: Path) -> dict:
for entry in entries: for entry in entries:
subkey = entry.get('type_variety', 'default') subkey = entry.get('type_variety', 'default')
if key == 'Edfa': if key == 'Edfa':
equipment[key][subkey] = Amp.from_json(filename, **entry) equipment[key][subkey] = Amp.from_json(extra_configs, **entry)
elif key == 'Fiber': elif key == 'Fiber':
equipment[key][subkey] = Fiber(**entry) equipment[key][subkey] = Fiber(**entry)
elif key == 'Span': elif key == 'Span':

View File

@@ -13,10 +13,12 @@ from gnpy.tools.json_io import load_network, load_equipment, load_json, _equipme
from pathlib import Path from pathlib import Path
import pytest import pytest
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
DATA_DIR = TEST_DIR / 'data' DATA_DIR = TEST_DIR / 'data'
test_network = DATA_DIR / 'test_network.json' test_network = DATA_DIR / 'test_network.json'
eqpt_library = DATA_DIR / 'eqpt_config.json' eqpt_library = DATA_DIR / 'eqpt_config.json'
extra_configs = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
# TODO in elements.py code: pytests doesn't pass with 1 channel: interpolate fail # TODO in elements.py code: pytests doesn't pass with 1 channel: interpolate fail
@@ -39,7 +41,7 @@ def bw():
def setup_edfa_variable_gain(): def setup_edfa_variable_gain():
"""init edfa class by reading test_network.json file """init edfa class by reading test_network.json file
remove all gain and nf ripple""" remove all gain and nf ripple"""
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = load_network(test_network, equipment) network = load_network(test_network, equipment)
build_network(network, equipment, 0, 20) build_network(network, equipment, 0, 20)
edfa = [n for n in network.nodes() if isinstance(n, Edfa)][0] edfa = [n for n in network.nodes() if isinstance(n, Edfa)][0]
@@ -51,7 +53,7 @@ def setup_edfa_variable_gain():
@pytest.fixture() @pytest.fixture()
def setup_edfa_fixed_gain(): def setup_edfa_fixed_gain():
"""init edfa class by reading the 2nd edfa in test_network.json file""" """init edfa class by reading the 2nd edfa in test_network.json file"""
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = load_network(test_network, equipment) network = load_network(test_network, equipment)
build_network(network, equipment, 0, 20) build_network(network, equipment, 0, 20)
edfa = [n for n in network.nodes() if isinstance(n, Edfa)][1] edfa = [n for n in network.nodes() if isinstance(n, Edfa)][1]
@@ -61,7 +63,7 @@ def setup_edfa_fixed_gain():
@pytest.fixture() @pytest.fixture()
def setup_trx(): def setup_trx():
"""init transceiver class to access snr and osnr calculations""" """init transceiver class to access snr and osnr calculations"""
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = load_network(test_network, equipment) network = load_network(test_network, equipment)
build_network(network, equipment, 0, 20) build_network(network, equipment, 0, 20)
trx = [n for n in network.nodes() if isinstance(n, Transceiver)][0] trx = [n for n in network.nodes() if isinstance(n, Transceiver)][0]
@@ -144,7 +146,7 @@ def test_compare_nf_models(gain, setup_edfa_variable_gain, si):
} }
} }
} }
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
extra_params = equipment['Edfa']['CienaDB_medium_gain'] extra_params = equipment['Edfa']['CienaDB_medium_gain']
temp = el_config.setdefault('params', {}) temp = el_config.setdefault('params', {})
temp = merge_amplifier_restrictions(temp, extra_params.__dict__) temp = merge_amplifier_restrictions(temp, extra_params.__dict__)
@@ -165,7 +167,7 @@ def test_ase_noise(gain, si, setup_trx, bw):
2-pout/pase afet propagate 2-pout/pase afet propagate
3-Transceiver osnr_ase_01nm 3-Transceiver osnr_ase_01nm
=> unitary test for Edfa.noise_profile (Edfa.interpol_params, Edfa.propagate)""" => unitary test for Edfa.noise_profile (Edfa.interpol_params, Edfa.propagate)"""
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = load_network(test_network, equipment) network = load_network(test_network, equipment)
edfa = next(n for n in network.nodes() if n.uid == 'Edfa1') edfa = next(n for n in network.nodes() if n.uid == 'Edfa1')
span = next(n for n in network.nodes() if n.uid == 'Span1') span = next(n for n in network.nodes() if n.uid == 'Span1')
@@ -225,7 +227,7 @@ def test_amp_behaviour(tilt_target, delta_p):
}], }],
"connections": [] "connections": []
} }
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
edfa = [n for n in network.nodes() if isinstance(n, Edfa)][0] edfa = [n for n in network.nodes() if isinstance(n, Edfa)][0]
fiber = [n for n in network.nodes() if isinstance(n, Fiber)][0] fiber = [n for n in network.nodes() if isinstance(n, Fiber)][0]
@@ -310,7 +312,7 @@ def test_amp_saturation(delta_pdb_per_channel, base_power, delta_p):
}], }],
"connections": [] "connections": []
} }
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
edfa = [n for n in network.nodes()][0] edfa = [n for n in network.nodes()][0]
frequency = 193e12 + array([0, 50e9, 150e9, 225e9, 275e9]) frequency = 193e12 + array([0, 50e9, 150e9, 225e9, 275e9])
@@ -354,7 +356,7 @@ def test_set_out_voa():
}], }],
"connections": [] "connections": []
} }
equipment = load_equipment(eqpt_library) equipment = load_equipment(eqpt_library, extra_configs)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
amp = [n for n in network.nodes()][0] amp = [n for n in network.nodes()][0]
print(amp.out_voa) print(amp.out_voa)
@@ -408,7 +410,7 @@ def test_multiband():
] ]
equipment_json['Edfa'].extend(amps) equipment_json['Edfa'].extend(amps)
equipment = _equipment_from_json(equipment_json, eqpt_library) equipment = _equipment_from_json(equipment_json, extra_configs)
el_config = { el_config = {
"uid": "Edfa1", "uid": "Edfa1",
@@ -510,6 +512,7 @@ def test_multiband():
def test_user_defined_config(): def test_user_defined_config():
"""Checks that a user defined config is correctly used instead of DEFAULT_EDFA_CONFIG """Checks that a user defined config is correctly used instead of DEFAULT_EDFA_CONFIG
""" """
extra_configs['user_edfa_config.json'] = DATA_DIR / 'user_edfa_config.json'
user_edfa = { user_edfa = {
"type_variety": "user_defined", "type_variety": "user_defined",
"type_def": "variable_gain", "type_def": "variable_gain",
@@ -526,7 +529,7 @@ def test_user_defined_config():
# add the reference to # add the reference to
json_data = load_json(eqpt_library) json_data = load_json(eqpt_library)
json_data['Edfa'].append(user_edfa) json_data['Edfa'].append(user_edfa)
equipment = _equipment_from_json(json_data, eqpt_library) equipment = _equipment_from_json(json_data, extra_configs)
json_data = { json_data = {
"elements": [{ "elements": [{
"uid": "Edfa1", "uid": "Edfa1",
@@ -554,6 +557,7 @@ def test_default_config():
"""Checks that a config using a file gives the exact same result as the default config if values are identical """Checks that a config using a file gives the exact same result as the default config if values are identical
to DEFAULT_EDFA_CONFIG to DEFAULT_EDFA_CONFIG
""" """
extra_configs['copy_default_edfa_config.json'] = DATA_DIR / 'copy_default_edfa_config.json'
user_edfa = { user_edfa = {
"type_variety": "user_defined", "type_variety": "user_defined",
"type_def": "variable_gain", "type_def": "variable_gain",
@@ -583,7 +587,7 @@ def test_default_config():
json_data = load_json(eqpt_library) json_data = load_json(eqpt_library)
json_data['Edfa'].append(user_edfa) json_data['Edfa'].append(user_edfa)
json_data['Edfa'].append(default_edfa) json_data['Edfa'].append(default_edfa)
equipment = _equipment_from_json(json_data, eqpt_library) equipment = _equipment_from_json(json_data, extra_configs)
json_data = { json_data = {
"elements": [{ "elements": [{
"uid": "Edfa1", "uid": "Edfa1",
@@ -617,7 +621,8 @@ def test_default_config():
assert_allclose(amp1.params.dgt, amp2.params.dgt, rtol=1e-13) assert_allclose(amp1.params.dgt, amp2.params.dgt, rtol=1e-13)
@pytest.mark.parametrize("file", [None, "copy_default_edfa_config.json"]) @pytest.mark.parametrize("file", [None, {"name": "copy_default_edfa_config.json",
"path": DATA_DIR / "copy_default_edfa_config.json"}])
def test_frequency_range(file): def test_frequency_range(file):
"""Checks that a frequency range is correctly read from the library and pre-empts DEFAULT_EDFA_CONFIG """Checks that a frequency range is correctly read from the library and pre-empts DEFAULT_EDFA_CONFIG
""" """
@@ -635,11 +640,12 @@ def test_frequency_range(file):
"allowed_for_design": True "allowed_for_design": True
} }
if file: if file:
user_edfa["default_config_from_json"] = file user_edfa["default_config_from_json"] = file['name']
extra_configs[file['name']] = file['path']
# add the reference to # add the reference to
json_data = load_json(eqpt_library) json_data = load_json(eqpt_library)
json_data['Edfa'].append(user_edfa) json_data['Edfa'].append(user_edfa)
equipment = _equipment_from_json(json_data, eqpt_library) equipment = _equipment_from_json(json_data, extra_configs)
json_data = { json_data = {
"elements": [{ "elements": [{
"uid": "Edfa1", "uid": "Edfa1",

View File

@@ -26,10 +26,13 @@ from gnpy.tools.json_io import load_network, load_equipment, requests_from_json,
_equipment_from_json _equipment_from_json
network_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_expected.json' data_dir = Path(__file__).parent.parent / 'tests/data'
service_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_testservices.json' network_file_name = data_dir / 'testTopology_expected.json'
result_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_testresults.json' service_file_name = data_dir / 'testTopology_testservices.json'
eqpt_library_name = Path(__file__).parent.parent / 'tests/data/eqpt_config.json' result_file_name = data_dir / 'testTopology_testresults.json'
eqpt_library_name = data_dir / 'eqpt_config.json'
extra_configs = {"std_medium_gain_advanced_config.json": data_dir / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": data_dir / "Juniper-BoosterHG.json"}
@pytest.mark.parametrize("net", [network_file_name]) @pytest.mark.parametrize("net", [network_file_name])
@@ -39,7 +42,7 @@ eqpt_library_name = Path(__file__).parent.parent / 'tests/data/eqpt_config.json'
'PS_SP64_1', 'mode 3', 'PS_SP64_1', 'PS_SP64_1', '16QAM', 'mode 1', 'PS_SP64_1', 'mode 3', 'PS_SP64_1', 'PS_SP64_1', '16QAM', 'mode 1',
'PS_SP64_1', 'PS_SP64_1', 'mode 1', 'mode 2', 'mode 1', 'mode 2', 'nok']]) 'PS_SP64_1', 'PS_SP64_1', 'mode 1', 'mode 2', 'mode 1', 'mode 2', 'nok']])
def test_automaticmodefeature(net, eqpt, serv, expected_mode): def test_automaticmodefeature(net, eqpt, serv, expected_mode):
equipment = load_equipment(eqpt) equipment = load_equipment(eqpt, extra_configs)
network = load_network(net, equipment) network = load_network(net, equipment)
data = load_requests(serv, eqpt, bidir=False, network=network, network_filename=net) data = load_requests(serv, eqpt, bidir=False, network=network, network_filename=net)
@@ -155,7 +158,7 @@ def test_propagate_and_optimize_mode(caplog):
# change default ROADM PDL so that crossing 2 ROADMs leasd to inifinte penalty for mode 1 # change default ROADM PDL so that crossing 2 ROADMs leasd to inifinte penalty for mode 1
eqpt_roadm = next(r for r in json_data['Roadm'] if 'type_variety' not in r) eqpt_roadm = next(r for r in json_data['Roadm'] if 'type_variety' not in r)
eqpt_roadm['pdl'] = 0.5 eqpt_roadm['pdl'] = 0.5
equipment = _equipment_from_json(json_data, eqpt_library_name) equipment = _equipment_from_json(json_data, extra_configs)
network = load_network(network_file_name, equipment) network = load_network(network_file_name, equipment)
data = load_requests(filename=Path(__file__).parent.parent / 'tests/data/testTopology_services_expected.json', data = load_requests(filename=Path(__file__).parent.parent / 'tests/data/testTopology_services_expected.json',
eqpt=eqpt_library_name, bidir=False, network=network, network_filename=network_file_name) eqpt=eqpt_library_name, bidir=False, network=network, network_filename=network_file_name)

View File

@@ -23,10 +23,14 @@ from gnpy.topology.request import (compute_path_dsjctn, isdisjoint, find_reverse
from gnpy.topology.spectrum_assignment import build_oms_list from gnpy.topology.spectrum_assignment import build_oms_list
from gnpy.tools.json_io import requests_from_json, load_requests, load_network, load_equipment, disjunctions_from_json from gnpy.tools.json_io import requests_from_json, load_requests, load_network, load_equipment, disjunctions_from_json
NETWORK_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_expected.json'
SERVICE_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_testservices.json' DATA_DIR = Path(__file__).parent.parent / 'tests/data'
RESULT_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_testresults.json' NETWORK_FILE_NAME = DATA_DIR / 'testTopology_expected.json'
EQPT_LIBRARY_NAME = Path(__file__).parent.parent / 'tests/data/eqpt_config.json' SERVICE_FILE_NAME = DATA_DIR / 'testTopology_testservices.json'
RESULT_FILE_NAME = DATA_DIR / 'testTopology_testresults.json'
EQPT_LIBRARY_NAME = DATA_DIR / 'eqpt_config.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
@pytest.fixture() @pytest.fixture()
@@ -43,7 +47,7 @@ def serv(test_setup):
@pytest.fixture() @pytest.fixture()
def test_setup(): def test_setup():
"""common setup for tests: builds network, equipment and oms only once""" """common setup for tests: builds network, equipment and oms only once"""
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_LIBRARY_NAME, EXTRA_CONFIGS)
network = load_network(NETWORK_FILE_NAME, equipment) network = load_network(NETWORK_FILE_NAME, equipment)
# Build the network once using the default power defined in SI in eqpt config # Build the network once using the default power defined in SI in eqpt config
# power density : db2linp(ower_dbm': 0)/power_dbm': 0 * nb channels as defined by # power density : db2linp(ower_dbm': 0)/power_dbm': 0 * nb channels as defined by
@@ -297,7 +301,7 @@ def test_aggregation(ids, modes, req_n, req_m, disjunction, final_ids, final_ns,
if mode is not defined, requests must not be merged, if mode is not defined, requests must not be merged,
if requests are in a synchronization vector, they should not be merged if requests are in a synchronization vector, they should not be merged
""" """
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_LIBRARY_NAME, EXTRA_CONFIGS)
requests = [] requests = []
for request_id, mode, req_n, req_m in zip(ids, modes, req_n, req_m): for request_id, mode, req_n, req_m in zip(ids, modes, req_n, req_m):
params = request_set params = request_set

View File

@@ -28,8 +28,11 @@ from gnpy.topology.spectrum_assignment import build_oms_list
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
NETWORK_FILENAME = TEST_DIR / 'data/testTopology_expected.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILENAME = DATA_DIR / 'testTopology_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
@pytest.mark.parametrize('degree, equalization_type, target, expected_pch_out_dbm, expected_si', @pytest.mark.parametrize('degree, equalization_type, target, expected_pch_out_dbm, expected_si',
@@ -151,7 +154,7 @@ def test_merge_equalization():
"type": "Roadm"}], "type": "Roadm"}],
"connections": [] "connections": []
} }
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
roadm = [n for n in network.nodes()][0] roadm = [n for n in network.nodes()][0]
assert roadm.target_pch_out_dbm == -20 assert roadm.target_pch_out_dbm == -20
@@ -352,7 +355,7 @@ def create_voyager_req(equipment, source, dest, bidir, nodes_list, loose_list, m
def test_initial_spectrum(mode, slot_width, power_dbm): def test_initial_spectrum(mode, slot_width, power_dbm):
"""checks that propagation using the user defined spectrum identical to SI, gives same result as SI""" """checks that propagation using the user defined spectrum identical to SI, gives same result as SI"""
# first propagate without any req.initial_spectrum attribute # first propagate without any req.initial_spectrum attribute
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'], req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
mode, slot_width, power_dbm) mode, slot_width, power_dbm)
network = net_setup(equipment) network = net_setup(equipment)
@@ -389,7 +392,7 @@ def test_initial_spectrum_not_identical():
"""checks that user defined spectrum overrides spectrum defined in SI """checks that user defined spectrum overrides spectrum defined in SI
""" """
# first propagate without any req.initial_spectrum attribute # first propagate without any req.initial_spectrum attribute
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'], req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
'mode 1', 50e9, 0) 'mode 1', 50e9, 0)
network = net_setup(equipment) network = net_setup(equipment)
@@ -424,7 +427,7 @@ def test_target_psd_or_psw(power_dbm, equalization, target_value):
"""checks that if target_out_mWperSlotWidth or target_psd_out_mWperGHz is defined, it is used as equalization """checks that if target_out_mWperSlotWidth or target_psd_out_mWperGHz is defined, it is used as equalization
and it gives same result if computed target is the same and it gives same result if computed target is the same
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = net_setup(equipment) network = net_setup(equipment)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'], req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
'mode 1', 50e9, power_dbm) 'mode 1', 50e9, power_dbm)
@@ -452,7 +455,7 @@ def test_target_psd_or_psw(power_dbm, equalization, target_value):
def ref_network(): def ref_network():
"""Create a network instance with a instance of propagated path""" """Create a network instance with a instance of propagated path"""
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = net_setup(equipment) network = net_setup(equipment)
req0 = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'], req0 = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
'mode 1', 50e9, 0) 'mode 1', 50e9, 0)
@@ -467,7 +470,7 @@ def test_target_psd_out_mwperghz_deltap(deltap):
Power over 1.18dBm saturate amp with this test: TODO add a test on this saturation Power over 1.18dBm saturate amp with this test: TODO add a test on this saturation
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = net_setup(equipment, deltap) network = net_setup(equipment, deltap)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'], req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
'mode 1', 50e9, deltap) 'mode 1', 50e9, deltap)
@@ -510,7 +513,7 @@ def test_equalization(case, deltap, target, mode, slot_width, equalization):
- per degree : target_pch_out_db / target_psd_out_mWperGHz - per degree : target_pch_out_db / target_psd_out_mWperGHz
for these cases with and without power from user for these cases with and without power from user
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
setattr(equipment['Roadm']['default'], 'target_pch_out_db', target) setattr(equipment['Roadm']['default'], 'target_pch_out_db', target)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Rennes_STA', False, req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Rennes_STA', False,
['east edfa in Brest_KLA to Quimper', 'roadm Lannion_CAS', 'trx Rennes_STA'], ['east edfa in Brest_KLA to Quimper', 'roadm Lannion_CAS', 'trx Rennes_STA'],
@@ -567,7 +570,7 @@ def test_equalization(case, deltap, target, mode, slot_width, equalization):
def test_power_option(req_power): def test_power_option(req_power):
"""check that --po option adds correctly power with spectral information """check that --po option adds correctly power with spectral information
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
setattr(equipment['Roadm']['default'], 'target_pch_out_db', None) setattr(equipment['Roadm']['default'], 'target_pch_out_db', None)
setattr(equipment['Roadm']['default'], 'target_psd_out_mWperGHz', power_dbm_to_psd_mw_ghz(-20, 32e9)) setattr(equipment['Roadm']['default'], 'target_psd_out_mWperGHz', power_dbm_to_psd_mw_ghz(-20, 32e9))
network = net_setup(equipment) network = net_setup(equipment)
@@ -718,7 +721,7 @@ def test_power_offset_trx_equalization_psw(slot_width, value):
"""Check that the equalization with the offset is giving the same result as with reference slot_width """Check that the equalization with the offset is giving the same result as with reference slot_width
Check that larger slot width but no offset takes larger slot width for equalization Check that larger slot width but no offset takes larger slot width for equalization
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
trx = transceiver(slot_width, value) trx = transceiver(slot_width, value)
equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx) equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx)
setattr(equipment['Roadm']['default'], 'target_pch_out_db', None) setattr(equipment['Roadm']['default'], 'target_pch_out_db', None)
@@ -751,7 +754,7 @@ def test_power_offset_trx_equalization_psw(slot_width, value):
def test_power_offset_trx_equalization_p(slot_width, value): def test_power_offset_trx_equalization_p(slot_width, value):
"""Check that the constant power equalization with the offset is applied """Check that the constant power equalization with the offset is applied
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
trx = transceiver(slot_width, value) trx = transceiver(slot_width, value)
equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx) equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx)
setattr(equipment['Roadm']['default'], 'target_pch_out_db', -20) setattr(equipment['Roadm']['default'], 'target_pch_out_db', -20)
@@ -777,7 +780,7 @@ def test_power_offset_automatic_mode_selection(slot_width, value, equalization,
"""Check that the same result is obtained if the mode is user defined or if it is """Check that the same result is obtained if the mode is user defined or if it is
automatically selected automatically selected
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
trx = transceiver(slot_width, value) trx = transceiver(slot_width, value)
equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx) equipment['Transceiver'][trx['type_variety']] = Transceiver(**trx)
setattr(equipment['Roadm']['default'], 'target_pch_out_db', None) setattr(equipment['Roadm']['default'], 'target_pch_out_db', None)
@@ -859,7 +862,7 @@ def test_tx_power(tx_power_dbm):
for el in json_data['elements']: for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS': if el['uid'] == 'roadm Lannion_CAS':
el['type_variety'] = 'example_detailed_impairments' el['type_variety'] = 'example_detailed_impairments'
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
default_spectrum = equipment['SI']['default'] default_spectrum = equipment['SI']['default']
p_db = default_spectrum.power_dbm p_db = default_spectrum.power_dbm

View File

@@ -22,8 +22,11 @@ from gnpy.topology.request import PathRequest, compute_constrained_path, propaga
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
NETWORK_FILENAME = TEST_DIR / 'data/perdegreemeshTopologyExampleV2_auto_design_expected.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILENAME = DATA_DIR / 'perdegreemeshTopologyExampleV2_auto_design_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
def net_setup(equipment): def net_setup(equipment):
@@ -71,7 +74,7 @@ def test_gain_mode(req_power, power_dbm):
in gain mode, whatever the value of equipment power_dbm or request power, the network is unchanged in gain mode, whatever the value of equipment power_dbm or request power, the network is unchanged
and the propagation remains the same as for power mode and 0dBm and the propagation remains the same as for power mode and 0dBm
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = net_setup(equipment) network = net_setup(equipment)
req = create_rq(equipment, 'trx Brest_KLA', 'trx Rennes_STA', False, req = create_rq(equipment, 'trx Brest_KLA', 'trx Rennes_STA', False,
['Edfa0_roadm Brest_KLA', 'roadm Lannion_CAS', 'trx Rennes_STA'], ['Edfa0_roadm Brest_KLA', 'roadm Lannion_CAS', 'trx Rennes_STA'],

View File

@@ -18,6 +18,8 @@ TEST_DIR = Path(__file__).parent
EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json' EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json'
MULTIBAND_EQPT_FILENAME = TEST_DIR / 'data/eqpt_config_multiband.json' MULTIBAND_EQPT_FILENAME = TEST_DIR / 'data/eqpt_config_multiband.json'
DATA_DIR = TEST_DIR / 'data' DATA_DIR = TEST_DIR / 'data'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
def test_jsonthing(caplog): def test_jsonthing(caplog):
@@ -97,7 +99,7 @@ def test_wrong_equipment(caplog, error, equipment, json_data, expected_msg):
def test_wrong_xls_service(xls_service_filename, xls_topo_filename, expected_msg): def test_wrong_xls_service(xls_service_filename, xls_topo_filename, expected_msg):
""" """
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(DATA_DIR / xls_topo_filename, equipment) network = load_network(DATA_DIR / xls_topo_filename, equipment)
with pytest.raises(ServiceError, match=expected_msg): with pytest.raises(ServiceError, match=expected_msg):
_ = load_requests(DATA_DIR / xls_service_filename, equipment, False, network, DATA_DIR / xls_topo_filename) _ = load_requests(DATA_DIR / xls_service_filename, equipment, False, network, DATA_DIR / xls_topo_filename)
@@ -293,7 +295,7 @@ def test_json_request(error, json_data, expected_msg):
""" """
Check that a missing key is correctly raisong the logger Check that a missing key is correctly raisong the logger
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
with pytest.raises(error, match=re.escape(expected_msg)): with pytest.raises(error, match=re.escape(expected_msg)):
_ = requests_from_json(json_data, equipment) _ = requests_from_json(json_data, equipment)
@@ -387,7 +389,7 @@ def test_json_network(error, json_data, expected_msg):
""" """
Check that a missing key is correctly raisong the logger Check that a missing key is correctly raisong the logger
""" """
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
with pytest.raises(error, match=re.escape(expected_msg)): with pytest.raises(error, match=re.escape(expected_msg)):
_ = network_from_json(json_data, equipment) _ = network_from_json(json_data, equipment)
@@ -568,7 +570,7 @@ def wrong_configs():
@pytest.mark.parametrize('el_config, error_type', wrong_configs()) @pytest.mark.parametrize('el_config, error_type', wrong_configs())
def test_wrong_multiband(el_config, error_type): def test_wrong_multiband(el_config, error_type):
equipment = load_equipment(MULTIBAND_EQPT_FILENAME) equipment = load_equipment(MULTIBAND_EQPT_FILENAME, EXTRA_CONFIGS)
fused_config = { fused_config = {
"uid": "[83/WR-2-4-SIG=>930/WRT-1-2-SIG]-Tl/9300", "uid": "[83/WR-2-4-SIG=>930/WRT-1-2-SIG]-Tl/9300",
"type": "Fused", "type": "Fused",

View File

@@ -18,9 +18,12 @@ from gnpy.core.parameters import SimParams, EdfaParams, MultiBandParams
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
EQPT_MULTBAND_FILENAME = TEST_DIR / 'data/eqpt_config_multiband.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILENAME = TEST_DIR / 'data/bugfixiteratortopo.json' EQPT_MULTBAND_FILENAME = DATA_DIR / 'eqpt_config_multiband.json'
NETWORK_FILENAME = DATA_DIR / 'bugfixiteratortopo.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
@pytest.mark.parametrize("node, attenuation", [ @pytest.mark.parametrize("node, attenuation", [
@@ -52,7 +55,7 @@ NETWORK_FILENAME = TEST_DIR / 'data/bugfixiteratortopo.json'
['Site_B', 0], ['Site_B', 0],
]) ])
def test_span_loss(node, attenuation): def test_span_loss(node, attenuation):
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(NETWORK_FILENAME, equipment) network = load_network(NETWORK_FILENAME, equipment)
for x in network.nodes(): for x in network.nodes():
if x.uid == node: if x.uid == node:
@@ -64,7 +67,7 @@ def test_span_loss(node, attenuation):
@pytest.mark.parametrize("node", ['fused4']) @pytest.mark.parametrize("node", ['fused4'])
def test_span_loss_unconnected(node): def test_span_loss_unconnected(node):
'''Fused node that has no next and no previous nodes should be detected''' '''Fused node that has no next and no previous nodes should be detected'''
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(NETWORK_FILENAME, equipment) network = load_network(NETWORK_FILENAME, equipment)
x = next(x for x in network.nodes() if x.uid == node) x = next(x for x in network.nodes() if x.uid == node)
with pytest.raises(NetworkTopologyError): with pytest.raises(NetworkTopologyError):
@@ -164,7 +167,7 @@ def test_eol(typ, expected_loss):
} }
] ]
} }
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
equipment['Span']['default'].EOL = 1 equipment['Span']['default'].EOL = 1
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
@@ -230,7 +233,7 @@ def test_design_non_amplified_link(elem1, elem2, expected_gain, expected_delta_p
} }
] ]
} }
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
equipment['Span']['default'].power_mode = power_mode equipment['Span']['default'].power_mode = power_mode
equipment['SI']['default'].power_dbm = p_db equipment['SI']['default'].power_dbm = p_db
equipment['SI']['default'].tx_power_dbm = p_db equipment['SI']['default'].tx_power_dbm = p_db
@@ -441,7 +444,7 @@ def test_design_band(case, site_type, amplifier_type, expected_design_bands, exp
EOL is then added on the first fiber only. EOL is then added on the first fiber only.
""" """
json_data = network_base(case, site_type, amplifier_type=amplifier_type) json_data = network_base(case, site_type, amplifier_type=amplifier_type)
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min, p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min,
@@ -462,7 +465,7 @@ def test_design_band(case, site_type, amplifier_type, expected_design_bands, exp
def test_select_edfa(caplog, raman_allowed, gain_target, power_target, target_extended_gain, warning, expected_selection): def test_select_edfa(caplog, raman_allowed, gain_target, power_target, target_extended_gain, warning, expected_selection):
""" """
""" """
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
edfa_eqpt = {n: a for n, a in equipment['Edfa'].items() if a.type_def != 'multi_band'} edfa_eqpt = {n: a for n, a in equipment['Edfa'].items() if a.type_def != 'multi_band'}
selection = select_edfa(raman_allowed, gain_target, power_target, edfa_eqpt, "toto", target_extended_gain, verbose=True) selection = select_edfa(raman_allowed, gain_target, power_target, edfa_eqpt, "toto", target_extended_gain, verbose=True)
assert selection == expected_selection assert selection == expected_selection
@@ -499,7 +502,7 @@ def test_select_edfa(caplog, raman_allowed, gain_target, power_target, target_ex
def test_get_node_restrictions(cls, defaultparams, variety_list, booster_list, band, expected_restrictions): def test_get_node_restrictions(cls, defaultparams, variety_list, booster_list, band, expected_restrictions):
"""Check that all combinations of restrictions are correctly captured """Check that all combinations of restrictions are correctly captured
""" """
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
edfa_config = {"uid": "Edfa1"} edfa_config = {"uid": "Edfa1"}
if cls == Multiband_amplifier: if cls == Multiband_amplifier:
edfa_config['amplifiers'] = {} edfa_config['amplifiers'] = {}
@@ -570,7 +573,7 @@ def test_multiband(case, site_type, band, expected_gain, expected_tilt, expected
EOL is then added on the first fiber only. EOL is then added on the first fiber only.
""" """
json_data = network_base(case, site_type) json_data = network_base(case, site_type)
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min, p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min,
@@ -600,12 +603,12 @@ def test_tilt_fused():
SimParams.set_params(load_json(TEST_DIR / 'data' / 'sim_params.json')) SimParams.set_params(load_json(TEST_DIR / 'data' / 'sim_params.json'))
input_powers = {'CBAND': 0.001, 'LBAND': 0.001} input_powers = {'CBAND': 0.001, 'LBAND': 0.001}
json_data = network_base("design", "Multiband_amplifier", length=100) json_data = network_base("design", "Multiband_amplifier", length=100)
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
node = next(n for n in network.nodes() if n.uid == 'fiber (SITE1 → ILA1)') node = next(n for n in network.nodes() if n.uid == 'fiber (SITE1 → ILA1)')
tilt_db, tilt_target = estimate_srs_power_deviation(network, node, equipment, design_bands, input_powers) tilt_db, tilt_target = estimate_srs_power_deviation(network, node, equipment, design_bands, input_powers)
json_data = network_base("design", "Fused", length=50) json_data = network_base("design", "Fused", length=50)
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
node = next(n for n in network.nodes() if n.uid == 'fiber (ILA1 → ILA2)') node = next(n for n in network.nodes() if n.uid == 'fiber (ILA1 → ILA2)')
fused_tilt_db, fused_tilt_target = \ fused_tilt_db, fused_tilt_target = \
@@ -734,7 +737,7 @@ def test_insert_amp(site_type, expected_type, bands, expected_bands):
EOL is then added on the first fiber only. EOL is then added on the first fiber only.
""" """
json_data = network_wo_booster(site_type, bands) json_data = network_wo_booster(site_type, bands)
equipment = load_equipment(EQPT_MULTBAND_FILENAME) equipment = load_equipment(EQPT_MULTBAND_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min, p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min,

View File

@@ -36,8 +36,10 @@ from gnpy.tools.service_sheet import read_service_sheet, correct_xls_route_list,
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
DATA_DIR = TEST_DIR / 'data' DATA_DIR = TEST_DIR / 'data'
eqpt_filename = DATA_DIR / 'eqpt_config.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
equipment = load_equipment(eqpt_filename) EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
@pytest.mark.parametrize('xls_input,expected_json_output', { @pytest.mark.parametrize('xls_input,expected_json_output', {
@@ -69,7 +71,7 @@ def test_excel_json_generation(tmpdir, xls_input, expected_json_output):
}.items()) }.items())
def test_auto_design_generation_fromxlsgainmode(tmpdir, xls_input, expected_json_output): def test_auto_design_generation_fromxlsgainmode(tmpdir, xls_input, expected_json_output):
"""tests generation of topology json and that the build network gives correct results in gain mode""" """tests generation of topology json and that the build network gives correct results in gain mode"""
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(xls_input, equipment) network = load_network(xls_input, equipment)
add_missing_elements_in_network(network, equipment) add_missing_elements_in_network(network, equipment)
# in order to test the Eqpt sheet and load gain target, # in order to test the Eqpt sheet and load gain target,
@@ -100,7 +102,7 @@ def test_auto_design_generation_fromxlsgainmode(tmpdir, xls_input, expected_json
}.items()) }.items())
def test_auto_design_generation_fromjson(tmpdir, json_input, power_mode): def test_auto_design_generation_fromjson(tmpdir, json_input, power_mode):
"""test that autodesign creates same file as an input file already autodesigned""" """test that autodesign creates same file as an input file already autodesigned"""
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(json_input, equipment) network = load_network(json_input, equipment)
# in order to test the Eqpt sheet and load gain target, # in order to test the Eqpt sheet and load gain target,
# change the power-mode to False (to be in gain mode) # change the power-mode to False (to be in gain mode)
@@ -127,7 +129,7 @@ def test_auto_design_generation_fromjson(tmpdir, json_input, power_mode):
}.items()) }.items())
def test_excel_service_json_generation(xls_input, expected_json_output): def test_excel_service_json_generation(xls_input, expected_json_output):
"""test services creation""" """test services creation"""
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(DATA_DIR / 'testTopology.xls', equipment) network = load_network(DATA_DIR / 'testTopology.xls', equipment)
# Build the network once using the default power defined in SI in eqpt config # Build the network once using the default power defined in SI in eqpt config
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
@@ -148,7 +150,7 @@ def test_excel_service_json_generation(xls_input, expected_json_output):
def test_csv_response_generation(tmpdir, json_input): def test_csv_response_generation(tmpdir, json_input):
"""tests if generated csv is consistant with expected generation same columns (order not important)""" """tests if generated csv is consistant with expected generation same columns (order not important)"""
json_data = load_json(json_input) json_data = load_json(json_input)
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
csv_filename = Path(tmpdir / json_input.name).with_suffix('.csv') csv_filename = Path(tmpdir / json_input.name).with_suffix('.csv')
with open(csv_filename, 'w', encoding='utf-8') as fcsv: with open(csv_filename, 'w', encoding='utf-8') as fcsv:
jsontocsv(json_data, equipment, fcsv) jsontocsv(json_data, equipment, fcsv)
@@ -213,7 +215,7 @@ def test_csv_response_generation(tmpdir, json_input):
def test_json_response_generation(xls_input, expected_response_file): def test_json_response_generation(xls_input, expected_response_file):
"""tests if json response is correctly generated for all combinations of requests""" """tests if json response is correctly generated for all combinations of requests"""
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(xls_input, equipment) network = load_network(xls_input, equipment)
p_db = equipment['SI']['default'].power_dbm p_db = equipment['SI']['default'].power_dbm
@@ -630,7 +632,7 @@ def test_roadm_type_variety(type_variety, target_pch_out_db, correct_variety):
# Do not add type variety in json_data to test that it creates a 'default' type_variety # Do not add type variety in json_data to test that it creates a 'default' type_variety
expected_roadm['type_variety'] = 'default' expected_roadm['type_variety'] = 'default'
expected_roadm['params']['target_pch_out_db'] = target_pch_out_db expected_roadm['params']['target_pch_out_db'] = target_pch_out_db
equipment = load_equipment(eqpt_filename) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
if correct_variety: if correct_variety:
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
roadm = [n for n in network.nodes()][0] roadm = [n for n in network.nodes()][0]

View File

@@ -23,7 +23,9 @@ DATA_DIR = TEST_DIR / 'data'
EQPT_FILENAME = DATA_DIR / 'eqpt_config.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json' NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json'
SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json' SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json'
equipment = load_equipment(EQPT_FILENAME) EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
@pytest.fixture() @pytest.fixture()

View File

@@ -18,8 +18,12 @@ from gnpy.core.network import build_network
from gnpy.tools.json_io import load_network, load_equipment, network_from_json from gnpy.tools.json_io import load_network, load_equipment, network_from_json
network_file_name = Path(__file__).parent.parent / 'tests/LinkforTest.json' TEST_DIR = Path(__file__).parent
eqpt_library_name = Path(__file__).parent.parent / 'tests/data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
network_file_name = DATA_DIR / 'LinkforTest.json'
eqpt_library_name = DATA_DIR / 'eqpt_config.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
@pytest.fixture(params=[(96, 0.05e12), (60, 0.075e12), (45, 0.1e12), (2, 0.1e12)], @pytest.fixture(params=[(96, 0.05e12), (60, 0.075e12), (45, 0.1e12), (2, 0.1e12)],
@@ -31,7 +35,7 @@ def nch_and_spacing(request):
def propagation(input_power, con_in, con_out, dest): def propagation(input_power, con_in, con_out, dest):
equipment = load_equipment(eqpt_library_name) equipment = load_equipment(eqpt_library_name, EXTRA_CONFIGS)
network = load_network(network_file_name, equipment) network = load_network(network_file_name, equipment)
# parametrize the network elements with the con losses and adapt gain # parametrize the network elements with the con losses and adapt gain
@@ -178,7 +182,7 @@ def test_json_element(error, json_data, expected_msg):
""" """
Check that a missing key is correctly raisong the logger Check that a missing key is correctly raisong the logger
""" """
equipment = load_equipment(eqpt_library_name) equipment = load_equipment(eqpt_library_name, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
elem = next(e for e in network.nodes() if e.uid == 'Elem') elem = next(e for e in network.nodes() if e.uid == 'Elem')
si = create_input_spectral_information(f_min=191.3e12, f_max=196.1e12, roll_off=0.15, si = create_input_spectral_information(f_min=191.3e12, f_max=196.1e12, roll_off=0.15,

View File

@@ -28,8 +28,11 @@ from gnpy.core.exceptions import ConfigurationError, NetworkTopologyError
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
EQPT_LIBRARY_NAME = TEST_DIR / 'data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
NETWORK_FILE_NAME = TEST_DIR / 'data/testTopology_expected.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILE_NAME = DATA_DIR / 'testTopology_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
# adding tests to check the roadm restrictions # adding tests to check the roadm restrictions
# mark node_uid amps as fused for testing purpose # mark node_uid amps as fused for testing purpose
@@ -39,7 +42,7 @@ def test_no_amp_feature(node_uid):
test_parser covers partly this behaviour. This test should guaranty that the test_parser covers partly this behaviour. This test should guaranty that the
feature is preserved even if convert is changed feature is preserved even if convert is changed
""" """
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
json_network = load_json(NETWORK_FILE_NAME) json_network = load_json(NETWORK_FILE_NAME)
for elem in json_network['elements']: for elem in json_network['elements']:
@@ -82,7 +85,7 @@ def test_no_amp_feature(node_uid):
@pytest.fixture() @pytest.fixture()
def equipment(): def equipment():
"""init transceiver class to access snr and osnr calculations""" """init transceiver class to access snr and osnr calculations"""
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
# define some booster and preamps # define some booster and preamps
restrictions_list = [ restrictions_list = [
{ {
@@ -131,7 +134,7 @@ def equipment():
}] }]
# add them to the library # add them to the library
for entry in restrictions_list: for entry in restrictions_list:
equipment['Edfa'][entry['type_variety']] = Amp.from_json(EQPT_LIBRARY_NAME, **entry) equipment['Edfa'][entry['type_variety']] = Amp.from_json(EXTRA_CONFIGS, **entry)
return equipment return equipment
@@ -223,7 +226,7 @@ def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm, roa
for the test where the prev_node in ROADM B is either an amplifier or a fused, so that the target for the test where the prev_node in ROADM B is either an amplifier or a fused, so that the target
power can not be met in this last case. power can not be met in this last case.
""" """
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
equipment['SI']['default'].power_dbm = power_dbm equipment['SI']['default'].power_dbm = power_dbm
json_network = load_json(TEST_DIR / 'data/twohops_roadm_power_test.json') json_network = load_json(TEST_DIR / 'data/twohops_roadm_power_test.json')
prev_node = next(n for n in json_network['elements'] if n['uid'] == 'west edfa in node B to ila2') prev_node = next(n for n in json_network['elements'] if n['uid'] == 'west edfa in node B to ila2')
@@ -448,7 +451,7 @@ def test_compare_design_propagation_settings(power_dbm, req_power, amp_with_delt
This test also checks all the possible combinations and expected before/after propagation This test also checks all the possible combinations and expected before/after propagation
gain differences. It also checks delta_p applied due to saturation during design. gain differences. It also checks delta_p applied due to saturation during design.
""" """
eqpt = load_equipment(EQPT_LIBRARY_NAME) eqpt = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
eqpt['SI']['default'].power_dbm = power_dbm eqpt['SI']['default'].power_dbm = power_dbm
json_network = load_json(NETWORK_FILE_NAME) json_network = load_json(NETWORK_FILE_NAME)
for element in json_network['elements']: for element in json_network['elements']:
@@ -550,14 +553,14 @@ def test_wrong_restrictions(restrictions, fail):
"""Check that sanity_check correctly raises an error when restriction is incorrect and that library """Check that sanity_check correctly raises an error when restriction is incorrect and that library
correctly includes restrictions. correctly includes restrictions.
""" """
json_data = load_json(EQPT_LIBRARY_NAME) json_data = load_json(EQPT_FILENAME)
# define wrong restriction # define wrong restriction
json_data['Roadm'][0]['restrictions'] = restrictions json_data['Roadm'][0]['restrictions'] = restrictions
if fail: if fail:
with pytest.raises(ConfigurationError): with pytest.raises(ConfigurationError):
_ = _equipment_from_json(json_data, EQPT_LIBRARY_NAME) _ = _equipment_from_json(json_data, EXTRA_CONFIGS)
else: else:
equipment = _equipment_from_json(json_data, EQPT_LIBRARY_NAME) equipment = _equipment_from_json(json_data, EXTRA_CONFIGS)
assert equipment['Roadm']['example_test'].restrictions == restrictions assert equipment['Roadm']['example_test'].restrictions == restrictions
@@ -574,7 +577,7 @@ def test_roadm_impairments(roadm, from_degree, to_degree, expected_impairment_id
for el in json_data['elements']: for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS': if el['uid'] == 'roadm Lannion_CAS':
el['type_variety'] = 'example_detailed_impairments' el['type_variety'] = 'example_detailed_impairments'
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment) network = network_from_json(json_data, equipment)
build_network(network, equipment, 0.0, 20.0) build_network(network, equipment, 0.0, 20.0)
roadm = next(n for n in network.nodes() if n.uid == roadm) roadm = next(n for n in network.nodes() if n.uid == roadm)
@@ -590,7 +593,7 @@ def test_roadm_impairments(roadm, from_degree, to_degree, expected_impairment_id
def test_roadm_per_degree_impairments(type_variety, from_degree, to_degree, impairment_id, expected_type): def test_roadm_per_degree_impairments(type_variety, from_degree, to_degree, impairment_id, expected_type):
"""Check that impairment type is correct also if per degree impairment is defined """Check that impairment type is correct also if per degree impairment is defined
""" """
json_data = load_json(EQPT_LIBRARY_NAME) json_data = load_json(EQPT_FILENAME)
assert 'type_variety' not in json_data['Roadm'][2] assert 'type_variety' not in json_data['Roadm'][2]
json_data['Roadm'][2]['roadm-path-impairments'] = [ json_data['Roadm'][2]['roadm-path-impairments'] = [
{ {
@@ -614,7 +617,7 @@ def test_roadm_per_degree_impairments(type_variety, from_degree, to_degree, impa
"roadm-noise-figure": 23 "roadm-noise-figure": 23
}] }]
}] }]
equipment = _equipment_from_json(json_data, EQPT_LIBRARY_NAME) equipment = _equipment_from_json(json_data, EXTRA_CONFIGS)
assert equipment['Roadm']['default'].type_variety == 'default' assert equipment['Roadm']['default'].type_variety == 'default'
json_data = load_json(NETWORK_FILE_NAME) json_data = load_json(NETWORK_FILE_NAME)
@@ -647,7 +650,7 @@ def test_roadm_per_degree_impairments(type_variety, from_degree, to_degree, impa
def test_wrong_roadm_per_degree_impairments(from_degree, to_degree, impairment_id, error, message): def test_wrong_roadm_per_degree_impairments(from_degree, to_degree, impairment_id, error, message):
"""Check that wrong per degree definitions are correctly catched """Check that wrong per degree definitions are correctly catched
""" """
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
json_data = load_json(NETWORK_FILE_NAME) json_data = load_json(NETWORK_FILE_NAME)
for el in json_data['elements']: for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS': if el['uid'] == 'roadm Lannion_CAS':
@@ -680,7 +683,7 @@ def test_impairment_initialization(path_type, type_variety, expected_pmd, expect
- use roadm detailed impairment for the corresponding path_type if roadm type_variety has detailed impairments - use roadm detailed impairment for the corresponding path_type if roadm type_variety has detailed impairments
- use roadm per degree impairment if they are defined - use roadm per degree impairment if they are defined
""" """
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
extra_params = equipment['Roadm'][type_variety].__dict__ extra_params = equipment['Roadm'][type_variety].__dict__
roadm_config = { roadm_config = {
"uid": "roadm Lannion_CAS", "uid": "roadm Lannion_CAS",

View File

@@ -30,6 +30,8 @@ DATA_DIR = TEST_DIR / 'data'
EQPT_FILENAME = DATA_DIR / 'eqpt_config.json' EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json' NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json'
SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json' SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
grid = 0.00625e12 grid = 0.00625e12
slot = 0.0125e12 slot = 0.0125e12
@@ -40,7 +42,7 @@ cband_freq_max = 196.1e12
@pytest.fixture() @pytest.fixture()
def equipment(): def equipment():
equipment = load_equipment(EQPT_FILENAME) equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
return equipment return equipment

View File

@@ -17,8 +17,11 @@ from gnpy.tools.json_io import load_equipment, load_json, _equipment_from_json
TEST_DIR = Path(__file__).parent TEST_DIR = Path(__file__).parent
EQPT_LIBRARY_NAME = TEST_DIR / 'data/eqpt_config.json' DATA_DIR = TEST_DIR / 'data'
NETWORK_FILE_NAME = TEST_DIR / 'data/testTopology_expected.json' EQPT_LIBRARY_NAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILE_NAME = DATA_DIR / 'testTopology_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": DATA_DIR / "std_medium_gain_advanced_config.json",
"Juniper-BoosterHG.json": DATA_DIR / "Juniper-BoosterHG.json"}
@pytest.mark.parametrize('trx_type, trx_mode, error_message, no_error, expected_result', @pytest.mark.parametrize('trx_type, trx_mode, error_message, no_error, expected_result',
@@ -91,7 +94,7 @@ def test_trx_mode_params(trx_type, trx_mode, error_message, no_error, expected_r
'penalties': None, 'penalties': None,
'cost': None 'cost': None
} }
equipment = load_equipment(EQPT_LIBRARY_NAME) equipment = load_equipment(EQPT_LIBRARY_NAME, EXTRA_CONFIGS)
if no_error: if no_error:
trx_params = trx_mode_params(equipment, trx_type, trx_mode, error_message) trx_params = trx_mode_params(equipment, trx_type, trx_mode, error_message)
print(trx_params) print(trx_params)
@@ -150,7 +153,7 @@ def test_wrong_baudrate_spacing(baudrate, spacing, error_message):
'equalization_offset_db': 0}] 'equalization_offset_db': 0}]
} }
json_data['Transceiver'].append(wrong_transceiver) json_data['Transceiver'].append(wrong_transceiver)
equipment = _equipment_from_json(json_data, EQPT_LIBRARY_NAME) equipment = _equipment_from_json(json_data, EXTRA_CONFIGS)
with pytest.raises(EquipmentConfigError, match=error_message): with pytest.raises(EquipmentConfigError, match=error_message):
_ = trx_mode_params(equipment, 'vendorB_trx-type1', 'wrong mode', error_message=False) _ = trx_mode_params(equipment, 'vendorB_trx-type1', 'wrong mode', error_message=False)