## Setting up your own virtual environment
Run `make virtual_env_4_oam` to create a virtual environment.
-then activate it with `source .venv/bin/activate`.
+then activate it with `source .oam/bin/activate`.
## Install the project in develop mode
"kml": {
"enabled": true,
"compressed": true
+ },
+ "teiv": {
+ "enabled": false,
+ "compressed": false
}
}
}
\ No newline at end of file
"day0Config": "to_directory",
"svg": "svg",
"kml": "kml",
+ "teiv": "teiv",
}
for task_key, method_name in task_to_method.items():
"kml": {
"description": "If enabled, a kml file will be generated.",
"$ref": "#/$defs/export"
+ },
+ "teiv": {
+ "description": "If enabled, a teiv data file will be generated.",
+ "$ref": "#/$defs/export"
}
}
}
def to_directory(self, parent_dir: str) -> None:
pass
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:NRCellDU",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ return super().add_teiv_data_entities(entity_type, attributes)
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = ""
+ ) -> dict[str, list[dict[str, Any]]]:
+ result: dict[str, list[dict[str, Any]]] = {}
+ return result
super().to_topology_links,
"to_topology_links",
)
+
+ def _extend_teiv_data_with_tower_references(
+ self: Any, tower_method_name: str
+ ) -> dict[str, list[dict[str, Any]]]:
+ """ """
+ result: dict[str, Any] = {}
+ for tower in self.towers:
+ tower_data = getattr(tower, tower_method_name)()
+ for key, value_list in tower_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return result
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ return self._extend_teiv_data_with_tower_references(
+ "add_teiv_data_entities"
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = ""
+ ) -> dict[str, list[dict[str, Any]]]:
+ return self._extend_teiv_data_with_tower_references(
+ "add_teiv_data_relationships"
+ )
A Class representing an O-RAN centralized unit (ORanCu)
and at the same time a location for an O-Cloud resource pool
"""
+import re
import xml.etree.ElementTree as ET
from typing import Any, cast
)
def to_topology_links(self) -> list[dict[str, Any]]:
- return self._extend_with_o_ran_cloud_du_references(
+ interface = "o1"
+ destination_node = self.parent.parent
+ link_id: str = "".join(
+ [interface, ":", self.name, "<->", destination_node.name]
+ )
+ source_tp: str = "-".join([self.name, interface.upper()])
+ dest_tp: str = "-".join([destination_node.name, interface.upper()])
+ result = self._extend_with_o_ran_cloud_du_references(
super().to_topology_links,
"to_topology_links",
)
+ result.append(
+ {
+ "link-id": link_id,
+ "source": {
+ "source-node": self.name,
+ "source-tp": source_tp,
+ },
+ "destination": {
+ "dest-node": destination_node.name,
+ "dest-tp": dest_tp,
+ },
+ }
+ )
+ return result
+
+ def _extend_teiv_data_with_o_ran_cloud_du_references(
+ self: Any,
+ teiv_data: dict[str, list[dict[str, Any]]],
+ o_ran_cloud_du_method_name: str,
+ ) -> dict[str, list[dict[str, Any]]]:
+ """ """
+ for o_ran_cloud_du in self.o_ran_cloud_dus:
+ o_ran_cloud_du_data = getattr(
+ o_ran_cloud_du, o_ran_cloud_du_method_name
+ )()
+ for key, value_list in o_ran_cloud_du_data.items():
+ if key not in teiv_data:
+ teiv_data[key] = []
+ teiv_data[key].extend(self.flatten_list(value_list))
+ return teiv_data
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:OCUCPFunction",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ id = int(re.sub(r"\D", "", self.name))
+ id_len = len(str(abs(id)))
+ attributes = {
+ "gNBCUName": self.name,
+ "gNBId": id,
+ "gNBIdLength": id_len,
+ }
+ result = super().add_teiv_data_entities(
+ entity_type, attributes
+ )
+ entity_type = "o-ran-smo-teiv-ran:OCUUPFunction"
+ attributes = {
+ "gNBId": id,
+ "gNBIdLength": id_len,
+ }
+ o_ran_cuip_data = super().add_teiv_data_entities(
+ entity_type, attributes
+ )
+ for key, value_list in o_ran_cuip_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return self._extend_teiv_data_with_o_ran_cloud_du_references(
+ result, "add_teiv_data_entities"
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = "o-ran-smo-teiv-ran:OCUCPFUNCTION_O1LINK_SMO"
+ ) -> dict[str, list[dict[str, Any]]]:
+ aside = self.name
+ bside = self.parent.parent.name
+ id = "".join(["o1", ":", aside, ":", bside])
+ return self._extend_teiv_data_with_o_ran_cloud_du_references(
+ super().add_teiv_data_relationships(id, aside, bside, rel_type),
+ "add_teiv_data_relationships",
+ )
A Class representing an O-RAN distributed unit (ORanDu)
"""
import os
+import re
import xml.etree.ElementTree as ET
from typing import Any, cast
os.makedirs(parent_path, exist_ok=True)
if not os.path.exists(path):
os.mkdir(path)
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:ODUFunction",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ id = int(re.sub(r"\D", "", self.name))
+ id_len = len(str(abs(id)))
+ attributes = {"gNBDUId": id, "gNBId": id, "gNBIdLength": id_len}
+ return super().add_teiv_data_entities(
+ entity_type, attributes
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = "o-ran-smo-teiv-ran:ODUFUNCTION_O1LINK_SMO"
+ ) -> dict[str, list[dict[str, Any]]]:
+ result = {}
+ aside = self.name
+ bside = self.parent.parent.parent.name
+ id = "".join(["o1", ":", aside, ":", bside])
+ result = super().add_teiv_data_relationships(
+ id, aside, bside, rel_type
+ )
+ aside = self.name
+ bside = self.parent.parent.name
+ id = "".join(["e2", ":", aside, ":", bside])
+ rel_type = "o-ran-smo-teiv-ran:ODUFUNCTION_E2LINK_NEARRTRICFUNCTION"
+ rel_data = super().add_teiv_data_relationships(
+ id, aside, bside, rel_type
+ )
+ for key, value_list in rel_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return result
A Class representing an O-RAN Near real-time intelligent controller
(ORanNearRtRic)
"""
+import re
import xml.etree.ElementTree as ET
from typing import Any, cast
)
def to_topology_links(self) -> list[dict[str, Any]]:
- return self._extend_with_o_ran_cu_references(
+ interface = "o1"
+ destination_node = self.parent
+ link_id: str = "".join(
+ [interface, ":", self.name, "<->", destination_node.name]
+ )
+ source_tp: str = "-".join([self.name, interface.upper()])
+ dest_tp: str = "-".join([destination_node.name, interface.upper()])
+ result = self._extend_with_o_ran_cu_references(
super().to_topology_links, "to_topology_links"
)
+ result.append(
+ {
+ "link-id": link_id,
+ "source": {
+ "source-node": self.name,
+ "source-tp": source_tp,
+ },
+ "destination": {
+ "dest-node": destination_node.name,
+ "dest-tp": dest_tp,
+ },
+ }
+ )
+ return result
+
+ def _extend_teiv_data_with_o_ran_cu_references(
+ self: Any,
+ teiv_data: dict[str, list[dict[str, Any]]],
+ o_ran_cu_method_name: str,
+ ) -> dict[str, list[dict[str, Any]]]:
+ """ """
+ for o_ran_cu in self.o_ran_cus:
+ o_ran_cu_data = getattr(o_ran_cu, o_ran_cu_method_name)()
+ for key, value_list in o_ran_cu_data.items():
+ if key not in teiv_data:
+ teiv_data[key] = []
+ teiv_data[key].extend(self.flatten_list(value_list))
+ return teiv_data
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:NearRTRICFunction",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ id = int(re.sub(r"\D", "", self.name))
+ attributes = {
+ "nearRtRicId": id,
+ }
+ result = super().add_teiv_data_entities(
+ entity_type, attributes
+ )
+ return self._extend_teiv_data_with_o_ran_cu_references(
+ result, "add_teiv_data_entities"
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = "o-ran-smo-teiv-ran:NEARRTRICFUNCTION_O1LINK_SMO"
+ ) -> dict[str, list[dict[str, Any]]]:
+ aside = self.name
+ bside = self.parent.name
+ id = "".join(["o1", ":", aside, ":", bside])
+ return self._extend_teiv_data_with_o_ran_cu_references(
+ super().add_teiv_data_relationships(id, aside, bside, rel_type),
+ "add_teiv_data_relationships",
+ )
def json(self) -> dict[str, Any]:
return super().json()
+
+ def to_teiv_data(self) -> dict[str, Any]:
+ entities: dict[str, list[dict[str, Any]]] = (
+ self._o_ran_smo.add_teiv_data_entities()
+ )
+ relationships: dict[str, list[dict[str, Any]]] = (
+ self._o_ran_smo.add_teiv_data_relationships()
+ )
+ return {"entities": [entities], "relationships": [relationships]}
self._address: AddressType = cast(
AddressType, o_ran_node_data["address"]
)
- self._geo_location: GeoLocation = o_ran_node_data["geoLocation"]
+ self._geo_location: GeoLocation = o_ran_node_data["geoLocation"]
self._url: str = str(o_ran_node_data["url"])
self._position: Hex = cast(Hex, o_ran_node_data["position"])
self._layout: Layout = cast(Layout, o_ran_node_data["layout"])
open.text = "1"
name: ET.Element = ET.SubElement(folder, "name")
name.text = self.name
-
+
placemark: ET.Element = ET.SubElement(folder, "Placemark")
name: ET.Element = ET.SubElement(placemark, "name")
name.text = self.name
str("%.6f" % float(point_gl.latitude)),
str("%.6f" % float(point_gl.aboveMeanSeaLevel)),
])
-
+
# link to parent
if (getattr(self.parent, 'geo_location', None) is not None):
line: ET.Element = ET.SubElement(multi_geometry, "LineString")
str("%.6f" % float(parent_gl.aboveMeanSeaLevel)),
]),
])
- return folder
+ return folder
@abstractmethod
def toSvg(self) -> ET.Element:
@abstractmethod
def to_directory(self, parent_dir: str) -> None:
pass
+
+ @abstractmethod
+ def add_teiv_data_entities(
+ self, entity_type: str, attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ sources = []
+ for tp in self.termination_points():
+ sources.append(tp.name)
+ result: dict[str, list[dict[str, Any]]] = {}
+ entity_id = self.name
+ entity: dict[str, Any] = {"id": entity_id}
+ if attributes:
+ entity["attributes"] = attributes
+ if sources:
+ entity["sourceIds"] = sources
+ result[entity_type] = [entity]
+ return result
+
+ @abstractmethod
+ def add_teiv_data_relationships(
+ self, id: str, aside: str, bside: str, rel_type: str
+ ) -> dict[str, list[dict[str, Any]]]:
+ result: dict[str, list[dict[str, Any]]] = {}
+ relationship = {"id": id, "aSide": aside, "bSide": bside}
+ result[rel_type] = [relationship]
+ return result
A Class representing an O-RAN radio unit (ORanRu)
"""
import os
+import re
import xml.etree.ElementTree as ET
from typing import Any, cast
for cell in self.cells:
o_ran_ru.append(cell.toKml())
o_ran_du.append(o_ran_ru)
- return o_ran_du
+ return o_ran_du
def toSvg(self) -> ET.Element:
return ET.Element("to-be-implemented")
for cell in self.cells:
result.extend(self.flatten_list(getattr(cell, cell_method_name)()))
return result
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:ORUFunction",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ id = int(re.sub(r"\D", "", self.name))
+ attributes = {"oruId": id}
+ result = super().add_teiv_data_entities(
+ entity_type, attributes
+ )
+ o_ran_du_data = self.oRanDu.add_teiv_data_entities()
+ for key, value_list in o_ran_du_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ for cell in self._cells:
+ cell_data = cell.add_teiv_data_entities()
+ for key, value_list in cell_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return result
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = "",
+ ) -> dict[str, list[dict[str, Any]]]:
+ result = {}
+ result = self.oRanDu.add_teiv_data_relationships()
+ for interface in ["OFHM", "OFHC", "OFHU", "OFHS"]:
+ aside = self.name
+ bside = self.oRanDu.name
+ id = "".join([interface, ":", aside, ":", bside])
+ rel_type = (
+ f"o-ran-smo-teiv-ran:ORUFUNCTION_{interface}LINK_ODUFUNCTION"
+ )
+ rel_data = super().add_teiv_data_relationships(
+ id, aside, bside, rel_type
+ )
+ for key, value_list in rel_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return result
"""
Class representing an O-RAN Service Management and Operation object.
"""
+
_interfaces = ["a1", "o1", "ofhm", "o2"]
def __init__(
return self._extend_with_ric_references(
super().to_topology_links, "to_topology_links"
)
+
+ def _extend_teiv_data_with_ric_references(
+ self: Any,
+ teiv_data: dict[str, list[dict[str, Any]]],
+ ric_method_name: str
+ ) -> dict[str, list[dict[str, Any]]]:
+ """ """
+ for ric in self.o_ran_near_rt_rics:
+ ric_data = getattr(ric, ric_method_name)()
+ for key, value_list in ric_data.items():
+ if key not in teiv_data:
+ teiv_data[key] = []
+ teiv_data[key].extend(self.flatten_list(value_list))
+ return teiv_data
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "o-ran-smo-teiv-ran:SMO",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ attributes = {
+ "smoName": self.name
+ }
+ return self._extend_teiv_data_with_ric_references(
+ super().add_teiv_data_entities(entity_type, attributes),
+ "add_teiv_data_entities"
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = ""
+ ) -> dict[str, list[dict[str, Any]]]:
+ return self._extend_teiv_data_with_ric_references(
+ {}, "add_teiv_data_relationships"
+ )
super().to_topology_links,
"to_topology_links",
)
+
+ def _extend_teiv_data_with_o_ran_ru_references(
+ self: Any, o_ran_ru_method_name: str
+ ) -> dict[str, list[dict[str, Any]]]:
+ """ """
+ result: dict[str, Any] = {}
+ for o_ran_ru in self.o_ran_rus:
+ o_ran_ru_data = getattr(o_ran_ru, o_ran_ru_method_name)()
+ for key, value_list in o_ran_ru_data.items():
+ if key not in result:
+ result[key] = []
+ result[key].extend(self.flatten_list(value_list))
+ return result
+
+ def add_teiv_data_entities(
+ self,
+ entity_type: str = "",
+ attributes: dict[str, Any] = {}
+ ) -> dict[str, list[dict[str, Any]]]:
+ return self._extend_teiv_data_with_o_ran_ru_references(
+ "add_teiv_data_entities"
+ )
+
+ def add_teiv_data_relationships(
+ self,
+ id: str = "",
+ aside: str = "",
+ bside: str = "",
+ rel_type: str = "",
+ ) -> dict[str, list[dict[str, Any]]]:
+ return self._extend_teiv_data_with_o_ran_ru_references(
+ "add_teiv_data_relationships"
+ )
json.dump(content, jf, ensure_ascii=False, indent=2)
print(f'File "{filename}" saved!')
+ def __save_on_disc_teiv_cloudevent(
+ self, filename: str, content: Any, cloud_event_header: str
+ ) -> None:
+ with open(f"{filename}", "w", encoding="utf-8") as jf:
+ jf.write(cloud_event_header)
+ json.dump(content, jf, ensure_ascii=False, indent=2)
+ print(f'File "{filename}" saved!')
+
def rfc8345(self, filename: str, compressed: bool = True) -> None:
"""
Method saving the class content to a file in json format.
file_extension: str = "-operational.json"
self.__save_on_disc(f"{filename}{file_extension}", compressed, output)
+ def teiv(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method for saving the class content as teiv cloud event
+ data in json format and in small chunks of cloud event text files.
+ :param filename: A valid path to a file on the system.
+ :param compressed: if True, json is stored as gz format.
+ :type filename: string
+ """
+ output: dict[str, list[dict[str, list[dict[str, Any]]]]] = (
+ self.__network.to_teiv_data()
+ )
+ chunk_size: int = 600
+ entities = output.get("entities", [])
+ relationships = output.get("relationships", [])
+ self.save_teiv_as_chunks("entities", entities, chunk_size, filename)
+ self.save_teiv_as_chunks(
+ "relationships", relationships, chunk_size, filename
+ )
+ self.__save_on_disc(f"{filename}-teiv-data.json", compressed, output)
+
+ def save_teiv_as_chunks(
+ self,
+ type: str,
+ data: list[dict[str, Any]],
+ chunk_size: int,
+ filename: str,
+ ) -> None:
+ """
+ Method for saving the class content as teiv cloud event
+ data in small chunks of cloud event text files.
+ :param type: type of node.
+ :param data: the data to be split.
+ :param chunk_size: amount of nodes per cloud event
+ :param filename: a valid path to a file on the system.
+ """
+ cloud_event_header = "ce_specversion:::1.0,ce_id:::a30e63c9-d29e" \
+ "-46ff-b99a-b63ed83fd237,ce_source:::dmi-plugin:nm-1,ce_type:::" \
+ "ran-logical-topology.merge,content-type:::application/yang-" \
+ "data+json,ce_time:::2023-11-30T09:05:00Z,ce_dataschema:::https" \
+ "://ties:8080/schemas/v1/r1-topology,,,"
+ idx = 1
+ for items in data:
+ for key, value in items.items():
+ for i in range(0, len(value), chunk_size):
+ full_filename = f"{filename}-teiv-{type}-data-part" \
+ f"-{idx}-{i // chunk_size + 1}.txt"
+ chunk: dict[str, Any] = {
+ type: [{key: value[i: i + chunk_size]}]
+ }
+ self.__save_on_disc_teiv_cloudevent(
+ full_filename,
+ chunk,
+ cloud_event_header,
+ )
+ idx = idx + 1
+
def readStylesFromFile(self) -> str:
"""
Method reading the css styles from known file
label_style = ET.SubElement(style, "LabelStyle")
label_scale = ET.SubElement(label_style, "scale")
label_scale.text = "0"
-
+
icon_style = ET.SubElement(style, "IconStyle")
icon_color = ET.SubElement(icon_style, "color")
icon_color.text = str(value["stroke"]["color"])
Code Review / oam.git / commitdiff