forked from INCF/BrainImagingPipelines
/
io.py
174 lines (145 loc) · 6.33 KB
/
io.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
import glob
import os
import shutil
import re
import tempfile
from warnings import warn
from write_report import report
import sqlite3
try:
import pyxnat
except:
pass
from nipype.interfaces.base import (TraitedSpec, traits, File, Directory,
BaseInterface, InputMultiPath, isdefined,
OutputMultiPath, DynamicTraitedSpec,
Undefined, BaseInterfaceInputSpec)
from nipype.utils.filemanip import (copyfile, list_to_filename,
filename_to_list, save_json)
from nipype.interfaces.io import IOBase
import logging
iflogger = logging.getLogger('interface')
class ReportSinkInputSpec(DynamicTraitedSpec, BaseInterfaceInputSpec):
base_directory = Directory(
desc='Path to the base directory for writing report.')
container = traits.Str(desc = 'Folder within base directory in which to store output')
_outputs = traits.Dict(traits.Str, value={}, usedefault=True)
_outputs_order = []
remove_dest_dir = traits.Bool(False, usedefault=True,
desc='remove dest directory when copying dirs')
report_name = traits.Str('Report',usedefault=True, desc='Name of report')
json_sink = Directory(desc="place to store json in addition to base_directory")
def __setattr__(self, key, value):
if key not in self.copyable_trait_names():
if not isdefined(value):
super(ReportSinkInputSpec, self).__setattr__(key, value)
self._outputs[key] = value
self._outputs_order.append(key)
else:
if key in self._outputs:
self._outputs[key] = value
self._outputs_order.append(key)
super(ReportSinkInputSpec, self).__setattr__(key, value)
class ReportSink(IOBase):
""" ReportSink module to write outputs to a pdf and save to json file
This interface allows arbitrary creation of input attributes. The names of
these attributes define the Report structure to create for display of images,
tables, and filenames.
This interface **cannot** be used in a MapNode as the inputs are
defined only when the connect statement is executed.
If an input ends with a .png or .jpg, the image will be displayed in the report
If an input is a list enclosed in more than 2 brackets,
a table will be displayed:
ex: [[ [['Month','Day'],[7,10],[12,25]] ]] --> 3x2 table with
'Month' and 'Day' as column headers, 7,10 in the first row
and 12,25 in the second
Anything else is displayed as text
Examples
--------
>>> rs = ReportSink()
>>> rs.inputs.base_directory = 'results_dir'
>>> rs.inputs.subject = 'Subject 5'
>>> rs.inputs.table = [[ [['Month','Day'],[7,10],[12,25]] ]]
>>> rs.inputs.image = 'structural.png'
>>> rs.run() # doctest: +SKIP
"""
input_spec = ReportSinkInputSpec
def __init__(self, orderfields=None, infields=None, **kwargs):
"""
Parameters
----------
infields : list of str
Indicates the input fields to be dynamically created
"""
super(ReportSink, self).__init__(**kwargs)
undefined_traits = {}
# used for mandatory inputs check
self._infields = infields
if infields:
for key in infields:
self.inputs.add_trait(key, traits.Any)
self.inputs._outputs[key] = Undefined
undefined_traits[key] = Undefined
self._orderfields = orderfields
self.inputs.trait_set(trait_change_notify=False, **undefined_traits)
def _list_outputs(self):
"""Execute this module.
"""
outdir = self.inputs.base_directory
if not isdefined(outdir):
outdir = '.'
outdir = os.path.abspath(outdir)
if isdefined(self.inputs.container):
outdir = os.path.join(outdir, self.inputs.container)
if not os.path.exists(outdir):
try:
os.makedirs(outdir)
except OSError, inst:
if 'File exists' in inst:
pass
else:
raise(inst)
# Begin Report
rep = report(os.path.abspath(os.path.join(outdir,self.inputs.report_name+'.pdf')),self.inputs.report_name)
# Loop through all inputs
#for key, files in self.inputs._outputs.items():
if self._orderfields:
order = self._orderfields
else:
order = self.inputs._outputs_order
for key in order:
files = self.inputs._outputs[key]
if not isdefined(files):
continue
iflogger.debug("key: %s files: %s"%(key, str(files)))
files = filename_to_list(files)
tempoutdir = outdir
# Add name of input as a title
rep.add_text('<b>%s</b>' % key.replace('_',' '))
# flattening list
if isinstance(files, list):
if isinstance(files[0], list):
files = [item for sublist in files for item in sublist]
for i, thing in enumerate(files):
# Add a table, image or text
if isinstance(thing,list):
rep.add_table(thing)
else:
if thing.endswith('.png') or thing.endswith('.jpg'):
rep.add_text(os.path.split(thing)[1])
rep.add_image(thing)
else:
rep.add_text(thing)
# write the report
rep.write()
# save json
save_json(os.path.join(outdir, self.inputs.report_name+'.json'), self.inputs._outputs)
if isdefined(self.inputs.json_sink):
if not isdefined(self.inputs.container):
save_json(os.path.join(self.inputs.json_sink, self.inputs.report_name+'.json'), self.inputs._outputs)
else:
if not os.path.exists(os.path.join(self.inputs.json_sink,self.inputs.container)):
os.mkdir(os.path.join(self.inputs.json_sink,self.inputs.container))
save_json(os.path.join(self.inputs.json_sink,self.inputs.container, self.inputs.report_name+'.json'), self.inputs._outputs)
print "json file " , os.path.join(outdir, self.inputs.report_name+'.json')
return None