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