Data

def __init__(self):
    self.__serialization_id = 'serialized_dlb_' + str(uuid.uuid4())
    self._serialization_validator = SerializationValidation()
    self._serialization_validator.update_validation_scheme(SerializationValidation.dlb_default_scheme())

def __init__(self, *args, **kwargs):
    super(DataLoadingPlan, self).__init__(*args, **kwargs)
    self.dlp_id = 'dlp_' + str(uuid.uuid4())
    self.desc = ""
    self.target_dataset_type = DatasetTypes.NONE
    self._serialization_validation = SerializationValidation()
    self._serialization_validation.update_validation_scheme(SerializationValidation.dlp_default_scheme())

def __init__(self):
    self._dlp = None

def __init__(self,
             dataset: Union[np.ndarray, pd.DataFrame, pd.Series, Dataset],
             target: Union[np.ndarray, pd.DataFrame, pd.Series] = None,
             **kwargs: dict) -> None:

    """Constructor of DataManager,

    Args:
        dataset: Dataset object. It can be an instance, PyTorch Dataset or Tuple.
        target: Target variable or variables.
        **kwargs: Additional parameters that are going to be used for data loader
    """

    # TODO: Improve datamanager for auto loading by given dataset_path and other information
    # such as inputs variable indexes and target variables indexes

    self._dataset = dataset
    self._target = target
    self._loader_arguments: Dict = kwargs
    self._data_manager_instance = None

def __init__(self):
    super().__init__()
    self.__flamby_fed_class = None
    self._transform = None

def __init__(self):
    super().__init__()
    self.metadata = {
        "flamby_dataset_name": None,
        "flamby_center_id": None
    }
    self._serialization_validator.update_validation_scheme(
        FlambyDatasetMetadataBlock._extra_validation_scheme())

def __init__(self):
    super(MapperBlock, self).__init__()
    self.map = {}
    self._serialization_validator.update_validation_scheme(MapperBlock._extra_validation_scheme())

def __init__(self, root: Union[str, Path, None] = None):
    """Constructs MedicalFolderBase

    Args:
        root: path to Medical Folder root folder.
    """
    super(MedicalFolderBase, self).__init__()

    if root is not None:
        root = self.validate_MedicalFolder_root_folder(root)

    self._root = root

def __init__(self, root: str = None):
    """Constructs MedicalFolderController

    Args:
        root: Folder path to dataset. Defaults to None.
    """
    super(MedicalFolderController, self).__init__(root=root)

def __init__(self,
             root: Union[str, PathLike, Path],
             data_modalities: Optional[Union[str, Iterable[str]]] = 'T1',
             transform: Union[Callable, Dict[str, Callable]] = None,
             target_modalities: Optional[Union[str, Iterable[str]]] = 'label',
             target_transform: Union[Callable, Dict[str, Callable]] = None,
             demographics_transform: Optional[Callable] = None,
             tabular_file: Union[str, PathLike, Path, None] = None,
             index_col: Union[int, str, None] = None,
             ):
    """Constructor for class `MedicalFolderDataset`.

    Args:
        root: Root folder containing all the subject directories.
        data_modalities (str, Iterable): Modality or modalities to be used as data sources.
        transform: A function or dict of function transform(s) that preprocess each data source.
        target_modalities: (str, Iterable): Modality or modalities to be used as target sources.
        target_transform: A function or dict of function transform(s) that preprocess each target source.
        demographics_transform: TODO
        tabular_file: Path to a CSV or Excel file containing the demographic information from the patients.
        index_col: Column name in the tabular file containing the subject ids which mush match the folder names.
    """
    super(MedicalFolderDataset, self).__init__(root=root)

    self._tabular_file = tabular_file
    self._index_col = index_col

    self._data_modalities = [data_modalities] if isinstance(data_modalities, str) else data_modalities
    self._target_modalities = [target_modalities] if isinstance(target_modalities, str) else target_modalities

    self._transform = self._check_and_reformat_transforms(transform, data_modalities)
    self._target_transform = self._check_and_reformat_transforms(target_transform, target_modalities)
    self._demographics_transform = demographics_transform if demographics_transform is not None else lambda x: {}

    # Image loader
    self._reader = Compose([
        LoadImage(ITKReader(), image_only=True),
        ToTensor()
    ])

def __init__(self, root: Union[str, PathLike, Path],
             transform: Union[Callable, None] = None,
             target_transform: Union[Callable, None] = None
             ):
    """Constructor of the class

    Args:
        root: folder where the data is located.
        transform: transforms to be applied on data.
        target_transform: transforms to be applied on target indexes.

    Raises:
        FedbiomedDatasetError: bad argument type
        FedbiomedDatasetError: bad root path
    """
    # check parameters type
    for tr, trname in ((transform, 'transform'), (target_transform, 'target_transform')):
        if not callable(tr) and tr is not None:
            raise FedbiomedDatasetError(f"{ErrorNumbers.FB612.value}: Parameter {trname} has incorrect "
                                        f"type {type(tr)}, cannot create dataset.")

    if not isinstance(root, str) and not isinstance(root, PathLike) and not isinstance(root, Path):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB612.value}: Parameter `root` has incorrect type "
                                    f"{type(root)}, cannot create dataset.")

    # initialize object variables
    self._files = []
    self._class_labels = []
    self._targets = []

    try:
        self._root_dir = Path(root).expanduser()
    except RuntimeError as e:
        raise FedbiomedDatasetError(
            f"{ErrorNumbers.FB612.value}: Cannot expand path {root}, error message is: {e}")

    self._transform = transform
    self._target_transform = target_transform
    self._reader = Compose([
        LoadImage(ITKReader(), image_only=True),
        ToTensor()
    ])

    self._explore_root_folder()

def __init__(self,
             dataset: np.ndarray,
             target: np.ndarray,
             batch_size: int = 1,
             shuffle: bool = False,
             random_seed: Optional[int] = None,
             drop_last: bool = False):
    """Construct numpy data loader

    Args:
        dataset: 2D Numpy array
        target: Numpy array of target values
        batch_size: batch size for each iteration
        shuffle: shuffle before iteration
        random_seed: an optional integer to set the numpy random seed for shuffling. If it equals
            None, then no attempt will be made to set the random seed.
        drop_last: whether to drop the last batch in case it does not fill the whole batch size
    """

    if not isinstance(dataset, np.ndarray) or not isinstance(target, np.ndarray):
        msg = f"{ErrorNumbers.FB609.value}. Wrong input type for `dataset` or `target` in NPDataLoader. " \
              f"Expected type np.ndarray for both, instead got {type(dataset)} and" \
              f"{type(target)} respectively."
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    # If the researcher gave a 1-dimensional dataset, we expand it to 2 dimensions
    if dataset.ndim == 1:
        logger.info(f"NPDataLoader expanding 1-dimensional dataset to become 2-dimensional.")
        dataset = dataset[:, np.newaxis]

    # If the researcher gave a 1-dimensional target, we expand it to 2 dimensions
    if target.ndim == 1:
        logger.info(f"NPDataLoader expanding 1-dimensional target to become 2-dimensional.")
        target = target[:, np.newaxis]

    if dataset.ndim != 2 or target.ndim != 2:
        msg = f"{ErrorNumbers.FB609.value}. Wrong shape for `dataset` or `target` in NPDataLoader. " \
              f"Expected 2-dimensional arrays, instead got {dataset.ndim}-dimensional " \
              f"and {target.ndim}-dimensional arrays respectively."
        logger.error(msg)
        raise FedbiomedValueError(msg)

    if len(dataset) != len(target):
        msg = f"{ErrorNumbers.FB609.value}. Inconsistent length for `dataset` and `target` in NPDataLoader. " \
              f"Expected same length, instead got len(dataset)={len(dataset)}, len(target)={len(target)}"
        logger.error(msg)
        raise FedbiomedValueError(msg)

    if not isinstance(batch_size, int):
        msg = f"{ErrorNumbers.FB609.value}. Wrong type for `batch_size` parameter of NPDataLoader. Expected a " \
              f"non-zero positive integer, instead got type {type(batch_size)}."
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    if batch_size <= 0:
        msg = f"{ErrorNumbers.FB609.value}. Wrong value for `batch_size` parameter of NPDataLoader. Expected a " \
              f"non-zero positive integer, instead got value {batch_size}."
        logger.error(msg)
        raise FedbiomedValueError(msg)

    if not isinstance(shuffle, bool):
        msg = f"{ErrorNumbers.FB609.value}. Wrong type for `shuffle` parameter of NPDataLoader. Expected `bool`, " \
              f"instead got {type(shuffle)}."
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    if not isinstance(drop_last, bool):
        msg = f"{ErrorNumbers.FB609.value}. Wrong type for `drop_last` parameter of NPDataLoader. " \
              f"Expected `bool`, instead got {type(drop_last)}."
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    if random_seed is not None and not isinstance(random_seed, int):
        msg = f"{ErrorNumbers.FB609.value}. Wrong type for `random_seed` parameter of NPDataLoader. " \
              f"Expected int or None, instead got {type(random_seed)}."
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    self._dataset = dataset
    self._target = target
    self._batch_size = batch_size
    self._shuffle = shuffle
    self._drop_last = drop_last
    self._rng = np.random.default_rng(random_seed)

def __init__(self):
    self._validation_scheme = {}

def __init__(self,
             inputs: Union[np.ndarray, pd.DataFrame, pd.Series],
             target: Union[np.ndarray, pd.DataFrame, pd.Series],
             **kwargs: dict):

    """ Construct a SkLearnDataManager from an array of inputs and an array of targets.

    The loader arguments will be passed to the [fedbiomed.common.data.NPDataLoader] classes instantiated
    when split is called. They may include batch_size, shuffle, drop_last, and others. Please see the
    [fedbiomed.common.data.NPDataLoader] class for more details.

    Args:
        inputs: Independent variables (inputs, features) for model training
        target: Dependent variable/s (target) for model training and validation
        **kwargs: Loader arguments
    """

    if not isinstance(inputs, (np.ndarray, pd.DataFrame, pd.Series)) or \
            not isinstance(target, (np.ndarray, pd.DataFrame, pd.Series)):
        msg = f"{ErrorNumbers.FB609.value}. Parameters `inputs` and `target` for " \
              f"initialization of {self.__class__.__name__} should be one of np.ndarray, pd.DataFrame, pd.Series"
        logger.error(msg)
        raise FedbiomedTypeError(msg)

    # Convert pd.DataFrame or pd.Series to np.ndarray for `inputs`
    if isinstance(inputs, (pd.DataFrame, pd.Series)):
        self._inputs = inputs.to_numpy()
    else:
        self._inputs = inputs

    # Convert pd.DataFrame or pd.Series to np.ndarray for `target`
    if isinstance(target, (pd.DataFrame, pd.Series)):
        self._target = target.to_numpy()
    else:
        self._target = target

    # Additional loader arguments
    self._loader_arguments = kwargs

    # Subset None means that train/validation split has not been performed
    self._subset_test: Union[Tuple[np.ndarray, np.ndarray], None] = None
    self._subset_train: Union[Tuple[np.ndarray, np.ndarray], None] = None

def __init__(self,
             inputs: Union[np.ndarray, pd.DataFrame, pd.Series],
             target: Union[np.ndarray, pd.DataFrame, pd.Series]):
    """Constructs PyTorch dataset object

    Args:
        inputs: Input variables that will be passed to network
        target: Target variable for output layer

    Raises:
        FedbiomedTorchDatasetError: If input variables and target variable does not have
            equal length/size
    """

    # Inputs and target variable should be converted to the torch tensors
    # PyTorch provides `from_numpy` function to convert numpy arrays to
    # torch tensor. Therefore, if the arguments `inputs` and `target` are
    # instance one of `pd.DataFrame` or `pd.Series`, they should be converted to
    # numpy arrays
    if isinstance(inputs, (pd.DataFrame, pd.Series)):
        self.inputs = inputs.to_numpy()
    elif isinstance(inputs, np.ndarray):
        self.inputs = inputs
    else:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB610.value}: The argument `inputs` should be "
                                                f"an instance one of np.ndarray, pd.DataFrame or pd.Series")
    # Configuring self.target attribute
    if isinstance(target, (pd.DataFrame, pd.Series)):
        self.target = target.to_numpy()
    elif isinstance(inputs, np.ndarray):
        self.target = target
    else:
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB610.value}: The argument `target` should be "
                                                f"an instance one of np.ndarray, pd.DataFrame or pd.Series")

    # The lengths should be equal
    if len(self.inputs) != len(self.target):
        raise FedbiomedDatasetError(f"{ErrorNumbers.FB610.value}: Length of input variables and target "
                                                f"variable does not match. Please make sure that they have "
                                                f"equal size while creating the method `training_data` of "
                                                f"TrainingPlan")

    # Convert `inputs` adn `target` to Torch floats
    self.inputs = from_numpy(self.inputs).float()
    self.target = from_numpy(self.target).float()

def __init__(self, dataset: Dataset, **kwargs: dict):
    """Construct  of class

    Args:
        dataset: Dataset object for torch.utils.data.DataLoader
        **kwargs: Arguments for PyTorch `DataLoader`

    Raises:
        FedbiomedTorchDataManagerError: If the argument `dataset` is not an instance of `torch.utils.data.Dataset`
    """

    # TorchDataManager should get `dataset` argument as an instance of torch.utils.data.Dataset
    if not isinstance(dataset, Dataset):
        raise FedbiomedTorchDataManagerError(
            f"{ErrorNumbers.FB608.value}: The attribute `dataset` should an instance "
            f"of `torch.utils.data.Dataset`, please use `Dataset` as parent class for"
            f"your custom torch dataset object")

    self._dataset = dataset
    self._loader_arguments = kwargs
    self._subset_test: Union[Subset, None] = None
    self._subset_train: Union[Subset, None] = None