Audio fingerprinting.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import scipy\n",
    "import matplotlib.pyplot as plt\n",
    "from scipy.ndimage.filters import maximum_filter\n",
    "import timeit\n",
    "import cv2\n",
    "import mysql.connector\n",
    "import sys\n",
    "import os\n",
    "import hashlib\n",
    "import sounddevice as sd\n",
    "import librosa\n",
    "import pyaudio\n",
    "import wave"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from Parameters import FRAME_SIZE, HOP_SIZE \n",
    "def spectrogram(recarr,sr):\n",
    "    \n",
    "    stft_recarr=librosa.stft(recarr,n_fft=FRAME_SIZE,hop_length=HOP_SIZE)\n",
    "    spec_matrix = (np.abs(stft_recarr))**2\n",
    "    \n",
    "    return spec_matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from Parameters import peak_neighborhood_size, min_amp, size\n",
    "\n",
    "def get_peaks(stft_array):\n",
    "    \n",
    "    local_max = maximum_filter(stft_array, size) == stft_array  #Comparing with the original array gives us the sparse array \n",
    "                                                                #in the form of a boolean matrix\n",
    "\n",
    "\n",
    "    frequency_list = []\n",
    "    time_list = []\n",
    "    \n",
    "    L = stft_array.shape[0]   # No of rows in the stft_array\n",
    "    B = stft_array.shape[1]   # No of columns in the stft_array\n",
    "\n",
    "    #The first 2 loops ar used to iterate through all the local maximas\n",
    "    #The following 2 nested if statements are used to check if a particular local maxima is in the required target zone\n",
    "    #Each time we find a peak, the third for loop is used which finds additional 5-6 peaks in the same vertical bin\n",
    "    for i in range(B):\n",
    "        for j in range(L):\n",
    "            if (local_max[j][i]): \n",
    "                \n",
    "                if (stft_array[j][i]>min_amp):           \n",
    "                    frequency_list.append(j)\n",
    "                    time_list.append(i)\n",
    "                    \n",
    "                else:\n",
    "                    local_max[j][i] = False\n",
    "                \n",
    "                \n",
    "                vertical_bin = np.argpartition(stft_array[:,i], -6)[-6:]\n",
    "                \n",
    "                for k in range(len(vertical_bin)):\n",
    "                    if (stft_array[vertical_bin[k]][i]>min_amp):\n",
    "                        frequency_list.append(vertical_bin[k])\n",
    "                        time_list.append(i)\n",
    "                        \n",
    "\n",
    "    #Use this code for plotting the peaks                \n",
    "#     plt.imshow(local_max, cmap='Greys',  interpolation='nearest')\n",
    "#     plt.show()\n",
    "\n",
    "    return time_list, frequency_list, L, B"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from Parameters import min_time_delta, max_time_delta, fan_value, range_frequencies\n",
    "\n",
    "def get_hashes(time_list, frequency_list, L, B,song_id):\n",
    "    t = time_list\n",
    "    f = frequency_list\n",
    "    \n",
    "    N = len(time_list)\n",
    "    hashes = []\n",
    "\n",
    "    count2 = 0\n",
    "    \n",
    "    for i in range(N):\n",
    "        for j in range(min_time_delta,max_time_delta):\n",
    "            if ((i+j)<N):\n",
    "                frequency1 = f[i]\n",
    "                frequency2 = f[i+j]\n",
    "                time_delta = t[i+j] - t[i]\n",
    "                \n",
    "                if ( (time_delta > min_time_delta) & (time_delta < max_time_delta) ):\n",
    "                    count2 += 1\n",
    "                    h = hashlib.sha224(f\"{str(frequency1)}|{str(frequency2)}|{time_delta}\".encode('utf-8'))\n",
    "                    hash1 = h.hexdigest()\n",
    "                    hashes.append((song_id,hash1[0:20],t[i]))\n",
    "                    if (count2>=fan_value):\n",
    "                        break\n",
    "                        \n",
    "    return hashes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mysql_config import host, user, password, database\n",
    "\n",
    "def add_songs_to_database():\n",
    "    \n",
    "    #Connecting to the MySQL server\n",
    "    try:\n",
    "        mydb = mysql.connector.connect(host = host,\n",
    "                                       user = user,\n",
    "                                       password = password,\n",
    "                                       database = database)\n",
    "    except:\n",
    "        print(\"Unable to conect to the database\")\n",
    "    \n",
    "    mycursor = mydb.cursor()\n",
    "    \n",
    "    #Reading the song\n",
    "    song_name = input(\"Enter song name \")\n",
    "    filename='mp3/'+song_name+'.wav'\n",
    "    recarr,sr=librosa.load(filename)\n",
    "    \n",
    "    \n",
    "    #Checking if the song is already fingerprinted and if not then adding it in the database\n",
    "    sql = \"SELECT count(*) from songs where song_name = %s and fingerprinted = %s\"\n",
    "    val = (song_name,\"1\")\n",
    "    mycursor.execute(sql,val)\n",
    "    x = mycursor.fetchall()\n",
    "    if (x[0][0]==1):\n",
    "        print(\"Song \", song_name,\" is already fingerprinted\")\n",
    "        return\n",
    "    else:\n",
    "        sql = \"INSERT INTO songs (song_name,fingerprinted) VALUES (%s,%s)\"\n",
    "        val = (song_name,\"1\")\n",
    "        mycursor.execute(sql,val)\n",
    "        sql = \"SELECT * from songs where song_name = %s and fingerprinted = %s\"\n",
    "        val = (song_name,\"1\")\n",
    "        mycursor.execute(sql,val)\n",
    "        try:\n",
    "            x = mycursor.fetchall()\n",
    "            song_id = x[0][0]\n",
    "        except:\n",
    "            song_id = 1\n",
    "        \n",
    "    print(\"Making spectrogram\")\n",
    "    spec_matrix = spectrogram(recarr,sr)\n",
    "    print(\"Spectrogram successfully made\")\n",
    "    \n",
    "    print(\"Finding peaks in the spectrogram\")\n",
    "    time_list, frequency_list, L, B = get_peaks(spec_matrix)\n",
    "    print(\"Found \", len(time_list),\" peaks\")\n",
    "    \n",
    "    print(\"Making hashes\")\n",
    "    hashes = get_hashes(time_list, frequency_list, L, B,song_id)\n",
    "    print(\"Made \",len(hashes),\" hashes\")\n",
    "    hashes = list(set(hashes))\n",
    "    print(\"Found \",len(hashes),\" unique hashes\")\n",
    "    \n",
    "    #Saving hashes in the databses\n",
    "    val = hashes\n",
    "    sql = \"INSERT INTO fingerprints (song_id,hash,offset) VALUES (%s,%s,%s)\"\n",
    "    mycursor.executemany(sql,val)\n",
    "    mydb.commit()\n",
    "    print(mycursor.rowcount, \"record(s) inserted for song \",song_name)\n",
    "    \n",
    "    #Closing the connection with the database\n",
    "    mycursor.close()\n",
    "    mydb.close()\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def recognize():\n",
    "    \n",
    "    sampling_rate = 44100\n",
    "    duration = 10\n",
    "    print(\"How do you want to recognize the song\")\n",
    "    print(\"    1. Through microphone\")\n",
    "    print(\"    2. Reading the sample file from system\")\n",
    "    choice = input(\"Enter your choice number: \")\n",
    "    \n",
    "    if(choice == '1'):\n",
    "        filename = \"test100\" + \".wav\"\n",
    "        chunk = 1024\n",
    "        FORMAT = pyaudio.paInt16\n",
    "        channels = 1\n",
    "        sample_rate = 44100\n",
    "        record_seconds = 10\n",
    "        p = pyaudio.PyAudio()\n",
    "        stream = p.open(format=FORMAT,\n",
    "                channels=channels,\n",
    "                rate=sample_rate,\n",
    "                input=True,\n",
    "                output=True,\n",
    "                frames_per_buffer=chunk)\n",
    "        frames = []\n",
    "        print(\"Recording...\")        \n",
    "        for i in range(int(44100 / chunk * record_seconds)):\n",
    "            data = stream.read(chunk)\n",
    "            frames.append(data)\n",
    "        \n",
    "        print(\"Recording ended...\")\n",
    "        stream.stop_stream()\n",
    "        stream.close()\n",
    "        p.terminate()\n",
    "        wf = wave.open(filename, \"wb\")\n",
    "        wf.setnchannels(channels)\n",
    "        wf.setsampwidth(p.get_sample_size(FORMAT))\n",
    "        wf.setframerate(sample_rate)\n",
    "        wf.writeframes(b\"\".join(frames))\n",
    "        wf.close()\n",
    "        \n",
    "\n",
    "        path = filename \n",
    "        recarr,sr=librosa.load(path)\n",
    "        recognize_the_song(recarr, sampling_rate)\n",
    "    \n",
    "    \n",
    "    elif (choice == '2'):\n",
    "        sample_name = input(\"Enter the name of the sample you want to recognize: \")        \n",
    "        path = 'test/' + sample_name + '.wav' \n",
    "        recarr,sr=librosa.load(path)\n",
    "        recognize_the_song(recarr,sr)\n",
    "        \n",
    "    else:\n",
    "        print(\"PLease enter a valid choice number\")\n",
    "        \n",
    "    return"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mysql_config import host, user, password, database\n",
    "\n",
    "def recognize_the_song(audio, sampling_rate):\n",
    "    \n",
    "    #Connecting to the MySQL server\n",
    "    try:\n",
    "        mydb = mysql.connector.connect(host = host,\n",
    "                                       user = user,\n",
    "                                       password = password,\n",
    "                                       database = database)\n",
    "    except:\n",
    "        print(\"Unable to conect to the database\")\n",
    "    \n",
    "    mycursor = mydb.cursor()\n",
    "    \n",
    "    print(\"Making spec\")\n",
    "    spec_matrix = spectrogram(audio,sampling_rate)\n",
    "    print(\"Spec done\")\n",
    "    \n",
    "    print(\"Finding peaks\")\n",
    "    time_list, frequency_list, L, B = get_peaks(spec_matrix)\n",
    "    print(\"Peaks found\", len(time_list))\n",
    "    \n",
    "    hashes = get_hashes(time_list, frequency_list, L, B,song_id = 0)\n",
    "    print(\"Hashes found\",len(hashes))\n",
    "    hashes = list(set(hashes))\n",
    "    print(\"Unique Hashes found\",len(hashes))\n",
    "    \n",
    "    unique_hashes = hashes\n",
    "    \n",
    "    bins = [{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}]\n",
    "\n",
    "    for sample_hash in unique_hashes:\n",
    "        \n",
    "        sql = (\"SELECT song_id, offset from fingerprints where hash = %s\")\n",
    "        val = (sample_hash[1],)\n",
    "        mycursor.execute(sql,val)\n",
    "        id_and_offset = mycursor.fetchall()\n",
    "        sample_offset = sample_hash[2] \n",
    "        for j in id_and_offset:\n",
    "            \n",
    "            song_id = j[0]\n",
    "            database_offset = j[1]\n",
    "            difference = database_offset - sample_offset\n",
    "            bins[song_id][difference] = bins[song_id].get(difference,0) + 1 \n",
    "    \n",
    "    score = {}\n",
    "    for i in range(len(bins)):\n",
    "        if bins[i] != {}:\n",
    "            score[i] = bins[i].get( max(bins[i], key = bins[i].get) )\n",
    "    try:\n",
    "        Best_match = max(score,key = score.get)\n",
    "    except:\n",
    "        print(\"Error\")\n",
    "        return \n",
    "    #Get the name of the song with the song_id equal to best_match\n",
    "    sql = (\"SELECT song_name from songs where song_id = %s\")\n",
    "    val = (Best_match,)\n",
    "    mycursor.execute(sql,val)\n",
    "    name_best_match = mycursor.fetchone()\n",
    "    \n",
    "    print(\"The best match song is \", name_best_match[0],\" with a score of \",score[Best_match])\n",
    "    \n",
    "    return\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"What do you wish to do:\")\n",
    "print(\"    1. Add songs to the data base\")\n",
    "print(\"    2. Recognize a song\")\n",
    "choice = input(\"Enter your choice number: \")\n",
    "if (choice == '1'):\n",
    "    add_songs_to_database()\n",
    "elif (choice =='2'):\n",
    "    recognize()\n",
    "else:\n",
    "    print(\"PLease enter a valid choice number\")"
   ]
  }
 ],
 "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}