data_loader.py

import numpy as np
import paths as pt
from abc import ABC, abstractmethod
from tools import file_reader
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split

class BaseDataLoader(ABC):
    def __init__(self, file_name, converters=None):
        """Initilizer method that takes a file name and optionally a converter"""
        self.file_name = file_name
        self.converters = converters
        
    @abstractmethod
    def load_data(self):
        """Loads the data from a data set at startup"""
        
    def get_data(self):
        """Returns the features and target"""
        return self.X, self.y
    
    def get_features(self):
        """Returns the feature names"""
        return self.X.columns
        
    def prepare_data(self, scaling_strategy=None):
        """Prepares the data from a data set"""
        X = np.array(self.X)
        y = np.array(self.y)
        
        if scaling_strategy != None:
            emb_cols = self.X.filter(regex='((\d+)[Ats])\w+', axis=1)
            n_scale_cols = self.X.shape[1] - emb_cols.shape[1]
            
            if scaling_strategy == "Standard":
                scaler = StandardScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
            else:
                scaler = MinMaxScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
            
        return X, y
        
    def prepare_data_split(self, test_size, scaling_strategy=None):
        """Prepares and splits the data from a data set"""
        X = np.array(self.X)
        y = np.array(self.y)

        if scaling_strategy != None:
            emb_cols = self.X.filter(regex='((\d+)[Ats])\w+', axis=1)
            n_scale_cols = self.X.shape[1] - emb_cols.shape[1]
            if scaling_strategy == "Standard":
                scaler = StandardScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
            else:
                scaler = MinMaxScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
        
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size,
                                                    stratify=y, random_state=0)
        
        return X_train, X_test, y_train, y_test

class CompleteDataLoader(BaseDataLoader):
    def load_data(self):
        df = file_reader.read_csv(pt.PROCESSED_DATA_DIR,
                                  self.file_name,
                                  converters=self.converters)
        X = df.drop(['Complete'], axis=1)
        y = df['Complete']
        self.X = X
        self.y = y
        return self
    
class ComplianceDataLoader(BaseDataLoader):
    def load_data(self):
        df = file_reader.read_csv(pt.PROCESSED_DATA_DIR,
                                  self.file_name,
                                  converters=self.converters)
        X = df.drop(['Compliance'], axis=1)
        y = df['Compliance']
        self.X = X
        self.y = y
        return self

class FallDataLoader(BaseDataLoader):
    def load_data(self):
        df = file_reader.read_csv(pt.PROCESSED_DATA_DIR,
                                  self.file_name,
                                  converters=self.converters)
        X = df.drop(['Fall'], axis=1)
        y = df['Fall']
        self.X = X
        self.y = y
        return self

class FallTestDataLoader(BaseDataLoader):
    def load_data(self):
        df = file_reader.read_csv(pt.PROCESSED_DATA_DIR,
                                  self.file_name,
                                  converters=self.converters)
        X = df.drop(['Fall'], axis=1)
        y = df['Fall']
        self.X = X
        self.y = y
        return self
    
    def prepare_data(self, scaling_strategy: str = None):
        X = np.array(self.X)
        y = np.array(self.y)
        
        if scaling_strategy != None:
            emb_cols = self.X.filter(regex='((\d+)[Ats|Ex])\w+', axis=1)
            n_scale_cols = self.X.shape[1] - emb_cols.shape[1]
            if scaling_strategy == "Standard":
                scaler = StandardScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
            else:
                scaler = MinMaxScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
        
        return X, y
    
    def prepare_data_split(self, test_size: float,
                           scaling_strategy: str = None):
        X = np.array(self.X)
        y = np.array(self.y)
    
        if scaling_strategy != None:
            emb_cols = self.X.filter(regex='((\d+)[Ats|Ex])\w+', axis=1)
            n_scale_cols = self.X.shape[1] - emb_cols.shape[1]
            if scaling_strategy == "Standard":
                scaler = StandardScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
            else:
                scaler = MinMaxScaler()
                X_sc = scaler.fit_transform(X[:,:n_scale_cols])
                X = np.concatenate([X_sc, X[:,n_scale_cols:]], axis=1)
        
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size,
                                                    stratify=y, random_state=0)
        
        return X_train, X_test, y_train, y_test