Display Workflows and save as file (#1300)

* Save workflow as symbolic workflow

* Save workflow to JSON format

* Add a Workflow.view method

* Refactor to define name variable

* Add optional title argument

* Fix docstring

* Fix input management

* Use graphviz.view instead of PIL

* Try catch pyvista import

* Update Workflow.view and add test

Signed-off-by: paul.profizi <[email protected]>

* Update logic and test

Signed-off-by: paul.profizi <[email protected]>

* Mention required GraphViz install in docstring

Signed-off-by: paul.profizi <[email protected]>

* Mention required GraphViz install in docstring

Signed-off-by: paul.profizi <[email protected]>

* Mention required GraphViz install in docstring

Signed-off-by: paul.profizi <[email protected]>

* Fix QA

Signed-off-by: paul.profizi <[email protected]>

* Add HAS_GRAPHVIZ conditional skip

Signed-off-by: paul.profizi <[email protected]>

---------

Signed-off-by: paul.profizi <[email protected]>

Author: PProfizi

src/ansys/dpf/core/workflow.py

@@ -5,10 +5,13 @@
 ========
 """
 import logging
+import os
 import traceback
 import warnings
 
 from enum import Enum
+from typing import Union
+
 from ansys import dpf
 from ansys.dpf.core import dpf_operator, inputs, outputs
 from ansys.dpf.core.check_version import server_meet_version, version_requires, server_meet_version_and_raise
@@ -844,6 +847,68 @@ def create_on_other_server(self, *args, **kwargs):
                 "or both ip and port inputs) or a server is required"
             )
 
+    def view(
+            self,
+            title: Union[None, str] = None,
+            save_as: Union[None, str, os.PathLike] = None,
+            off_screen: bool = False,
+            keep_dot_file: bool = False,
+    ) -> Union[str, None]:
+        """Run a viewer to show a rendering of the workflow.
+
+        .. warning::
+            The workflow is rendered using GraphViz and requires:
+            - installation of GraphViz on your computer (see `<https://graphviz.org/download/>`_)
+            - installation of the ``graphviz`` library in your Python environment.
+
+
+        Parameters
+        ----------
+        title:
+            Name to use in intermediate files and in the viewer.
+        save_as:
+            Path to a file to save the workflow view as.
+        off_screen:
+            Render the image off_screen.
+        keep_dot_file:
+            Whether to keep the intermediate DOT file generated.
+
+        Returns
+        -------
+        Returns the path to the image file rendered is ``save_as``, else None.
+        """
+        try:
+            import graphviz
+        except ImportError:
+            raise ValueError("To render workflows using graphviz, run 'pip install graphviz'.")
+
+        if title is None:
+            name = f"workflow_{repr(self).split()[-1][:-1]}"
+        else:
+            name = title
+
+        if save_as:
+            dot_path = os.path.splitext(str(save_as))[0]+".dot"
+            image_path = save_as
+        else:
+            dot_path = os.path.join(os.getcwd(), f"{name}.dot")
+            image_path = os.path.join(os.getcwd(), f"{name}.png")
+
+        # Create graphviz file of workflow
+        self.to_graphviz(dot_path)
+        # Render workflow
+        graphviz.render(engine='dot', filepath=dot_path, outfile=image_path)
+        if not off_screen:
+            # View workflow
+            graphviz.view(filepath=image_path)
+        if not keep_dot_file:
+            os.remove(dot_path)
+        return image_path
+
+    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))
+
     def __del__(self):
         try:
             if hasattr(self, "_internal_obj"):

tests/test_workflow.py

@@ -1,17 +1,67 @@
+import os
+
 import numpy as np
 import pytest
 import platform
 
 import ansys.dpf.core.operators as op
 import conftest
 from ansys import dpf
+from ansys.dpf.core import misc
+
+if misc.module_exists("graphviz"):
+    HAS_GRAPHVIZ = True
+else:
+    HAS_GRAPHVIZ = False
 
 
 def test_create_workflow(server_type):
     wf = dpf.core.Workflow(server=server_type)
     assert wf._internal_obj
 
 
+@pytest.fixture()
+def remove_dot_file(request):
+    """Cleanup a testing directory once we are finished."""
+
+    dot_path = os.path.join(os.getcwd(), "test.dot")
+    png_path = os.path.join(os.getcwd(), "test.png")
+    png_path1 = os.path.join(os.getcwd(), "test1.png")
+
+    def remove_files():
+        if os.path.exists(dot_path):
+            os.remove(os.path.join(os.getcwd(), dot_path))
+        if os.path.exists(png_path):
+            os.remove(os.path.join(os.getcwd(), png_path))
+        if os.path.exists(png_path1):
+            os.remove(os.path.join(os.getcwd(), png_path1))
+
+    request.addfinalizer(remove_files)
+
+
+@pytest.mark.skipif(not HAS_GRAPHVIZ, reason="Please install pyvista")
+def test_workflow_view(server_in_process, remove_dot_file):
+    pre_wf = dpf.core.Workflow(server=server_in_process)
+    pre_op = dpf.core.operators.utility.forward(server=server_in_process)
+    pre_wf.add_operator(pre_op)
+    pre_wf.set_input_name("prewf_input", pre_op.inputs.any)
+    pre_wf.set_output_name("prewf_output", pre_op.outputs.any)
+
+    wf = dpf.core.Workflow(server=server_in_process)
+    forward_op = dpf.core.operators.utility.forward(server=server_in_process)
+    wf.add_operator(forward_op)
+    wf.set_input_name("wf_input", forward_op.inputs.any)
+    wf.set_output_name("wf_output", forward_op.outputs.any)
+
+    wf.connect_with(pre_wf, {"prewf_output": "wf_input"})
+    wf.view(off_screen=True, title="test1")
+    assert not os.path.exists("test1.dot")
+    assert os.path.exists("test1.png")
+    wf.view(off_screen=True, save_as="test.png", keep_dot_file=True)
+    assert os.path.exists("test.dot")
+    assert os.path.exists("test.png")
+
+
 def test_connect_field_workflow(server_type):
     wf = dpf.core.Workflow(server=server_type)
     wf.progress_bar = False