Telecominfraproject/oopt-gnpy
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/oopt-gnpy.git synced 2026-01-27 02:21:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Using new pch and ptot definitions

Browse Source 
Using the new defined attribute in a coherent way along the code.
Still no changes in the behaviour

Change-Id: I6e43db1e28a5456e0522e52c0f74e79969307ed9
This commit is contained in:
AndreaDAmico
2025-12-02 13:42:16 -05:00
parent e152f66546
commit 13cf2d2d01
5 changed files with 82 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
gnpy/core/elements.py
View File

@@ -219,13 +219,12 @@ class Transceiver(_Node):
with errstate(divide='ignore'):
self.propagated_labels = spectral_info.label
self.baud_rate = spectral_info.baud_rate
ratio_01nm = lin2db(12.5e9 / self.baud_rate)
# set raw values to record original calculation, before update_snr()
self.raw_osnr_ase = lin2db(spectral_info.signal / spectral_info.ase)
self.raw_osnr_ase_01nm = self.raw_osnr_ase - ratio_01nm
self.raw_osnr_nli = lin2db(spectral_info.signal / spectral_info.nli)
self.raw_snr = lin2db(spectral_info.signal / (spectral_info.ase + spectral_info.nli))
self.raw_snr_01nm = self.raw_snr - ratio_01nm
self.raw_osnr_ase = spectral_info.snr_lin_db
self.raw_osnr_ase_01nm = spectral_info.opt_snr_lin_db
self.raw_osnr_nli = spectral_info.snr_nli_db
self.raw_snr = spectral_info.gsnr_db
self.raw_snr_01nm = spectral_info.opt_gsnr_db
self.osnr_ase = self.raw_osnr_ase
self.osnr_ase_01nm = self.raw_osnr_ase_01nm
@@ -588,12 +587,12 @@ class Roadm(_Node):
:type from_degree: str
"""
# record input powers to compute the actual loss at the end of the process
input_power_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
input_pch_dbm = spectral_info.pch_dbm
# apply min ROADM loss if it exists
roadm_maxloss_db = self.get_impairment('roadm-maxloss', spectral_info.frequency, from_degree, degree)
spectral_info.apply_attenuation_db(roadm_maxloss_db)
# records the total power after applying minimum loss
net_input_power_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
net_input_pch_dbm = spectral_info.pch_dbm
# find the target power for the reference carrier
ref_per_degree_pch = self.get_per_degree_ref_power(degree)
# find the target powers for each signal carrier
@@ -630,9 +629,9 @@ class Roadm(_Node):
# that had the min power.
# This change corresponds to a discussion held during coders call. Please look at this document for
# a reference: https://telecominfraproject.atlassian.net/wiki/spaces/OOPT/pages/669679645/PSE+Meeting+Minutes
correction = calculate_absolute_min_or_zero(net_input_power_dbm - target_power_per_channel)
correction = calculate_absolute_min_or_zero(net_input_pch_dbm - target_power_per_channel)
new_target = target_power_per_channel - correction
delta_power = net_input_power_dbm - new_target
delta_power = net_input_pch_dbm - new_target
spectral_info.apply_attenuation_db(delta_power)
@@ -645,10 +644,10 @@ class Roadm(_Node):
spectral_info.pdl = sqrt(spectral_info.pdl ** 2 + pdl_impairment ** 2)
# Update the per channel power with the result of propagation
self.pch_out_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
self.pch_out_dbm = spectral_info.pch_dbm
# Update the loss per channel and the labels
self.loss_pch_db = input_power_dbm - self.pch_out_dbm
self.loss_pch_db = input_pch_dbm - self.pch_out_dbm
self.propagated_labels = spectral_info.label
def set_roadm_paths(self, from_degree, to_degree, path_type, impairment_id=None):
@@ -1138,7 +1137,7 @@ class Fiber(_Node):
# apply the attenuation due to the output connector loss
attenuation_out_db = self.params.con_out
spectral_info.apply_attenuation_db(attenuation_out_db)
self.pch_out_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
self.pch_out_dbm = spectral_info.pch_dbm
self.propagated_labels = spectral_info.label
def __call__(self, spectral_info):
@@ -1242,7 +1241,7 @@ class RamanFiber(Fiber):
# apply the attenuation due to the output connector loss
attenuation_out_db = self.params.con_out
spectral_info.apply_attenuation_db(attenuation_out_db)
self.pch_out_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
self.pch_out_dbm = watt2dbm(spectral_info.pch)
self.propagated_labels = spectral_info.label
pout = watt2dbm(sum(spectral_info.signal))
self.actual_raman_gain = self.loss + pout - pin
@@ -1433,8 +1432,8 @@ class Edfa(_Node):
self.interpol_nf_ripple = interp(spectral_info.frequency, amplifier_freq, self.params.nf_ripple)
self.nch = spectral_info.number_of_channels
pin = spectral_info.signal + spectral_info.ase + spectral_info.nli
self.pin_db = watt2dbm(sum(pin))
pch_in = spectral_info.pch
self.pin_db = watt2dbm(spectral_info.ptot)
# The following should be changed when we have the new spectral information including slot widths.
# For now, with homogeneous spectrum, we can calculate it as the difference between neighbouring channels.
self.slot_width = self.channel_freq[1] - self.channel_freq[0]
@@ -1448,10 +1447,10 @@ class Edfa(_Node):
# check power saturation and correct target_gain accordingly:
self.nf = self._calc_nf()
self.gprofile = self._gain_profile(pin)
self.gprofile = self._gain_profile(pch_in)
pout = (pin + self.noise_profile(spectral_info)) * db2lin(self.gprofile)
self.pout_db = lin2db(sum(pout * 1e3))
pch_out = (pch_in + self.noise_profile(spectral_info)) * db2lin(self.gprofile)
self.pout_db = watt2dbm(sum(pch_out))
# ase & nli are only calculated in signal bandwidth
# pout_db is not the absolute full output power (negligible if sufficient channels)
@@ -1626,13 +1625,13 @@ class Edfa(_Node):
# second estimate of amp ch gain using the channel input profile
g2nd = g1st - voa
pout_db = lin2db(sum(pin * 1e3 * db2lin(g2nd)))
pout_db = watt2dbm(sum(pin * db2lin(g2nd)))
dgts2 = self.effective_gain - (pout_db - tot_in_power_db)
# center estimate of amp ch gain
xcent = dgts2
gcent = g1st - voa + array(self.interpol_dgt) * xcent
pout_db = lin2db(sum(pin * 1e3 * db2lin(gcent)))
pout_db = watt2dbm(sum(pin * db2lin(gcent)))
gavg_cent = pout_db - tot_in_power_db
# Lower estimate of amp ch gain
@@ -1644,13 +1643,13 @@ class Edfa(_Node):
xlow = dgts2 - deltax
glow = g1st - voa + array(self.interpol_dgt) * xlow
pout_db = lin2db(sum(pin * 1e3 * db2lin(glow)))
pout_db = watt2dbm(sum(pin * db2lin(glow)))
gavg_low = pout_db - tot_in_power_db
# upper gain estimate
xhigh = dgts2 + deltax
ghigh = g1st - voa + array(self.interpol_dgt) * xhigh
pout_db = lin2db(sum(pin * 1e3 * db2lin(ghigh)))
pout_db = watt2dbm(sum(pin * db2lin(ghigh)))
gavg_high = pout_db - tot_in_power_db
# compute slope
@@ -1683,7 +1682,7 @@ class Edfa(_Node):
spectral_info.apply_gain_db(self.gprofile - self.out_voa)
spectral_info.pmd = sqrt(spectral_info.pmd ** 2 + self.params.pmd ** 2)
spectral_info.pdl = sqrt(spectral_info.pdl ** 2 + self.params.pdl ** 2)
self.pch_out_dbm = watt2dbm(spectral_info.signal + spectral_info.nli + spectral_info.ase)
self.pch_out_dbm = spectral_info.pch_dbm
self.propagated_labels = spectral_info.label
def __call__(self, spectral_info):

48
gnpy/core/info.py
View File

@@ -20,7 +20,7 @@ from typing import Union, List, Optional
from dataclasses import dataclass
from numpy import argsort, mean, array, append, ones, ceil, any, zeros, outer, full, ndarray, asarray
from gnpy.core.utils import automatic_nch, db2lin, watt2dbm
from gnpy.core.utils import automatic_nch, db2lin, watt2dbm, lin2db
from gnpy.core.exceptions import SpectrumError
DEFAULT_SLOT_WIDTH_STEP = 12.5e9 # Hz
@@ -112,12 +112,20 @@ class SpectralInformation(object):
@property
def pch(self):
return self._pch
return array(self._pch)
@property
def pch_dbm(self):
return watt2dbm(self.pch)
@property
def ptot(self):
return sum(self._pch)
@property
def ptot_dbm(self):
return watt2dbm(self.ptot)
@pch.setter
def pch(self, pch):
self._pch = pch
@@ -126,10 +134,18 @@ class SpectralInformation(object):
def signal(self):
return self._signal_ratio * self._pch
@property
def signal_dbm(self):
return watt2dbm(self.signal)
@property
def nli(self):
return self._nli_ratio * self._pch
@property
def nli_dbm(self):
return watt2dbm(self.nli)
def add_nli(self, nli):
pch = self.pch + nli
self._signal_ratio *= self.pch / pch
@@ -141,6 +157,10 @@ class SpectralInformation(object):
def ase(self):
return self._ase_ratio * self._pch
@property
def ase_dbm(self):
return watt2dbm(self.ase)
def add_ase(self, ase):
pch = self.pch + ase
self._signal_ratio *= self.pch / pch
@@ -152,14 +172,38 @@ class SpectralInformation(object):
def snr_lin(self):
return self._signal_ratio / self._ase_ratio
@property
def snr_lin_db(self):
return lin2db(self.snr_lin)
@property
def opt_snr_lin_db(self):
return self.snr_lin_db - lin2db(12.5e9/self.baud_rate)
@property
def snr_nli(self):
return self._signal_ratio / self._nli_ratio
@property
def snr_nli_db(self):
return lin2db(self.snr_nli)
@property
def opt_snr_nli_db(self):
return self.snr_nli_db - lin2db(12.5e9/self.baud_rate)
@property
def gsnr(self):
return self._signal_ratio / (self._ase_ratio + self._nli_ratio)
@property
def gsnr_db(self):
return lin2db(self.gsnr)
@property
def opt_gsnr_db(self):
return self.gsnr_db - lin2db(12.5e9/self.baud_rate)
@property
def roll_off(self):
return self._roll_off

10
tests/test_amplifier.py
View File

@@ -146,7 +146,7 @@ def test_fixed_gain_nf(gain, nf_expected, setup_edfa_fixed_gain, si):
def test_si(si, nch_and_spacing):
"""basic total power check of the channel comb generation"""
nb_channel = nch_and_spacing[0]
p_tot = sum(si.signal + si.ase + si.nli)
p_tot = si.ptot
expected_p_tot = si.signal[0] * nb_channel
assert pytest.approx(expected_p_tot, abs=0.01) == p_tot
@@ -219,12 +219,12 @@ def test_ase_noise(gain, si, setup_trx, bw):
edfa.interpol_params(si)
nf = edfa.nf
print('nf', nf)
pin = lin2db((si.signal[0] + si.ase[0] + si.nli[0]) * 1e3)
pin = watt2dbm(si.pch[0])
osnr_expected = pin - nf[0] + 58
si = edfa(si)
print(edfa)
osnr = lin2db(si.signal[0] / si.ase[0]) - lin2db(12.5e9 / bw)
osnr = si.opt_snr_lin_db[0]
assert pytest.approx(osnr_expected, abs=0.01) == osnr
trx = setup_trx
@@ -280,7 +280,7 @@ def test_amp_behaviour(tilt_target, delta_p):
expected_total_power_out = total_sig_powerin * 100 * db2lin(delta_p) if delta_p else total_sig_powerin * 100
assert pytest.approx(total_sig_powerout, abs=1e-6) == min(expected_total_power_out, dbm2watt(21))
assert pytest.approx(edfa.effective_gain, 1e-5) == gain
assert watt2dbm(sum(si.signal + si.nli + si.ase)) <= 21.01
assert si.ptot_dbm <= 21.01
# If there is no tilt on the amp: the gain is identical for all carriers
if tilt_target == 0:
assert_allclose(sig_in + gain, sig_out, rtol=1e-13)
@@ -363,7 +363,7 @@ def test_amp_saturation(delta_pdb_per_channel, base_power, delta_p):
sig_out = lin2db(si.signal)
total_sig_powerout = sum(si.signal)
gain = lin2db(total_sig_powerout / total_sig_powerin)
assert watt2dbm(sum(si.signal + si.nli + si.ase)) <= 21.02
assert si.ptot_dbm <= 21.02
assert pytest.approx(edfa.effective_gain, 1e-13) == gain
assert_allclose(sig_in + gain, sig_out, rtol=1e-13)

6
tests/test_equalization.py
View File

@@ -594,10 +594,10 @@ def test_equalization(case, deltap, target, mode, slot_width, equalization):
si = el(si, degree=path[i + 1].uid, from_degree=path[i - 1].uid)
if case in ['SI', 'nodes', 'degrees']:
if equalization == 'target_psd_out_mWperGHz':
assert_allclose(power_dbm_to_psd_mw_ghz(watt2dbm(si.signal + si.ase + si.nli), si.baud_rate),
assert_allclose(power_dbm_to_psd_mw_ghz(si.pch_dbm, si.baud_rate),
target_psd, rtol=1e-3)
if equalization == 'target_out_mWperSlotWidth':
assert_allclose(power_dbm_to_psd_mw_ghz(watt2dbm(si.signal + si.ase + si.nli), si.slot_width),
assert_allclose(power_dbm_to_psd_mw_ghz(si.pch_dbm, si.slot_width),
target_psd, rtol=1e-3)
else:
si = el(si)
@@ -988,4 +988,4 @@ def test_tx_power(tx_power_dbm):
si = node(si, path[i + 1], path[i - 1])
else:
si = node(si)
assert_allclose(watt2dbm(si.signal + si.ase + si.nli), array([-20, -20, -20]), rtol=1e-5)
assert_allclose(si.pch_dbm, array([-20, -20, -20]), rtol=1e-5)

10
tests/test_roadm_restrictions.py
View File

@@ -309,9 +309,9 @@ def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm, roa
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
pch_in_roadm = si.pch
si = el(si, degree=path[i + 1].uid, from_degree=path[i - 1].uid)
power_out_roadm = si.signal + si.ase + si.nli
pch_out_roadm = si.pch
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)
@@ -324,7 +324,7 @@ def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm, roa
# 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)
assert_allclose(pch_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.
@@ -332,9 +332,9 @@ def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm, roa
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_allclose(pch_out_roadm, pch_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)
pytest.approx(lin2db(min(pch_in_roadm) * 1e3), rel=1e-3)
else:
si = el(si)