-
Notifications
You must be signed in to change notification settings - Fork 5
/
LoadImageXNATd.py
90 lines (69 loc) · 3.92 KB
/
LoadImageXNATd.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
"""
MONAI MapTransform for importing image data from XNAT
"""
import tempfile
import xnat
import os
import glob
from monai.transforms import MapTransform, LoadImage
from monai.config import KeysCollection
from mlops.utils.logger import logger
from monai.transforms import Transform
class LoadImageXNATd(MapTransform):
"""
MapTransform for importing image data from XNAT
"""
def __init__(self, keys: KeysCollection, xnat_configuration: dict = None,
image_loader: Transform = LoadImage(), validate_data: bool = False, expected_filetype_ext: str = '.dcm'):
super().__init__(keys)
self.image_loader = image_loader
self.xnat_configuration = xnat_configuration
self.expected_filetype = expected_filetype_ext
self.validate_data = validate_data
def __call__(self, data):
"""
Checks for the requested keys in the input data dictionary.
If specified key is found then will loop over actions in action list and apply each to the value of the
requested key, if no action is triggered then raise a warning actions arefunctions that return any of projects,
subjects, experiments, scans, or resources XNAT object along with a key to be used in the data dictionary
Each action function should locate a single image object in XNAT. This image object is then downloaded to a
temporary directory and loaded into memory as the value defined by key set by the actions' data_label.
If validate_data is true then NO data will be downloaded. In this case the transform will loop over the actions
but will instead return a true/false value for each data sample. This can be used to remove samples where the
data is not present in XNAT.
:param data: dictionary of data
:return:
"""
d = dict(data)
for key in self.keys:
if key in data:
# data_label = d['data_label']
with xnat.connect(server=self.xnat_configuration['server'],
user=self.xnat_configuration['user'],
password=self.xnat_configuration['password'],
verify=self.xnat_configuration['verify'],
) as session:
"Check data list has no duplicate keys"
if len(set([x['data_label'] for x in d[key]])) != len([x['data_label'] for x in d[key]]):
logger.warn('Multiple images with identical labels found')
raise
"Download image from XNAT"
for item in d[key]:
data_label = item['data_label']
with tempfile.TemporaryDirectory() as tmpdirname:
session_obj = session.create_object(item['xnat_uri'])
session_obj.download_dir(tmpdirname)
images_path = glob.glob(os.path.join(tmpdirname, '**/*' + self.expected_filetype), recursive=True)
# image loader needs full path to load single images
logger.info(f"Downloading images: {images_path}")
if len(images_path) == 1:
image = self.image_loader(images_path)
# image loader needs directory path to load 3D images
else:
"find unique directories in list of image paths"
image_dirs = list(set(os.path.dirname(image_path) for image_path in images_path))
if len(image_dirs) > 1:
raise ValueError(f'More than one image series found in {images_path}')
image = self.image_loader(image_dirs[0])
d[data_label] = image
return d