6 - Classification using a Bayesian Gaussian Mixture.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Classification using Bayesian Gaussian Mixture\n",
    "Even though the results of the previus classifier are not too sensible to different choices for the number of mixtures  a little drawback of the that classifier is that we need to specify the number of components in the mixture models. Here we present a similar approach that instead of fixing the number of components and brutally applying expectation-maximization, only fixes tha maximum number of components and then adds a regularization by integrating information from prior distributions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#load libraries\n",
    "import matplotlib\n",
    "import matplotlib.pyplot as plt\n",
    "import sklearn\n",
    "import sklearn.discriminant_analysis\n",
    "import sklearn.mixture\n",
    "import librosa.display\n",
    "import numpy\n",
    "import random\n",
    "import pandas\n",
    "import seaborn\n",
    "import json\n",
    "import operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load Features\n",
    "After the extraction of the features using the code provided in **Feature_Extraction.ipynb** we can simply load them from the .txt files (this will save time since feature extration can take a while)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('x.txt', 'r') as filehandle:\n",
    "    y = json.load(filehandle)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('SR.txt', 'r') as filehandle:\n",
    "    SR = json.load(filehandle)\n",
    "with open('MFCC.txt', 'r') as filehandle:\n",
    "    MFCC = json.load(filehandle)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "##convert mfccs from list to numpy array that are faster to work with\n",
    "x={}\n",
    "for sr in SR:\n",
    "    x[sr]={}\n",
    "    for n_mfcc in MFCC:\n",
    "        x[sr][n_mfcc]=[]\n",
    "            \n",
    "for sr in y:\n",
    "    for n_mfcc in y[sr]:\n",
    "        for t in y[sr][n_mfcc]:\n",
    "            x[int(sr)][int(n_mfcc)].append({\"artist\": t[\"artist\"],\n",
    "                                            \"song\": t[\"song\"],\n",
    "                                            \"mfcc\":numpy.asarray(t[\"mfcc\"])})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Balanced Train-Test split\n",
    "Now we create a random, but balanced, train-test split, meaning that we control randomness in a way to use for training the model the same number of songs per artist (around 3/4 for train and 1/4 for test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create list of artists\n",
    "artistlist=[]\n",
    "for t in x[SR[0]][MFCC[0]]:\n",
    "    if t[\"artist\"] not in artistlist:\n",
    "        artistlist.append(t[\"artist\"])\n",
    "artistlist.sort()\n",
    "#print(artistlist)\n",
    "\n",
    "\n",
    "#count number of songs for each artist\n",
    "songcount={}\n",
    "for t in x[SR[0]][MFCC[0]]:\n",
    "    if t[\"artist\"] not in songcount:\n",
    "        songcount[t[\"artist\"]] = 1\n",
    "    else:\n",
    "        songcount[t[\"artist\"]]+=1\n",
    "#print(songcount)\n",
    "\n",
    "\n",
    "#find number of songs for the artist having the smallest number of songs\n",
    "artist_with_min = min(songcount.keys(), key=(lambda k: songcount[k]))\n",
    "min_song=songcount[artist_with_min]\n",
    "\n",
    "\n",
    "#number of song per artist to use as train\n",
    "N=int(min_song*3/4)\n",
    "#print(N)\n",
    "\n",
    "\n",
    "#Create a list of songs per each artist\n",
    "songs={}\n",
    "for t in x[SR[0]][MFCC[0]]:\n",
    "    if t[\"artist\"] not in songs:\n",
    "        songs[t[\"artist\"]]=[]\n",
    "    songs[t[\"artist\"]].append(t[\"song\"])\n",
    "#print(songs)\n",
    "\n",
    "###create split\n",
    "split={}\n",
    "for artist in artistlist:\n",
    "    split[artist]={}\n",
    "    split[artist][\"train\"],split[artist][\"test\"]=sklearn.model_selection.train_test_split(songs[artist]\n",
    "                                                                                          ,train_size=N\n",
    "                                                                                          ,random_state=1)\n",
    "songtrain=[]\n",
    "songtest=[]\n",
    "for artist in artistlist:\n",
    "    songtrain = songtrain + split[artist][\"train\"]\n",
    "    songtest = songtest + split[artist][\"test\"]\n",
    "                     \n",
    "\n",
    "#print(songtrain)\n",
    "#print(songtest)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "##create dictionary such that to each artist correspond the list of its real songs\n",
    "##this will be usefull to compare predictions with real values\n",
    "song_and_artist_test={}\n",
    "for t in x[2500][4]:\n",
    "    if t[\"artist\"] not in song_and_artist_test:\n",
    "        song_and_artist_test[t[\"artist\"]] = []\n",
    "    if t[\"song\"] in songtest:\n",
    "        song_and_artist_test[t[\"artist\"]].append(t[\"song\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Classification using a BGM to model each artist:\n",
    "Here we run the model letting vary the \"quality\" of features extracted."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#SR=[2500]    #to test\n",
    "#MFCC=[4]     #to test\n",
    "\n",
    "\n",
    "#initialize CM, this will store the confusion matrices of all models\n",
    "CM={}\n",
    "for sr in SR:\n",
    "    CM[sr]={}\n",
    "    for n_mfcc in MFCC:\n",
    "        CM[sr][n_mfcc]=0\n",
    "\n",
    "#loop over sampling rates and number of mfcc\n",
    "for sr in SR:\n",
    "    for n_mfcc in MFCC:\n",
    "        print((sr,n_mfcc))\n",
    "\n",
    "        \n",
    "        #train a BGM for each artist\n",
    "        bgmdict={}\n",
    "        for artist in artistlist:\n",
    "            features_train=numpy.empty((1,n_mfcc))\n",
    "            for t in x[sr][n_mfcc]:\n",
    "                if t[\"artist\"]==artist:\n",
    "                    if t[\"song\"] in songtrain:\n",
    "                        features_train=numpy.vstack((features_train, t[\"mfcc\"]))\n",
    "            features_train=numpy.delete(features_train,0,0)\n",
    "            bgm = sklearn.mixture.BayesianGaussianMixture(n_components=64, covariance_type='diag'\n",
    "                                                          ,weight_concentration_prior_type='dirichlet_distribution'\n",
    "                                                          ,max_iter=30,tol=0.01)\n",
    "            bgm.fit(features_train)\n",
    "            bgmdict[artist]=bgm    \n",
    "\n",
    "            \n",
    "        #create a matrix with test features and also ordered labels for song and arstist\n",
    "        features_test=numpy.empty((1,n_mfcc))\n",
    "        artist_labels_test=[]\n",
    "        song_labels_test=[]\n",
    "        for t in x[sr][n_mfcc]:\n",
    "            if t[\"song\"] in songtest:\n",
    "                features_test=numpy.vstack((features_test, t[\"mfcc\"]))\n",
    "                for _ in range(t[\"mfcc\"].shape[0]):\n",
    "                    artist_labels_test.append(t[\"artist\"])\n",
    "                    song_labels_test.append(t[\"song\"])\n",
    "        features_test=numpy.delete(features_test,0,0)\n",
    "        \n",
    "        #frame predictions\n",
    "        artist_pred_bgm=[]\n",
    "        for t in features_test:\n",
    "            score={}\n",
    "            for artist in artistlist:\n",
    "                model = bgmdict[artist]\n",
    "                score[artist]= model.score(numpy.reshape(t,(1,-1)))\n",
    "            pred = max(score, key=lambda key: score[key])\n",
    "            artist_pred_bgm.append(pred)\n",
    "        \n",
    "   \n",
    "        #songs predictions\n",
    "        out_bgm={}\n",
    "        for song in songtest:\n",
    "            v=[artist_pred_bgm[u] for u in range(len(artist_pred_bgm)) if song_labels_test[u] == song]\n",
    "            out_bgm[song]=max(set(v), key=v.count)\n",
    "        out_bgm\n",
    "            \n",
    "            \n",
    "        ##create dictioary such that to each artist correspond the list of its predicted songs\n",
    "        song_and_artist_pred_bgm={}\n",
    "        for key,value in out_bgm.items():\n",
    "            if value not in song_and_artist_pred_bgm:\n",
    "                song_and_artist_pred_bgm[value] = []\n",
    "            song_and_artist_pred_bgm[value].append(key)\n",
    "            \n",
    "        ##initialize confusion matrix   \n",
    "        N_art=len(artistlist)\n",
    "        conf_matrix_bgm = pandas.DataFrame(numpy.zeros(shape=(N_art,N_art)), columns = artistlist, index=artistlist)\n",
    "\n",
    "        ##place in position (i,j) the number of songs from artist i predicted as of artist j\n",
    "        for artist_t, listsong_t in song_and_artist_test.items():\n",
    "            for artist_p, listsong_p in song_and_artist_pred_bgm.items():\n",
    "                tot=len(set(listsong_t).intersection(listsong_p))\n",
    "                conf_matrix_bgm[artist_p][artist_t]=tot\n",
    "        #print(conf_matrix_gmm)\n",
    "\n",
    "        #store confusion matrix\n",
    "        CM[sr][n_mfcc]=conf_matrix_bgm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Performances Evaluation\n",
    "Now we use all the confusion matrices stored in **CM** to evaluate performances. To this end we calculate class specific *precison* and *recall* and then we will average the values. This *macro*-averaged measures are suitable for assessing performaces of our models because every class has almost the same number of observations. However we will also use *micro*-averaged measures."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "#inizialize dictionaries for macro and micro measures of performance\n",
    "MACRO={}\n",
    "micro={}\n",
    "for sr in SR:\n",
    "    MACRO[sr]={}\n",
    "    micro[sr]={}\n",
    "    for n_mfcc in MFCC:\n",
    "        MACRO[sr][n_mfcc]=0\n",
    "        micro[sr][n_mfcc]=0\n",
    "\n",
    "        \n",
    "\n",
    "#calculate precison and recall (Macro-averaged)\n",
    "for sr in SR:\n",
    "    for n_mfcc in MFCC:\n",
    "        cm=numpy.asarray(CM[sr][n_mfcc])\n",
    "        recall = numpy.mean(numpy.diag(cm) / numpy.sum(cm, axis = 1))\n",
    "        precision =numpy.mean([u/v if v>0 else 0 for u,v in zip(numpy.diag(cm),numpy.asarray(numpy.sum(cm,axis=0)))])\n",
    "        F1 = (2*precision*recall)/(precision+recall)\n",
    "        MACRO[sr][n_mfcc]={\"recall\":round(recall,3),\"precision\":round(precision,3),\"F1\":round(F1,3)}\n",
    "        \n",
    "#calculate precison and recall (Micro-averaged)\n",
    "for sr in SR:\n",
    "    for n_mfcc in MFCC:\n",
    "        cm=numpy.asarray(CM[sr][n_mfcc])\n",
    "        recall = (numpy.trace(cm) / numpy.sum(numpy.sum(cm, axis = 1)))\n",
    "        precision = (numpy.trace(cm) / numpy.sum(numpy.sum(cm, axis = 0)))\n",
    "        F1 = (2*precision*recall)/(precision+recall)\n",
    "        micro[sr][n_mfcc]={\"recall\":round(recall,3),\"precision\":round(precision,3),\"F1\":round(F1,3)}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#visualization of MACRO-F1 index\n",
    "MF1 = pandas.DataFrame(numpy.zeros(shape=(len(SR),len(MFCC))), columns = MFCC, index=SR)\n",
    "for sr in SR:\n",
    "    for n_mfcc in MFCC:\n",
    "        MF1[n_mfcc][sr]=MACRO[sr][n_mfcc][\"F1\"]\n",
    "\n",
    "#nice heatmap\n",
    "seaborn.set(font_scale=1.4)\n",
    "seaborn.heatmap(MF1, annot=True,annot_kws={\"size\": 12}\n",
    "                  ,linewidths = 1,linecolor=\"gray\",cmap=\"BuGn\")\n",
    "plt.yticks(rotation=0, horizontalalignment='right')\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#visualization of micro-F1 index\n",
    "mF1 = pandas.DataFrame(numpy.zeros(shape=(len(SR),len(MFCC))), columns = MFCC, index=SR)\n",
    "for sr in SR:\n",
    "    for n_mfcc in MFCC:\n",
    "        mF1[n_mfcc][sr]=micro[sr][n_mfcc][\"F1\"]\n",
    "\n",
    "#nice heatmap\n",
    "seaborn.set(font_scale=1.4)\n",
    "seaborn.heatmap(mF1, annot=True,annot_kws={\"size\": 12}\n",
    "                  ,linewidths = 1,linecolor=\"gray\",cmap=\"BuGn\")\n",
    "plt.yticks(rotation=0, horizontalalignment='right')\n",
    "plt.show()"
   ]
  },
  {
   "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}