feat: skip path computation when path is explicit

and add tests for explicit path

Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Change-Id: I95aaf5b56a7ea04f24153d5cb6612cd09401401c

gnpy/core/utils.py

@@ -444,6 +444,16 @@ def restore_order(elements, order):
     return [elements[i[0]] for i in sorted(enumerate(order), key=lambda x:x[1]) if elements[i[0]] is not None]
 
 
+def unique_ordered(elements):
+    """
+    """
+    unique_elements = []
+    for element in elements:
+        if element not in unique_elements:
+            unique_elements.append(element)
+    return unique_elements
+
+
 def calculate_absolute_min_or_zero(x: array) -> array:
     """Calculates the element-wise absolute minimum between the x and zero.
 

gnpy/topology/request.py

@@ -24,7 +24,7 @@ from networkx.utils import pairwise
 from numpy import mean, argmin
 
 from gnpy.core.elements import Transceiver, Roadm, Edfa, Multiband_amplifier
-from gnpy.core.utils import lin2db, find_common_range
+from gnpy.core.utils import lin2db, unique_ordered, find_common_range
 from gnpy.core.info import create_input_spectral_information, carriers_to_spectral_information, \
     demuxed_spectral_information, muxed_spectral_information, SpectralInformation
 from gnpy.core import network as network_module
@@ -301,7 +301,9 @@ def compute_constrained_path(network, req):
     nodes_list = []
     for node in req.nodes_list[:-1]:
         nodes_list.append(next(el for el in network if el.uid == node))
-
+    total_path = explicit_path(nodes_list, source, destination, network)
+    if total_path is not None:
+        return total_path
     try:
         path_generator = shortest_simple_paths(network, source, destination, weight='weight')
         total_path = next(path for path in path_generator if ispart(nodes_list, path))
@@ -1253,6 +1255,44 @@ def _penalty_msg(total_path, msg, min_ind):
     return msg
 
 
+def is_adjacent(oms1, oms2):
+    """ oms1's egress ROADM is oms2's ingress ROADM
+    """
+    return oms1.el_list[-1] == oms2.el_list[0]
+
+
+def explicit_path(node_list, source, destination, network):
+    """ if list of nodes leads to adjacent oms, then means that the path is explicit, and no need to compute
+    the function returns the explicit path (including source and destination ROADMs)
+    """
+    path_oms = []
+    for elem in node_list:
+        if hasattr(elem, 'oms'):
+            path_oms.append(elem.oms)
+    if not path_oms:
+        return None
+    path_oms = unique_ordered(path_oms)
+    try:
+        next_node = next(network.successors(source))
+        source_roadm = next_node if isinstance(next_node, Roadm) else source
+        previous_node = next(network.predecessors(destination))
+        destination_roadm = previous_node if isinstance(previous_node, Roadm) else destination
+        if not (path_oms[0].el_list[0] == source_roadm and path_oms[-1].el_list[-1] == destination_roadm):
+            return None
+    except StopIteration:
+        return None
+
+    oms0 = path_oms[0]
+    path = [source] + oms0.el_list
+    for oms in path_oms[1:]:
+        if not is_adjacent(oms0, oms):
+            return None
+        oms0 = oms
+        path.extend(oms.el_list)
+    path.append(destination)
+    return unique_ordered(path)
+
+
 def find_elements_common_range(el_list: list, equipment: dict) -> List[dict]:
     """Find the common frequency range of amps of a given list of elements (for example an OMS or a path)
     If there are no amplifiers in the path, then use the SI

tests/test_path_computation_functions.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+# Module name: test_path_computation_functions.py
+# Version:
+# License: BSD 3-Clause Licence
+# Copyright (c) 2018, Telecom Infra Project
+
+"""
+@author: esther.lerouzic
+
+"""
+
+from pathlib import Path
+import pytest
+from gnpy.core.network import build_network
+from gnpy.core.utils import lin2db, automatic_nch
+from gnpy.topology.request import explicit_path
+from gnpy.topology.spectrum_assignment import build_oms_list
+from gnpy.tools.json_io import load_equipment, load_network, requests_from_json
+
+
+TEST_DIR = Path(__file__).parent
+DATA_DIR = TEST_DIR / 'data'
+EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
+NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json'
+SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json'
+equipment = load_equipment(EQPT_FILENAME)
+
+
+@pytest.fixture()
+def setup_without_oms():
+    """ common setup for tests: builds network, equipment and oms only once
+    """
+    network = load_network(NETWORK_FILENAME, equipment)
+    spectrum = equipment['SI']['default']
+    p_db = spectrum.power_dbm
+    p_total_db = p_db + lin2db(automatic_nch(spectrum.f_min, spectrum.f_max, spectrum.spacing))
+    build_network(network, equipment, p_db, p_total_db)
+    return network
+
+
+def some_request(explicit_route):
+    """Create a request with an explicit route
+    """
+    route = {
+        "route-object-include-exclude": [
+            {
+                "explicit-route-usage": "route-include-ero",
+                "index": i,
+                "num-unnum-hop": {
+                    "node-id": node_id,
+                    "link-tp-id": "link-tp-id is not used",
+                    "hop-type": "STRICT"
+                }
+            } for i, node_id in enumerate(explicit_route)
+        ]
+    }
+    return {
+        "path-request": [{
+            "request-id": "2",
+            "source": explicit_route[0],
+            "destination": explicit_route[-1],
+            "src-tp-id": explicit_route[0],
+            "dst-tp-id": explicit_route[-1],
+            "bidirectional": False,
+            "path-constraints": {
+                "te-bandwidth": {
+                    "technology": "flexi-grid",
+                    "trx_type": "Voyager",
+                    "trx_mode": "mode 1",
+                    "spacing": 75000000000.0,
+                    "path_bandwidth": 100000000000.0
+                }
+            },
+            "explicit-route-objects": route}
+        ]
+    }
+
+
+@pytest.mark.parametrize('setup', ["with_oms", "without_oms"])
+@pytest.mark.parametrize('explicit_route, expected_path', [
+    (['trx Brest_KLA', 'trx Vannes_KBE'], None),
+    # path contains one element per oms
+    (['trx Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'trx Lannion_CAS'],
+     ['trx Brest_KLA', 'roadm Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'fiber (Brest_KLA → Morlaix)-F060',
+      'east fused spans in Morlaix', 'fiber (Morlaix → Lannion_CAS)-F059', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'trx Lannion_CAS']),
+    # path contains several elements per oms
+    (['trx Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'trx Lannion_CAS'],
+     ['trx Brest_KLA', 'roadm Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'fiber (Brest_KLA → Morlaix)-F060',
+      'east fused spans in Morlaix', 'fiber (Morlaix → Lannion_CAS)-F059', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'trx Lannion_CAS']),
+    # path contains all elements
+    (['trx Brest_KLA', 'roadm Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'fiber (Brest_KLA → Morlaix)-F060',
+      'east fused spans in Morlaix', 'fiber (Morlaix → Lannion_CAS)-F059', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'trx Lannion_CAS'],
+     ['trx Brest_KLA', 'roadm Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'fiber (Brest_KLA → Morlaix)-F060',
+      'east fused spans in Morlaix', 'fiber (Morlaix → Lannion_CAS)-F059', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'trx Lannion_CAS']),
+    # path conteains element for only 1 oms (2 oms path)
+    (['trx Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'trx Rennes_STA'], None),
+    # path contains roadm edges for all OMS, but no element of the OMS
+    (['trx Brest_KLA', 'roadm Brest_KLA', 'roadm Lannion_CAS', 'trx Lannion_CAS'], None),
+    # path contains one element for all 3 OMS
+    (['trx Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'east edfa in Lannion_CAS to Corlay',
+      'east edfa in Lorient_KMA to Vannes_KBE', 'trx Vannes_KBE'],
+     ['trx Brest_KLA', 'roadm Brest_KLA', 'east edfa in Brest_KLA to Morlaix', 'fiber (Brest_KLA → Morlaix)-F060',
+      'east fused spans in Morlaix', 'fiber (Morlaix → Lannion_CAS)-F059', 'west edfa in Lannion_CAS to Morlaix',
+      'roadm Lannion_CAS', 'east edfa in Lannion_CAS to Corlay', 'fiber (Lannion_CAS → Corlay)-F061',
+      'west fused spans in Corlay', 'fiber (Corlay → Loudeac)-F010', 'west fused spans in Loudeac',
+      'fiber (Loudeac → Lorient_KMA)-F054', 'west edfa in Lorient_KMA to Loudeac', 'roadm Lorient_KMA',
+      'east edfa in Lorient_KMA to Vannes_KBE', 'fiber (Lorient_KMA → Vannes_KBE)-F055',
+      'west edfa in Vannes_KBE to Lorient_KMA', 'roadm Vannes_KBE', 'trx Vannes_KBE'])])
+def test_explicit_path(setup, setup_without_oms, explicit_route, expected_path):
+    """tests that explicit path correctly returns the full path if it is possible else that it returns None
+    """
+    network = setup_without_oms
+    if setup == "with_oms":
+        # OMS are initiated in elements, so that explicit path can be verified
+        build_oms_list(network, equipment)
+    else:
+        # OMS are not initiated, explicit path can not be computed.
+        expected_path = None
+    json_data = some_request(explicit_route)
+    [req] = requests_from_json(json_data, equipment)
+
+    node_list = []
+    for node in req.nodes_list:
+        node_list.append(next(el for el in network if el.uid == node))
+    source = node_list[0]
+    destination = node_list[-1]
+    if expected_path is None:
+        assert explicit_path(node_list, source, destination, network) is None
+    else:
+        actual_path = [e.uid for e in explicit_path(node_list, source, destination, network)]
+        assert actual_path == expected_path