First version

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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.

from typing import Dict, Tuple

from common.proto.context_pb2 import Device, DeviceId, EndPoint, EndPointId

class NameMappings:

    def __init__(self) -> None:

        self._device_uuid_to_name   : Dict[str,             str] = dict()

        self._endpoint_uuid_to_name : Dict[Tuple[str, str], str] = dict()

    def store_device_name(self, device : Device) -> None:

        device_uuid = device.device_id.device_uuid.uuid

        device_name = device.name

        self._device_uuid_to_name[device_uuid] = device_name

        self._device_uuid_to_name[device_name] = device_name

    def store_endpoint_name(self, device : Device, endpoint : EndPoint) -> None:

        device_uuid = device.device_id.device_uuid.uuid

        device_name = device.name

        endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

        endpoint_name = endpoint.name

        self._endpoint_uuid_to_name[(device_uuid, endpoint_uuid)] = endpoint_name

        self._endpoint_uuid_to_name[(device_name, endpoint_uuid)] = endpoint_name

        self._endpoint_uuid_to_name[(device_uuid, endpoint_name)] = endpoint_name

        self._endpoint_uuid_to_name[(device_name, endpoint_name)] = endpoint_name

    def get_device_name(self, device_id : DeviceId) -> str:

        device_uuid = device_id.device_uuid.uuid

        return self._device_uuid_to_name.get(device_uuid, device_uuid)

    def get_endpoint_name(self, endpoint_id : EndPointId) -> str:

        device_uuid = endpoint_id.device_id.device_uuid.uuid

        endpoint_uuid = endpoint_id.endpoint_uuid.uuid

        return self._endpoint_uuid_to_name.get((device_uuid, endpoint_uuid), endpoint_uuid)

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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.

from enum import Enum

from typing import Optional

class NetworkTypeEnum(Enum):

    TE_OTN_TOPOLOGY      = 'otn'

    TE_ETH_TRAN_TOPOLOGY = 'eth-tran'

def get_network_topology_type(topology_id : str) -> Optional[NetworkTypeEnum]:

    return NetworkTypeEnum._value2member_map_.get(topology_id)

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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, logging

from flask import request

from flask.json import jsonify

from flask_restful import Resource

from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME

from common.tools.context_queries.Topology import get_topology_details

from context.client.ContextClient import ContextClient

from nbi.service.rest_server.nbi_plugins.tools.Authentication import HTTP_AUTH

from nbi.service.rest_server.nbi_plugins.tools.HttpStatusCodes import HTTP_OK, HTTP_SERVERERROR

from .YangHandler import YangHandler

from .ManualFixes import manual_fixes

LOGGER = logging.getLogger(__name__)

TE_TOPOLOGY_NAMES = [

    'providerId-10-clientId-0-topologyId-1',

    'providerId-10-clientId-0-topologyId-2'

]

class Networks(Resource):

    @HTTP_AUTH.login_required

    def get(self):

        LOGGER.info('Request: {:s}'.format(str(request)))

        topology_id = ''

        try:

            context_client = ContextClient()

            #target = get_slice_by_uuid(context_client, vpn_id, rw_copy=True)

            #if target is None:

            #    raise Exception('VPN({:s}) not found in database'.format(str(vpn_id)))

            topology_details = get_topology_details(

                context_client, DEFAULT_TOPOLOGY_NAME, context_uuid=DEFAULT_CONTEXT_NAME, #rw_copy=True

            )

            if topology_details is None:

                MSG = 'Topology({:s}/{:s}) not found'

                raise Exception(MSG.format(DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME))

            network_list_reply = []

            yang_handler = YangHandler()

            for te_topology_name in TE_TOPOLOGY_NAMES:

                network_reply = yang_handler.compose_network(te_topology_name, topology_details)

                network_list_reply.append(network_reply)

            # TODO: improve these workarounds to enhance performance

            yang_handler.destroy()

            response = jsonify(network_list_reply)

            # Workaround; pyangbind does not allow to set otn_topology / eth-tran-topology

            manual_fixes(response)

            response.status_code = HTTP_OK

        except Exception as e: # pylint: disable=broad-except

            LOGGER.exception('Something went wrong Retrieving Topology({:s})'.format(str(topology_id)))

            response = jsonify({'error': str(e)})

            response.status_code = HTTP_SERVERERROR

        return response

 No newline at end of file

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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, libyang, logging, os, re

from typing import Any

from common.DeviceTypes import DeviceTypeEnum

from common.proto.context_pb2 import TopologyDetails, Device, Link, EndPoint

from .NameMapping import NameMappings

from .NetworkTypeEnum import NetworkTypeEnum, get_network_topology_type

LOGGER = logging.getLogger(__name__)

YANG_DIR = os.path.join(os.path.dirname(__file__), 'yang')

YANG_MODULES = [

    'ietf-network',

    'ietf-network-topology'

]

MAPPINGS_TE_NODE_NAME = {

    '10.0.10.1'  : 'OA',

    '10.0.20.1'  : 'P',

    '10.0.30.1'  : 'OE',

    '10.0.40.1'  : 'P',

    '128.32.10.1': 'ONT1',

    '128.32.20.1': 'ONT2',

    '128.32.33.5': 'OLT',

}

IGNORE_ENDPOINT_NAMES = {'mgmt', 'eth1'}

IGNORE_DEVICE_TYPES = {

    DeviceTypeEnum.CLIENT.value,

    DeviceTypeEnum.DATACENTER.value,

    DeviceTypeEnum.EMULATED_CLIENT.value,

    DeviceTypeEnum.EMULATED_DATACENTER.value,

    DeviceTypeEnum.EMULATED_IP_SDN_CONTROLLER,

    DeviceTypeEnum.EMULATED_MICROWAVE_RADIO_SYSTEM.value,

    DeviceTypeEnum.EMULATED_OPEN_LINE_SYSTEM.value,

    DeviceTypeEnum.EMULATED_XR_CONSTELLATION.value,

    DeviceTypeEnum.IP_SDN_CONTROLLER,

    DeviceTypeEnum.MICROWAVE_RADIO_SYSTEM.value,

    DeviceTypeEnum.NETWORK.value,

    DeviceTypeEnum.OPEN_LINE_SYSTEM.value,

    DeviceTypeEnum.XR_CONSTELLATION.value,

}

IGNORE_DEVICE_NAMES = {

    NetworkTypeEnum.TE_OTN_TOPOLOGY: {

        'nce-t', '128.32.10.1', '128.32.33.5', '128.32.20.5', '128.32.20.1', '128.32.10.5',

    },

    NetworkTypeEnum.TE_ETH_TRAN_TOPOLOGY: {

        'nce-t',

    },

}

TE_TOPOLOGY_NAME = 'Huawei-Network'

class YangHandler:

    def __init__(self) -> None:

        self._yang_context = libyang.Context(YANG_DIR)

        for yang_module_name in YANG_MODULES:

            LOGGER.info('Loading module: {:s}'.format(str(yang_module_name)))

            self._yang_context.load_module(yang_module_name).feature_enable_all()

    def compose_network(self, te_topology_name : str, topology_details : TopologyDetails) -> None:

        networks = self._yang_context.create_data_path('/ietf-network:networks')

        network = networks.create_path('network[network-id="{:s}"]'.format(te_topology_name))

        network.create_path('network-id', te_topology_name)

        network_types = network.create_path('network-types') #augmentation, mirar RFC8346

        network_types.create_path('ietf-l3-unicast-topology:l3-unicast-topology') #no se si es asi

        network.create_path('supporting-network', ) #lista de networks-id, NO SE COMO VERLO

        attributes = network.create_path('ietf-l3-unicast-topology:l3-topology-attributes')

        #NO SE QUE ATRIBUTOS, CASO ESPECIAL OSPF?

        attributes.create_path('name',)

        attributes.create_path('flag',)

        ''' NO SEEEEEE SI HACE FALTA

        # TODO: resolve setting of otn_topology/eth_tran_topology network type; not working in bindings.

        # See "../ManualFixes.py".

        topology_id = ietf_network_obj.te_topology_identifier.topology_id

        topology_id = {

            '1': NetworkTypeEnum.TE_OTN_TOPOLOGY.value,

            '2': NetworkTypeEnum.TE_ETH_TRAN_TOPOLOGY.value,

        }.get(topology_id, topology_id)

        network_type = get_network_topology_type(topology_id)

        if network_type == NetworkTypeEnum.TE_OTN_TOPOLOGY:

            #ietf_network_obj.network_types.te_topology.otn_topology._set_present()

            pass

        elif network_type == NetworkTypeEnum.TE_ETH_TRAN_TOPOLOGY:

            #ietf_network_obj.network_types.te_topology.eth_tran_topology._set_present()

            pass

        else:

            raise Exception('Unsupported TopologyId({:s})'.format(str(topology_id)))

        '''

        name_mappings = NameMappings()

        ignore_device_uuids = set()

        for device in topology_details.devices:

            device_uuid = device.device_id.device_uuid.uuid

            device_type = device.device_type

            if device_type in IGNORE_DEVICE_TYPES:

                ignore_device_uuids.add(device_uuid)

                continue

            device_name = device.name

            if device_name in IGNORE_DEVICE_NAMES.get(network_type, set()): #NETWORK_TYPE?

                ignore_device_uuids.add(device_uuid)

                continue

            self.compose_node(device, name_mappings, network)

        for link in topology_details.links:

            link_device_uuids = {

                endpoint_id.device_id.device_uuid.uuid

                for endpoint_id in link.link_endpoint_ids

            }

            if len(ignore_device_uuids.intersection(link_device_uuids)) > 0:

                continue

            link_name = link.name

            self.compose_link(link_name, link, name_mappings, network)

    def compose_node(self, device : Device, name_mappings : NameMappings, network : Any

    ) -> None:

        device_name = device.name

        name_mappings.store_device_name(device)

        node = network.create_path('node[node-id="{:s}"]'.format(device_name))

        node.create_path('node-id', device_name)

        #supporting node (nw-ref, node-ref)?

        node_attributes = node.create_path('ietf-l3-unicast-topology:l3-node-attributes')

        #no se que va en cada atributo

        node_attributes.create_path('name',)

        node_attributes.create_path('flag',)

        node_attributes.create_path('router-id',)

        prefix = node_attributes.create_path('prefix[prefix="{:s}"]'.format())

        prefix.create_path('prefix',)

        prefix.create_path('metric',)

        prefix.create_path('flag',)

        for endpoint in device.device_endpoints:

            endpoint_name = endpoint.name

            if endpoint_name in IGNORE_ENDPOINT_NAMES: continue

            tp = node.create_path('ietf-network-topology:termination-point[tp-id="{:s}"]'.format(endpoint_name))

            tp.create_path('tp-id', endpoint_name)

            #supporting termination point?

            tp_attributes = tp.create_path('ietf-l3-unicast-topology:l3-termination-point-attributes')

            tp_attributes.create_path('termination-point-type',endpoint.endpoint_type) 

    def compose_link(self, link_name : Any, link_specs : Link, name_mappings : NameMappings, network : Any

    ) -> None:

        links = network.create_path('ietf-network-topology:link[link-id="{:s}"]'.format(link_name))

        links.create_path('link-id', link_name)

        src_endpoint_id = link_specs.link_endpoint_ids[0]

        source = links.create_path('source')

        source.create_path('source_node', name_mappings.get_device_name(src_endpoint_id.device_id))

        source.create_path('source_tp', name_mappings.get_endpoint_name(src_endpoint_id))

        dst_endpoint_id = link_specs.link_endpoint_ids[-1]

        destination = links.create_path('destination')

        destination.create_path('dest_node', name_mappings.get_device_name(dst_endpoint_id.device_id))

        destination.create_path('dest_tp', name_mappings.get_endpoint_name(dst_endpoint_id))

        #supporting-link???

        link_attributes = links.create_path('ietf-l3-unicast-topology:l3-link-attributes')

        #no se que atributos poner

        link_attributes.create_path('name',)

        link_attributes.create_path('flag',)

        link_attributes.create_path('metric',)