|
1
|
|
|
import copy |
|
2
|
|
|
import numbers |
|
3
|
|
|
from abc import ABC, abstractmethod |
|
4
|
|
|
from typing import Optional, Union, Tuple, List |
|
5
|
|
|
|
|
6
|
|
|
import torch |
|
7
|
|
|
import numpy as np |
|
8
|
|
|
import nibabel as nib |
|
9
|
|
|
import SimpleITK as sitk |
|
10
|
|
|
|
|
11
|
|
|
from .. import TypeData, DATA, AFFINE, TypeNumber |
|
12
|
|
|
from ..data.subject import Subject |
|
13
|
|
|
from ..data.image import Image, ScalarImage |
|
14
|
|
|
from ..utils import nib_to_sitk, sitk_to_nib, to_tuple |
|
15
|
|
|
from .interpolation import Interpolation |
|
16
|
|
|
|
|
17
|
|
|
|
|
18
|
|
|
TypeTransformInput = Union[ |
|
19
|
|
|
Subject, |
|
20
|
|
|
Image, |
|
21
|
|
|
torch.Tensor, |
|
22
|
|
|
np.ndarray, |
|
23
|
|
|
sitk.Image, |
|
24
|
|
|
dict, |
|
25
|
|
|
nib.Nifti1Image, |
|
26
|
|
|
] |
|
27
|
|
|
|
|
28
|
|
|
|
|
29
|
|
|
class Transform(ABC): |
|
30
|
|
|
"""Abstract class for all TorchIO transforms. |
|
31
|
|
|
|
|
32
|
|
|
All subclasses should overwrite |
|
33
|
|
|
:py:meth:`torchio.tranforms.Transform.apply_transform`, |
|
34
|
|
|
which takes data, applies some transformation and returns the result. |
|
35
|
|
|
|
|
36
|
|
|
The input can be an instance of |
|
37
|
|
|
:py:class:`torchio.Subject`, |
|
38
|
|
|
:py:class:`torchio.Image`, |
|
39
|
|
|
:py:class:`numpy.ndarray`, |
|
40
|
|
|
:py:class:`torch.Tensor`, |
|
41
|
|
|
:py:class:`SimpleITK.image`, |
|
42
|
|
|
or a Python dictionary. |
|
43
|
|
|
|
|
44
|
|
|
Args: |
|
45
|
|
|
p: Probability that this transform will be applied. |
|
46
|
|
|
copy: Make a shallow copy of the input before applying the transform. |
|
47
|
|
|
keys: Mandatory if the input is a Python dictionary. The transform will |
|
48
|
|
|
be applied only to the data in each key. |
|
49
|
|
|
""" |
|
50
|
|
|
def __init__( |
|
51
|
|
|
self, |
|
52
|
|
|
p: float = 1, |
|
53
|
|
|
copy: bool = True, |
|
54
|
|
|
keys: Optional[List[str]] = None, |
|
55
|
|
|
): |
|
56
|
|
|
self.probability = self.parse_probability(p) |
|
57
|
|
|
self.copy = copy |
|
58
|
|
|
self.keys = keys |
|
59
|
|
|
|
|
60
|
|
|
def __call__( |
|
61
|
|
|
self, |
|
62
|
|
|
data: TypeTransformInput, |
|
63
|
|
|
) -> TypeTransformInput: |
|
64
|
|
|
"""Transform data and return a result of the same type. |
|
65
|
|
|
|
|
66
|
|
|
Args: |
|
67
|
|
|
data: Instance of 1) :py:class:`~torchio.Subject`, 4D |
|
68
|
|
|
:py:class:`torch.Tensor` or NumPy array with dimensions |
|
69
|
|
|
:math:`(C, W, H, D)`, where :math:`C` is the number of channels |
|
70
|
|
|
and :math:`W, H, D` are the spatial dimensions. If the input is |
|
71
|
|
|
a tensor, the affine matrix will be set to identity. Other |
|
72
|
|
|
valid input types are a SimpleITK image, a |
|
73
|
|
|
:py:class:`torch.Image`, a NiBabel Nifti1 Image or a Python |
|
74
|
|
|
dictionary. The output type is the same as te input type. |
|
75
|
|
|
""" |
|
76
|
|
|
if torch.rand(1).item() > self.probability: |
|
77
|
|
|
return data |
|
78
|
|
|
data_parser = DataToSubject(data, keys=self.keys) |
|
79
|
|
|
subject = data_parser.get_subject() |
|
80
|
|
|
if self.copy: |
|
81
|
|
|
subject = copy.copy(subject) |
|
82
|
|
|
with np.errstate(all='raise'): |
|
83
|
|
|
transformed = self.apply_transform(subject) |
|
84
|
|
|
self.add_transform_to_subject_history(transformed) |
|
85
|
|
|
for image in transformed.get_images(intensity_only=False): |
|
86
|
|
|
ndim = image[DATA].ndim |
|
87
|
|
|
assert ndim == 4, f'Output of {self.name} is {ndim}D' |
|
88
|
|
|
output = data_parser.get_output(transformed) |
|
89
|
|
|
return output |
|
90
|
|
|
|
|
91
|
|
|
def __repr__(self): |
|
92
|
|
|
if hasattr(self, 'args_names'): |
|
93
|
|
|
names = self.args_names |
|
94
|
|
|
args_strings = [f'{arg}={getattr(self, arg)}' for arg in names] |
|
95
|
|
|
if hasattr(self, 'invert_transform') and self.invert_transform: |
|
96
|
|
|
args_strings.append('invert=True') |
|
97
|
|
|
args_string = ', '.join(args_strings) |
|
98
|
|
|
return f'{self.name}({args_string})' |
|
99
|
|
|
else: |
|
100
|
|
|
return super().__repr__() |
|
101
|
|
|
|
|
102
|
|
|
@abstractmethod |
|
103
|
|
|
def apply_transform(self, subject: Subject): |
|
104
|
|
|
raise NotImplementedError |
|
105
|
|
|
|
|
106
|
|
|
def add_transform_to_subject_history(self, subject): |
|
107
|
|
|
from .augmentation import RandomTransform |
|
108
|
|
|
from . import Compose, OneOf, CropOrPad |
|
109
|
|
|
call_others = ( |
|
110
|
|
|
RandomTransform, |
|
111
|
|
|
Compose, |
|
112
|
|
|
OneOf, |
|
113
|
|
|
CropOrPad, |
|
114
|
|
|
) |
|
115
|
|
|
if not isinstance(self, call_others): |
|
116
|
|
|
subject.add_transform(self, self.get_arguments()) |
|
117
|
|
|
|
|
118
|
|
|
@staticmethod |
|
119
|
|
|
def to_range(n, around): |
|
120
|
|
|
if around is None: |
|
121
|
|
|
return 0, n |
|
122
|
|
|
else: |
|
123
|
|
|
return around - n, around + n |
|
124
|
|
|
|
|
125
|
|
|
def parse_params(self, params, around, name, make_ranges=True, **kwargs): |
|
126
|
|
|
params = to_tuple(params) |
|
127
|
|
|
if len(params) == 1 or (len(params) == 2 and make_ranges): # d or (a, b) |
|
128
|
|
|
params *= 3 # (d, d, d) or (a, b, a, b, a, b) |
|
129
|
|
|
if len(params) == 3 and make_ranges: # (a, b, c) |
|
130
|
|
|
items = [self.to_range(n, around) for n in params] |
|
131
|
|
|
# (-a, a, -b, b, -c, c) or (1-a, 1+a, 1-b, 1+b, 1-c, 1+c) |
|
132
|
|
|
params = [n for prange in items for n in prange] |
|
133
|
|
|
if make_ranges and len(params) != 6: |
|
134
|
|
|
if len(params) != 6: |
|
135
|
|
|
message = ( |
|
136
|
|
|
f'If "{name}" is a sequence, it must have length 2, 3 or 6,' |
|
137
|
|
|
f' not {len(params)}' |
|
138
|
|
|
) |
|
139
|
|
|
raise ValueError(message) |
|
140
|
|
|
for param_range in zip(params[::2], params[1::2]): |
|
141
|
|
|
self.parse_range(param_range, name, **kwargs) |
|
142
|
|
|
return tuple(params) |
|
143
|
|
|
|
|
144
|
|
|
@staticmethod |
|
145
|
|
|
def parse_range( |
|
146
|
|
|
nums_range: Union[TypeNumber, Tuple[TypeNumber, TypeNumber]], |
|
147
|
|
|
name: str, |
|
148
|
|
|
min_constraint: TypeNumber = None, |
|
149
|
|
|
max_constraint: TypeNumber = None, |
|
150
|
|
|
type_constraint: type = None, |
|
151
|
|
|
) -> Tuple[TypeNumber, TypeNumber]: |
|
152
|
|
|
r"""Adapted from ``torchvision.transforms.RandomRotation``. |
|
153
|
|
|
|
|
154
|
|
|
Args: |
|
155
|
|
|
nums_range: Tuple of two numbers :math:`(n_{min}, n_{max})`, |
|
156
|
|
|
where :math:`n_{min} \leq n_{max}`. |
|
157
|
|
|
If a single positive number :math:`n` is provided, |
|
158
|
|
|
:math:`n_{min} = -n` and :math:`n_{max} = n`. |
|
159
|
|
|
name: Name of the parameter, so that an informative error message |
|
160
|
|
|
can be printed. |
|
161
|
|
|
min_constraint: Minimal value that :math:`n_{min}` can take, |
|
162
|
|
|
default is None, i.e. there is no minimal value. |
|
163
|
|
|
max_constraint: Maximal value that :math:`n_{max}` can take, |
|
164
|
|
|
default is None, i.e. there is no maximal value. |
|
165
|
|
|
type_constraint: Precise type that :math:`n_{max}` and |
|
166
|
|
|
:math:`n_{min}` must take. |
|
167
|
|
|
|
|
168
|
|
|
Returns: |
|
169
|
|
|
A tuple of two numbers :math:`(n_{min}, n_{max})`. |
|
170
|
|
|
|
|
171
|
|
|
Raises: |
|
172
|
|
|
ValueError: if :attr:`nums_range` is negative |
|
173
|
|
|
ValueError: if :math:`n_{max}` or :math:`n_{min}` is not a number |
|
174
|
|
|
ValueError: if :math:`n_{max} \lt n_{min}` |
|
175
|
|
|
ValueError: if :attr:`min_constraint` is not None and |
|
176
|
|
|
:math:`n_{min}` is smaller than :attr:`min_constraint` |
|
177
|
|
|
ValueError: if :attr:`max_constraint` is not None and |
|
178
|
|
|
:math:`n_{max}` is greater than :attr:`max_constraint` |
|
179
|
|
|
ValueError: if :attr:`type_constraint` is not None and |
|
180
|
|
|
:math:`n_{max}` and :math:`n_{max}` are not of type |
|
181
|
|
|
:attr:`type_constraint`. |
|
182
|
|
|
""" |
|
183
|
|
|
if isinstance(nums_range, numbers.Number): # single number given |
|
184
|
|
|
if nums_range < 0: |
|
185
|
|
|
raise ValueError( |
|
186
|
|
|
f'If {name} is a single number,' |
|
187
|
|
|
f' it must be positive, not {nums_range}') |
|
188
|
|
|
if min_constraint is not None and nums_range < min_constraint: |
|
189
|
|
|
raise ValueError( |
|
190
|
|
|
f'If {name} is a single number, it must be greater' |
|
191
|
|
|
f' than {min_constraint}, not {nums_range}' |
|
192
|
|
|
) |
|
193
|
|
|
if max_constraint is not None and nums_range > max_constraint: |
|
194
|
|
|
raise ValueError( |
|
195
|
|
|
f'If {name} is a single number, it must be smaller' |
|
196
|
|
|
f' than {max_constraint}, not {nums_range}' |
|
197
|
|
|
) |
|
198
|
|
|
if type_constraint is not None: |
|
199
|
|
|
if not isinstance(nums_range, type_constraint): |
|
200
|
|
|
raise ValueError( |
|
201
|
|
|
f'If {name} is a single number, it must be of' |
|
202
|
|
|
f' type {type_constraint}, not {nums_range}' |
|
203
|
|
|
) |
|
204
|
|
|
min_range = -nums_range if min_constraint is None else nums_range |
|
205
|
|
|
return (min_range, nums_range) |
|
206
|
|
|
|
|
207
|
|
|
try: |
|
208
|
|
|
min_value, max_value = nums_range |
|
209
|
|
|
except (TypeError, ValueError): |
|
210
|
|
|
raise ValueError( |
|
211
|
|
|
f'If {name} is not a single number, it must be' |
|
212
|
|
|
f' a sequence of len 2, not {nums_range}' |
|
213
|
|
|
) |
|
214
|
|
|
|
|
215
|
|
|
min_is_number = isinstance(min_value, numbers.Number) |
|
216
|
|
|
max_is_number = isinstance(max_value, numbers.Number) |
|
217
|
|
|
if not min_is_number or not max_is_number: |
|
218
|
|
|
message = ( |
|
219
|
|
|
f'{name} values must be numbers, not {nums_range}') |
|
220
|
|
|
raise ValueError(message) |
|
221
|
|
|
|
|
222
|
|
|
if min_value > max_value: |
|
223
|
|
|
raise ValueError( |
|
224
|
|
|
f'If {name} is a sequence, the second value must be' |
|
225
|
|
|
f' equal or greater than the first, but it is {nums_range}') |
|
226
|
|
|
|
|
227
|
|
|
if min_constraint is not None and min_value < min_constraint: |
|
228
|
|
|
raise ValueError( |
|
229
|
|
|
f'If {name} is a sequence, the first value must be greater' |
|
230
|
|
|
f' than {min_constraint}, but it is {min_value}' |
|
231
|
|
|
) |
|
232
|
|
|
|
|
233
|
|
|
if max_constraint is not None and max_value > max_constraint: |
|
234
|
|
|
raise ValueError( |
|
235
|
|
|
f'If {name} is a sequence, the second value must be smaller' |
|
236
|
|
|
f' than {max_constraint}, but it is {max_value}' |
|
237
|
|
|
) |
|
238
|
|
|
|
|
239
|
|
|
if type_constraint is not None: |
|
240
|
|
|
min_type_ok = isinstance(min_value, type_constraint) |
|
241
|
|
|
max_type_ok = isinstance(max_value, type_constraint) |
|
242
|
|
|
if not min_type_ok or not max_type_ok: |
|
243
|
|
|
raise ValueError( |
|
244
|
|
|
f'If "{name}" is a sequence, its values must be of' |
|
245
|
|
|
f' type "{type_constraint}", not "{type(nums_range)}"' |
|
246
|
|
|
) |
|
247
|
|
|
return nums_range |
|
248
|
|
|
|
|
249
|
|
|
@staticmethod |
|
250
|
|
|
def parse_interpolation(interpolation: str) -> str: |
|
251
|
|
|
interpolation = interpolation.lower() |
|
252
|
|
|
is_string = isinstance(interpolation, str) |
|
253
|
|
|
supported_values = [key.name.lower() for key in Interpolation] |
|
254
|
|
|
is_supported = interpolation.lower() in supported_values |
|
255
|
|
|
if is_string and is_supported: |
|
256
|
|
|
return interpolation |
|
257
|
|
|
message = ( |
|
258
|
|
|
f'Interpolation "{interpolation}" of type {type(interpolation)}' |
|
259
|
|
|
f' must be a string among the supported values: {supported_values}' |
|
260
|
|
|
) |
|
261
|
|
|
raise TypeError(message) |
|
262
|
|
|
|
|
263
|
|
|
@staticmethod |
|
264
|
|
|
def parse_probability(probability: float) -> float: |
|
265
|
|
|
is_number = isinstance(probability, numbers.Number) |
|
266
|
|
|
if not (is_number and 0 <= probability <= 1): |
|
267
|
|
|
message = ( |
|
268
|
|
|
'Probability must be a number in [0, 1],' |
|
269
|
|
|
f' not {probability}' |
|
270
|
|
|
) |
|
271
|
|
|
raise ValueError(message) |
|
272
|
|
|
return probability |
|
273
|
|
|
|
|
274
|
|
|
@staticmethod |
|
275
|
|
|
def nib_to_sitk(data: TypeData, affine: TypeData) -> sitk.Image: |
|
276
|
|
|
return nib_to_sitk(data, affine) |
|
277
|
|
|
|
|
278
|
|
|
@staticmethod |
|
279
|
|
|
def sitk_to_nib(image: sitk.Image) -> Tuple[torch.Tensor, np.ndarray]: |
|
280
|
|
|
return sitk_to_nib(image) |
|
281
|
|
|
|
|
282
|
|
|
@property |
|
283
|
|
|
def name(self): |
|
284
|
|
|
return self.__class__.__name__ |
|
285
|
|
|
|
|
286
|
|
|
def get_arguments(self): |
|
287
|
|
|
""" |
|
288
|
|
|
Return a dictionary with the arguments that would be necessary to |
|
289
|
|
|
reproduce the transform exactly. |
|
290
|
|
|
""" |
|
291
|
|
|
return {name: getattr(self, name) for name in self.args_names} |
|
292
|
|
|
|
|
293
|
|
|
def is_invertible(self): |
|
294
|
|
|
return hasattr(self, 'invert_transform') |
|
295
|
|
|
|
|
296
|
|
|
def inverse(self): |
|
297
|
|
|
if not self.is_invertible(): |
|
298
|
|
|
raise RuntimeError(f'{self.name} is not invertible') |
|
299
|
|
|
new = copy.deepcopy(self) |
|
300
|
|
|
new.invert_transform = not self.invert_transform |
|
301
|
|
|
return new |
|
302
|
|
|
|
|
303
|
|
|
|
|
304
|
|
|
class DataToSubject: |
|
305
|
|
|
def __init__( |
|
306
|
|
|
self, |
|
307
|
|
|
data: TypeTransformInput, |
|
308
|
|
|
keys: Optional[List[str]] = None, |
|
309
|
|
|
): |
|
310
|
|
|
self.data = data |
|
311
|
|
|
self.keys = keys |
|
312
|
|
|
self.default_image_name = 'default_image_name' |
|
313
|
|
|
self.is_tensor = False |
|
314
|
|
|
self.is_array = False |
|
315
|
|
|
self.is_dict = False |
|
316
|
|
|
self.is_image = False |
|
317
|
|
|
self.is_sitk = False |
|
318
|
|
|
self.is_nib = False |
|
319
|
|
|
|
|
320
|
|
|
def get_subject(self): |
|
321
|
|
|
if isinstance(self.data, nib.Nifti1Image): |
|
322
|
|
|
tensor = self.data.get_fdata(dtype=np.float32) |
|
323
|
|
|
data = ScalarImage(tensor=tensor, affine=self.data.affine) |
|
324
|
|
|
subject = self._get_subject_from_image(data) |
|
325
|
|
|
self.is_nib = True |
|
326
|
|
|
elif isinstance(self.data, (np.ndarray, torch.Tensor)): |
|
327
|
|
|
subject = self._parse_tensor(self.data) |
|
328
|
|
|
self.is_array = isinstance(self.data, np.ndarray) |
|
329
|
|
|
self.is_tensor = True |
|
330
|
|
|
elif isinstance(self.data, Image): |
|
331
|
|
|
subject = self._get_subject_from_image(self.data) |
|
332
|
|
|
self.is_image = True |
|
333
|
|
|
elif isinstance(self.data, Subject): |
|
334
|
|
|
subject = self.data |
|
335
|
|
|
elif isinstance(self.data, sitk.Image): |
|
336
|
|
|
subject = self._get_subject_from_sitk_image(self.data) |
|
337
|
|
|
self.is_sitk = True |
|
338
|
|
|
elif isinstance(self.data, dict): # e.g. Eisen or MONAI dicts |
|
339
|
|
|
if self.keys is None: |
|
340
|
|
|
message = ( |
|
341
|
|
|
'If input is a dictionary, a value for "keys" must be' |
|
342
|
|
|
' specified when instantiating the transform' |
|
343
|
|
|
) |
|
344
|
|
|
raise RuntimeError(message) |
|
345
|
|
|
subject = self._get_subject_from_dict(self.data, self.keys) |
|
346
|
|
|
self.is_dict = True |
|
347
|
|
|
else: |
|
348
|
|
|
raise ValueError(f'Input type not recognized: {type(self.data)}') |
|
349
|
|
|
self._parse_subject(subject) |
|
350
|
|
|
return subject |
|
351
|
|
|
|
|
352
|
|
|
def get_output(self, transformed): |
|
353
|
|
|
if self.is_tensor or self.is_sitk: |
|
354
|
|
|
image = transformed[self.default_image_name] |
|
355
|
|
|
transformed = image[DATA] |
|
356
|
|
|
if self.is_array: |
|
357
|
|
|
transformed = transformed.numpy() |
|
358
|
|
|
elif self.is_sitk: |
|
359
|
|
|
transformed = nib_to_sitk(image[DATA], image[AFFINE]) |
|
360
|
|
|
elif self.is_image: |
|
361
|
|
|
transformed = transformed[self.default_image_name] |
|
362
|
|
|
elif self.is_dict: |
|
363
|
|
|
transformed = dict(transformed) |
|
364
|
|
|
for key, value in transformed.items(): |
|
365
|
|
|
if isinstance(value, Image): |
|
366
|
|
|
transformed[key] = value.data |
|
367
|
|
|
elif self.is_nib: |
|
368
|
|
|
image = transformed[self.default_image_name] |
|
369
|
|
|
data = image[DATA] |
|
370
|
|
|
if len(data) > 1: |
|
371
|
|
|
message = ( |
|
372
|
|
|
'Multichannel images not supported for input of type' |
|
373
|
|
|
' nibabel.nifti.Nifti1Image' |
|
374
|
|
|
) |
|
375
|
|
|
raise RuntimeError(message) |
|
376
|
|
|
transformed = nib.Nifti1Image(data[0].numpy(), image[AFFINE]) |
|
377
|
|
|
return transformed |
|
378
|
|
|
|
|
379
|
|
|
@staticmethod |
|
380
|
|
|
def _parse_subject(subject: Subject) -> None: |
|
381
|
|
|
if not isinstance(subject, Subject): |
|
382
|
|
|
message = ( |
|
383
|
|
|
'Input to a transform must be a tensor or an instance' |
|
384
|
|
|
f' of torchio.Subject, not "{type(subject)}"' |
|
385
|
|
|
) |
|
386
|
|
|
raise RuntimeError(message) |
|
387
|
|
|
|
|
388
|
|
|
def _parse_tensor(self, data: TypeData) -> Subject: |
|
389
|
|
|
if data.ndim != 4: |
|
390
|
|
|
message = ( |
|
391
|
|
|
'The input must be a 4D tensor with dimensions' |
|
392
|
|
|
f' (channels, x, y, z) but it has shape {tuple(data.shape)}' |
|
393
|
|
|
) |
|
394
|
|
|
raise ValueError(message) |
|
395
|
|
|
return self._get_subject_from_tensor(data) |
|
396
|
|
|
|
|
397
|
|
|
def _get_subject_from_tensor(self, tensor: torch.Tensor) -> Subject: |
|
398
|
|
|
image = ScalarImage(tensor=tensor) |
|
399
|
|
|
return self._get_subject_from_image(image) |
|
400
|
|
|
|
|
401
|
|
|
def _get_subject_from_image(self, image: Image) -> Subject: |
|
402
|
|
|
subject = Subject({self.default_image_name: image}) |
|
403
|
|
|
return subject |
|
404
|
|
|
|
|
405
|
|
|
@staticmethod |
|
406
|
|
|
def _get_subject_from_dict( |
|
407
|
|
|
data: dict, |
|
408
|
|
|
image_keys: List[str], |
|
409
|
|
|
) -> Subject: |
|
410
|
|
|
subject_dict = {} |
|
411
|
|
|
for key, value in data.items(): |
|
412
|
|
|
if key in image_keys: |
|
413
|
|
|
value = ScalarImage(tensor=value) |
|
414
|
|
|
subject_dict[key] = value |
|
415
|
|
|
return Subject(subject_dict) |
|
416
|
|
|
|
|
417
|
|
|
def _get_subject_from_sitk_image(self, image): |
|
418
|
|
|
tensor, affine = sitk_to_nib(image) |
|
419
|
|
|
image = ScalarImage(tensor=tensor, affine=affine) |
|
420
|
|
|
return self._get_subject_from_image(image) |
|
421
|
|
|
|