From: Martin Skorupski <martin.skorupski@highstreet-technologies.com>
Date: Wed, 25 Sep 2024 12:45:39 +0000 (+0200)
Subject: Define the schema
X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F56%2F13456%2F1;p=oam.git

Define the schema

- house keeping

Issue-ID: OAM-415
Change-Id: I1bff3eea69513c1baf7a7df0933f8d910385a7f3
Signed-off-by: Martin Skorupski <martin.skorupski@highstreet-technologies.com>
---

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 5dd83fb..4fd1cd2 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
@@ -52,10 +52,11 @@ class ORanSmo(ORanNode):
     ) -> None:
         o_ran_smo_data: IORanSmo = self._to_o_ran_smo_data(data)
         super().__init__(cast(dict[str, Any], o_ran_smo_data), **kwargs)
+        self.type = "o-ran-sc-network:smo"
         if self.parent is not None:
-            self._o_ran_near_rt_rics: list[
-                ORanNearRtRic
-            ] = self._calculate_near_rt_rics()
+            self._o_ran_near_rt_rics: list[ORanNearRtRic] = (
+                self._calculate_near_rt_rics()
+            )
 
     def _calculate_near_rt_rics(self) -> list[ORanNearRtRic]:
         hex_ring_radius: int = (
@@ -63,10 +64,10 @@ class ORanSmo(ORanNode):
             if self.parent is not None
             else 1
         )
-        hex_list: list[
-            Hex
-        ] = self.parent.spiral_radius_profile.oRanNearRtRicSpiral(
-            self.position, hex_ring_radius
+        hex_list: list[Hex] = (
+            self.parent.spiral_radius_profile.oRanNearRtRicSpiral(
+                self.position, hex_ring_radius
+            )
         )
         result: list[ORanNearRtRic] = []
         for index, hex in enumerate(hex_list):
@@ -86,6 +87,7 @@ class ORanSmo(ORanNode):
                         "position": hex,
                         "layout": self.layout,
                         "parent": self,
+                        "network": self.network,
                     },
                 )
             )
@@ -114,31 +116,18 @@ class ORanSmo(ORanNode):
     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}))
+        result.append(ORanTerminationPoint({
+            "name": phy_tp,
+            "type": "o-ran-sc-network:phy"
+        }))
         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
-
-    def to_topology_nodes(self) -> list[dict[str, Any]]:
-        result: list[dict[str, Any]] = 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, Any]]:
-        result: list[dict[str, Any]] = []  # super().to_topology_links()
-        for ric in self.o_ran_near_rt_rics:
-            result.extend(ric.to_topology_links())
+            result.append(ORanTerminationPoint({
+                     "name": id,
+                     "type": ":".join(["o-ran-sc-network", interface]),
+                     "supporter": phy_tp,
+                     "parent": self
+            }))
         return result
 
     def toKml(self) -> ET.Element:
@@ -157,3 +146,30 @@ class ORanSmo(ORanNode):
     def to_directory(self, parent_dir: str) -> None:
         for ric in self.o_ran_near_rt_rics:
             ric.to_directory(parent_dir)
+
+    def _extend_with_ric_references(
+        self: Any, super_method: Any, ric_method_name: str
+    ) -> list[dict[str, Any]]:
+        """
+        Helper method to extend results with references from
+        o_ran_near_rt_rics.
+
+        :param super_method: The superclass method to call for the initial
+                             result.
+        :param ric_method_name: The method name to call on each ric.
+        :return: A list of dictionaries with the combined results.
+        """
+        result = super_method()
+        for ric in self.o_ran_near_rt_rics:
+            result.extend(self.flatten_list(getattr(ric, ric_method_name)()))
+        return result
+
+    def to_topology_nodes(self) -> list[dict[str, Any]]:
+        return self._extend_with_ric_references(
+            super().to_topology_nodes, "to_topology_nodes"
+        )
+
+    def to_topology_links(self) -> list[dict[str, Any]]:
+        return self._extend_with_ric_references(
+            super().to_topology_links, "to_topology_links"
+        )