code/network-topology-instance-generator/model/python/tapi_node.py

   1 # Copyright 2022 highstreet technologies GmbH
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 #!/usr/bin/python
  16 """
  17 Module containing the class for a TAPI Node.
  18 """
  19 import uuid
  20 from typing import Dict
  21 from lxml import etree
  22 from model.python.svg.near_tr_ric import NearRtRic
  23 from model.python.svg.amf import Amf
  24 from model.python.svg.o_cloud import OCloud
  25 from model.python.svg.o_cu_cp import OCuCp
  26 from model.python.svg.o_cu_up import OCuUp
  27 from model.python.svg.o_du import ODu
  28 from model.python.svg.ue import Ue
  29 from model.python.svg.fronthaul_gateway import FronthaulGateway
  30 from model.python.svg.node import Node
  31 from model.python.tapi_node_edge_point import TapiNodeEdgePoint
  32 from model.python.top import Top
  33
  34
  35 class TapiNode(Top):
  36     """
  37     Class representing a TAPI Node.
  38     """
  39
  40     __data: dict = {}
  41     __configuration: dict = {}
  42     __parent: 'TapiNode' = None
  43     __width: int = 0  # default SVG width, should be overritten by constructor
  44
  45     # constructor
  46     def __init__(self, parent: 'TapiNode', configuration: dict):
  47         super().__init__(configuration)
  48         self.__parent = parent
  49         self.__configuration = configuration
  50         self.width((4 + 1) * (2.2*self.FONTSIZE))  # 4x nep
  51         self.__data = {
  52             "uuid": str(uuid.uuid4()),
  53             "name": [
  54                 {
  55                     "value-name": "topology-node-name",
  56                     "value": self.name()
  57                 },
  58                 {
  59                     "value-name": "topology-node-local-id",
  60                     "value": configuration['node']['localId']
  61                 }
  62             ],
  63             "owned-node-edge-point": [],
  64             "administrative-state": "LOCKED",
  65             "operational-state": "ENABLED",
  66             "lifecycle-state": "INSTALLED",
  67             "layer-protocol-name": ["ETH"],
  68             "cost-characteristic": [
  69                 {
  70                     "cost-name": "cost",
  71                     "cost-algorithm": "alg1",
  72                     "cost-value": "value-1"
  73                 }
  74             ],
  75             "latency-characteristic": [{
  76                 "traffic-property-name": "property-1",
  77                 "queing-latency-characteristic": "queue-1",
  78                 "fixed-latency-characteristic": "latency-1",
  79                 "jitter-characteristic": "jitter-1",
  80                 "wander-characteristic": "wander-1"
  81             }],
  82             "o-ran-sc-topology:function": configuration['node']['function'],
  83             "o-ran-sc-topology:geolocation": {
  84                 "longitude": "0",
  85                 "latitude": "0",
  86                 "altitude": "20000"
  87             }
  88         }
  89
  90     # getter
  91     def x_offset_by_cep_name(self, name: str, local_id: int) -> int:
  92         cep_of_interest = len(self.data()['owned-node-edge-point'])-2
  93         mapping: Dict[str, int] = {
  94             "o2-rest-consumer": -4*6*self.FONTSIZE,
  95             "a1-rest-consumer": -2*6*self.FONTSIZE,
  96             "oam-netconf-consumer": -0*6*self.FONTSIZE,
  97             "o1-ves-provider": 2*6*self.FONTSIZE,
  98             "o1-file-consumer": 4*6*self.FONTSIZE,
  99
 100             "o2-rest-provider": 0*self.FONTSIZE,
 101             "a1-rest-provider": -8*self.FONTSIZE,
 102             "e2-rest-consumer": 0*self.FONTSIZE,
 103
 104             "n1-nas-provider": -2*self.FONTSIZE,
 105             "n2-nas-provider": +2*self.FONTSIZE,
 106             "n3-nas-provider": 0*self.FONTSIZE,
 107
 108             "f1-c-unknown-consumer": 0*self.FONTSIZE,
 109             "f1-u-unknown-consumer": 0*self.FONTSIZE,
 110
 111             "e1-unknown-provider": -5*2*self.FONTSIZE,
 112             "e1-unknown-consumer": 5*2*self.FONTSIZE,
 113
 114             "e2-rest-provider": -4*2*self.FONTSIZE,
 115             "n2-nas-consumer": -2*2*self.FONTSIZE,
 116             "n3-nas-consumer": -2*2*self.FONTSIZE,
 117             "f1-c-unknown-provider": -2*2*self.FONTSIZE,
 118             "f1-u-unknown-provider": 2*2*self.FONTSIZE,
 119             "f1-unknown-provider": -2*2*self.FONTSIZE,
 120             "o1-netconf-provider": 4*self.FONTSIZE,
 121             "o1-ves-consumer": 8*self.FONTSIZE,
 122             "o1-file-provider": 12*self.FONTSIZE,
 123             "ofh-netconf-consumer": -2*self.FONTSIZE,
 124
 125             "eth-ofh-provider": -2*self.FONTSIZE,
 126             "oam-netconf-provider": 2*self.FONTSIZE,
 127             #            "eth-ofh-consumer": -(len(self.data()['owned-node-edg
 128             # e-point']) - 2)/2 * 2 * self.FONTSIZE,
 129             "eth-ofh-consumer": 0-(cep_of_interest/2)*4*self.FONTSIZE + 2*self.FONTSIZE,
 130
 131             "ofh-netconf-provider": 0*self.FONTSIZE,
 132             "uu-radio-provider": 0*self.FONTSIZE,
 133
 134             "uu-radio-consumer": -2*self.FONTSIZE,
 135             "n1-nas-consumer":2*self.FONTSIZE
 136         }
 137         if name in mapping:
 138             return mapping[name] + 4*self.FONTSIZE * local_id
 139
 140         print("Node: CEP name", name, "for x postion calculation not found")
 141         return 0
 142
 143     def y_offset_by_cep_name(self, name: str) -> int:
 144         mapping: Dict[str, int] = {
 145             "o2-rest-consumer": 3*self.FONTSIZE,
 146             "a1-rest-consumer": 3*self.FONTSIZE,
 147             "oam-netconf-consumer": 3*self.FONTSIZE,
 148             "o1-ves-provider": 3*self.FONTSIZE,
 149             "o1-file-consumer": 3*self.FONTSIZE,
 150
 151             "o2-rest-provider": -3*self.FONTSIZE,
 152             "a1-rest-provider": -3*self.FONTSIZE,
 153             "e2-rest-consumer": 3*self.FONTSIZE,
 154
 155             "n1-nas-provider": 3*self.FONTSIZE,
 156             "n2-nas-provider": 3*self.FONTSIZE,
 157             "n3-nas-provider": 3*self.FONTSIZE,
 158
 159             "e1-unknown-provider": 1*self.FONTSIZE,
 160             "e1-unknown-consumer": 1*self.FONTSIZE,
 161
 162             "f1-c-unknown-consumer": 3*self.FONTSIZE,
 163             "f1-u-unknown-consumer": 3*self.FONTSIZE,
 164             "f1-unknown-consumer": 3*self.FONTSIZE,
 165
 166             "e2-rest-provider": -3*self.FONTSIZE,
 167             "n2-nas-consumer": -3*self.FONTSIZE,
 168             "n3-nas-consumer": -3*self.FONTSIZE,
 169             "f1-c-unknown-provider": -3*self.FONTSIZE,
 170             "f1-u-unknown-provider": -3*self.FONTSIZE,
 171             "f1-unknown-provider": -3*self.FONTSIZE,
 172             "o1-netconf-provider": -3*self.FONTSIZE,
 173             "o1-ves-consumer": -3*self.FONTSIZE,
 174             "o1-file-provider": -3*self.FONTSIZE,
 175             "ofh-netconf-consumer": 3*self.FONTSIZE,
 176
 177             "eth-ofh-provider": -3*self.FONTSIZE,
 178             "oam-netconf-provider": -3*self.FONTSIZE,
 179             "eth-ofh-consumer": +3*self.FONTSIZE,
 180
 181             "ofh-netconf-provider": -3*self.FONTSIZE,
 182             "uu-radio-provider": 3*self.FONTSIZE,
 183
 184             "uu-radio-consumer": -3*self.FONTSIZE,
 185             "n1-nas-consumer": -3*self.FONTSIZE
 186         }
 187         if name in mapping:
 188             return mapping[name]
 189
 190         print("Node: CEP name", name, "for y postion calculation not found")
 191         return 0
 192
 193     def configuration(self) -> dict:
 194         """
 195         Getter for a json object representing the TAPI Node configuration.
 196         :return TAPI Node configuration as json object.
 197         """
 198         return self.__configuration
 199
 200     def data(self) -> dict:
 201         """
 202         Getter for a json object representing the TAPI Link.
 203         :return TAPI Link as json object.
 204         """
 205         return self.__data
 206
 207     def local_id(self) -> int:
 208         return self.configuration()["node"]["localId"]
 209
 210     def function(self) -> str:
 211         """
 212         Getter returning the network-function type
 213         :return The type of the network-function as yang IDENTITY.
 214         """
 215         return self.__configuration['node']['function']
 216
 217     def function_label(self) -> str:
 218         """
 219         Getter returning the network-function label
 220         :return The type of the network-function as human readable string.
 221         """
 222         mapping = {
 223             "o-ran-sc-topology-common:smo": "SMO",
 224             "o-ran-sc-topology-common:o-cloud": "O-Cloud",
 225             "o-ran-sc-topology-common:near-rt-ric": "Near-RT-RIC",
 226             "o-ran-sc-topology-common:access-and-mobility-management-function": "AMF",
 227             "o-ran-sc-topology-common:user-plane-function": "UPF",
 228             "o-ran-sc-topology-common:o-cu": "O-CU",
 229             "o-ran-sc-topology-common:o-cu-cp": "O-CU-CP",
 230             "o-ran-sc-topology-common:o-cu-up": "O-CU-UP",
 231             "o-ran-sc-topology-common:o-du": "O-DU",
 232             "o-ran-sc-topology-common:fronthaul-gateway": "FHGW",
 233             "o-ran-sc-topology-common:o-ru": "O-RU",
 234             "o-ran-sc-topology-common:user-equipment": "UE"
 235         }
 236         if mapping[self.function()]:
 237             return mapping[self.function()]
 238         else:
 239             return self.function()
 240
 241     def identifier(self) -> str:
 242         """
 243         Getter returning the TAPI Node identifier.
 244         :return Object identifier as UUID.
 245         """
 246         return self.__data["uuid"]
 247
 248     def json(self) -> dict:
 249         """
 250         Getter for a json object representing the TAPI Node.
 251         :return TAPI Node as json object.
 252         """
 253         result = self.__data.copy()
 254         result['owned-node-edge-point'] = []
 255         for nep in self.__data['owned-node-edge-point']:
 256             result['owned-node-edge-point'].append(nep.json())
 257         return result
 258
 259     def name(self) -> str:
 260         """
 261         Getter for TAPI Node name.
 262         :return TAPI Node as json object.
 263         """
 264         return "".join([
 265             self.__configuration['node']['type'],
 266             "-",
 267             str(self.__configuration['node']['localId'])
 268         ])
 269
 270     def node_edge_point_by_cep_name(self, cep_name, local_id) -> TapiNodeEdgePoint:
 271         """
 272         Method returning a NEP based on a given interface name
 273         :param interface_name: Search string
 274         :return The NEP uuid or "not found"
 275         """
 276         result = []
 277         for nep in self.__data["owned-node-edge-point"]:
 278             for cep in nep.connection_edge_points():
 279                 if cep.name() == cep_name:
 280                     result.append(nep)
 281         if len(result) == 0:
 282             for nep in self.__data["owned-node-edge-point"]:
 283                 print("# Check", cep_name, nep.json()["name"][0]["value"], nep.json()[
 284                       "tapi-connectivity:cep-list"]["connection-end-point"][0]["name"][0]["value"])
 285             return
 286         if len(result) > 1:
 287             for nep in result:
 288                 if nep.name().endswith(str(local_id[-1])):
 289                     return nep
 290         return result[0]
 291
 292     def parent(self) -> 'TapiNode':
 293         """
 294         Getter for a TAPI Node object representing the TAPI Node configuration.
 295         :return TAPI Node configuration as json object.
 296         """
 297         return self.__parent
 298
 299     def svg(self, x: int, y: int) -> etree.Element:
 300         """
 301         Getter for a xml Element object representing the TAPI Node.
 302         :return TAPI Node as svg object.
 303         """
 304         self.__svg_x = x
 305         self.__svg_y = y
 306
 307         svg_nep = None
 308         if type(self).__name__ == "TapiNodeSmo":
 309             svg_nep = Node(self, x, y)
 310         elif type(self).__name__ == "TapiNodeOCloud":
 311             svg_nep = OCloud(self, x, y)
 312         elif type(self).__name__ == "TapiNodeNearRtRic":
 313             svg_nep = NearRtRic(self, x, y)
 314         elif type(self).__name__ == "TapiNodeAmf":
 315             svg_nep = Amf(self, x, y)
 316         elif type(self).__name__ == "TapiNodeOCuCp":
 317             svg_nep = OCuCp(self, x, y)
 318         elif type(self).__name__ == "TapiNodeOCuUp":
 319             svg_nep = OCuUp(self, x, y)
 320         elif type(self).__name__ == "TapiNodeODu":
 321             svg_nep = ODu(self, x, y)
 322         elif type(self).__name__ == "TapiNodeUserEquipment":
 323             svg_nep = Ue(self, x, y)
 324         elif type(self).__name__ == "TapiNodeFronthaulGateway":
 325             svg_nep = FronthaulGateway(self, x, y)
 326         # elif type(self).__name__ == "TapiNodeORu":
 327         #     svg_nep = Node(self, x, y)
 328         # elif type(self).__name__ == "TapiNodeUserEquipment":
 329         #     svg_nep = Node(self, x, y)
 330         else:
 331             svg_nep = Node(self, x, y)
 332
 333         group: etree.Element = svg_nep.svg_element()
 334
 335         for nep in self.data()['owned-node-edge-point']:
 336             localId = 0
 337             if "local-id" in nep.configuration()["nodeEdgePoint"]:
 338                 localId = nep.configuration()["nodeEdgePoint"]["local-id"]
 339
 340             nep_x = x + \
 341                 self.x_offset_by_cep_name(
 342                     nep.connection_edge_points()[0].name(), localId)
 343             nep_y = y + \
 344                 self.y_offset_by_cep_name(
 345                     nep.connection_edge_points()[0].name())
 346             group.append(nep.svg(nep_x, nep_y))
 347         return group
 348
 349     def width(self, width: int) -> None:
 350         """
 351         Setter for the SVG width in px.
 352         :param width as integer with unit "px" (pixel)
 353         """
 354         self.__width = width
 355
 356     # methods
 357
 358     def add(self, nep: TapiNodeEdgePoint) -> 'TapiNode':
 359         """
 360         Method adding a TAPI Node Edge Point object.
 361         :return TAPI Node as object.
 362         """
 363         self.__data['owned-node-edge-point'].append(nep)
 364         return self