Source code for

# -*- coding: utf-8 -*-
"""This module loads images from csv files and outputs numpy arrays."""
from __future__ import absolute_import, division, print_function

from copy import deepcopy
import argparse
import numpy as np
import pandas
import tensorflow as tf
from six import string_types

from import dtype_casting
from import COLUMN_UNIQ_ID
from import ImageFactory
from niftynet.layer.base_layer import Layer, DataDependentLayer, RandomisedLayer
from niftynet.utilities.user_parameters_helper import make_input_tuple
from niftynet.utilities.util_common import print_progress_bar, ParserNamespace
from import ImageSetsPartitioner
from niftynet.utilities.util_common import look_up_operations


    'csv_file', 'path_to_search', 'csv_data_file', 'filename_removefromid',
    'filename_contains', 'filename_not_contains', 'to_ohe',
    'interp_order', 'loader', 'pixdim', 'axcodes', 'spatial_window_size'}

[docs]def infer_tf_dtypes(image_array): """ Choosing a suitable tf dtype based on the dtype of input numpy array. """ return dtype_casting( image_array.dtype[0], image_array.interp_order[0], as_tf=True)
[docs]class ImageReader(Layer): """ For a concrete example:: _input_sources define multiple modality mappings, e.g., _input_sources {'image': ('T1', 'T2'), 'label': ('manual_map',)} means: 'image' consists of two components, formed by concatenating 'T1' and 'T2' input source images. 'label' consists of one component, loading from 'manual_map' :param self._names: a tuple of the output names of this reader. ``('image', 'labels')`` :param self._shapes: the shapes after combining input sources ``{'image': (192, 160, 192, 1, 2), 'label': (192, 160, 192, 1, 1)}`` :param self._dtypes: store the dictionary of tensorflow shapes ``{'image': tf.float32, 'label': tf.float32}`` :param self.output_list: a list of dictionaries, with each item:: {'image': < object>, 'label': < object>} """ def __init__(self, names=None): # list of file names self._file_list = None self._input_sources = None self._spatial_ranks = None self._shapes = None self._dtypes = None self._names = None if names: self.names = names # list of image objects self.output_list = None self.current_id = -1 self.preprocessors = [] super(ImageReader, self).__init__(name='image_reader')
[docs] def initialise(self, data_param, task_param=None, file_list=None): """ ``task_param`` specifies how to combine user input modalities. e.g., for multimodal segmentation 'image' corresponds to multiple modality sections, 'label' corresponds to one modality section This function converts elements of ``file_list`` into dictionaries of image objects, and save them to ``self.output_list``. e.g.:: data_param = {'T1': {'path_to_search': 'path/to/t1'} 'T2': {'path_to_search': 'path/to/t2'}} loads pairs of T1 and T1 images (grouped by matching the filename). The reader's output is in the form of ``{'T1': np.array, 'T2': np.array}``. If the (optional) ``task_param`` is specified:: task_param = {'image': ('T1', 'T2')} the reader loads pairs of T1 and T2 and returns the concatenated image (both modalities should have the same spatial dimensions). The reader's output is in the form of ``{'image': np.array}``. :param data_param: dictionary of input sections :param task_param: dictionary of grouping :param file_list: a dataframe generated by ImagePartitioner for cross validation, so that the reader only loads files in training/inference phases. :return: the initialised reader instance """ data_param = param_to_dict(data_param) if not task_param: task_param = {mod: (mod,) for mod in list(data_param)} try: if not isinstance(task_param, dict): task_param = vars(task_param) except ValueError: tf.logging.fatal( "To concatenate multiple input data arrays,\n" "task_param should be a dictionary in the form:\n" "{'new_modality_name': ['modality_1', 'modality_2',...]}.") raise if file_list is None: # defaulting to all files detected by the input specification file_list = ImageSetsPartitioner().initialise(data_param).all_files if not self.names: # defaulting to load all sections defined in the task_param self.names = list(task_param) valid_names = [name for name in self.names if task_param.get(name, None)] if not valid_names: tf.logging.fatal("Reader requires task input keywords %s, but " "not exist in the config file.\n" "Available task keywords: %s", self.names, list(task_param)) raise ValueError self.names = valid_names self._input_sources = dict((name, task_param.get(name)) for name in self.names) required_sections = \ sum([list(task_param.get(name)) for name in self.names], []) for required in required_sections: try: if (file_list is None) or \ (required not in list(file_list)) or \ (file_list[required].isnull().all()): tf.logging.fatal('Reader required input section ' 'name [%s], but in the filename list ' 'the column is empty.', required) raise ValueError except (AttributeError, TypeError, ValueError): tf.logging.fatal( 'file_list parameter should be a ' 'pandas.DataFrame instance and has input ' 'section name [%s] as a column name.', required) if required_sections: tf.logging.fatal('Reader requires section(s): %s', required_sections) if file_list is not None: tf.logging.fatal('Configuration input sections are: %s', list(file_list)) raise self.output_list, self._file_list = _filename_to_image_list( file_list, self._input_sources, data_param) for name in self.names: 'Image reader: loading %d subjects ' 'from sections %s as input [%s]', len(self.output_list), self.input_sources[name], name) return self
[docs] def prepare_preprocessors(self): """ Some preprocessors requires an initial step to initialise data dependent internal parameters. This function find these preprocessors and run the initialisations. """ for layer in self.preprocessors: if isinstance(layer, DataDependentLayer): layer.train(self.output_list)
[docs] def add_preprocessing_layers(self, layers): """ Adding a ``niftynet.layer`` or a list of layers as preprocessing steps. """ assert self.output_list is not None, \ 'Please initialise the reader first, ' \ 'before adding preprocessors.' if isinstance(layers, Layer): self.preprocessors.append(layers) else: self.preprocessors.extend(layers) self.prepare_preprocessors()
# pylint: disable=arguments-differ,too-many-branches
[docs] def layer_op(self, idx=None, shuffle=True): """ this layer returns dictionaries:: keys: self.output_fields values: image volume array """ if idx is None: if shuffle: # training, with random list output idx = np.random.randint(len(self.output_list)) else: # testing, with sequential output # accessing self.current_id, not suitable for multi-thread idx = self.current_id + 1 self.current_id = idx try: image_dict = self.output_list[idx] except (IndexError, TypeError): return -1, None, None image_data_dict = \ {field: image.get_data() for (field, image) in image_dict.items()} interp_order_dict = \ {field: image.interp_order for ( field, image) in image_dict.items()} preprocessors = [deepcopy(layer) for layer in self.preprocessors] # dictionary of masks is cached mask = None for layer in preprocessors: # import time; local_time = time.time() if layer is None: continue if isinstance(layer, RandomisedLayer): if "random_elastic_deformation" not in layer.randomise() else: layer.randomise(image_data_dict) image_data_dict = layer(image_data_dict, interp_order_dict) elif isinstance(layer, Layer): image_data_dict, mask = layer(image_data_dict, mask) # print('%s, %.3f sec'%(layer, -local_time + time.time())) return idx, image_data_dict, interp_order_dict
@property def spatial_ranks(self): """ Number of spatial dimensions of the images. :return: integers of spatial rank """ if not self.output_list: tf.logging.fatal("Please initialise the reader first.") raise RuntimeError if not self._spatial_ranks: first_image = self.output_list[0] self._spatial_ranks = {field: first_image[field].spatial_rank for field in self.names} return self._spatial_ranks @property def shapes(self): """ Image shapes before any preprocessing. :return: tuple of integers as image shape .. caution:: To have fast access, the spatial dimensions are not accurate 1. only read from the first image in list 2. not considering effects of random augmentation layers but time and modality dimensions should be correct """ if not self.output_list: tf.logging.fatal("Please initialise the reader first.") raise RuntimeError if not self._shapes: first_image = self.output_list[0] self._shapes = {field: first_image[field].shape for field in self.names} return self._shapes @property def tf_dtypes(self): """ Infer input data dtypes in TF (using the first image in the file list). """ if not self.output_list: tf.logging.fatal("Please initialise the reader first.") raise RuntimeError if not self._dtypes: first_image = self.output_list[0] self._dtypes = {field: infer_tf_dtypes(first_image[field]) for field in self.names} return self._dtypes @property def input_sources(self): """ returns mapping of input keywords and input sections e.g., input_sources:: {'image': ('T1', 'T2'), 'label': ('manual_map',)} map task parameter keywords ``image`` and ``label`` to section names ``T1``, ``T2``, and ``manual_map`` respectively. """ if not self._input_sources: tf.logging.fatal("Please initialise the reader first.") raise RuntimeError return self._input_sources @property def names(self): """ :return: the keys of ``self.input_sources`` dictionary """ return self._names @names.setter def names(self, fields_tuple): """ output_fields is a sequence of output names each name might correspond to a list of multiple input sources this should be specified in CUSTOM section in the config """ self._names = make_input_tuple(fields_tuple, string_types) @property def num_subjects(self): """ :return: number of subjects in the reader """ if not self.output_list: return 0 return len(self.output_list)
[docs] def get_subject_id(self, image_index): """ Given an integer id returns the subject id. """ try: return self._file_list.iloc[image_index][COLUMN_UNIQ_ID] except KeyError: tf.logging.warning('Unknown subject id in reader file list.') raise
[docs] def get_image_index(self, subject_id): """ Given a subject id, return the file_list index :param subject_id: a string with the subject id :return: an int with the file list index """ return np.flatnonzero(self._file_list['subject_id'] == subject_id)[0]
[docs] def get_subject(self, image_index=None): """ Given an integer id returns the corresponding row of the file list. returns: a dictionary of the row """ try: if image_index is None: return self._file_list.iloc[:].to_dict() return self._file_list.iloc[image_index].to_dict() except (KeyError, AttributeError): tf.logging.warning('Unknown subject id in reader file list.') raise
def _filename_to_image_list(file_list, mod_dict, data_param): """ Converting a list of filenames to a list of image objects, Properties (e.g. interp_order) are added to each object """ volume_list = [] valid_idx = [] for idx in range(len(file_list)): # create image instance for each subject print_progress_bar(idx, len(file_list), prefix='reading datasets headers', decimals=1, length=10, fill='*') # combine fieldnames and volumes as a dictionary _dict = {} for field, modalities in mod_dict.items(): _dict[field] = _create_image( file_list, idx, modalities, data_param) # skipping the subject if there're missing image components if _dict and None not in list(_dict.values()): volume_list.append(_dict) valid_idx.append(idx) if not volume_list: tf.logging.fatal( "Empty filename lists, please check the csv " "files. (removing csv_file keyword if it is" " in the config file " "to automatically search folders and generate new csv " "files again)\n\n" "Please note in the matched file names, each subject id are " "created by removing all keywords listed `filename_contains` " "in the config.\n\n" "E.g., `filename_contains=foo, bar` will match file " "foo_subject42_bar.nii.gz, and the subject id is _subject42_.") raise IOError return volume_list, file_list.iloc[valid_idx] def _create_image(file_list, idx, modalities, data_param): """ data_param consists of description of each modality This function combines modalities according to the 'modalities' parameter and create <*D> """ try: file_path = tuple(file_list.iloc[idx][mod] for mod in modalities) any_missing = any([pandas.isnull(file_name) or not bool(file_name) for file_name in file_path]) if any_missing: # todo: enable missing modalities again # the file_path of a multimodal image will contain `nan`, e.g. # this should be handled by `ImageFactory.create_instance` # ('testT1.nii.gz', 'testT2.nii.gz', nan, 'testFlair.nii.gz') return None interp_order, pixdim, axcodes, loader = [], [], [], [] for mod in modalities: mod_spec = data_param[mod] \ if isinstance(data_param[mod], dict) else vars(data_param[mod]) interp_order.append(mod_spec.get('interp_order', DEFAULT_INTERP_ORDER)) pixdim.append(mod_spec.get('pixdim', None)) axcodes.append(mod_spec.get('axcodes', None)) loader.append(mod_spec.get('loader', None)) except KeyError: tf.logging.fatal( "Specified modality names %s " "not found in config: input sections %s.", modalities, list(data_param)) raise except AttributeError: tf.logging.fatal( "Data params must contain: interp_order, pixdim, axcodes.\n" "Reader must be initialised with a dataframe as file_list.") raise image_properties = {'file_path': file_path, 'name': modalities, 'interp_order': interp_order, 'output_pixdim': pixdim, 'output_axcodes': axcodes, 'loader': loader} return ImageFactory.create_instance(**image_properties)
[docs]def param_to_dict(input_data_param): """ Validate the user input ``input_data_param`` raise an error if it's invalid. :param input_data_param: :return: input data specifications as a nested dictionary """ error_msg = 'Unknown ``data_param`` type. ' \ 'It should be a nested dictionary: '\ '{"modality_name": {"input_property": value}} '\ 'or a dictionary of: {"modality_name": '\ 'niftynet.utilities.util_common.ParserNamespace}' data_param = deepcopy(input_data_param) if isinstance(data_param, (ParserNamespace, argparse.Namespace)): data_param = vars(data_param) if not isinstance(data_param, dict): raise ValueError(error_msg) for mod in data_param: mod_param = data_param[mod] if isinstance(mod_param, (ParserNamespace, argparse.Namespace)): dict_param = vars(mod_param) elif isinstance(mod_param, dict): dict_param = mod_param else: raise ValueError(error_msg) for data_key in dict_param: look_up_operations(data_key, SUPPORTED_DATA_SPEC) data_param[mod] = dict_param return data_param