--- /dev/null
+# Copyright 2022 highstreet technologies GmbH
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#!/usr/bin/python
+"""
+Module containing the main class for this project for a TAPI Topology.
+"""
+import uuid
+from typing import Dict, List, Union
+from lxml import etree
+
+from model.python.top import Top
+from model.python.tapi_node import TapiNode
+from model.python.tapi_node_smo import TapiNodeSmo
+from model.python.tapi_node_o_cloud import TapiNodeOCloud
+from model.python.tapi_node_near_rt_ric import TapiNodeNearRtRic
+from model.python.tapi_node_o_cu_cp import TapiNodeOCuCp
+from model.python.tapi_node_o_cu_up import TapiNodeOCuUp
+from model.python.tapi_node_o_du import TapiNodeODu
+from model.python.tapi_node_fronthaul_gateway import TapiNodeFronthaulGateway
+from model.python.tapi_node_o_ru import TapiNodeORu
+from model.python.tapi_node_user_equipment import TapiNodeUserEquipment
+from model.python.tapi_link import TapiLink
+
+
+class TapiTopology(Top):
+ """
+ Class representing a TAPI Topology
+ """
+
+ __data: Dict[str, Union[str, List[Union[Dict, TapiNode, TapiLink]]]] = None
+ __configuration: dict = None
+
+ # constructor
+ def __init__(self, configuration: dict):
+ super().__init__(configuration)
+ self.__configuration = configuration
+ self.__data = {
+ "uuid": str(uuid.uuid4()),
+ "name": [{
+ "value-name": "network-name",
+ "value": configuration['network']['name']}],
+ "layer-protocol-name": ["ETH"],
+ "node": [],
+ "link": []}
+
+ topology_structure: dict = configuration['network']['pattern']
+ network_function_type: str = next(iter(topology_structure))
+ count: int = configuration['network']['pattern'][network_function_type]
+
+ if network_function_type == "smo":
+ self.__create_smos(None, topology_structure, count)
+ elif network_function_type == "near-rt-ric":
+ self.__create_near_rt_rics(None, topology_structure, count)
+ elif network_function_type == "o-cu":
+ self.__create_o_cus(None, topology_structure, count)
+ elif network_function_type == "o-du":
+ self.__create_o_dus(None, topology_structure, count)
+ elif network_function_type == "o-ru":
+ self.__create_o_rus(None, topology_structure, count)
+ elif network_function_type == "ue":
+ self.__create_ues(None, topology_structure, count)
+ else:
+ print("Unknown network function type", network_function_type)
+
+ # getter
+ def configuration(self) -> dict:
+ """
+ Getter for a json object representing the TAPI Topology configuration.
+ :return TAPI Topology configuration as json object.
+ """
+ return self.__configuration
+
+ def data(self) -> dict:
+ """
+ Getter for a json object representing the TAPI Topology.
+ :return TAPI Topology as json object.
+ """
+ return self.__data
+
+ def identifier(self) -> str:
+ """
+ Getter returning the TAPI Topology identifier.
+ :return Object identifier as UUID.
+ """
+ return self.__data["uuid"]
+
+ def name(self) -> str:
+ """
+ Getter for TAPI Topology name. The TAPI topology is a representation of
+ the network. Therefore, the TAPI Topology name has the same value as the
+ Network name.
+ :return TAPI Topology name as string.
+ """
+ return self.__configuration['network']['name']
+
+ def json(self) -> dict:
+ """
+ Getter for a json object representing the TAPI Topology.
+ :return TAPI Topology Context as json object.
+ """
+ result = self.data().copy()
+
+ # nodes handling
+ result["node"] = []
+ for node in self.__data["node"]:
+ result["node"].append(node.json())
+
+ # link handling
+ result["link"] = []
+ for link in self.__data["link"]:
+ result["link"].append(link.json())
+
+ return result
+
+ def svg(self, svg_x: int, svg_y: int) -> etree.Element:
+ """
+ Getter for a xml Element object representing the TAPI Topology Context.
+ :return TAPI Topology Context as svg object.
+ """
+ group = etree.Element("g")
+ title = etree.Element("title")
+ title.text = "\n TAPI Topology \n id: " + \
+ self.identifier() # + "\n name: " + self.name()
+ group.append(title)
+
+ # nodes handling
+ index_per_type: Dict = {}
+ svg_nodes = []
+ for node in self.__data["node"]:
+ if type(node) in index_per_type:
+ index_per_type[type(node)] = index_per_type[type(node)] + 1
+ else:
+ index_per_type[type(node)] = 0
+ index = index_per_type[type(node)]
+ node_x = svg_x + \
+ index*self.__svg_dynamic_x_offset_by_node_type(type(node)) + \
+ self.__svg_static_x_offset_by_node_type(type(node))
+ node_y = svg_y + self.__svg_y_offset_by_node_type(type(node))
+ svg_nodes.append(node.svg(node_x, node_y))
+ # group.append(node.svg(node_x, node_y))
+
+ # handling and drawing links
+ for link in self.__data["link"]:
+ group.append(link.svg(0, 0))
+
+ # drawing nodes
+ for svg_node in svg_nodes:
+ group.append(svg_node)
+
+ return group
+
+ def __svg_static_x_offset_by_node_type(self, node_type) -> int:
+ """
+ Mapping function from node types to y position in svg
+ return: int value
+ """
+ pattern = self.configuration()['network']['pattern']
+ padding = 0 # self.FONTSIZE
+ width_unit = (2 * 2 * self.FONTSIZE + padding)
+
+ ru = (pattern['user-equipment']-1) * width_unit / 2
+ fhgw = (pattern['o-ru'] *
+ pattern['user-equipment'] - 1) * width_unit / 2
+ odu = (pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
+ - 1) * width_unit/2
+ ocu = (pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
+ - 1) * width_unit / 2
+ ric = (pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
+ - 1) * width_unit / 2
+ smo = (pattern['smo'] * pattern['near-rt-ric'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment']
+ -0.5) * width_unit
+
+ x_mapping: Dict[type, int] = {
+ TapiNodeSmo: smo,
+ TapiNodeOCloud: ric,
+ TapiNodeNearRtRic: ric,
+ TapiNodeOCuCp: ocu - 12.5*self.FONTSIZE,
+ TapiNodeOCuUp: ocu + 12.5*self.FONTSIZE,
+ TapiNodeODu: odu,
+ TapiNodeFronthaulGateway: fhgw,
+ TapiNodeORu: ru,
+ TapiNodeUserEquipment: 0
+ }
+ if node_type in x_mapping:
+ return x_mapping[node_type]
+ return 0
+
+ def __svg_dynamic_x_offset_by_node_type(self, node_type) -> int:
+ """
+ Mapping function from node types to y position in svg
+ return: int value
+ """
+ padding = 0
+ pattern = self.configuration()['network']['pattern']
+ width_unit = (2 * 2 * self.FONTSIZE + padding)
+ x_mapping: Dict[type, int] = {
+ TapiNodeSmo: pattern['near-rt-ric'] * pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeOCloud: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeNearRtRic: pattern['o-cu'] * pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeOCuCp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeOCuUp: pattern['o-du'] * pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeODu: pattern['fronthaul-gateway'] * pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeFronthaulGateway: pattern['o-ru'] * pattern['user-equipment'],
+ TapiNodeORu: pattern['user-equipment'],
+ TapiNodeUserEquipment: 1
+ }
+ if node_type in x_mapping:
+ return x_mapping[node_type] * width_unit
+ return 0
+
+ def __svg_y_offset_by_node_type(self, node_type) -> int:
+ """
+ Mapping function from node types to y position in svg
+ return: int value
+ """
+ offset = 11*self.FONTSIZE
+ y_mapping: Dict[type, int] = {
+ TapiNodeSmo: 0 * offset,
+ TapiNodeOCloud: 1 * offset,
+ TapiNodeNearRtRic: 2 * offset,
+ TapiNodeOCuCp: 3.5 * offset - 4 * self.FONTSIZE,
+ TapiNodeOCuUp: 3.5 * offset + 4 * self.FONTSIZE,
+ TapiNodeODu: 5 * offset,
+ TapiNodeFronthaulGateway: 6 * offset,
+ TapiNodeORu: 7 * offset,
+ TapiNodeUserEquipment: 8 * offset
+ }
+ if node_type in y_mapping:
+ return y_mapping[node_type]
+ return 0
+
+ # methods
+ def add_node(self, node: TapiNode):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :return TAPI Topology object.
+ """
+ self.__data["node"].append(node)
+ return self
+
+ def add_link(self, link: TapiLink):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :return TAPI Topology object.
+ """
+ self.__data["link"].append(link)
+ return self
+
+ def __create_smos(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "smo"
+ next_type = "near-rt-ric"
+ for local_id in range(count):
+ prefix = ""
+
+ if parent is not None:
+ prefix = parent.data()["name"][1]["value"]
+ config = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": "o-ran-sc-topology-common:"+current_type}}
+ node = TapiNodeSmo(parent, config)
+ self.add_node(node)
+
+ # add O-Clouds
+ if "o-cloud" in topology_structure:
+ structure = topology_structure.copy()
+ self.__create_o_clouds(
+ node, structure, structure["o-cloud"])
+
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure["o-cloud"]
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_near_rt_rics(
+ node, structure, structure[next_type])
+
+ return self
+
+ def __create_o_clouds(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "o-cloud"
+ for local_id in range(count):
+ # add node
+ prefix = ""
+ if parent is not None:
+ prefix = parent.json()["name"][1]["value"]
+ function = "o-ran-sc-topology-common:"+current_type
+ node_configuration = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": function}}
+ node = TapiNodeOCloud(parent, node_configuration)
+ self.add_node(node)
+
+ # add links
+ # O2
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o2-rest",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+ return self
+
+ def __create_near_rt_rics(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "near-rt-ric"
+ next_type = "o-cu"
+ for local_id in range(count):
+ # add node
+ prefix = ""
+ if parent is not None:
+ prefix = parent.json()["name"][1]["value"]
+ function = "o-ran-sc-topology-common:"+current_type
+ node_configuration = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": function}}
+ node = TapiNodeNearRtRic(parent, node_configuration)
+ self.add_node(node)
+
+ # add links
+ # A1
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "a1-rest",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 NETCONF
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-netconf",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 FILE
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-file",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 VES
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-ves",
+ "provider": parent,
+ "consumer": node
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # continue
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_o_cus(node, structure, structure[next_type])
+
+ return self
+
+ def __function_identity(self, function_type: str, plane: str) -> str:
+ """
+ Method to calculate the Function IDENTITY
+ """
+ return "".join([
+ "o-ran-sc-topology-common:",
+ function_type,
+ "-",
+ plane
+ ])
+
+ def __create_o_cus(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "o-cu"
+ next_type = "o-du"
+ for local_id in range(count):
+ prefix = ""
+ if parent is not None:
+ prefix = parent.data()["name"][1]["value"]
+
+ node: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {}
+ for plane in ["cp", "up"]:
+ config = {"node": {"localId": prefix + str(local_id),
+ "type": "-".join([current_type, plane]),
+ "function": self.__function_identity(current_type, plane)}}
+ classes: Dict[str, Union[TapiNodeOCuCp, TapiNodeOCuUp]] = {
+ "cp": TapiNodeOCuCp,
+ "up": TapiNodeOCuUp}
+ node[plane] = classes[plane](parent, config)
+ self.add_node(node[plane])
+
+ # add links
+ # E2
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "e2-rest",
+ "provider": node[plane],
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 NETCONF
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-netconf",
+ "provider": node[plane],
+ "consumer": parent.parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 FILE
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-file",
+ "provider": node[plane],
+ "consumer": parent.parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 VES
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-ves",
+ "provider": parent.parent(),
+ "consumer": node[plane]
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # continue
+ # E1 Interface between O-CU-UP and O-CU-CP
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "e1-unknown",
+ "provider": node["up"],
+ "consumer": node["cp"]
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_o_dus(
+ node, structure, structure[next_type])
+ return self
+
+ def __create_o_dus(self, parents: Dict[str, TapiNode], topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "o-du"
+ next_type = "fronthaul-gateway"
+ for local_id in range(count):
+ prefix = "000"
+ if parents["cp"] is not None:
+ prefix = parents["cp"].data()["name"][1]["value"]
+ config = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": "o-ran-sc-topology-common:"+current_type}}
+ node = TapiNodeODu(parents["cp"], config)
+ self.add_node(node)
+
+ for plane, parent in parents.items():
+
+ # add links
+ # E2
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "e2-rest",
+ "provider": node,
+ "consumer": parent.parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 NETCONF
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-netconf",
+ "provider": node,
+ "consumer": parent.parent().parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 FILE
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-file",
+ "provider": node,
+ "consumer": parent.parent().parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # O1 VES
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "o1-ves",
+ "provider": parent.parent().parent(),
+ "consumer": node
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # F1 User Plane or Control Plane
+ interfaces: Dict[str, str] = {"cp": "f1-c", "up": "f1-u"}
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": interfaces[plane]+"-unknown",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # continue
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_fronthaul_gateways(
+ node, structure, structure[next_type])
+ return self
+
+ def __create_fronthaul_gateways(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "fronthaul-gateway"
+ next_type = "o-ru"
+ for local_id in range(count):
+ prefix = ""
+ if parent is not None:
+ prefix = parent.data()["name"][1]["value"]
+ node_configuration = {
+ "node": {
+ "localId": prefix + str(local_id),
+ "type": current_type,
+ "function": "o-ran-sc-topology-common:"+current_type,
+ "southbound-nep-count": topology_structure[next_type]
+ }
+ }
+ node = TapiNodeFronthaulGateway(parent, node_configuration)
+ self.add_node(node)
+
+ # add links
+
+ # Eth NBI
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "oam-netconf",
+ "provider": node,
+ "consumer": parent.parent().parent().parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # Eth SBI
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "eth-ofh",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # continue
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_o_rus(node, structure, structure[next_type])
+ return self
+
+ def __create_o_rus(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "o-ru"
+ next_type = "user-equipment"
+ for local_id in range(count):
+ prefix = ""
+ if parent is not None:
+ prefix = parent.data()["name"][1]["value"]
+ config = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": "o-ran-sc-topology-common:"+current_type}}
+ node = TapiNodeORu(parent, config)
+ self.add_node(node)
+
+ # add links
+
+ # O1 NETCONF
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "ofh-netconf",
+ "provider": node,
+ "consumer": parent.parent().parent().parent().parent()
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # OFH M-Plane to O-DU via fronthaul-gateway
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "ofh-netconf",
+ "provider": node,
+ "consumer": parent
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ # continue
+ if next_type in topology_structure:
+ structure = topology_structure.copy()
+ if current_type in structure:
+ del structure[current_type]
+ self.__create_ues(node, structure, structure[next_type])
+ return self
+
+ def __create_ues(self, parent: TapiNode, topology_structure: dict, count: int):
+ """
+ Method adding a TAPI node to TAPI Topology.
+ :param parent: A TAPI node which acts a a parent node in the topology.
+ :param topology_structure: Information about the next topology levels.
+ :param count: Number of instance to be created
+ :return TAPI Topology object.
+ """
+ current_type = "user-equipment"
+ for local_id in range(count):
+ prefix = ""
+ if parent is not None:
+ prefix = parent.data()["name"][1]["value"]
+ config = {"node": {"localId": prefix + str(local_id),
+ "type": current_type,
+ "function": "o-ran-sc-topology-common:"+current_type}}
+ node = TapiNodeUserEquipment(parent, config)
+ self.add_node(node)
+
+ # add links
+ # Uu unknown
+ link_configuration = {
+ "topology_reference": self.data()["uuid"],
+ "name_prefix": "uu-unknown",
+ "provider": parent,
+ "consumer": node
+ }
+ self.add_link(TapiLink(link_configuration))
+
+ if "key" in topology_structure:
+ print("Implement missing topology level.")
+
+ return self
Code Review / oam.git / commitdiff