-
Notifications
You must be signed in to change notification settings - Fork 15
/
bids.py
99 lines (77 loc) · 3.37 KB
/
bids.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
"""Adapted interfaces from Niworkflows."""
from json import loads
from pathlib import Path
from bids.layout import Config
from nipype.interfaces.base import (
BaseInterfaceInputSpec,
OutputMultiObject,
SimpleInterface,
Str,
TraitedSpec,
traits,
)
from niworkflows.interfaces.bids import DerivativesDataSink as BaseDerivativesDataSink
from pkg_resources import resource_filename as pkgrf
from aslprep import config
# NOTE: Modified for aslprep's purposes
aslprep_spec = loads(Path(pkgrf("aslprep", "data/aslprep_bids_config.json")).read_text())
bids_config = Config.load("bids")
deriv_config = Config.load("derivatives")
aslprep_entities = {v["name"]: v["pattern"] for v in aslprep_spec["entities"]}
merged_entities = {**bids_config.entities, **deriv_config.entities}
merged_entities = {k: v.pattern for k, v in merged_entities.items()}
merged_entities = {**merged_entities, **aslprep_entities}
merged_entities = [{"name": k, "pattern": v} for k, v in merged_entities.items()]
config_entities = frozenset({e["name"] for e in merged_entities})
class _BIDSDataGrabberInputSpec(BaseInterfaceInputSpec):
subject_data = traits.Dict(Str, traits.Any)
subject_id = Str()
class _BIDSDataGrabberOutputSpec(TraitedSpec):
out_dict = traits.Dict(desc="output data structure")
fmap = OutputMultiObject(desc="output fieldmaps")
bold = OutputMultiObject(desc="output functional images")
sbref = OutputMultiObject(desc="output sbrefs")
t1w = OutputMultiObject(desc="output T1w images")
roi = OutputMultiObject(desc="output ROI images")
t2w = OutputMultiObject(desc="output T2w images")
flair = OutputMultiObject(desc="output FLAIR images")
asl = OutputMultiObject(desc="output ASL images")
class BIDSDataGrabber(SimpleInterface):
"""Collect files from a BIDS directory structure."""
input_spec = _BIDSDataGrabberInputSpec
output_spec = _BIDSDataGrabberOutputSpec
_require_funcs = True
def __init__(self, *args, **kwargs):
anat_only = kwargs.pop("anat_only")
super(BIDSDataGrabber, self).__init__(*args, **kwargs)
if anat_only is not None:
self._require_funcs = not anat_only
def _run_interface(self, runtime):
bids_dict = self.inputs.subject_data
self._results["out_dict"] = bids_dict
self._results.update(bids_dict)
if not bids_dict["t1w"]:
raise FileNotFoundError(
f"No T1w images found for subject sub-{self.inputs.subject_id}"
)
if self._require_funcs and not bids_dict["asl"]:
raise FileNotFoundError(
f"No ASL images found for subject sub-{self.inputs.subject_id}"
)
for imtype in ["t2w", "flair", "fmap", "sbref", "roi", "asl"]:
if not bids_dict[imtype]:
config.loggers.interface.info(
'No "%s" images found for sub-%s',
imtype,
self.inputs.subject_id,
)
return runtime
class DerivativesDataSink(BaseDerivativesDataSink):
"""Store derivative files.
A child class of the niworkflows DerivativesDataSink, using aslprep's configuration files.
"""
out_path_base = "aslprep"
_allowed_entities = set(config_entities)
_config_entities = config_entities
_config_entities_dict = merged_entities
_file_patterns = aslprep_spec["default_path_patterns"]