3 - Classification at Frame-Level using KNN.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Classification usign Frame-Level Features\n",
    "Here we try to classify songs to their respective artist using a two step procedure that consists in classifying frames to artists using standard classifiers (here we use KNN, but SVM gives similar results) and then classifying songs to to the artist for which the majority of its frame were classified in the previous step (this is far from the state-of-the-art but, despite being simple, it actually works well)."
   ]
  },
  {
   "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 librosa.display\n",
    "import numpy\n",
    "import random\n",
    "import pandas\n",
    "import seaborn\n",
    "import json\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 extration can take a while, more then one hour for our dataset)."
   ]
  },
  {
   "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": null,
   "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",
    "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=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": "markdown",
   "metadata": {},
   "source": [
    "# Two step classification of songs to their artists using KNN:\n",
    "Now as claimed we perform the two step classification"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "##num of neighbours, default is 5, we decided to use k = 10, because gave us best performances on smaller models\n",
    "k = 10 \n",
    "#initialize knn object\n",
    "knn = sklearn.neighbors.KNeighborsClassifier(n_neighbors = k, metric='euclidean')\n",
    "\n",
    "#SR = [5000]    #to test\n",
    "#MFCC = [8]          #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",
    "        #create a big matrix containing all mfcc of all songs of all artists for train and test\n",
    "        features_train = numpy.empty((1,n_mfcc))\n",
    "        features_test = numpy.empty((1,n_mfcc))\n",
    "        for t in x[sr][n_mfcc]:\n",
    "            if t[\"song\"] in songtrain:\n",
    "                features_train = numpy.vstack((features_train, t[\"mfcc\"]))\n",
    "            else:\n",
    "                features_test = numpy.vstack((features_test, t[\"mfcc\"]))\n",
    "        features_train = numpy.delete(features_train,0,0)\n",
    "        features_test = numpy.delete(features_test,0,0)\n",
    "        \n",
    "        #creating labels of each frame relative to the artist and to the song\n",
    "        artist_labels_train = []\n",
    "        artist_labels_test = []\n",
    "        song_labels_train = []\n",
    "        song_labels_test = []\n",
    "        for t in x[sr][n_mfcc]:\n",
    "            if t[\"song\"] in songtrain:\n",
    "                for _ in range(t[\"mfcc\"].shape[0]):\n",
    "                    artist_labels_train.append(t[\"artist\"])\n",
    "                    song_labels_train.append(t[\"song\"])\n",
    "            else:\n",
    "                for _ in range(t[\"mfcc\"].shape[0]):\n",
    "                    artist_labels_test.append(t[\"artist\"])\n",
    "                    song_labels_test.append(t[\"song\"])\n",
    "        \n",
    "        #train and test knn: this is the first step of the classification\n",
    "        knn.fit(features_train, artist_labels_train)\n",
    "        artist_pred_knn = knn.predict(features_test)\n",
    "        \n",
    "        #assing song to artist for which majority of its frame were calssified\n",
    "        out_knn = {}\n",
    "        for song in songtest:\n",
    "            u = [artist_pred_knn[u] for u in range(len(artist_pred_knn)) if song_labels_test[u] == song]\n",
    "            out_knn[song] = max(set(u), key=u.count)\n",
    "         \n",
    "        \n",
    "        #initialize confusion matrix\n",
    "        N_art = len(artistlist)\n",
    "        conf_matrix_knn = pandas.DataFrame(numpy.zeros(shape = (N_art,N_art)), columns = artistlist, index=artistlist)\n",
    "        \n",
    "        ##create dictionary such that to each artist correspond the list of its real songs\n",
    "        song_and_artist_test = {}\n",
    "        for t in x[sr][n_mfcc]:\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\"])\n",
    "        \n",
    "        \n",
    "        ##create dictioary such that to each artist correspond the list of\n",
    "        ##its songs as preddicted by knn two step procedure\n",
    "        song_and_artist_pred_knn = {}\n",
    "        for key,value in out_knn.items():\n",
    "            if value not in song_and_artist_pred_knn:\n",
    "                song_and_artist_pred_knn[value] = []\n",
    "            song_and_artist_pred_knn[value].append(key)  \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_knn.items():\n",
    "                tot = len(set(listsong_t).intersection(listsong_p))\n",
    "                conf_matrix_knn[artist_p][artist_t] = tot\n",
    "        \n",
    "        #store confusion matrix\n",
    "        CM[sr][n_mfcc] = conf_matrix_knn\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save the final model  then one can load it as model = joblib.load('knn.joblib')\n",
    "joblib.dump(knn, 'knn.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_knn.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, moreover, we will consider also the harmonic mean of the global precision and recall, called F1 score. 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 that in this setting of multiclass classification are all equal to the accuracy of the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "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_knn.txt', 'r') as filehandle:\n",
    "    CM2 = json.load(filehandle)\n",
    "            \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": 28,
   "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": {
    "scrolled": true
   },
   "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 (or accuracy)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": [
    "CM[10000][20]"
   ]
  },
  {
   "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
}