From: JvD_Ericsson Date: Thu, 14 Nov 2024 14:22:58 +0000 (+0000) Subject: Added a way to generate TEIV data X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=555fffabc96a61e6559086ffd4e5f75a5a1c5b4e;p=oam.git Added a way to generate TEIV data network-genorator now has a way to generate TEIV cloud event data based on the topology disabled by default in the config Issue-ID: SMO-166 Change-Id: Ia08f5d0e0ed79c8ff0c6962151fe76c2f71eb61c Signed-off-by: JvD_Ericsson --- diff --git a/code/network-generator/CONTRIBUTING.md b/code/network-generator/CONTRIBUTING.md index 7a22c5d..0dfc55e 100644 --- a/code/network-generator/CONTRIBUTING.md +++ b/code/network-generator/CONTRIBUTING.md @@ -23,7 +23,7 @@ sudo apt install python3.10-venv ## 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 diff --git a/code/network-generator/config.json b/code/network-generator/config.json index 0a80fb6..fbbf578 100644 --- a/code/network-generator/config.json +++ b/code/network-generator/config.json @@ -54,6 +54,10 @@ "kml": { "enabled": true, "compressed": true + }, + "teiv": { + "enabled": false, + "compressed": false } } } \ No newline at end of file diff --git a/code/network-generator/network_generation/cli.py b/code/network-generator/network_generation/cli.py index 2c7b6ad..de055b3 100644 --- a/code/network-generator/network_generation/cli.py +++ b/code/network-generator/network_generation/cli.py @@ -73,6 +73,7 @@ def main() -> None: # pragma: no cover "day0Config": "to_directory", "svg": "svg", "kml": "kml", + "teiv": "teiv", } for task_key, method_name in task_to_method.items(): diff --git a/code/network-generator/network_generation/model/jsonSchema/configuration.schema.json b/code/network-generator/network_generation/model/jsonSchema/configuration.schema.json index d1110f9..16f6228 100644 --- a/code/network-generator/network_generation/model/jsonSchema/configuration.schema.json +++ b/code/network-generator/network_generation/model/jsonSchema/configuration.schema.json @@ -317,6 +317,10 @@ "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" } } } diff --git a/code/network-generator/network_generation/model/python/nr_cell_du.py b/code/network-generator/network_generation/model/python/nr_cell_du.py index ca46b0b..b6bc262 100644 --- a/code/network-generator/network_generation/model/python/nr_cell_du.py +++ b/code/network-generator/network_generation/model/python/nr_cell_du.py @@ -229,3 +229,20 @@ class NrCellDu(ORanNode): 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 diff --git a/code/network-generator/network_generation/model/python/o_ran_cloud_du.py b/code/network-generator/network_generation/model/python/o_ran_cloud_du.py index e527cfc..23edb48 100644 --- a/code/network-generator/network_generation/model/python/o_ran_cloud_du.py +++ b/code/network-generator/network_generation/model/python/o_ran_cloud_du.py @@ -144,3 +144,36 @@ class ORanCloudDu(ORanNode): 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" + ) diff --git a/code/network-generator/network_generation/model/python/o_ran_cu.py b/code/network-generator/network_generation/model/python/o_ran_cu.py index ed26a19..d214e0c 100644 --- a/code/network-generator/network_generation/model/python/o_ran_cu.py +++ b/code/network-generator/network_generation/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 """ +import re import xml.etree.ElementTree as ET from typing import Any, cast @@ -145,7 +146,90 @@ class ORanCu(ORanNode): ) 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", + ) diff --git a/code/network-generator/network_generation/model/python/o_ran_du.py b/code/network-generator/network_generation/model/python/o_ran_du.py index 8dca087..d57040f 100644 --- a/code/network-generator/network_generation/model/python/o_ran_du.py +++ b/code/network-generator/network_generation/model/python/o_ran_du.py @@ -18,6 +18,7 @@ A Class representing an O-RAN distributed unit (ORanDu) """ import os +import re import xml.etree.ElementTree as ET from typing import Any, cast @@ -110,3 +111,42 @@ class ORanDu(ORanNode): 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 diff --git a/code/network-generator/network_generation/model/python/o_ran_near_rt_ric.py b/code/network-generator/network_generation/model/python/o_ran_near_rt_ric.py index a513f9c..e2f2981 100644 --- a/code/network-generator/network_generation/model/python/o_ran_near_rt_ric.py +++ b/code/network-generator/network_generation/model/python/o_ran_near_rt_ric.py @@ -18,6 +18,7 @@ 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 @@ -147,6 +148,72 @@ class ORanNearRtRic(ORanNode): ) 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", + ) diff --git a/code/network-generator/network_generation/model/python/o_ran_network.py b/code/network-generator/network_generation/model/python/o_ran_network.py index 2073d42..5011d1c 100644 --- a/code/network-generator/network_generation/model/python/o_ran_network.py +++ b/code/network-generator/network_generation/model/python/o_ran_network.py @@ -213,3 +213,12 @@ class ORanNetwork(ORanObject): 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]} diff --git a/code/network-generator/network_generation/model/python/o_ran_node.py b/code/network-generator/network_generation/model/python/o_ran_node.py index 9586de8..f7d83f0 100644 --- a/code/network-generator/network_generation/model/python/o_ran_node.py +++ b/code/network-generator/network_generation/model/python/o_ran_node.py @@ -106,7 +106,7 @@ class ORanNode(ORanObject): 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"]) @@ -278,7 +278,7 @@ class ORanNode(ORanObject): 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 @@ -296,7 +296,7 @@ class ORanNode(ORanObject): 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") @@ -320,7 +320,7 @@ class ORanNode(ORanObject): str("%.6f" % float(parent_gl.aboveMeanSeaLevel)), ]), ]) - return folder + return folder @abstractmethod def toSvg(self) -> ET.Element: @@ -329,3 +329,29 @@ class ORanNode(ORanObject): @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 diff --git a/code/network-generator/network_generation/model/python/o_ran_ru.py b/code/network-generator/network_generation/model/python/o_ran_ru.py index 71b0b3c..f85059d 100644 --- a/code/network-generator/network_generation/model/python/o_ran_ru.py +++ b/code/network-generator/network_generation/model/python/o_ran_ru.py @@ -18,6 +18,7 @@ A Class representing an O-RAN radio unit (ORanRu) """ import os +import re import xml.etree.ElementTree as ET from typing import Any, cast @@ -168,7 +169,7 @@ class ORanRu(ORanNode): 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") @@ -198,3 +199,51 @@ class ORanRu(ORanNode): 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 diff --git a/code/network-generator/network_generation/model/python/o_ran_smo.py b/code/network-generator/network_generation/model/python/o_ran_smo.py index 8da06e2..02dc27f 100644 --- a/code/network-generator/network_generation/model/python/o_ran_smo.py +++ b/code/network-generator/network_generation/model/python/o_ran_smo.py @@ -41,6 +41,7 @@ class ORanSmo(ORanNode): """ Class representing an O-RAN Service Management and Operation object. """ + _interfaces = ["a1", "o1", "ofhm", "o2"] def __init__( @@ -149,3 +150,41 @@ class ORanSmo(ORanNode): 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" + ) diff --git a/code/network-generator/network_generation/model/python/tower.py b/code/network-generator/network_generation/model/python/tower.py index 214e6e1..498b75b 100644 --- a/code/network-generator/network_generation/model/python/tower.py +++ b/code/network-generator/network_generation/model/python/tower.py @@ -151,3 +151,36 @@ class Tower(ORanNode): 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" + ) diff --git a/code/network-generator/network_generation/view/network_viewer.py b/code/network-generator/network_generation/view/network_viewer.py index 67fa024..fae65a3 100644 --- a/code/network-generator/network_generation/view/network_viewer.py +++ b/code/network-generator/network_generation/view/network_viewer.py @@ -78,6 +78,14 @@ class NetworkViewer: 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. @@ -89,6 +97,62 @@ class NetworkViewer: 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 @@ -141,7 +205,7 @@ class NetworkViewer: 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"])