train_DL_singleSubject_5classes_RNN.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import h5py\n",
    "from scipy import stats\n",
    "import scipy.io\n",
    "import mne\n",
    "\n",
    "mne.set_log_level('error')\n",
    "\n",
    "from random import shuffle\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "from sklearn.metrics import accuracy_score\n",
    "from sklearn.model_selection import KFold\n",
    "\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "\n",
    "from torchsummary import summary\n",
    "\n",
    "import optuna\n",
    "\n",
    "\n",
    "from utils.load import Load\n",
    "from config.default import cfg\n",
    "from utils.eval import accuracy\n",
    "\n",
    "%load_ext autoreload\n",
    "%autoreload 2\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "subject_id = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "cuda\n"
     ]
    }
   ],
   "source": [
    "device_name = 'cuda' if torch.cuda.is_available() else 'cpu'\n",
    "device = torch.device(device_name)\n",
    "print(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "index (50, 30, 158, 2)\n",
      "little (50, 30, 158, 2)\n",
      "middle (50, 30, 158, 2)\n",
      "ring (50, 30, 158, 2)\n",
      "thumb (50, 30, 158, 2)\n"
     ]
    }
   ],
   "source": [
    "# Load the data  from the HDF5 file\n",
    "target_dir = 'features'\n",
    "tag = '0_25powers'\n",
    "file_path = os.path.join(target_dir, tag+'_'+cfg['subjects'][subject_id] + '.h5')\n",
    "\n",
    "\n",
    "data = {}\n",
    "with h5py.File(file_path, 'r') as h5file:\n",
    "    for key in h5file.keys():\n",
    "        data[key] = np.array(h5file[key])\n",
    "\n",
    "# Time first [Time, Channels, Features]\n",
    "for key, value in data.items():\n",
    "    data[key] = np.transpose(value, (0, 2, 1, 3))\n",
    "\n",
    "# Print the loaded data dictionary\n",
    "for key, value in data.items():\n",
    "    print(key, value.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "X = np.concatenate(list(data.values()), axis=0)\n",
    "y = np.concatenate([np.ones(data[finger].shape[0]) * i for i, finger in enumerate(data)], axis=0)\n",
    "\n",
    "# Normalize the data\n",
    "# orig_shape = X.shape\n",
    "# X = X.reshape(X.shape[0], -1)\n",
    "# scaler = StandardScaler()\n",
    "# X = scaler.fit_transform(X)\n",
    "# X = X.reshape(orig_shape)\n",
    "\n",
    "# Split the data into training and test sets\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "class CustomDataset(Dataset):\n",
    "    def __init__(self, X, y):\n",
    "        self.X = X\n",
    "        self.y = y\n",
    "\n",
    "        self.X = torch.from_numpy(X).float().to(device)\n",
    "        self.y = torch.from_numpy(y).long().to(device)\n",
    "\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.y)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        features = self.X[idx]\n",
    "        label = self.y[idx]\n",
    "        \n",
    "        return features, label\n",
    "\n",
    "train_dataset = CustomDataset(X_train, y_train)\n",
    "train_dataloader = DataLoader(train_dataset, batch_size=10, shuffle=True)\n",
    "\n",
    "test_dataset = CustomDataset(X_test, y_test)\n",
    "test_dataloader = DataLoader(test_dataset, batch_size=10, shuffle=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "torch.Size([10, 30, 158, 2])\n",
      "tensor([2, 3, 1, 0, 0, 2, 4, 0, 3, 2], device='cuda:0')\n"
     ]
    }
   ],
   "source": [
    "for features, label in train_dataloader:\n",
    "    print(features.shape)\n",
    "    print(label)\n",
    "    break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "class SingleLayerMLP(nn.Module):\n",
    "    def __init__(self, time_stamps, channels, rnn_hidden_size,  output_size, activation):\n",
    "        super(SingleLayerMLP, self).__init__()\n",
    "\n",
    "        self.channels = channels\n",
    "        self.fc1 = nn.Linear(2, 1)\n",
    "        self.rnn = nn.RNN(channels, rnn_hidden_size, batch_first=True)\n",
    "        self.lstm = nn.LSTM(channels, rnn_hidden_size, bias = False, batch_first=True, bidirectional=True)\n",
    "        self.activation = nn.ReLU()\n",
    "        self.fc2 = nn.Linear(rnn_hidden_size*2, output_size)\n",
    "        #self.fc2 = nn.Linear(rnn_hidden_size*time_stamps, output_size)\n",
    "\n",
    "        self.dropout = nn.Dropout(0.1)\n",
    "        \n",
    "        \n",
    "    def forward(self, x):\n",
    "\n",
    "        ###\n",
    "        ###Merge Mu and Beta band power features\n",
    "        ###\n",
    "        # A version\n",
    "        x = x.reshape(-1,2)\n",
    "        x = self.fc1(x)         # Apply the linear layer\n",
    "        x = x.view(-1, time_stamps, channels)         # Reshape the tensor back to its original shape: (26, 158)\n",
    "        \n",
    "        # B version\n",
    "        #x = torch.mean(x, -1)\n",
    "\n",
    "        # C version\n",
    "        #x = x.reshape(10, -1, self.channels*2)\n",
    "\n",
    "        # RNN\n",
    "        #x, hn = self.rnn(x)\n",
    "        x, (hn, cn) = self.lstm(x)\n",
    "        x = x[:, -1, :]\n",
    "       \n",
    "        \n",
    "        x = self.activation(x)\n",
    "        x = self.dropout(x)\n",
    "\n",
    "        x = self.fc2(x)\n",
    "\n",
    "        return x\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_epochs = 100\n",
    "learning_rate = 1e-3\n",
    "rnn_hidden_size = 8\n",
    "\n",
    "\n",
    "\n",
    "time_stamps = X.shape[1]\n",
    "channels = X.shape[2]\n",
    "model = SingleLayerMLP(time_stamps, channels, rnn_hidden_size, 5, nn.ReLU())\n",
    "#summary(model, input_size=(X[0].shape));\n",
    "model.to(device)\n",
    "\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "optimizer = optim.Adam(model.parameters(), lr=learning_rate)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch 10/100, Loss: 31.797123670578003, Train accuracy: 23.00%, Test accuracy: 24.00%\n",
      "Epoch 20/100, Loss: 30.235387682914734, Train accuracy: 41.50%, Test accuracy: 16.00%\n",
      "Epoch 30/100, Loss: 27.36820948123932, Train accuracy: 60.00%, Test accuracy: 20.00%\n",
      "Epoch 40/100, Loss: 24.012503504753113, Train accuracy: 69.50%, Test accuracy: 18.00%\n",
      "Epoch 50/100, Loss: 20.028084576129913, Train accuracy: 78.50%, Test accuracy: 24.00%\n",
      "Epoch 60/100, Loss: 16.078268885612488, Train accuracy: 82.00%, Test accuracy: 22.00%\n",
      "Epoch 70/100, Loss: 13.006169825792313, Train accuracy: 85.50%, Test accuracy: 20.00%\n",
      "Epoch 80/100, Loss: 11.960670351982117, Train accuracy: 90.50%, Test accuracy: 22.00%\n",
      "Epoch 90/100, Loss: 9.182256996631622, Train accuracy: 89.50%, Test accuracy: 20.00%\n",
      "Epoch 100/100, Loss: 7.874237805604935, Train accuracy: 95.50%, Test accuracy: 14.00%\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "18.0"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def train(num_epochs, verbose = False):\n",
    "    # single train\n",
    "    model.train()\n",
    "    for epoch in range(num_epochs):\n",
    "\n",
    "        epoch_loss = 0.0\n",
    "        for X, y in train_dataloader:\n",
    "            optimizer.zero_grad()\n",
    "            outputs = model(X)\n",
    "            loss = criterion(outputs, y)\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            epoch_loss += loss.item()\n",
    "\n",
    "        if  verbose and (epoch+1) % 10 == 0:\n",
    "            train_accuracy = accuracy(train_dataloader)\n",
    "            test_accuracy = accuracy(test_dataloader)\n",
    "            print(f\"Epoch {epoch + 1}/{num_epochs}, Loss: {epoch_loss}, Train accuracy: {train_accuracy:.2f}%, Test accuracy: {test_accuracy:.2f}%\")\n",
    "\n",
    "    return accuracy(test_dataloader)\n",
    "\n",
    "train(num_epochs, verbose=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def objective(trial):\n",
    "    learning_rate = trial.suggest_float(\"learning_rate\", 1e-5, 1e-1, log=True)\n",
    "    num_epochs = trial.suggest_int(\"num_epochs\", 20, 500)\n",
    "    hidden_size = trial.suggest_int(\"hidden_size\", 4, 32)\n",
    "    activation_name = trial.suggest_categorical(\"activation\", [\"relu\", \"elu\", \"leaky_relu\"])\n",
    "    optimizer = trial.suggest_categorical(\"optimizer\", [\"SGD\", \"Adam\"])\n",
    "\n",
    "    if activation_name == \"relu\":\n",
    "        activation = nn.ReLU()\n",
    "    elif activation_name == \"elu\":\n",
    "        activation = nn.ELU()\n",
    "    elif activation_name == \"leaky_relu\":\n",
    "        activation = nn.LeakyReLU()\n",
    "\n",
    "    if optimizer == \"SGD\":\n",
    "        optimizer = optim.SGD\n",
    "    elif optimizer == \"Adam\":\n",
    "        optimizer = optim.Adam\n",
    "\n",
    "\n",
    "    model = SingleLayerMLP(time_stamps=time_stamps, channels=channels, rnn_hidden_size=hidden_size, output_size=5, activation=activation)\n",
    "    model.to(device)\n",
    "\n",
    "    criterion = nn.CrossEntropyLoss()\n",
    "    optimizer = optimizer(model.parameters(), lr=learning_rate)\n",
    "    return train(num_epochs)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\u001b[32m[I 2023-04-27 16:49:28,404]\u001b[0m A new study created in memory with name: no-name-2496c584-dd02-4594-9a30-fd5a68f1fecf\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:49:45,492]\u001b[0m Trial 0 finished with value: 12.0 and parameters: {'learning_rate': 0.00778406180620629, 'num_epochs': 286, 'hidden_size': 20, 'activation': 'leaky_relu', 'optimizer': 'Adam'}. Best is trial 0 with value: 12.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:50:11,562]\u001b[0m Trial 1 finished with value: 14.000000000000002 and parameters: {'learning_rate': 0.0005476655042602315, 'num_epochs': 430, 'hidden_size': 11, 'activation': 'relu', 'optimizer': 'SGD'}. Best is trial 1 with value: 14.000000000000002.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:50:13,683]\u001b[0m Trial 2 finished with value: 16.0 and parameters: {'learning_rate': 0.014318479037639343, 'num_epochs': 36, 'hidden_size': 20, 'activation': 'elu', 'optimizer': 'SGD'}. Best is trial 2 with value: 16.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:50:17,923]\u001b[0m Trial 3 finished with value: 14.000000000000002 and parameters: {'learning_rate': 0.0004593623820428704, 'num_epochs': 60, 'hidden_size': 28, 'activation': 'leaky_relu', 'optimizer': 'SGD'}. Best is trial 2 with value: 16.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:50:37,745]\u001b[0m Trial 4 finished with value: 8.0 and parameters: {'learning_rate': 0.011436833050779128, 'num_epochs': 338, 'hidden_size': 16, 'activation': 'relu', 'optimizer': 'SGD'}. Best is trial 2 with value: 16.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:50:59,623]\u001b[0m Trial 5 finished with value: 18.0 and parameters: {'learning_rate': 1.5283688884631135e-05, 'num_epochs': 351, 'hidden_size': 18, 'activation': 'leaky_relu', 'optimizer': 'SGD'}. Best is trial 5 with value: 18.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:51:28,146]\u001b[0m Trial 6 finished with value: 14.000000000000002 and parameters: {'learning_rate': 5.843840936126841e-05, 'num_epochs': 316, 'hidden_size': 11, 'activation': 'leaky_relu', 'optimizer': 'Adam'}. Best is trial 5 with value: 18.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:51:50,093]\u001b[0m Trial 7 finished with value: 16.0 and parameters: {'learning_rate': 0.009033494542599437, 'num_epochs': 246, 'hidden_size': 31, 'activation': 'leaky_relu', 'optimizer': 'Adam'}. Best is trial 5 with value: 18.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:52:03,302]\u001b[0m Trial 8 finished with value: 18.0 and parameters: {'learning_rate': 0.0010535027394744467, 'num_epochs': 213, 'hidden_size': 17, 'activation': 'relu', 'optimizer': 'SGD'}. Best is trial 5 with value: 18.0.\u001b[0m\n",
      "\u001b[32m[I 2023-04-27 16:52:18,099]\u001b[0m Trial 9 finished with value: 16.0 and parameters: {'learning_rate': 0.00020950173453692506, 'num_epochs': 263, 'hidden_size': 26, 'activation': 'leaky_relu', 'optimizer': 'Adam'}. Best is trial 5 with value: 18.0.\u001b[0m\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Best trial params: {'learning_rate': 1.5283688884631135e-05, 'num_epochs': 351, 'hidden_size': 18, 'activation': 'leaky_relu', 'optimizer': 'SGD'}\n",
      "Best trial accuracy: 18.0%\n"
     ]
    }
   ],
   "source": [
    "# Hyperparameter tuning\n",
    "n_trials = 10\n",
    "\n",
    "study = optuna.create_study(direction=\"maximize\")\n",
    "study.optimize(lambda trial: objective(trial), n_trials=n_trials)\n",
    "\n",
    "best_trial = study.best_trial\n",
    "\n",
    "print(f'Best trial params: {best_trial.params}')\n",
    "print(f'Best trial accuracy: {best_trial.value}%')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.0"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "9cefe38f745df9e33a66570f2e5a410ba71c4ae3bf929b6ad1b474ac5f904d76"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}