record_og.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'seaborn'",
     "output_type": "error",
     "traceback": [
      "\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
      "\u001B[0;31mModuleNotFoundError\u001B[0m                       Traceback (most recent call last)",
      "Cell \u001B[0;32mIn[2], line 7\u001B[0m\n\u001B[1;32m      5\u001B[0m \u001B[38;5;28;01mfrom\u001B[39;00m \u001B[38;5;21;01mtime\u001B[39;00m \u001B[38;5;28;01mimport\u001B[39;00m sleep\n\u001B[1;32m      6\u001B[0m \u001B[38;5;28;01mimport\u001B[39;00m \u001B[38;5;21;01mnumpy\u001B[39;00m \u001B[38;5;28;01mas\u001B[39;00m \u001B[38;5;21;01mnp\u001B[39;00m\n\u001B[0;32m----> 7\u001B[0m \u001B[38;5;28;01mimport\u001B[39;00m \u001B[38;5;21;01mseaborn\u001B[39;00m \u001B[38;5;28;01mas\u001B[39;00m \u001B[38;5;21;01msns\u001B[39;00m\n\u001B[1;32m      8\u001B[0m \u001B[38;5;28;01mimport\u001B[39;00m \u001B[38;5;21;01mpandas\u001B[39;00m \u001B[38;5;28;01mas\u001B[39;00m \u001B[38;5;21;01mpd\u001B[39;00m\n\u001B[1;32m      9\u001B[0m \u001B[38;5;28;01mfrom\u001B[39;00m \u001B[38;5;21;01mtqdm\u001B[39;00m\u001B[38;5;21;01m.\u001B[39;00m\u001B[38;5;21;01mnotebook\u001B[39;00m \u001B[38;5;28;01mimport\u001B[39;00m tqdm\n",
      "\u001B[0;31mModuleNotFoundError\u001B[0m: No module named 'seaborn'"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import time\n",
    "\n",
    "import sounddevice\n",
    "import sounddevice as sd\n",
    "import soundfile as sf\n",
    "from time import sleep\n",
    "import numpy as np\n",
    "import seaborn as sns\n",
    "import pandas as pd\n",
    "from tqdm.notebook import tqdm\n",
    "import glob\n",
    "import pathlib\n",
    "import matplotlib\n",
    "import matplotlib.pyplot as plt\n",
    "from scipy import fft\n",
    "from scipy.signal import chirp\n",
    "import math\n",
    "import subprocess\n",
    "import librosa\n",
    "import torchaudio\n",
    "from torch.nn import functional as F\n",
    "import torchaudio.functional as Fa\n",
    "import torch\n",
    "import torch.fft\n",
    "from IPython import embed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "outputs": [],
   "source": [
    "def play_android(wav_file_path, fs_rec):\n",
    "    # print(wav_file_path)\n",
    "    data, fs_org = sf.read(wav_file_path)\n",
    "    duration = librosa.get_duration(filename=wav_file_path)\n",
    "    print('File duration = ', duration)\n",
    "    file_name = '/'.join(wav_file_path.split('/')[-3:])\n",
    "    print(file_name)\n",
    "    subprocess.run([\n",
    "        \"adb shell am start -a android.intent.action.VIEW -d file:///storage/emulated/0/{}  -t video/* -f 0x00008000\".format(\n",
    "            file_name)], shell=True)\n",
    "    print('Start Recording at fs = {}:'.format(fs_rec))\n",
    "    rec = sd.rec(int(duration * fs_rec), samplerate=fs_rec, channels=2, blocking=True)\n",
    "    print(\"Recording Done!\")\n",
    "    rec_in = rec[:, 0]  # inside box\n",
    "    rec_out = rec[:, 1]  # outside box\n",
    "    return data, rec_in, rec_out, duration\n",
    "\n",
    "\n",
    "def get_f0_Q_tube(L, r, correct=False):\n",
    "    # open_ended tube:\n",
    "    n = 1\n",
    "    T = 24  ## Temp in Celsius\n",
    "    v = 20.05 * math.sqrt(T + 273.15)  #343   # sound speed in air in m/s\n",
    "    f0 = n * v / 2 / L\n",
    "    ## Corrected frequency (wiki)\n",
    "    if correct:\n",
    "        print('Correction is Applied!')\n",
    "        f0 = n * v / 2 / (L + 0.8 * 2 * r)\n",
    "    print(\"resonant frequency = \", f0)\n",
    "    # Quality factor method 1:\n",
    "    A = math.pi * r ** 2  # radius in meters\n",
    "    d_rad = 5.34 * 10 ** -5 * A * f0 ** 2\n",
    "    d_wall = 5.71 * 10 ** -3 * (A * f0) ** -0.5\n",
    "    Q = round(1 / (d_rad + d_wall))\n",
    "    print(\"Quality Factor = \", Q)\n",
    "    # # Quality factor method 2:\n",
    "    # x = 3.8  # 3500                 # x = (r/d)**2\n",
    "    # Q2 = round(math.sqrt(x)/1.5)\n",
    "    # print(\"Quality Factor 2 = \", Q2)\n",
    "    return f0, Q\n",
    "\n",
    "\n",
    "def resonant_filter_torch(wav_path, f0, Q, num_harmonics, save_audio=True):\n",
    "    waveform, fs = torchaudio.load(wav_path)\n",
    "    # print('Original wav limits:', torch.max(waveform), torch.min(waveform))\n",
    "    # print(waveform.shape)\n",
    "    y_sum = torch.zeros(waveform.shape)\n",
    "    for i in range(1, num_harmonics + 1):\n",
    "        y_sum += Fa.band_biquad(waveform, fs, i * f0, Q, noise=False)\n",
    "    y_sum = y_sum / torch.max(torch.abs(y_sum))\n",
    "\n",
    "    z_sum = torch.zeros(waveform.shape)\n",
    "    for i in range(1, num_harmonics + 1):\n",
    "        z_sum += Fa.bandpass_biquad(waveform, fs, f0 * i, Q, const_skirt_gain=False)\n",
    "    z_sum = z_sum / torch.max(torch.abs(z_sum))\n",
    "\n",
    "    return waveform, y_sum, z_sum, fs\n",
    "\n",
    "\n",
    "def get_transfer_function_scipy(sig_org, sig_out, sig_in, fs, fs_org, T, save_fig=False, title=''):\n",
    "    \"\"\"\n",
    "    :param org: original signal\n",
    "    :param sig_in: signal recorded Inside the box\n",
    "    :param sig_out: signal recorded Outside the box\n",
    "    :param fs: sampling rate\n",
    "    :return: TF: transfer function = FFT(sig2)/FFT(sig1)\n",
    "    \"\"\"\n",
    "    l1 = 15\n",
    "    l2 = 10\n",
    "    d1 = 1\n",
    "    d2 = 2\n",
    "    ratio = (d2 / d1) ** 2\n",
    "    x = np.arange(1, 7500, 1)\n",
    "    out = -1 / np.tan(2 * math.pi * x * l1 / 34350) - ratio * 1 / np.tan(2 * math.pi * x * l2 / 34350)\n",
    "    zero_crossings = np.where(np.diff(np.sign(out)).astype(bool) & (out[:-1] < 0.5) & (out[:-1] > -0.5))[0]\n",
    "\n",
    "    num_fft = len(sig_in)\n",
    "    sig_org_fft = np.abs(fft.rfft(sig_org, norm=\"ortho\", n=num_fft))\n",
    "    sig_in_fft = np.abs(fft.rfft(sig_in, norm=\"ortho\"))\n",
    "    sig_out_fft = np.abs(fft.rfft(sig_out, norm=\"ortho\"))\n",
    "    freq_org = fft.rfftfreq(num_fft, d=1 / fs_org)\n",
    "    freq = fft.rfftfreq(num_fft, d=1 / fs)\n",
    "\n",
    "    fig, ax = plt.subplots(3, 1, sharex=True)\n",
    "    plt.setp(ax, xticks=np.arange(0, fs / 2, 250))\n",
    "    plt.xticks(rotation=45)\n",
    "    plt.xlim([-1, 4000])\n",
    "    ax[0].plot(freq_org, sig_org_fft, color='black', alpha=0.8, label='Original')\n",
    "    ax[0].legend(loc='upper right')\n",
    "    ax[1].plot(freq, sig_out_fft, color='blue', alpha=0.8, label='out_box')\n",
    "    ax[1].legend(loc='upper right')\n",
    "    ax[2].plot(freq, sig_in_fft, color='red', alpha=0.8, label='In_box')\n",
    "    ax[2].legend(loc='upper right')\n",
    "    for j in zero_crossings:\n",
    "        ax[2].axvline(x=j, c='b', ls=\"--\", alpha=0.3)\n",
    "    plt.tight_layout()\n",
    "    if save_fig:\n",
    "        fig_save_dir = title.replace(\"out\", \"figures\").replace('.wav', '_FFT.png')\n",
    "        pathlib.Path(fig_save_dir).parent.mkdir(parents=True, exist_ok=True)\n",
    "        plt.savefig(fig_save_dir)\n",
    "    plt.show()\n",
    "\n",
    "\n",
    "def load_play_save(dataset, save_dir, fs_rec=16000, save=True):\n",
    "    if '16' in dataset:\n",
    "        fs_org = 16000\n",
    "    else:\n",
    "        fs_org = 8000\n",
    "    for i, wav_file_path in enumerate(glob.glob(\"../Dataset/VoxCeleb/*/*.wav\")):\n",
    "        sleep(3)\n",
    "        try:\n",
    "            print('Original Dataset fs =', fs_org)\n",
    "            print('Now playing file {}: '.format(i), os.path.basename(wav_file_path))\n",
    "            ## Play using android:\n",
    "            org, rec_in, rec_out, T = play_android(wav_file_path.replace(\"VoxCeleb\", dataset), fs_rec)\n",
    "            ## Play using PC connected speakers:\n",
    "            # org, rec_in, rec_out, fs = play_record_2mic(wav_file_path)\n",
    "            outbox_path = wav_file_path.replace(\"VoxCeleb\", save_dir + \"/out\")\n",
    "            inbox_path = wav_file_path.replace(\"VoxCeleb\", save_dir + \"/in\")\n",
    "            inbox_path_normalized = wav_file_path.replace(\"VoxCeleb\", save_dir + \"/in_normalized\")\n",
    "            if save:\n",
    "                pathlib.Path(outbox_path).parent.mkdir(parents=True, exist_ok=True)\n",
    "                pathlib.Path(inbox_path).parent.mkdir(parents=True, exist_ok=True)\n",
    "                pathlib.Path(inbox_path_normalized).parent.mkdir(parents=True, exist_ok=True)\n",
    "                sf.write(inbox_path, rec_in, fs_rec)\n",
    "                rec_in_norm = rec_in / max(abs(rec_in)) * max(abs(org))\n",
    "                rec_out_norm = rec_out / max(abs(rec_out)) * max(abs(org))\n",
    "                sf.write(inbox_path_normalized, rec_in_norm, fs_rec)\n",
    "                sf.write(outbox_path, rec_out_norm, fs_rec)\n",
    "            get_transfer_function_scipy(org, rec_out_norm, rec_in_norm, fs_rec, fs_org, T, save, title=outbox_path)\n",
    "\n",
    "        except KeyboardInterrupt:\n",
    "            print('\\nPausing...  (Hit ENTER to continue, type quit to exit.)')\n",
    "            try:\n",
    "                response = input()\n",
    "                if response == 'quit':\n",
    "                    break\n",
    "                print('Resuming...')\n",
    "            except KeyboardInterrupt:\n",
    "                print('Resuming...')\n",
    "                continue\n",
    "\n",
    "\n",
    "def load_play_save_dataset(dataset, save_dir, fs_rec=16000, save=True):\n",
    "    if '16' in dataset:\n",
    "        fs_org = 16000\n",
    "    else:\n",
    "        fs_org = 8000\n",
    "    for i, wav_file_path in enumerate(glob.glob(\"../Dataset/{}/*/*.wav\".format(dataset))):\n",
    "        try:\n",
    "            if os.path.exists(wav_file_path.replace(dataset, save_dir + \"/in\")):\n",
    "                print(wav_file_path, 'exists already')\n",
    "                continue\n",
    "            print('Original Dataset fs =', fs_org)\n",
    "            print('Now playing file {}: '.format(i), os.path.basename(wav_file_path))\n",
    "            ## Play using android:\n",
    "            org, rec_in, rec_out, T = play_android(wav_file_path, fs_rec)\n",
    "            ## Play using PC connected speakers:\n",
    "            # org, rec_in, rec_out, fs = play_record_2mic(wav_file_path)\n",
    "            outbox_path = wav_file_path.replace(dataset, save_dir + \"/out\")\n",
    "            inbox_path = wav_file_path.replace(dataset, save_dir + \"/in\")\n",
    "            inbox_path_normalized = wav_file_path.replace(dataset, save_dir + \"/in_normalized\")\n",
    "            if save:\n",
    "                pathlib.Path(outbox_path).parent.mkdir(parents=True, exist_ok=True)\n",
    "                pathlib.Path(inbox_path).parent.mkdir(parents=True, exist_ok=True)\n",
    "                pathlib.Path(inbox_path_normalized).parent.mkdir(parents=True, exist_ok=True)\n",
    "                sf.write(inbox_path, rec_in, fs_rec)\n",
    "                rec_in_norm = rec_in / max(abs(rec_in)) * max(abs(org))\n",
    "                rec_out_norm = rec_out / max(abs(rec_out)) * max(abs(org))\n",
    "                sf.write(inbox_path_normalized, rec_in_norm, fs_rec)\n",
    "                sf.write(outbox_path, rec_out_norm, fs_rec)\n",
    "                # sf.write(outbox_path, rec_out, fs_rec)\n",
    "\n",
    "            get_transfer_function_scipy(org, rec_out_norm, rec_in_norm, fs_rec, fs_org, T, save, title=outbox_path)\n",
    "        except KeyboardInterrupt:\n",
    "            print('\\nPausing...  (Hit ENTER to continue, type quit to exit.)')\n",
    "            try:\n",
    "                response = input()\n",
    "                if response == 'quit':\n",
    "                    break\n",
    "                print('Resuming...')\n",
    "            except KeyboardInterrupt:\n",
    "                print('Resuming...')\n",
    "                continue\n",
    "\n",
    "def swept_sin(save, fs_org, fs_rec):\n",
    "    T = 3\n",
    "    # fs_org = 8000\n",
    "    t = np.arange(0, int(T * fs_org)) / fs_org\n",
    "    f0 = 7800\n",
    "    f1 = 100\n",
    "\n",
    "    w = chirp(t, f0=f0, f1=f1, t1=T, method='linear')\n",
    "    # print(w)\n",
    "    file_name = 'chirp_{}_{}_{}s_{}.wav'.format(f1, f0, T, fs_org)\n",
    "    # sf.write('../Dataset/chirp/'+file_name, w, fs_org)\n",
    "    # plt.plot(t,w)\n",
    "    # plt.show()\n",
    "    # w_fft = np.abs(fft.rfft(w, norm=\"ortho\"))\n",
    "    # freq = fft.rfftfreq(len(w), d=1/fs_org)\n",
    "    # plt.plot(freq, w_fft, alpha=0.8)\n",
    "    # plt.show()\n",
    "    # rec = sd.playrec(w, samplerate=fs, channels=2, blocking=True)\n",
    "\n",
    "    # file_name = os.path.basename(wav_file_path)\n",
    "    # subprocess.run([\"adb push ../Dataset/chirp/{} /sdcard/chirp/\".format(file_name)], shell=True)\n",
    "    subprocess.run([\n",
    "                       \"adb shell am start -a android.intent.action.VIEW -d file:///storage/emulated/0/chirp/{}  -t video/* -f 0x00008000\".format(\n",
    "                           file_name)], shell=True)\n",
    "    # subprocess.run([\"adb shell am start -a android.intent.action.VIEW -d file:///storage/emulated/0/{}  -t video/* -f 0x00008000\".format(\"id00012_21Uxsk56VDQ_00006_00000.wav\")], shell=True)\n",
    "\n",
    "    # fs = 8*1000\n",
    "    print('Start Recording at fs :')\n",
    "    fs = fs_rec\n",
    "    rec = sd.rec(int(T * fs), samplerate=fs, channels=2, blocking=True)\n",
    "    print(\"Recording Done!\")\n",
    "    rec_in = rec[:, 0]  # inside box\n",
    "    rec_out = rec[:, 1]  # outside box\n",
    "    rec_in_norm = rec_in / max(abs(rec_in)) * max(abs(w))\n",
    "    rec_out_norm = rec_out / max(abs(rec_out)) * max(abs(w))\n",
    "    # sf.write('../Dataset/Box_phone_chirp/direct_far_out_{}.wav'.format(fs), rec_out, fs)\n",
    "    if save:\n",
    "        sf.write('../Dataset/Box_phone_chirp/{}_{}_play{}_org.wav'.format(L_in, r, fs_org), w, fs)\n",
    "        sf.write('../Dataset/Box_phone_chirp/{}_{}_play{}_in.wav'.format(L_in, r, fs_org), rec_in, fs)\n",
    "        sf.write('../Dataset/Box_phone_chirp/{}_{}_play{}_out.wav'.format(L_in, r, fs_org), rec_out, fs)\n",
    "    # _, band_filter, bandpass_filter, _ = resonant_filter_torch('../Dataset/'+file_name, f0, Q, round(fs/2/f0), False)\n",
    "    # band_filter = band_filter.cpu().detach().numpy().flatten()\n",
    "    # bandpass_filter = bandpass_filter.cpu().detach().numpy().flatten()\n",
    "    get_transfer_function_scipy(w, rec_out_norm, rec_in_norm, fs, fs_org, T, save_fig=save,\n",
    "                                title='../Dataset/Box_phone_chirp/{}_{}_play{}_16k.wav'.format(L_in, d, fs_org))\n",
    "    # get_transfer_function_scipy(rec_in, band_filter, bandpass_filter, fs, save_fig=False, title='sin_swept')\n"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "start_time": "2023-04-30T16:00:53.978762Z",
     "end_time": "2023-04-30T16:00:53.980076Z"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "File duration =  5.52634375\n",
      "m_hamil.wav\n",
      "Starting: Intent { act=android.intent.action.VIEW dat=file:///storage/emulated/0/m_hamil.wav typ=video/* flg=0x8000 }\n",
      "Start Recording at fs = 16000:\n",
      "Recording Done!\n"
     ]
    }
   ],
   "source": [
    "import librosa\n",
    "import subprocess\n",
    "\n",
    "fs_rec = 16000\n",
    "wav_file_path = 'm_hamil.wav'\n",
    "data, fs_org = sf.read(wav_file_path)\n",
    "duration = librosa.get_duration(path=wav_file_path)\n",
    "print('File duration = ', duration)\n",
    "file_name = '/'.join(wav_file_path.split('/')[-3:])\n",
    "print(file_name)\n",
    "subprocess.run([\n",
    "    \"adb shell am start -a android.intent.action.VIEW -d file:///storage/emulated/0/{}  -t video/* -f 0x00008000\".format(\n",
    "        file_name)], shell=True)\n",
    "print('Start Recording at fs = {}:'.format(fs_rec))\n",
    "rec = sd.rec(int(duration * fs_rec), samplerate=fs_rec, channels=2, blocking=True, device=1)\n",
    "print(\"Recording Done!\")\n",
    "rec_in = rec[:, 0]  # inside box\n",
    "rec_out = rec[:, 1]  # outside box\n"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "start_time": "2023-04-30T16:16:26.237727Z",
     "end_time": "2023-04-30T16:16:32.041409Z"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "outputs": [],
   "source": [
    "import torchaudio\n",
    "import torch\n",
    "\n",
    "torchaudio.save('m_hamil_16k.wav', torch.from_numpy(rec.T), fs_rec)"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "start_time": "2023-04-30T16:16:48.156951Z",
     "end_time": "2023-04-30T16:16:48.160809Z"
    }
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}