Machine learning utilities to help speed up the prototyping process.
from .data_helpers import DatasetSplitter,DataSubSampler
from .plot_helpers import plot_history_curves,plot_confusion_matrix
deepfastmlu/extra/__init__.py
fabprezja/deep-fast-machine-learning-utils
import os
import shutil
import numpy as np
import sys

class DatasetSplitter:
    """
    DatasetSplitter class for splitting a dataset into train, validation, and test sets.

    Attributes:
        data_dir (str): Directory containing the dataset.
        destination_dir (str): Directory where the train, validation, and test sets will be saved.
        train_ratio (float): Ratio of train set.
        val_ratio (float): Ratio of validation set.
        test_ratio (float): Ratio of test set.
        test_ratio_2 (float): Ratio of an additional test set, if needed.
        seed (int): Seed for random number generator.
    """
    def __init__(self, data_dir, destination_dir, train_ratio=0.7, val_ratio=0.15, test_ratio=0.15, test_ratio_2=None,
                 seed=None):
        self.data_dir = data_dir
        self.destination_dir = destination_dir
        self.train_ratio = train_ratio
        self.val_ratio = val_ratio
        self.test_ratio = test_ratio
        self.test_ratio_2 = test_ratio_2
        self.seed = seed
        if self.seed is not None:
            np.random.seed(self.seed)

    def run(self):
        """
        Main method to execute dataset splitting and copying files.
        """
        if not os.path.exists(self.destination_dir):
            os.makedirs(self.destination_dir)

        subdirs = self._get_subdirs()
        total_files = self._count_total_files(subdirs)

        processed_files = 0
        for subdir in subdirs:
            subdir_path = os.path.join(self.data_dir, subdir)
            files = os.listdir(subdir_path)
            np.random.shuffle(files)

            train_split, val_split, test_split, test_split_2 = self._calculate_split_indices(len(files))

            subdir_processed_files = self._copy_files(files, subdir, train_split, val_split, test_split,
                                                      test_split_2)
            processed_files += subdir_processed_files
            self._update_progress_bar(processed_files, total_files)

    def _get_subdirs(self):
        """
        Gets the list of subdirectories in the data directory.

        Returns:
            list: List of subdirectories.
        """
        return [d for d in os.listdir(self.data_dir) if os.path.isdir(os.path.join(self.data_dir, d))]

    def _count_total_files(self, subdirs):
        """
        Counts the total number of files in the given subdirectories.

        Args:
            subdirs (list): List of subdirectories.

        Returns:
            int: Total number of files.
        """
        return sum(len(os.listdir(os.path.join(self.data_dir, subdir))) for subdir in subdirs)

    def _calculate_split_indices(self, num_files):
        """
        Calculates the split indices for train, validation, test, and additional test set (if applicable).

        Args:
            num_files (int): Total number of files in the subdirectory.

        Returns:
            tuple: Train split index, validation split index, test split index, and additional test split index (if applicable).
        """
        train_split = int(self.train_ratio * num_files)
        val_split = int((self.train_ratio + self.val_ratio) * num_files)
        test_split = int((self.train_ratio + self.val_ratio + self.test_ratio) * num_files) if self.test_ratio else None
        test_split_2 = int((self.train_ratio + self.val_ratio + self.test_ratio + self.test_ratio_2) * num_files) if self.test_ratio_2 else None
        return train_split, val_split, test_split, test_split_2

    def _create_destination_path(self, file_index, subdir, train_split, val_split, test_split):
        """
        Creates the destination path for a file based on its index.

        Args:
            file_index (int): Index of the file in the subdirectory.
            subdir (str): Subdirectory name.
            train_split (int): Train split index.
            val_split (int): Validation split index.
            test_split (int): Test split index.

        Returns:
            str: Destination path for the file.
        """
        if file_index < train_split:
            return os.path.join(self.destination_dir, 'train', subdir)
        elif file_index < val_split:
            return os.path.join(self.destination_dir, 'val', subdir)
        elif self.test_ratio and file_index < test_split:
            return os.path.join(self.destination_dir, 'test', subdir)
        elif self.test_ratio_2:
            return os.path.join(self.destination_dir, 'test_set_2', subdir)
        else:
            return None

    def _copy_files(self, files, subdir, train_split, val_split, test_split, test_split_2):
        """
        Copies files from the source directory to the destination directory based on the split indices.

        Args:
            files (list): List of file names.
            subdir (str): Subdirectory name.
            train_split (int): Train split index.
            val_split (int): Validation split index.
            test_split (int): Test split index.
            test_split_2 (int): Second test set split index, if applicable.

        Returns:
            int: Number of processed files.
        """
        processed_files = 0

        for i, file in enumerate(files):
            src_path = os.path.join(self.data_dir, subdir, file)
            dest_path = self._create_destination_path(i, subdir, train_split, val_split, test_split)

            if dest_path is None:
                continue

            os.makedirs(dest_path, exist_ok=True)
            shutil.copy(src_path, os.path.join(dest_path, file))
            processed_files += 1

        return processed_files

    def _update_progress_bar(self,current,total):
        """
        Updates the progress bar based on the current number of processed files and the total number of files.

        Args:
            current (int): Current number of processed files.
            total (int): Total number of files.
        """
        progress = int((current/total)*100)
        bar = '='*progress + ' '*(100 - progress)
        print(f'\rSplitting Data: [{bar}] {progress}%',end='',flush=True)

class DataSubSampler:
    """
    DataSubSampler class for creating a smaller dataset by randomly sampling a fraction of files from the original dataset.

    Attributes:
        data_dir (str): Directory containing the original dataset.
        destination_dir (str): Directory where the sampled dataset will be saved.
        fraction (float): Fraction of files to sample from the original dataset.
        seed (int): Seed for random number generator.
    """
    def __init__(self, data_dir, destination_dir, fraction, seed=None):
        self.data_dir = data_dir
        self.destination_dir = destination_dir
        self.fraction = fraction
        self.seed = seed
        if self.seed is not None:
            np.random.seed(self.seed)

    def create_miniature_dataset(self):
        """
        Creates a copy of all folders and subfolders in the path, but only samples a fraction of files.
        """
        for root, dirs, files in os.walk(self.data_dir):
            destination_root = root.replace(self.data_dir, self.destination_dir)

            if not os.path.exists(destination_root):
                os.makedirs(destination_root)

            files_to_sample = int(len(files) * self.fraction)
            sampled_files = np.random.choice(files, files_to_sample, replace=False)

            for file in sampled_files:
                src_path = os.path.join(root, file)
                dest_path = os.path.join(destination_root, file)
                shutil.copy(src_path, dest_path)
deepfastmlu/extra/data_helpers.py
fabprezja/deep-fast-machine-learning-utils
import numpy as np
import seaborn as sns
import matplotlib.pyplot as plt
from sklearn.metrics import confusion_matrix

def plot_confusion_matrix(model, generator, generator_name, target_type):
    """
    Plot the confusion matrix of the model predictions on a given dataset.

    Args:
        model (Model): A trained Keras model.
        generator (ImageDataGenerator): A Keras ImageDataGenerator.
        generator_name (str): Name of the generator, used in the plot title.
        target_type (str): Type of target labels ('binary' or 'categorical').
    """
    true_labels = generator.classes
    pred_prob = model.predict(generator)

    if target_type == 'binary':
        threshold = 0.5
        pred_labels = (pred_prob > threshold).astype(int)
    else:
        pred_labels = np.argmax(pred_prob, axis=1)

    cm = confusion_matrix(true_labels, pred_labels)
    cm_norm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
    class_labels = list(generator.class_indices.keys())
    plt.figure()
    sns.heatmap(cm_norm, annot=True, cmap="Blues", fmt='.2f', xticklabels=class_labels, yticklabels=class_labels)
    plt.xlabel('Predicted label')
    plt.ylabel('True label')
    plt.title(f'{generator_name} Normalized Confusion Matrix')
    plt.show()


def plot_history_curves(history, show_min_max_plot, user_metric):
    """
    Plot the training and validation loss and user-specified metric curves.

    Args:
        history (History): Keras training history object.
        show_min_max_plot (bool): Whether to plot the maximum/minimum value lines.
        user_metric (str): User-specified metric to be plotted.
    """
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(6, 8))

    _plot_curve(ax1, history, 'loss', 'Loss', show_min_max_plot)
    train_metric_key = next(key for key in history.history.keys() if key.lower().startswith(user_metric))
    val_metric_key = next(key for key in history.history.keys() if key.lower().startswith(f'val_{user_metric}'))

    _plot_curve(ax2, history, train_metric_key, user_metric.capitalize(), show_min_max_plot)

    plt.tight_layout()
    plt.show()


def _plot_curve(ax, history, metric, ylabel, show_min_max_plot):
    """
    Plot a curve of a specific metric.

    Args:
        ax (Axes): Matplotlib Axes object.
        history (History): Keras training history object.
        metric (str): Metric to be plotted.
        ylabel (str): Label for the y-axis.
        show_min_max_plot (bool): Whether to plot the maximum/minimum value lines.
    """
    ax.plot(history.history[metric], label=f'Train {ylabel}')
    ax.plot(history.history[f'val_{metric}'], label=f'Validation {ylabel}')

    if show_min_max_plot:
        _plot_max_min_lines(ax, history, metric, ylabel)

    _set_plot_labels(ax, history, metric, ylabel)

    ax.legend()
    ax.set_xticks(range(0, len(history.history[metric])))
    ax.set_xticklabels([str(x + 1) for x in ax.get_xticks()])


def _plot_max_min_lines(ax, history, metric, ylabel):
    """
    Plot the maximum/minimum value lines for a specific metric.

    Args:
        ax (Axes): Matplotlib Axes object.
        history (History): Keras training history object.
        metric (str): Metric to be plotted.
        ylabel (str): Label for the y-axis.
    """
    val_metric_key = f'val_{metric}'
    if val_metric_key in history.history:
        val_metric_epoch = np.argmax(history.history[val_metric_key]) if ylabel != 'Loss' \
            else np.argmin(history.history[val_metric_key])
        max_val_metric = np.max(history.history[val_metric_key]) if ylabel != 'Loss' \
            else np.min(history.history[val_metric_key])
        ax.axvline(x=val_metric_epoch, color='g', linestyle='--', alpha=0.5,
                   label=(f'Maximum Validation {ylabel}: {max_val_metric:.4f}' if ylabel != 'Loss'
                          else f'Minimum Validation Loss: {max_val_metric:.4f}'))


def _set_plot_labels(ax, history, metric, ylabel):
    """
    Set the plot labels, grid, legend, and tick labels for a specific metric.

    Args:
        ax (Axes): Matplotlib Axes object.
        history (History): Keras training history object.
        metric (str): Metric to be plotted.
        ylabel (str): Label for the y-axis.
    """
    ax.set_title(f'{ylabel} Curves')
    ax.set_xlabel('Epoch')
    ax.set_ylabel(ylabel)
    ax.grid(True,linestyle='--',alpha=0.5)
    ax.legend()
    ax.set_xticks(range(0,len(history.history[metric])))
    ax.set_xticklabels([str(x + 1) for x in ax.get_xticks()])
deepfastmlu/extra/plot_helpers.py
fabprezja/deep-fast-machine-learning-utils
import numpy as np
from sklearn.feature_selection import VarianceThreshold

class AdaptiveVarianceThreshold:
    """
    AdaptiveVarianceThreshold is a feature selector that dynamically determines a variance threshold based on the
    provided percentile of the feature variances. Features with a variance below this threshold are dropped.
    Traditional variance-based feature selection uses a fixed threshold, which is not optimal for all datasets.

    Attributes:
        percentile (float): The percentile of the feature variances used to determine the threshold. A higher
        percentile will result in a higher threshold and potentially more features being dropped.

        variances (np.ndarray, optional): Holds the variances of each feature in the dataset. Calculated during the
        `fit` method.

        threshold (float, optional): The calculated variance threshold. Features with a variance below this value
        will be dropped during the `transform` method.

        indices_to_drop (np.ndarray, optional): After fitting and transforming, provides the indices of features
        that were dropped due to their variance.

        selector (VarianceThreshold, optional): An instance of Scikit-learn's VarianceThreshold class, used to
        perform the feature selection based on the calculated threshold.

        verbose (bool): If set to True, prints additional information during the processing, such as the calculated
        variance threshold.

    Args:
        percentile (float, optional): The desired percentile of feature variances to use for determining the
        threshold. Defaults to 1.5.

        verbose (bool, optional): Whether to print additional information during processing. Defaults to False.

    ...
    """

    def __init__(self, percentile=1.5, verbose=False):
        """Initialize the AdaptiveVarianceThreshold object.

        Args:
            percentile (float, optional): The variances percentile for determining the threshold. Defaults to 1.5.
            verbose (bool, optional): Whether to print additional information during processing. Defaults to False.
        """
        self.percentile = percentile
        self.variances = None
        self.threshold = None
        self.indices_to_drop = None
        self.selector = None
        self.verbose = verbose

    def fit(self, features):
        """Calculate the variances and threshold, and fit the selector.

        Args:
            features (np.array): The feature set.
        """
        self.variances = np.var(features, axis=0)
        self.threshold = np.percentile(self.variances, self.percentile)
        if self.verbose:
            print(f"Variance threshold: {self.threshold}")
        self.selector = VarianceThreshold(threshold=self.threshold)
        self.selector.fit(features)

    def transform(self, features):
        """Apply the selector to the features and store the indices of the dropped features.

        Args:
            features (np.array): The feature set.

        Returns:
            np.array: The transformed feature set.
        """
        features_transformed = self.selector.transform(features)
        self.indices_to_drop = np.where(self.selector.variances_ < self.threshold)
        return features_transformed

    def fit_transform(self, features, y=None):
        """
        Call the fit and transform methods successively.

        Args:
            features (np.array): The feature set.
            y (np.array, optional): Target values. This parameter is included for compatibility with scikit-learn's
            transformer API but is not used in this method.

        Returns:
            np.array: The transformed feature set.
        """
        self.fit(features)
        return self.transform(features)

deepfastmlu/feature_select/AdaptiveVarianceThreshold.py
fabprezja/deep-fast-machine-learning-utils
import numpy as np
from typing import List, Optional, Any

class ChainedFS:
    """
    ChainedFS is a feature selector that sequentially applies a list of feature selection methods. This class allows
    for the chaining of multiple feature selection methods, where the output of one method becomes the input for the
    next. This can be particularly useful when one wants to combine the strengths of different feature selection
    techniques or when a sequence of operations is required to refine the feature set.

    For instance, one might first want to use a variance threshold to remove low-variance features and then apply
    a more computationally intensive method on the reduced set.

    Attributes:
        methods (List[Any]): A list of feature selection methods to be applied in sequence. Each method should have
        `fit` and `transform` methods. The order of the list determines the order of application of the methods.

        indices_ (np.ndarray): After fitting, provides the indices of features retained after all methods are applied.

    Args:
        methods (List[Any], optional): Feature selection methods to apply in sequence. Defaults to an empty list.

    ...
    """

    def __init__(self, methods: Optional[List[Any]] = None):
        """Initialize the ChainedFS object.

        Args:
            methods (List[Any], optional): Feature selection methods to apply in sequence. Defaults to an empty list.
        """
        self.methods = methods if methods is not None else []
        self.indices_ = np.array([])

    def fit(self, X: np.ndarray, y: Optional[np.ndarray] = None):
        """Fit the ChainedFS to the data.

        Args:
            X (np.ndarray): Training data.
            y (np.ndarray, optional): Target values. Defaults to None.
        """
        X_new = X
        self.indices_ = np.arange(X.shape[1])

        for method in self.methods:
            method.fit(X_new, y)
            if hasattr(method, 'get_support'):
                support = method.get_support()
                X_new = X_new[:, support]
                self.indices_ = self.indices_[support]

        return self

    def transform(self, X: np.ndarray) -> np.ndarray:
        """Transform data based on the sequence of methods applied.

        Args:
            X (np.ndarray): Data to transform.

        Returns:
            np.ndarray: Transformed data.
        """
        X_new = X
        for method in self.methods:
            X_new = method.transform(X_new)
        return X_new

    def fit_transform(self, X: np.ndarray, y: Optional[np.ndarray] = None) -> np.ndarray:
        """Fit to data, then transform it.

        Args:
            X (np.ndarray): Training data.
            y (np.ndarray, optional): Target values. Defaults to None.

        Returns:
            np.ndarray: Transformed data.
        """
        self.fit(X, y)
        return self.transform(X)
deepfastmlu/feature_select/ChainedFS.py
fabprezja/deep-fast-machine-learning-utils
import numpy as np
from typing import List, Optional, Any, Union

class RankAggregatedFS:
    """
    RankAggregatedFS is a feature selector that aggregates the rankings of features from multiple feature selection
    methods. It combines the scores or rankings of features from different methods to provide a unified ranking of
    features. This approach can be useful when there's uncertainty about which feature selection method to use, as
    it combines the strengths of multiple methods.

    Attributes:
        ranking_ (np.ndarray): After fitting, provides the indices of features ranked based on their aggregated scores.
        aggregated_scores_ (np.ndarray): After fitting, provides the aggregated scores of features.

    Args:
        methods (List[Any], optional): A list of feature selection methods to aggregate. Each method should have a
        `fit` method and either provide a `scores_` attribute after fitting or be compatible with the `transform`
        method. Defaults to an empty list.

        k (int): The number of top-ranked features to select after aggregation. Features are ranked based on their
        aggregated scores, and the top `k` features are selected.

        weights (Optional[List[float]], optional): If provided, assigns weights to the feature selection methods. This
        can be used to give more importance to certain methods over others. If None, all methods are equally weighted.

    ...
    """

    def __init__(self, methods: Optional[List[Any]] = None, k: int = 3, weights: Optional[List[float]] = None):
        """
        Initialize the RankAggregatedFS object.

        Args:
            methods (List[Any], optional): Feature selection methods to aggregate. Defaults to an empty list.
            k (int): Number of top-ranked features to select.
            weights (Optional[List[float]]): Weights for each method. If None, all methods are equally weighted.
        """
        self.methods = methods if methods is not None else []
        self.k = k
        self.weights = weights
        self.ranking_ = np.array([])
        self.aggregated_scores_ = np.array([])

        if self.weights is not None and len(self.weights) != len(self.methods):
            raise ValueError("The number of weights must match the number of methods.")

    def fit(self, X: np.ndarray, y: Optional[np.ndarray] = None) -> 'RankAggregatedFS':
        """
        Fit the RankAggregatedFS to the data.

        Args:
            X (np.ndarray): Training data.
            y (np.ndarray, optional): Target values. Defaults to None.
        """
        scores = []

        for method in self.methods:
            method.fit(X, y)
            if hasattr(method, 'scores_'):
                scores.append(method.scores_)
            else:
                scores.append(np.ones(X.shape[1]))

        if self.weights is None:
            self.aggregated_scores_ = np.mean(scores, axis=0)
        else:
            self.aggregated_scores_ = np.average(scores, axis=0, weights=self.weights)

        self.ranking_ = np.argsort(self.aggregated_scores_)[::-1][:self.k]

        return self

    def transform(self, X: np.ndarray) -> np.ndarray:
        """
        Transform data to retain only the selected features.

        Args:
            X (np.ndarray): Data to transform.

        Returns:
            np.ndarray: Transformed data.
        """
        return X[:, self.ranking_]

    def fit_transform(self, X: np.ndarray, y: Optional[np.ndarray] = None) -> np.ndarray:
        """
        Fit to data, then transform it.

        Args:
            X (np.ndarray): Training data.
            y (np.ndarray, optional): Target values. Defaults to None.

        Returns:
            np.ndarray: Transformed data.
        """
        self.fit(X, y)
        return self.transform(X)

    def get_feature_rankings(self) -> np.ndarray:
        """
        Return features ranked by their aggregated scores.

        Returns:
            np.ndarray: Features ranked by their aggregated scores.
        """
        return np.argsort(self.aggregated_scores_)[::-1]
deepfastmlu/feature_select/RankAggregatedFS.py
fabprezja/deep-fast-machine-learning-utils
from .AdaptiveVarianceThreshold import AdaptiveVarianceThreshold
from .ChainedFS import ChainedFS
from .RankAggregatedFS import RankAggregatedFS
deepfastmlu/feature_select/__init__.py
fabprezja/deep-fast-machine-learning-utils
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.layers import BatchNormalization, Dense, Dropout
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.regularizers import l1, l2


class PCCDNAS:
    """
    Principal Component Cascade Dense Neural Architecture Search (PCCDNAS).

    PCCDNAS provides an automated method for designing neural networks. Using PCA
    (Principal Component Analysis), it systematically sets the number of neurons
    in each layer of the network. After applying PCA to the initial data, the
    neuron count for the first layer is determined based on the principal components
    (PCs) for a given variance threshold. Subsequently, the cascade mechanism ensures
    that the activations from each trained layer undergo PCA again. This process,
    in turn, determines the neuron count for the subsequent layers using the same
    principal component variance threshold criteria.
    """

    def __init__(self):
        """
        Initializes an empty Sequential model, an empty list for layer neurons,
        and a scaler for data normalization.
        """
        self._model = Sequential()
        self._num_neurons = []
        self._scaler = None

    def data_init(self, X_train, y_train, validation=None, normalize=True, unit=False):
        """
        Initialize data for searching the model.

        Args:
            X_train (ndarray): Training data.
            y_train (ndarray): Training labels/targets.
            validation (float, tuple, optional): Validation data. Can be a percentage (float) or a tuple (X_val, y_val).
            normalize (bool, optional): Whether to normalize the data or not. Default is True.
            unit (bool, optional): If True, standard deviation is used for normalization. Default is False.
        """
        self._X_train = X_train
        self._y_train = y_train
        self._validation = validation
        self._normalize = normalize
        self._unit = unit

        self._process_validation_data()
        if self._normalize:
            self._normalize_data()

    def initialize_model_search(self, **kwargs):
        """
        Initialize the model's hyperparameters based on provided keyword arguments.

        Note:
            All parameters are passed as keyword arguments.

        Args:
            epochs (int, optional): Number of training epochs. Default is 15.
            layers (int, optional): Number of layers in the model. Default is 2.
            activation (str, optional): Activation function for the layers. Default is 'elu'.
            pca_variance (float or list of float, optional): Desired explained variance for PCA. Default is 0.95.
            loss (str, optional): Loss function for the model. Default is 'categorical_crossentropy'.
            optimizer (str, optional): Optimizer for the model. Default is 'adam'.
            metrics (list, optional): List of metrics to be evaluated. Default is ['accuracy'].
            output_neurons (int, optional): Number of neurons in the output layer. Default is 1.
            out_activation (str, optional): Activation function for the output layer. Default is 'sigmoid'.
            stop_criteria (str, optional): Criteria for early stopping. Default is 'val_loss'.
            es_mode (str, optional): Mode for early stopping. Default is 'max'.
            dropout (float, optional): Dropout rate for dropout layers (includes). Default is None.
            regularize (tuple (str, float), optional): Regularization type and value. Default is None.
            batch_size (int, optional): Batch size for training. Default is 32.
            kernel_initializer (str, optional): Kernel initializer for the dense layers. Default is 'he_normal'.
            batch_norm (bool, optional): Whether to include batch normalization layers. Default is True.
            es_patience (int, optional): Number of epochs with no improvement for early stopping. Default is 5.
            verbose (int, optional): Verbosity mode. 0 = silent, 1 = progress bar, 2 = one line per epoch. Default is 1.
            custom_callback (callback, optional): Custom callback for training. Default is None.
            learn_rate (float, optional): Learning rate for the optimizer. Default is 0.0001.
        """

        # Default values
        defaults = {
            'epochs': 15,
            'layers': 2,
            'activation': 'elu',
            'pca_variance': 0.95,
            'loss': 'categorical_crossentropy',
            'optimizer': 'adam',
            'metrics': ['accuracy'],
            'output_neurons': 1,
            'out_activation': 'sigmoid',
            'stop_criteria': 'val_loss',
            'es_mode': 'max',
            'dropout': None,
            'regularize': None,
            'batch_size': 32,
            'kernel_initializer': 'he_normal',
            'batch_norm': True,
            'es_patience': 5,
            'verbose': 1,
            'custom_callback': None,
            'learn_rate': 0.0001
        }

        for key, default_value in defaults.items():
            setattr(self, f"_{key}", kwargs.get(key, default_value))

        self._process_pca_variance()

    def _process_pca_variance(self):
        """
        Processes PCA variance to ensure it's valid. Either a single number
        or a list of length equal to the number of layers is expected.
        """
        if isinstance(self._pca_variance, (int, float)):
            self._variance_list = [self._pca_variance] * self._layers
        elif isinstance(self._pca_variance, list) and len(self._pca_variance) == self._layers:
            self._variance_list = self._pca_variance
        else:
            raise ValueError(
                'PCA variance must be a single number or a list of length equal to the number of layers.')

    def _process_validation_data(self):
        """
        Processes validation data based on its type (float or tuple).
        """
        if isinstance(self._validation, float):
            self._validation_split = self._validation
            self._validation_data = None
        elif isinstance(self._validation, tuple) and len(self._validation) == 2:
            self._validation_data = self._validation
            self._validation_split = None
        else:
            raise ValueError("Validation must be either float or a tuple in form (X_val, y_val)")

    def _normalize_data(self):
        """
        Normalize the training data using the Standard Scaler.
        If validation data is present, it is also transformed (not fitted).
        """
        self._scaler = StandardScaler(with_std=self._unit)
        self._X_train = self._scaler.fit_transform(self._X_train)
        if self._validation_data:
            self._validation_data = (
                self._scaler.transform(self._validation_data[0]), self._validation_data[1]
            )

    def _create_callback(self):
        """
        Creates an early stopping callback for the Keras model based on the
        specified criteria.

        Returns:
            EarlyStopping: Configured early stopping callback.
        """
        return EarlyStopping(
            monitor=self._stop_criteria, mode=self._es_mode, verbose=self._verbose,
            patience=self._es_patience, restore_best_weights=True
        )

    def _perform_pca(self, data, variance):
        """
        Performs PCA on the provided data to determine the number of components
        that explain the given variance.

        Args:
            data (ndarray): Input data for PCA.
            variance (float): Desired explained variance for PCA.

        Returns:
            int: Number of PCA components.
        """
        pca = PCA(variance)
        pca.fit(data)
        return pca.n_components_

    def _build_layer(self, layer_idx):
        """
        Builds an individual layer of the neural network based on the PCA result.

        Args:
            layer_idx (int): Index of the layer to be built.

        Returns:
            int: Number of PCA components for the current layer unit count.
        """
        regularizer = self._get_regularizer()

        if layer_idx == 0:
            n_pca_components = self._perform_pca(self._X_train, self._variance_list[layer_idx])
        else:
            self._append_and_train_previous_layer(regularizer)
            layer_output_model = Model(inputs=self._model.input, outputs=self._model.layers[-1].output)
            X_train_transformed = layer_output_model.predict(self._X_train)
            n_pca_components = self._perform_pca(X_train_transformed, self._variance_list[layer_idx])

        self._num_neurons.append(n_pca_components)
        return n_pca_components

    def _get_regularizer(self):
        """
        Returns the appropriate regularizer (L1 or L2) based on the provided type.

        Returns:
            Regularizer: Configured regularizer.
        """
        if self._regularize:
            return l1(self._regularize[1]) if self._regularize[0].lower() == 'l1' else l2(self._regularize[1])

    def _append_and_train_previous_layer(self, regularizer):
        """
        Appends necessary layers to the model, trains it, and pops the last layer.
        This method prepares the model for the next layer of neurons.

        Args:
            regularizer (Regularizer): Regularizer to be applied to the layer.
        """
        self._append_batch_norm_if_needed()
        self._append_dense_dropout_compile(regularizer)
        self._train_model()
        self._model.pop()

    def _append_batch_norm_if_needed(self):
        """
        Adds a batch normalization layer to the model if specified in the hyperparameters.
        """
        if self._batch_norm:
            self._model.add(BatchNormalization())

    def _append_dense_dropout_compile(self, regularizer):
        """
        Adds dense and dropout layers to the model and then compiles it.

        Args:
            regularizer (Regularizer): Regularizer to be applied to the layer.
        """
        self._model.add(Dense(
            self._output_neurons, activation=self._out_activation,
            kernel_initializer=self._kernel_initializer, kernel_regularizer=regularizer
        ))
        if self._dropout:
            self._model.add(Dropout(self._dropout))
        self._model.compile(loss=self._loss, optimizer=Adam(learning_rate=self._learn_rate), metrics=self._metrics)

    def _train_model(self):
        """
        Trains the neural network model on the provided data and hyperparameters.
        """
        history = self._model.fit(
            self._X_train, self._y_train, epochs=self._epochs, verbose=self._verbose,
            validation_split=self._validation_split, validation_data=self._validation_data,
            batch_size=self._batch_size, callbacks=[self._create_callback()]
        )
        self._print_layer_info(history, len(self._num_neurons))

    def _print_layer_info(self, history, layer_idx):
        """
        Prints the information about the trained layer, including the number of neurons
        and the best validation metric.

        Args:
            history (History): Training history from Keras model training.
            layer_idx (int or str): Index or descriptor of the layer.
        """
        best_val_stop = min(history.history[self._stop_criteria]) if self._es_mode == 'min' else max(
            history.history[self._stop_criteria])
        print(f'Layer {layer_idx}: {self._num_neurons[-1]} neurons, best {self._stop_criteria}: {best_val_stop}')

    def _reset_model(self):
        """
        Resets the neural network model and the list of neurons.
        """
        self._model = Sequential()
        self._num_neurons = []

    def build(self):
        """
        Builds and trains the complete neural network model based on the PCA specification
        and the specified hyperparameters.

        Returns:
            tuple: A tuple containing the trained Keras model and a list of the number of neurons
            for each layer.
        """
        self._reset_model()
        regularizer = self._get_regularizer()

        for layer_idx in range(self._layers):
            n_pca_components = self._build_layer(layer_idx)
            self._append_batch_norm_if_needed()
            self._model.add(Dense(
                n_pca_components, input_dim=self._X_train.shape[1], use_bias=False,
                activation=self._activation, kernel_initializer=self._kernel_initializer,
                kernel_regularizer=regularizer
            ))
            if self._dropout:
                self._model.add(Dropout(self._dropout))

        self._model.add(Dense(
            self._output_neurons, activation=self._out_activation,
            kernel_regularizer=regularizer, use_bias=False
        ))
        self._model.compile(loss=self._loss, optimizer=Adam(learning_rate=self._learn_rate), metrics=self._metrics)

        callbacks = [self._create_callback()]
        if self._custom_callback:
            callbacks.append(self._custom_callback)

        history = self._model.fit(
            self._X_train, self._y_train, epochs=self._epochs, verbose=self._verbose,
            validation_split=self._validation_split, validation_data=self._validation_data,
            batch_size=self._batch_size, callbacks=callbacks
        )
        self._print_layer_info(history, 'Final')
        return self._model, self._num_neurons
deepfastmlu/model_search/PCCDNAS.py
fabprezja/deep-fast-machine-learning-utils
from .PCCDNAS import PCCDNAS
deepfastmlu/model_search/__init__.py
fabprezja/deep-fast-machine-learning-utils
import sys
import os
sys.path.insert(0, os.path.abspath('../'))
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html

# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information

project = 'deepfastmlu'
copyright = '2023, Fabi Prezja'
author = 'Fabi Prezja'
release = '0.1'

# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

extensions = ['sphinx.ext.autodoc']
add_module_names = False
templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']



# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output

html_theme = 'sphinx_rtd_theme'
html_static_path = ['_static']

docs/conf.py
fabprezja/deep-fast-machine-learning-utils
from setuptools import setup, find_packages

setup(
    name="deepfastmlu",
    version="0.1",
    description="Machine learning utilities to help speed up your prototyping process.",
    author="Fabi Prezja",
    author_email="faprezja@fairn.fi",
    url="https://github.com/fabprezja/deep-fast-machine-learning-utils",
    packages=find_packages(),
    install_requires=[
        "numpy",
        "tensorflow>=2.3.0",
        "scikit-learn",
        "pandas",
        "seaborn",
        "matplotlib",
    ],
    extras_require={
        "gpu": ["tensorflow-gpu>=2.3.0"]
    },
    classifiers=[
        "Intended Audience :: Developers",
        "Intended Audience :: Science/Research",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
        "Programming Language :: Python :: 3.7",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
        "Programming Language :: Python :: 3.10",
    ],
    python_requires=">=3.7",
)

setup.py
fabprezja/deep-fast-machine-learning-utils