Skip to content
Snippets Groups Projects
DescriptorTools.py 3.48 KiB
Newer Older
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
# 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.

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
import copy, json
from typing import Dict, List, Optional, Tuple, Union
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

def get_descriptors_add_contexts(contexts : List[Dict]) -> List[Dict]:
    contexts_add = copy.deepcopy(contexts)
    for context in contexts_add:
        context['topology_ids'] = []
        context['service_ids'] = []
    return contexts_add

def get_descriptors_add_topologies(topologies : List[Dict]) -> List[Dict]:
    topologies_add = copy.deepcopy(topologies)
    for topology in topologies_add:
        topology['device_ids'] = []
        topology['link_ids'] = []
    return topologies_add

def get_descriptors_add_services(services : List[Dict]) -> List[Dict]:
    services_add = []
    for service in services:
        service_copy = copy.deepcopy(service)
        service_copy['service_endpoint_ids'] = []
        service_copy['service_constraints'] = []
        service_copy['service_config'] = {'config_rules': []}
        services_add.append(service_copy)
    return services_add

def get_descriptors_add_slices(slices : List[Dict]) -> List[Dict]:
    slices_add = []
    for slice in slices:
        slice_copy = copy.deepcopy(slice)
        slice_copy['slice_endpoint_ids'] = []
        slice_copy['slice_constraints'] = []
        slice_copy['slice_config'] = {'config_rules': []}
        slices_add.append(slice_copy)
    return slices_add

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
TypeResourceValue = Union[str, int, bool, float, dict, list]
def format_custom_config_rules(config_rules : List[Dict]) -> List[Dict]:
    for config_rule in config_rules:
        if 'custom' not in config_rule: continue
        custom_resource_value : TypeResourceValue = config_rule['custom']['resource_value']
        if isinstance(custom_resource_value, (dict, list)):
            custom_resource_value = json.dumps(custom_resource_value, sort_keys=True, indent=0)
            config_rule['custom']['resource_value'] = custom_resource_value
    return config_rules

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
def split_devices_by_rules(devices : List[Dict]) -> Tuple[List[Dict], List[Dict]]:
    devices_add = []
    devices_config = []
    for device in devices:
        connect_rules = []
        config_rules = []
        for config_rule in device.get('device_config', {}).get('config_rules', []):
            custom_resource_key : Optional[str] = config_rule.get('custom', {}).get('resource_key')
            if custom_resource_key is not None and custom_resource_key.startswith('_connect/'):
                connect_rules.append(config_rule)
            else:
                config_rules.append(config_rule)

        if len(connect_rules) > 0:
            device_add = copy.deepcopy(device)
            device_add['device_endpoints'] = []
            device_add['device_config'] = {'config_rules': connect_rules}
            devices_add.append(device_add)

        if len(config_rules) > 0:
            device['device_config'] = {'config_rules': config_rules}
            devices_config.append(device)

    return devices_add, devices_config