record_old_working.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "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": "markdown",
   "metadata": {},
   "source": [
    "# TODO\n",
    "Figure out cd, pwd, problem with Dataset/chirp folder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-04-30T16:00:53.980076Z",
     "start_time": "2023-04-30T16:00:53.978762Z"
    },
    "collapsed": false
   },
   "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",
    "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",
    "    \n",
    "    T = 3\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('/home/gabe/Work/audio_recorder/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 /home/gabe/Work/audio_recorder/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 audio/* -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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-04-30T16:16:32.041409Z",
     "start_time": "2023-04-30T16:16:26.237727Z"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/home/gabe/Work/audio_recorder/Dataset/chirp/chirp_100_7800_3s_16000.wav: 1 file pushed. 0.1 MB/s (96044 bytes in 0.628s)\n",
      "Starting: Intent { act=android.intent.action.VIEW dat=file:///storage/emulated/0/chirp/chirp_100_7800_3s_16000.wav typ=audio/* flg=0x8000 }\n",
      "Start Recording at fs :\n",
      "Recording Done!\n"
     ]
    }
   ],
   "source": [
    "import librosa\n",
    "import subprocess\n",
    "\n",
    "fs_rec = 16000\n",
    "fs_org = 16000\n",
    "swept_sin(True, fs_org, fs_rec)\n",
    "\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",
    "\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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-04-30T16:16:48.160809Z",
     "start_time": "2023-04-30T16:16:48.156951Z"
    },
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import torchaudio\n",
    "import torch\n",
    "\n",
    "torchaudio.save('m_hamil_16k.wav', torch.from_numpy(rec.T), fs_rec)"
   ]
  }
 ],
 "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}