From c3b5ba6a8b36c41d550b9b0faf3e1bc316b34a21 Mon Sep 17 00:00:00 2001 From: demx8as6 Date: Tue, 22 Feb 2022 12:41:35 +0100 Subject: [PATCH] Create a script to generate a Topology - add tapi topology generation Issue-ID: OAM-249 Change-Id: Ic2d34b2951be07780e1c318ff751ec8c78c95cfe Signed-off-by: demx8as6 --- .../model/python/tapi_topology.py | 702 +++++++++++++++++++++ 1 file changed, 702 insertions(+) create mode 100644 code/network-topology-instance-generator/model/python/tapi_topology.py diff --git a/code/network-topology-instance-generator/model/python/tapi_topology.py b/code/network-topology-instance-generator/model/python/tapi_topology.py new file mode 100644 index 0000000..803dd1b --- /dev/null +++ b/code/network-topology-instance-generator/model/python/tapi_topology.py @@ -0,0 +1,702 @@ +# Copyright 2022 highstreet technologies GmbH +# +# 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. + +#!/usr/bin/python +""" +Module containing the main class for this project for a TAPI Topology. +""" +import uuid +from typing import Dict, List, Union +from lxml import etree + +from model.python.top import Top +from model.python.tapi_node import TapiNode +from model.python.tapi_node_smo import TapiNodeSmo +from model.python.tapi_node_o_cloud import TapiNodeOCloud +from model.python.tapi_node_near_rt_ric import TapiNodeNearRtRic +from model.python.tapi_node_o_cu_cp import TapiNodeOCuCp +from model.python.tapi_node_o_cu_up import TapiNodeOCuUp +from model.python.tapi_node_o_du import TapiNodeODu +from model.python.tapi_node_fronthaul_gateway import TapiNodeFronthaulGateway +from model.python.tapi_node_o_ru import TapiNodeORu +from model.python.tapi_node_user_equipment import TapiNodeUserEquipment +from model.python.tapi_link import TapiLink + + +class TapiTopology(Top): + """ + Class representing a TAPI Topology + """ + + __data: Dict[str, Union[str, List[Union[Dict, TapiNode, TapiLink]]]] = None + __configuration: dict = None + + # constructor + def __init__(self, configuration: dict): + super().__init__(configuration) + self.__configuration = configuration + self.__data = { + "uuid": str(uuid.uuid4()), + "name": [{ + "value-name": "network-name", + "value": configuration['network']['name']}], + "layer-protocol-name": ["ETH"], + "node": [], + "link": []} + + topology_structure: dict = configuration['network']['pattern'] + network_function_type: str = next(iter(topology_structure)) + count: int = configuration['network']['pattern'][network_function_type] + + if network_function_type == "smo": + self.__create_smos(None, topology_structure, count) + elif network_function_type == "near-rt-ric": + self.__create_near_rt_rics(None, topology_structure, count) + elif network_function_type == "o-cu": + self.__create_o_cus(None, topology_structure, count) + elif network_function_type == "o-du": + self.__create_o_dus(None, topology_structure, count) + elif network_function_type == "o-ru": + self.__create_o_rus(None, topology_structure, count) + elif network_function_type == "ue": + self.__create_ues(None, topology_structure, count) + else: + print("Unknown network function type", network_function_type) + + # getter + def configuration(self) -> dict: + """ + Getter for a json object representing the TAPI Topology configuration. + :return TAPI Topology configuration as json object. + """ + return self.__configuration + + def data(self) -> dict: + """ + Getter for a json object representing the TAPI Topology. + :return TAPI Topology as json object. + """ + return self.__data + + def identifier(self) -> str: + """ + Getter returning the TAPI Topology identifier. + :return Object identifier as UUID. + """ + return self.__data["uuid"] + + def name(self) -> str: + """ + Getter for TAPI Topology name. The TAPI topology is a representation of + the network. Therefore, the TAPI Topology name has the same value as the + Network name. + :return TAPI Topology name as string. + """ + return self.__configuration['network']['name'] + + def json(self) -> dict: + """ + Getter for a json object representing the TAPI Topology. + :return TAPI Topology Context as json object. + """ + result = self.data().copy() + + # nodes handling + result["node"] = [] + for node in self.__data["node"]: + result["node"].append(node.json()) + + # link handling + result["link"] = [] + for link in self.__data["link"]: + result["link"].append(link.json()) + + return result + + def svg(self, svg_x: int, svg_y: int) -> etree.Element: + """ + Getter for a xml Element object representing the TAPI Topology Context. + :return TAPI Topology Context as svg object. + """ + group = etree.Element("g") + title = etree.Element("title") + title.text = "\n TAPI Topology \n id: " + \ + self.identifier() # + "\n name: " + self.name() + group.append(title) + + # nodes handling + index_per_type: Dict = {} + svg_nodes = [] + for node in self.__data["node"]: + if type(node) in index_per_type: + index_per_type[type(node)] = index_per_type[type(node)] + 1 + else: + index_per_type[type(node)] = 0 + index = index_per_type[type(node)] + node_x = svg_x + \ + index*self.__svg_dynamic_x_offset_by_node_type(type(node)) + \ + self.__svg_static_x_offset_by_node_type(type(node)) + node_y = svg_y + self.__svg_y_offset_by_node_type(type(node)) + svg_nodes.append(node.svg(node_x, node_y)) + # group.append(node.svg(node_x, node_y)) + + # handling and drawing links + for link in self.__data["link"]: + group.append(link.svg(0, 0)) + + # drawing nodes + for svg_node in svg_nodes: + group.append(svg_node) + + return group + + def __svg_static_x_offset_by_node_type(self, node_type) -> int: + """ + Mapping function from node types to y position in svg + return: int value + """ + pattern = self.configuration()['network']['pattern'] + padding = 0 # self.FONTSIZE + width_unit = (2 * 2 * self.FONTSIZE + padding) + + ru = (pattern['user-equipment']-1) * width_unit / 2 + fhgw = (pattern['o-ru'] * + pattern['user-equipment'] - 1) * width_unit / 2 + odu = (pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'] + - 1) * width_unit/2 + ocu = (pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'] + - 1) * width_unit / 2 + ric = (pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'] + - 1) * width_unit / 2 + smo = (pattern['smo'] * pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'] + -0.5) * width_unit + + x_mapping: Dict[type, int] = { + TapiNodeSmo: smo, + TapiNodeOCloud: ric, + TapiNodeNearRtRic: ric, + TapiNodeOCuCp: ocu - 12.5*self.FONTSIZE, + TapiNodeOCuUp: ocu + 12.5*self.FONTSIZE, + TapiNodeODu: odu, + TapiNodeFronthaulGateway: fhgw, + TapiNodeORu: ru, + TapiNodeUserEquipment: 0 + } + if node_type in x_mapping: + return x_mapping[node_type] + return 0 + + def __svg_dynamic_x_offset_by_node_type(self, node_type) -> int: + """ + Mapping function from node types to y position in svg + return: int value + """ + padding = 0 + pattern = self.configuration()['network']['pattern'] + width_unit = (2 * 2 * self.FONTSIZE + padding) + x_mapping: Dict[type, int] = { + TapiNodeSmo: pattern['near-rt-ric'] * pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeOCloud: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeNearRtRic: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeOCuCp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeOCuUp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeODu: pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'], + TapiNodeFronthaulGateway: pattern['o-ru'] * pattern['user-equipment'], + TapiNodeORu: pattern['user-equipment'], + TapiNodeUserEquipment: 1 + } + if node_type in x_mapping: + return x_mapping[node_type] * width_unit + return 0 + + def __svg_y_offset_by_node_type(self, node_type) -> int: + """ + Mapping function from node types to y position in svg + return: int value + """ + offset = 11*self.FONTSIZE + y_mapping: Dict[type, int] = { + TapiNodeSmo: 0 * offset, + TapiNodeOCloud: 1 * offset, + TapiNodeNearRtRic: 2 * offset, + TapiNodeOCuCp: 3.5 * offset - 4 * self.FONTSIZE, + TapiNodeOCuUp: 3.5 * offset + 4 * self.FONTSIZE, + TapiNodeODu: 5 * offset, + TapiNodeFronthaulGateway: 6 * offset, + TapiNodeORu: 7 * offset, + TapiNodeUserEquipment: 8 * offset + } + if node_type in y_mapping: + return y_mapping[node_type] + return 0 + + # methods + def add_node(self, node: TapiNode): + """ + Method adding a TAPI node to TAPI Topology. + :return TAPI Topology object. + """ + self.__data["node"].append(node) + return self + + def add_link(self, link: TapiLink): + """ + Method adding a TAPI node to TAPI Topology. + :return TAPI Topology object. + """ + self.__data["link"].append(link) + return self + + def __create_smos(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "smo" + next_type = "near-rt-ric" + for local_id in range(count): + prefix = "" + + if parent is not None: + prefix = parent.data()["name"][1]["value"] + config = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": "o-ran-sc-topology-common:"+current_type}} + node = TapiNodeSmo(parent, config) + self.add_node(node) + + # add O-Clouds + if "o-cloud" in topology_structure: + structure = topology_structure.copy() + self.__create_o_clouds( + node, structure, structure["o-cloud"]) + + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure["o-cloud"] + if current_type in structure: + del structure[current_type] + self.__create_near_rt_rics( + node, structure, structure[next_type]) + + return self + + def __create_o_clouds(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "o-cloud" + for local_id in range(count): + # add node + prefix = "" + if parent is not None: + prefix = parent.json()["name"][1]["value"] + function = "o-ran-sc-topology-common:"+current_type + node_configuration = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": function}} + node = TapiNodeOCloud(parent, node_configuration) + self.add_node(node) + + # add links + # O2 + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o2-rest", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + return self + + def __create_near_rt_rics(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "near-rt-ric" + next_type = "o-cu" + for local_id in range(count): + # add node + prefix = "" + if parent is not None: + prefix = parent.json()["name"][1]["value"] + function = "o-ran-sc-topology-common:"+current_type + node_configuration = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": function}} + node = TapiNodeNearRtRic(parent, node_configuration) + self.add_node(node) + + # add links + # A1 + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "a1-rest", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # O1 NETCONF + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-netconf", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # O1 FILE + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-file", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # O1 VES + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-ves", + "provider": parent, + "consumer": node + } + self.add_link(TapiLink(link_configuration)) + + # continue + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure[current_type] + self.__create_o_cus(node, structure, structure[next_type]) + + return self + + def __function_identity(self, function_type: str, plane: str) -> str: + """ + Method to calculate the Function IDENTITY + """ + return "".join([ + "o-ran-sc-topology-common:", + function_type, + "-", + plane + ]) + + def __create_o_cus(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "o-cu" + next_type = "o-du" + for local_id in range(count): + prefix = "" + if parent is not None: + prefix = parent.data()["name"][1]["value"] + + node: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {} + for plane in ["cp", "up"]: + config = {"node": {"localId": prefix + str(local_id), + "type": "-".join([current_type, plane]), + "function": self.__function_identity(current_type, plane)}} + classes: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = { + "cp": TapiNodeOCuCp, + "up": TapiNodeOCuUp} + node[plane] = classes[plane](parent, config) + self.add_node(node[plane]) + + # add links + # E2 + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "e2-rest", + "provider": node[plane], + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # O1 NETCONF + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-netconf", + "provider": node[plane], + "consumer": parent.parent() + } + self.add_link(TapiLink(link_configuration)) + + # O1 FILE + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-file", + "provider": node[plane], + "consumer": parent.parent() + } + self.add_link(TapiLink(link_configuration)) + + # O1 VES + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-ves", + "provider": parent.parent(), + "consumer": node[plane] + } + self.add_link(TapiLink(link_configuration)) + + # continue + # E1 Interface between O-CU-UP and O-CU-CP + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "e1-unknown", + "provider": node["up"], + "consumer": node["cp"] + } + self.add_link(TapiLink(link_configuration)) + + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure[current_type] + self.__create_o_dus( + node, structure, structure[next_type]) + return self + + def __create_o_dus(self, parents: Dict[str, TapiNode], topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "o-du" + next_type = "fronthaul-gateway" + for local_id in range(count): + prefix = "000" + if parents["cp"] is not None: + prefix = parents["cp"].data()["name"][1]["value"] + config = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": "o-ran-sc-topology-common:"+current_type}} + node = TapiNodeODu(parents["cp"], config) + self.add_node(node) + + for plane, parent in parents.items(): + + # add links + # E2 + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "e2-rest", + "provider": node, + "consumer": parent.parent() + } + self.add_link(TapiLink(link_configuration)) + + # O1 NETCONF + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-netconf", + "provider": node, + "consumer": parent.parent().parent() + } + self.add_link(TapiLink(link_configuration)) + + # O1 FILE + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-file", + "provider": node, + "consumer": parent.parent().parent() + } + self.add_link(TapiLink(link_configuration)) + + # O1 VES + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "o1-ves", + "provider": parent.parent().parent(), + "consumer": node + } + self.add_link(TapiLink(link_configuration)) + + # F1 User Plane or Control Plane + interfaces: Dict[str, str] = {"cp": "f1-c", "up": "f1-u"} + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": interfaces[plane]+"-unknown", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # continue + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure[current_type] + self.__create_fronthaul_gateways( + node, structure, structure[next_type]) + return self + + def __create_fronthaul_gateways(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "fronthaul-gateway" + next_type = "o-ru" + for local_id in range(count): + prefix = "" + if parent is not None: + prefix = parent.data()["name"][1]["value"] + node_configuration = { + "node": { + "localId": prefix + str(local_id), + "type": current_type, + "function": "o-ran-sc-topology-common:"+current_type, + "southbound-nep-count": topology_structure[next_type] + } + } + node = TapiNodeFronthaulGateway(parent, node_configuration) + self.add_node(node) + + # add links + + # Eth NBI + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "oam-netconf", + "provider": node, + "consumer": parent.parent().parent().parent() + } + self.add_link(TapiLink(link_configuration)) + + # Eth SBI + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "eth-ofh", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # continue + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure[current_type] + self.__create_o_rus(node, structure, structure[next_type]) + return self + + def __create_o_rus(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "o-ru" + next_type = "user-equipment" + for local_id in range(count): + prefix = "" + if parent is not None: + prefix = parent.data()["name"][1]["value"] + config = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": "o-ran-sc-topology-common:"+current_type}} + node = TapiNodeORu(parent, config) + self.add_node(node) + + # add links + + # O1 NETCONF + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "ofh-netconf", + "provider": node, + "consumer": parent.parent().parent().parent().parent() + } + self.add_link(TapiLink(link_configuration)) + + # OFH M-Plane to O-DU via fronthaul-gateway + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "ofh-netconf", + "provider": node, + "consumer": parent + } + self.add_link(TapiLink(link_configuration)) + + # continue + if next_type in topology_structure: + structure = topology_structure.copy() + if current_type in structure: + del structure[current_type] + self.__create_ues(node, structure, structure[next_type]) + return self + + def __create_ues(self, parent: TapiNode, topology_structure: dict, count: int): + """ + Method adding a TAPI node to TAPI Topology. + :param parent: A TAPI node which acts a a parent node in the topology. + :param topology_structure: Information about the next topology levels. + :param count: Number of instance to be created + :return TAPI Topology object. + """ + current_type = "user-equipment" + for local_id in range(count): + prefix = "" + if parent is not None: + prefix = parent.data()["name"][1]["value"] + config = {"node": {"localId": prefix + str(local_id), + "type": current_type, + "function": "o-ran-sc-topology-common:"+current_type}} + node = TapiNodeUserEquipment(parent, config) + self.add_node(node) + + # add links + # Uu unknown + link_configuration = { + "topology_reference": self.data()["uuid"], + "name_prefix": "uu-unknown", + "provider": parent, + "consumer": node + } + self.add_link(TapiLink(link_configuration)) + + if "key" in topology_structure: + print("Implement missing topology level.") + + return self -- 2.16.6