Added the graph

Author: mahdieh-dst

nipype2pydra/workflow.py

@@ -1,4 +1,7 @@
 import json
+import tempfile
+from pathlib import Path
+import subprocess as sp
 from collections import defaultdict
 import black
 from nipype.interfaces.base import isdefined
@@ -82,3 +85,13 @@ def _parse_workflow_args(cls, args):
                 )
             dct[name] = val
         return dct
+
+    def save_graph(self, out_path: Path, format: str = "svg", work_dir: Path = None):
+        if work_dir is None:
+            work_dir = tempfile.mkdtemp()
+        work_dir = Path(work_dir)
+        graph_dot_path = work_dir / "wf-graph.dot"
+        self.wf.write_hierarchical_dotfile(graph_dot_path)
+        dot_path = sp.check_output("which dot", shell=True).decode('utf-8').strip()
+        sp.check_call(f"{dot_path} -T{format} {str(graph_dot_path)} > {str(out_path)}",
+                      shell=True)

tests/test_smriprep.py

@@ -1,9 +1,11 @@
 from pathlib import Path
+import yaml
 from nipype2pydra.cli import workflow
 from nipype2pydra.utils import show_cli_trace
+from nipype2pydra.workflow import WorkflowConverter
 
 
-def test_smriprep(pkg_dir, cli_runner):
+def test_smriprep_conversion(pkg_dir, cli_runner):
 
     output_dir = Path(pkg_dir) / "outputs"
 
@@ -19,3 +21,18 @@ def test_smriprep(pkg_dir, cli_runner):
     )
 
     assert result.exit_code == 0, show_cli_trace(result)
+
+
+def test_smriprep_graph(pkg_dir, cli_runner):
+
+    output_dir = Path(pkg_dir) / "outputs"
+
+    if not output_dir.exists():
+        output_dir.mkdir()
+
+    with open(f"{pkg_dir}/example-specs/smriprep.yaml") as f:
+        spec = yaml.safe_load(f)
+
+    converter = WorkflowConverter(spec)
+
+    converter.save_graph(output_dir / "smriprep-graph.svg")