insert multiband_amplifier if needed

the automatic add_missing_elements function is updated to insert multiband booster, preamp or inline amplifiers based on the OMS nature. If nothing is stated, then uses Edfas. Signed-off-by: EstherLerouzic <[email protected]> Change-Id: I968a2fc0a3da97aecb7b513ff211491b20cdd4f2
Orange-OpenSource · Sep 30, 2024 · 9ed4ec6 · 9ed4ec6
1 parent c8c80ae
commit 9ed4ec6
Show file tree

Hide file tree

Showing 2 changed files with 266 additions and 48 deletions.
diff --git a/gnpy/core/network.py b/gnpy/core/network.py
@@ -23,7 +23,7 @@
 from gnpy.core.utils import round2float, convert_length, psd2powerdbm, lin2db, watt2dbm, dbm2watt, automatic_nch, \
  find_common_range
 from gnpy.core.info import ReferenceCarrier, create_input_spectral_information
-from gnpy.core.parameters import SimParams, EdfaParams, find_band_name, FrequencyBand
+from gnpy.core.parameters import SimParams, EdfaParams, find_band_name, FrequencyBand, MultiBandParams
 from gnpy.core.science_utils import RamanSolver
 
 
@@ -469,6 +469,26 @@ def get_oms_edge_list(oms_ingress_node: Union[elements.Roadm, elements.Transceiv
  return oms_edges
 
 
+def get_oms_edge_list_from_egress(oms_egress_node, network: DiGraph) -> List[Tuple]:
+ """get the list of OMS edges (node, neighbour next node) starting from its ingress down to its egress
+ oms_ingress_node can be a ROADM or a Transceiver
+ """
+ oms_edges = []
+ node = oms_egress_node
+ visited_nodes = []
+ # collect the OMS element list (ROADM to ROADM, or Transceiver to ROADM)
+ while not (isinstance(node, elements.Roadm) or isinstance(node, elements.Transceiver)):
+ previous_node = get_previous_node(node, network)
+ visited_nodes.append(node.uid)
+ if previous_node.uid in visited_nodes:
+ raise NetworkTopologyError(f'Loop detected for {type(node).__name__} {node.uid}, '
+ + 'please check network topology')
+ oms_edges.append((node, previous_node))
+ node = previous_node
+
+ return oms_edges
+
+
 def check_oms_single_type(oms_edges: List[Tuple]) -> List[str]:
  """Verifies that the OMS only contains all single band amplifiers or all multi band amplifiers
  No mixed OMS are permitted for the time being.
@@ -899,6 +919,10 @@ def set_egress_amplifier(network: DiGraph, this_node: Union[elements.Roadm, elem
  if isinstance(node, elements.Edfa):
  band_name, _ = next((n, b) for n, b in _design_bands.items())
  restrictions = get_node_restrictions(node, prev_node, next_node, equipment, _design_bands)
+ if not restrictions:
+ raise ConfigurationError(f'{node.uid}: Auto_design could not find any amplifier in equipment '
+ + f'library matching the design bands{_design_bands} '
+ + 'and the restrictions (roadm or amplifier restictions)')
  dp[band_name], voa[band_name] = set_one_amplifier(node, prev_node, next_node, power_mode,
  prev_voa[band_name], prev_dp[band_name],
  pref_ch_db, pref_total_db,
@@ -1203,21 +1227,39 @@ def add_roadm_booster(network, roadm):
  # no amplification for fused spans or TRX
  for next_node in next_nodes:
  network.remove_edge(roadm, next_node)
- amp = elements.Edfa(
- uid=f'Edfa_booster_{roadm.uid}_to_{next_node.uid}',
- params=EdfaParams.default_values,
- metadata={
- 'location': {
- 'latitude': roadm.lat,
- 'longitude': roadm.lng,
- 'city': roadm.loc.city,
- 'region': roadm.loc.region,
- }
- },
- operational={
- 'gain_target': None,
- 'tilt_target': 0,
- })
+ oms_edges = get_oms_edge_list(next_node, network)
+ amps_type = check_oms_single_type(oms_edges)
+ if 'Multiband_amplifier' in amps_type or ('Edfa' not in amps_type and len(roadm.design_bands) > 1):
+ amp = elements.Multiband_amplifier(
+ uid=f'Edfa_booster_{roadm.uid}_to_{next_node.uid}',
+ params=MultiBandParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': roadm.lat,
+ 'longitude': roadm.lng,
+ 'city': roadm.loc.city,
+ 'region': roadm.loc.region,
+ }
+ },
+ amplifiers=[])
+ else:
+ # if 'Edfa' or no amplifier type is set in the OMS, then assumes single band
+ amp = elements.Edfa(
+ uid=f'Edfa_booster_{roadm.uid}_to_{next_node.uid}',
+ params=EdfaParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': roadm.lat,
+ 'longitude': roadm.lng,
+ 'city': roadm.loc.city,
+ 'region': roadm.loc.region,
+ }
+ },
+ operational={
+ 'gain_target': None,
+ 'tilt_target': 0,
+ })
+
  network.add_node(amp)
  network.add_edge(roadm, amp, weight=0.01)
  network.add_edge(amp, next_node, weight=0.01)
@@ -1230,21 +1272,37 @@ def add_roadm_preamp(network, roadm):
  # no amplification for fused spans or TRX
  for prev_node in prev_nodes:
  network.remove_edge(prev_node, roadm)
- amp = elements.Edfa(
- uid=f'Edfa_preamp_{roadm.uid}_from_{prev_node.uid}',
- params=EdfaParams.default_values,
- metadata={
- 'location': {
- 'latitude': roadm.lat,
- 'longitude': roadm.lng,
- 'city': roadm.loc.city,
- 'region': roadm.loc.region,
- }
- },
- operational={
- 'gain_target': None,
- 'tilt_target': 0,
- })
+ oms_edges = get_oms_edge_list_from_egress(prev_node, network)
+ amps_type = check_oms_single_type(oms_edges)
+ if 'Multiband_amplifier' in amps_type:
+ amp = elements.Multiband_amplifier(
+ uid=f'Edfa_preamp_{roadm.uid}_from_{prev_node.uid}',
+ params=MultiBandParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': roadm.lat,
+ 'longitude': roadm.lng,
+ 'city': roadm.loc.city,
+ 'region': roadm.loc.region,
+ }
+ },
+ amplifiers=[])
+ else:
+ amp = elements.Edfa(
+ uid=f'Edfa_preamp_{roadm.uid}_from_{prev_node.uid}',
+ params=EdfaParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': roadm.lat,
+ 'longitude': roadm.lng,
+ 'city': roadm.loc.city,
+ 'region': roadm.loc.region,
+ }
+ },
+ operational={
+ 'gain_target': None,
+ 'tilt_target': 0,
+ })
  network.add_node(amp)
  if isinstance(prev_node, elements.Fiber):
  edgeweight = prev_node.params.length
@@ -1259,21 +1317,37 @@ def add_inline_amplifier(network, fiber):
  if isinstance(next_node, elements.Fiber) or isinstance(next_node, elements.RamanFiber):
  # no amplification for fused spans or TRX
  network.remove_edge(fiber, next_node)
- amp = elements.Edfa(
- uid=f'Edfa_{fiber.uid}',
- params=EdfaParams.default_values,
- metadata={
- 'location': {
- 'latitude': (fiber.lat + next_node.lat) / 2,
- 'longitude': (fiber.lng + next_node.lng) / 2,
- 'city': fiber.loc.city,
- 'region': fiber.loc.region,
- }
- },
- operational={
- 'gain_target': None,
- 'tilt_target': 0,
- })
+ oms_edges = get_oms_edge_list(next_node, network)
+ amps_type = check_oms_single_type(oms_edges)
+ if 'Multiband_amplifier' in amps_type:
+ amp = elements.Multiband_amplifier(
+ uid=f'Edfa_{fiber.uid}',
+ params=MultiBandParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': (fiber.lat + next_node.lat) / 2,
+ 'longitude': (fiber.lng + next_node.lng) / 2,
+ 'city': fiber.loc.city,
+ 'region': fiber.loc.region,
+ }
+ },
+ amplifiers=[])
+ else:
+ amp = elements.Edfa(
+ uid=f'Edfa_{fiber.uid}',
+ params=EdfaParams.default_values,
+ metadata={
+ 'location': {
+ 'latitude': (fiber.lat + next_node.lat) / 2,
+ 'longitude': (fiber.lng + next_node.lng) / 2,
+ 'city': fiber.loc.city,
+ 'region': fiber.loc.region,
+ }
+ },
+ operational={
+ 'gain_target': None,
+ 'tilt_target': 0,
+ })
  network.add_node(amp)
  network.add_edge(fiber, amp, weight=fiber.params.length)
  network.add_edge(amp, next_node, weight=0.01)
@@ -1313,6 +1387,17 @@ def get_next_node(node, network):
  f'{type(node).__name__} {node.uid} is not properly connected, please check network topology')
 
 
+def get_previous_node(node, network):
+ """get previous node else raise the appropriate error
+ """
+ try:
+ previous_node = next(network.predecessors(node))
+ return previous_node
+ except StopIteration:
+ raise NetworkTopologyError(
+ f'{type(node).__name__} {node.uid} is not properly connected, please check network topology')
+
+
 def split_fiber(network, fiber, bounds, target_length):
  """If fiber length exceeds boundary then assume this is a link "intent", and replace this one-span link
  with an n_spans link, with identical fiber types.

diff --git a/tests/test_network_functions.py b/tests/test_network_functions.py
@@ -8,9 +8,9 @@
 import pytest
 from numpy.testing import assert_allclose
 
-from gnpy.core.exceptions import NetworkTopologyError
+from gnpy.core.exceptions import NetworkTopologyError, ConfigurationError
 from gnpy.core.network import span_loss, build_network, select_edfa, get_node_restrictions, \
- estimate_srs_power_deviation
+ estimate_srs_power_deviation, add_missing_elements_in_network, get_next_node
 from gnpy.tools.json_io import load_equipment, load_network, network_from_json, load_json
 from gnpy.core.utils import lin2db, automatic_nch, merge_amplifier_restrictions
 from gnpy.core.elements import Fiber, Edfa, Roadm, Multiband_amplifier
@@ -614,3 +614,136 @@ def test_tilt_fused():
  SimParams.set_params(save_sim_params)
  assert fused_tilt_db == tilt_db
  assert fused_tilt_target == tilt_target
+
+
+def network_wo_booster(site_type, bands):
+ return {
+ 'elements': [
+ {
+ 'uid': 'trx SITE1',
+ 'type': 'Transceiver'
+ },
+ {
+ 'uid': 'trx SITE2',
+ 'type': 'Transceiver'
+ },
+ {
+ 'uid': 'roadm SITE1',
+ 'params': {
+ 'design_bands': bands
+ },
+ 'type': 'Roadm'
+ },
+ {
+ 'uid': 'roadm SITE2',
+ 'type': 'Roadm'
+ },
+ {
+ 'uid': 'fiber (SITE1 → ILA1)',
+ 'type': 'Fiber',
+ 'type_variety': 'SSMF',
+ 'params': {
+ 'length': 50.0,
+ 'loss_coef': 0.2,
+ 'length_units': 'km'
+ }
+ },
+ {
+ 'uid': 'fiber (ILA1 → ILA2)',
+ 'type': 'Fiber',
+ 'type_variety': 'SSMF',
+ 'params': {
+ 'length': 50.0,
+ 'loss_coef': 0.2,
+ 'length_units': 'km'
+ }
+ },
+ {
+ 'uid': 'fiber (ILA2 → SITE2)',
+ 'type': 'Fiber',
+ 'type_variety': 'SSMF',
+ 'params': {
+ 'length': 50.0,
+ 'loss_coef': 0.2,
+ 'length_units': 'km'
+ }
+ },
+ {
+ 'uid': 'east edfa or fused in ILA1',
+ 'type': site_type
+ }
+ ],
+ 'connections': [
+ {
+ 'from_node': 'trx SITE1',
+ 'to_node': 'roadm SITE1'
+ },
+ {
+ 'from_node': 'roadm SITE1',
+ 'to_node': 'fiber (SITE1 → ILA1)'
+ },
+ {
+ 'from_node': 'fiber (SITE1 → ILA1)',
+ 'to_node': 'east edfa or fused in ILA1'
+ },
+ {
+ 'from_node': 'east edfa or fused in ILA1',
+ 'to_node': 'fiber (ILA1 → ILA2)'
+ },
+ {
+ 'from_node': 'fiber (ILA1 → ILA2)',
+ 'to_node': 'fiber (ILA2 → SITE2)'
+ },
+ {
+ 'from_node': 'fiber (ILA2 → SITE2)',
+ 'to_node': 'roadm SITE2'
+ },
+ {
+ 'from_node': 'roadm SITE2',
+ 'to_node': 'trx SITE2'
+ }
+ ]
+ }
+
+
+@pytest.mark.parametrize('site_type, expected_type, bands, expected_bands', [
+ ('Multiband_amplifier', Multiband_amplifier,
+ [{'f_min': 187.0e12, 'f_max': 190.0e12}, {'f_min': 191.3e12, 'f_max': 196.0e12}],
+ [{'f_min': 187.0e12, 'f_max': 190.0e12}, {'f_min': 191.3e12, 'f_max': 196.0e12}]),
+ ('Edfa', Edfa,
+ [{'f_min': 191.4e12, 'f_max': 196.1e12}],
+ [{'f_min': 191.4e12, 'f_max': 196.1e12}]),
+ ('Edfa', Edfa,
+ [{'f_min': 191.2e12, 'f_max': 196.0e12}],
+ []),
+ ('Fused', Multiband_amplifier,
+ [{'f_min': 187.0e12, 'f_max': 190.0e12}, {'f_min': 191.3e12, 'f_max': 196.0e12}],
+ [{'f_min': 187.0e12, 'f_max': 190.0e12}, {'f_min': 191.3e12, 'f_max': 196.0e12}]),
+ ('Fused', Edfa,
+ [{'f_min': 191.3e12, 'f_max': 196.0e12}],
+ [{'f_min': 191.3e12, 'f_max': 196.0e12}])])
+def test_insert_amp(site_type, expected_type, bands, expected_bands):
+ """Check:
+ - if amplifiers are defined in multiband they are used for design,
+ - if no design is defined,
+ - if type variety is defined: use it for determining bands
+ - if no type_variety autodesign is as expected, design uses OMS defined set of bands
+ EOL is added only once on spans. One span can be one fiber or several fused fibers
+ EOL is then added on the first fiber only.
+ """
+ json_data = network_wo_booster(site_type, bands)
+ equipment = load_equipment(EQPT_MULTBAND_FILENAME)
+ network = network_from_json(json_data, equipment)
+ p_db = equipment['SI']['default'].power_dbm
+ p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min,
+ equipment['SI']['default'].f_max, equipment['SI']['default'].spacing))
+ add_missing_elements_in_network(network, equipment)
+ if not expected_bands:
+ with pytest.raises(ConfigurationError):
+ build_network(network, equipment, p_db, p_total_db)
+ else:
+ build_network(network, equipment, p_db, p_total_db)
+ roadm1 = next(n for n in network.nodes() if n.uid == 'roadm SITE1')
+ amp1 = get_next_node(roadm1, network)
+ assert isinstance(amp1, expected_type)
+ assert roadm1.per_degree_design_bands['Edfa_booster_roadm SITE1_to_fiber (SITE1 → ILA1)'] == expected_bands