feat: add WorkflowTopology class and workflow_to_workflow_topology operator

src/ansys/dpf/core/__init__.py

@@ -118,6 +118,7 @@
 CustomTypeFieldsCollection:type = _CollectionFactory(CustomTypeField)
 GenericDataContainersCollection:type = _CollectionFactory(GenericDataContainer)
 StringFieldsCollection:type = _CollectionFactory(StringField)
+OperatorsCollection: type = _CollectionFactory(Operator)
 AnyCollection:type = _Collection
 
 # for matplotlib

src/ansys/dpf/core/custom_container_base.py

@@ -0,0 +1,28 @@
+# Copyright (C) 2020 - 2024 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from ansys.dpf.core.generic_data_container import GenericDataContainer
+
+
+class CustomContainerBase:
+    def __init__(self, container: GenericDataContainer) -> None:
+        self._container = container

src/ansys/dpf/core/dpf_operator.py

@@ -385,6 +385,7 @@ def _type_to_output_method(self):
             collection_base,
             any,
         )
+        from ansys.dpf.core.workflow_topology import workflow_topology
 
         out = [
             (any.Any, self._api.operator_getoutput_as_any),
@@ -481,6 +482,11 @@ def _type_to_output_method(self):
                 self._api.operator_getoutput_as_any,
                 lambda obj, type: any.Any(server=self._server, any_dpf=obj).cast(type),
             ),
+            (
+                workflow_topology.WorkflowTopology,
+                None,
+                "WorkflowTopology",
+            ),
         ]
         if hasattr(self._api, "operator_getoutput_generic_data_container"):
             out.append(
@@ -726,8 +732,10 @@ def default_config(name, server=None):
 
     def __del__(self):
         try:
-            if self._internal_obj is not None:
-                self._deleter_func[0](self._deleter_func[1](self))
+            if hasattr(self, "_deleter_func"):
+                obj = self._deleter_func[1](self)
+                if obj is not None:
+                    self._deleter_func[0](obj)
         except:
             warnings.warn(traceback.format_exc())
 

src/ansys/dpf/core/helpers/utils.py

@@ -48,3 +48,19 @@
         warnings.warn(txt)
     # Return the accepted arguments
     return kwargs_in
+
+
+def indent(text, subsequent_indent="", initial_indent=None):
+    if initial_indent is None:
+        initial_indent = subsequent_indent
+
+    if not isinstance(text, str):
+        text = str(text)
+
+    lines = text.rstrip().splitlines()
+    indented_lines = [
+        f"{initial_indent if index == 0 else subsequent_indent}{line}"
+        for (index, line) in enumerate(lines)
+    ]
+
+    return "\n".join(indented_lines)

src/ansys/dpf/core/workflow.py

@@ -953,6 +953,15 @@
         """Saves the workflow to a GraphViz file."""
         return self._api.work_flow_export_graphviz(self, str(path))
 
+    def get_topology(self):
+        workflow_to_workflow_topology_op = dpf_operator.Operator(
+            "workflow_to_workflow_topology", server=self._server
+        )
+        workflow_to_workflow_topology_op.inputs.workflow.connect(self)
+        workflow_topology_container = workflow_to_workflow_topology_op.outputs.workflow_topology()
+
+        return workflow_topology_container
+
     def __del__(self):
         try:
             if hasattr(self, "_internal_obj"):

src/ansys/dpf/core/workflow_topology/data_connection.py

@@ -0,0 +1,89 @@
+# Copyright (C) 2020 - 2024 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from ansys.dpf.core.custom_container_base import CustomContainerBase
+
+
+class DataConnection(CustomContainerBase):
+    def __init__(self, container):
+        super().__init__(container)
+        self._source_data = None
+        self._target_operator = None
+        self._target_pin_id = None
+
+    @property
+    def source_data(self):
+        if self._source_data is None:
+            self._source_data = self._container.get_property("source_data")
+
+        return self._source_data
+
+    @property
+    def target_operator(self):
+        from ansys.dpf.core.dpf_operator import Operator
+
+        if self._target_operator is None:
+            self._target_operator = self._container.get_property("target_operator", Operator)
+
+        return self._target_operator
+
+    @property
+    def target_pin_id(self):
+        if self._target_pin_id is None:
+            self._target_pin_id = self._container.get_property("target_pin_id", int)
+
+        return self._target_pin_id
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = "     "
+        return (
+            "DataConnection with properties:\n"
+            " - source_data:\n"
+            f"{indent(self.source_data, indents)}\n"
+            " - target_operator:\n"
+            f"{indent(self.target_operator.name, indents)}\n"
+            " - target_pin_id:\n"
+            f"{indent(self.target_pin_id, indents)}"
+        )
+
+
+class DataConnectionsCollection:
+    def __init__(self, collection):
+        self._collection = collection
+
+    def __len__(self):
+        return len(self._collection)
+
+    def __getitem__(self, index):
+        return DataConnection(self._collection[index])
+
+    def __iter__(self):
+        for i in range(len(self)):
+            yield self[i]
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = ("   ", " - ")
+        return "\n".join([indent(data_connection, *indents) for data_connection in self])

src/ansys/dpf/core/workflow_topology/exposed_pin.py

@@ -0,0 +1,89 @@
+# Copyright (C) 2020 - 2024 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from ansys.dpf.core.custom_container_base import CustomContainerBase
+
+
+class ExposedPin(CustomContainerBase):
+    def __init__(self, container):
+        super().__init__(container)
+        self._name = None
+        self._operator = None
+        self._pin_id = None
+
+    @property
+    def name(self):
+        if self._name is None:
+            self._name = self._container.get_property("name", str)
+
+        return self._name
+
+    @property
+    def operator(self):
+        from ansys.dpf.core.dpf_operator import Operator
+
+        if self._operator is None:
+            self._operator = self._container.get_property("operator", Operator)
+
+        return self._operator
+
+    @property
+    def pin_id(self):
+        if self._pin_id is None:
+            self._pin_id = self._container.get_property("pin_id", int)
+
+        return self._pin_id
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = "     "
+        return (
+            "ExposedPin with properties:\n"
+            " - name:\n"
+            f"{indent(self.name, indents)}\n"
+            " - operator:\n"
+            f"{indent(self.operator.name, indents)}\n"
+            " - pin_id:\n"
+            f"{indent(self.pin_id, indents)}"
+        )
+
+
+class ExposedPinsCollection:
+    def __init__(self, collection):
+        self._collection = collection
+
+    def __len__(self):
+        return len(self._collection)
+
+    def __getitem__(self, index):
+        return ExposedPin(self._collection[index])
+
+    def __iter__(self):
+        for i in range(len(self)):
+            yield self[i]
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = ("   ", " - ")
+        return "\n".join([indent(exposed_pin, *indents) for exposed_pin in self])

src/ansys/dpf/core/workflow_topology/operator_connection.py

@@ -0,0 +1,101 @@
+# Copyright (C) 2020 - 2024 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from ansys.dpf.core.custom_container_base import CustomContainerBase
+
+
+class OperatorConnection(CustomContainerBase):
+    def __init__(self, container):
+        super().__init__(container)
+        self._source_operator = None
+        self._source_pin_id = None
+        self._target_operator = None
+        self._target_pin_id = None
+
+    @property
+    def source_operator(self):
+        from ansys.dpf.core.dpf_operator import Operator
+
+        if self._source_operator is None:
+            self._source_operator = self._container.get_property("source_operator", Operator)
+
+        return self._source_operator
+
+    @property
+    def source_pin_id(self):
+        if self._source_pin_id is None:
+            self._source_pin_id = self._container.get_property("source_pin_id", int)
+
+        return self._source_pin_id
+
+    @property
+    def target_operator(self):
+        from ansys.dpf.core.dpf_operator import Operator
+
+        if self._target_operator is None:
+            self._target_operator = self._container.get_property("target_operator", Operator)
+
+        return self._target_operator
+
+    @property
+    def target_pin_id(self):
+        if self._target_pin_id is None:
+            self._target_pin_id = self._container.get_property("target_pin_id", int)
+
+        return self._target_pin_id
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = "     "
+        return (
+            "OperatorConnection with properties:\n"
+            " - source_operator:\n"
+            f"{indent(self.source_operator.name, indents)}\n"
+            " - source_pin_id:\n"
+            f"{indent(self.source_pin_id, indents)}\n"
+            " - target_operator:\n"
+            f"{indent(self.target_operator.name, indents)}\n"
+            " - target_pin_id:\n"
+            f"{indent(self.target_pin_id, indents)}"
+        )
+
+
+class OperatorConnectionsCollection:
+    def __init__(self, collection):
+        self._collection = collection
+
+    def __len__(self):
+        return len(self._collection)
+
+    def __getitem__(self, index):
+        return OperatorConnection(self._collection[index])
+
+    def __iter__(self):
+        for i in range(len(self)):
+            yield self[i]
+
+    def __str__(self):
+        from ansys.dpf.core.helpers.utils import indent
+
+        indents = ("   ", " - ")
+        return "\n".join([indent(operator_connection, *indents) for operator_connection in self])