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.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
35 Class representing a TAPI Node.
39 __configuration: dict = {}
40 __parent: 'TapiNode' = None
41 __width: int = 0 # default SVG width, should be overritten by 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
50 "uuid": str(uuid.uuid4()),
53 "value-name": "topology-node-name",
57 "value-name": "topology-node-local-id",
58 "value": configuration['node']['localId']
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": [
69 "cost-algorithm": "alg1",
70 "cost-value": "value-1"
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"
80 "o-ran-sc-topology:function": configuration['node']['function'],
81 "o-ran-sc-topology:geolocation": {
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,
98 "o2-rest-provider": 0*self.FONTSIZE,
99 "a1-rest-provider": -8*self.FONTSIZE,
100 "e2-rest-consumer": 0*self.FONTSIZE,
102 "f1-c-unknown-consumer": 0*self.FONTSIZE,
103 "f1-u-unknown-consumer": 0*self.FONTSIZE,
105 "e1-unknown-provider": -5*2*self.FONTSIZE,
106 "e1-unknown-consumer": 5*2*self.FONTSIZE,
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,
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,
123 "ofh-netconf-provider": 0*self.FONTSIZE,
124 "uu-unknown-provider": 0*self.FONTSIZE,
126 "uu-unknown-consumer": 0*self.FONTSIZE
129 return mapping[name] + 4*self.FONTSIZE * local_id
131 print("Node: CEP name", name, "for x postion calculation not found")
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,
142 "o2-rest-provider": -3*self.FONTSIZE,
143 "a1-rest-provider": -3*self.FONTSIZE,
144 "e2-rest-consumer": 3*self.FONTSIZE,
146 "e1-unknown-provider": 1*self.FONTSIZE,
147 "e1-unknown-consumer": 1*self.FONTSIZE,
149 "f1-c-unknown-consumer": 3*self.FONTSIZE,
150 "f1-u-unknown-consumer": 3*self.FONTSIZE,
151 "f1-unknown-consumer": 3*self.FONTSIZE,
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,
162 "eth-ofh-provider": -3*self.FONTSIZE,
163 "oam-netconf-provider": -3*self.FONTSIZE,
164 "eth-ofh-consumer": +3*self.FONTSIZE,
166 "ofh-netconf-provider": -3*self.FONTSIZE,
167 "uu-unknown-provider": 3*self.FONTSIZE,
169 "uu-unknown-consumer": -3*self.FONTSIZE
174 print("Node: CEP name", name, "for y postion calculation not found")
177 def configuration(self) -> dict:
179 Getter for a json object representing the TAPI Node configuration.
180 :return TAPI Node configuration as json object.
182 return self.__configuration
184 def data(self) -> dict:
186 Getter for a json object representing the TAPI Link.
187 :return TAPI Link as json object.
191 def local_id(self) -> int:
192 return self.configuration()["node"]["localId"]
194 def function(self) -> str:
196 Getter returning the network-function type
197 :return The type of the network-function as yang IDENTITY.
199 return self.__configuration['node']['function']
201 def function_label(self) -> str:
203 Getter returning the network-function label
204 :return The type of the network-function as human readable string.
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"
218 if mapping[self.function()]:
219 return mapping[self.function()]
221 return self.function()
223 def identifier(self) -> str:
225 Getter returning the TAPI Node identifier.
226 :return Object identifier as UUID.
228 return self.__data["uuid"]
230 def json(self) -> dict:
232 Getter for a json object representing the TAPI Node.
233 :return TAPI Node as json object.
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())
241 def name(self) -> str:
243 Getter for TAPI Node name.
244 :return TAPI Node as json object.
247 self.__configuration['node']['type'],
249 str(self.__configuration['node']['localId'])
252 def node_edge_point_by_cep_name(self, cep_name, local_id) -> TapiNodeEdgePoint:
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"
259 for nep in self.__data["owned-node-edge-point"]:
260 for cep in nep.connection_edge_points():
261 if cep.name() == cep_name:
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"])
269 if nep.name().endswith(str(local_id[-1])):
273 def parent(self) -> 'TapiNode':
275 Getter for a TAPI Node object representing the TAPI Node configuration.
276 :return TAPI Node configuration as json object.
280 def svg(self, x: int, y: int) -> etree.Element:
282 Getter for a xml Element object representing the TAPI Node.
283 :return TAPI Node as svg object.
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)
308 svg_nep = Node(self, x, y)
310 group: etree.Element = svg_nep.svg_element()
312 for nep in self.data()['owned-node-edge-point']:
314 if "local-id" in nep.configuration()["nodeEdgePoint"]:
315 localId = nep.configuration()["nodeEdgePoint"]["local-id"]
318 self.x_offset_by_cep_name(
319 nep.connection_edge_points()[0].name(), localId)
321 self.y_offset_by_cep_name(
322 nep.connection_edge_points()[0].name())
323 group.append(nep.svg(nep_x, nep_y))
326 def width(self, width: int) -> None:
328 Setter for the SVG width in px.
329 :param width as integer with unit "px" (pixel)
335 def add(self, nep: TapiNodeEdgePoint) -> 'TapiNode':
337 Method adding a TAPI Node Edge Point object.
338 :return TAPI Node as object.
340 self.__data['owned-node-edge-point'].append(nep)