From 45a11e5456b7f35e01267a665b67de55ff1fd176 Mon Sep 17 00:00:00 2001 From: Linqi Xiao Date: Sat, 7 Feb 2026 22:06:33 -0600 Subject: [PATCH] Add total_loss parameter as alternative to loss_coef for fiber attenuation Support OTDR-measured total span loss via new Fiber.total_loss parameter. When provided, it overrides loss_coef with an effective per-km value. - Add total_loss field to Fiber element with validation - Compute effective loss_coef = total_loss / length when total_loss is set - Add tests for total_loss functionality and edge cases - Add author entry to AUTHORS.rst --- AUTHORS.rst | 1 + gnpy/core/elements.py | 9 +++- gnpy/core/parameters.py | 44 ++++++++++++++--- gnpy/tools/convert_legacy_yang.py | 5 ++ gnpy/tools/yang_convert_utils.py | 42 ++++++++++++++++ tests/test_parameters.py | 80 +++++++++++++++++++++++++++++++ 6 files changed, 174 insertions(+), 7 deletions(-) diff --git a/AUTHORS.rst b/AUTHORS.rst index c9c56e65c..65dba9fe5 100644 --- a/AUTHORS.rst +++ b/AUTHORS.rst @@ -18,6 +18,7 @@ To learn how to contribute, please see CONTRIBUTING.md - Gabriele Galimberti (Cisco) - Gert Grammel (Juniper Networks) - Giacomo Borraccini (NEC Laboratories America) +- Linqi Xiao (University of Texas at Dallas) , , - Gilad Goldfarb (Facebook) - James Powell (Telecom Infra Project) - Jan Kundrát (Telecom Infra Project) diff --git a/gnpy/core/elements.py b/gnpy/core/elements.py index 791838934..ad22c55c5 100644 --- a/gnpy/core/elements.py +++ b/gnpy/core/elements.py @@ -903,7 +903,14 @@ def to_json(self): 'con_in': self.params.con_in, 'con_out': self.params.con_out } - if isinstance(self.params.loss_coef, ndarray): + if self.params.total_loss is not None: + if self.params.total_loss.size > 1: + params["total_loss_per_frequency"] = [ + {"frequency": frequency, "total_loss_value": round(float(tl), 6)} + for frequency, tl in zip(self.params.f_loss_ref, self.params.total_loss)] + else: + params["total_loss"] = round(float(self.params.total_loss), 6) + elif isinstance(self.params.loss_coef, ndarray): params["loss_coef_per_frequency"] = [ {"frequency": frequency, "loss_coef_value": round(loss * 1e3, 6)} for frequency, loss in zip(self.params.f_loss_ref, self.params.loss_coef)] diff --git a/gnpy/core/parameters.py b/gnpy/core/parameters.py index d42f67229..54ac5c30f 100644 --- a/gnpy/core/parameters.py +++ b/gnpy/core/parameters.py @@ -351,12 +351,32 @@ def __init__(self, **kwargs): self._pmd_coef_defined = kwargs.get('pmd_coef_defined', kwargs['pmd_coef'] is True) # Loss Coefficient - if isinstance(kwargs['loss_coef'], dict): - self._loss_coef = asarray(kwargs['loss_coef']['value']) * 1e-3 # lineic loss dB/m - self._f_loss_ref = asarray(kwargs['loss_coef']['frequency']) # Hz + # Support total_loss (dB) as an alternative to loss_coef (dB/km). + # When total_loss is provided, it represents the total fiber attenuation + # (excluding connectors), as typically measured by OTDR. + # The effective loss_coef is then derived as total_loss / length. + self._total_loss = None + if 'total_loss' in kwargs and kwargs['total_loss'] is not None: + total_loss = kwargs['total_loss'] + if isinstance(total_loss, dict): + # Frequency-dependent total loss: {value: [...], frequency: [...]} + self._total_loss = asarray(total_loss['value']) # dB + self._f_loss_ref = asarray(total_loss['frequency']) # Hz + self._loss_coef = self._total_loss / (self._length * 1e-3) # dB/km -> dB/m (*1e-3) + self._loss_coef = self._loss_coef * 1e-3 # dB/m + else: + self._total_loss = asarray(float(total_loss)) # dB + self._f_loss_ref = asarray(self._ref_frequency) # Hz + self._loss_coef = asarray(self._total_loss / (self._length * 1e-3)) * 1e-3 # dB/m + elif 'loss_coef' in kwargs: + if isinstance(kwargs['loss_coef'], dict): + self._loss_coef = asarray(kwargs['loss_coef']['value']) * 1e-3 # lineic loss dB/m + self._f_loss_ref = asarray(kwargs['loss_coef']['frequency']) # Hz + else: + self._loss_coef = asarray(kwargs['loss_coef']) * 1e-3 # lineic loss dB/m + self._f_loss_ref = asarray(self._ref_frequency) # Hz else: - self._loss_coef = asarray(kwargs['loss_coef']) * 1e-3 # lineic loss dB/m - self._f_loss_ref = asarray(self._ref_frequency) # Hz + raise KeyError('loss_coef') # Lumped Losses self._lumped_losses = kwargs['lumped_losses'] if 'lumped_losses' in kwargs else array([]) self._latency = self._length / (c / self._n1) # s @@ -446,6 +466,10 @@ def ref_wavelength(self): def ref_frequency(self): return self._ref_frequency + @property + def total_loss(self): + return self._total_loss + @property def loss_coef(self): return self._loss_coef @@ -464,7 +488,15 @@ def latency(self): def asdict(self): dictionary = super().asdict() - dictionary['loss_coef'] = self.loss_coef * 1e3 + if self._total_loss is not None: + if self._total_loss.size > 1: + dictionary['total_loss'] = {'value': self._total_loss.tolist(), + 'frequency': self._f_loss_ref.tolist()} + else: + dictionary['total_loss'] = float(self._total_loss) + dictionary.pop('loss_coef', None) + else: + dictionary['loss_coef'] = self.loss_coef * 1e3 dictionary['length_units'] = 'm' if len(self.lumped_losses) == 0: dictionary.pop('lumped_losses') diff --git a/gnpy/tools/convert_legacy_yang.py b/gnpy/tools/convert_legacy_yang.py index a7e3be975..1c0666480 100644 --- a/gnpy/tools/convert_legacy_yang.py +++ b/gnpy/tools/convert_legacy_yang.py @@ -25,6 +25,7 @@ convert_design_band, convert_back_design_band, \ convert_none_to_empty, convert_empty_to_none, \ convert_loss_coeff_list, convert_back_loss_coeff_list, \ + convert_total_loss_list, convert_back_total_loss_list, \ ELEMENTS_KEY, PATH_REQUEST_KEY, RESPONSE_KEY, SPECTRUM_KEY, EQPT_TYPES, EDFA_CONFIG_KEYS, SIM_PARAMS_KEYS, \ TOPO_NMSP, SERV_NMSP, EQPT_NMSP, SPECTRUM_NMSP, SIM_PARAMS_NMSP, EDFA_CONFIG_NMSP, RESP_NMSP, \ dump_data, add_missing_default_type_variety, \ @@ -56,6 +57,7 @@ def legacy_to_yang(json_data: Dict) -> Dict: json_data = convert_degree(json_data) json_data = convert_design_band(json_data) json_data = convert_loss_coeff_list(json_data) + json_data = convert_total_loss_list(json_data) json_data = convert_raman_coef(json_data) json_data = {TOPO_NMSP: json_data} elif TOPO_NMSP in json_data: @@ -63,6 +65,7 @@ def legacy_to_yang(json_data: Dict) -> Dict: json_data[TOPO_NMSP] = convert_degree(json_data[TOPO_NMSP]) json_data[TOPO_NMSP] = convert_design_band(json_data[TOPO_NMSP]) json_data[TOPO_NMSP] = convert_loss_coeff_list(json_data[TOPO_NMSP]) + json_data[TOPO_NMSP] = convert_total_loss_list(json_data[TOPO_NMSP]) json_data[TOPO_NMSP] = remove_null_region_city(json_data[TOPO_NMSP]) # case of equipment json @@ -149,11 +152,13 @@ def yang_to_legacy(json_data: Dict) -> Dict: json_data = convert_back_degree(json_data) json_data = convert_back_design_band(json_data) json_data = convert_back_loss_coeff_list(json_data) + json_data = convert_back_total_loss_list(json_data) json_data = convert_back_raman_coef(json_data) elif TOPO_NMSP in json_data: json_data = convert_back_degree(json_data[TOPO_NMSP]) json_data = convert_back_design_band(json_data) json_data = convert_back_loss_coeff_list(json_data) + json_data = convert_back_total_loss_list(json_data) json_data = convert_back_raman_coef(json_data) # case of equipment json diff --git a/gnpy/tools/yang_convert_utils.py b/gnpy/tools/yang_convert_utils.py index c1750626f..eeb9242db 100644 --- a/gnpy/tools/yang_convert_utils.py +++ b/gnpy/tools/yang_convert_utils.py @@ -34,6 +34,8 @@ SPECTRUM_KEY = 'spectrum' LOSS_COEF_KEY = 'loss_coef' LOSS_COEF_KEY_PER_FREQ = 'loss_coef_per_frequency' +TOTAL_LOSS_KEY = 'total_loss' +TOTAL_LOSS_KEY_PER_FREQ = 'total_loss_per_frequency' RAMAN_COEF_KEY = 'raman_coefficient' RAMAN_EFFICIENCY_KEY = 'raman_efficiency' EQPT_TYPES = ['Edfa', 'Transceiver', 'Fiber', 'Roadm'] @@ -399,6 +401,46 @@ def convert_back_loss_coeff_list(json_data: Dict) -> Dict: return json_data +def convert_total_loss_list(json_data: Dict) -> Dict: + """Convert total_loss per-frequency dict format to YANG list format. + + :param json_data: The input JSON topology data to convert. + :type json_data: Dict + :return: the converted JSON data + :rtype: Dict + """ + for elem in json_data[ELEMENTS_KEY]: + if PARAMS_KEY in elem and TOTAL_LOSS_KEY in elem[PARAMS_KEY] \ + and isinstance(elem[PARAMS_KEY][TOTAL_LOSS_KEY], dict): + total_loss_per_frequency = elem[PARAMS_KEY].pop(TOTAL_LOSS_KEY) + total_loss_list = total_loss_per_frequency.pop('value', None) + frequency_list = total_loss_per_frequency.pop('frequency', None) + if total_loss_list: + new_total_loss_per_frequency = [{'frequency': f, 'total_loss_value': v} + for f, v in zip(frequency_list, total_loss_list)] + elem[PARAMS_KEY][TOTAL_LOSS_KEY_PER_FREQ] = new_total_loss_per_frequency + return json_data + + +def convert_back_total_loss_list(json_data: Dict) -> Dict: + """Convert YANG total_loss list format back to dict format. + + :param json_data: The input JSON topology data to convert back + :type json_data: Dict + :return: the converted JSON data + :rtype: Dict + """ + for elem in json_data[ELEMENTS_KEY]: + if PARAMS_KEY in elem and TOTAL_LOSS_KEY_PER_FREQ in elem[PARAMS_KEY]: + total_loss_per_frequency = elem[PARAMS_KEY].pop(TOTAL_LOSS_KEY_PER_FREQ) + if total_loss_per_frequency: + new_total_loss_per_frequency = { + 'frequency': [item['frequency'] for item in total_loss_per_frequency], + 'value': [item['total_loss_value'] for item in total_loss_per_frequency]} + elem[PARAMS_KEY][TOTAL_LOSS_KEY] = new_total_loss_per_frequency + return json_data + + def convert_design_band(json_data: Dict) -> Dict: """Convert legacy json topology format to gnpy yang format revision 2025-01-20: diff --git a/tests/test_parameters.py b/tests/test_parameters.py index b667c9429..a3cedab00 100644 --- a/tests/test_parameters.py +++ b/tests/test_parameters.py @@ -61,3 +61,83 @@ def test_fiber_parameters(): norm_gamma_raman_default_g0_no_ssmf = no_ssmf_fiber_params.raman_coefficient.normalized_gamma_raman assert_allclose(norm_gamma_raman_explicit_g0, norm_gamma_raman_default_g0_no_ssmf, rtol=1e-10) + + +def test_fiber_total_loss_scalar(): + """Test that total_loss parameter correctly computes loss_coef from total_loss / length.""" + fiber_dict = { + 'length': 80, + 'length_units': 'km', + 'att_in': 0, + 'con_in': 0.5, + 'con_out': 0.5, + 'dispersion': 1.67e-05, + 'effective_area': 8.3e-11, + 'pmd_coef': 1.265e-15, + 'total_loss': 16.0, # 16 dB total loss for 80 km → 0.2 dB/km + } + params = FiberParams(**fiber_dict) + # total_loss = 16 dB, length = 80 km → loss_coef should be 0.2 dB/km = 0.0002 dB/m + assert_allclose(params.loss_coef, 0.0002, rtol=1e-10) + assert_allclose(params.total_loss, 16.0, rtol=1e-10) + + +def test_fiber_total_loss_per_frequency(): + """Test that frequency-dependent total_loss correctly computes loss_coef per frequency.""" + fiber_dict = { + 'length': 80, + 'length_units': 'km', + 'att_in': 0, + 'con_in': 0.5, + 'con_out': 0.5, + 'dispersion': 1.67e-05, + 'effective_area': 8.3e-11, + 'pmd_coef': 1.265e-15, + 'total_loss': { + 'value': [16.0, 17.6], # dB at two frequencies + 'frequency': [191.35e12, 196.1e12], + }, + } + params = FiberParams(**fiber_dict) + # 16.0 / 80 = 0.2 dB/km = 0.0002 dB/m + # 17.6 / 80 = 0.22 dB/km = 0.00022 dB/m + assert_allclose(params.loss_coef, [0.0002, 0.00022], rtol=1e-10) + assert_allclose(params.total_loss, [16.0, 17.6], rtol=1e-10) + + +def test_fiber_total_loss_backward_compat(): + """Ensure loss_coef still works when total_loss is not provided.""" + fiber_dict = { + 'length': 80, + 'length_units': 'km', + 'att_in': 0, + 'con_in': 0.5, + 'con_out': 0.5, + 'dispersion': 1.67e-05, + 'effective_area': 8.3e-11, + 'pmd_coef': 1.265e-15, + 'loss_coef': 0.2, # dB/km + } + params = FiberParams(**fiber_dict) + assert_allclose(params.loss_coef, 0.0002, rtol=1e-10) + assert params.total_loss is None + + +def test_fiber_total_loss_asdict(): + """Test that asdict preserves total_loss when it was the original input.""" + fiber_dict = { + 'length': 80, + 'length_units': 'km', + 'att_in': 0, + 'con_in': 0.5, + 'con_out': 0.5, + 'dispersion': 1.67e-05, + 'effective_area': 8.3e-11, + 'pmd_coef': 1.265e-15, + 'total_loss': 16.0, + } + params = FiberParams(**fiber_dict) + d = params.asdict() + assert 'total_loss' in d + assert 'loss_coef' not in d + assert_allclose(d['total_loss'], 16.0, rtol=1e-10)