mirror of
https://github.com/Telecominfraproject/oopt-gnpy.git
synced 2025-10-30 17:47:50 +00:00
56e615c713
Correctly uses the oms band and spacing for computing the nb of channel and total power for design per band. In order to keep the SI values as reference, introduce a new parameter in SI to indicate wether to use this feature or not. If "use_si_channel_count_for_design": true, then the f_min, f_max and spacing from SI are used for all OMSes else, the f_min, f_max, spacing defined per OMS (design_bands) is used. This impacts tests where the artificial C-band boudaries were hardcoded, and it also has an impact on performances when SI's defined nb of channels is larger than the one defined per OMS. In this case the design was considering a larger total power than the one finally propagated which resulted in reduced performance. This feature now corrects this case (if "use_si_channel_count_for_design": false which is the default setting). Overall autodesign are thus improved. Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com> Change-Id: I471a2c45200894ca354c90b46b662f42414b48ad tous les test marche et les jeu de tests aussi. Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com> Change-Id: If25b47aa10f97301fde7f17daa2a9478aed46db2
243 lines
8.7 KiB
Python
243 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# SPDX-License-Identifier: BSD-3-Clause
|
|
# test_propagation
|
|
# Copyright (C) 2025 Telecom Infra Project and GNPy contributors
|
|
# see AUTHORS.rst for a list of contributors
|
|
|
|
"""
|
|
Check that propagation example give expected results
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from pathlib import Path
|
|
from networkx import dijkstra_path
|
|
from numpy import mean, sqrt, ones
|
|
import re
|
|
|
|
from gnpy.core.exceptions import SpectrumError
|
|
from gnpy.core.elements import Transceiver, Fiber, Edfa, Roadm
|
|
from gnpy.core.utils import db2lin, dbm2watt
|
|
from gnpy.core.info import create_input_spectral_information
|
|
from gnpy.core.network import build_network
|
|
from gnpy.tools.json_io import load_network, load_equipment, network_from_json
|
|
from gnpy.topology.request import PathRequest
|
|
|
|
TEST_DIR = Path(__file__).parent
|
|
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)],
|
|
ids=['50GHz spacing', '75GHz spacing', '100GHz spacing', '2 channels'])
|
|
# TODO in elements.py code: pytests doesn't pass with 1 channel: interpolate fail
|
|
def nch_and_spacing(request):
|
|
"""parametrize channel count vs channel spacing (Hz)"""
|
|
yield request.param
|
|
|
|
|
|
def pathrequest(pch_dbm, p_tot_dbm):
|
|
"""create ref channel for defined power settings
|
|
"""
|
|
params = {
|
|
"power": dbm2watt(pch_dbm),
|
|
"tx_power": dbm2watt(pch_dbm),
|
|
"nb_channel": 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)
|
|
|
|
|
|
def propagation(input_power, con_in, con_out, dest):
|
|
equipment = load_equipment(eqpt_library_name, EXTRA_CONFIGS)
|
|
network = load_network(network_file_name, equipment)
|
|
|
|
# parametrize the network elements with the con losses and adapt gain
|
|
# (assumes all spans are identical)
|
|
for e in network.nodes():
|
|
if isinstance(e, Fiber):
|
|
loss = e.params.loss_coef * e.params.length
|
|
e.params.con_in = con_in
|
|
e.params.con_out = con_out
|
|
if isinstance(e, Edfa):
|
|
e.operational.gain_target = loss + con_in + con_out
|
|
|
|
build_network(network, equipment, pathrequest(0, 20))
|
|
|
|
transceivers = {n.uid: n for n in network.nodes() if isinstance(n, Transceiver)}
|
|
|
|
p = input_power
|
|
p = db2lin(p) * 1e-3
|
|
spacing = 50e9 # THz
|
|
si = create_input_spectral_information(f_min=191.3e12, f_max=191.3e12 + 79 * spacing, roll_off=0.15,
|
|
baud_rate=32e9, spacing=spacing, tx_osnr=None,
|
|
tx_power=p)
|
|
source = next(transceivers[uid] for uid in transceivers if uid == 'trx A')
|
|
sink = next(transceivers[uid] for uid in transceivers if uid == dest)
|
|
path = dijkstra_path(network, source, sink)
|
|
for el in path:
|
|
si = el(si)
|
|
print(el) # remove this line when sweeping across several powers
|
|
edfa_sample = next(el for el in path if isinstance(el, Edfa))
|
|
nf = mean(edfa_sample.nf)
|
|
|
|
print(f'pw: {input_power} conn in: {con_in} con out: {con_out}',
|
|
f'OSNR@0.1nm: {round(mean(sink.osnr_ase_01nm),2)}',
|
|
f'SNR@bandwitdth: {round(mean(sink.snr),2)}')
|
|
return sink, nf, path
|
|
|
|
|
|
test = {'a': (-1, 1, 0), 'b': (-1, 1, 1), 'c': (0, 1, 0), 'd': (1, 1, 1)}
|
|
expected = {'a': (-2, 0, 0), 'b': (-2, 0, 1), 'c': (-1, 0, 0), 'd': (0, 0, 1)}
|
|
|
|
|
|
@pytest.mark.parametrize("dest", ['trx B', 'trx F'])
|
|
@pytest.mark.parametrize("osnr_test", ['a', 'b', 'c', 'd'])
|
|
def test_snr(osnr_test, dest):
|
|
pw = test[osnr_test][0]
|
|
conn_in = test[osnr_test][1]
|
|
conn_out = test[osnr_test][2]
|
|
sink, nf, _ = propagation(pw, conn_in, conn_out, dest)
|
|
osnr = round(mean(sink.osnr_ase), 3)
|
|
nli = 1.0 / db2lin(round(mean(sink.snr), 3)) - 1.0 / db2lin(osnr)
|
|
pw = expected[osnr_test][0]
|
|
conn_in = expected[osnr_test][1]
|
|
conn_out = expected[osnr_test][2]
|
|
sink, exp_nf, _ = propagation(pw, conn_in, conn_out, dest)
|
|
expected_osnr = round(mean(sink.osnr_ase), 3)
|
|
expected_nli = 1.0 / db2lin(round(mean(sink.snr), 3)) - 1.0 / db2lin(expected_osnr)
|
|
# compare OSNR taking into account nf change of amps
|
|
osnr_diff = abs(osnr - expected_osnr + nf - exp_nf)
|
|
nli_diff = abs((nli - expected_nli) / nli)
|
|
assert osnr_diff < 0.01 and nli_diff < 0.01
|
|
|
|
|
|
@pytest.mark.parametrize("dest", ['trx B', 'trx F'])
|
|
@pytest.mark.parametrize("cd_test", ['a', 'b', 'c', 'd'])
|
|
def test_chromatic_dispersion(cd_test, dest):
|
|
pw = test[cd_test][0]
|
|
conn_in = test[cd_test][1]
|
|
conn_out = test[cd_test][2]
|
|
sink, _, path = propagation(pw, conn_in, conn_out, dest)
|
|
|
|
chromatic_dispersion = sink.chromatic_dispersion
|
|
|
|
num_ch = len(chromatic_dispersion)
|
|
expected_cd = 0
|
|
for el in path:
|
|
expected_cd += el.params.dispersion * el.params.length if isinstance(el, Fiber) else 0
|
|
expected_cd = expected_cd * ones(num_ch) * 1e3
|
|
assert chromatic_dispersion == pytest.approx(expected_cd)
|
|
|
|
|
|
@pytest.mark.parametrize("dest", ['trx B', 'trx F'])
|
|
@pytest.mark.parametrize("dgd_test", ['a', 'b', 'c', 'd'])
|
|
def test_dgd(dgd_test, dest):
|
|
pw = test[dgd_test][0]
|
|
conn_in = test[dgd_test][1]
|
|
conn_out = test[dgd_test][2]
|
|
sink, _, path = propagation(pw, conn_in, conn_out, dest)
|
|
|
|
pmd = sink.pmd
|
|
|
|
num_ch = len(pmd)
|
|
expected_pmd = 0
|
|
for el in path:
|
|
expected_pmd += el.params.pmd_coef**2 * el.params.length if isinstance(el, Fiber) else 0
|
|
expected_pmd += el.params.pmd**2 if isinstance(el, Roadm) else 0
|
|
expected_pmd = sqrt(expected_pmd) * ones(num_ch) * 1e12
|
|
assert pmd == pytest.approx(expected_pmd)
|
|
|
|
|
|
def wrong_element_propagate():
|
|
"""
|
|
"""
|
|
data = []
|
|
data.append({
|
|
"error": SpectrumError,
|
|
"json_data": {
|
|
"elements": [{
|
|
"uid": "Elem",
|
|
"type": "Fiber",
|
|
"type_variety": "SSMF",
|
|
"params": {
|
|
"dispersion_per_frequency": {
|
|
"frequency": [
|
|
185.49234135667396e12,
|
|
186.05251641137855e12,
|
|
188.01312910284463e12,
|
|
189.99124726477024e12],
|
|
"value": [
|
|
1.60e-05,
|
|
1.67e-05,
|
|
1.7e-05,
|
|
1.8e-05]
|
|
},
|
|
"length": 1.02,
|
|
"loss_coef": 2.85,
|
|
"length_units": "km",
|
|
"att_in": 0.0,
|
|
"con_in": 0.0,
|
|
"con_out": 0.0
|
|
}
|
|
}],
|
|
"connections": []
|
|
},
|
|
"expected_msg": 'The spectrum bandwidth exceeds the frequency interval used to define the fiber Chromatic '
|
|
+ 'Dispersion in "Fiber Elem".\nSpectrum f_min-f_max: 191.35-196.1\nChromatic Dispersion '
|
|
+ 'f_min-f_max: 185.49-189.99'
|
|
})
|
|
return data
|
|
|
|
|
|
@pytest.mark.parametrize('error, json_data, expected_msg',
|
|
[(e['error'], e['json_data'], e['expected_msg']) for e in wrong_element_propagate()])
|
|
def test_json_element(error, json_data, expected_msg):
|
|
"""
|
|
Check that a missing key is correctly raisong the logger
|
|
"""
|
|
equipment = load_equipment(eqpt_library_name, EXTRA_CONFIGS)
|
|
network = network_from_json(json_data, equipment)
|
|
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,
|
|
baud_rate=32e9, tx_power=1.0e-3, spacing=50.0e9, tx_osnr=45)
|
|
with pytest.raises(error, match=re.escape(expected_msg)):
|
|
_ = elem(si)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from logging import getLogger, basicConfig, INFO
|
|
logger = getLogger(__name__)
|
|
basicConfig(level=INFO)
|
|
|
|
for a in test:
|
|
test_snr(a, 'trx F')
|
|
print('\n')
|