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

Author: LynchXLQ

AUTHORS.rst

@@ -18,6 +18,7 @@ To learn how to contribute, please see CONTRIBUTING.md
 - Gabriele Galimberti (Cisco) <ggalimbe@cisco.com>
 - Gert Grammel (Juniper Networks) <ggrammel@juniper.net>
 - Giacomo Borraccini (NEC Laboratories America) <gborraccini@nec-labs.com>
+- Linqi Xiao (University of Texas at Dallas) <linqi.xiao@utdallas.edu>, <linqixiao2021@gmail.com>, <linqi.xiao@ieee.org>
 - Gilad Goldfarb (Facebook) <giladg@fb.com>
 - James Powell (Telecom Infra Project) <james.powell@telecominfraproject.com>
 - Jan Kundrát (Telecom Infra Project) <jkt@jankundrat.com>

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)]

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')

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,13 +57,15 @@ 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:
         # then this is a new format topology json, ensure that there are no issues
         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

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:
 

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)