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 self.__configuration = configuration
52 "uuid": str(uuid.uuid4()),
54 "value-name": "network-name",
55 "value": configuration['network']['name']}],
56 "layer-protocol-name": ["ETH"],
60 topology_structure: dict = configuration['network']['pattern']
61 network_function_type: str = next(iter(topology_structure))
62 count: int = configuration['network']['pattern'][network_function_type]
64 if network_function_type == "smo":
65 self.__create_smos(None, topology_structure, count)
66 elif network_function_type == "near-rt-ric":
67 self.__create_near_rt_rics(None, topology_structure, count)
68 elif network_function_type == "o-cu":
69 self.__create_o_cus(None, topology_structure, count)
70 elif network_function_type == "o-du":
71 self.__create_o_dus(None, topology_structure, count)
72 elif network_function_type == "o-ru":
73 self.__create_o_rus(None, topology_structure, count)
74 elif network_function_type == "ue":
75 self.__create_ues(None, topology_structure, count)
77 print("Unknown network function type", network_function_type)
80 def configuration(self) -> dict:
82 Getter for a json object representing the TAPI Topology configuration.
83 :return TAPI Topology configuration as json object.
85 return self.__configuration
87 def data(self) -> dict:
89 Getter for a json object representing the TAPI Topology.
90 :return TAPI Topology as json object.
94 def identifier(self) -> str:
96 Getter returning the TAPI Topology identifier.
97 :return Object identifier as UUID.
99 return self.__data["uuid"]
101 def name(self) -> str:
103 Getter for TAPI Topology name. The TAPI topology is a representation of
104 the network. Therefore, the TAPI Topology name has the same value as the
106 :return TAPI Topology name as string.
108 return self.__configuration['network']['name']
110 def json(self) -> dict:
112 Getter for a json object representing the TAPI Topology.
113 :return TAPI Topology Context as json object.
115 result = self.data().copy()
119 for node in self.__data["node"]:
120 result["node"].append(node.json())
124 for link in self.__data["link"]:
125 result["link"].append(link.json())
129 def svg(self, svg_x: int, svg_y: int) -> etree.Element:
131 Getter for a xml Element object representing the TAPI Topology Context.
132 :return TAPI Topology Context as svg object.
134 group = etree.Element("g")
135 title = etree.Element("title")
136 title.text = "\n TAPI Topology \n id: " + \
137 self.identifier() # + "\n name: " + self.name()
141 index_per_type: Dict = {}
143 for node in self.__data["node"]:
144 if type(node) in index_per_type:
145 index_per_type[type(node)] = index_per_type[type(node)] + 1
147 index_per_type[type(node)] = 0
148 index = index_per_type[type(node)]
150 index*self.__svg_dynamic_x_offset_by_node_type(type(node)) + \
151 self.__svg_static_x_offset_by_node_type(type(node))
152 node_y = svg_y + self.__svg_y_offset_by_node_type(type(node))
153 svg_nodes.append(node.svg(node_x, node_y))
154 # group.append(node.svg(node_x, node_y))
156 # handling and drawing links
157 for link in self.__data["link"]:
158 group.append(link.svg(0, 0))
161 for svg_node in svg_nodes:
162 group.append(svg_node)
166 def __svg_static_x_offset_by_node_type(self, node_type) -> int:
168 Mapping function from node types to y position in svg
171 pattern = self.configuration()['network']['pattern']
172 padding = 0 # self.FONTSIZE
173 width_unit = (2 * 2 * self.FONTSIZE + padding)
175 ru = (pattern['user-equipment']-1) * width_unit - 2*self.FONTSIZE
176 fhgw = (pattern['o-ru'] *
177 pattern['user-equipment'] - 1) * width_unit - 2*self.FONTSIZE
178 odu = (pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
179 - 1) * width_unit - 2*self.FONTSIZE
180 ocu = (pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
181 - 1) * width_unit - 2*self.FONTSIZE
182 ric = (pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
183 - 1) * width_unit - 2*self.FONTSIZE
184 smo = (pattern['smo'] * pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
185 -0.5) * width_unit * 2 - 2*self.FONTSIZE
187 x_mapping: Dict[type, int] = {
190 TapiNodeNearRtRic: ric,
191 TapiNodeAmf: ric + 16*self.FONTSIZE,
192 TapiNodeUpf: ric + 24*self.FONTSIZE,
193 TapiNodeOCuCp: ocu - 12.5*self.FONTSIZE,
194 TapiNodeOCuUp: ocu + 12.5*self.FONTSIZE,
196 TapiNodeFronthaulGateway: fhgw,
198 TapiNodeUserEquipment: 0
200 if node_type in x_mapping:
201 return x_mapping[node_type]
204 def __svg_dynamic_x_offset_by_node_type(self, node_type) -> int:
206 Mapping function from node types to y position in svg
210 pattern = self.configuration()['network']['pattern']
211 width_unit = (2 * 2 * self.FONTSIZE + padding)
212 x_mapping: Dict[type, int] = {
213 TapiNodeSmo: pattern['near-rt-ric'] * pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
214 TapiNodeOCloud: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
215 TapiNodeNearRtRic: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
216 TapiNodeAmf: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
217 TapiNodeUpf: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
218 TapiNodeOCuCp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
219 TapiNodeOCuUp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
220 TapiNodeODu: pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']*2,
221 TapiNodeFronthaulGateway: pattern['o-ru'] * pattern['user-equipment']*2,
222 TapiNodeORu: pattern['user-equipment']*2,
223 TapiNodeUserEquipment: 2
225 if node_type in x_mapping:
226 return x_mapping[node_type] * width_unit
229 def __svg_y_offset_by_node_type(self, node_type) -> int:
231 Mapping function from node types to y position in svg
234 offset = 11*self.FONTSIZE
235 y_mapping: Dict[type, int] = {
236 TapiNodeSmo: 0 * offset,
237 TapiNodeOCloud: 1 * offset,
238 TapiNodeNearRtRic: 2 * offset,
239 TapiNodeAmf: 2 * offset,
240 TapiNodeUpf: 2 * offset,
241 TapiNodeOCuCp: 3.5 * offset - 4 * self.FONTSIZE,
242 TapiNodeOCuUp: 3.5 * offset + 4 * self.FONTSIZE,
243 TapiNodeODu: 5 * offset,
244 TapiNodeFronthaulGateway: 6 * offset,
245 TapiNodeORu: 7 * offset,
246 TapiNodeUserEquipment: 8 * offset
248 if node_type in y_mapping:
249 return y_mapping[node_type]
253 def add_node(self, node: TapiNode):
255 Method adding a TAPI node to TAPI Topology.
256 :return TAPI Topology object.
258 self.__data["node"].append(node)
261 def add_link(self, link: TapiLink):
263 Method adding a TAPI node to TAPI Topology.
264 :return TAPI Topology object.
266 self.__data["link"].append(link)
269 def find_node(self, type_name:str, client_local_id:str) -> TapiNode:
271 Method finding the parent no in the hierarchy based on the node type name
272 and the client local identifier.
273 :param type_name: the TAPI NODE Type name of interest.
274 :param client_local_id: which starts with the parent node local id.
276 turn TAPI Node object.
278 for node in self.__data["node"]:
279 if type(node).__name__ == type_name:
280 if client_local_id.startswith(node.local_id()):
284 def __create_smos(self, parent: TapiNode, topology_structure: dict, count: int):
286 Method adding a TAPI node to TAPI Topology.
287 :param parent: A TAPI node which acts a a parent node in the topology.
288 :param topology_structure: Information about the next topology levels.
289 :param count: Number of instance to be created
290 :return TAPI Topology object.
293 next_type = "near-rt-ric"
294 for local_id in range(count):
297 if parent is not None:
298 prefix = parent.data()["name"][1]["value"]
299 config = {"node": {"localId": prefix + str(local_id),
300 "type": current_type,
301 "function": "o-ran-sc-topology-common:"+current_type}}
302 node = TapiNodeSmo(parent, config)
306 if "o-cloud" in topology_structure:
307 structure = topology_structure.copy()
308 self.__create_o_clouds(
309 node, structure, structure["o-cloud"])
311 if next_type in topology_structure:
312 structure = topology_structure.copy()
313 if current_type in structure:
314 del structure["o-cloud"]
315 if current_type in structure:
316 del structure[current_type]
317 self.__create_near_rt_rics(
318 node, structure, structure[next_type])
322 def __create_o_clouds(self, parent: TapiNode, topology_structure: dict, count: int):
324 Method adding a TAPI node to TAPI Topology.
325 :param parent: A TAPI node which acts a a parent node in the topology.
326 :param topology_structure: Information about the next topology levels.
327 :param count: Number of instance to be created
328 :return TAPI Topology object.
330 current_type = "o-cloud"
331 for local_id in range(count):
334 if parent is not None:
335 prefix = parent.json()["name"][1]["value"]
336 function = "o-ran-sc-topology-common:"+current_type
337 node_configuration = {"node": {"localId": prefix + str(local_id),
338 "type": current_type,
339 "function": function}}
340 node = TapiNodeOCloud(parent, node_configuration)
345 link_configuration = {
346 "topology_reference": self.data()["uuid"],
347 "name_prefix": "o2-rest",
351 self.add_link(TapiLink(link_configuration))
354 def __create_near_rt_rics(self, parent: TapiNode, topology_structure: dict, count: int):
356 Method adding a TAPI node to TAPI Topology.
357 :param parent: A TAPI node which acts a a parent node in the topology.
358 :param topology_structure: Information about the next topology levels.
359 :param count: Number of instance to be created
360 :return TAPI Topology object.
362 current_type = "near-rt-ric"
364 for local_id in range(count):
367 if parent is not None:
368 prefix = parent.json()["name"][1]["value"]
371 # create 1 5G-Core AMF and 1 5G-Core UPF instance per Near-RT-RIC
374 current_type = "access-and-mobility-management-function"
375 function = "o-ran-sc-topology-common:"+current_type
376 node_configuration = {"node": {"localId": prefix + str(local_id),
377 "type": current_type,
378 "function": function}}
379 node = TapiNodeAmf(parent, node_configuration)
383 current_type = "user-plane-function"
384 function = "o-ran-sc-topology-common:"+current_type
385 node_configuration = {"node": {"localId": prefix + str(local_id),
386 "type": current_type,
387 "function": function}}
388 node = TapiNodeUpf(parent, node_configuration)
392 current_type = "near-rt-ric"
393 function = "o-ran-sc-topology-common:"+current_type
394 node_configuration = {"node": {"localId": prefix + str(local_id),
395 "type": current_type,
396 "function": function}}
397 node = TapiNodeNearRtRic(parent, node_configuration)
402 link_configuration = {
403 "topology_reference": self.data()["uuid"],
404 "name_prefix": "a1-rest",
408 self.add_link(TapiLink(link_configuration))
411 link_configuration = {
412 "topology_reference": self.data()["uuid"],
413 "name_prefix": "o1-netconf",
417 self.add_link(TapiLink(link_configuration))
420 link_configuration = {
421 "topology_reference": self.data()["uuid"],
422 "name_prefix": "o1-file",
426 self.add_link(TapiLink(link_configuration))
429 link_configuration = {
430 "topology_reference": self.data()["uuid"],
431 "name_prefix": "o1-ves",
435 self.add_link(TapiLink(link_configuration))
438 if next_type in topology_structure:
439 structure = topology_structure.copy()
440 if current_type in structure:
441 del structure[current_type]
442 self.__create_o_cus(node, structure, structure[next_type])
446 def __function_identity(self, function_type: str, plane: str) -> str:
448 Method to calculate the Function IDENTITY
451 "o-ran-sc-topology-common:",
457 def __create_o_cus(self, parent: TapiNode, topology_structure: dict, count: int):
459 Method adding a TAPI node to TAPI Topology.
460 :param parent: A TAPI node which acts a a parent node in the topology.
461 :param topology_structure: Information about the next topology levels.
462 :param count: Number of instance to be created
463 :return TAPI Topology object.
465 current_type = "o-cu"
467 for local_id in range(count):
469 if parent is not None:
470 prefix = parent.data()["name"][1]["value"]
472 node: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {}
473 for plane in ["cp", "up"]:
474 config = {"node": {"localId": prefix + str(local_id),
475 "type": "-".join([current_type, plane]),
476 "function": self.__function_identity(current_type, plane)}}
477 classes: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {
480 node[plane] = classes[plane](parent, config)
481 self.add_node(node[plane])
485 link_configuration = {
486 "topology_reference": self.data()["uuid"],
487 "name_prefix": "e2-rest",
488 "provider": node[plane],
491 self.add_link(TapiLink(link_configuration))
494 link_configuration = {
495 "topology_reference": self.data()["uuid"],
496 "name_prefix": "o1-netconf",
497 "provider": node[plane],
498 "consumer": parent.parent()
500 self.add_link(TapiLink(link_configuration))
503 link_configuration = {
504 "topology_reference": self.data()["uuid"],
505 "name_prefix": "o1-file",
506 "provider": node[plane],
507 "consumer": parent.parent()
509 self.add_link(TapiLink(link_configuration))
512 link_configuration = {
513 "topology_reference": self.data()["uuid"],
514 "name_prefix": "o1-ves",
515 "provider": parent.parent(),
516 "consumer": node[plane]
518 self.add_link(TapiLink(link_configuration))
521 # E1 Interface between O-CU-UP and O-CU-CP
522 link_configuration = {
523 "topology_reference": self.data()["uuid"],
524 "name_prefix": "e1-unknown",
525 "provider": node["up"],
526 "consumer": node["cp"]
528 self.add_link(TapiLink(link_configuration))
530 # N2 Interface between O-CU-CP and AMF
531 link_configuration = {
532 "topology_reference": self.data()["uuid"],
533 "name_prefix": "n2-nas",
534 "provider": self.find_node("TapiNodeAmf", node["cp"].local_id()),
535 "consumer": node["cp"]
537 self.add_link(TapiLink(link_configuration))
539 # N3 Interface between O-CU-UP and UPF
540 link_configuration = {
541 "topology_reference": self.data()["uuid"],
542 "name_prefix": "n3-nas",
543 "provider": self.find_node("TapiNodeUpf", node["up"].local_id()),
544 "consumer": node["up"]
546 self.add_link(TapiLink(link_configuration))
548 if next_type in topology_structure:
549 structure = topology_structure.copy()
550 if current_type in structure:
551 del structure[current_type]
553 node, structure, structure[next_type])
556 def __create_o_dus(self, parents: Dict[str, TapiNode], topology_structure: dict, count: int):
558 Method adding a TAPI node to TAPI Topology.
559 :param parent: A TAPI node which acts a a parent node in the topology.
560 :param topology_structure: Information about the next topology levels.
561 :param count: Number of instance to be created
562 :return TAPI Topology object.
564 current_type = "o-du"
565 next_type = "fronthaul-gateway"
566 for local_id in range(count):
568 if parents["cp"] is not None:
569 prefix = parents["cp"].data()["name"][1]["value"]
570 config = {"node": {"localId": prefix + str(local_id),
571 "type": current_type,
572 "function": "o-ran-sc-topology-common:"+current_type}}
573 node = TapiNodeODu(parents["cp"], config)
576 for plane, parent in parents.items():
580 link_configuration = {
581 "topology_reference": self.data()["uuid"],
582 "name_prefix": "e2-rest",
584 "consumer": parent.parent()
586 self.add_link(TapiLink(link_configuration))
589 link_configuration = {
590 "topology_reference": self.data()["uuid"],
591 "name_prefix": "o1-netconf",
593 "consumer": parent.parent().parent()
595 self.add_link(TapiLink(link_configuration))
598 link_configuration = {
599 "topology_reference": self.data()["uuid"],
600 "name_prefix": "o1-file",
602 "consumer": parent.parent().parent()
604 self.add_link(TapiLink(link_configuration))
607 link_configuration = {
608 "topology_reference": self.data()["uuid"],
609 "name_prefix": "o1-ves",
610 "provider": parent.parent().parent(),
613 self.add_link(TapiLink(link_configuration))
615 # F1 User Plane or Control Plane
616 interfaces: Dict[str, str] = {"cp": "f1-c", "up": "f1-u"}
617 link_configuration = {
618 "topology_reference": self.data()["uuid"],
619 "name_prefix": interfaces[plane]+"-unknown",
623 self.add_link(TapiLink(link_configuration))
626 if next_type in topology_structure:
627 structure = topology_structure.copy()
628 if current_type in structure:
629 del structure[current_type]
630 self.__create_fronthaul_gateways(
631 node, structure, structure[next_type])
634 def __create_fronthaul_gateways(self, parent: TapiNode, topology_structure: dict, count: int):
636 Method adding a TAPI node to TAPI Topology.
637 :param parent: A TAPI node which acts a a parent node in the topology.
638 :param topology_structure: Information about the next topology levels.
639 :param count: Number of instance to be created
640 :return TAPI Topology object.
642 current_type = "fronthaul-gateway"
644 for local_id in range(count):
646 if parent is not None:
647 prefix = parent.data()["name"][1]["value"]
648 node_configuration = {
650 "localId": prefix + str(local_id),
651 "type": current_type,
652 "function": "o-ran-sc-topology-common:"+current_type,
653 "southbound-nep-count": topology_structure[next_type]
656 node = TapiNodeFronthaulGateway(parent, node_configuration)
662 link_configuration = {
663 "topology_reference": self.data()["uuid"],
664 "name_prefix": "oam-netconf",
666 "consumer": parent.parent().parent().parent()
668 self.add_link(TapiLink(link_configuration))
671 link_configuration = {
672 "topology_reference": self.data()["uuid"],
673 "name_prefix": "eth-ofh",
677 self.add_link(TapiLink(link_configuration))
680 if next_type in topology_structure:
681 structure = topology_structure.copy()
682 if current_type in structure:
683 del structure[current_type]
684 self.__create_o_rus(node, structure, structure[next_type])
687 def __create_o_rus(self, parent: TapiNode, topology_structure: dict, count: int):
689 Method adding a TAPI node to TAPI Topology.
690 :param parent: A TAPI node which acts a a parent node in the topology.
691 :param topology_structure: Information about the next topology levels.
692 :param count: Number of instance to be created
693 :return TAPI Topology object.
695 current_type = "o-ru"
696 next_type = "user-equipment"
697 for local_id in range(count):
699 if parent is not None:
700 prefix = parent.data()["name"][1]["value"]
701 config = {"node": {"localId": prefix + str(local_id),
702 "type": current_type,
703 "function": "o-ran-sc-topology-common:"+current_type}}
704 node = TapiNodeORu(parent, config)
710 link_configuration = {
711 "topology_reference": self.data()["uuid"],
712 "name_prefix": "ofh-netconf",
714 "consumer": parent.parent().parent().parent().parent()
716 self.add_link(TapiLink(link_configuration))
718 # OFH M-Plane to O-DU via fronthaul-gateway
719 link_configuration = {
720 "topology_reference": self.data()["uuid"],
721 "name_prefix": "ofh-netconf",
725 self.add_link(TapiLink(link_configuration))
728 if next_type in topology_structure:
729 structure = topology_structure.copy()
730 if current_type in structure:
731 del structure[current_type]
732 self.__create_ues(node, structure, structure[next_type])
735 def __create_ues(self, parent: TapiNode, topology_structure: dict, count: int):
737 Method adding a TAPI node to TAPI Topology.
738 :param parent: A TAPI node which acts a a parent node in the topology.
739 :param topology_structure: Information about the next topology levels.
740 :param count: Number of instance to be created
741 :return TAPI Topology object.
743 current_type = "user-equipment"
744 for local_id in range(count):
746 if parent is not None:
747 prefix = parent.data()["name"][1]["value"]
748 config = {"node": {"localId": prefix + str(local_id),
749 "type": current_type,
750 "function": "o-ran-sc-topology-common:"+current_type}}
751 node = TapiNodeUserEquipment(parent, config)
756 link_configuration = {
757 "topology_reference": self.data()["uuid"],
758 "name_prefix": "uu-radio",
762 self.add_link(TapiLink(link_configuration))
765 link_configuration = {
766 "topology_reference": self.data()["uuid"],
767 "name_prefix": "n1-nas",
768 "provider": self.find_node("TapiNodeAmf", node.local_id()),
771 self.add_link(TapiLink(link_configuration))
773 if "key" in topology_structure:
774 print("Implement missing topology level.")