1 # Copyright 2022 highstreet technologies GmbH
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 Module containing the class for a TAPI Node.
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
37 Class representing a TAPI Node.
41 __configuration: dict = {}
42 __parent: 'TapiNode' = None
43 __width: int = 0 # default SVG width, should be overritten by 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
52 "uuid": str(uuid.uuid4()),
55 "value-name": "topology-node-name",
59 "value-name": "topology-node-local-id",
60 "value": configuration['node']['localId']
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": [
71 "cost-algorithm": "alg1",
72 "cost-value": "value-1"
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"
82 "o-ran-sc-topology:function": configuration['node']['function'],
83 "o-ran-sc-topology:geolocation": {
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,
100 "o2-rest-provider": 0*self.FONTSIZE,
101 "a1-rest-provider": -8*self.FONTSIZE,
102 "e2-sctp-consumer": 0*self.FONTSIZE,
104 "n1-nas-provider": -2*self.FONTSIZE,
105 "n2-nas-provider": +2*self.FONTSIZE,
106 "n3-nas-provider": 0*self.FONTSIZE,
108 "f1-c-unknown-consumer": 0*self.FONTSIZE,
109 "f1-u-unknown-consumer": 0*self.FONTSIZE,
111 "e1-unknown-provider": -5*2*self.FONTSIZE,
112 "e1-unknown-consumer": 5*2*self.FONTSIZE,
114 "e2-sctp-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,
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,
131 "ofh-netconf-provider": 0*self.FONTSIZE,
132 "uu-radio-provider": 0*self.FONTSIZE,
134 "uu-radio-consumer": -2*self.FONTSIZE,
135 "n1-nas-consumer":2*self.FONTSIZE
138 return mapping[name] + 4*self.FONTSIZE * local_id
140 print("Node: CEP name", name, "for x postion calculation not found")
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,
151 "o2-rest-provider": -3*self.FONTSIZE,
152 "a1-rest-provider": -3*self.FONTSIZE,
153 "e2-sctp-consumer": 3*self.FONTSIZE,
155 "n1-nas-provider": 3*self.FONTSIZE,
156 "n2-nas-provider": 3*self.FONTSIZE,
157 "n3-nas-provider": 3*self.FONTSIZE,
159 "e1-unknown-provider": 1*self.FONTSIZE,
160 "e1-unknown-consumer": 1*self.FONTSIZE,
162 "f1-c-unknown-consumer": 3*self.FONTSIZE,
163 "f1-u-unknown-consumer": 3*self.FONTSIZE,
164 "f1-unknown-consumer": 3*self.FONTSIZE,
166 "e2-sctp-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,
177 "eth-ofh-provider": -3*self.FONTSIZE,
178 "oam-netconf-provider": -3*self.FONTSIZE,
179 "eth-ofh-consumer": +3*self.FONTSIZE,
181 "ofh-netconf-provider": -3*self.FONTSIZE,
182 "uu-radio-provider": 3*self.FONTSIZE,
184 "uu-radio-consumer": -3*self.FONTSIZE,
185 "n1-nas-consumer": -3*self.FONTSIZE
190 print("Node: CEP name", name, "for y postion calculation not found")
193 def configuration(self) -> dict:
195 Getter for a json object representing the TAPI Node configuration.
196 :return TAPI Node configuration as json object.
198 return self.__configuration
200 def data(self) -> dict:
202 Getter for a json object representing the TAPI Link.
203 :return TAPI Link as json object.
207 def local_id(self) -> int:
208 return self.configuration()["node"]["localId"]
210 def function(self) -> str:
212 Getter returning the network-function type
213 :return The type of the network-function as yang IDENTITY.
215 return self.__configuration['node']['function']
217 def function_label(self) -> str:
219 Getter returning the network-function label
220 :return The type of the network-function as human readable string.
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"
236 if mapping[self.function()]:
237 return mapping[self.function()]
239 return self.function()
241 def identifier(self) -> str:
243 Getter returning the TAPI Node identifier.
244 :return Object identifier as UUID.
246 return self.__data["uuid"]
248 def json(self) -> dict:
250 Getter for a json object representing the TAPI Node.
251 :return TAPI Node as json object.
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())
259 def name(self) -> str:
261 Getter for TAPI Node name.
262 :return TAPI Node as json object.
265 self.__configuration['node']['type'],
267 str(self.__configuration['node']['localId'])
270 def node_edge_point_by_cep_name(self, cep_name, local_id) -> TapiNodeEdgePoint:
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"
277 for nep in self.__data["owned-node-edge-point"]:
278 for cep in nep.connection_edge_points():
279 if cep.name() == cep_name:
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"])
288 if nep.name().endswith(str(local_id[-1])):
292 def parent(self) -> 'TapiNode':
294 Getter for a TAPI Node object representing the TAPI Node configuration.
295 :return TAPI Node configuration as json object.
299 def svg(self, x: int, y: int) -> etree.Element:
301 Getter for a xml Element object representing the TAPI Node.
302 :return TAPI Node as svg object.
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)
331 svg_nep = Node(self, x, y)
333 group: etree.Element = svg_nep.svg_element()
335 for nep in self.data()['owned-node-edge-point']:
337 if "local-id" in nep.configuration()["nodeEdgePoint"]:
338 localId = nep.configuration()["nodeEdgePoint"]["local-id"]
341 self.x_offset_by_cep_name(
342 nep.connection_edge_points()[0].name(), localId)
344 self.y_offset_by_cep_name(
345 nep.connection_edge_points()[0].name())
346 group.append(nep.svg(nep_x, nep_y))
349 def width(self, width: int) -> None:
351 Setter for the SVG width in px.
352 :param width as integer with unit "px" (pixel)
358 def add(self, nep: TapiNodeEdgePoint) -> 'TapiNode':
360 Method adding a TAPI Node Edge Point object.
361 :return TAPI Node as object.
363 self.__data['owned-node-edge-point'].append(nep)