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 main class for this project for a TAPI Topology.
20 from typing import Dict, List, Union
21 from lxml import etree
23 from model.python.top import Top
24 from model.python.tapi_node import TapiNode
25 from model.python.tapi_node_smo import TapiNodeSmo
26 from model.python.tapi_node_o_cloud import TapiNodeOCloud
27 from model.python.tapi_node_near_rt_ric import TapiNodeNearRtRic
28 from model.python.tapi_node_o_cu_cp import TapiNodeOCuCp
29 from model.python.tapi_node_o_cu_up import TapiNodeOCuUp
30 from model.python.tapi_node_o_du import TapiNodeODu
31 from model.python.tapi_node_fronthaul_gateway import TapiNodeFronthaulGateway
32 from model.python.tapi_node_o_ru import TapiNodeORu
33 from model.python.tapi_node_user_equipment import TapiNodeUserEquipment
34 from model.python.tapi_link import TapiLink
37 class TapiTopology(Top):
39 Class representing a TAPI Topology
42 __data: Dict[str, Union[str, List[Union[Dict, TapiNode, TapiLink]]]] = None
43 __configuration: dict = None
46 def __init__(self, configuration: dict):
47 super().__init__(configuration)
48 self.__configuration = configuration
50 "uuid": str(uuid.uuid4()),
52 "value-name": "network-name",
53 "value": configuration['network']['name']}],
54 "layer-protocol-name": ["ETH"],
58 topology_structure: dict = configuration['network']['pattern']
59 network_function_type: str = next(iter(topology_structure))
60 count: int = configuration['network']['pattern'][network_function_type]
62 if network_function_type == "smo":
63 self.__create_smos(None, topology_structure, count)
64 elif network_function_type == "near-rt-ric":
65 self.__create_near_rt_rics(None, topology_structure, count)
66 elif network_function_type == "o-cu":
67 self.__create_o_cus(None, topology_structure, count)
68 elif network_function_type == "o-du":
69 self.__create_o_dus(None, topology_structure, count)
70 elif network_function_type == "o-ru":
71 self.__create_o_rus(None, topology_structure, count)
72 elif network_function_type == "ue":
73 self.__create_ues(None, topology_structure, count)
75 print("Unknown network function type", network_function_type)
78 def configuration(self) -> dict:
80 Getter for a json object representing the TAPI Topology configuration.
81 :return TAPI Topology configuration as json object.
83 return self.__configuration
85 def data(self) -> dict:
87 Getter for a json object representing the TAPI Topology.
88 :return TAPI Topology as json object.
92 def identifier(self) -> str:
94 Getter returning the TAPI Topology identifier.
95 :return Object identifier as UUID.
97 return self.__data["uuid"]
99 def name(self) -> str:
101 Getter for TAPI Topology name. The TAPI topology is a representation of
102 the network. Therefore, the TAPI Topology name has the same value as the
104 :return TAPI Topology name as string.
106 return self.__configuration['network']['name']
108 def json(self) -> dict:
110 Getter for a json object representing the TAPI Topology.
111 :return TAPI Topology Context as json object.
113 result = self.data().copy()
117 for node in self.__data["node"]:
118 result["node"].append(node.json())
122 for link in self.__data["link"]:
123 result["link"].append(link.json())
127 def svg(self, svg_x: int, svg_y: int) -> etree.Element:
129 Getter for a xml Element object representing the TAPI Topology Context.
130 :return TAPI Topology Context as svg object.
132 group = etree.Element("g")
133 title = etree.Element("title")
134 title.text = "\n TAPI Topology \n id: " + \
135 self.identifier() # + "\n name: " + self.name()
139 index_per_type: Dict = {}
141 for node in self.__data["node"]:
142 if type(node) in index_per_type:
143 index_per_type[type(node)] = index_per_type[type(node)] + 1
145 index_per_type[type(node)] = 0
146 index = index_per_type[type(node)]
148 index*self.__svg_dynamic_x_offset_by_node_type(type(node)) + \
149 self.__svg_static_x_offset_by_node_type(type(node))
150 node_y = svg_y + self.__svg_y_offset_by_node_type(type(node))
151 svg_nodes.append(node.svg(node_x, node_y))
152 # group.append(node.svg(node_x, node_y))
154 # handling and drawing links
155 for link in self.__data["link"]:
156 group.append(link.svg(0, 0))
159 for svg_node in svg_nodes:
160 group.append(svg_node)
164 def __svg_static_x_offset_by_node_type(self, node_type) -> int:
166 Mapping function from node types to y position in svg
169 pattern = self.configuration()['network']['pattern']
170 padding = 0 # self.FONTSIZE
171 width_unit = (2 * 2 * self.FONTSIZE + padding)
173 ru = (pattern['user-equipment']-1) * width_unit / 2
174 fhgw = (pattern['o-ru'] *
175 pattern['user-equipment'] - 1) * width_unit / 2
176 odu = (pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
178 ocu = (pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
179 - 1) * width_unit / 2
180 ric = (pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
181 - 1) * width_unit / 2
182 smo = (pattern['smo'] * pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
185 x_mapping: Dict[type, int] = {
188 TapiNodeNearRtRic: ric,
189 TapiNodeOCuCp: ocu - 12.5*self.FONTSIZE,
190 TapiNodeOCuUp: ocu + 12.5*self.FONTSIZE,
192 TapiNodeFronthaulGateway: fhgw,
194 TapiNodeUserEquipment: 0
196 if node_type in x_mapping:
197 return x_mapping[node_type]
200 def __svg_dynamic_x_offset_by_node_type(self, node_type) -> int:
202 Mapping function from node types to y position in svg
206 pattern = self.configuration()['network']['pattern']
207 width_unit = (2 * 2 * self.FONTSIZE + padding)
208 x_mapping: Dict[type, int] = {
209 TapiNodeSmo: pattern['near-rt-ric'] * pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
210 TapiNodeOCloud: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
211 TapiNodeNearRtRic: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
212 TapiNodeOCuCp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
213 TapiNodeOCuUp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
214 TapiNodeODu: pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
215 TapiNodeFronthaulGateway: pattern['o-ru'] * pattern['user-equipment'],
216 TapiNodeORu: pattern['user-equipment'],
217 TapiNodeUserEquipment: 1
219 if node_type in x_mapping:
220 return x_mapping[node_type] * width_unit
223 def __svg_y_offset_by_node_type(self, node_type) -> int:
225 Mapping function from node types to y position in svg
228 offset = 11*self.FONTSIZE
229 y_mapping: Dict[type, int] = {
230 TapiNodeSmo: 0 * offset,
231 TapiNodeOCloud: 1 * offset,
232 TapiNodeNearRtRic: 2 * offset,
233 TapiNodeOCuCp: 3.5 * offset - 4 * self.FONTSIZE,
234 TapiNodeOCuUp: 3.5 * offset + 4 * self.FONTSIZE,
235 TapiNodeODu: 5 * offset,
236 TapiNodeFronthaulGateway: 6 * offset,
237 TapiNodeORu: 7 * offset,
238 TapiNodeUserEquipment: 8 * offset
240 if node_type in y_mapping:
241 return y_mapping[node_type]
245 def add_node(self, node: TapiNode):
247 Method adding a TAPI node to TAPI Topology.
248 :return TAPI Topology object.
250 self.__data["node"].append(node)
253 def add_link(self, link: TapiLink):
255 Method adding a TAPI node to TAPI Topology.
256 :return TAPI Topology object.
258 self.__data["link"].append(link)
261 def __create_smos(self, parent: TapiNode, topology_structure: dict, count: int):
263 Method adding a TAPI node to TAPI Topology.
264 :param parent: A TAPI node which acts a a parent node in the topology.
265 :param topology_structure: Information about the next topology levels.
266 :param count: Number of instance to be created
267 :return TAPI Topology object.
270 next_type = "near-rt-ric"
271 for local_id in range(count):
274 if parent is not None:
275 prefix = parent.data()["name"][1]["value"]
276 config = {"node": {"localId": prefix + str(local_id),
277 "type": current_type,
278 "function": "o-ran-sc-topology-common:"+current_type}}
279 node = TapiNodeSmo(parent, config)
283 if "o-cloud" in topology_structure:
284 structure = topology_structure.copy()
285 self.__create_o_clouds(
286 node, structure, structure["o-cloud"])
288 if next_type in topology_structure:
289 structure = topology_structure.copy()
290 if current_type in structure:
291 del structure["o-cloud"]
292 if current_type in structure:
293 del structure[current_type]
294 self.__create_near_rt_rics(
295 node, structure, structure[next_type])
299 def __create_o_clouds(self, parent: TapiNode, topology_structure: dict, count: int):
301 Method adding a TAPI node to TAPI Topology.
302 :param parent: A TAPI node which acts a a parent node in the topology.
303 :param topology_structure: Information about the next topology levels.
304 :param count: Number of instance to be created
305 :return TAPI Topology object.
307 current_type = "o-cloud"
308 for local_id in range(count):
311 if parent is not None:
312 prefix = parent.json()["name"][1]["value"]
313 function = "o-ran-sc-topology-common:"+current_type
314 node_configuration = {"node": {"localId": prefix + str(local_id),
315 "type": current_type,
316 "function": function}}
317 node = TapiNodeOCloud(parent, node_configuration)
322 link_configuration = {
323 "topology_reference": self.data()["uuid"],
324 "name_prefix": "o2-rest",
328 self.add_link(TapiLink(link_configuration))
331 def __create_near_rt_rics(self, parent: TapiNode, topology_structure: dict, count: int):
333 Method adding a TAPI node to TAPI Topology.
334 :param parent: A TAPI node which acts a a parent node in the topology.
335 :param topology_structure: Information about the next topology levels.
336 :param count: Number of instance to be created
337 :return TAPI Topology object.
339 current_type = "near-rt-ric"
341 for local_id in range(count):
344 if parent is not None:
345 prefix = parent.json()["name"][1]["value"]
346 function = "o-ran-sc-topology-common:"+current_type
347 node_configuration = {"node": {"localId": prefix + str(local_id),
348 "type": current_type,
349 "function": function}}
350 node = TapiNodeNearRtRic(parent, node_configuration)
355 link_configuration = {
356 "topology_reference": self.data()["uuid"],
357 "name_prefix": "a1-rest",
361 self.add_link(TapiLink(link_configuration))
364 link_configuration = {
365 "topology_reference": self.data()["uuid"],
366 "name_prefix": "o1-netconf",
370 self.add_link(TapiLink(link_configuration))
373 link_configuration = {
374 "topology_reference": self.data()["uuid"],
375 "name_prefix": "o1-file",
379 self.add_link(TapiLink(link_configuration))
382 link_configuration = {
383 "topology_reference": self.data()["uuid"],
384 "name_prefix": "o1-ves",
388 self.add_link(TapiLink(link_configuration))
391 if next_type in topology_structure:
392 structure = topology_structure.copy()
393 if current_type in structure:
394 del structure[current_type]
395 self.__create_o_cus(node, structure, structure[next_type])
399 def __function_identity(self, function_type: str, plane: str) -> str:
401 Method to calculate the Function IDENTITY
404 "o-ran-sc-topology-common:",
410 def __create_o_cus(self, parent: TapiNode, topology_structure: dict, count: int):
412 Method adding a TAPI node to TAPI Topology.
413 :param parent: A TAPI node which acts a a parent node in the topology.
414 :param topology_structure: Information about the next topology levels.
415 :param count: Number of instance to be created
416 :return TAPI Topology object.
418 current_type = "o-cu"
420 for local_id in range(count):
422 if parent is not None:
423 prefix = parent.data()["name"][1]["value"]
425 node: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {}
426 for plane in ["cp", "up"]:
427 config = {"node": {"localId": prefix + str(local_id),
428 "type": "-".join([current_type, plane]),
429 "function": self.__function_identity(current_type, plane)}}
430 classes: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {
433 node[plane] = classes[plane](parent, config)
434 self.add_node(node[plane])
438 link_configuration = {
439 "topology_reference": self.data()["uuid"],
440 "name_prefix": "e2-rest",
441 "provider": node[plane],
444 self.add_link(TapiLink(link_configuration))
447 link_configuration = {
448 "topology_reference": self.data()["uuid"],
449 "name_prefix": "o1-netconf",
450 "provider": node[plane],
451 "consumer": parent.parent()
453 self.add_link(TapiLink(link_configuration))
456 link_configuration = {
457 "topology_reference": self.data()["uuid"],
458 "name_prefix": "o1-file",
459 "provider": node[plane],
460 "consumer": parent.parent()
462 self.add_link(TapiLink(link_configuration))
465 link_configuration = {
466 "topology_reference": self.data()["uuid"],
467 "name_prefix": "o1-ves",
468 "provider": parent.parent(),
469 "consumer": node[plane]
471 self.add_link(TapiLink(link_configuration))
474 # E1 Interface between O-CU-UP and O-CU-CP
475 link_configuration = {
476 "topology_reference": self.data()["uuid"],
477 "name_prefix": "e1-unknown",
478 "provider": node["up"],
479 "consumer": node["cp"]
481 self.add_link(TapiLink(link_configuration))
483 if next_type in topology_structure:
484 structure = topology_structure.copy()
485 if current_type in structure:
486 del structure[current_type]
488 node, structure, structure[next_type])
491 def __create_o_dus(self, parents: Dict[str, TapiNode], topology_structure: dict, count: int):
493 Method adding a TAPI node to TAPI Topology.
494 :param parent: A TAPI node which acts a a parent node in the topology.
495 :param topology_structure: Information about the next topology levels.
496 :param count: Number of instance to be created
497 :return TAPI Topology object.
499 current_type = "o-du"
500 next_type = "fronthaul-gateway"
501 for local_id in range(count):
503 if parents["cp"] is not None:
504 prefix = parents["cp"].data()["name"][1]["value"]
505 config = {"node": {"localId": prefix + str(local_id),
506 "type": current_type,
507 "function": "o-ran-sc-topology-common:"+current_type}}
508 node = TapiNodeODu(parents["cp"], config)
511 for plane, parent in parents.items():
515 link_configuration = {
516 "topology_reference": self.data()["uuid"],
517 "name_prefix": "e2-rest",
519 "consumer": parent.parent()
521 self.add_link(TapiLink(link_configuration))
524 link_configuration = {
525 "topology_reference": self.data()["uuid"],
526 "name_prefix": "o1-netconf",
528 "consumer": parent.parent().parent()
530 self.add_link(TapiLink(link_configuration))
533 link_configuration = {
534 "topology_reference": self.data()["uuid"],
535 "name_prefix": "o1-file",
537 "consumer": parent.parent().parent()
539 self.add_link(TapiLink(link_configuration))
542 link_configuration = {
543 "topology_reference": self.data()["uuid"],
544 "name_prefix": "o1-ves",
545 "provider": parent.parent().parent(),
548 self.add_link(TapiLink(link_configuration))
550 # F1 User Plane or Control Plane
551 interfaces: Dict[str, str] = {"cp": "f1-c", "up": "f1-u"}
552 link_configuration = {
553 "topology_reference": self.data()["uuid"],
554 "name_prefix": interfaces[plane]+"-unknown",
558 self.add_link(TapiLink(link_configuration))
561 if next_type in topology_structure:
562 structure = topology_structure.copy()
563 if current_type in structure:
564 del structure[current_type]
565 self.__create_fronthaul_gateways(
566 node, structure, structure[next_type])
569 def __create_fronthaul_gateways(self, parent: TapiNode, topology_structure: dict, count: int):
571 Method adding a TAPI node to TAPI Topology.
572 :param parent: A TAPI node which acts a a parent node in the topology.
573 :param topology_structure: Information about the next topology levels.
574 :param count: Number of instance to be created
575 :return TAPI Topology object.
577 current_type = "fronthaul-gateway"
579 for local_id in range(count):
581 if parent is not None:
582 prefix = parent.data()["name"][1]["value"]
583 node_configuration = {
585 "localId": prefix + str(local_id),
586 "type": current_type,
587 "function": "o-ran-sc-topology-common:"+current_type,
588 "southbound-nep-count": topology_structure[next_type]
591 node = TapiNodeFronthaulGateway(parent, node_configuration)
597 link_configuration = {
598 "topology_reference": self.data()["uuid"],
599 "name_prefix": "oam-netconf",
601 "consumer": parent.parent().parent().parent()
603 self.add_link(TapiLink(link_configuration))
606 link_configuration = {
607 "topology_reference": self.data()["uuid"],
608 "name_prefix": "eth-ofh",
612 self.add_link(TapiLink(link_configuration))
615 if next_type in topology_structure:
616 structure = topology_structure.copy()
617 if current_type in structure:
618 del structure[current_type]
619 self.__create_o_rus(node, structure, structure[next_type])
622 def __create_o_rus(self, parent: TapiNode, topology_structure: dict, count: int):
624 Method adding a TAPI node to TAPI Topology.
625 :param parent: A TAPI node which acts a a parent node in the topology.
626 :param topology_structure: Information about the next topology levels.
627 :param count: Number of instance to be created
628 :return TAPI Topology object.
630 current_type = "o-ru"
631 next_type = "user-equipment"
632 for local_id in range(count):
634 if parent is not None:
635 prefix = parent.data()["name"][1]["value"]
636 config = {"node": {"localId": prefix + str(local_id),
637 "type": current_type,
638 "function": "o-ran-sc-topology-common:"+current_type}}
639 node = TapiNodeORu(parent, config)
645 link_configuration = {
646 "topology_reference": self.data()["uuid"],
647 "name_prefix": "ofh-netconf",
649 "consumer": parent.parent().parent().parent().parent()
651 self.add_link(TapiLink(link_configuration))
653 # OFH M-Plane to O-DU via fronthaul-gateway
654 link_configuration = {
655 "topology_reference": self.data()["uuid"],
656 "name_prefix": "ofh-netconf",
660 self.add_link(TapiLink(link_configuration))
663 if next_type in topology_structure:
664 structure = topology_structure.copy()
665 if current_type in structure:
666 del structure[current_type]
667 self.__create_ues(node, structure, structure[next_type])
670 def __create_ues(self, parent: TapiNode, topology_structure: dict, count: int):
672 Method adding a TAPI node to TAPI Topology.
673 :param parent: A TAPI node which acts a a parent node in the topology.
674 :param topology_structure: Information about the next topology levels.
675 :param count: Number of instance to be created
676 :return TAPI Topology object.
678 current_type = "user-equipment"
679 for local_id in range(count):
681 if parent is not None:
682 prefix = parent.data()["name"][1]["value"]
683 config = {"node": {"localId": prefix + str(local_id),
684 "type": current_type,
685 "function": "o-ran-sc-topology-common:"+current_type}}
686 node = TapiNodeUserEquipment(parent, config)
691 link_configuration = {
692 "topology_reference": self.data()["uuid"],
693 "name_prefix": "uu-unknown",
697 self.add_link(TapiLink(link_configuration))
699 if "key" in topology_structure:
700 print("Implement missing topology level.")