1
1
from __future__ import annotations
2
+ import typing as ty
2
3
import json
3
4
import tempfile
4
5
from pathlib import Path
@@ -16,12 +17,21 @@ def __init__(self, spec):
16
17
17
18
self .wf = load_class_or_func (self .spec ["function" ])(
18
19
** self ._parse_workflow_args (self .spec ["args" ])
19
- ) # loads the 'function' in smriprep.yaml, and implement the args (creates a dictionary)
20
+ )
21
+ # loads the 'function' in smriprep.yaml, and implement the args (creates a
22
+ # dictionary)
20
23
21
- def node_connections (self , workflow , functions : dict [str , dict ], wf_inputs : dict [str , str ], wf_outputs : dict [str , str ]):
22
- connections = defaultdict (dict )
24
+ def node_connections (
25
+ self ,
26
+ workflow ,
27
+ functions : dict [str , dict ],
28
+ wf_inputs : dict [str , str ],
29
+ wf_outputs : dict [str , str ],
30
+ ):
31
+ connections : defaultdict = defaultdict (dict )
23
32
24
- # iterates over wf graph, Get connections from workflow graph, store connections in a dictionary
33
+ # iterates over wf graph, Get connections from workflow graph, store connections
34
+ # in a dictionary
25
35
for edge , props in workflow ._graph .edges .items ():
26
36
src_node = edge [0 ].name
27
37
dest_node = edge [1 ].name
@@ -88,9 +98,11 @@ def _parse_workflow_args(cls, args):
88
98
dct [name ] = val
89
99
return dct
90
100
91
- def save_graph (self , out_path : Path , format : str = "svg" , work_dir : Path = None ):
101
+ def save_graph (
102
+ self , out_path : Path , format : str = "svg" , work_dir : ty .Optional [Path ] = None
103
+ ):
92
104
if work_dir is None :
93
- work_dir = tempfile .mkdtemp ()
105
+ work_dir = Path ( tempfile .mkdtemp () )
94
106
work_dir = Path (work_dir )
95
107
graph_dot_path = work_dir / "wf-graph.dot"
96
108
self .wf .write_hierarchical_dotfile (graph_dot_path )
0 commit comments