Newer
Older
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
#
# 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
from typing import Dict, List, Set, Tuple
from common.proto.context_pb2 import EndPointId, EndPointIdList
from context.client.ContextClient import ContextClient
LOGGER = logging.getLogger(__name__)
TYPE_DEVICE_NAMES = Dict[str, str] # device_uuid => device_name
TYPE_ENDPOINTS_DATA = Dict[str, Tuple[str, str]] # endpoint_uuid => (endpoint_name, endpoint_type)
TYPE_NAME_MAPS = Tuple[TYPE_DEVICE_NAMES, TYPE_ENDPOINTS_DATA]
def get_endpoint_names(context_client : ContextClient, endpoint_ids : List[EndPointId]) -> TYPE_NAME_MAPS:
endpoint_uuids_to_names : Set[str] = set()
endpoint_id_list = EndPointIdList()
for endpoint_id in endpoint_ids:
endpoint_uuid = endpoint_id.endpoint_uuid.uuid
if endpoint_uuid in endpoint_uuids_to_names: continue
endpoint_id_list.endpoint_ids.add().CopyFrom(endpoint_id) # pylint: disable=no-member
endpoint_uuids_to_names.add(endpoint_uuid)
endpoint_names = context_client.ListEndPointNames(endpoint_id_list)
device_names : TYPE_DEVICE_NAMES = dict()
endpoints_data : TYPE_ENDPOINTS_DATA = dict()
for endpoint_name in endpoint_names.endpoint_names:
device_uuid = endpoint_name.endpoint_id.device_id.device_uuid.uuid
device_names[device_uuid] = endpoint_name.device_name
endpoint_uuid = endpoint_name.endpoint_id.endpoint_uuid.uuid
endpoints_data[endpoint_uuid] = (endpoint_name.endpoint_name, endpoint_name.endpoint_type)
return device_names, endpoints_data