#!/usr/bin/env python3 # -*- coding: utf-8 -*- # @Author: Esther Le Rouzic # @Date: 2019-05-22 """ @author: esther.lerouzic checks that fused placed in amp type is correctly converted to a fused element instead of an edfa and that no additional amp is added. checks that restrictions in roadms are correctly applied during autodesign """ from pathlib import Path import pytest from numpy.testing import assert_allclose from gnpy.core.utils import lin2db, automatic_nch from gnpy.core.elements import Fused, Roadm, Edfa from gnpy.core.network import build_network from gnpy.tools.json_io import network_from_json, load_equipment, load_json, Amp from gnpy.core.equipment import trx_mode_params from gnpy.topology.request import PathRequest, compute_constrained_path, ref_carrier from gnpy.core.info import create_input_spectral_information from gnpy.core.utils import db2lin TEST_DIR = Path(__file__).parent EQPT_LIBRARY_NAME = TEST_DIR / 'data/eqpt_config.json' NETWORK_FILE_NAME = TEST_DIR / 'data/testTopology_expected.json' # adding tests to check the roadm restrictions # mark node_uid amps as fused for testing purpose @pytest.mark.parametrize("node_uid", ['east edfa in Lannion_CAS to Stbrieuc']) def test_no_amp_feature(node_uid): """Check that booster is not placed on a roadm if fused is specified test_parser covers partly this behaviour. This test should guaranty that the feature is preserved even if convert is changed """ equipment = load_equipment(EQPT_LIBRARY_NAME) json_network = load_json(NETWORK_FILE_NAME) for elem in json_network['elements']: if elem['uid'] == node_uid: # replace edfa node by a fused node in the topology elem['type'] = 'Fused' elem.pop('type_variety') elem.pop('operational') elem['params'] = {'loss': 0} next_node_uid = next(conn['to_node'] for conn in json_network['connections'] if conn['from_node'] == node_uid) previous_node_uid = next(conn['from_node'] for conn in json_network['connections'] if conn['to_node'] == node_uid) network = network_from_json(json_network, equipment) # 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 # spacing, f_min and f_max p_db = equipment['SI']['default'].power_dbm p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min, equipment['SI']['default'].f_max, equipment['SI']['default'].spacing)) build_network(network, equipment, p_db, p_total_db) node = next(nd for nd in network.nodes() if nd.uid == node_uid) next_node = next(network.successors(node)) previous_node = next(network.predecessors(node)) if not isinstance(node, Fused): raise AssertionError() if not node.params.loss == 0.0: raise AssertionError() if not next_node_uid == next_node.uid: raise AssertionError() if not previous_node_uid == previous_node.uid: raise AssertionError() @pytest.fixture() def equipment(): """init transceiver class to access snr and osnr calculations""" equipment = load_equipment(EQPT_LIBRARY_NAME) # define some booster and preamps restrictions_list = [ { 'type_variety': 'booster_medium_gain', 'type_def': 'variable_gain', 'gain_flatmax': 25, 'gain_min': 15, 'p_max': 21, 'nf_min': 5.8, 'nf_max': 10, 'out_voa_auto': False, 'allowed_for_design': False }, { 'type_variety': 'preamp_medium_gain', 'type_def': 'variable_gain', 'gain_flatmax': 26, 'gain_min': 15, 'p_max': 23, 'nf_min': 6, 'nf_max': 10, 'out_voa_auto': False, 'allowed_for_design': False }, { 'type_variety': 'preamp_high_gain', 'type_def': 'variable_gain', 'gain_flatmax': 35, 'gain_min': 25, 'p_max': 21, 'nf_min': 5.5, 'nf_max': 7, 'out_voa_auto': False, 'allowed_for_design': False }, { 'type_variety': 'preamp_low_gain', 'type_def': 'variable_gain', 'gain_flatmax': 16, 'gain_min': 8, 'p_max': 23, 'nf_min': 6.5, 'nf_max': 11, 'out_voa_auto': False, 'allowed_for_design': False }] # add them to the library for entry in restrictions_list: equipment['Edfa'][entry['type_variety']] = Amp.from_json(EQPT_LIBRARY_NAME, **entry) return equipment @pytest.mark.parametrize("restrictions", [ { 'preamp_variety_list': [], 'booster_variety_list':[] }, { 'preamp_variety_list': [], 'booster_variety_list':['booster_medium_gain'] }, { 'preamp_variety_list': ['preamp_medium_gain', 'preamp_high_gain', 'preamp_low_gain'], 'booster_variety_list':[] }]) def test_restrictions(restrictions, equipment): """test that restriction is correctly applied if provided in eqpt_config and if no Edfa type were provided in the network json """ # add restrictions equipment['Roadm']['default'].restrictions = restrictions # build network json_network = load_json(NETWORK_FILE_NAME) network = network_from_json(json_network, equipment) amp_nodes_nobuild_uid = [nd.uid for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.predecessors(nd)), Roadm)] preamp_nodes_nobuild_uid = [nd.uid for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.successors(nd)), Roadm)] amp_nodes_nobuild = {nd.uid: nd for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.predecessors(nd)), Roadm)} preamp_nodes_nobuild = {nd.uid: nd for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.successors(nd)), Roadm)} # roadm dict with restrictions before build roadms = {nd.uid: nd for nd in network.nodes() if isinstance(nd, Roadm)} # 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 # spacing, f_min and f_max p_db = equipment['SI']['default'].power_dbm p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min, equipment['SI']['default'].f_max, equipment['SI']['default'].spacing)) build_network(network, equipment, p_db, p_total_db) amp_nodes = [nd for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.predecessors(nd)), Roadm) and next(network.predecessors(nd)).restrictions['booster_variety_list']] preamp_nodes = [nd for nd in network.nodes() if isinstance(nd, Edfa) and isinstance(next(network.successors(nd)), Roadm) and next(network.successors(nd)).restrictions['preamp_variety_list']] # check that previously existing amp are not changed for amp in amp_nodes: if amp.uid in amp_nodes_nobuild_uid: print(amp.uid, amp.params.type_variety) if not amp.params.type_variety == amp_nodes_nobuild[amp.uid].params.type_variety: raise AssertionError() for amp in preamp_nodes: if amp.uid in preamp_nodes_nobuild_uid: if not amp.params.type_variety == preamp_nodes_nobuild[amp.uid].params.type_variety: raise AssertionError() # check that restrictions are correctly applied for amp in amp_nodes: if amp.uid not in amp_nodes_nobuild_uid: # and if roadm had no restrictions before build: if restrictions['booster_variety_list'] and \ not roadms[next(network.predecessors(amp)).uid]\ .restrictions['booster_variety_list']: if amp.params.type_variety not in restrictions['booster_variety_list']: raise AssertionError() for amp in preamp_nodes: if amp.uid not in preamp_nodes_nobuild_uid: if restrictions['preamp_variety_list'] and\ not roadms[next(network.successors(amp)).uid].restrictions['preamp_variety_list']: if amp.params.type_variety not in restrictions['preamp_variety_list']: raise AssertionError() @pytest.mark.parametrize('power_dbm', [0, +1, -2]) @pytest.mark.parametrize('prev_node_type, effective_pch_out_db', [('edfa', -20.0), ('fused', -22.0)]) def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm): """Check that egress power of roadm is equal to target power if input power is greater than target power else, that it is equal to input power. Use a simple two hops A-B-C topology 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. """ equipment = load_equipment(EQPT_LIBRARY_NAME) 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') json_network['elements'].remove(prev_node) if prev_node_type == 'edfa': prev_node = {'uid': 'west edfa in node B to ila2', 'type': 'Edfa'} elif prev_node_type == 'fused': prev_node = {'uid': 'west edfa in node B to ila2', 'type': 'Fused'} prev_node['params'] = {'loss': 0} json_network['elements'].append(prev_node) network = network_from_json(json_network, equipment) nb_channel = automatic_nch(equipment['SI']['default'].f_min, equipment['SI']['default'].f_max, equipment['SI']['default'].spacing) p_total_db = power_dbm + lin2db(nb_channel) build_network(network, equipment, power_dbm, p_total_db) params = {'request_id': 0, 'trx_type': '', 'trx_mode': '', 'source': 'trx node A', 'destination': 'trx node C', 'bidir': False, 'nodes_list': ['trx node C'], 'loose_list': ['strict'], 'format': '', 'path_bandwidth': 100e9, 'effective_freq_slot': None, 'nb_channel': nb_channel } trx_params = trx_mode_params(equipment) params.update(trx_params) req = PathRequest(**params) req.power = db2lin(power_dbm - 30) path = compute_constrained_path(network, req) si = create_input_spectral_information( f_min=req.f_min, f_max=req.f_max, roll_off=req.roll_off, baud_rate=req.baud_rate, power=req.power, spacing=req.spacing, tx_osnr=req.tx_osnr, ref_carrier=ref_carrier(equipment)) for i, el in enumerate(path): if isinstance(el, Roadm): power_in_roadm = si.signal + si.ase + si.nli si = el(si, degree=path[i + 1].uid) power_out_roadm = si.signal + si.ase + si.nli if el.uid == 'roadm node B': # if previous was an EDFA, power level at ROADM input is enough for the ROADM to apply its # target power (as specified in equipment ie -20 dBm) # if it is a Fused, the input power to the ROADM is smaller than the target power, and the # ROADM cannot apply this target. In this case, it is assumed that the ROADM has 0 dB loss # so the output power will be the same as the input power, which for this particular case # corresponds to -22dBm + power_dbm # next step (for ROADM modelling) will be to apply a minimum loss for ROADMs ! if prev_node_type == 'edfa': # edfa prev_node sets input power to roadm to a high enough value: # check that target power is correctly set in the ROADM assert_allclose(el.ref_pch_out_dbm, effective_pch_out_db, rtol=1e-3) # Check that egress power of roadm is equal to target power assert_allclose(power_out_roadm, db2lin(effective_pch_out_db - 30), rtol=1e-3) elif prev_node_type == 'fused': # fused prev_node does reamplfy power after fiber propagation, so input power # to roadm is low. # check that target power correctly reports power_dbm from previous propagation assert_allclose(el.ref_pch_out_dbm, effective_pch_out_db + power_dbm, rtol=1e-3) # Check that egress power of roadm is not equalized power out is the same as power in. assert_allclose(power_out_roadm, power_in_roadm, rtol=1e-3) else: si = el(si)