<jupyter_start><jupyter_text>Data Preprocessing  Preprocessing steps : 1. **Read data**2. **Power transform**3. **Data Normalization**4. **Reshaping**5. **Train/test/validation split**<jupyter_code>import os
import pandas as pd

import matplotlib.pyplot as plt
import seaborn as sns 
import numpy as np
from scipy.stats import normaltest

sns.set()

#path
import sys
sys.path.append("..")

import config
from src.features.build_features import *
from src.utils.preprocessing import save_preprocessing_object
filename = "mnist-64s.csv"
csv_path = os.path.join(config.RAW_EEG_DIR,filename)
csv_path
df = pd.read_csv(csv_path,index_col=0)
df.head()
df.drop(labels="mnist_class",axis=1,inplace=True,errors="ignore")
df.head()<jupyter_output><empty_output><jupyter_text>Transform distributionAs we know from data analysis VAE needs data that comes from Gausian distribution. In order to complete this assumption I can use two type os solutions : - Data Science : Use data transformation and map the data into Gausian distribution based on KDE- Deep learning : And random noise to intermediate layers during network initialization<jupyter_code>sensors_list = df.columns[:14]
_ = df[sensors_list].hist(figsize=(15,15),bins=50)
_ = plt.suptitle("Before transformation",size = 20)
_ = df[sensors_list].plot(figsize=(20,8)).legend(bbox_to_anchor=(1.1,1.0))
_ = plt.title("EEG Before transformation",size = 20)
_ = plt.xlabel("Samples")
_ = plt.ylabel("Voltage")<jupyter_output><empty_output><jupyter_text>Applying quantile transformation  For this task **QuantileTransformer** was applied.This scaller a non-linear transformation such that the probability density function of each feature will be mapped to a normal distribution distribution.As RobustScaler, QuantileTransformer is robust to outliers in the sense that adding or removing outliers in the training set will yield approximately the same transformation on held out data<jupyter_code>"""
The transformation is applied on each feature independently.
First an estimate of the cumulative distribution function of a feature is used to map the original values to a uniform distribution.
The obtained values are then mapped to the desired output distribution using the associated quantile function.
"""
df[sensors_list] = quantile_transformer(df[sensors_list])
_ = df[sensors_list].hist(figsize=(15,15),bins=50)
_ = plt.suptitle("After transformation",size=20)
_ = df[sensors_list].plot(figsize=(20,8)).legend(bbox_to_anchor=(1.1,1.0))
_ = plt.title("EEG After transformation",size = 20)
_ = plt.xlabel("Samples")
_ = plt.ylabel("Voltage")<jupyter_output><empty_output><jupyter_text>Check hypothesisTo check if the electrode value distribution has a gauss distribution we need to perform statistical test.For this particular situation **D’Agostino’s K^2 test** was selectedThe D’Agostino’s K^2 test calculates summary statistics from the data, namely kurtosis and skewness, to determine if the data distribution departs from the normal distribution, named for <NAME>.Test Indicators : * **Skew** is a quantification of how much a distribution is pushed left or right, a measure of asymmetry in the distribution.* **Kurtosis** quantifies how much of the distribution is in the tail. It is a simple and commonly used statistical test for normality.Hypothesis :        * **H0**: the sample has a Gaussian distribution.   * **H1**: the sample does not have a Gaussian distribution.<jupyter_code>stat,p_list = normaltest(df[sensors_list].values)
for p in range(len(p_list)):
    val = p_list[p]
    sensor = df[sensors_list[p]].name
    if p > 0.05:
        print(f'Sensor {sensor} probably Gaussian')
    else:
        print(f'Sensor {sensor} Probably not Gaussian')<jupyter_output>Sensor AF3 Probably not Gaussian
Sensor F7 probably Gaussian
Sensor F3 probably Gaussian
Sensor FC5 probably Gaussian
Sensor T7 probably Gaussian
Sensor P7 probably Gaussian
Sensor O1 probably Gaussian
Sensor O2 probably Gaussian
Sensor P8 probably Gaussian
Sensor T8 probably Gaussian
Sensor FC6 probably Gaussian
Sensor F4 probably Gaussian
Sensor F8 probably Gaussian
Sensor AF4 probably Gaussian<jupyter_text>Data normalization  EEG The basic assuption of the analysis was not to remove outliers because its not neccesary for deep learning architectures.In order to deal with problem we need to select appriopriate scalling method immune for outliers.Scalers not immune for outliers * **StandardScaler*** **MinMaxScaler** * **MaxAbsScaler**Scalers not immune for outliers : * **Robust scaler**Centering and scaling statistics of this scaler are **based on percentiles** and are therefore not influenced by a few number of very large marginal outliers. The problem with robust scaler is related with range of values. I need to transform data into uniform distribion. In order to do that i need to use  :<jupyter_code>df[sensors_list], scaler = normalize_min_max(df[sensors_list])
df.head()
# as you can see during normalization step I use min max scaller 
df[sensors_list].agg(["min","max"])
_ = df[sensors_list].hist(figsize=(15,15),bins=50)
#Sprawdzić
_ = df[sensors_list].plot(figsize=(20,8)).legend(bbox_to_anchor=(1.1,1.0))
_ = plt.title("EEG After scaling",size = 20)
_ = plt.xlabel("Samples")
_ = plt.ylabel("Voltage")<jupyter_output><empty_output><jupyter_text>Save scaler<jupyter_code># save scaler for new samples
scaling_object = scaler
filepath = os.path.join(config.MODEL_DIR,"normalizer.h5")
save_preprocessing_object(scaling_object,filepath)<jupyter_output><empty_output><jupyter_text>Visual Stimulus<jupyter_code>from keras.datasets import mnist
(x_train, y_train), (_,_) = mnist.load_data()
x_train = x_train[0:1202]
y_train = y_train[0:1202]
x_train = x_train.astype("float")/255.0
n_examples = 3
plt.figure(figsize=(15,5))

for i in range(n_examples):
    plt.subplot(1,3,i+1)
    plt.imshow(x_train[i],cmap=plt.cm.binary)
    plt.axis('off')

plt.suptitle("Visual stimulus normalized ",size = 20)
plt.grid(False)
plt.colorbar()<jupyter_output><empty_output><jupyter_text>Data reshapingDuring this step I need to reshape data form suitable for Variational autoencoder. In order to do this I need to :1. Batch EEG samples based on visual stimuly. (64 samples 1 image)2. Reshape eeg and img to the same shape 2. Flatten EEG signal (4 autoencoder input)3. Flatten IMG samples (4 autoencoder input)4. Perform train/test/validation split<jupyter_code>mnist_indexes = df.mnist_index.value_counts()
eeg_batched = [df[df["mnist_index"]==i][sensors_list].values for i in range(len(mnist_indexes))]
images_batched = [x_train[i] for i in range(len(mnist_indexes))]
df2 = pd.DataFrame({"eeg" : eeg_batched,
                   "img" : images_batched})
df2.head()
import cv2
# resize with interpolation to prefered shape
df2["eeg"] = df2["eeg"].apply(lambda x : cv2.resize(x,(30,30)))
df2["img"] = df2["img"].apply(lambda x : cv2.resize(x,(30,30)))
jeb
# Flatten to VAE Dense input
df2["eeg"] = df2["eeg"].apply(lambda x : x.flatten())
df2["img"] = df2["img"].apply(lambda x : x.flatten())<jupyter_output><empty_output><jupyter_text>Train-test-split<jupyter_code>from sklearn.model_selection import train_test_split
X = df2.eeg.values
y = df2.img.values
#add dimension 
X = np.array([X[i] for i in range(len(X))]).squeeze()
y = np.array([y[i] for i in range(len(X))]).squeeze()
print(f"Features shape : {X.shape}")
print(f"Outputs  shape : {y.shape}")
def train_test_validation_split(X,y,train_ratio = 0.75,test_ratio = 0.15,validation_ratio = 0.10 ):
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=1 - train_ratio)
    # train is now 75% of the entire data set
    # the _junk suffix means that we drop that variable completely

    X_val, X_test, y_val, y_test = train_test_split(X_test, y_test, test_size=test_ratio/(test_ratio + validation_ratio))
    # test is now 10% of the initial data set
    # validation is now 15% of the initial data set

    return X_train,X_test,X_val,y_train,y_test,y_val
X_train,X_test,X_val,y_train,y_test,y_val = train_test_validation_split(X,y)
print("Shapes info : \n")
print(f"X_train shape : {X_train.shape}")
print(f"X_test shape : {X_test.shape}")
print(f"X_valid shape : {X_val.shape}")

print(f"y_train shape : {y_train.shape}")
print(f"y_test shape : {y_test.shape}")
print(f"X_valid shape : {y_val.shape}")

np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"X_train"),X_train)
np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"X_test"),X_test)
np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"X_valid"),X_val)

np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"y_train"),y_train)
np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"y_test"),y_test)
np.save(os.path.join(config.DATA_PREPROCESSED_DIR,"y_valid"),y_val)
print("Done")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11054072553045859, 'fraction_numerical': 0.01756787588409765, 'mean_word_length': 6.162581699346405, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 38, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '612725', 'license': 'MIT', 'n_tokens_mistral': 2984, 'n_tokens_neox': 2885, 'n_words': 820, 'path': 'notebooks/2.0_MikeG27_Preprocessing.ipynb', 'repo_name': 'MikeG27/BCI_Image_Generation', 'size': '8776'}