torchio.utils.create_dummy_dataset() - Code Metrics - Inspection of "Add examples gallery to documentation" - fepegar/torchio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#508)

by Fernando

created 2021-04-25 20:39 UTC

torchio.utils.create_dummy_dataset() B

↳ Parent: torchio.utils

Complexity

Conditions

Size

Total Lines	59
Code Lines	49

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	49
nop	6
dl	0
loc	59
rs	7.269
c	0
b	0
f	0

How to fix Long Method

import ast
import gzip
import shutil
import tempfile
from pathlib import Path
from typing import Union, Iterable, Tuple, Any, Optional, List, Sequence

from torch.utils.data._utils.collate import default_collate
import numpy as np
import nibabel as nib
import SimpleITK as sitk
from tqdm import trange

from .typing import TypeNumber, TypePath


def to_tuple(
        value: Union[TypeNumber, Iterable[TypeNumber]],
        length: int = 1,
        ) -> Tuple[TypeNumber, ...]:
    """
    to_tuple(1, length=1) -> (1,)
    to_tuple(1, length=3) -> (1, 1, 1)

    If value is an iterable, n is ignored and tuple(value) is returned
    to_tuple((1,), length=1) -> (1,)
    to_tuple((1, 2), length=1) -> (1, 2)
    to_tuple([1, 2], length=3) -> (1, 2)
    """
    try:
        iter(value)
        value = tuple(value)
    except TypeError:
        value = length * (value,)
    return value


def get_stem(
        path: Union[TypePath, List[TypePath]]
        ) -> Union[str, List[str]]:
    """
    '/home/user/image.nii.gz' -> 'image'
    """
    def _get_stem(path_string):
        return Path(path_string).name.split('.')[0]
    if isinstance(path, (str, Path)):
        return _get_stem(path)
    return [_get_stem(p) for p in path]


def create_dummy_dataset(
        num_images: int,
        size_range: Tuple[int, int],
        directory: Optional[TypePath] = None,
        suffix: str = '.nii.gz',
        force: bool = False,
        verbose: bool = False,
        ):
    from .data import ScalarImage, LabelMap, Subject
    output_dir = tempfile.gettempdir() if directory is None else directory
    output_dir = Path(output_dir)
    images_dir = output_dir / 'dummy_images'
    labels_dir = output_dir / 'dummy_labels'

    if force:
        shutil.rmtree(images_dir)
        shutil.rmtree(labels_dir)

    subjects: List[Subject] = []
    if images_dir.is_dir():
        for i in trange(num_images):
            image_path = images_dir / f'image_{i}{suffix}'
            label_path = labels_dir / f'label_{i}{suffix}'
            subject = Subject(
                one_modality=ScalarImage(image_path),
                segmentation=LabelMap(label_path),
            )
            subjects.append(subject)
    else:
        images_dir.mkdir(exist_ok=True, parents=True)
        labels_dir.mkdir(exist_ok=True, parents=True)
        if verbose:
            print('Creating dummy dataset...')  # noqa: T001
            iterable = trange(num_images)
        else:
            iterable = range(num_images)
        for i in iterable:
            shape = np.random.randint(*size_range, size=3)
            affine = np.eye(4)
            image = np.random.rand(*shape)
            label = np.ones_like(image)
            label[image < 0.33] = 0
            label[image > 0.66] = 2
            image *= 255

            image_path = images_dir / f'image_{i}{suffix}'
            nii = nib.Nifti1Image(image.astype(np.uint8), affine)
            nii.to_filename(str(image_path))

            label_path = labels_dir / f'label_{i}{suffix}'
            nii = nib.Nifti1Image(label.astype(np.uint8), affine)
            nii.to_filename(str(label_path))

            subject = Subject(
                one_modality=ScalarImage(image_path),
                segmentation=LabelMap(label_path),
            )
            subjects.append(subject)
    return subjects


def apply_transform_to_file(
        input_path: TypePath,
        transform,  # : Transform seems to create a circular import
        output_path: TypePath,
        class_: str = 'ScalarImage',
        verbose: bool = False,
        ):
    from . import data
    image = getattr(data, class_)(input_path)
    subject = data.Subject(image=image)
    transformed = transform(subject)
    transformed.image.save(output_path)
    if verbose and transformed.history:
        print('Applied transform:', transformed.history[0])  # noqa: T001


def guess_type(string: str) -> Any:
    # Adapted from
    # https://www.reddit.com/r/learnpython/comments/4599hl/module_to_guess_type_from_a_string/czw3f5s
    string = string.replace(' ', '')
    try:
        value = ast.literal_eval(string)
    except ValueError:
        result_type = str
    else:
        result_type = type(value)
    if result_type in (list, tuple):
        string = string[1:-1]  # remove brackets
        split = string.split(',')
        list_result = [guess_type(n) for n in split]
        value = tuple(list_result) if result_type is tuple else list_result
        return value
    try:
        value = result_type(string)
    except TypeError:
        value = None
    return value


def get_torchio_cache_dir():
    return Path('~/.cache/torchio').expanduser()


def compress(input_path, output_path):
    with open(input_path, 'rb') as f_in:
        with gzip.open(output_path, 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)


def check_sequence(sequence: Sequence, name: str):
    try:
        iter(sequence)
    except TypeError:
        message = f'"{name}" must be a sequence, not {type(name)}'
        raise TypeError(message)


def get_major_sitk_version() -> int:
    # This attribute was added in version 2
    # https://github.com/SimpleITK/SimpleITK/pull/1171
    version = getattr(sitk, '__version__', None)
    major_version = 1 if version is None else 2
    return major_version


def history_collate(batch: Sequence, collate_transforms=True):
    attr = 'history' if collate_transforms else 'applied_transforms'
    # Adapted from
    # https://github.com/romainVala/torchQC/blob/master/segmentation/collate_functions.py
    from .data import Subject
    first_element = batch[0]
    if isinstance(first_element, Subject):
        dictionary = {
            key: default_collate([d[key] for d in batch])
            for key in first_element
        }
        if hasattr(first_element, attr):
            dictionary.update({attr: [getattr(d, attr) for d in batch]})
        return dictionary


def get_subclasses(target_class: type) -> List[type]:
    subclasses = target_class.__subclasses__()
    subclasses += sum((get_subclasses(cls) for cls in subclasses), [])

    return subclasses


def get_first_item(data_loader):
    return next(iter(data_loader))


1			import ast
2			import gzip
3			import shutil
4			import tempfile
5			from pathlib import Path
6			from typing import Union, Iterable, Tuple, Any, Optional, List, Sequence
7
8			from torch.utils.data._utils.collate import default_collate
9			import numpy as np
10			import nibabel as nib
11			import SimpleITK as sitk
12			from tqdm import trange
13
14			from .typing import TypeNumber, TypePath
15
16
17			def to_tuple(
18			value: Union[TypeNumber, Iterable[TypeNumber]],
19			length: int = 1,
20			) -> Tuple[TypeNumber, ...]:
21			"""
22			to_tuple(1, length=1) -> (1,)
23			to_tuple(1, length=3) -> (1, 1, 1)
24
25			If value is an iterable, n is ignored and tuple(value) is returned
26			to_tuple((1,), length=1) -> (1,)
27			to_tuple((1, 2), length=1) -> (1, 2)
28			to_tuple([1, 2], length=3) -> (1, 2)
29			"""
30			try:
31			iter(value)
32			value = tuple(value)
33			except TypeError:
34			value = length * (value,)
35			return value
36
37
38			def get_stem(
39			path: Union[TypePath, List[TypePath]]
40			) -> Union[str, List[str]]:
41			"""
42			'/home/user/image.nii.gz' -> 'image'
43			"""
44			def _get_stem(path_string):
45			return Path(path_string).name.split('.')[0]
46			if isinstance(path, (str, Path)):
47			return _get_stem(path)
48			return [_get_stem(p) for p in path]
49
50
51			def create_dummy_dataset(
52			num_images: int,
53			size_range: Tuple[int, int],
54			directory: Optional[TypePath] = None,
55			suffix: str = '.nii.gz',
56			force: bool = False,
57			verbose: bool = False,
58			):
59			from .data import ScalarImage, LabelMap, Subject
60			output_dir = tempfile.gettempdir() if directory is None else directory
61			output_dir = Path(output_dir)
62			images_dir = output_dir / 'dummy_images'
63			labels_dir = output_dir / 'dummy_labels'
64
65			if force:
66			shutil.rmtree(images_dir)
67			shutil.rmtree(labels_dir)
68
69			subjects: List[Subject] = []
70			if images_dir.is_dir():
71			for i in trange(num_images):
72			image_path = images_dir / f'image_{i}{suffix}'
73			label_path = labels_dir / f'label_{i}{suffix}'
74			subject = Subject(
75			one_modality=ScalarImage(image_path),
76			segmentation=LabelMap(label_path),
77			)
78			subjects.append(subject)
79			else:
80			images_dir.mkdir(exist_ok=True, parents=True)
81			labels_dir.mkdir(exist_ok=True, parents=True)
82			if verbose:
83			print('Creating dummy dataset...') # noqa: T001
84			iterable = trange(num_images)
85			else:
86			iterable = range(num_images)
87			for i in iterable:
88			shape = np.random.randint(*size_range, size=3)
89			affine = np.eye(4)
90			image = np.random.rand(*shape)
91			label = np.ones_like(image)
92			label[image < 0.33] = 0
93			label[image > 0.66] = 2
94			image *= 255
95
96			image_path = images_dir / f'image_{i}{suffix}'
97			nii = nib.Nifti1Image(image.astype(np.uint8), affine)
98			nii.to_filename(str(image_path))
99
100			label_path = labels_dir / f'label_{i}{suffix}'
101			nii = nib.Nifti1Image(label.astype(np.uint8), affine)
102			nii.to_filename(str(label_path))
103
104			subject = Subject(
105			one_modality=ScalarImage(image_path),
106			segmentation=LabelMap(label_path),
107			)
108			subjects.append(subject)
109			return subjects
110
111
112			def apply_transform_to_file(
113			input_path: TypePath,
114			transform, # : Transform seems to create a circular import
115			output_path: TypePath,
116			class_: str = 'ScalarImage',
117			verbose: bool = False,
118			):
119			from . import data
120			image = getattr(data, class_)(input_path)
121			subject = data.Subject(image=image)
122			transformed = transform(subject)
123			transformed.image.save(output_path)
124			if verbose and transformed.history:
125			print('Applied transform:', transformed.history[0]) # noqa: T001
126
127
128			def guess_type(string: str) -> Any:
129			# Adapted from
130			# https://www.reddit.com/r/learnpython/comments/4599hl/module_to_guess_type_from_a_string/czw3f5s
131			string = string.replace(' ', '')
132			try:
133			value = ast.literal_eval(string)
134			except ValueError:
135			result_type = str
136			else:
137			result_type = type(value)
138			if result_type in (list, tuple):
139			string = string[1:-1] # remove brackets
140			split = string.split(',')
141			list_result = [guess_type(n) for n in split]
142			value = tuple(list_result) if result_type is tuple else list_result
143			return value
144			try:
145			value = result_type(string)
146			except TypeError:
147			value = None
148			return value
149
150
151			def get_torchio_cache_dir():
152			return Path('~/.cache/torchio').expanduser()
153
154
155			def compress(input_path, output_path):
156			with open(input_path, 'rb') as f_in:
157			with gzip.open(output_path, 'wb') as f_out:
158			shutil.copyfileobj(f_in, f_out)
159
160
161			def check_sequence(sequence: Sequence, name: str):
162			try:
163			iter(sequence)
164			except TypeError:
165			message = f'"{name}" must be a sequence, not {type(name)}'
166			raise TypeError(message)
167
168
169			def get_major_sitk_version() -> int:
170			# This attribute was added in version 2
171			# https://github.com/SimpleITK/SimpleITK/pull/1171
172			version = getattr(sitk, '__version__', None)
173			major_version = 1 if version is None else 2
174			return major_version
175
176
177			def history_collate(batch: Sequence, collate_transforms=True):
178			attr = 'history' if collate_transforms else 'applied_transforms'
179			# Adapted from
180			# https://github.com/romainVala/torchQC/blob/master/segmentation/collate_functions.py
181			from .data import Subject
182			first_element = batch[0]
183			if isinstance(first_element, Subject):
184			dictionary = {
185			key: default_collate([d[key] for d in batch])
186			for key in first_element
187			}
188			if hasattr(first_element, attr):
189			dictionary.update({attr: [getattr(d, attr) for d in batch]})
190			return dictionary
191
192
193			def get_subclasses(target_class: type) -> List[type]:
194			subclasses = target_class.__subclasses__()
195			subclasses += sum((get_subclasses(cls) for cls in subclasses), [])
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-29 14:23 UTC by Report Bug Copy Issue Report The variable `cls` does not seem to be defined. Loading history...
196			return subclasses
197
198
199			def get_first_item(data_loader):
200			return next(iter(data_loader))
201

fepegar / torchio

Pull Request — master (#508)

torchio.utils.create_dummy_dataset() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like