feat(workflow): add Workflow.required_plugins

src/ansys/dpf/core/workflow.py

@@ -1032,3 +1032,31 @@ def __str__(self):
         from ansys.dpf.core.core import _description
 
         return _description(self._internal_obj, self._server)
+
+    def required_plugins(self) -> list[str]:
+        """List of plugins required by the workflow based on registered operators.
+
+        Returns
+        -------
+        plugins:
+            List of plugins used by the workflow ordered alphabetically.
+            The plugin name reported is the one set when loading the plugin.
+
+        Examples
+        --------
+        >>> from ansys.dpf import core as dpf
+        >>> wf = dpf.Workflow()
+        >>> op1 = dpf.Operator("csv_to_field")  # from 'csv' plugin
+        >>> op2 = dpf.Operator("U")         # from 'core' plugin
+        >>> wf.add_operators([op1, op2])
+        >>> wf.required_plugins()
+        ['core', 'csv']
+        """
+        num = self._api.work_flow_number_of_operators(self)
+        out = []
+        for i in range(num):
+            op_name = self._api.work_flow_operator_name_by_index(self, i)
+            spec = dpf.core.dpf_operator.Operator.operator_specification(op_name, self._server)
+            plugin_name = spec.properties["plugin"]
+            out.append(plugin_name)
+        return sorted(list(set(out)))

tests/test_workflow.py

@@ -1101,6 +1101,18 @@ def test_workflow_get_output_derived_class(server_type):
     assert workflow_topology
 
 
+def test_required_plugins(server_type):
+    wf = dpf.core.Workflow(server=server_type)
+    op1 = dpf.core.Operator("csv_to_field", server=server_type)  # from 'csv' plugin
+    op2 = dpf.core.Operator("U", server=server_type)  # from 'core' plugin
+    wf.add_operators([op1, op2])
+    plugins = wf.required_plugins()
+    assert isinstance(plugins, list)
+    assert op1.specification.properties["plugin"] in plugins
+    assert op2.specification.properties["plugin"] in plugins
+    assert len(plugins) >= 2
+
+
 def main():
     test_connect_field_workflow()
     velocity_acceleration = conftest.resolve_test_file("velocity_acceleration.rst", "rst_operators")