# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 copy from typing import Dict, List, Optional, Tuple def json_endpoint_id(device_id : Dict, endpoint_uuid : str, topology_id : Optional[Dict] = None): result = {'device_id': copy.deepcopy(device_id), 'endpoint_uuid': {'uuid': endpoint_uuid}} if topology_id is not None: result['topology_id'] = copy.deepcopy(topology_id) return result def json_endpoint_ids( device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]], topology_id : Optional[Dict] = None ): return [ json_endpoint_id(device_id, endpoint_uuid, topology_id=topology_id) for endpoint_uuid, _, _ in endpoint_descriptors ] def json_endpoint( device_id : Dict, endpoint_uuid : str, endpoint_type : str, topology_id : Optional[Dict] = None, kpi_sample_types : List[int] = [] ): result = { 'endpoint_id': json_endpoint_id(device_id, endpoint_uuid, topology_id=topology_id), 'endpoint_type': endpoint_type, } if len(kpi_sample_types) > 0: result['kpi_sample_types'] = copy.deepcopy(kpi_sample_types) return result def json_endpoints( device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]], topology_id : Optional[Dict] = None ): return [ json_endpoint( device_id, endpoint_uuid, endpoint_type, topology_id=topology_id, kpi_sample_types=endpoint_sample_types) for endpoint_uuid, endpoint_type, endpoint_sample_types in endpoint_descriptors ]