<filename>KFolder.py
from sklearn.model_selection import KFold
from abc import ABC, abstractmethod
from typing import Tuple, Union, Dict, List, Any
from torch.utils.data import Dataset, DataLoader
import torch
import albumentations as A
from albumentations.pytorch.transforms import ToTensorV2
import random
import pandas as pd
import cv2


class PytorchImagesDataset(Dataset):
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.cancer_type, self.img_array = self.df['cancer_type'], self.df['img_array']

    def __len__(self):
        return len(self.df)

    def __getitem__(self, idx):
        img = self.img_array.iloc[idx]
        label = torch.tensor(int(self.cancer_type.iloc[idx]))
        return img, label


class AlbTrxs:
    resize_factor: int = 347  # each image is 1388x1040 pixels, so this value represents a 4x downsize
    tl_means: Tuple[float, float, float] = (0.485, 0.456, 0.406)
    tl_stds: Tuple[float, float, float] = (0.229, 0.224, 0.225)
    n_passes: int
    p: float = 0.0  # initialize

    def __init__(self):
        self.train_trxs = A.Compose([A.HorizontalFlip(p=self.p),
                                     A.VerticalFlip(p=self.p),
                                     A.ColorJitter(p=self.p),
                                     A.Rotate(limit=10, interpolation=cv2.BORDER_CONSTANT, p=self.p),
                                     A.RGBShift(p=self.p)
                                     ])
        self.normalize_resize_tensorize = A.Compose([A.Normalize(mean=self.tl_means, std=self.tl_stds),
                                                     A.LongestMaxSize(self.resize_factor),
                                                     ToTensorV2()
                                                     ])


class KFoldIndices(KFold):
    """this class creates indices for the outer and inner folds of the nested k-fold.
       k-folded dataframes will be constructed using these indices"""
    outer_train_idxs: Dict[int, Any]
    test_idxs: Dict[int, Any]
    train_idxs: Dict[Tuple[int, int], Any]
    valid_idxs: Dict[Tuple[int, int], Any]

    def __init__(self, image_data: pd.DataFrame, n_outer_splits: int, n_inner_splits: int, shuffle: bool = True, random_state: int = 42):
        super().__init__(shuffle=shuffle, random_state=random_state)
        self.image_df = image_data
        self.n_outer_splits, self.n_inner_splits = n_outer_splits, n_inner_splits
        self.shuffle, self.random_state = shuffle, random_state
        self.nested_outer()
        self.nested_inner()

    def nested_outer(self):
        """outer pass, return dicts of train and test set indices"""
        self.n_splits = self.n_outer_splits
        self.outer_train_idxs = {}
        self.test_idxs = {}

        i = 0
        """get indices of train and test splits and store in dicts"""
        for outer_train_idxs, test_idxs in self.split(self.image_df):
            self.outer_train_idxs[i] = outer_train_idxs
            self.test_idxs[i] = test_idxs
            i += 1

    def nested_inner(self):
        """inner pass, return dicts of train and val set indices"""
        self.n_splits = self.n_inner_splits
        self.train_idxs = {}
        self.valid_idxs = {}

        """get indices of train and val splits and store in dicts"""
        i = 0
        for n_fold, outer_train_idxs in self.outer_train_idxs.items():
            j = 0
            for train_idxs, valid_idxs in self.split(outer_train_idxs):
                self.train_idxs[(i, j)] = outer_train_idxs[train_idxs]
                self.valid_idxs[(i, j)] = outer_train_idxs[valid_idxs]
                j += 1
            i += 1


class KFoldedDatasets(ABC):
    """this abstract base class is used for constructing the train, valid, and test dataframes
       based on the k-fold indices generated by the KFoldIndices class object"""
    phase: str

    @abstractmethod
    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        self.image_df = kfold_idxs.image_df
        self.kfold_idxs = kfold_idxs
        self.n_outer_fold, self.n_inner_fold = n_outer_fold, n_inner_fold
        self.nkf_df = self.get_nkf_dataframe(self.n_outer_fold, self.n_inner_fold)

    @abstractmethod
    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        pass


class DFTrainKFolded(KFoldedDatasets):
    phase: str = 'train'

    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        """get the full dataframes from the train and valid idxs"""
        df = self.image_df.iloc[self.kfold_idxs.train_idxs[(n_outer_fold, n_inner_fold)]]
        return df.reset_index(drop=True)


class DFValidKFolded(KFoldedDatasets):
    phase: str = 'valid'

    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        df = self.image_df.iloc[self.kfold_idxs.valid_idxs[(n_outer_fold, n_inner_fold)]]
        return df.reset_index(drop=True)


class DFTestKFolded(KFoldedDatasets):
    phase: str = 'test'

    def __init__(self, n_outer_fold: int, kfold_idxs: KFoldIndices, n_inner_fold=None):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold):
        df = self.image_df.iloc[self.kfold_idxs.test_idxs[self.n_outer_fold]]
        return df.reset_index(drop=True)


class DataloaderBaseClass(ABC):
    """this abstract base class is used to create pytorch dataloaders based on the dataframes passed
       to it by any of the DFTrainKFolded, DFValidKFolded, or DFTestKFolded class objects"""
    dataset: PytorchImagesDataset
    dataset_len: int
    dataloader: DataLoader

    @abstractmethod
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int, num_workers: int = 0):
        self.nkf_df = kfolded_data.nkf_df
        self.transformations = transformations
        self.batch_size = batch_size
        self.num_workers = num_workers
        self.transformed_data = pd.DataFrame()

    def normalize_resize_data(self):
        """normalize and resize original data for PyTorch"""
        original_data = self.nkf_df.copy(deep=True)
        for i in range(len(original_data)):
            original_data.at[i, 'img_array'] = \
                self.transformations.normalize_resize_tensorize(image=original_data['img_array'].values[i])['image']
        self.transformed_data = original_data

    def create_dataloader(self):
        self.dataset = PytorchImagesDataset(self.transformed_data)
        self.dataloader = DataLoader(self.dataset, batch_size=self.batch_size, num_workers=self.num_workers)


class DFTrainDataloader(DataloaderBaseClass):
    generated_data: Dict[str, List[pd.Series]]

    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.generate_data()
        self.normalize_resize_data()
        self.combine_data()
        self.create_dataloader()

    def generate_data(self):
        original_data = self.nkf_df.copy(deep=True)
        label, image = original_data.columns[0], original_data.columns[1]

        self.generated_data = {label: [], image: []}
        aug_pass_counter = 0
        while aug_pass_counter < self.transformations.n_passes:
            for i in range(len(original_data)):
                transformed_array = self.transformations.train_trxs(image=original_data['img_array'].values[i])['image']
                transformed_array = self.transformations.normalize_resize_tensorize(image=transformed_array)['image']

                labeled_array = {label: original_data[label].iloc[i],  # get original subtype label
                                 image: transformed_array
                                 }
                self.generated_data[label].append(labeled_array[label])
                self.generated_data[image].append(labeled_array[image])
            aug_pass_counter += 1

    def combine_data(self):
        """concat the generated data with the normalized and resized original data"""

        generated_data = pd.DataFrame(data=self.generated_data)
        self.transformed_data = pd.concat([generated_data, self.transformed_data], ignore_index=True)


class DFValidDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()
        self.create_dataloader()


class DFTestDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()
        self.batch_size = len(self.nkf_df)
        self.create_dataloader()


class OverfitDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()

    def normalize_resize_data(self):
        rand_idx = random.randint(0, len(self.nkf_df))
        rand_entry = self.nkf_df.iloc[[rand_idx]].copy(deep=True)
        rand_entry.at[rand_idx, 'img_array'] = self.transformations.normalize_resize_tensorize(image=rand_entry.at[rand_idx, 'img_array'])['image']
        self.transformed_data = rand_entry

<filename>KFolder.py
from sklearn.model_selection import KFold
from abc import ABC, abstractmethod
from typing import Tuple, Union, Dict, List, Any
from torch.utils.data import Dataset, DataLoader
import torch
import albumentations as A
from albumentations.pytorch.transforms import ToTensorV2
import random
import pandas as pd
import cv2


class PytorchImagesDataset(Dataset):
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.cancer_type, self.img_array = self.df['cancer_type'], self.df['img_array']

    def __len__(self):
        return len(self.df)

    def __getitem__(self, idx):
        img = self.img_array.iloc[idx]
        label = torch.tensor(int(self.cancer_type.iloc[idx]))
        return img, label


class AlbTrxs:
    resize_factor: int = 347  # each image is 1388x1040 pixels, so this value represents a 4x downsize
    tl_means: Tuple[float, float, float] = (0.485, 0.456, 0.406)
    tl_stds: Tuple[float, float, float] = (0.229, 0.224, 0.225)
    n_passes: int
    p: float = 0.0  # initialize

    def __init__(self):
        self.train_trxs = A.Compose([A.HorizontalFlip(p=self.p),
                                     A.VerticalFlip(p=self.p),
                                     A.ColorJitter(p=self.p),
                                     A.Rotate(limit=10, interpolation=cv2.BORDER_CONSTANT, p=self.p),
                                     A.RGBShift(p=self.p)
                                     ])
        self.normalize_resize_tensorize = A.Compose([A.Normalize(mean=self.tl_means, std=self.tl_stds),
                                                     A.LongestMaxSize(self.resize_factor),
                                                     ToTensorV2()
                                                     ])


class KFoldIndices(KFold):
    """this class creates indices for the outer and inner folds of the nested k-fold.
       k-folded dataframes will be constructed using these indices"""
    outer_train_idxs: Dict[int, Any]
    test_idxs: Dict[int, Any]
    train_idxs: Dict[Tuple[int, int], Any]
    valid_idxs: Dict[Tuple[int, int], Any]

    def __init__(self, image_data: pd.DataFrame, n_outer_splits: int, n_inner_splits: int, shuffle: bool = True, random_state: int = 42):
        super().__init__(shuffle=shuffle, random_state=random_state)
        self.image_df = image_data
        self.n_outer_splits, self.n_inner_splits = n_outer_splits, n_inner_splits
        self.shuffle, self.random_state = shuffle, random_state
        self.nested_outer()
        self.nested_inner()

    def nested_outer(self):
        """outer pass, return dicts of train and test set indices"""
        self.n_splits = self.n_outer_splits
        self.outer_train_idxs = {}
        self.test_idxs = {}

        i = 0
        """get indices of train and test splits and store in dicts"""
        for outer_train_idxs, test_idxs in self.split(self.image_df):
            self.outer_train_idxs[i] = outer_train_idxs
            self.test_idxs[i] = test_idxs
            i += 1

    def nested_inner(self):
        """inner pass, return dicts of train and val set indices"""
        self.n_splits = self.n_inner_splits
        self.train_idxs = {}
        self.valid_idxs = {}

        """get indices of train and val splits and store in dicts"""
        i = 0
        for n_fold, outer_train_idxs in self.outer_train_idxs.items():
            j = 0
            for train_idxs, valid_idxs in self.split(outer_train_idxs):
                self.train_idxs[(i, j)] = outer_train_idxs[train_idxs]
                self.valid_idxs[(i, j)] = outer_train_idxs[valid_idxs]
                j += 1
            i += 1


class KFoldedDatasets(ABC):
    """this abstract base class is used for constructing the train, valid, and test dataframes
       based on the k-fold indices generated by the KFoldIndices class object"""
    phase: str

    @abstractmethod
    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        self.image_df = kfold_idxs.image_df
        self.kfold_idxs = kfold_idxs
        self.n_outer_fold, self.n_inner_fold = n_outer_fold, n_inner_fold
        self.nkf_df = self.get_nkf_dataframe(self.n_outer_fold, self.n_inner_fold)

    @abstractmethod
    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        pass


class DFTrainKFolded(KFoldedDatasets):
    phase: str = 'train'

    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        """get the full dataframes from the train and valid idxs"""
        df = self.image_df.iloc[self.kfold_idxs.train_idxs[(n_outer_fold, n_inner_fold)]]
        return df.reset_index(drop=True)


class DFValidKFolded(KFoldedDatasets):
    phase: str = 'valid'

    def __init__(self, n_outer_fold: int, n_inner_fold: int, kfold_idxs: KFoldIndices):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold: int):
        df = self.image_df.iloc[self.kfold_idxs.valid_idxs[(n_outer_fold, n_inner_fold)]]
        return df.reset_index(drop=True)


class DFTestKFolded(KFoldedDatasets):
    phase: str = 'test'

    def __init__(self, n_outer_fold: int, kfold_idxs: KFoldIndices, n_inner_fold=None):
        super().__init__(n_outer_fold, n_inner_fold, kfold_idxs)

    def get_nkf_dataframe(self, n_outer_fold: int, n_inner_fold):
        df = self.image_df.iloc[self.kfold_idxs.test_idxs[self.n_outer_fold]]
        return df.reset_index(drop=True)


class DataloaderBaseClass(ABC):
    """this abstract base class is used to create pytorch dataloaders based on the dataframes passed
       to it by any of the DFTrainKFolded, DFValidKFolded, or DFTestKFolded class objects"""
    dataset: PytorchImagesDataset
    dataset_len: int
    dataloader: DataLoader

    @abstractmethod
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int, num_workers: int = 0):
        self.nkf_df = kfolded_data.nkf_df
        self.transformations = transformations
        self.batch_size = batch_size
        self.num_workers = num_workers
        self.transformed_data = pd.DataFrame()

    def normalize_resize_data(self):
        """normalize and resize original data for PyTorch"""
        original_data = self.nkf_df.copy(deep=True)
        for i in range(len(original_data)):
            original_data.at[i, 'img_array'] = \
                self.transformations.normalize_resize_tensorize(image=original_data['img_array'].values[i])['image']
        self.transformed_data = original_data

    def create_dataloader(self):
        self.dataset = PytorchImagesDataset(self.transformed_data)
        self.dataloader = DataLoader(self.dataset, batch_size=self.batch_size, num_workers=self.num_workers)


class DFTrainDataloader(DataloaderBaseClass):
    generated_data: Dict[str, List[pd.Series]]

    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.generate_data()
        self.normalize_resize_data()
        self.combine_data()
        self.create_dataloader()

    def generate_data(self):
        original_data = self.nkf_df.copy(deep=True)
        label, image = original_data.columns[0], original_data.columns[1]

        self.generated_data = {label: [], image: []}
        aug_pass_counter = 0
        while aug_pass_counter < self.transformations.n_passes:
            for i in range(len(original_data)):
                transformed_array = self.transformations.train_trxs(image=original_data['img_array'].values[i])['image']
                transformed_array = self.transformations.normalize_resize_tensorize(image=transformed_array)['image']

                labeled_array = {label: original_data[label].iloc[i],  # get original subtype label
                                 image: transformed_array
                                 }
                self.generated_data[label].append(labeled_array[label])
                self.generated_data[image].append(labeled_array[image])
            aug_pass_counter += 1

    def combine_data(self):
        """concat the generated data with the normalized and resized original data"""

        generated_data = pd.DataFrame(data=self.generated_data)
        self.transformed_data = pd.concat([generated_data, self.transformed_data], ignore_index=True)


class DFValidDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()
        self.create_dataloader()


class DFTestDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()
        self.batch_size = len(self.nkf_df)
        self.create_dataloader()


class OverfitDataloader(DataloaderBaseClass):
    def __init__(self, kfolded_data: Union[DFTrainKFolded, DFValidKFolded, DFTestKFolded],
                 transformations: AlbTrxs, batch_size: int):
        super().__init__(kfolded_data, transformations, batch_size)
        self.normalize_resize_data()

    def normalize_resize_data(self):
        rand_idx = random.randint(0, len(self.nkf_df))
        rand_entry = self.nkf_df.iloc[[rand_idx]].copy(deep=True)
        rand_entry.at[rand_idx, 'img_array'] = self.transformations.normalize_resize_tensorize(image=rand_entry.at[rand_idx, 'img_array'])['image']
        self.transformed_data = rand_entry

# each image is 1388x1040 pixels, so this value represents a 4x downsize # initialize this class creates indices for the outer and inner folds of the nested k-fold.        k-folded dataframes will be constructed using these indices outer pass, return dicts of train and test set indices get indices of train and test splits and store in dicts inner pass, return dicts of train and val set indices get indices of train and val splits and store in dicts this abstract base class is used for constructing the train, valid, and test dataframes        based on the k-fold indices generated by the KFoldIndices class object get the full dataframes from the train and valid idxs this abstract base class is used to create pytorch dataloaders based on the dataframes passed        to it by any of the DFTrainKFolded, DFValidKFolded, or DFTestKFolded class objects normalize and resize original data for PyTorch # get original subtype label concat the generated data with the normalized and resized original data