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_amf import TapiNodeAmf
29 from model.python.tapi_node_upf import TapiNodeUpf
30 from model.python.tapi_node_o_cu_cp import TapiNodeOCuCp
31 from model.python.tapi_node_o_cu_up import TapiNodeOCuUp
32 from model.python.tapi_node_o_du import TapiNodeODu
33 from model.python.tapi_node_fronthaul_gateway import TapiNodeFronthaulGateway
34 from model.python.tapi_node_o_ru import TapiNodeORu
35 from model.python.tapi_node_user_equipment import TapiNodeUserEquipment
36 from model.python.tapi_link import TapiLink
39 class TapiTopology(Top):
41 Class representing a TAPI Topology
44 __data: Dict[str, Union[str, List[Union[Dict, TapiNode, TapiLink]]]] = None
45 __configuration: dict = None
48 def __init__(self, configuration: dict):
49 super().__init__(configuration)
50 name = "o-ran-sc-topology-view"
51 if "name" in configuration['network']:
52 name = configuration['network']['name']
54 self.__configuration = configuration
56 "uuid": str(uuid.uuid4()),
58 "value-name": "network-name",
60 "layer-protocol-name": ["ETH"],
64 topology_structure: dict = configuration['network']['pattern']
65 network_function_type: str = next(iter(topology_structure))
66 count: int = configuration['network']['pattern'][network_function_type]
68 if network_function_type == "smo":
69 self.__create_smos(None, topology_structure, count)
70 elif network_function_type == "near-rt-ric":
71 self.__create_near_rt_rics(None, topology_structure, count)
72 elif network_function_type == "o-cu":
73 self.__create_o_cus(None, topology_structure, count)
74 elif network_function_type == "o-du":
75 self.__create_o_dus(None, topology_structure, count)
76 elif network_function_type == "o-ru":
77 self.__create_o_rus(None, topology_structure, count)
78 elif network_function_type == "ue":
79 self.__create_ues(None, topology_structure, count)
81 print("Unknown network function type", network_function_type)
84 def configuration(self) -> dict:
86 Getter for a json object representing the TAPI Topology configuration.
87 :return TAPI Topology configuration as json object.
89 return self.__configuration
91 def data(self) -> dict:
93 Getter for a json object representing the TAPI Topology.
94 :return TAPI Topology as json object.
98 def identifier(self) -> str:
100 Getter returning the TAPI Topology identifier.
101 :return Object identifier as UUID.
103 return self.__data["uuid"]
105 def name(self) -> str:
107 Getter for TAPI Topology name. The TAPI topology is a representation of
108 the network. Therefore, the TAPI Topology name has the same value as the
110 :return TAPI Topology name as string.
112 return self.__configuration['network']['name']
114 def json(self) -> dict:
116 Getter for a json object representing the TAPI Topology.
117 :return TAPI Topology Context as json object.
119 result = self.data().copy()
123 for node in self.__data["node"]:
124 result["node"].append(node.json())
128 for link in self.__data["link"]:
129 result["link"].append(link.json())
133 def svg(self, svg_x: int, svg_y: int) -> etree.Element:
135 Getter for a xml Element object representing the TAPI Topology Context.
136 :return TAPI Topology Context as svg object.
138 group = etree.Element("g")
139 title = etree.Element("title")
140 title.text = "\n TAPI Topology \n id: " + \
141 self.identifier() # + "\n name: " + self.name()
145 index_per_type: Dict = {}
147 for node in self.__data["node"]:
148 if type(node) in index_per_type:
149 index_per_type[type(node)] = index_per_type[type(node)] + 1
151 index_per_type[type(node)] = 0
152 index = index_per_type[type(node)]
154 index*self.__svg_dynamic_x_offset_by_node_type(type(node)) + \
155 self.__svg_static_x_offset_by_node_type(type(node))
156 node_y = svg_y + self.__svg_y_offset_by_node_type(type(node))
157 svg_nodes.append(node.svg(node_x, node_y))
158 # group.append(node.svg(node_x, node_y))
160 # handling and drawing links
161 for link in self.__data["link"]:
162 group.append(link.svg(0, 0))
165 for svg_node in svg_nodes:
166 group.append(svg_node)
170 def __svg_static_x_offset_by_node_type(self, node_type) -> int:
172 Mapping function from node types to y position in svg
175 pattern = self.configuration()['network']['pattern']
176 padding = 0 # self.FONTSIZE
177 width_unit = (2 * 2 * self.FONTSIZE + padding)
179 ru = (pattern['user-equipment']-1) * width_unit - 2*self.FONTSIZE
180 fhgw = (pattern['o-ru'] *
181 pattern['user-equipment'] - 1) * width_unit - 2*self.FONTSIZE
182 odu = (pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
183 - 1) * width_unit - 2*self.FONTSIZE
184 ocu = (pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
185 - 1) * width_unit - 2*self.FONTSIZE
186 ric = (pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
187 - 1) * width_unit - 2*self.FONTSIZE
188 smo = (pattern['smo'] * pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
189 -0.5) * width_unit * 2 - 2*self.FONTSIZE
191 x_mapping: Dict[type, int] = {
194 TapiNodeNearRtRic: ric,
195 TapiNodeAmf: ric + 16*self.FONTSIZE,
196 TapiNodeUpf: ric + 24*self.FONTSIZE,
197 TapiNodeOCuCp: ocu - 12.5*self.FONTSIZE,
198 TapiNodeOCuUp: ocu + 12.5*self.FONTSIZE,
200 TapiNodeFronthaulGateway: fhgw,
202 TapiNodeUserEquipment: 0
204 if node_type in x_mapping:
205 return x_mapping[node_type]
208 def __svg_dynamic_x_offset_by_node_type(self, node_type) -> int:
210 Mapping function from node types to y position in svg
214 pattern = self.configuration()['network']['pattern']
215 width_unit = (2 * 2 * self.FONTSIZE + padding)
216 x_mapping: Dict[type, int] = {
217 TapiNodeSmo: pattern['near-rt-ric'] * pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
218 TapiNodeOCloud: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
219 TapiNodeNearRtRic: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
220 TapiNodeAmf: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
221 TapiNodeUpf: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
222 TapiNodeOCuCp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
223 TapiNodeOCuUp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
224 TapiNodeODu: pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
225 TapiNodeFronthaulGateway: pattern['o-ru'] * pattern['user-equipment']*2,
226 TapiNodeORu: pattern['user-equipment']*2,
227 TapiNodeUserEquipment: 2
229 if node_type in x_mapping:
230 return x_mapping[node_type] * width_unit
233 def __svg_y_offset_by_node_type(self, node_type) -> int:
235 Mapping function from node types to y position in svg
238 offset = 11*self.FONTSIZE
239 y_mapping: Dict[type, int] = {
240 TapiNodeSmo: 0 * offset,
241 TapiNodeOCloud: 1 * offset,
242 TapiNodeNearRtRic: 2 * offset,
243 TapiNodeAmf: 2 * offset,
244 TapiNodeUpf: 2 * offset,
245 TapiNodeOCuCp: 3.5 * offset - 4 * self.FONTSIZE,
246 TapiNodeOCuUp: 3.5 * offset + 4 * self.FONTSIZE,
247 TapiNodeODu: 5 * offset,
248 TapiNodeFronthaulGateway: 6 * offset,
249 TapiNodeORu: 7 * offset,
250 TapiNodeUserEquipment: 8 * offset
252 if node_type in y_mapping:
253 return y_mapping[node_type]
257 def add_node(self, node: TapiNode):
259 Method adding a TAPI node to TAPI Topology.
260 :return TAPI Topology object.
262 self.__data["node"].append(node)
265 def add_link(self, link: TapiLink):
267 Method adding a TAPI node to TAPI Topology.
268 :return TAPI Topology object.
270 self.__data["link"].append(link)
273 def find_node(self, type_name:str, client_local_id:str) -> TapiNode:
275 Method finding the parent no in the hierarchy based on the node type name
276 and the client local identifier.
277 :param type_name: the TAPI NODE Type name of interest.
278 :param client_local_id: which starts with the parent node local id.
280 turn TAPI Node object.
282 for node in self.__data["node"]:
283 if type(node).__name__ == type_name:
284 if client_local_id.startswith(node.local_id()):
288 def __create_smos(self, parent: TapiNode, topology_structure: dict, count: int):
290 Method adding a TAPI node to TAPI Topology.
291 :param parent: A TAPI node which acts a a parent node in the topology.
292 :param topology_structure: Information about the next topology levels.
293 :param count: Number of instance to be created
294 :return TAPI Topology object.
297 next_type = "near-rt-ric"
298 for local_id in range(count):
301 if parent is not None:
302 prefix = parent.data()["name"][1]["value"]
303 config = {"node": {"localId": prefix + str(local_id),
304 "type": current_type,
305 "function": "o-ran-sc-topology-common:"+current_type}}
306 node = TapiNodeSmo(parent, config)
310 if "o-cloud" in topology_structure:
311 structure = topology_structure.copy()
312 self.__create_o_clouds(
313 node, structure, structure["o-cloud"])
315 if next_type in topology_structure:
316 structure = topology_structure.copy()
317 if current_type in structure:
318 del structure["o-cloud"]
319 if current_type in structure:
320 del structure[current_type]
321 self.__create_near_rt_rics(
322 node, structure, structure[next_type])
326 def __create_o_clouds(self, parent: TapiNode, topology_structure: dict, count: int):
328 Method adding a TAPI node to TAPI Topology.
329 :param parent: A TAPI node which acts a a parent node in the topology.
330 :param topology_structure: Information about the next topology levels.
331 :param count: Number of instance to be created
332 :return TAPI Topology object.
334 current_type = "o-cloud"
335 for local_id in range(count):
338 if parent is not None:
339 prefix = parent.json()["name"][1]["value"]
340 function = "o-ran-sc-topology-common:"+current_type
341 node_configuration = {"node": {"localId": prefix + str(local_id),
342 "type": current_type,
343 "function": function}}
344 node = TapiNodeOCloud(parent, node_configuration)
349 link_configuration = {
350 "topology_reference": self.data()["uuid"],
351 "name_prefix": "o2-rest",
355 self.add_link(TapiLink(link_configuration))
358 def __create_near_rt_rics(self, parent: TapiNode, topology_structure: dict, count: int):
360 Method adding a TAPI node to TAPI Topology.
361 :param parent: A TAPI node which acts a a parent node in the topology.
362 :param topology_structure: Information about the next topology levels.
363 :param count: Number of instance to be created
364 :return TAPI Topology object.
366 current_type = "near-rt-ric"
368 for local_id in range(count):
371 if parent is not None:
372 prefix = parent.json()["name"][1]["value"]
375 # create 1 5G-Core AMF and 1 5G-Core UPF instance per Near-RT-RIC
378 current_type = "access-and-mobility-management-function"
379 function = "o-ran-sc-topology-common:"+current_type
380 node_configuration = {"node": {"localId": prefix + str(local_id),
381 "type": current_type,
382 "function": function}}
383 node = TapiNodeAmf(parent, node_configuration)
387 current_type = "user-plane-function"
388 function = "o-ran-sc-topology-common:"+current_type
389 node_configuration = {"node": {"localId": prefix + str(local_id),
390 "type": current_type,
391 "function": function}}
392 node = TapiNodeUpf(parent, node_configuration)
396 current_type = "near-rt-ric"
397 function = "o-ran-sc-topology-common:"+current_type
398 node_configuration = {"node": {"localId": prefix + str(local_id),
399 "type": current_type,
400 "function": function}}
401 node = TapiNodeNearRtRic(parent, node_configuration)
406 link_configuration = {
407 "topology_reference": self.data()["uuid"],
408 "name_prefix": "a1-rest",
412 self.add_link(TapiLink(link_configuration))
415 link_configuration = {
416 "topology_reference": self.data()["uuid"],
417 "name_prefix": "o1-netconf",
421 self.add_link(TapiLink(link_configuration))
424 link_configuration = {
425 "topology_reference": self.data()["uuid"],
426 "name_prefix": "o1-file",
430 self.add_link(TapiLink(link_configuration))
433 link_configuration = {
434 "topology_reference": self.data()["uuid"],
435 "name_prefix": "o1-ves",
439 self.add_link(TapiLink(link_configuration))
442 if next_type in topology_structure:
443 structure = topology_structure.copy()
444 if current_type in structure:
445 del structure[current_type]
446 self.__create_o_cus(node, structure, structure[next_type])
450 def __function_identity(self, function_type: str, plane: str) -> str:
452 Method to calculate the Function IDENTITY
455 "o-ran-sc-topology-common:",
461 def __create_o_cus(self, parent: TapiNode, topology_structure: dict, count: int):
463 Method adding a TAPI node to TAPI Topology.
464 :param parent: A TAPI node which acts a a parent node in the topology.
465 :param topology_structure: Information about the next topology levels.
466 :param count: Number of instance to be created
467 :return TAPI Topology object.
469 current_type = "o-cu"
471 for local_id in range(count):
473 if parent is not None:
474 prefix = parent.data()["name"][1]["value"]
476 node: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {}
477 for plane in ["cp", "up"]:
478 config = {"node": {"localId": prefix + str(local_id),
479 "type": "-".join([current_type, plane]),
480 "function": self.__function_identity(current_type, plane)}}
481 classes: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {
484 node[plane] = classes[plane](parent, config)
485 self.add_node(node[plane])
489 link_configuration = {
490 "topology_reference": self.data()["uuid"],
491 "name_prefix": "e2-rest",
492 "provider": node[plane],
495 self.add_link(TapiLink(link_configuration))
498 link_configuration = {
499 "topology_reference": self.data()["uuid"],
500 "name_prefix": "o1-netconf",
501 "provider": node[plane],
502 "consumer": parent.parent()
504 self.add_link(TapiLink(link_configuration))
507 link_configuration = {
508 "topology_reference": self.data()["uuid"],
509 "name_prefix": "o1-file",
510 "provider": node[plane],
511 "consumer": parent.parent()
513 self.add_link(TapiLink(link_configuration))
516 link_configuration = {
517 "topology_reference": self.data()["uuid"],
518 "name_prefix": "o1-ves",
519 "provider": parent.parent(),
520 "consumer": node[plane]
522 self.add_link(TapiLink(link_configuration))
525 # E1 Interface between O-CU-UP and O-CU-CP
526 link_configuration = {
527 "topology_reference": self.data()["uuid"],
528 "name_prefix": "e1-unknown",
529 "provider": node["up"],
530 "consumer": node["cp"]
532 self.add_link(TapiLink(link_configuration))
534 # N2 Interface between O-CU-CP and AMF
535 link_configuration = {
536 "topology_reference": self.data()["uuid"],
537 "name_prefix": "n2-nas",
538 "provider": self.find_node("TapiNodeAmf", node["cp"].local_id()),
539 "consumer": node["cp"]
541 self.add_link(TapiLink(link_configuration))
543 # N3 Interface between O-CU-UP and UPF
544 link_configuration = {
545 "topology_reference": self.data()["uuid"],
546 "name_prefix": "n3-nas",
547 "provider": self.find_node("TapiNodeUpf", node["up"].local_id()),
548 "consumer": node["up"]
550 self.add_link(TapiLink(link_configuration))
552 if next_type in topology_structure:
553 structure = topology_structure.copy()
554 if current_type in structure:
555 del structure[current_type]
557 node, structure, structure[next_type])
560 def __create_o_dus(self, parents: Dict[str, TapiNode], topology_structure: dict, count: int):
562 Method adding a TAPI node to TAPI Topology.
563 :param parent: A TAPI node which acts a a parent node in the topology.
564 :param topology_structure: Information about the next topology levels.
565 :param count: Number of instance to be created
566 :return TAPI Topology object.
568 current_type = "o-du"
569 next_type = "fronthaul-gateway"
570 for local_id in range(count):
572 if parents["cp"] is not None:
573 prefix = parents["cp"].data()["name"][1]["value"]
574 config = {"node": {"localId": prefix + str(local_id),
575 "type": current_type,
576 "function": "o-ran-sc-topology-common:"+current_type}}
577 node = TapiNodeODu(parents["cp"], config)
580 for plane, parent in parents.items():
584 link_configuration = {
585 "topology_reference": self.data()["uuid"],
586 "name_prefix": "e2-rest",
588 "consumer": parent.parent()
590 self.add_link(TapiLink(link_configuration))
593 link_configuration = {
594 "topology_reference": self.data()["uuid"],
595 "name_prefix": "o1-netconf",
597 "consumer": parent.parent().parent()
599 self.add_link(TapiLink(link_configuration))
602 link_configuration = {
603 "topology_reference": self.data()["uuid"],
604 "name_prefix": "o1-file",
606 "consumer": parent.parent().parent()
608 self.add_link(TapiLink(link_configuration))
611 link_configuration = {
612 "topology_reference": self.data()["uuid"],
613 "name_prefix": "o1-ves",
614 "provider": parent.parent().parent(),
617 self.add_link(TapiLink(link_configuration))
619 # F1 User Plane or Control Plane
620 interfaces: Dict[str, str] = {"cp": "f1-c", "up": "f1-u"}
621 link_configuration = {
622 "topology_reference": self.data()["uuid"],
623 "name_prefix": interfaces[plane]+"-unknown",
627 self.add_link(TapiLink(link_configuration))
630 if next_type in topology_structure:
631 structure = topology_structure.copy()
632 if current_type in structure:
633 del structure[current_type]
634 self.__create_fronthaul_gateways(
635 node, structure, structure[next_type])
638 def __create_fronthaul_gateways(self, parent: TapiNode, topology_structure: dict, count: int):
640 Method adding a TAPI node to TAPI Topology.
641 :param parent: A TAPI node which acts a a parent node in the topology.
642 :param topology_structure: Information about the next topology levels.
643 :param count: Number of instance to be created
644 :return TAPI Topology object.
646 current_type = "fronthaul-gateway"
648 for local_id in range(count):
650 if parent is not None:
651 prefix = parent.data()["name"][1]["value"]
652 node_configuration = {
654 "localId": prefix + str(local_id),
655 "type": current_type,
656 "function": "o-ran-sc-topology-common:"+current_type,
657 "southbound-nep-count": topology_structure[next_type]
660 node = TapiNodeFronthaulGateway(parent, node_configuration)
666 link_configuration = {
667 "topology_reference": self.data()["uuid"],
668 "name_prefix": "oam-netconf",
670 "consumer": parent.parent().parent().parent()
672 self.add_link(TapiLink(link_configuration))
675 link_configuration = {
676 "topology_reference": self.data()["uuid"],
677 "name_prefix": "eth-ofh",
681 self.add_link(TapiLink(link_configuration))
684 if next_type in topology_structure:
685 structure = topology_structure.copy()
686 if current_type in structure:
687 del structure[current_type]
688 self.__create_o_rus(node, structure, structure[next_type])
691 def __create_o_rus(self, parent: TapiNode, topology_structure: dict, count: int):
693 Method adding a TAPI node to TAPI Topology.
694 :param parent: A TAPI node which acts a a parent node in the topology.
695 :param topology_structure: Information about the next topology levels.
696 :param count: Number of instance to be created
697 :return TAPI Topology object.
699 current_type = "o-ru"
700 next_type = "user-equipment"
701 for local_id in range(count):
703 if parent is not None:
704 prefix = parent.data()["name"][1]["value"]
705 config = {"node": {"localId": prefix + str(local_id),
706 "type": current_type,
707 "function": "o-ran-sc-topology-common:"+current_type}}
708 node = TapiNodeORu(parent, config)
714 link_configuration = {
715 "topology_reference": self.data()["uuid"],
716 "name_prefix": "ofh-netconf",
718 "consumer": parent.parent().parent().parent().parent()
720 self.add_link(TapiLink(link_configuration))
722 # OFH M-Plane to O-DU via fronthaul-gateway
723 link_configuration = {
724 "topology_reference": self.data()["uuid"],
725 "name_prefix": "ofh-netconf",
729 self.add_link(TapiLink(link_configuration))
732 if next_type in topology_structure:
733 structure = topology_structure.copy()
734 if current_type in structure:
735 del structure[current_type]
736 self.__create_ues(node, structure, structure[next_type])
739 def __create_ues(self, parent: TapiNode, topology_structure: dict, count: int):
741 Method adding a TAPI node to TAPI Topology.
742 :param parent: A TAPI node which acts a a parent node in the topology.
743 :param topology_structure: Information about the next topology levels.
744 :param count: Number of instance to be created
745 :return TAPI Topology object.
747 current_type = "user-equipment"
748 for local_id in range(count):
750 if parent is not None:
751 prefix = parent.data()["name"][1]["value"]
752 config = {"node": {"localId": prefix + str(local_id),
753 "type": current_type,
754 "function": "o-ran-sc-topology-common:"+current_type}}
755 node = TapiNodeUserEquipment(parent, config)
760 link_configuration = {
761 "topology_reference": self.data()["uuid"],
762 "name_prefix": "uu-radio",
766 self.add_link(TapiLink(link_configuration))
769 link_configuration = {
770 "topology_reference": self.data()["uuid"],
771 "name_prefix": "n1-nas",
772 "provider": self.find_node("TapiNodeAmf", node.local_id()),
775 self.add_link(TapiLink(link_configuration))
777 if "key" in topology_structure:
778 print("Implement missing topology level.")