BiHDM_example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from sklearn.base import clone\n",
    "from sklearn.model_selection import cross_val_score, StratifiedKFold\n",
    "\n",
    "from BiHDM import BiHDMClassifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define 64 EEG channels using 10-20 standard (on a 64-channel BioSemi cap)\n",
    "ch_names = ['Fp1', 'AF7', 'AF3', 'F1', 'F3', 'F5', 'F7', 'FT7', 'FC5', 'FC3', 'FC1', \n",
    "            'C1', 'C3', 'C5', 'T7', 'TP7', 'CP5', 'CP3', 'CP1', 'P1', 'P3', 'P5', \n",
    "            'P7', 'P9', 'PO7', 'PO3', 'O1', 'Iz', 'Oz', 'POz', 'Pz', 'CPz', 'Fpz', \n",
    "            'Fp2', 'AF8', 'AF4', 'AFz', 'Fz', 'F2', 'F4', 'F6', 'F8', 'FT8', 'FC6', \n",
    "            'FC4', 'FC2', 'FCz', 'Cz', 'C2', 'C4', 'C6', 'T8', 'TP8', 'CP6', 'CP4', \n",
    "            'CP2', 'P2', 'P4', 'P6', 'P8', 'P10', 'PO8', 'PO4', 'O2']\n",
    "\n",
    "lh_chs = ['Fp1', 'AF7', 'AF3', 'F7', 'F5', 'F3', 'F1', 'FT7', 'FC5', 'FC3', 'FC1', \n",
    "          'T7', 'C5', 'C3', 'C1', 'TP7', 'CP5', 'CP3', 'CP1', 'P7', 'P5', 'P3', 'P1', \n",
    "          'PO7', 'PO3', 'O1']\n",
    "rh_chs = ['Fp2', 'AF8', 'AF4', 'F8', 'F6', 'F4', 'F2', 'FT8', 'FC6', 'FC4', 'FC2', \n",
    "          'T8', 'C6', 'C4', 'C2','TP8', 'CP6', 'CP4', 'CP2', 'P8', 'P6', 'P4', 'P2', \n",
    "          'PO8', 'PO4', 'O2']\n",
    "lv_chs = ['Fp1', 'AF7', 'F7', 'FT7', 'T7', 'TP7', 'P7', 'PO7', 'AF3', 'F5', 'FC5', \n",
    "          'C5', 'CP5', 'P5', 'O1', 'F3', 'FC3', 'C3', 'CP3', 'P3', 'PO3', 'F1', 'FC1', \n",
    "          'C1', 'CP1', 'P1']\n",
    "rv_chs = ['Fp2', 'AF8', 'F8', 'FT8', 'T8', 'TP8', 'P8', 'PO8', 'AF4', 'F6', 'FC6', \n",
    "          'C6', 'CP6', 'P6', 'O2', 'F4', 'FC4', 'C4', 'CP4', 'P4', 'PO4', 'F2', 'FC2', \n",
    "          'C2', 'CP2', 'P2']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate some data for classification\n",
    "X = np.ones((1000, 64, 5)) # 1000 samples x 64 channels x 5 bands per channel (delta, theta, alpha, beta, gamma)\n",
    "y = np.repeat([0,1], 500)\n",
    "\n",
    "# Let's simulate a frontal alpha-asymmetry for the classifier to learn from\n",
    "left_frontal_chs = ['Fp1', 'AF3', 'AF7', 'F1', 'F3', 'F5', 'FC3', 'FC1']\n",
    "X[:500,np.isin(ch_names, left_frontal_chs),2] -= 1\n",
    "\n",
    "# And let's add some gaussian noise\n",
    "rng = np.random.default_rng(42)\n",
    "X += rng.normal(scale=0.5, size=X.shape)\n",
    "\n",
    "# Reshape X to meet sklearn standard\n",
    "X = X.reshape(1000, -1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.783\n",
      "[0.745 0.76  0.775 0.805 0.83 ]\n"
     ]
    }
   ],
   "source": [
    "clf = BiHDMClassifier(ch_names, lh_chs, rh_chs, lv_chs, rv_chs, \n",
    "                    d_stream=32, d_pair=32, d_global=32, d_out=16, \n",
    "                    k=6, a=0.01, pairwise_operation='subtraction', \n",
    "                    rnn_stream_kwargs={}, rnn_global_kwargs={}, \n",
    "                    loss='NLLLoss', domain_loss='NLLLoss', optimizer='SGD', lr=0.003, \n",
    "                    epochs=8, batch_size=200, loss_kwargs={}, domain_loss_kwargs={}, \n",
    "                    optimizer_kwargs=dict(momentum=0.9, weight_decay=0.95),\n",
    "                    random_state=42, use_gpu=True, verbose=False)\n",
    "\n",
    "# first let's test the model without performing the domain adversarial strategy\n",
    "scores = cross_val_score(clf, X, y)\n",
    "print(np.mean(scores))\n",
    "print(scores)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then let's test the model again, but this time with the domain adversarial strategy."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def custom_cross_val_score(clf, X, y):\n",
    "    cv = StratifiedKFold()\n",
    "    scores = []\n",
    "    for train_index, test_index in cv.split(X, y):\n",
    "        cloned_clf = clone(clf)\n",
    "        X_train = X[train_index]\n",
    "        y_train = y[train_index]\n",
    "        X_test = X[test_index]\n",
    "        y_test = y[test_index]\n",
    "        \n",
    "        # provide X_test for performing the domain adversarial strategy\n",
    "        cloned_clf.fit(X_train, y_train, X_test=X_test)\n",
    "        score = cloned_clf.score(X_test, y_test)\n",
    "        scores.append(score)\n",
    "    return np.array(scores)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.7870000000000001\n",
      "[0.795 0.78  0.835 0.85  0.675]\n"
     ]
    }
   ],
   "source": [
    "scores_d = custom_cross_val_score(clf, X, y)\n",
    "print(np.mean(scores_d))\n",
    "print(scores_d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "prosody_1",
   "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.9.15"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "3f38051f1ddded3e326d0b827e329a10034216e79b2f401ba9b72226d19a5307"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}