scottk/oopt-gnpy
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/oopt-gnpy.git synced 2025-10-29 01:02:32 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity
Files
master
oopt-gnpy/tests/test_roadm_restrictions.py
EstherLerouzic f2039fbe1c fix: use loaded json instead of Path for extra configs 
In order to be used by API.

Co-authored-by: Renato Ambrosone <renato.ambrosone@polito.it>

Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Change-Id: I12111427c8a90b85b3158cdd95f4ee771cb39316
2025-09-26 11:17:45 +02:00

753 lines
38 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: BSD-3-Clause
# test_roadm_restrictions
# Copyright (C) 2025 Telecom Infra Project and GNPy contributors
# see AUTHORS.rst for a list of contributors
"""
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 numpy import ndarray, mean
from copy import deepcopy
from gnpy.core.utils import lin2db, automatic_nch
from gnpy.core.elements import Fused, Roadm, Edfa, Transceiver, EdfaOperational, EdfaParams, Fiber
from gnpy.core.parameters import FiberParams, RoadmParams, FusedParams
from gnpy.core.network import build_network, design_network
from gnpy.tools.json_io import network_from_json, load_equipment, load_json, Amp, _equipment_from_json
from gnpy.core.equipment import trx_mode_params
from gnpy.topology.request import PathRequest, compute_constrained_path, propagate
from gnpy.core.info import create_input_spectral_information, Carrier
from gnpy.core.utils import db2lin, dbm2watt, merge_amplifier_restrictions
from gnpy.core.exceptions import ConfigurationError, NetworkTopologyError
TEST_DIR = Path(__file__).parent
DATA_DIR = TEST_DIR / 'data'
EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
NETWORK_FILE_NAME = DATA_DIR / 'testTopology_expected.json'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": load_json(DATA_DIR / "std_medium_gain_advanced_config.json")}
def pathrequest(pch_dbm: float, p_tot_dbm: float = None, nb_channels: int = None):
"""create ref channel for defined power settings
"""
params = {
"power": dbm2watt(pch_dbm),
"tx_power": dbm2watt(pch_dbm),
"nb_channel": nb_channels if nb_channels else round(dbm2watt(p_tot_dbm) / dbm2watt(pch_dbm), 0),
'request_id': None,
'trx_type': None,
'trx_mode': None,
'source': None,
'destination': None,
'bidir': False,
'nodes_list': [],
'loose_list': [],
'format': '',
'baud_rate': None,
'bit_rate': None,
'roll_off': None,
'OSNR': None,
'penalties': None,
'path_bandwidth': None,
'effective_freq_slot': None,
'f_min': None,
'f_max': None,
'spacing': None,
'min_spacing': None,
'cost': None,
'equalization_offset_db': None,
'tx_osnr': None
}
return PathRequest(**params)
# 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_FILENAME, EXTRA_CONFIGS)
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
nb_channels = automatic_nch(equipment['SI']['default'].f_min,
equipment['SI']['default'].f_max, equipment['SI']['default'].spacing)
build_network(network, equipment, pathrequest(p_db, nb_channels=nb_channels))
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_FILENAME, EXTRA_CONFIGS)
# 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(EXTRA_CONFIGS, **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
nb_channels = automatic_nch(equipment['SI']['default'].f_min,
equipment['SI']['default'].f_max, equipment['SI']['default'].spacing)
build_network(network, equipment, pathrequest(p_db, nb_channels=nb_channels))
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('roadm_type_variety, roadm_b_maxloss', [('default', 0),
('example_detailed_impairments', 16.5)])
@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, roadm_type_variety, roadm_b_maxloss):
"""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_FILENAME, EXTRA_CONFIGS)
equipment['SI']['default'].power_dbm = power_dbm
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)
roadm_b = next(element for element in json_network['elements'] if element['uid'] == 'roadm node B')
roadm_b['type_variety'] = roadm_type_variety
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)
build_network(network, equipment, pathrequest(power_dbm, nb_channels=nb_channel))
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,
'power': dbm2watt(power_dbm),
'tx_power': dbm2watt(power_dbm)
}
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,
spacing=req.spacing, tx_osnr=req.tx_osnr, tx_power=req.tx_power)
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, from_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. If the ROADM has 0 dB loss the output power will be the same
# as the input power, which for this particular case corresponds to -22dBm + power_dbm.
# If ROADM has a minimum losss, then output power will be -22dBm + power_dbm - ROADM loss. !
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)
if prev_node_type == 'fused':
# fused prev_node does not reamplify 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 - roadm_b_maxloss, rtol=1e-3)
# Check that egress power of roadm is not equalized:
# power out is the same as power in minus the ROADM loss.
assert_allclose(power_out_roadm, power_in_roadm / db2lin(roadm_b_maxloss), rtol=1e-3)
assert effective_pch_out_db + power_dbm ==\
pytest.approx(lin2db(min(power_in_roadm) * 1e3), rel=1e-3)
else:
si = el(si)
def create_per_oms_request(network, eqpt, req_power):
"""Create requests between every adjacent ROADMs + one additional request crossing several ROADMs
"""
nb_channel = automatic_nch(eqpt['SI']['default'].f_min, eqpt['SI']['default'].f_max,
eqpt['SI']['default'].spacing)
params = {
'trx_type': '',
'trx_mode': '',
'bidir': False,
'loose_list': ['strict', 'strict'],
'format': '',
'path_bandwidth': 100e9,
'effective_freq_slot': None,
'nb_channel': nb_channel,
'power': dbm2watt(req_power),
'tx_power': dbm2watt(req_power)
}
trx_params = trx_mode_params(eqpt)
params.update(trx_params)
trxs = [e for e in network if isinstance(e, Transceiver)]
req_list = []
req_id = 0
for trx in trxs:
source = trx.uid
roadm = next(n for n in network.successors(trx) if isinstance(n, Roadm))
for degree in roadm.per_degree_pch_out_dbm.keys():
node = next(n for n in network.nodes() if n.uid == degree)
# find next roadm
while not isinstance(node, Roadm):
node = next(n for n in network.successors(node))
next_roadm = node
destination = next(n.uid for n in network.successors(next_roadm) if isinstance(n, Transceiver))
params['request_id'] = req_id
req_id += 1
params['source'] = source
params['destination'] = destination
params['nodes_list'] = [degree, destination]
req = PathRequest(**params)
req.power = dbm2watt(req_power)
carrier = {key: getattr(req, key) for key in ['baud_rate', 'roll_off', 'tx_osnr']}
carrier['label'] = ""
carrier['slot_width'] = req.spacing
carrier['delta_pdb'] = 0
carrier['tx_power'] = 1e-3
req.initial_spectrum = {(req.f_min + req.spacing * f): Carrier(**carrier)
for f in range(1, req.nb_channel + 1)}
req_list.append(req)
# add one additional request crossing several roadms to have a complete view
params['source'] = 'trx Rennes_STA'
params['destination'] = 'trx Vannes_KBE'
params['nodes_list'] = ['roadm Lannion_CAS', 'trx Vannes_KBE']
params['bidir'] = True
req = PathRequest(**params)
req.power = dbm2watt(req_power)
carrier = {key: getattr(req, key) for key in ['baud_rate', 'roll_off', 'tx_osnr']}
carrier['label'] = ""
carrier['slot_width'] = req.spacing
carrier['delta_pdb'] = 0
carrier['tx_power'] = 1e-3
req.initial_spectrum = {(req.f_min + req.spacing * f): Carrier(**carrier) for f in range(1, req.nb_channel + 1)}
req_list.append(req)
return req_list
def list_element_attr(element):
"""Return the list of keys to be checked depending on element type. List only the keys that are not
created upon element effective propagation
"""
if isinstance(element, Roadm):
return ['uid', 'name', 'metadata', 'operational', 'type_variety', 'target_pch_out_dbm',
'passive', 'restrictions', 'per_degree_pch_out_dbm',
'target_psd_out_mWperGHz', 'per_degree_pch_psd']
# Dynamically created: 'effective_loss',
if isinstance(element, RoadmParams):
return ['target_pch_out_dbm', 'target_psd_out_mWperGHz', 'per_degree_pch_out_db', 'per_degree_pch_psd',
'add_drop_osnr', 'pmd', 'restrictions']
if isinstance(element, Edfa):
return ['variety_list', 'uid', 'name', 'params', 'metadata', 'operational',
'passive', 'effective_gain', 'delta_p', 'tilt_target', 'out_voa']
# TODO this exhaustive test highlighted that type_variety is not correctly updated from EdfaParams to
# attributes in preamps
# Dynamically created only with channel propagation: 'att_in', 'channel_freq', 'effective_pch_out_db'
# 'gprofile', 'interpol_dgt', 'interpol_gain_ripple', 'interpol_nf_ripple', 'nch', 'nf', 'pin_db', 'pout_db',
# 'target_pch_out_db',
if isinstance(element, FusedParams):
return ['loss']
if isinstance(element, EdfaOperational):
return ['delta_p', 'gain_target', 'out_voa', 'tilt_target']
if isinstance(element, EdfaParams):
return ['f_min', 'f_max', 'type_variety', 'type_def', 'gain_flatmax', 'gain_min', 'p_max', 'nf_model',
'dual_stage_model', 'nf_fit_coeff', 'nf_ripple', 'dgt', 'gain_ripple', 'out_voa_auto',
'allowed_for_design', 'raman']
if isinstance(element, Fiber):
return ['uid', 'name', 'params', 'metadata', 'operational', 'type_variety', 'passive',
'lumped_losses', 'z_lumped_losses']
# Dynamically created 'output_total_power', 'pch_out_db'
if isinstance(element, FiberParams):
return ['_length', '_att_in', '_con_in', '_con_out', '_ref_frequency', '_ref_wavelength',
'_dispersion', '_dispersion_slope', '_dispersion', '_f_dispersion_ref',
'_gamma', '_pmd_coef', '_loss_coef',
'_f_loss_ref', '_lumped_losses']
if isinstance(element, Fused):
return ['uid', 'name', 'params', 'metadata', 'operational', 'loss', 'passive']
if isinstance(element, FusedParams):
return ['loss']
return ['should never come here']
# all initial delta_p are null in topo file, so add random places to change this value
@pytest.mark.parametrize('amp_with_deltap_one', [[],
['east edfa in Lorient_KMA to Vannes_KBE',
'east edfa in Stbrieuc to Rennes_STA',
'west edfa in Lannion_CAS to Morlaix',
'east edfa in a to b',
'west edfa in b to a']])
@pytest.mark.parametrize('power_dbm, req_power', [(0, 0), (0, -3), (3, 3), (0, 3), (3, 0),
(3, 1), (3, 5), (3, 2), (3, 4), (2, 4)])
def test_compare_design_propagation_settings(power_dbm, req_power, amp_with_deltap_one):
"""Check that network design does not change after propagation except for gain in
case of power_saturation during design and/or during propagation:
- in power mode only:
expected behaviour: target power out of roadm does not change
so gain of booster should be reduced/augmented by the exact power difference;
the following amplifiers on the OMS have unchanged gain except if augmentation
of channel power on booster leads to total_power above amplifier max power,
ie if amplifier saturates.
roadm -----booster (pmax 21dBm, 96 channels= 19.82dB)
pdesign=0dBm pch= 0dBm, ^ -20dBm ^G=20dB, Pch=0dBm, Ptot=19.82dBm
pdesign=0dBm pch= -3dBm ^ -20dBm ^G=17dB, Pch=-3dBm, Ptot=16.82dBm
pdesign=3dBm pch= 3dBm ^ -20dBm ^G=23-1.82dB, Pch=1.18dBm, Ptot=21dBm
amplifier can not handle 96x3dBm channels, amplifier saturation is considered
for the choice of amplifier during design
pdesign=0dBm pch= 3dBm ^ -20dBm ^G=23-1.82dB, Pch=1.18dBm, Ptot=21dBm
amplifier can not handle 96x3dBm channels during propagation, amplifier selection
has been done for 0dBm. Saturation is applied for all amps only during propagation
Design applies a saturation verification on amplifiers.
This saturation leads to a power reduction to the max power in the amp library, which
is also applied on the amp delta_p and independantly from propagation.
After design, upon propagation, the amplifier gain and applied delta_p may also change
if total power exceeds max power (eg not the same nb of channels, not the same power per channel
compared to design).
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.
"""
eqpt = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
eqpt['SI']['default'].power_dbm = power_dbm
json_network = load_json(NETWORK_FILE_NAME)
for element in json_network['elements']:
# Initialize a value for delta_p
if element['type'] == 'Edfa':
element['operational']['delta_p'] = 0 + element['operational']['out_voa'] \
if element['operational']['out_voa'] is not None else 0
# apply a 1 dB delta_p on the set of amps
if element['uid'] in amp_with_deltap_one:
element['operational']['delta_p'] = 1
network = network_from_json(json_network, eqpt)
# Build the network once using the default power defined in SI in eqpt config
p_db = power_dbm
nb_channels = automatic_nch(eqpt['SI']['default'].f_min,
eqpt['SI']['default'].f_max,
eqpt['SI']['default'].spacing)
build_network(network, eqpt, pathrequest(p_db, nb_channels=nb_channels), verbose=False)
# record network settings before propagating
# propagate on each oms
req_list = create_per_oms_request(network, eqpt, req_power)
paths = [compute_constrained_path(network, r) for r in req_list]
# systematic comparison of elements settings before and after propagation
# all amps have 21 dBm max power
pch_max = 21 - lin2db(96)
for path, req in zip(paths, req_list):
# check all elements except source and destination trx
# in order to have clean initialization, use deecopy of paths
design_network(req, network, eqpt, verbose=False)
network_copy = deepcopy(network)
pth = deepcopy(path)
_ = propagate(pth, req, eqpt)
for i, element in enumerate(pth[1:-1]):
element_is_first_amp = False
# index of previous element in path is i
if (isinstance(element, Edfa) and isinstance(pth[i], Roadm)) or element.uid == 'west edfa in d to c':
# oms c to d has no booster but one preamp: the power difference is hold there
element_is_first_amp = True
# find the element with the same id in the network_copy
element_copy = next(n for n in network_copy.nodes() if n.uid == element.uid)
for key in list_element_attr(element):
if not isinstance(getattr(element, key),
(EdfaOperational, EdfaParams, FiberParams, RoadmParams, FusedParams)):
if not key == 'effective_gain':
# for all keys, before and after design should be the same except for gain (in power mode)
if isinstance(getattr(element, key), ndarray):
if len(getattr(element, key)) > 0:
assert getattr(element, key) == getattr(element_copy, key)
else:
assert len(getattr(element_copy, key)) == 0
else:
assert getattr(element, key) == getattr(element_copy, key)
else:
dp = element.out_voa if element.uid not in amp_with_deltap_one else element.out_voa + 1
# check that target power is correctly set
assert element.target_pch_out_dbm == req_power + dp
# check that designed gain is exactly applied except if target power exceeds max power, then
# gain is slightly less than the one computed during design for the noiseless reference,
# because during propagation, noise has accumulated, additing to signal.
# check that delta_p is unchanged unless for saturation
if element.target_pch_out_dbm > pch_max:
assert element.effective_gain == pytest.approx(element_copy.effective_gain, abs=2e-2)
else:
assert element.effective_gain == element_copy.effective_gain
# check that delta_p is unchanged unless for saturation
assert element.delta_p == element_copy.delta_p
if element_is_first_amp:
# if element is first amp on path, then it is the one that will saturate if req_power is
# too high
assert mean(element.pch_out_dbm) ==\
pytest.approx(min(pch_max, req_power + element.delta_p - element.out_voa), abs=2e-2)
# check that delta_p is unchanged unless due to saturation
assert element.delta_p == pytest.approx(min(req_power + dp, pch_max) - req_power, abs=1e-2)
# check that delta_p is unchanged unless for saturation
else:
# for all subkeys, before and after design should be the same
for subkey in list_element_attr(getattr(element, key)):
if isinstance(getattr(getattr(element, key), subkey), list):
assert getattr(getattr(element, key), subkey) == getattr(getattr(element_copy, key), subkey)
elif isinstance(getattr(getattr(element, key), subkey), dict):
for value1, value2 in zip(getattr(getattr(element, key), subkey).values(),
getattr(getattr(element_copy, key), subkey).values()):
assert all(value1 == value2)
elif isinstance(getattr(getattr(element, key), subkey), ndarray):
assert_allclose(getattr(getattr(element, key), subkey),
getattr(getattr(element_copy, key), subkey), rtol=1e-12)
else:
assert getattr(getattr(element, key), subkey) == getattr(getattr(element_copy, key), subkey)
@pytest.mark.parametrize("restrictions, fail", [
({'preamp_variety_list': [], 'booster_variety_list':[]}, False),
({'preamp_variety_list': ['std_medium_gain', 'std_low_gain'], 'booster_variety_list':['std_medium_gain']}, False),
# the two next amp type_variety do not exist
({'preamp_variety_list': [], 'booster_variety_list':['booster_medium_gain']}, True),
({'preamp_variety_list': ['std_medium_gain', 'preamp_high_gain'], 'booster_variety_list':[]}, True)])
def test_wrong_restrictions(restrictions, fail):
"""Check that sanity_check correctly raises an error when restriction is incorrect and that library
correctly includes restrictions.
"""
json_data = load_json(EQPT_FILENAME)
# define wrong restriction
json_data['Roadm'][0]['restrictions'] = restrictions
if fail:
with pytest.raises(ConfigurationError):
_ = _equipment_from_json(json_data, EXTRA_CONFIGS)
else:
equipment = _equipment_from_json(json_data, EXTRA_CONFIGS)
assert equipment['Roadm']['example_test'].restrictions == restrictions
@pytest.mark.parametrize('roadm, from_degree, to_degree, expected_impairment_id, expected_type', [
('roadm Lannion_CAS', 'trx Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 1, 'add'),
('roadm Lannion_CAS', 'west edfa in Lannion_CAS to Stbrieuc', 'east edfa in Lannion_CAS to Corlay', 0, 'express'),
('roadm Lannion_CAS', 'west edfa in Lannion_CAS to Stbrieuc', 'trx Lannion_CAS', 2, 'drop'),
('roadm h', 'west edfa in h to g', 'trx h', None, 'drop')
])
def test_roadm_impairments(roadm, from_degree, to_degree, expected_impairment_id, expected_type):
"""Check that impairment id and types are correct
"""
json_data = load_json(NETWORK_FILE_NAME)
for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS':
el['type_variety'] = 'example_detailed_impairments'
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = network_from_json(json_data, equipment)
build_network(network, equipment, pathrequest(0.0, 20.0))
roadm = next(n for n in network.nodes() if n.uid == roadm)
assert roadm.get_roadm_path(from_degree, to_degree).path_type == expected_type
assert roadm.get_roadm_path(from_degree, to_degree).impairment_id == expected_impairment_id
@pytest.mark.parametrize('type_variety, from_degree, to_degree, impairment_id, expected_type', [
(None, 'trx Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 1, 'add'),
('default', 'trx Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 3, 'add'),
(None, 'west edfa in Lannion_CAS to Stbrieuc', 'east edfa in Lannion_CAS to Corlay', None, 'express')
])
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
"""
json_data = load_json(EQPT_FILENAME)
assert 'type_variety' not in json_data['Roadm'][2]
json_data['Roadm'][2]['roadm-path-impairments'] = [
{
"roadm-path-impairments-id": 1,
"roadm-add-path": [{
"frequency-range": {
"lower-frequency": 191.3e12,
"upper-frequency": 196.1e12
},
"roadm-osnr": 41,
}]
}, {
"roadm-path-impairments-id": 3,
"roadm-add-path": [{
"frequency-range": {
"lower-frequency": 191.3e12,
"upper-frequency": 196.1e12
},
"roadm-inband-crosstalk": 0,
"roadm-osnr": 20,
"roadm-noise-figure": 23
}]
}]
equipment = _equipment_from_json(json_data, EXTRA_CONFIGS)
assert equipment['Roadm']['default'].type_variety == 'default'
json_data = load_json(NETWORK_FILE_NAME)
for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS' and type_variety is not None:
el['type_variety'] = type_variety
el['params'] = {
"per_degree_impairments": [
{
"from_degree": from_degree,
"to_degree": to_degree,
"impairment_id": impairment_id
}]
}
network = network_from_json(json_data, equipment)
build_network(network, equipment, pathrequest(0.0, 20.0))
roadm = next(n for n in network.nodes() if n.uid == 'roadm Lannion_CAS')
assert roadm.get_roadm_path(from_degree, to_degree).path_type == expected_type
assert roadm.get_roadm_path(from_degree, to_degree).impairment_id == impairment_id
@pytest.mark.parametrize('from_degree, to_degree, impairment_id, error, message', [
('trx Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 2, NetworkTopologyError,
'Roadm roadm Lannion_CAS path_type is defined as drop but it should be add'), # wrong path_type
('trx Lannion_CAS', 'east edfa toto', 1, ConfigurationError,
'Roadm roadm Lannion_CAS has wrong from-to degree uid trx Lannion_CAS - east edfa toto'), # wrong degree
('trx Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 11, NetworkTopologyError,
'ROADM roadm Lannion_CAS: impairment profile id 11 is not defined in library') # wrong impairment_id
])
def test_wrong_roadm_per_degree_impairments(from_degree, to_degree, impairment_id, error, message):
"""Check that wrong per degree definitions are correctly catched
"""
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
json_data = load_json(NETWORK_FILE_NAME)
for el in json_data['elements']:
if el['uid'] == 'roadm Lannion_CAS':
el['type_variety'] = 'example_detailed_impairments'
el['params'] = {
"per_degree_impairments": [
{
"from_degree": from_degree,
"to_degree": to_degree,
"impairment_id": impairment_id
}]
}
network = network_from_json(json_data, equipment)
with pytest.raises(error, match=message):
build_network(network, equipment, pathrequest(0.0, 20.0))
@pytest.mark.parametrize('path_type, type_variety, expected_pmd, expected_pdl, expected_osnr, freq', [
('express', 'default', 5.0e-12, 0.5, None, [191.3e12]), # roadm instance parameters pre-empts library
('express', 'example_test', 5.0e-12, 0.5, None, [191.3e12]),
('express', 'example_detailed_impairments', 0, 0, None, [191.3e12]), # detailed parameters pre-empts global ones
('add', 'default', 5.0e-12, 0.5, None, [191.3e12]),
('add', 'example_test', 5.0e-12, 0.5, None, [191.3e12]),
('add', 'example_detailed_impairments', 0, 0, 41, [191.3e12]),
('add', 'example_detailed_impairments', [0, 0], [0.5, 0], [35, 41], [188.5e12, 191.3e12])])
def test_impairment_initialization(path_type, type_variety, expected_pmd, expected_pdl, expected_osnr, freq):
"""Check that impairments are correctly initialized, with this order:
- use equipment roadm impairments if no impairment are set in the ROADM instance
- use roadm global impairment if roadm global impairment are set
- 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
"""
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
extra_params = equipment['Roadm'][type_variety].__dict__
roadm_config = {
"uid": "roadm Lannion_CAS",
"params": {
"add_drop_osnr": 38,
"pmd": 5.0e-12,
"pdl": 0.5
}
}
if type_variety != 'default':
roadm_config["type_variety"] = type_variety
roadm_config['params'] = merge_amplifier_restrictions(roadm_config['params'], extra_params)
roadm = Roadm(**roadm_config)
roadm.set_roadm_paths(from_degree='tata', to_degree='toto', path_type=path_type)
assert roadm.get_roadm_path(from_degree='tata', to_degree='toto').path_type == path_type
assert_allclose(roadm.get_impairment('roadm-pmd', freq, from_degree='tata', degree='toto'),
expected_pmd, rtol=1e-12)
assert_allclose(roadm.get_impairment('roadm-pdl', freq, from_degree='tata', degree='toto'),
expected_pdl, rtol=1e-12)
if path_type == 'add':
# we assume for simplicity that add contribution is the same as drop contribution
# add_drop_osnr_db = 10log10(1/add_osnr + 1/drop_osnr)
if type_variety in ['default', 'example_test']:
assert_allclose(roadm.get_impairment('roadm-osnr', freq, from_degree='tata', degree='toto'),
roadm.params.add_drop_osnr + lin2db(2), rtol=1e-12)
else:
assert_allclose(roadm.get_impairment('roadm-osnr', freq, from_degree='tata', degree='toto'),
expected_osnr, rtol=1e-12)
Reference in New Issue View Git Blame Copy Permalink