1
1
import re
2
2
import attrs
3
+ import inspect
3
4
from .base import BaseTaskConverter
5
+ from fileformats .core .mixin import WithClassifiers
4
6
5
7
6
8
@attrs .define
@@ -23,21 +25,32 @@ def generate_task_str(self, filename, input_fields, nonstd_types, output_fields)
23
25
executable = self .nipype_interface .cmd
24
26
if not isinstance (executable , str ):
25
27
raise RuntimeError (
26
- f"Could not find executable for { self .nipype_interface } "
28
+ f"Could not find executable for { self .nipype_interface } , "
29
+ "try the FunctionTaskConverter class instead"
27
30
)
28
31
32
+ def unwrap_field_type (t ):
33
+ if issubclass (t , WithClassifiers ) and t .is_classified :
34
+ unwraped_classifiers = ", " .join (unwrap_field_type (c ) for c in t .classifiers )
35
+ return f"{ t .unclassified .__name__ } [{ unwraped_classifiers } ]"
36
+ return t .__name__
37
+
29
38
def types_to_names (spec_fields ):
30
39
spec_fields_str = []
31
40
for el in spec_fields :
32
41
el = list (el )
33
- tp_str = str ( el [1 ])
34
- if tp_str . startswith ( "<class " ):
35
- tp_str = el [ 1 ]. __name__
42
+ field_type = el [1 ]
43
+ if inspect . isclass ( field_type ) and issubclass ( field_type , WithClassifiers ):
44
+ field_type_str = unwrap_field_type ( field_type )
36
45
else :
37
- # Alter modules in type string to match those that will be imported
38
- tp_str = tp_str .replace ("typing" , "ty" )
39
- tp_str = re .sub (r"(\w+\.)+(?<!ty\.)(\w+)" , r"\2" , tp_str )
40
- el [1 ] = "#" + tp_str + "#"
46
+ field_type_str = str (field_type )
47
+ if field_type_str .startswith ("<class " ):
48
+ field_type_str = el [1 ].__name__
49
+ else :
50
+ # Alter modules in type string to match those that will be imported
51
+ field_type_str = field_type_str .replace ("typing" , "ty" )
52
+ field_type_str = re .sub (r"(\w+\.)+(?<!ty\.)(\w+)" , r"\2" , field_type_str )
53
+ el [1 ] = "#" + field_type_str + "#"
41
54
spec_fields_str .append (tuple (el ))
42
55
return spec_fields_str
43
56
0 commit comments