PathComp component - FrontEnd:

import logging, sys

import logging, sys

from common.proto.context_pb2 import ServiceTypeEnum

from common.proto.context_pb2 import ServiceTypeEnum

from pathcomp.frontend.service.algorithms.tools.ComputeSubServices import convert_explicit_path_hops_to_connections

from .data import path_hops, device_dict

from .data import path_hops, device_dict

from .ComputeSubServices import convert_explicit_path_hops_to_connections

logging.basicConfig(level=logging.DEBUG)

logging.basicConfig(level=logging.DEBUG)

LOGGER = logging.getLogger(__name__)

LOGGER = logging.getLogger(__name__)

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

# Convert the path defined as explicit hops with ingress and egress endpoints per device into a set of connections and

# compute the dependencies among them.

#

# Example:

# o-- int DC1 eth1 -- 10/1 CS1 1/2 -- 1/2 R2 2/1 -- a7.. OLS 60.. -- 2/1 R3 1/1 -- 1/1 CS2 10/1 -- eth1 DC2 int --o

#         APP              PKT            PKT            CTRL            PKT           PKT              APP

#

# path_hops = [

#     {'device': 'DC1-GW', 'ingress_ep': 'int', 'egress_ep': 'eth1'},

#     {'device': 'CS1-GW1', 'ingress_ep': '10/1', 'egress_ep': '1/2'},

#     {'device': 'TN-R2', 'ingress_ep': '1/2', 'egress_ep': '2/1'},

#     {'device': 'TN-OLS', 'ingress_ep': 'a7a80b23a703', 'egress_ep': '60519106029e'},

#     {'device': 'TN-R3', 'ingress_ep': '2/1', 'egress_ep': '1/1'},

#     {'device': 'CS2-GW1', 'ingress_ep': '1/1', 'egress_ep': '10/1'},

#     {'device': 'DC2-GW', 'ingress_ep': 'eth1', 'egress_ep': 'int'}

# ]

#

# connections=[

#     (UUID('7548edf7-ee7c-4adf-ac0f-c7a0c0dfba8e'), ServiceTypeEnum.SERVICETYPE_TAPI_CONNECTIVITY_SERVICE, [

#             {'device': 'TN-OLS', 'ingress_ep': '833760219d0f', 'egress_ep': 'cf176771a4b9'}

#         ], []),

#     (UUID('c2e57966-5d82-4705-a5fe-44cf6487219e'), ServiceTypeEnum.SERVICETYPE_L2NM, [

#             {'device': 'CS1-GW1', 'ingress_ep': '10/1', 'egress_ep': '1/2'},

#             {'device': 'TN-R2', 'ingress_ep': '1/2', 'egress_ep': '2/1'},

#             {'device': 'TN-R3', 'ingress_ep': '2/1', 'egress_ep': '1/1'},

#             {'device': 'CS2-GW1', 'ingress_ep': '1/1', 'egress_ep': '10/1'}

#         ], [UUID('7548edf7-ee7c-4adf-ac0f-c7a0c0dfba8e')]),

#     (UUID('1e205c82-f6ea-4977-9e97-dc27ef1f4802'), ServiceTypeEnum.SERVICETYPE_L2NM, [

#             {'device': 'DC1-GW', 'ingress_ep': 'int', 'egress_ep': 'eth1'},

#             {'device': 'DC2-GW', 'ingress_ep': 'eth1', 'egress_ep': 'int'}

#         ], [UUID('c2e57966-5d82-4705-a5fe-44cf6487219e')])

# ]

import logging, queue, uuid

from typing import Dict, List, Optional, Tuple

from common.DeviceTypes import DeviceTypeEnum

from common.proto.context_pb2 import Device, ServiceTypeEnum

from .ResourceGroups import IGNORED_DEVICE_TYPES, get_resource_classification

from .ServiceTypes import get_service_type

LOGGER = logging.getLogger(__name__)

def convert_explicit_path_hops_to_connections(

    path_hops : List[Dict], device_dict : Dict[str, Tuple[Dict, Device]],

    main_service_uuid : str, main_service_type : ServiceTypeEnum

) -> List[Tuple[str, int, List[str], List[str]]]:

    LOGGER.debug('path_hops={:s}'.format(str(path_hops)))

    connection_stack = queue.LifoQueue()

    connections : List[Tuple[str, int, List[str], List[str]]] = list()

    prv_device_uuid = None

    prv_res_class : Tuple[Optional[int], Optional[DeviceTypeEnum], Optional[str]] = None, None, None

    for path_hop in path_hops:

        device_uuid = path_hop['device']

        if prv_device_uuid == device_uuid: continue

        device_tuple = device_dict.get(device_uuid)

        if device_tuple is None: raise Exception('Device({:s}) not found'.format(str(device_uuid)))

        _,grpc_device = device_tuple

        res_class = get_resource_classification(grpc_device, device_dict)

        if res_class[1] in IGNORED_DEVICE_TYPES: continue

        if prv_res_class[0] is None:

            # path ingress

            connection_stack.put((main_service_uuid, main_service_type, [path_hop], []))

        elif prv_res_class[0] > res_class[0]:

            # create underlying connection

            connection_uuid = str(uuid.uuid4())

            prv_service_type = connection_stack.queue[-1][1]

            service_type = get_service_type(res_class[1], prv_service_type)

            connection_stack.put((connection_uuid, service_type, [path_hop], []))

        elif prv_res_class[0] == res_class[0]:

            # same resource group kind

            if prv_res_class[1] == res_class[1] and prv_res_class[2] == res_class[2]:

                # same device type and device controller: connection continues

                connection_stack.queue[-1][2].append(path_hop)

            else:

                # different device type or device controller: chain connections

                connection = connection_stack.get()

                connections.append(connection)

                connection_stack.queue[-1][3].append(connection[0])

                connection_uuid = str(uuid.uuid4())

                prv_service_type = connection_stack.queue[-1][1]

                service_type = get_service_type(res_class[1], prv_service_type)

                connection_stack.put((connection_uuid, service_type, [path_hop], []))

        elif prv_res_class[0] < res_class[0]:

            # underlying connection ended

            connection = connection_stack.get()

            connections.append(connection)

            connection_stack.queue[-1][3].append(connection[0])

            connection_stack.queue[-1][2].append(path_hop)

        else:

            raise Exception('Uncontrolled condition')

        prv_device_uuid = device_uuid

        prv_res_class = res_class

    # path egress

    connections.append(connection_stack.get())

    LOGGER.debug('connections={:s}'.format(str(connections)))

    assert connection_stack.empty()

    return connections

def convert_explicit_path_hops_to_plain_connection(

    path_hops : List[Dict], main_service_uuid : str, main_service_type : ServiceTypeEnum

) -> List[Tuple[str, int, List[str], List[str]]]:

    connection : Tuple[str, int, List[str], List[str]] = \

        (main_service_uuid, main_service_type, [], [])

    prv_device_uuid = None

    for path_hop in path_hops:

        device_uuid = path_hop['device']

        if prv_device_uuid == device_uuid: continue

        connection[2].append(path_hop)

        prv_device_uuid = device_uuid

    return [connection]

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import json

from typing import Dict, Optional, Tuple

from common.DeviceTypes import DeviceTypeEnum

from common.proto.context_pb2 import Device

from common.tools.grpc.Tools import grpc_message_to_json_string

DEVICE_TYPE_TO_DEEPNESS = {

    DeviceTypeEnum.EMULATED_DATACENTER.value             : 90,

    DeviceTypeEnum.DATACENTER.value                      : 90,

    DeviceTypeEnum.NETWORK.value                         : 90,

    DeviceTypeEnum.TERAFLOWSDN_CONTROLLER.value          : 80,

    DeviceTypeEnum.EMULATED_PACKET_ROUTER.value          : 70,

    DeviceTypeEnum.PACKET_ROUTER.value                   : 70,

    DeviceTypeEnum.EMULATED_PACKET_SWITCH.value          : 60,

    DeviceTypeEnum.PACKET_SWITCH.value                   : 60,

    DeviceTypeEnum.EMULATED_P4_SWITCH.value              : 60,

    DeviceTypeEnum.P4_SWITCH.value                       : 60,

    DeviceTypeEnum.EMULATED_MICROWAVE_RADIO_SYSTEM.value : 40,

    DeviceTypeEnum.MICROWAVE_RADIO_SYSTEM.value          : 40,

    DeviceTypeEnum.EMULATED_XR_CONSTELLATION.value       : 40,

    DeviceTypeEnum.XR_CONSTELLATION.value                : 40,

    DeviceTypeEnum.EMULATED_OPEN_LINE_SYSTEM.value       : 30,

    DeviceTypeEnum.OPEN_LINE_SYSTEM.value                : 30,

    DeviceTypeEnum.EMULATED_PACKET_RADIO_ROUTER.value    : 10,

    DeviceTypeEnum.PACKET_RADIO_ROUTER.value             : 10,

    DeviceTypeEnum.EMULATED_OPTICAL_TRANSPONDER.value    : 10,

    DeviceTypeEnum.OPTICAL_TRANSPONDER.value             : 10,

    DeviceTypeEnum.EMULATED_OPTICAL_ROADM.value          : 10,

    DeviceTypeEnum.OPTICAL_ROADM.value                   : 10,

    DeviceTypeEnum.EMULATED_OPTICAL_SPLITTER.value       :  0,

}

IGNORED_DEVICE_TYPES = {DeviceTypeEnum.EMULATED_OPTICAL_SPLITTER}

def get_device_controller_uuid(

    device : Device

) -> Optional[str]:

    for config_rule in device.device_config.config_rules:

        if config_rule.WhichOneof('config_rule') != 'custom': continue

        if config_rule.custom.resource_key != '_controller': continue

        device_controller_id = json.loads(config_rule.custom.resource_value)

        return device_controller_id['uuid']

    return None

def _map_device_type(device : Device) -> DeviceTypeEnum:

    device_type = DeviceTypeEnum._value2member_map_.get(device.device_type) # pylint: disable=no-member

    if device_type is None:

        MSG = 'Unsupported DeviceType({:s}) for Device({:s})'

        raise Exception(MSG.format(str(device.device_type), grpc_message_to_json_string(device)))

    return device_type

def _map_resource_to_deepness(device_type : DeviceTypeEnum) -> int:

    deepness = DEVICE_TYPE_TO_DEEPNESS.get(device_type.value)

    if deepness is None: raise Exception('Unsupported DeviceType({:s})'.format(str(device_type.value)))

    return deepness

def get_device_type(

    device : Device, device_dict : Dict[str, Tuple[Dict, Device]], device_controller_uuid : Optional[str]

) -> DeviceTypeEnum:

    if device_controller_uuid is None: return _map_device_type(device)

    device_controller_tuple = device_dict.get(device_controller_uuid)

    if device_controller_tuple is None: raise Exception('Device({:s}) not found'.format(str(device_controller_uuid)))

    _,device = device_controller_tuple

    return _map_device_type(device)

def get_resource_classification(

    device : Device, device_dict : Dict[str, Tuple[Dict, Device]]

) -> Tuple[int, DeviceTypeEnum, Optional[str]]:

    device_controller_uuid = get_device_controller_uuid(device)

    device_type = get_device_type(device, device_dict, device_controller_uuid)

    resource_deepness = _map_resource_to_deepness(device_type)

    return resource_deepness, device_type, device_controller_uuid