From d7a950fc14c5dd9e27336051fb0ac9b61f83397b Mon Sep 17 00:00:00 2001 From: demx8as6 Date: Tue, 22 Feb 2022 12:42:31 +0100 Subject: [PATCH] Create a script to generate a Topology - add (abstract) tapi node generation Issue-ID: OAM-249 Change-Id: Ic67de8b49c3e82f221d610078863f4a1d15d2194 Signed-off-by: demx8as6 --- .../model/python/tapi_node.py | 341 +++++++++++++++++++++ 1 file changed, 341 insertions(+) create mode 100644 code/network-topology-instance-generator/model/python/tapi_node.py diff --git a/code/network-topology-instance-generator/model/python/tapi_node.py b/code/network-topology-instance-generator/model/python/tapi_node.py new file mode 100644 index 0000000..20da86d --- /dev/null +++ b/code/network-topology-instance-generator/model/python/tapi_node.py @@ -0,0 +1,341 @@ +# 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 class for a TAPI Node. +""" +import uuid +from typing import Dict +from lxml import etree +from model.python.svg.near_tr_ric import NearRtRic +from model.python.svg.o_cloud import OCloud +from model.python.svg.o_cu_cp import OCuCp +from model.python.svg.o_cu_up import OCuUp +from model.python.svg.o_du import ODu +from model.python.svg.fronthaul_gateway import FronthaulGateway +from model.python.svg.node import Node +from model.python.tapi_node_edge_point import TapiNodeEdgePoint +from model.python.top import Top + + +class TapiNode(Top): + """ + Class representing a TAPI Node. + """ + + __data: dict = {} + __configuration: dict = {} + __parent: 'TapiNode' = None + __width: int = 0 # default SVG width, should be overritten by constructor + + # constructor + def __init__(self, parent: 'TapiNode', configuration: dict): + super().__init__(configuration) + self.__parent = parent + self.__configuration = configuration + self.width((4 + 1) * (2.2*self.FONTSIZE)) # 4x nep + self.__data = { + "uuid": str(uuid.uuid4()), + "name": [ + { + "value-name": "topology-node-name", + "value": self.name() + }, + { + "value-name": "topology-node-local-id", + "value": configuration['node']['localId'] + } + ], + "owned-node-edge-point": [], + "administrative-state": "LOCKED", + "operational-state": "ENABLED", + "lifecycle-state": "INSTALLED", + "layer-protocol-name": ["ETH"], + "cost-characteristic": [ + { + "cost-name": "cost", + "cost-algorithm": "alg1", + "cost-value": "value-1" + } + ], + "latency-characteristic": [{ + "traffic-property-name": "property-1", + "queing-latency-characteristic": "queue-1", + "fixed-latency-characteristic": "latency-1", + "jitter-characteristic": "jitter-1", + "wander-characteristic": "wander-1" + }], + "o-ran-sc-topology:function": configuration['node']['function'], + "o-ran-sc-topology:geolocation": { + "longitude": "0", + "latitude": "0", + "altitude": "20000" + } + } + + # getter + def x_offset_by_cep_name(self, name: str, local_id: int) -> int: + cep_of_interest = len(self.data()['owned-node-edge-point'])-2 + mapping: Dict[str, int] = { + "o2-rest-consumer": -4*6*self.FONTSIZE, + "a1-rest-consumer": -2*6*self.FONTSIZE, + "oam-netconf-consumer": -0*6*self.FONTSIZE, + "o1-ves-provider": 2*6*self.FONTSIZE, + "o1-file-consumer": 4*6*self.FONTSIZE, + + "o2-rest-provider": 0*self.FONTSIZE, + "a1-rest-provider": -8*self.FONTSIZE, + "e2-rest-consumer": 0*self.FONTSIZE, + + "f1-c-unknown-consumer": 0*self.FONTSIZE, + "f1-u-unknown-consumer": 0*self.FONTSIZE, + + "e1-unknown-provider": -5*2*self.FONTSIZE, + "e1-unknown-consumer": 5*2*self.FONTSIZE, + + "e2-rest-provider": -4*2*self.FONTSIZE, + "f1-c-unknown-provider": -2*2*self.FONTSIZE, + "f1-u-unknown-provider": 2*2*self.FONTSIZE, + "f1-unknown-provider": -2*2*self.FONTSIZE, + "o1-netconf-provider": 4*self.FONTSIZE, + "o1-ves-consumer": 8*self.FONTSIZE, + "o1-file-provider": 12*self.FONTSIZE, + "ofh-netconf-consumer": -2*self.FONTSIZE, + + "eth-ofh-provider": -2*self.FONTSIZE, + "oam-netconf-provider": 2*self.FONTSIZE, + # "eth-ofh-consumer": -(len(self.data()['owned-node-edg + # e-point']) - 2)/2 * 2 * self.FONTSIZE, + "eth-ofh-consumer": 0-(cep_of_interest/2)*4*self.FONTSIZE + 2*self.FONTSIZE, + + "ofh-netconf-provider": 0*self.FONTSIZE, + "uu-unknown-provider": 0*self.FONTSIZE, + + "uu-unknown-consumer": 0*self.FONTSIZE + } + if name in mapping: + return mapping[name] + 4*self.FONTSIZE * local_id + + print("Node: CEP name", name, "for x postion calculation not found") + return 0 + + def y_offset_by_cep_name(self, name: str) -> int: + mapping: Dict[str, int] = { + "o2-rest-consumer": 3*self.FONTSIZE, + "a1-rest-consumer": 3*self.FONTSIZE, + "oam-netconf-consumer": 3*self.FONTSIZE, + "o1-ves-provider": 3*self.FONTSIZE, + "o1-file-consumer": 3*self.FONTSIZE, + + "o2-rest-provider": -3*self.FONTSIZE, + "a1-rest-provider": -3*self.FONTSIZE, + "e2-rest-consumer": 3*self.FONTSIZE, + + "e1-unknown-provider": 1*self.FONTSIZE, + "e1-unknown-consumer": 1*self.FONTSIZE, + + "f1-c-unknown-consumer": 3*self.FONTSIZE, + "f1-u-unknown-consumer": 3*self.FONTSIZE, + "f1-unknown-consumer": 3*self.FONTSIZE, + + "e2-rest-provider": -3*self.FONTSIZE, + "f1-c-unknown-provider": -3*self.FONTSIZE, + "f1-u-unknown-provider": -3*self.FONTSIZE, + "f1-unknown-provider": -3*self.FONTSIZE, + "o1-netconf-provider": -3*self.FONTSIZE, + "o1-ves-consumer": -3*self.FONTSIZE, + "o1-file-provider": -3*self.FONTSIZE, + "ofh-netconf-consumer": 3*self.FONTSIZE, + + "eth-ofh-provider": -3*self.FONTSIZE, + "oam-netconf-provider": -3*self.FONTSIZE, + "eth-ofh-consumer": +3*self.FONTSIZE, + + "ofh-netconf-provider": -3*self.FONTSIZE, + "uu-unknown-provider": 3*self.FONTSIZE, + + "uu-unknown-consumer": -3*self.FONTSIZE + } + if name in mapping: + return mapping[name] + + print("Node: CEP name", name, "for y postion calculation not found") + return 0 + + def configuration(self) -> dict: + """ + Getter for a json object representing the TAPI Node configuration. + :return TAPI Node configuration as json object. + """ + return self.__configuration + + def data(self) -> dict: + """ + Getter for a json object representing the TAPI Link. + :return TAPI Link as json object. + """ + return self.__data + + def local_id(self) -> int: + return self.configuration()["node"]["localId"] + + def function(self) -> str: + """ + Getter returning the network-function type + :return The type of the network-function as yang IDENTITY. + """ + return self.__configuration['node']['function'] + + def function_label(self) -> str: + """ + Getter returning the network-function label + :return The type of the network-function as human readable string. + """ + mapping = { + "o-ran-sc-topology-common:smo": "SMO", + "o-ran-sc-topology-common:o-cloud": "O-Cloud", + "o-ran-sc-topology-common:near-rt-ric": "Near-RT-RIC", + "o-ran-sc-topology-common:o-cu": "O-CU", + "o-ran-sc-topology-common:o-cu-cp": "O-CU-CP", + "o-ran-sc-topology-common:o-cu-up": "O-CU-UP", + "o-ran-sc-topology-common:o-du": "O-DU", + "o-ran-sc-topology-common:fronthaul-gateway": "FHGW", + "o-ran-sc-topology-common:o-ru": "O-RU", + "o-ran-sc-topology-common:user-equipment": "UE" + } + if mapping[self.function()]: + return mapping[self.function()] + else: + return self.function() + + def identifier(self) -> str: + """ + Getter returning the TAPI Node identifier. + :return Object identifier as UUID. + """ + return self.__data["uuid"] + + def json(self) -> dict: + """ + Getter for a json object representing the TAPI Node. + :return TAPI Node as json object. + """ + result = self.__data.copy() + result['owned-node-edge-point'] = [] + for nep in self.__data['owned-node-edge-point']: + result['owned-node-edge-point'].append(nep.json()) + return result + + def name(self) -> str: + """ + Getter for TAPI Node name. + :return TAPI Node as json object. + """ + return "".join([ + self.__configuration['node']['type'], + "-", + str(self.__configuration['node']['localId']) + ]) + + def node_edge_point_by_cep_name(self, cep_name, local_id) -> TapiNodeEdgePoint: + """ + Method returning a NEP based on a given interface name + :param interface_name: Search string + :return The NEP uuid or "not found" + """ + result = [] + for nep in self.__data["owned-node-edge-point"]: + for cep in nep.connection_edge_points(): + if cep.name() == cep_name: + result.append(nep) + if len(result) == 0: + for nep in self.__data["owned-node-edge-point"]: + print("# Check", cep_name, nep.json()["name"][0]["value"], nep.json()[ + "tapi-connectivity:cep-list"]["connection-end-point"][0]["name"][0]["value"]) + if len(result) > 1: + for nep in result: + if nep.name().endswith(str(local_id[-1])): + return nep + return result[0] + + def parent(self) -> 'TapiNode': + """ + Getter for a TAPI Node object representing the TAPI Node configuration. + :return TAPI Node configuration as json object. + """ + return self.__parent + + def svg(self, x: int, y: int) -> etree.Element: + """ + Getter for a xml Element object representing the TAPI Node. + :return TAPI Node as svg object. + """ + self.__svg_x = x + self.__svg_y = y + + svg_nep = None + if type(self).__name__ == "TapiNodeSmo": + svg_nep = Node(self, x, y) + elif type(self).__name__ == "TapiNodeOCloud": + svg_nep = OCloud(self, x, y) + elif type(self).__name__ == "TapiNodeNearRtRic": + svg_nep = NearRtRic(self, x, y) + elif type(self).__name__ == "TapiNodeOCuCp": + svg_nep = OCuCp(self, x, y) + elif type(self).__name__ == "TapiNodeOCuUp": + svg_nep = OCuUp(self, x, y) + elif type(self).__name__ == "TapiNodeODu": + svg_nep = ODu(self, x, y) + elif type(self).__name__ == "TapiNodeFronthaulGateway": + svg_nep = FronthaulGateway(self, x, y) + # elif type(self).__name__ == "TapiNodeORu": + # svg_nep = Node(self, x, y) + # elif type(self).__name__ == "TapiNodeUserEquipment": + # svg_nep = Node(self, x, y) + else: + svg_nep = Node(self, x, y) + + group: etree.Element = svg_nep.svg_element() + + for nep in self.data()['owned-node-edge-point']: + localId = 0 + if "local-id" in nep.configuration()["nodeEdgePoint"]: + localId = nep.configuration()["nodeEdgePoint"]["local-id"] + + nep_x = x + \ + self.x_offset_by_cep_name( + nep.connection_edge_points()[0].name(), localId) + nep_y = y + \ + self.y_offset_by_cep_name( + nep.connection_edge_points()[0].name()) + group.append(nep.svg(nep_x, nep_y)) + return group + + def width(self, width: int) -> None: + """ + Setter for the SVG width in px. + :param width as integer with unit "px" (pixel) + """ + self.__width = width + + # methods + + def add(self, nep: TapiNodeEdgePoint) -> 'TapiNode': + """ + Method adding a TAPI Node Edge Point object. + :return TAPI Node as object. + """ + self.__data['owned-node-edge-point'].append(nep) + return self -- 2.16.6