import os
import numpy as np
import mne
import matplotlib.pyplot as plt
import pandas as pd
from sklearn import neighbors
from sklearn.model_selection import train_test_split
from keras.models import Sequential
from keras.layers import LSTM, Dense


class BrainWaves:
    def __init__(self):
        self.__file_num = 5
        raw_data_files = []
        self.__raw = []
        self.__brain_frames = []
        self.__chan_num = 32
        self.__KNN_fit_frames = []
        self.__prediction_frame = None
        self.__TAES_frame = None
        for i in range(self.__file_num):
            raw_data_files.append(os.path.join(r"/kaggle/input/eeg-edf/", r"bckg_00" + str(i) + r"_a_.edf"))  # Change this for Kaggle (r"/kaggle/input/eeg-edf/")
            self.__raw.append(mne.io.read_raw_edf(raw_data_files[i]))
            self.__brain_frames.append(pd.DataFrame(self.__raw[i].get_data()))
            self.__brain_frames[i] = self.__brain_frames[i].transpose()   # To get the correct dimension
            self.__brain_frames[i] = self.__brain_frames[i].iloc[1000:5000, :self.__chan_num]   # Ahhh!
  ​def print_waves(self):
        print("Just some preliminary info: ")
        for i in range(self.__file_num):
            print(self.__raw[i])
            print(self.__raw[i].info)
            print(self.__raw[i].ch_names)
            print(self.__brain_frames[i])
            print("Saving raw data ...")
            self.__brain_frames[i].to_csv(r"raw-brain-frame-subject-" + str(i+1) + r".csv")
        print("\n---------------------------------------------------\n")
​
    def plot_waves(self):
        for i in range(self.__file_num):
            self.__raw[i].plot(color='r', start=50)  # Not as interesting before 50
            plt.show()
            self.__raw[i].plot_psd()    # Freq-power spectrum
            plt.show()
            self.__brain_frames[i].plot()
            plt.grid()
            plt.show()
​
    def fit_KNN(self):
        print("Fitting KNN ...")
        for i in range(self.__file_num):
            X = list(range(len(self.__brain_frames[i].index)))
            X = np.array(X)
            X = X.reshape(-1, 1)
            y, neighbour, y_predicted = [], [], []
            for j in range(self.__chan_num):
                y.append(self.__brain_frames[i].iloc[:, j].to_numpy())
                neighbour.append(neighbors.KNeighborsRegressor(n_neighbors=3))
                neighbour[j].fit(X, y[j])
                y_predicted.append(neighbour[j].predict(X))
            self.__KNN_fit_frames.append(pd.DataFrame(y_predicted))
            self.__KNN_fit_frames[i] = self.__KNN_fit_frames[i].transpose()
            print("Saving KNN fits ...")
            self.__KNN_fit_frames[i].to_csv(r"KNN-fit-frame-subject-" + str(i+1) + r".csv")
        print("\n---------------------------------------------------\n")
​
    def split_sequences(self, sequences, n_steps):
        X, y = list(), list()
        for i in range(len(sequences)):
            end_ix = i + n_steps
            if end_ix > len(sequences) - 1:
                break
            seq_x, seq_y = sequences[i:end_ix, :], sequences[end_ix, :]
            X.append(seq_x)
            y.append(seq_y)
        return np.array(X), np.array(y)
​
    def save_data(self, X, y, n_steps, filename):
        print("Saving data ...")
        df = pd.DataFrame()
        for i in range(self.__chan_num):
            for j in range(n_steps):
                df["channel: " + str(i+1), "X_" + str(j)] = X[:, j, i]
            df["channel: " + str(i+1), "y"] = y[:, i]
        df.to_csv(filename)
        print("\n---------------------------------------------------\n")
​
    def train_predictions(self):
        print("Starting prediction routine ...")
        working_frame = pd.concat(self.__brain_frames, ignore_index=True)   # Change this for KNN before
        train, test = train_test_split(working_frame, test_size=0.20, shuffle=True)
        n_steps = 30
        n_features = self.__chan_num
        X, y = self.split_sequences(train.to_numpy(), n_steps)
        self.save_data(X, y, n_steps, "training-data.csv")
        print("Now training model ...")
        model = Sequential()
        model.add(LSTM(400, activation="relu", return_sequences=True, input_shape=(n_steps, n_features)))
        model.add(LSTM(600, activation="relu"))
        model.add(Dense(n_features))
        model.compile(optimizer="adam", loss="mse", metrics=['accuracy'])
        model.fit(X, y, epochs= 100, verbose=1, validation_data=(X,y))
        X, y = self.split_sequences(test.to_numpy(), n_steps)
        self.save_data(X, y, n_steps, "testing-data.csv")
        print("Now making predictions ...")
        y_hat = list()
        for i in range(len(X)):
            x_input = X[i]
            x_input = x_input.reshape((1, n_steps, n_features))
            y_hat.append(model.predict(x_input, verbose=1))
        y_hat = np.array(y_hat)
        y_hat = y_hat.squeeze(axis=1)
        self.__prediction_frame = pd.DataFrame()
        for i in range(self.__chan_num):
            self.__prediction_frame.loc[:, "predicted_values_channel: " + str(i+1)] = y_hat[:, i]
            self.__prediction_frame.loc[:, "actual_values_channel: " + str(i+1)] = y[:, i]
        self.__prediction_frame.to_csv("results.csv")
        print("\n---------------------------------------------------\n")
​
    def show_predictions(self):
        for i in range(self.__chan_num):
            self.__prediction_frame.loc[:, "actual_values_channel: " + str(i+1)].plot()
            self.__prediction_frame.loc[:, "predicted_values_channel: " + str(i+1)].plot(linestyle="dashed")
            plt.grid()
            plt.show()
​
    def calc_TAES(self):
        print("Calculating TAES ...")
        self.__TAES_frame = pd.DataFrame()
        TAES = [0] * self.__chan_num
        for i in range(self.__chan_num):
            y_hat = self.__prediction_frame.loc[:, "predicted_values_channel: " + str(i+1)].to_numpy()
            y = self.__prediction_frame.loc[:, "actual_values_channel: " + str(i+1)].to_numpy()
            FP = 0
            TN = 0
            for j in range(len(y_hat)):
                if y_hat[j] > 0 and y[j] < 0:
                    FP += 1
                if y[j] < 0 and y_hat[j] < 0:
                    TN += 1
            TNR = TN / len(y_hat)
            TAES[i] = 1 - TNR
            self.__TAES_frame.loc[:, "Channel: " + str(i+1)] = FP, TN, TNR, len(y_hat), TAES[i]
        self.__TAES_frame.rename(index={0: "FP", 1: "TN", 2: "TNR", 3: "N", 4: "TAES"}, inplace=True)
        self.__TAES_frame.to_csv("TAES-metrics.csv")
        print("\n---------------------------------------------------\n")
        
        def main():
    seek = BrainWaves()    # Change this for Kaggle
    print("\n---------------------------------------------------\n")
    print("Starting ...")
    seek.print_waves()
    #seek.plot_waves()
    seek.fit_KNN()
    seek.train_predictions()
    seek.show_predictions()
    #seek.calc_TAES()
    print("Program finished successfully.")
    print("\n---------------------------------------------------\n")


main()
​ 
by

Python Online Compiler

Write, Run & Share Python code online using OneCompiler's Python online compiler for free. It's one of the robust, feature-rich online compilers for python language. Getting started with the OneCompiler's Python editor is easy and fast. The editor shows sample boilerplate code when you choose language as Python or Python2. OneCompiler also has reference programs, where you can look for the sample code and start coding.

Taking inputs (stdin)

OneCompiler's python online editor supports stdin and users can give inputs to programs using the STDIN textbox under the I/O tab. Following is a sample python program which takes name as input and print your name with hello.

import sys
name = sys.stdin.readline()
print("Hello "+ name)

About Python

Python is a very popular general-purpose programming language which was created by Guido van Rossum, and released in 1991. It is very popular for web development and you can build almost anything like mobile apps, web apps, tools, data analytics, machine learning etc. It is designed to be simple and easy like english language. It's is highly productive and efficient making it a very popular language.

Tutorial & Syntax help

Loops

1. If-Else:

When ever you want to perform a set of operations based on a condition IF-ELSE is used.

if conditional-expression
    #code
elif conditional-expression
    #code
else:
    #code

Note:

Indentation is very important in Python, make sure the indentation is followed correctly

2. For:

For loop is used to iterate over arrays(list, tuple, set, dictionary) or strings.

Example:

mylist=("Iphone","Pixel","Samsung")
for i in mylist:
    print(i)

3. While:

While is also used to iterate a set of statements based on a condition. Usually while is preferred when number of iterations are not known in advance.

while condition  
    #code 

Collections

There are four types of collections in Python.

1. List:

List is a collection which is ordered and can be changed. Lists are specified in square brackets.

Example:

mylist=["iPhone","Pixel","Samsung"]
print(mylist)

2. Tuple:

Tuple is a collection which is ordered and can not be changed. Tuples are specified in round brackets.

Example:

myTuple=("iPhone","Pixel","Samsung")
print(myTuple)

Below throws an error if you assign another value to tuple again.

myTuple=("iPhone","Pixel","Samsung")
print(myTuple)
myTuple[1]="onePlus"
print(myTuple)

3. Set:

Set is a collection which is unordered and unindexed. Sets are specified in curly brackets.

Example:

myset{"iPhone","Pixel","Samsung"}
print{myset}

4. Dictionary:

Dictionary is a collection of key value pairs which is unordered, can be changed, and indexed. They are written in curly brackets with key - value pairs.

Example:

mydict = {
    "brand" :"iPhone",
    "model": "iPhone 11"
}
print(mydict)