audio_similarity_prototype.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Audio similarity - prototype"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MFCC descriptor + DTW distance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import librosa\n",
    "import librosa.display\n",
    "import matplotlib.pyplot as plt\n",
    "from dtw import dtw, accelerated_dtw\n",
    "from numpy.linalg import norm\n",
    "import numpy as np\n",
    "import math\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Extract MFCC feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_mfcc(file):\n",
    "    signal, sr = librosa.load('data/' + file)\n",
    "    mfcc = librosa.feature.mfcc(signal, n_mfcc=13, sr=sr)\n",
    "    return mfcc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Split one audio track into multiple samples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 306,
   "metadata": {},
   "outputs": [],
   "source": [
    "N = 400 # Sample size\n",
    "\n",
    "def split_mfcc(mfcc):\n",
    "    n_samples = math.floor(mfcc.shape[1] / SAMPLE_SIZE)\n",
    "    last_sample_size = mfcc.shape[1] % SAMPLE_SIZE\n",
    "#    print('Splitting into %d samples of size %d. Last sample has size %d.' % (n_samples, SAMPLE_SIZE, last_sample_size))\n",
    "    \n",
    "    samples = []\n",
    "    for i in range(n_samples):\n",
    "        samples.append(mfcc[:, i*N:(i+1)*N])\n",
    "    samples.append(mfcc[:, n_samples*N:])\n",
    "    \n",
    "    return samples"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build reference database of audio samples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 299,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_db(files):\n",
    "    db = []\n",
    "    \n",
    "    for file in files:\n",
    "        mfcc = extract_mfcc(file)\n",
    "        samples = split_mfcc(mfcc)\n",
    "\n",
    "        for sample in samples:\n",
    "            db.append({ 'file': file, 'mfcc': sample })\n",
    "    \n",
    "    print(\"Added %d samples of %d audiotracks\" % (len(db), len(files)))\n",
    "    return np.array(db, dtype=object)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distance measure"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 300,
   "metadata": {},
   "outputs": [],
   "source": [
    "def distance(a, b):\n",
    "    dist, cost, acc_cost, path = dtw(a.T, b.T, dist=lambda x, y: norm(x - y, ord=1))\n",
    "    return dist"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find the most similar match"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 301,
   "metadata": {},
   "outputs": [],
   "source": [
    "def find_match(mfcc, db):\n",
    "    start = time.time()\n",
    "    matches = []\n",
    "    \n",
    "    # Loop over all samples in the database and find the best match\n",
    "    for sample in db:\n",
    "        dist = None\n",
    "        \n",
    "        # Align both samples to the same size\n",
    "        if mfcc.shape[1] == sample['mfcc'].shape[1]:\n",
    "            dist = distance(mfcc, sample['mfcc'])\n",
    "        elif mfcc.shape[1] < sample['mfcc'].shape[1]:\n",
    "            dist = distance(mfcc, sample['mfcc'][:, mfcc.shape[1]])\n",
    "        else:\n",
    "            dist = distance(mfcc[:, sample['mfcc'].shape[1]], sample['mfcc'])\n",
    "        \n",
    "        matches.append({ 'file': sample['file'], 'dist': dist })\n",
    "\n",
    "    matches = sorted(matches, key=lambda x: x['dist'])\n",
    "    print(\"Time: %.2f s\" % (time.time() - start))\n",
    "    return matches"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 307,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/site-packages/librosa/core/audio.py:162: UserWarning: PySoundFile failed. Trying audioread instead.\n",
      "  warnings.warn(\"PySoundFile failed. Trying audioread instead.\")\n",
      "/usr/local/lib/python3.8/site-packages/librosa/core/audio.py:162: UserWarning: PySoundFile failed. Trying audioread instead.\n",
      "  warnings.warn(\"PySoundFile failed. Trying audioread instead.\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Added 6 samples of 4 audiotracks\n"
     ]
    }
   ],
   "source": [
    "files = ['sample_1.wav', 'dubstep_beat_1.wav', 'dubstep_drum.mp3', 'trance.m4a']\n",
    "db = build_db(files)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 308,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/site-packages/librosa/core/audio.py:162: UserWarning: PySoundFile failed. Trying audioread instead.\n",
      "  warnings.warn(\"PySoundFile failed. Trying audioread instead.\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time: 4.62 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[{'file': 'sample_1.wav', 'dist': 35945.62905883789},\n",
       " {'file': 'dubstep_beat_1.wav', 'dist': 86426.36378479004},\n",
       " {'file': 'sample_1.wav', 'dist': 119476.18923950195},\n",
       " {'file': 'trance.m4a', 'dist': 148126.52734375},\n",
       " {'file': 'dubstep_drum.mp3', 'dist': 149414.47550964355},\n",
       " {'file': 'dubstep_beat_1.wav', 'dist': 209982.72225952148}]"
      ]
     },
     "execution_count": 308,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mfcc = extract_mfcc('recorded_sample_1.m4a')[:, :N]\n",
    "find_match(mfcc, db)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "mfcc = extract_mfcc('guitar.wav')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2905"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(mfcc[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 310,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/site-packages/librosa/core/audio.py:162: UserWarning: PySoundFile failed. Trying audioread instead.\n",
      "  warnings.warn(\"PySoundFile failed. Trying audioread instead.\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time: 4.61 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[{'file': 'sample_1.wav', 'dist': 37445.249908447266},\n",
       " {'file': 'dubstep_beat_1.wav', 'dist': 85355.85360717773},\n",
       " {'file': 'trance.m4a', 'dist': 141331.39974975586},\n",
       " {'file': 'dubstep_drum.mp3', 'dist': 149616.13061523438},\n",
       " {'file': 'sample_1.wav', 'dist': 152272.89219665527},\n",
       " {'file': 'dubstep_beat_1.wav', 'dist': 178349.8076171875}]"
      ]
     },
     "execution_count": 310,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mfcc = extract_mfcc('trance_2.m4a')[:, :N]\n",
    "find_match(mfcc, db)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fingerprinting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fingeprint audiotrack using Chromaprint library / tool"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "def parse_chromaprint_output(output):\n",
    "    output = output.decode('utf-8')\n",
    "    key = 'FINGERPRINT='\n",
    "    strip_index = output.find(key) + len(key)\n",
    "    return list(map(int, output[strip_index:-1].split(',')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fingeprint_file(path):\n",
    "    output = !fpcalc $path  -raw\n",
    "    return parse_chromaprint_output(output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "import subprocess\n",
    "\n",
    "def fingeprint_file(path):\n",
    "        cmd = 'fpcalc %s -raw' % path\n",
    "        process = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)\n",
    "        output, error = process.communicate()\n",
    "\n",
    "        if error:\n",
    "            raise Exception(error)\n",
    "        return parse_chromaprint_output(output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "fp1 = fingeprint_file('data/sample_1.wav')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "fp2 = fingeprint_file('data/trance.m4a')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compare fingeprints\n",
    "\n",
    "Source: https://gist.github.com/lalinsky/1132166\n",
    "\n",
    "Another way with correlation: https://medium.com/@shivama205/audio-signals-comparison-23e431ed2207"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "popcnt_table_8bit = [\n",
    "    0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,\n",
    "    1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,\n",
    "    1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,\n",
    "    2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,\n",
    "    1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,\n",
    "    2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,\n",
    "    2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,\n",
    "    3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8,\n",
    "]\n",
    "\n",
    "def popcnt(x):\n",
    "    \"\"\"\n",
    "    Count the number of set bits in the given 32-bit integer.\n",
    "    \"\"\"\n",
    "    return (popcnt_table_8bit[(x >>  0) & 0xFF] +\n",
    "            popcnt_table_8bit[(x >>  8) & 0xFF] +\n",
    "            popcnt_table_8bit[(x >> 16) & 0xFF] +\n",
    "            popcnt_table_8bit[(x >> 24) & 0xFF])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "def compare_fingerprints(fp1, fp2):\n",
    "    error = 0\n",
    "    for x, y in zip(fp1, fp2):\n",
    "        error += popcnt(x ^ y)\n",
    "    return 1.0 - error / 32.0 / min(len(fp1), len(fp2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Similarity: 0.482031\n"
     ]
    }
   ],
   "source": [
    "sim = compare_fingerprints(fp1, fp2)\n",
    "print(\"Similarity: %f\"% sim)"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}