Device component:

import json, logging, requests, threading

from typing import Any, Iterator, List, Optional, Tuple, Union

from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method

from common.tools.client.RestConfClient import RestConfClient

from common.type_checkers.Checkers import chk_string, chk_type

from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOURCE_SERVICES

from .handlers.EthtServiceHandler import EthtServiceHandler

from .handlers.OsuTunnelHandler import OsuTunnelHandler

from .handlers.NetworkTopologyHandler import NetworkTopologyHandler

from .handlers.RestApiClient import RestApiClient

from .Tools import get_etht_services, get_osu_tunnels, parse_resource_key

        self.__started = threading.Event()

        self.__terminate = threading.Event()

        self._rest_api_client = RestApiClient(address, port, settings=settings)

        self._handler_osu_tunnel = OsuTunnelHandler(self._rest_api_client)

        self._rest_conf_client = RestConfClient(address, port, **settings)

        self._handler_etht_service = EthtServiceHandler(self._rest_api_client)

        self._handler_net_topology = NetworkTopologyHandler(self._rest_conf_client, **settings)

        self._handler_osu_tunnel = OsuTunnelHandler(self._rest_api_client)

    def Connect(self) -> bool:

        with self.__lock:

                    if resource_key == RESOURCE_ENDPOINTS:

                        # Add mgmt endpoint by default

                        resource_key = '/endpoints/endpoint[mgmt]'

                        resource_value = {'uuid': 'mgmt', 'name': 'mgmt', 'type': 'mgmt'}

                        results.append((resource_key, resource_value))

                        #resource_key = '/endpoints/endpoint[mgmt]'

                        #resource_value = {'uuid': 'mgmt', 'name': 'mgmt', 'type': 'mgmt'}

                        #results.append((resource_key, resource_value))

                        results.extend(self._handler_net_topology.get())

                    elif resource_key == RESOURCE_SERVICES:

                        get_osu_tunnels(self._handler_osu_tunnel, _results)

                        get_etht_services(self._handler_etht_service, _results)

# Copyright 2022-2025 ETSI 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 logging, requests

from typing import Dict, List, Optional, Tuple, Union

from common.Constants import DEFAULT_TOPOLOGY_NAME

from common.DeviceTypes import DeviceTypeEnum

from common.proto.context_pb2 import (

    DEVICEDRIVER_UNDEFINED, DEVICEOPERATIONALSTATUS_DISABLED,

    DEVICEOPERATIONALSTATUS_ENABLED, DeviceOperationalStatusEnum

)

from common.tools.client.RestApiClient import RestApiClient

from common.tools.client.RestConfClient import RestConfClient

from device.service.driver_api.ImportTopologyEnum import (

    ImportTopologyEnum, get_import_topology

)

from .RestApiClient import (

    HTTP_STATUS_CREATED, HTTP_STATUS_NO_CONTENT, HTTP_STATUS_OK,

    RestApiClient

)

GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'

GET_DEVICES_URL     = '/tfs-api/devices'

GET_LINKS_URL       = '/tfs-api/links'

L3VPN_URL           = '/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services'

MAPPING_STATUS = {

    'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,

    'DEVICEOPERATIONALSTATUS_DISABLED' : 1,

    'DEVICEOPERATIONALSTATUS_ENABLED'  : 2,

}

MAPPING_DRIVER = {

    'DEVICEDRIVER_UNDEFINED'            : 0,

    'DEVICEDRIVER_OPENCONFIG'           : 1,

    'DEVICEDRIVER_TRANSPORT_API'        : 2,

    'DEVICEDRIVER_P4'                   : 3,

    'DEVICEDRIVER_IETF_NETWORK_TOPOLOGY': 4,

    'DEVICEDRIVER_ONF_TR_532'           : 5,

    'DEVICEDRIVER_XR'                   : 6,

    'DEVICEDRIVER_IETF_L2VPN'           : 7,

    'DEVICEDRIVER_GNMI_OPENCONFIG'      : 8,

    'DEVICEDRIVER_OPTICAL_TFS'          : 9,

    'DEVICEDRIVER_IETF_ACTN'            : 10,

    'DEVICEDRIVER_OC'                   : 11,

    'DEVICEDRIVER_QKD'                  : 12,

    'DEVICEDRIVER_IETF_L3VPN'           : 13,

    'DEVICEDRIVER_IETF_SLICE'           : 14,

    'DEVICEDRIVER_NCE'                  : 15,

    'DEVICEDRIVER_SMARTNIC'             : 16,

    'DEVICEDRIVER_MORPHEUS'             : 17,

    'DEVICEDRIVER_RYU'                  : 18,

}

LOGGER = logging.getLogger(__name__)

class NetworkTopologyHandler:

    def __init__(self, rest_conf_client : RestConfClient, **settings) -> None:

        self._rest_conf_client = rest_conf_client

        self._object_name  = 'NetworkTopology'

        self._subpath_root = '/ietf-network:networks'

        self._subpath_item = self._subpath_root + '/network="{network_id:s}"'

        # Options are:

        #    disabled --> just import endpoints as usual

        #    devices  --> imports sub-devices but not links connecting them.

        #                 (a remotely-controlled transport domain might exist between them)

        #    topology --> imports sub-devices and links connecting them.

        #                 (not supported by XR driver)

        self._import_topology = get_import_topology(settings, default=ImportTopologyEnum.TOPOLOGY)

    def get(self, network_id : Optional[str] = None) -> List[Dict]:

        if network_id is None: network_id = DEFAULT_TOPOLOGY_NAME

        endpoint = self._subpath_item.format(network_id=network_id)

        networks = self._rest_conf_client.get(endpoint)

        if 'ietf-network:networks' not in networks:

            raise Exception('Malformed reply. "ietf-network:networks" missing')

        networks = networks['ietf-network:networks']

        if 'network' not in networks: return list()

        networks = networks['network']

        if len(networks) == 0: return list()

        network = next(iter([

            n for n in networks if n['network-id'] == network_id

        ]), default=None)

        if network is None:

            raise Exception('Network({:s}) not found'.format(str(network_id)))

        MSG = '[get] import_topology={:s}'

        LOGGER.debug(MSG.format(str(self._import_topology)))

        result = list()

        if self._import_topology == ImportTopologyEnum.DISABLED:

            LOGGER.debug('[get] abstract controller; returning')

            return result

        device_type = DeviceTypeEnum.EMULATED_PACKET_SWITCH.value

        endpoint_type = ''

        if 'network-types' in network:

            nnt = network['network-types']

            if 'ietf-te-topology:te-topology' in nnt:

                nnt_tet = nnt['ietf-te-topology:te-topology']

                if 'ietf-otn-topology:otn-topology' in nnt_tet:

                    device_type = DeviceTypeEnum.EMULATED_OPTICAL_ROADM.value

                    endpoint_type = 'optical'

                elif 'ietf-eth-te-topology:eth-tran-topology' in nnt_tet:

                    device_type = DeviceTypeEnum.EMULATED_PACKET_SWITCH.value

                    endpoint_type = 'copper'

                elif 'ietf-l3-unicast-topology:l3-unicast-topology' in nnt_tet:

                    device_type = DeviceTypeEnum.EMULATED_PACKET_ROUTER.value

                    endpoint_type = 'copper'

        for node in network['node']:

            node_id = node['node-id']

            node_name = node_id

            node_is_up = True

            if 'ietf-te-topology:te' in node:

                nte = node['ietf-te-topology:te']

                if 'oper-status' in nte:

                    node_is_up = nte['oper-status'] == 'up'

                if 'te-node-attributes' in nte:

                    ntea = nte['te-node-attributes']

                    if 'name' in ntea:

                        node_name = ntea['name']

            device_url = '/devices/device[{:s}]'.format(node_id)

            device_data = {

                'uuid': node_id,

                'name': node_name,

                'type': device_type,

                'status': DEVICEOPERATIONALSTATUS_ENABLED if node_is_up else DEVICEOPERATIONALSTATUS_DISABLED,

                'drivers': [DEVICEDRIVER_UNDEFINED],

            }

            result.append((device_url, device_data))

            for tp in node['ietf-network-topology:termination-point']:

                tp_id = tp['tp-id']

                tp_name = tp_id

                if 'ietf-te-topology:te' in tp:

                    tpte = tp['ietf-te-topology:te']

                    if 'name' in tpte:

                        tp_name = tpte['name']

                endpoint_url = '/endpoints/endpoint[{:s}, {:s}]'.format(node_id, tp_id)

                endpoint_data = {

                    'device_uuid': node_id,

                    'uuid': tp_id,

                    'name': tp_name,

                    'type': endpoint_type,

                }

                result.append((endpoint_url, endpoint_data))

        if self._import_topology == ImportTopologyEnum.DEVICES:

            LOGGER.debug('[get] devices only; returning')

            return result

#        for json_link in links['links']:

#            link_uuid : str = json_link['link_id']['link_uuid']['uuid']

#            link_url = '/links/link[{:s}]'.format(link_uuid)

#            link_endpoint_ids = [

#                (

#                    json_endpoint_id['device_id']['device_uuid']['uuid'],

#                    json_endpoint_id['endpoint_uuid']['uuid'],

#                )

#                for json_endpoint_id in json_link['link_endpoint_ids']

#            ]

#            link_data = {

#                'uuid': json_link['link_id']['link_uuid']['uuid'],

#                'name': json_link['name'],

#                'endpoints': link_endpoint_ids,

#            }

#            result.append((link_url, link_data))

        LOGGER.debug('[get] topology; returning')

        return result