feat: add WorkflowTopology class and workflow_to_workflow_topology operator (#1902)

Author: Matteo-Baussart-ANSYS

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/common.py

@@ -32,6 +32,7 @@
 import re
 import sys
 from enum import Enum
+from typing import Dict
 
 from ansys.dpf.core.misc import module_exists
 from ansys.dpf.gate.common import locations, ProgressBarBase  # noqa: F401
@@ -430,6 +431,51 @@ def type_to_special_dpf_constructors():
     return _type_to_special_dpf_constructors
 
 
+_derived_class_name_to_type = None
+
+
+def derived_class_name_to_type() -> Dict[str, type]:
+    """
+    Returns a mapping of derived class names to their corresponding Python classes.
+
+    Returns
+    -------
+    dict[str, type]
+        A dictionary mapping derived class names (str) to their corresponding
+        Python class objects.
+    """
+    global _derived_class_name_to_type
+    if _derived_class_name_to_type is None:
+        from ansys.dpf.core.workflow_topology import WorkflowTopology
+
+        _derived_class_name_to_type = {"WorkflowTopology": WorkflowTopology}
+    return _derived_class_name_to_type
+
+
+def record_derived_class(class_name: str, py_class: type, overwrite: bool = False):
+    """
+    Records a new derived class in the mapping of class names to their corresponding Python classes.
+
+    This function updates the global dictionary that maps derived class names (str) to their corresponding
+    Python class objects (type). If the provided class name already exists in the dictionary, it will either
+    overwrite the existing mapping or leave it unchanged based on the `overwrite` flag.
+
+    Parameters
+    ----------
+    class_name : str
+        The name of the derived class to be recorded.
+    py_class : type
+        The Python class type corresponding to the derived class.
+    overwrite : bool, optional
+        A flag indicating whether to overwrite an existing entry for the `class_name`.
+        If `True`, the entry will be overwritten. If `False` (default), the entry will
+        not be overwritten if it already exists.
+    """
+    recorded_classes = derived_class_name_to_type()
+    if overwrite or class_name not in recorded_classes:
+        recorded_classes[class_name] = py_class
+
+
 def create_dpf_instance(type, internal_obj, server):
     spe_constructors = type_to_special_dpf_constructors()
     if type in spe_constructors:

src/ansys/dpf/core/custom_container_base.py

@@ -0,0 +1,54 @@
+# 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.
+
+"""
+CustomContainerBase
+===================
+This module contains the `CustomContainerBase` class, which serves as a base
+for creating wrappers around `GenericDataContainer` objects.
+
+These wrappers provide an interface for accessing and managing data in
+generic containers, enabling more intuitive usage and the addition of custom
+behaviors tailored to specific use cases.
+"""
+
+from ansys.dpf.core.generic_data_container import GenericDataContainer
+
+
+class CustomContainerBase:
+    """
+    Base class for custom container wrappers.
+
+    This class provides a common interface for managing an underlying
+    `GenericDataContainer` object.
+    """
+
+    def __init__(self, container: GenericDataContainer) -> None:
+        """
+        Initialize the base container with a `GenericDataContainer`.
+
+        Parameters
+        ----------
+        container : GenericDataContainer
+            The underlying data container to be wrapped by this class.
+        """
+        self._container = container

src/ansys/dpf/core/dpf_operator.py

@@ -384,6 +384,7 @@ def _type_to_output_method(self):
             mesh_info,
             collection_base,
             any,
+            custom_container_base,
         )
 
         out = [
@@ -481,6 +482,15 @@ 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),
             ),
+            (
+                custom_container_base.CustomContainerBase,
+                self._api.operator_getoutput_generic_data_container,
+                lambda obj, type: type(
+                    container=generic_data_container.GenericDataContainer(
+                        generic_data_container=obj, server=self._server
+                    )
+                ),
+            ),
         ]
         if hasattr(self._api, "operator_getoutput_generic_data_container"):
             out.append(
@@ -726,8 +736,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

@@ -22,6 +22,7 @@
 
 import inspect
 import sys
+from typing import Any, Optional
 
 
 def _sort_supported_kwargs(bound_method, **kwargs):
@@ -48,3 +49,42 @@ def _sort_supported_kwargs(bound_method, **kwargs):
         warnings.warn(txt)
     # Return the accepted arguments
     return kwargs_in
+
+
+def indent(text: Any, subsequent_indent: str = "", initial_indent: Optional[str] = None) -> str:
+    """Indents each line of a given text.
+
+    Parameters
+    ----------
+    text : Any
+        The input text to be indented. If it is not already a string, it will be converted to one.
+    subsequent_indent : str, optional
+        The string to prefix all lines of the text after the first line. Default is an empty string.
+    initial_indent : Optional[str], optional
+        The string to prefix the first line of the text. If not provided, `subsequent_indent` will be used.
+
+    Returns
+    -------
+    str
+        The indented text with specified prefixes applied to each line.
+
+    Examples
+    --------
+    >>> text = "Hello\\nWorld"
+    >>> print(indent(text, subsequent_indent="    ", initial_indent="--> "))
+    --> Hello
+        World
+    """
+    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/outputs.py

@@ -81,17 +81,24 @@ def get_data(self):
         elif type_output == "int32":
             type_output = types.int
 
+        output = self._operator.get_output(self._pin, type_output)
+
         type_output_derive_class = self._spec.name_derived_class
+        if type_output_derive_class == "":
+            return output
+
+        from ansys.dpf.core.common import derived_class_name_to_type
+
+        derived_type = derived_class_name_to_type().get(type_output_derive_class)
+        if derived_type is not None:
+            return derived_type(output)
 
-        if type_output_derive_class != "":
-            out_type = [
-                type_tuple
-                for type_tuple in self._operator._type_to_output_method
-                if type_output_derive_class in type_tuple
-            ]
-            return out_type[0][0](self._operator.get_output(self._pin, type_output))
-        else:
-            return self._operator.get_output(self._pin, type_output)
+        derived_types = [
+            type_tuple
+            for type_tuple in self._operator._type_to_output_method
+            if type_output_derive_class in type_tuple
+        ]
+        return derived_types[0][0](output)
 
     def __call__(self):
         return self.get_data()

src/ansys/dpf/core/workflow.py

@@ -333,6 +333,7 @@ def _type_to_output_method(self):
             collection_base,
             streams_container,
         )
+        from ansys.dpf.core.custom_container_base import CustomContainerBase
 
         out = [
             (streams_container.StreamsContainer, self._api.work_flow_getoutput_streams),
@@ -421,6 +422,15 @@ def _type_to_output_method(self):
                 self._api.work_flow_getoutput_as_any,
                 lambda obj, type: any.Any(server=self._server, any_dpf=obj).cast(type),
             ),
+            (
+                CustomContainerBase,
+                self._api.work_flow_getoutput_generic_data_container,
+                lambda obj, type: type(
+                    container=generic_data_container.GenericDataContainer(
+                        generic_data_container=obj, server=self._server
+                    )
+                ),
+            ),
         ]
         if hasattr(self._api, "work_flow_connect_generic_data_container"):
             out.append(
@@ -953,6 +963,26 @@ def to_graphviz(self, path: Union[os.PathLike, str]):
         """Saves the workflow to a GraphViz file."""
         return self._api.work_flow_export_graphviz(self, str(path))
 
+    @version_requires("10.0")
+    def get_topology(self):
+        """Get the topology of the workflow.
+
+        Returns
+        -------
+        workflow_topology : workflow_topology.WorkflowTopology
+
+        Notes
+        -----
+        Available from 10.0 server version.
+        """
+        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 = workflow_to_workflow_topology_op.outputs.workflow_topology()
+
+        return workflow_topology
+
     def __del__(self):
         try:
             if hasattr(self, "_internal_obj"):

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

@@ -0,0 +1,26 @@
+# 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 .workflow_topology import WorkflowTopology
+from .operator_connection import OperatorConnection
+from .data_connection import DataConnection
+from .exposed_pin import ExposedPin