From 718e37645399f0d1be0c563c070b310d1ac66304 Mon Sep 17 00:00:00 2001 From: Martin Skorupski Date: Fri, 3 Nov 2023 12:55:06 +0100 Subject: [PATCH] Create ietf-network-topology json - the generation of itef-topology-nodes and links effects all object classes Issue-ID: OAM-382 Change-Id: I0a685461b4730be8d10eb23daf9ac7ae055e1d36 Signed-off-by: Martin Skorupski --- code/network-generator/model/python/nr_cell_du.py | 50 +++++++++---- .../model/python/o_ran_cloud_du.py | 24 ++++++ code/network-generator/model/python/o_ran_cu.py | 28 +++++++ code/network-generator/model/python/o_ran_du.py | 35 ++++++++- .../model/python/o_ran_near_rt_ric.py | 26 ++++++- .../model/python/o_ran_network.py | 20 ++--- code/network-generator/model/python/o_ran_node.py | 43 ++++++++--- .../network-generator/model/python/o_ran_object.py | 10 +-- code/network-generator/model/python/o_ran_ru.py | 87 +++++++++++++++++++--- code/network-generator/model/python/o_ran_smo.py | 25 +++++++ .../model/python/o_ran_spiral_radius_profile.py | 5 +- .../model/python/o_ran_termination_point.py | 49 ++++++++++-- code/network-generator/model/python/top.py | 9 +-- code/network-generator/model/python/tower.py | 32 ++++++++ code/network-generator/view/network_viewer.py | 5 +- 15 files changed, 369 insertions(+), 79 deletions(-) diff --git a/code/network-generator/model/python/nr_cell_du.py b/code/network-generator/model/python/nr_cell_du.py index b6cea7e..3f090ae 100644 --- a/code/network-generator/model/python/nr_cell_du.py +++ b/code/network-generator/model/python/nr_cell_du.py @@ -17,6 +17,9 @@ """ A Class representing a 3GPP new radio cell du (NrCellDu) """ +from typing import overload + +from model.python.o_ran_termination_point import ORanTerminationPoint from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode import model.python.hexagon as Hexagon @@ -29,7 +32,7 @@ import xml.etree.ElementTree as ET class INrCellDu(IORanObject): def __init__(self, cell_angel: int, azimuth: int, **kwargs): super().__init__(**kwargs) - self._cell_angle = cell_angle + self._cell_angle = cell_angel self._azimuth = azimuth @@ -44,6 +47,22 @@ class NrCellDu(ORanNode, INrCellDu): cell_data["azimuth"] if cell_data and "azimuth" in cell_data else 0 ) + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + result.append(ORanTerminationPoint({"id": self.name, "name": self.name})) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + # a cell is not a node it is a Termination Point + result: list[dict[str, dict]] = [] # super().to_topology_nodes() + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + # as a cell is not a node, it does not have links + result: list[dict[str, dict]] = [] # super().to_topology_links() + return result + def toKml(self) -> ET.Element: placemark: ET.Element = ET.Element("Placemark") name: ET.Element = ET.SubElement(placemark, "name") @@ -63,23 +82,28 @@ class NrCellDu(ORanNode, INrCellDu): geo_locations: list[GeoLocation] = list(map(method, points)) text: list[str] = [] - - index: int = 1 + int(self._azimuth/self._cell_angle) - network_center:GeoLocation = GeoLocation(self.parent.parent.parent.parent.parent.parent.geoLocation) + index: int = 1 + int(self._azimuth / self._cell_angle) + network_center: GeoLocation = GeoLocation( + self.parent.parent.parent.parent.parent.parent.geoLocation + ) intersect1: Point = Point( - (points[(2 * index +1) % 6].x + points[(2 * index +2) % 6].x) / 2, - (points[(2 * index +1) % 6].y + points[(2 * index +2) % 6].y) / 2, + (points[(2 * index + 1) % 6].x + points[(2 * index + 2) % 6].x) / 2, + (points[(2 * index + 1) % 6].y + points[(2 * index + 2) % 6].y) / 2, + ) + intersect_geo_location1: GeoLocation = network_center.point_to_geo_location( + intersect1 ) - intersect_geo_location1: GeoLocation = network_center.point_to_geo_location(intersect1) - + intersect2: Point = Point( - (points[(2 * index +3) % 6].x + points[(2 * index +4) % 6].x) / 2, - (points[(2 * index +3) % 6].y + points[(2 * index +4) % 6].y) / 2, + (points[(2 * index + 3) % 6].x + points[(2 * index + 4) % 6].x) / 2, + (points[(2 * index + 3) % 6].y + points[(2 * index + 4) % 6].y) / 2, + ) + intersect_geo_location2: GeoLocation = network_center.point_to_geo_location( + intersect2 ) - intersect_geo_location2: GeoLocation = network_center.point_to_geo_location(intersect2) - tower:GeoLocation = GeoLocation(self.geoLocation) + tower: GeoLocation = GeoLocation(self.geoLocation) cell_polygon: list[GeoLocation] = [] cell_polygon.append(tower) @@ -89,7 +113,7 @@ class NrCellDu(ORanNode, INrCellDu): cell_polygon.append(intersect_geo_location2) # close polygon cell_polygon.append(tower) - + for geo_location in cell_polygon: text.append( f"{'%.6f' % geo_location.longitude},{'%.6f' % geo_location.latitude},{'%.6f' % geo_location.aboveMeanSeaLevel}" diff --git a/code/network-generator/model/python/o_ran_cloud_du.py b/code/network-generator/model/python/o_ran_cloud_du.py index 617827a..2581920 100644 --- a/code/network-generator/model/python/o_ran_cloud_du.py +++ b/code/network-generator/model/python/o_ran_cloud_du.py @@ -20,12 +20,14 @@ By default all O-RAN-DUs associated with the towers around are deployed here. Maybe dedicated hardware is required to host O-DUs, but it is expected that the O-Cloud mechanism and concepts can be applied here. """ +from typing import overload import model.python.hexagon as Hexagon from model.python.hexagon import Hex from model.python.cube import Cube from model.python.tower import Tower from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode +from model.python.o_ran_termination_point import ORanTerminationPoint import xml.etree.ElementTree as ET @@ -72,6 +74,28 @@ class ORanCloudDu(ORanNode, IORanCloudDu): def towers(self) -> list[Tower]: return self._towers + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append(ORanTerminationPoint({"id": phy_tp, "name": phy_tp})) + for interface in ["o2"]: + id:str = "-".join([self.name, interface.upper()]) + result.append(ORanTerminationPoint({"id": id, "name":id, "supporter": phy_tp, "parent":self})) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + for tower in self.towers: + result.extend(tower.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + for tower in self.towers: + result.extend(tower.to_topology_links()) + return result + def toKml(self) -> ET.Element: o_ran_cloud_du: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(o_ran_cloud_du, "open") diff --git a/code/network-generator/model/python/o_ran_cu.py b/code/network-generator/model/python/o_ran_cu.py index fba55d7..e2f5520 100644 --- a/code/network-generator/model/python/o_ran_cu.py +++ b/code/network-generator/model/python/o_ran_cu.py @@ -18,6 +18,7 @@ A Class representing an O-RAN centralized unit (ORanCu) and at the same time a location for an O-Cloud resource pool """ +from typing import overload from model.python.cube import Cube from model.python.hexagon import Hex import model.python.hexagon as Hexagon @@ -25,6 +26,7 @@ from model.python.o_ran_cloud_du import ORanCloudDu from model.python.tower import Tower from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode +from model.python.o_ran_termination_point import ORanTerminationPoint import xml.etree.ElementTree as ET @@ -80,6 +82,32 @@ class ORanCu(ORanNode, IORanCu): result.append(tower) return result + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append({"tp-id": phy_tp, "name": phy_tp}) + for interface in ["e2", "o1"]: + id:str = "-".join([self.name, interface.upper()]) + result.append(ORanTerminationPoint({"id": id, "name":id, "supporter": phy_tp, "parent":self})) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + # for o_ran_du in self.o_ran_dus: # TODO + # result.extend(o_ran_du.to_topology_nodes()) + for o_ran_cloud_du in self.o_ran_cloud_dus: + result.extend(o_ran_cloud_du.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + # for o_ran_du in self.o_ran_dus: + # result.extend(o_ran_du.to_topology_links()) + for o_ran_cloud_du in self.o_ran_cloud_dus: + result.extend(o_ran_cloud_du.to_topology_links()) + return result + def toKml(self) -> ET.Element: o_ran_cu: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(o_ran_cu, "open") diff --git a/code/network-generator/model/python/o_ran_du.py b/code/network-generator/model/python/o_ran_du.py index 3774c8e..7f1e5c6 100644 --- a/code/network-generator/model/python/o_ran_du.py +++ b/code/network-generator/model/python/o_ran_du.py @@ -17,12 +17,10 @@ """ A Class representing an O-RAN distributed unit (ORanDu) """ -import model.python.hexagon as Hexagon -from model.python.hexagon import Hex -from model.python.cube import Cube -from model.python.o_ran_ru import ORanRu +from typing import overload from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode +from model.python.o_ran_termination_point import ORanTerminationPoint import xml.etree.ElementTree as ET @@ -41,6 +39,35 @@ class ORanDu(ORanNode, IORanDu): o_ran_du_data["oRanRuCount"] if o_ran_du_data and "oRanRuCount" in o_ran_du_data else 1 ) + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append(ORanTerminationPoint({"id": phy_tp, "name": phy_tp})) + for interface in ["e2", "o1", "ofhm", "ofhc", "ofhu","ofhs"]: + id:str = "-".join([self.name, interface.upper()]) + result.append(ORanTerminationPoint({"id": id, "name":id, "supporter": phy_tp, "parent":self})) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + for interface in ["e2", "o1"]: + link_id: str = "".join([interface, ":", self.name, "<->", self.parent.name]) + source_tp: str = "-".join([self.name, interface.upper()]) + dest_tp: str = "-".join([self.parent.name, interface.upper()]) + result.append( + { + "link-id": link_id, + "source": {"source-node": self.name, "source-tp": source_tp}, + "destination": {"dest-node": self.parent.name, "dest-tp": dest_tp}, + } + ) + return result + def toKml(self) -> ET.Element: o_ran_du: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(o_ran_du, "open") diff --git a/code/network-generator/model/python/o_ran_near_rt_ric.py b/code/network-generator/model/python/o_ran_near_rt_ric.py index a375e68..f533328 100644 --- a/code/network-generator/model/python/o_ran_near_rt_ric.py +++ b/code/network-generator/model/python/o_ran_near_rt_ric.py @@ -17,10 +17,12 @@ """ A Class representing an O-RAN Near real-time intelligent controller (ORanNearRtRic) """ +from typing import overload from model.python.tower import Tower from model.python.o_ran_cu import ORanCu from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode +from model.python.o_ran_termination_point import ORanTerminationPoint from model.python.hexagon import Hex import model.python.hexagon as Hexagon import xml.etree.ElementTree as ET @@ -36,7 +38,7 @@ class IORanNearRtRic(IORanObject): class ORanNearRtRic(ORanNode, IORanNearRtRic): def __init__(self, o_ran_near_rt_ric_data: IORanNearRtRic = None, **kwargs): super().__init__(o_ran_near_rt_ric_data, **kwargs) - self._o_ran_cus: list(ORanCu) = self._calculate_o_ran_cus() + self._o_ran_cus: list[ORanCu] = self._calculate_o_ran_cus() def _calculate_o_ran_cus(self) -> list[ORanCu]: hex_ring_radius: int = self.spiralRadiusProfile.oRanNearRtRicSpiralRadiusOfOCus @@ -79,6 +81,28 @@ class ORanNearRtRic(ORanNode, IORanNearRtRic): result.append(tower) return result + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append({"tp-id": phy_tp, "name": phy_tp}) + for interface in ["a1", "o1", "o2", "e2"]: + id:str = "-".join([self.name, interface.upper()]) + result.append(ORanTerminationPoint({"id": id, "name":id, "supporter": phy_tp, "parent":self})) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + for o_ran_cu in self.o_ran_cus: + result.extend(o_ran_cu.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + for o_ran_cu in self.o_ran_cus: + result.extend(o_ran_cu.to_topology_links()) + return result + def toKml(self) -> ET.Element: ric: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(ric, "open") diff --git a/code/network-generator/model/python/o_ran_network.py b/code/network-generator/model/python/o_ran_network.py index 6339335..4576530 100644 --- a/code/network-generator/model/python/o_ran_network.py +++ b/code/network-generator/model/python/o_ran_network.py @@ -16,8 +16,6 @@ """ Module for a class representing a O-RAN Network """ -from typing import Any - from model.python.o_ran_smo import ORanSmo from model.python.o_ran_spiral_radius_profile import SpiralRadiusProfile from model.python.o_ran_object import IORanObject, ORanObject @@ -33,7 +31,7 @@ class ORanNetwork(ORanObject): """ # constructor - def __init__(self, configuration: dict[str, Any], of: IORanObject = None, **kwargs): + def __init__(self, configuration: dict[str, dict], of: IORanObject = None, **kwargs): super().__init__(of, **kwargs) self.__configuration = configuration self.name = configuration["name"] @@ -76,25 +74,21 @@ class ORanNetwork(ORanObject): """ return self.__configuration - def __appendNodes(self) -> list[dict[str, Any]]: - result: list[dict[str, Any]] = [] - for tower in self._o_ran_smo.towers: - result.append(tower.toTopology()) - return result - - def toTopology(self) -> dict[str, Any]: + def to_topology(self) -> dict[str, dict]: + nodes: dict[str, dict] = self._o_ran_smo.to_topology_nodes() + links: dict[str, dict] = self._o_ran_smo.to_topology_links() return { "ietf-network:networks": { "network": [ { "network-id": self.id, - "node": self.__appendNodes(), - "ietf-network-topology:link": [], + "node": nodes, + "ietf-network-topology:link": links, } ], } } - + def toKml(self) -> ET.Element: root: ET.Element = ET.Element("kml", xmlns="http://www.opengis.net/kml/2.2") document = ET.SubElement(root, "Document") diff --git a/code/network-generator/model/python/o_ran_node.py b/code/network-generator/model/python/o_ran_node.py index f57279d..219b2c4 100644 --- a/code/network-generator/model/python/o_ran_node.py +++ b/code/network-generator/model/python/o_ran_node.py @@ -17,9 +17,10 @@ """ An abstract Class for O-RAN Node """ -from abc import abstractmethod +from abc import abstractmethod, abstractproperty from typing import Any import xml.etree.ElementTree as ET +import json from model.python.geo_location import GeoLocation from model.python.o_ran_object import IORanObject, ORanObject import model.python.hexagon as Hexagon @@ -76,7 +77,7 @@ class ORanNode(ORanObject, IORanNode): else SpiralRadiusProfile() ) self.parent = of["parent"] if of and "parent" in of else None - self._terminationPoints = [] + self._termination_points: list[ORanTerminationPoint] = [] @property def address(self) -> str: @@ -134,11 +135,11 @@ class ORanNode(ORanObject, IORanNode): def parent(self, value: Any): self._parent = value - @property - def terminationPoints(self) -> list[ORanTerminationPoint]: - return self._terminationPoints + @abstractproperty + def termination_points(self) -> list[ORanTerminationPoint]: + return self._termination_points - def json(self) -> dict[str, Any]: + def json(self) -> dict[str, dict]: result: dict = super().json() result["address"] = self.address result["geoLocation"] = self.geoLocation @@ -148,11 +149,33 @@ class ORanNode(ORanObject, IORanNode): result["parent"] = self.parent return result - def toTopology(self) -> dict[str, Any]: - result: dict[str, Any] = { + @abstractmethod + def to_topology_nodes(self) -> list[dict[str, dict]]: + tps: list[dict[str, dict]] = [] + for tp in self.termination_points: + if str(type(tp)) == "": + tps.append(tp.to_topology()) + + result: list[dict[str, dict]] = [] + result.append({ "node-id": self.name, - "ietf-network-topology:termination-point": self.terminationPoints, - } + "ietf-network-topology:termination-point": tps, + }) + return result + + @abstractmethod + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = [] + source_tp: str = "-".join([self.name, "phy".upper()]) + dest_tp: str = "-".join([self.parent.name, "phy".upper()]) + if self.parent and not "Tower" in source_tp and not "Tower" in dest_tp: + link_id: str = "".join(["phy", ":", self.name, "<->", self.parent.name]) + link = { + "link-id": link_id, + "source": {"source-node": self.name, "source-tp": source_tp}, + "destination": {"dest-node": self.parent.name, "dest-tp": dest_tp}, + } + result.append(link) return result @abstractmethod diff --git a/code/network-generator/model/python/o_ran_object.py b/code/network-generator/model/python/o_ran_object.py index 792cc30..3cc8a81 100644 --- a/code/network-generator/model/python/o_ran_object.py +++ b/code/network-generator/model/python/o_ran_object.py @@ -17,8 +17,6 @@ """ An abstract Class for O-RAN Objects """ -from abc import abstractmethod -from typing import Any from model.python.top import ITop, Top @@ -33,13 +31,9 @@ class ORanObject(Top, IORanObject): def __init__(self, of: IORanObject = None, **kwargs): super().__init__(of, **kwargs) - def json(self) -> dict[str, Any]: - result: dict[str, Any] = super().json() + def json(self) -> dict[str, dict]: + result: dict[str, dict] = super().json() return result def __str__(self) -> str: return str(self.json()) - - @abstractmethod - def toTopology(self) -> dict[str, Any]: - pass diff --git a/code/network-generator/model/python/o_ran_ru.py b/code/network-generator/model/python/o_ran_ru.py index a02db29..02ad0c2 100644 --- a/code/network-generator/model/python/o_ran_ru.py +++ b/code/network-generator/model/python/o_ran_ru.py @@ -17,6 +17,10 @@ """ A Class representing an O-RAN radio unit (ORanRu) """ +from typing import overload + +from model.python.o_ran_du import ORanDu +from model.python.o_ran_termination_point import ORanTerminationPoint from model.python.nr_cell_du import NrCellDu from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode @@ -25,7 +29,7 @@ import xml.etree.ElementTree as ET # Define the "IORanRu" interface class IORanRu(IORanObject): - def __init__(self, cell_count: int, ru_angle:int, ru_azimuth:int, **kwargs): + def __init__(self, cell_count: int, ru_angle: int, ru_azimuth: int, **kwargs): super().__init__(**kwargs) self._cell_count = cell_count self._ru_angle = ru_angle @@ -36,23 +40,46 @@ class IORanRu(IORanObject): class ORanRu(ORanNode, IORanRu): def __init__(self, o_ran_ru_data: IORanRu = None, **kwargs): super().__init__(o_ran_ru_data, **kwargs) - self._cell_count = o_ran_ru_data["cellCount"] if o_ran_ru_data and "cellCount" in o_ran_ru_data else 1 - self._ru_angle = o_ran_ru_data["ruAngle"] if o_ran_ru_data and "ruAngle" in o_ran_ru_data else 120 - self._ru_azimuth = o_ran_ru_data["ruAzimuth"] if o_ran_ru_data and "ruAzimuth" in o_ran_ru_data else 0 + self._cell_count = ( + o_ran_ru_data["cellCount"] + if o_ran_ru_data and "cellCount" in o_ran_ru_data + else 1 + ) + self._ru_angle = ( + o_ran_ru_data["ruAngle"] + if o_ran_ru_data and "ruAngle" in o_ran_ru_data + else 120 + ) + self._ru_azimuth = ( + o_ran_ru_data["ruAzimuth"] + if o_ran_ru_data and "ruAzimuth" in o_ran_ru_data + else 0 + ) self._cells: list[NrCellDu] = self._create_cells() - + name: str = self.name.replace("RU", "DU") + self._oRanDu: ORanDu = ORanDu( + { + "name": name, + "geoLocation": self.parent.geoLocation, + "position": self.parent.position, + "layout": self.layout, + "parent": self.parent.parent.parent, + } + ) def _create_cells(self) -> list[NrCellDu]: result: list[NrCellDu] = [] - cell_angle : int = self.parent.parent.parent.parent.parent.parent.configuration()[ - "pattern" - ]["nr-cell-du"]["cell-angle"] + cell_angle: int = ( + self.parent.parent.parent.parent.parent.parent.configuration()["pattern"][ + "nr-cell-du" + ]["cell-angle"] + ) for index in range(self._cell_count): s: str = "00" + str(index) name: str = "-".join( [self.name.replace("RU", "NRCellDu"), s[len(s) - 2 : len(s)]] ) - azimuth: int = index * cell_angle +self._ru_azimuth + azimuth: int = index * cell_angle + self._ru_azimuth result.append( NrCellDu( { @@ -73,8 +100,48 @@ class ORanRu(ORanNode, IORanRu): def cells(self) -> list[NrCellDu]: return self._cells + @property + def oRanDu(self) -> ORanDu: + return self._oRanDu + + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append(ORanTerminationPoint({"id": phy_tp, "name": phy_tp})) + for interface in ["ofhm", "ofhc", "ofhu", "ofhs"]: + id: str = "-".join([self.name, interface.upper()]) + result.append( + ORanTerminationPoint( + {"id": id, "name": id, "supporter": phy_tp, "parent": self} + ) + ) + for cell in self.cells: + result.extend(cell.termination_points) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + result.extend(self.oRanDu.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + result.extend(self.oRanDu.to_topology_links()) + for interface in ["phy", "ofhm", "ofhc", "ofhu", "ofhs"]: + link_id: str = "".join([interface, ":", self.name, "<->", self.oRanDu.name]) + source_tp: str = "-".join([self.name, interface.upper()]) + dest_tp: str = "-".join([self.oRanDu.name, interface.upper()]) + result.append( + { + "link-id": link_id, + "source": {"source-node": self.name, "source-tp": source_tp}, + "destination": {"dest-node": self.oRanDu.name, "dest-tp": dest_tp}, + } + ) + return result + def toKml(self) -> ET.Element: - print("ru-tower", self.position, self.parent.position) if self.position.q is not self.parent.position.q else "ok" o_ran_ru: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(o_ran_ru, "open") open.text = "1" diff --git a/code/network-generator/model/python/o_ran_smo.py b/code/network-generator/model/python/o_ran_smo.py index fe04088..ee0de31 100644 --- a/code/network-generator/model/python/o_ran_smo.py +++ b/code/network-generator/model/python/o_ran_smo.py @@ -17,11 +17,13 @@ """ A Class representing an O-RAN Service Management and Orchestration Framework (SMO) """ +from typing import overload from model.python.tower import Tower from model.python.o_ran_near_rt_ric import ORanNearRtRic from model.python.o_ran_object import IORanObject from model.python.o_ran_node import ORanNode from model.python.hexagon import Hex +from model.python.o_ran_termination_point import ORanTerminationPoint import model.python.hexagon as Hexagon import xml.etree.ElementTree as ET @@ -67,10 +69,21 @@ class ORanSmo(ORanNode, IORanSmo): ) return result + @property def o_ran_near_rt_rics(self) -> list[ORanNearRtRic]: return self._o_ran_near_rt_rics + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append(ORanTerminationPoint({"id": phy_tp, "name": phy_tp})) + for interface in ["a1", "o1", "o2"]: + id:str = "-".join([self.name, interface.upper()]) + result.append(ORanTerminationPoint({"id": id, "name":id, "supporter": phy_tp, "parent":self})) + return result + @property def towers(self) -> list[Tower]: result: list[Tower] = [] @@ -79,6 +92,18 @@ class ORanSmo(ORanNode, IORanSmo): result.append(tower) return result + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + for ric in self.o_ran_near_rt_rics: + result.extend(ric.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = [] # super().to_topology_links() + for ric in self.o_ran_near_rt_rics: + result.extend(ric.to_topology_links()) + return result + def toKml(self) -> ET.Element: smo: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(smo, "open") diff --git a/code/network-generator/model/python/o_ran_spiral_radius_profile.py b/code/network-generator/model/python/o_ran_spiral_radius_profile.py index c0fd44d..a47a486 100644 --- a/code/network-generator/model/python/o_ran_spiral_radius_profile.py +++ b/code/network-generator/model/python/o_ran_spiral_radius_profile.py @@ -14,16 +14,13 @@ #!/usr/bin/python -from dataclasses import dataclass -from typing import Any - from model.python.cube import Cube from model.python.hexagon import Hex import model.python.hexagon as Hexagon class SpiralRadiusProfile: - def __init__(self, data: [dict[str, Any] | None] = None): + def __init__(self, data: [dict[str, dict] | None] = None): self._oRanSmoSpiralRadiusOfNearRtRics = ( data.get("oRanSmoSpiralRadiusOfNearRtRics", 1) if data else 1 ) diff --git a/code/network-generator/model/python/o_ran_termination_point.py b/code/network-generator/model/python/o_ran_termination_point.py index ff9dc89..6d575e7 100644 --- a/code/network-generator/model/python/o_ran_termination_point.py +++ b/code/network-generator/model/python/o_ran_termination_point.py @@ -18,15 +18,50 @@ An abstract Class for O-RAN TerminationPoint """ from abc import abstractmethod -from typing import Any -from model.python.o_ran_object import ORanObject +from model.python.o_ran_object import IORanObject, ORanObject -# Define an abstract O-RAN Node class -class ORanTerminationPoint(ORanObject): - def __init__(self, **kwargs): +# Define the "IORanObject" interface +class IORanTerminationPointData(IORanObject): + def __init__(self, supporter: str = None, parent=None, **kwargs): super().__init__(**kwargs) + self.supporter = supporter + self.parent = parent - def toTopology(self): - result: dict[str, Any] = {"tp-id": self.name} + +# Define an O-RAN Termination Point (ietf-interface, onf:logical-termination-point) class +class ORanTerminationPoint(ORanObject, IORanTerminationPointData): + def __init__(self, tp: IORanTerminationPointData = None, **kwargs): + super().__init__(tp, **kwargs) + self.supporter = tp["supporter"] if tp and "supporter" in tp else None + self.parent = tp["parent"] if tp and "parent" in tp else None + + def to_topology(self): + result: dict[str, dict] = {"tp-id": self.name} + if self.supporter: + network_ref: str = "" + match str(type(self.parent)): + case "": + network_ref = self.parent.parent.id + case "": + network_ref = self.parent.parent.parent.id + case "": + network_ref = self.parent.parent.parent.parent.id + case "": + network_ref = self.parent.parent.parent.parent.parent.id + case "": + network_ref = self.parent.parent.parent.parent.parent.id + case "": + network_ref = self.parent.parent.parent.parent.parent.parent.id + case _: + print("unknown: implement " + str(type(self.parent))) + network_ref = "unknown: implement " + str(type(self.parent)) + + result["supporting-termination-point"] = [ + { + "network-ref": network_ref, + "node-ref": self.parent.name, + "tp-ref": self.supporter, + } + ] return result diff --git a/code/network-generator/model/python/top.py b/code/network-generator/model/python/top.py index ea6010a..82701e5 100644 --- a/code/network-generator/model/python/top.py +++ b/code/network-generator/model/python/top.py @@ -18,10 +18,8 @@ An abstract Class for all classes """ import uuid -from abc import ABC, abstractmethod -from typing import Any +from abc import ABC from model.python.type_definitions import ( - AddressType, AdministrativeState, OperationalState, UsageState, @@ -29,7 +27,6 @@ from model.python.type_definitions import ( LifeCycleState, AlarmState, ) -from model.python.geo_location import GeoLocation # Define the ITop interface @@ -57,7 +54,7 @@ class ITop: # Define the Top class class Top(ABC, ITop): - def __init__(self, data: [dict[str, Any] | None] = None): + def __init__(self, data: [dict[str, dict] | None] = None): self._id = data["id"] if data and "id" in data else str(uuid.uuid4()) self._name = ( data["name"] @@ -149,7 +146,7 @@ class Top(ABC, ITop): def utilization(self, value: Utilization): self._utilization = value - def json(self) -> dict[str, Any]: + def json(self) -> dict[str, dict]: return { "id": self.id, "name": self.name, diff --git a/code/network-generator/model/python/tower.py b/code/network-generator/model/python/tower.py index 7def5e6..d012e32 100644 --- a/code/network-generator/model/python/tower.py +++ b/code/network-generator/model/python/tower.py @@ -19,9 +19,11 @@ A Class representing a Tower to mount O-RAN RUs It can be interpreted as 'resource pool' for physical network functions. """ +from typing import overload from model.python.o_ran_object import IORanObject from model.python.o_ran_ru import ORanRu from model.python.o_ran_node import ORanNode +from model.python.o_ran_termination_point import ORanTerminationPoint import xml.etree.ElementTree as ET @@ -79,6 +81,36 @@ class Tower(ORanNode): def o_ran_rus(self) -> list[ORanRu]: return self._o_ran_rus + @property + def termination_points(self) -> list[ORanTerminationPoint]: + result: list[ORanTerminationPoint] = super().termination_points + phy_tp: str = "-".join([self.name, "phy".upper()]) + result.append({"tp-id": phy_tp}) + for interface in ["e2", "o1", "ofhm", "ofhc", "ofhu","ofhs"]: + result.append( { + "tp-id": "-".join([self.name, interface.upper()]), + "supporting-termination-point": [ + { + "network-ref": type(self.parent.parent.parent.parent), + "node-ref":self.name, + "tp-ref": phy_tp + } + ] + }) + return result + + def to_topology_nodes(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_nodes() + for o_ran_ru in self.o_ran_rus: + result.extend(o_ran_ru.to_topology_nodes()) + return result + + def to_topology_links(self) -> list[dict[str, dict]]: + result: list[dict[str, dict]] = super().to_topology_links() + for o_ran_ru in self.o_ran_rus: + result.extend(o_ran_ru.to_topology_links()) + return result + def toKml(self) -> ET.Element: tower: ET.Element = ET.Element("Folder") open: ET.Element = ET.SubElement(tower, "open") diff --git a/code/network-generator/view/network_viewer.py b/code/network-generator/view/network_viewer.py index fdd810e..8e0fa0e 100644 --- a/code/network-generator/view/network_viewer.py +++ b/code/network-generator/view/network_viewer.py @@ -18,7 +18,6 @@ Provides functions to convert the Network into different formats """ import json -from typing import Any from model.python.o_ran_network import ORanNetwork import xml.etree.ElementTree as ET @@ -43,7 +42,7 @@ class NetworkViewer: """ return self - def show_as_json(self) -> dict[str, Any]: + def show_as_json(self) -> dict[str, dict]: """ Method printing the class in json format. """ @@ -62,7 +61,7 @@ class NetworkViewer: :type filename: string """ with open(filename, "w", encoding="utf-8") as json_file: - output = self.__network.toTopology() + output:dict[str, dict] = self.__network.to_topology() json.dump(output, json_file, ensure_ascii=False, indent=2) print("File '" + filename + "' saved!") -- 2.16.6