feat(topology): add support for nGraphResourceTransform

src/videoipath_automation_tool/apps/topology/model/n_graph_elements/topology_n_graph_resource_transform.py

@@ -0,0 +1,45 @@
+from typing import List, Literal
+
+from pydantic import Field
+
+from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_n_graph_element import (
+    Descriptor,
+    NGraphElement,
+)
+
+
+class NGraphResourceTransform(NGraphElement):
+    """
+    Represents the attributes of a nGraphResourceTransform.
+
+    """
+
+    active: bool = Field(default=True)
+    descriptor: Descriptor
+    fDescriptor: Descriptor
+    fromId: str
+    tags: List[str] = Field(default=[], description="List of tags.")
+    toId: str
+    fResourceIds: List[str] = Field(default=[], description="Factory resource ids")
+    resourceIds: List[str] = Field(default=[], description="User resource ids")
+    type: Literal["nGraphResourceTransform"] = "nGraphResourceTransform"
+
+    @property
+    def from_id(self) -> str:
+        """The source vertex ID."""
+        return self.fromId
+
+    @property
+    def to_id(self) -> str:
+        """The destination vertex ID."""
+        return self.toId
+
+    @property
+    def factory_resource_ids(self) -> List[str]:
+        """The factory resource IDs."""
+        return self.fResourceIds
+
+    @property
+    def resource_ids(self) -> List[str]:
+        """The user defined resource IDs."""
+        return self.resourceIds

src/videoipath_automation_tool/apps/topology/model/topology_device_configuration.py

@@ -17,6 +17,9 @@
     MapsElement,
     SdpStrategy,
 )
+from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_n_graph_resource_transform import (
+    NGraphResourceTransform,
+)
 from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_unidirectional_edge import (
     UnidirectionalEdge,
 )
@@ -34,6 +37,7 @@ class TopologyDeviceConfiguration(BaseModel):
         codec_vertices (List[CodecVertex]): Configuration of the codec vertices of the device
         internal_edges (List[UnidirectionalEdge]): Configuration of the internal unidirectional edges of the device (All edges that connect vertices within the same device)
         external_edges (List[UnidirectionalEdge]): Configuration of the external unidirectional edges of the device (All edges that connect this device to other devices)
+        resource_transform_edges (List[NGraphResourceTransform]): Configuration of the resource transform edges of the device
     """
 
     base_device: BaseDevice
@@ -42,6 +46,7 @@ class TopologyDeviceConfiguration(BaseModel):
     codec_vertices: List[CodecVertex] = Field(default_factory=list)
     internal_edges: List[UnidirectionalEdge] = Field(default_factory=list)
     external_edges: List[UnidirectionalEdge] = Field(default_factory=list)
+    resource_transform_edges: List[NGraphResourceTransform] = Field(default_factory=list)
 
     # --- Setters and Getters ---
 
@@ -245,7 +250,7 @@ def get_vertices_by_module_label(
 
     def get_nGraphElement_by_id(
         self, element_id: str
-    ) -> Optional[BaseDevice | GenericVertex | IpVertex | CodecVertex | UnidirectionalEdge]:
+    ) -> Optional[BaseDevice | GenericVertex | IpVertex | CodecVertex | UnidirectionalEdge | NGraphResourceTransform]:
         """Get an nGraphElement by its ID. Method will return the first element found with the specified ID.
 
         Args:
@@ -261,6 +266,7 @@ def get_nGraphElement_by_id(
             self.codec_vertices,
             self.internal_edges,
             self.external_edges,
+            self.resource_transform_edges,
         )
 
         matching_element = next((element for element in all_elements if element.id == element_id), None)

src/videoipath_automation_tool/apps/topology/model/topology_device_configuration_compare.py

@@ -7,6 +7,9 @@
 from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_generic_vertex import GenericVertex
 from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_ip_vertex import IpVertex
 from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_n_graph_element import NGraphElement
+from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_n_graph_resource_transform import (
+    NGraphResourceTransform,
+)
 from videoipath_automation_tool.apps.topology.model.n_graph_elements.topology_unidirectional_edge import (
     UnidirectionalEdge,
 )
@@ -249,6 +252,7 @@ class TopologyDeviceComparison(BaseModel):
     codec_vertices: NGraphElementListComparison
     internal_edges: NGraphElementListComparison
     external_edges: NGraphElementListComparison
+    resource_transform_edges: NGraphElementListComparison
 
     @classmethod
     def analyze_topology_devices(
@@ -273,6 +277,10 @@ def analyze_topology_devices(
         external_edges = cls.create_compare_list(
             reference_device.configuration.external_edges, staged_device.configuration.external_edges
         )
+        resource_transform_edges = cls.create_compare_list(
+            reference_device.configuration.resource_transform_edges,
+            staged_device.configuration.resource_transform_edges,
+        )
 
         return cls(
             reference_device=reference_device,
@@ -283,6 +291,7 @@ def analyze_topology_devices(
             codec_vertices=codec_vertices,
             internal_edges=internal_edges,
             external_edges=external_edges,
+            resource_transform_edges=resource_transform_edges,
         )
 
     @staticmethod
@@ -291,12 +300,14 @@ def create_compare_list(
         | List[GenericVertex]
         | List[IpVertex]
         | List[CodecVertex]
-        | List[UnidirectionalEdge],
+        | List[UnidirectionalEdge]
+        | List[NGraphResourceTransform],
         staged_elements: List[NGraphElement]
         | List[GenericVertex]
         | List[IpVertex]
         | List[CodecVertex]
-        | List[UnidirectionalEdge],
+        | List[UnidirectionalEdge]
+        | List[NGraphResourceTransform],
     ) -> NGraphElementListComparison:
         """Method to create a comparison list between two lists of nGraphElements."""
         reference_element_ids = [element.id for element in reference_elements]
@@ -353,6 +364,9 @@ def get_changed_elements(self, include_rev=False) -> List[NGraphElementDiff]:
         if len(self.external_edges.get_changed(include_rev)) > 0:
             changed_elements += self.external_edges.get_changed(include_rev)
 
+        if len(self.resource_transform_edges.get_changed(include_rev)) > 0:
+            changed_elements += self.resource_transform_edges.get_changed(include_rev)
+
         return changed_elements
 
     def get_added_elements(self) -> List[NGraphElement]:
@@ -363,6 +377,7 @@ def get_added_elements(self) -> List[NGraphElement]:
         added_elements += self.codec_vertices.get_added()
         added_elements += self.internal_edges.get_added()
         added_elements += self.external_edges.get_added()
+        added_elements += self.resource_transform_edges.get_added()
         return added_elements
 
     def get_removed_elements(self) -> List[NGraphElement]:
@@ -373,4 +388,5 @@ def get_removed_elements(self) -> List[NGraphElement]:
         removed_elements += self.codec_vertices.get_removed()
         removed_elements += self.internal_edges.get_removed()
         removed_elements += self.external_edges.get_removed()
+        removed_elements += self.resource_transform_edges.get_removed()
         return removed_elements