statistical_feature_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "import pickle\n",
    "\n",
    "from data_utils.data_utils import (\n",
    "    load_data_from_dir,\n",
    "    AUDIO_BLOCKS,\n",
    "    get_block_raw_data_by_marker,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Extract statistical features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### CHANGE ME\n",
    "marker = 'EGG'\n",
    "channeel_no = 0\n",
    "data_dir = \"../data/\"  # Replace with your own data dir\n",
    "###\n",
    "\n",
    "data_raw, subject_list = [], []\n",
    "subject_to_data = {}\n",
    "for i, d in enumerate(os.listdir(data_dir)):\n",
    "    dir_name = data_dir + d\n",
    "    if not os.path.isdir(dir_name):\n",
    "        continue\n",
    "\n",
    "    data = load_data_from_dir(dir_name)\n",
    "    block_to_data = get_block_raw_data_by_marker(\n",
    "        data, AUDIO_BLOCKS, marker, channeel_no,\n",
    "    )\n",
    "    data_raw.append(block_to_data)\n",
    "    subject_list.append(d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from features.stats_features import (\n",
    "    FEATURE_TO_FUNC,\n",
    "    StatsFeature,\n",
    ")\n",
    "\n",
    "\n",
    "def _extact_block_features(raw_data, feature_key):\n",
    "    features = []\n",
    "    for b in AUDIO_BLOCKS:\n",
    "        for bd in raw_data[b]:\n",
    "            f = FEATURE_TO_FUNC[feature_key](bd)\n",
    "            features.append(f)\n",
    "    return np.array(features)\n",
    "\n",
    "\n",
    "## CHANGE ME\n",
    "feature_list = [\n",
    "    StatsFeature.MAXIM,\n",
    "    StatsFeature.MINIM,\n",
    "]\n",
    "###\n",
    "\n",
    "all_features = []\n",
    "for i in range(len(subject_list)):\n",
    "    features = [\n",
    "        _extact_block_features(data_raw[i], f)\n",
    "        for f in feature_list\n",
    "    ]\n",
    "    all_features.append(np.swapaxes(features, 0, 1))\n",
    "\n",
    "all_features = np.array(all_features)\n",
    "all_features.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load behavioral labels\n",
    "Please look into eeg_features_analysis.ipynb to see how to extract labels"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read labels pkl file\n",
    "with open(\"./data/replace_me.pkl\", \"rb\") as fp:\n",
    "    behavioral_labels = pickle.load(fp)\n",
    "\n",
    "valence_labels, arousal_labels, label_thresholds = (\n",
    "    behavioral_labels[\"valence_labels\"],\n",
    "    behavioral_labels[\"arousal_labels\"],\n",
    "    behavioral_labels[\"label_thresholds\"],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Optional inspect feature correlation with user rating"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy import stats\n",
    "\n",
    "feature_key=StatsFeature.ABS_DIFF\n",
    "for l, labels in {\"valence\": valence_labels, \"arousal\": arousal_labels}.items():\n",
    "    n_row, n_col = (5, 8)\n",
    "    fig, axes = plt.subplots(\n",
    "        nrows=n_row,\n",
    "        sharey=False,\n",
    "        ncols=n_col,\n",
    "        figsize=(n_col * 3, n_row * 3),\n",
    "    )\n",
    "\n",
    "    for i, ax in enumerate(axes.flat):\n",
    "        s = str(subject_list[i])\n",
    "\n",
    "        feature = _extact_block_features(data_raw[i], feature_key)\n",
    "        r_v, p_v = stats.pearsonr(labels[i], feature)\n",
    "\n",
    "        color = \"red\" if p_v < 0.05 else \"grey\"\n",
    "        result = pd.DataFrame({\"user rating\": labels[i], \"feature\": feature})\n",
    "        g1 = sns.regplot(data=result, x=\"user rating\", y=\"feature\", ax=ax, color=color)\n",
    "        g1.set(xlabel=None, ylabel=None)\n",
    "        g1.set_title(f\"{s} r:{r_v:2.4f}, p:{p_v:2.4f}\", fontsize=12, color=color)\n",
    "\n",
    "    fig.suptitle(f\"{feature_key.name} vs user rating - {l}\", y=1, size=24)\n",
    "    fig.tight_layout(pad=1.8)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model training with PCA + KNN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from training_utils.dataset import get_consecutive_validation_indexes\n",
    "from training_utils.dataset import DatasetBuilder\n",
    "\n",
    "n_trial_per_block = 13\n",
    "n_step_trial = 3\n",
    "val_indexes = [\n",
    "    get_consecutive_validation_indexes(\n",
    "        len(valence_labels[0]), len(AUDIO_BLOCKS), 1, i, n_step_trial\n",
    "    )\n",
    "    for i in range(1, n_trial_per_block, n_step_trial)\n",
    "]\n",
    "print(len(val_indexes), val_indexes)\n",
    "\n",
    "dataset_builder = DatasetBuilder(len(valence_labels[0]), val_indexes_group=val_indexes)\n",
    "len(valence_labels[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cross validation\n",
    "Check out eeg_feature_analysis.ipynb for further evaluation helpers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from training_utils.training import decode_marker_data, get_metadata\n",
    "\n",
    "subject_accuracy_summary = {\n",
    "    \"subject\": [],\n",
    "    \"channel\": [],\n",
    "    \"feature\": [],\n",
    "    \"label_type\": [],\n",
    "    \"cv_scores\": [],\n",
    "    \"cv_mean_score\": [],\n",
    "}\n",
    "\n",
    "###CHANGE ME####\n",
    "method = \"PCA\"\n",
    "feature_name = \"\"\n",
    "output_dim = 4\n",
    "###############\n",
    "\n",
    "subject_to_embedding = {s: {\"valence\": [], \"arousal\": []} for s in subject_list}\n",
    "\n",
    "for idx in range(len(subject_list)):\n",
    "    subj = subject_list[idx]\n",
    "    print(\"decoding subject...\", subj)\n",
    "\n",
    "    v_thred, a_thred = label_thresholds[idx]\n",
    "    for lt in [\"valence\", \"arousal\"]:\n",
    "        labels = valence_labels[idx] if lt == \"valence\" else arousal_labels[idx]\n",
    "        thred = v_thred if lt == \"valence\" else a_thred\n",
    "\n",
    "        dataset_dict = {\n",
    "            marker: {feature_name: dataset_builder.train_test_split(features, labels)}\n",
    "        }\n",
    "\n",
    "        subject_to_embedding[subj][lt], accuracy = decode_marker_data(\n",
    "            dataset_dict,\n",
    "            lt,\n",
    "            v_thred,\n",
    "            a_thred,\n",
    "            method,\n",
    "            output_dim,\n",
    "            thred,\n",
    "        )\n",
    "\n",
    "        all_channels, all_feature_name, cv_scores = get_metadata(accuracy)\n",
    "\n",
    "        subject_accuracy_summary[\"subject\"].extend([subj] * len(all_feature_name))\n",
    "        subject_accuracy_summary[\"channel\"].extend(all_channels)\n",
    "        subject_accuracy_summary[\"feature\"].extend(all_feature_name)\n",
    "        subject_accuracy_summary[\"cv_mean_score\"].extend([round(np.mean(cv_scores), 2)])\n",
    "        subject_accuracy_summary[\"cv_scores\"].extend(cv_scores)\n",
    "        subject_accuracy_summary[\"label_type\"].extend([lt] * len(all_feature_name))\n",
    "\n",
    "subject_accuracy_summary = pd.DataFrame(subject_accuracy_summary)\n",
    "subject_accuracy_summary[\"subject\"] = subject_accuracy_summary[\"subject\"].astype(int)"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}