fedbiomed.common.data

@abstractmethod
def apply(self, *args, **kwargs):
    """Abstract method representing an application of the DataLoadingBlock
    """
    pass

def deserialize(self, load_from: dict) -> TDataLoadingBlock:
    """Reconstruct the DataLoadingBlock from a serialized version.

    Args:
        load_from (dict): a dictionary as obtained by the serialize function.
    Returns:
        the self instance
    """
    self._serialization_validator.validate(load_from, FedbiomedLoadingBlockValueError)
    self.__serialization_id = load_from['dlb_id']
    return self

def get_serialization_id(self):
    """Expose serialization id as read-only"""
    return self.__serialization_id

@staticmethod
def instantiate_class(loading_block: dict) -> TDataLoadingBlock:
    """Instantiate one [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock]
    object of the type defined in the arguments.

    Uses the `loading_block_module` and `loading_block_class` fields of the loading_block argument to
    identify the type of [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock]
    to be instantiated, then calls its default constructor.
    Note that this function **does not call deserialize**.

    Args:
        loading_block (dict): [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock]
            metadata in the format returned by the serialize function.
    Returns:
        A default-constructed instance of a
            [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock]
            of the type defined in the metadata.
    Raises:
       FedbiomedLoadingBlockError: if the instantiation process raised any exception.
    """
    try:
        dlb_module = import_module(loading_block['loading_block_module'])
        dlb = eval(f"dlb_module.{loading_block['loading_block_class']}()")
    except Exception as e:
        msg = f"{ErrorNumbers.FB614.value}: could not instantiate DataLoadingBlock from the following metadata: " +\
              f"{loading_block} because of {type(e).__name__}: {e}"
        logger.debug(msg)
        raise FedbiomedLoadingBlockError(msg)
    return dlb

@staticmethod
def instantiate_key(key_module: str, key_classname: str, loading_block_key_str: str) -> DataLoadingBlockTypes:
    """Imports and loads [DataLoadingBlockTypes][fedbiomed.common.constants.DataLoadingBlockTypes]
    regarding the passed arguments

    Args:
        key_module (str): _description_
        key_classname (str): _description_
        loading_block_key_str (str): _description_

    Raises:
        FedbiomedDataLoadingPlanError: _description_

    Returns:
        DataLoadingBlockTypes: _description_
    """
    try:
        keys = import_module(key_module)
        loading_block_key = eval(f"keys.{key_classname}('{loading_block_key_str}')")
    except Exception as e:
        msg = f"{ErrorNumbers.FB615.value} Error deserializing loading block key " + \
              f"{loading_block_key_str} with path {key_module}.{key_classname} " + \
              f"because of {type(e).__name__}: {e}"
        logger.debug(msg)
        raise FedbiomedDataLoadingPlanError(msg)
    return loading_block_key

def serialize(self) -> dict:
    """Serializes the class in a format similar to json.

    Returns:
        a dictionary of key-value pairs sufficient for reconstructing
        the DataLoadingBlock.
    """
    return dict(
        loading_block_class=self.__class__.__qualname__,
        loading_block_module=self.__module__,
        dlb_id=self.__serialization_id
    )

def deserialize(self, serialized_dlp: dict, serialized_loading_blocks: List[dict]) -> TDataLoadingPlan:
    """Reconstruct the DataLoadingPlan][fedbiomed.common.data._data_loading_plan.DataLoadingPlan] from a serialized version.

    !!! warning "Calling this function will *clear* the contained [DataLoadingBlockTypes]."
        This function may not be used to "update" nor to "append to"
        a [DataLoadingPlan][fedbiomed.common.data._data_loading_plan.DataLoadingPlan].

    Args:
        serialized_dlp: a dictionary of data loading plan metadata, as obtained from the first output of the
            serialize function
        serialized_loading_blocks: a list of dictionaries of loading_block metadata, as obtained from the
            second output of the serialize function
    Returns:
        the self instance
    """
    self._serialization_validation.validate(serialized_dlp, FedbiomedDataLoadingPlanValueError)

    self.clear()
    self.dlp_id = serialized_dlp['dlp_id']
    self.desc = serialized_dlp['dlp_name']
    self.target_dataset_type = DatasetTypes(serialized_dlp['target_dataset_type'])
    for loading_block_key_str, dlb_id in serialized_dlp['loading_blocks'].items():
        key_module, key_classname = serialized_dlp['key_paths'][loading_block_key_str]
        loading_block_key = DataLoadingBlock.instantiate_key(key_module, key_classname, loading_block_key_str)
        loading_block = next(filter(lambda x: x['dlb_id'] == dlb_id,
                                    serialized_loading_blocks))
        dlb = DataLoadingBlock.instantiate_class(loading_block)
        self[loading_block_key] = dlb.deserialize(loading_block)
    return self

@staticmethod
def infer_dataset_type(dataset: Any) -> DatasetTypes:
    """Infer the type of a given dataset.

    This function provides the mapping between a dataset's class and the DatasetTypes enum. If the dataset exposes
    the correct interface (i.e. the get_dataset_type method) then it directly calls that, otherwise it tries to
    apply some heuristics to guess the type of dataset.

    Args:
        dataset: the dataset whose type we want to infer.
    Returns:
        a DatasetTypes enum element which identifies the type of the dataset.
    Raises:
        FedbiomedDataLoadingPlanValueError: if the dataset does not have a `get_dataset_type` method and moreover
            the type could not be guessed.
    """
    if hasattr(dataset, 'get_dataset_type'):
        return dataset.get_dataset_type()
    elif dataset.__class__.__name__ == 'ImageFolder':
        # ImageFolder could be both an images type or mednist. Try to identify mednist with some heuristic.
        if hasattr(dataset, 'classes') and \
                all([x in dataset.classes for x in ['AbdomenCT', 'BreastMRI', 'CXR', 'ChestCT', 'Hand', 'HeadCT']]):
            return DatasetTypes.MEDNIST
        else:
            return DatasetTypes.IMAGES
    elif dataset.__class__.__name__ == 'MNIST':
        return DatasetTypes.DEFAULT
    msg = f"{ErrorNumbers.FB615.value} Trying to infer dataset type of {dataset} is not supported " + \
        f"for datasets of type {dataset.__class__.__qualname__}"
    logger.debug(msg)
    raise FedbiomedDataLoadingPlanValueError(msg)

def serialize(self) -> Tuple[dict, List]:
    """Serializes the class in a format similar to json.

    Returns:
        a tuple sufficient for reconstructing the DataLoading plan. It includes:
            - a dictionary of key-value pairs with the
            [DataLoadingPlan][fedbiomed.common.data._data_loading_plan.DataLoadingPlan] parameters.
            - a list of dict containing the data for reconstruction all the DataLoadingBlock
                of the [DataLoadingPlan][fedbiomed.common.data._data_loading_plan.DataLoadingPlan] 
    """
    return dict(
        dlp_id=self.dlp_id,
        dlp_name=self.desc,
        target_dataset_type=self.target_dataset_type.value,
        loading_blocks={key.value: dlb.get_serialization_id() for key, dlb in self.items()},
        key_paths={key.value: (f"{key.__module__}", f"{key.__class__.__qualname__}") for key in self.keys()}
    ), [dlb.serialize() for dlb in self.values()]

def apply_dlb(self, default_ret_value: Any, dlb_key: DataLoadingBlockTypes,
              *args: Optional[Any], **kwargs: Optional[Any]) -> Any:
    """Apply one DataLoadingBlock identified by its key.

    Note that we want to easily support the case where the DataLoadingPlan
    is not activated, or the requested loading block is not contained in the
    DataLoadingPlan. This is achieved by providing a default return value
    to be returned when the above conditions are met. Hence, most of the
    calls to apply_dlb will look like this:
    ```
    value = self.apply_dlb(value, 'my-loading-block', my_apply_args)
    ```
    This will ensure that value is not changed if the DataLoadingPlan is
    not active.

    Args:
        default_ret_value: the value to be returned in case that the dlp
            functionality is not required
        dlb_key: the key of the DataLoadingBlock to be applied
        *args: forwarded to the DataLoadingBlock's apply function
        **kwargs: forwarded to the DataLoadingBlock's apply function
    Returns:
        the output of the DataLoadingBlock's apply function, or
            the default_ret_value when dlp is None or it does not contain
            the requested loading block
    """
    if not isinstance(dlb_key, DataLoadingBlockTypes):
        raise FedbiomedDataLoadingPlanValueError(f"Key {dlb_key} is not of enum type DataLoadingBlockTypes"
                                                 f" in DataLoadingPlanMixin.apply_dlb")
    if self._dlp is not None and dlb_key in self._dlp:
        return self._dlp[dlb_key].apply(*args, **kwargs)
    else:
        return default_ret_value

def clear_dlp(self):
    self._dlp = None

def set_dlp(self, dlp: DataLoadingPlan):
    """Sets the dlp if the target dataset type is appropriate"""
    if not isinstance(dlp, DataLoadingPlan):
        msg = f"{ErrorNumbers.FB615.value} Trying to set a DataLoadingPlan but the argument is of type " + \
              f"{type(dlp).__name__}"
        logger.debug(msg)
        raise FedbiomedDataLoadingPlanValueError(msg)

    dataset_type = DataLoadingPlan.infer_dataset_type(self)  # `self` here will refer to the Dataset instance
    if dlp.target_dataset_type != DatasetTypes.NONE and dataset_type != dlp.target_dataset_type:
        raise FedbiomedDataLoadingPlanValueError(f"Trying to set {dlp} on dataset of type {dataset_type.value} but "
                                                 f"the target type is {dlp.target_dataset_type}")
    elif dlp.target_dataset_type == DatasetTypes.NONE:
        dlp.target_dataset_type = dataset_type
    self._dlp = dlp

def load(self, tp_type: TrainingPlans):
    """Loads proper DataManager based on given TrainingPlan and
    `dataset`, `target` attributes.

    Args:
        tp_type: Enumeration instance of TrainingPlans that stands for type of training plan.

    Raises:
        FedbiomedDataManagerError: If requested DataManager does not match with given arguments.

    """

    # Training plan is type of TorcTrainingPlan
    if tp_type == TrainingPlans.TorchTrainingPlan:
        if self._target is None and isinstance(self._dataset, Dataset):
            # Create Dataset for pytorch
            self._data_manager_instance = TorchDataManager(dataset=self._dataset, **self._loader_arguments)
        elif isinstance(self._dataset, (pd.DataFrame, pd.Series, np.ndarray)) and \
                isinstance(self._target, (pd.DataFrame, pd.Series, np.ndarray)):
            # If `dataset` and `target` attributes are array-like object
            # create TabularDataset object to instantiate a TorchDataManager
            torch_dataset = TabularDataset(inputs=self._dataset, target=self._target)
            self._data_manager_instance = TorchDataManager(dataset=torch_dataset, **self._loader_arguments)
        else:
            raise FedbiomedDataManagerError(f"{ErrorNumbers.FB607.value}: Invalid arguments for torch based "
                                            f"training plan, either provide the argument  `dataset` as PyTorch "
                                            f"Dataset instance, or provide `dataset` and `target` arguments as "
                                            f"an instance one of pd.DataFrame, pd.Series or np.ndarray ")

    elif tp_type == TrainingPlans.SkLearnTrainingPlan:
        # Try to convert `torch.utils.Data.Dataset` to SkLearnBased dataset/datamanager
        if self._target is None and isinstance(self._dataset, Dataset):
            torch_data_manager = TorchDataManager(dataset=self._dataset)
            try:
                self._data_manager_instance = torch_data_manager.to_sklearn()
            except Exception as e:
                raise FedbiomedDataManagerError(f"{ErrorNumbers.FB607.value}: PyTorch based `Dataset` object "
                                                "has been instantiated with DataManager. An error occurred while"
                                                "trying to convert torch.utils.data.Dataset to numpy based "
                                                f"dataset: {str(e)}")

        # For scikit-learn based training plans, the arguments `dataset` and `target` should be an instance
        # one of `pd.DataFrame`, `pd.Series`, `np.ndarray`
        elif isinstance(self._dataset, (pd.DataFrame, pd.Series, np.ndarray)) and \
                isinstance(self._target, (pd.DataFrame, pd.Series, np.ndarray)):
            # Create Dataset for SkLearn training plans
            self._data_manager_instance = SkLearnDataManager(inputs=self._dataset, target=self._target,
                                                             **self._loader_arguments)
        else:
            raise FedbiomedDataManagerError(f"{ErrorNumbers.FB607.value}: The argument `dataset` and `target` "
                                            f"should be instance of pd.DataFrame, pd.Series or np.ndarray ")
    else:
        raise FedbiomedDataManagerError(f"{ErrorNumbers.FB607.value}: Undefined training plan")

def clear_dlp(self):
    """Clears dlp and automatically clears the FedClass

    Tries to guarantee some semblance of integrity by also clearing the FedClass, since setting the dlp
    initializes it.
    """
    super().clear_dlp()
    self._clear()

@_check_fed_class_initialization_status(require_initialized=True,
                                        require_uninitialized=False,
                                        message="Flamby dataset is in an inconsistent state: a Data Loading Plan "
                                                "is set but the wrapped FedClass was not initialized.")
@_requires_dlp
def get_center_id(self) -> int:
    """Returns the center id. Requires that the DataLoadingPlan has already been set.

    Returns:
        the center id (int).
    Raises:
        FedbiomedDatasetError: in one of the two scenarios below
            - if the data loading plan is not set or is malformed.
            - if the wrapped FedClass is not initialized but the dlp exists
    """
    return self.apply_dlb(None, FlambyLoadingBlockTypes.FLAMBY_DATASET_METADATA)['flamby_center_id']

@staticmethod
def get_dataset_type() -> DatasetTypes:
    """Returns the Flamby DatasetType"""
    return DatasetTypes.FLAMBY

def get_flamby_fed_class(self):
    """Returns the instance of the wrapped Flamby FedClass"""
    return self.__flamby_fed_class

def get_transform(self):
    """Gets the transform attribute"""
    return self._transform

@_check_fed_class_initialization_status(require_initialized=False,
                                        require_uninitialized=True,
                                        message="Calling init_transform is not allowed if the wrapped FedClass "
                                                "has already been initialized. At your own risk, you may call "
                                                "clear_dlp to reset the full FlambyDataset")
def init_transform(self, transform: Union[MonaiCompose, TorchCompose]) -> Union[MonaiCompose, TorchCompose]:
    """Initializes the transform attribute. Must be called before initialization of the wrapped FedClass.

    Arguments:
        transform: a composed transform of type torchvision.transforms.Compose or monai.transforms.Compose

    Raises:
        FedbiomedDatasetError: if the wrapped FedClass was already initialized.
        FedbiomedDatasetValueError: if the input is not of the correct type.
    """
    if not isinstance(transform, (MonaiCompose, TorchCompose)):
        msg = f"{ErrorNumbers.FB618.value}. FlambyDataset transform must be of type " \
              f"torchvision.transforms.Compose or monai.transforms.Compose"
        logger.critical(msg)
        raise FedbiomedDatasetValueError(msg)

    self._transform = transform
    return self._transform

def set_dlp(self, dlp):
    """Sets the Data Loading Plan and ensures that the flamby_fed_class is initialized

    Overrides the set_dlp function from the DataLoadingPlanMixin to make sure that self._init_flamby_fed_class
    is also called immediately after.
    """
    super().set_dlp(dlp)
    try:
        self._init_flamby_fed_class()
    except FedbiomedDatasetError as e:
        # clean up
        super().clear_dlp()
        raise FedbiomedDatasetError from e

@_check_fed_class_initialization_status(require_initialized=True,
                                        require_uninitialized=False,
                                        message="Cannot compute shape because FedClass was not initialized.")
def shape(self) -> List[int]:
    """Returns the shape of the flamby_fed_class"""
    return [len(self)] + list(self.__getitem__(0)[0].shape)

def apply(self) -> dict:
    """Returns a dictionary of dataset metadata.

    The metadata dictionary contains:
    - flamby_dataset_name: (str) the name of the selected flamby dataset.
    - flamby_center_id: (int) the center id selected at dataset add time.

    Note that the flamby_dataset_name will be the same as the module name required to instantiate the FedClass.
    However, it will not contain the full module path, hence to properly import this module it must be
    prepended with `flamby.datasets`, for example `import flamby.datasets.flamby_dataset_name`

    Returns:
        this data loading block's metadata
    """
    if any([v is None for v in self.metadata.values()]):
        msg = f"{ErrorNumbers.FB316}. Attempting to read Flamby dataset metadata, but " \
              f"the {[k for k,v in self.metadata.items() if v is None]} keys were not previously set."
        logger.critical(msg)
        raise FedbiomedLoadingBlockError(msg)
    return self.metadata

def deserialize(self, load_from: dict) -> DataLoadingBlock:
    """Reconstruct the DataLoadingBlock from a serialized version.

    Args:
        load_from: a dictionary as obtained by the serialize function.
    Returns:
        the self instance
    """
    super().deserialize(load_from)
    self.metadata['flamby_dataset_name'] = load_from['flamby_dataset_name']
    self.metadata['flamby_center_id'] = load_from['flamby_center_id']
    return self

def serialize(self) -> dict:
    """Serializes the class in a format similar to json.

    Returns:
         a dictionary of key-value pairs sufficient for reconstructing
         the DataLoadingBlock.
    """
    ret = super().serialize()
    ret.update({'flamby_dataset_name': self.metadata['flamby_dataset_name'],
                'flamby_center_id': self.metadata['flamby_center_id']
                })
    return ret

def apply(self, key):
    """Returns the value mapped to the key, if it exists.

    Raises:
        FedbiomedLoadingBlockError: if map is not a dict or the key does not exist.
    """
    if not isinstance(self.map, dict) or key not in self.map:
        msg = f"{ErrorNumbers.FB614.value} Mapper block error: no key '{key}' in mapping dictionary"
        logger.debug(msg)
        raise FedbiomedLoadingBlockError(msg)
    return self.map[key]

def deserialize(self, load_from: dict) -> DataLoadingBlock:
    """Reconstruct the [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock]
    from a serialized version.

    Args:
        load_from (dict): a dictionary as obtained by the serialize function.
    Returns:
        the self instance
    """
    super(MapperBlock, self).deserialize(load_from)
    self.map = load_from['map']
    return self

def serialize(self) -> dict:
    """Serializes the class in a format similar to json.

    Returns:
        a dictionary of key-value pairs sufficient for reconstructing
        the [DataLoadingBlock][fedbiomed.common.data._data_loading_plan.DataLoadingBlock].
    """
    ret = super(MapperBlock, self).serialize()
    ret.update({'map': self.map})
    return ret

def available_subjects(self,
                       subjects_from_index: Union[list, pd.Series],
                       subjects_from_folder: list = None) -> tuple[list[str], list[str], list[str]]:
    """Checks missing subject folders and missing entries in demographics

    Args:
        subjects_from_index: Given subject folder names in demographics
        subjects_from_folder: List of subject folder names to get intersection of given subject_from_index

    Returns:
        available_subjects: subjects that have an imaging data folder and are also present in the demographics file
        missing_subject_folders: subjects that are in the demographics file but do not have an imaging data folder
        missing_entries: subjects that have an imaging data folder but are not present in the demographics file
    """

    # Select all subject folders if it is not given
    if subjects_from_folder is None:
        subjects_from_folder = self.subjects_with_imaging_data_folders()

    # Missing subject that will cause warnings
    missing_subject_folders = list(set(subjects_from_index) - set(subjects_from_folder))

    # Missing entries that will cause errors
    missing_entries = list(set(subjects_from_folder) - set(subjects_from_index))

    # Intersection
    available_subjects = list(set(subjects_from_index).intersection(set(subjects_from_folder)))

    return available_subjects, missing_subject_folders, missing_entries

def complete_subjects(self, subjects: List[str], modalities: List[str]) -> List[str]:
    """Retrieves subjects that have given all the modalities.

    Args:
        subjects: List of subject folder names
        modalities: List of required modalities

    Returns:
        List of subject folder names that have required modalities
    """
    return [subject for subject in subjects if all(self.is_modalities_existing(subject, modalities))]

@staticmethod
def demographics_column_names(path: Union[str, Path]):
    return MedicalFolderBase.read_demographics(path).columns.values

@staticmethod
def get_dataset_type() -> DatasetTypes:
    return DatasetTypes.MEDICAL_FOLDER

def is_modalities_existing(self, subject: str, modalities: List[str]) -> List[bool]:
    """Checks whether given modalities exists in the subject directory

    Args:
        subject: Subject ID or subject folder name
        modalities: List of modalities to check

    Returns:
        List of `bool` that represents whether modality is existing respectively for each of modality.

    Raises:
        FedbiomedDatasetError: bad argument type
    """
    if not isinstance(subject, str):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Expected string for subject folder/ID, "
                                    f"but got {type(subject)}")
    if not isinstance(modalities, list):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Expected a list for modalities, "
                                    f"but got {type(modalities)}")
    if not all([type(m) is str for m in modalities]):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Expected a list of string for modalities, "
                                    f"but some modalities are "
                                    f"{' '.join([ str(type(m) for m in modalities if type(m) != str)])}")
    are_modalities_existing = list()
    for modality in modalities:
        modality_folder = self._subject_modality_folder(subject, modality)
        are_modalities_existing.append(bool(modality_folder) and
                                       self._root.joinpath(subject, modality_folder).is_dir())
    return are_modalities_existing

def modalities(self) -> Tuple[list, list]:
    """Gets all modalities based either on all possible candidates or those provided by the DataLoadingPlan.

    Returns:
         List of unique available modalities
         List of all encountered modality folders in each subject folder, appearing once per folder
    """
    modality_candidates, modality_folders_list = self.modalities_candidates_from_subfolders()
    if self._dlp is not None and MedicalFolderLoadingBlockTypes.MODALITIES_TO_FOLDERS in self._dlp:
        modalities = list(self._dlp[MedicalFolderLoadingBlockTypes.MODALITIES_TO_FOLDERS].map.keys())
        return modalities, modality_folders_list
    else:
        return modality_candidates, modality_folders_list

def modalities_candidates_from_subfolders(self) -> Tuple[list, list]:
    """ Gets all possible modality folders under root directory

    Returns:
         List of unique available modality folders appearing at least once
         List of all encountered modality folders in each subject folder, appearing once per folder
    """

    # Accept only folders that don't start with "." and "_"
    modalities = [f.name for f in self._root.glob("*/*") if f.is_dir() and not f.name.startswith((".", "_"))]
    return sorted(list(set(modalities))), modalities

@staticmethod
def read_demographics(path: Union[str, Path], index_col: Optional[int] = None):
    """ Read demographics tabular file for Medical Folder dataset

    Raises:
        FedbiomedDatasetError: bad file format
    """
    path = Path(path)
    if not path.is_file() or path.suffix.lower() not in [".csv", ".tsv"]:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Demographics should be CSV or TSV files")

    return pd.read_csv(path, index_col=index_col, engine='python')

@property
def root(self):
    """Root property of MedicalFolderController"""
    return self._root

def subjects_with_imaging_data_folders(self) -> List[str]:
    """Retrieves subject folder names under Medical Folder root directory.

    Returns:
        subject folder names under Medical Folder root directory.
    """
    return [f.name for f in self._root.iterdir() if f.is_dir() and not f.name.startswith(".")]

@staticmethod
def validate_MedicalFolder_root_folder(path: Union[str, Path]) -> Path:
    """ Validates Medical Folder root directory by checking folder structure

    Args:
        path: path to root directory

    Returns:
        Path to root folder of Medical Folder dataset

    Raises:
        FedbiomedDatasetError: - If path is not an instance of `str` or `pathlib.Path`
                               - If path is not a directory
    """
    if not isinstance(path, (Path, str)):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: The argument root should an instance of "
                                    f"`Path` or `str`, but got {type(path)}")

    if not isinstance(path, Path):
        path = Path(path)

    path = Path(path).expanduser().resolve()

    if not path.exists():
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Folder or file {path} not found on system")
    if not path.is_dir():
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Root for Medical Folder dataset "
                                    f"should be a directory.")

    directories = [f for f in path.iterdir() if f.is_dir()]
    if len(directories) == 0:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Root folder of Medical Folder should "
                                    f"contain subject folders, but no sub folder has been found. ")

    modalities = [f for f in path.glob("*/*") if f.is_dir()]
    if len(modalities) == 0:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value} Subject folders for Medical Folder should "
                                    f"contain modalities as folders. Folder structure should be "
                                    f"root/<subjects>/<modalities>")

    return path

def load_MedicalFolder(self,
                       tabular_file: Union[str, Path] = None,
                       index_col: Union[str, int] = None) -> MedicalFolderDataset:
    """ Load Medical Folder dataset with given tabular_file and index_col

    Args:
        tabular_file: File path to demographics data set
        index_col: Column index that represents subject folder names

    Returns:
        MedicalFolderDataset object

    Raises:
        FedbiomedDatasetError: If Medical Folder dataset is not successfully loaded
    """
    if self._root is None:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Can not load Medical Folder dataset without "
                                    f"declaring root directory. Please set root or build MedicalFolderController "
                                    f"with by providing `root` argument use")

    modalities, _ = self.modalities()

    try:
        dataset = MedicalFolderDataset(root=self._root,
                                       tabular_file=tabular_file,
                                       index_col=index_col,
                                       data_modalities=modalities,
                                       target_modalities=modalities)
    except FedbiomedError as e:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Can not create Medical Folder dataset. {e}")

    if self._dlp is not None:
        dataset.set_dlp(self._dlp)
    return dataset

def subject_modality_status(self, index: Union[List, pd.Series] = None) -> Dict:
    """Scans subjects and checks which modalities are existing for each subject

    Args:
        index: Array-like index that comes from reference csv file of Medical Folder dataset. It represents subject
            folder names. Defaults to None.
    Returns:
        Modality status for each subject that indicates which modalities are available
    """

    modalities, _ = self.modalities()
    subjects = self.subjects_with_imaging_data_folders()
    modality_status = {"columns": [*modalities], "data": [], "index": []}

    if index is not None:
        _, missing_subjects, missing_entries = self.available_subjects(subjects_from_index=index)
        modality_status["columns"].extend(["in_folder", "in_index"])

    for subject in subjects:
        modality_report = self.is_modalities_existing(subject, modalities)
        status_list = [status for status in modality_report]
        if index is not None:
            status_list.append(False if subject in missing_subjects else True)
            status_list.append(False if subject in missing_entries else True)

        modality_status["data"].append(status_list)
        modality_status["index"].append(subject)

    return modality_status

@property
@cache
def demographics(self) -> pd.DataFrame:
    """Loads tabular data file (supports excel, csv, tsv and colon separated value files)."""

    if self._tabular_file is None or self._index_col is None:
        # If there is no tabular file return empty data frame
        return None

    # Read demographics CSV
    try:
        demographics = self.read_demographics(self._tabular_file, self._index_col)
    except Exception as e:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Can not load demographics tabular file. "
                                    f"Error message is: {e}")

    # Keep the first one in duplicated subjects
    return demographics.loc[~demographics.index.duplicated(keep="first")]

def get_nontransformed_item(self, item):
    # For the first item retrieve complete subject folders
    subjects = self.subject_folders()

    if not subjects:
        # case where subjects is an empty list (subject folders have not been found)
        raise FedbiomedDatasetError(
            f"{ErrorNumbers.FB613.value}: Cannot find complete subject folders with all the modalities")
    # Get subject folder
    subject_folder = subjects[item]

    # Load data modalities
    data = self.load_images(subject_folder, modalities=self._data_modalities)

    # Load target modalities
    targets = self.load_images(subject_folder, modalities=self._target_modalities)

    # Demographics
    demographics = self._get_from_demographics(subject_id=subject_folder.name)
    return (data, demographics), targets

@property
def index_col(self):
    """Getter/setter of the column containing folder's name (in the tabular file)"""
    return self._index_col

def load_images(self, subject_folder: Path, modalities: list) -> Dict[str, torch.Tensor]:
    """Loads modality images in given subject folder

    Args:
        subject_folder: Subject folder where modalities are stored
        modalities: List of available modalities

    Returns:
        Subject image data as victories where keys represent each modality.
    """
    subject_data = {}

    for modality in modalities:
        modality_folder = self._subject_modality_folder(subject_folder, modality)
        image_folder = subject_folder.joinpath(modality_folder)
        nii_files = [p.resolve() for p in image_folder.glob("**/*")
                     if ''.join(p.suffixes) in self.ALLOWED_EXTENSIONS]

        # Load the first, we assume there is going to be a single image per modality for now.
        img_path = nii_files[0]
        img = self._reader(img_path)
        subject_data[modality] = img

    return subject_data

def set_dataset_parameters(self, parameters: dict):
    """Sets dataset parameters.

    Args:
        parameters: Parameters to initialize

    Raises:
        FedbiomedDatasetError: If given parameters are not of `dict` type
    """
    if not isinstance(parameters, dict):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Expected type for `parameters` is `dict, "
                                    f"but got {type(parameters)}`")

    for key, value in parameters.items():
        if hasattr(self, key):
            setattr(self, key, value)
        else:
            raise FedbiomedDatasetError(f"{ErrorNumbers.FB613.value}: Trying to set non existing attribute '{key}'")

def shape(self) -> dict:
    """Retrieves shape information for modalities and demographics csv"""

    # Get all modalities
    data_modalities = list(set(self._data_modalities))
    target_modalities = list(set(self._target_modalities))
    modalities = list(set(self._data_modalities + self._target_modalities))
    (image, _), targets = self.get_nontransformed_item(0)

    result = {modality: list(image[modality].shape) for modality in data_modalities}

    result.update({modality: list(targets[modality].shape) for modality in target_modalities})
    num_modalities = len(modalities)
    demographics_shape = self.demographics.shape if self.demographics is not None else None
    result.update({"demographics": demographics_shape, "num_modalities": num_modalities})

    return result