Device component - ACTN Driver:

# 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

import logging

from typing import Dict, List, Optional

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

    DEVICEOPERATIONALSTATUS_ENABLED

)

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__)

        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}"'

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

        # Options are:

        #    disabled --> just import endpoints as usual

    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)

        reply = 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 'ietf-network:network' not in reply:

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

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

        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)))

        if len(network) == 0:

            MSG = '[get] Network({:s}) not found; returning'

            LOGGER.debug(MSG.format(str(network_id)))

            return list()

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

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

            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))

        for link in network['ietf-network-topology:link']:

            link_uuid = link['link-id']

            link_src  = link['source']

            link_dst  = link['destination']

            link_src_dev_id = link_src['source-node']

            link_src_ep_id  = link_src['source-tp']

            link_dst_dev_id = link_dst['dest-node']

            link_dst_ep_id  = link_dst['dest-tp']

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

            link_endpoint_ids = [

                (link_src_dev_id, link_src_ep_id),

                (link_dst_dev_id, link_dst_ep_id),

            ]

            link_data = {

                'uuid': link_uuid,

                'name': link_uuid,

                'endpoints': link_endpoint_ids,

            }

            result.append((link_url, link_data))

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

        return result