4 - Classification using GMM.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Classification using GMM\n",
    "Here we try a different approach for classification. For each artist in the dataset we model the mfccs coming from their songs as a guassian mixture model with a \"big\" number of components. So we train for each artist a GMM (using the train songs) and estimate its parameters (the weights, the means and the covariances) by expected maximization. Then for every frame of every song in the test set we compute the log-likelihood for that frame for coming from the model of each artist. We then assign the frame to the artist with the highest score and then the song according to the artist that has received the majority of \"votes\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "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\n",
    "import joblib\n",
    "import pickle"
   ]
  },
  {
   "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 extraction can take a while)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('x.txt', 'r') as filehandle:\n",
    "    y = json.load(filehandle)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "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": 4,
   "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": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "40\n"
     ]
    }
   ],
   "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",
    "N = N-1\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 = 3)\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": 6,
   "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 GMM 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": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(10000, 20)\n"
     ]
    }
   ],
   "source": [
    "#SR = [10000]    #to test\n",
    "#MFCC = [20]     #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 samplig 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 GMM for each artist\n",
    "        gmmdict = {}\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",
    "            gmm = sklearn.mixture.GaussianMixture(n_components = 64, covariance_type = 'diag', max_iter = 20, tol = 0.01)\n",
    "            gmm.fit(features_train)\n",
    "            gmmdict[artist] = gmm    \n",
    "\n",
    "            \n",
    "        #create a matrix with test features and also ordered labels for song adn 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_gmm = []\n",
    "        for t in features_test:\n",
    "            score = {}\n",
    "            for artist in artistlist:\n",
    "                model = gmmdict[artist]\n",
    "                score[artist] = model.score(numpy.reshape(t,(1,-1)))\n",
    "            pred = max(score, key = lambda key: score[key])\n",
    "            artist_pred_gmm.append(pred)\n",
    "        \n",
    "   \n",
    "        #songs predictions\n",
    "        out_gmm = {}\n",
    "        for song in songtest:\n",
    "            v = [artist_pred_gmm[u] for u in range(len(artist_pred_gmm)) if song_labels_test[u] == song]\n",
    "            out_gmm[song] = max(set(v), key=v.count)\n",
    "        #out_gmm\n",
    "            \n",
    "            \n",
    "        ##create dictioary such that to each artist correspond the list of its predicted songs\n",
    "        song_and_artist_pred_gmm = {}\n",
    "        for key,value in out_gmm.items():\n",
    "            if value not in song_and_artist_pred_gmm:\n",
    "                song_and_artist_pred_gmm[value] = []\n",
    "            song_and_artist_pred_gmm[value].append(key)\n",
    "            \n",
    "        ##initialize confusion matrix   \n",
    "        N_art=len(artistlist)\n",
    "        conf_matrix_gmm = 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_gmm.items():\n",
    "                tot=len(set(listsong_t).intersection(listsong_p))\n",
    "                conf_matrix_gmm[artist_p][artist_t] = tot\n",
    "        #print(conf_matrix_gmm)\n",
    "\n",
    "        #store confusion matrix\n",
    "        CM[sr][n_mfcc] = conf_matrix_gmm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "CM[10000][20]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save the final model  then one can load it as model = joblib.load('GMM.joblib')\n",
    "joblib.dump(gmmdict, 'GMM.joblib') \n",
    "\n",
    "#and save the confusion matrix dictionary\n",
    "CM1={}\n",
    "for sr in SR:\n",
    "    CM1[sr] = {}\n",
    "    for n_mfcc in MFCC:\n",
    "        CM1[sr][n_mfcc] = CM[sr][n_mfcc].rename_axis('ID').values.tolist()\n",
    "with open('CM_GMM.txt', 'w') as filehandle:\n",
    "    json.dump(CM1, filehandle)"
   ]
  },
  {
   "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": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#artistlist=['Bruce Springsteen','Coldplay','Ed Sheeran', \"Guns N' Roses\"\n",
    "            #, 'Michael Jackson', 'Passenger', 'Pink Floyd', 'Queen', 'Simon & Garfunkel' , 'The Beatles']\n",
    "#load CM if needed\n",
    "with open('CM_GMM.txt', 'r') as filehandle:\n",
    "    CM2 = json.load(filehandle)\n",
    "CM = {}\n",
    "for sr in SR:\n",
    "    CM[sr] = {}\n",
    "    for n_mfcc in MFCC:\n",
    "        CM[sr][n_mfcc]=pandas.DataFrame(numpy.asarray(CM2[str(sr)][str(n_mfcc)]),\n",
    "                                        columns = artistlist, index=artistlist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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": [
    "#confusion matrix of the best model\n",
    "CM[10000][20]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "seaborn.set(font_scale=1.4)\n",
    "seaborn.heatmap(CM[10000][20], annot = True,annot_kws = {\"size\": 12},\n",
    "                  linewidths = 1, linecolor = \"gray\", cmap = \"Blues\")\n",
    "plt.yticks(rotation = 0, horizontalalignment = 'right')\n",
    "plt.xticks(rotation = 45, horizontalalignment = 'right')\n",
    "plt.savefig(\"cm.svg\")"
   ]
  },
  {
   "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
}