lab.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import librosa\n",
    "import numpy as np\n",
    "from numpy.fft import rfft\n",
    "from numpy import pi\n",
    "from matplotlib import pyplot as plt\n",
    "from IPython.display import Audio\n",
    "import cmath\n",
    "import scipy\n",
    "from cache_no_hash import cache\n",
    "from blindDescend import blindDescend\n",
    "from yin import yin\n",
    "from harmonicSynth import HarmonicSynth, Harmonic\n",
    "\n",
    "TWO_PI = np.pi * 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "SR = 22050\n",
    "NYQUIST = SR // 2\n",
    "PAGE_LEN = 1024\n",
    "DTYPE = np.float32\n",
    "N_HARMONICS = 8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "HANN = scipy.signal.get_window('hann', PAGE_LEN, True)\n",
    "IMAGINARY_LADDER = np.linspace(0, TWO_PI * 1j, PAGE_LEN)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sino(freq, length):\n",
    "    return np.sin(np.arange(length) * freq * TWO_PI / SR)\n",
    "\n",
    "def playHard(data):\n",
    "    return Audio(data, rate = SR)\n",
    "def play(data, soft = .02):\n",
    "    t = np.concatenate([data, [1]])\n",
    "    length = round(soft * SR)\n",
    "    t[:length ] = np.multiply(t[:length ], np.linspace(0, 1, length))\n",
    "    t[-length:] = np.multiply(t[-length:], np.linspace(1, 0, length))\n",
    "    return playHard(t)\n",
    "\n",
    "def findPeaks(energy):\n",
    "    slope = np.sign(energy[1:] - energy[:-1])\n",
    "    extrema = slope[1:] - slope[:-1]\n",
    "    return np.argpartition(\n",
    "        (extrema == -2) * energy[1:-1], - N_HARMONICS,\n",
    "    )[- N_HARMONICS:] + 1\n",
    "\n",
    "def sft(signal, freq_bin):\n",
    "    # Slow Fourier Transform\n",
    "    return np.abs(np.sum(signal * np.exp(IMAGINARY_LADDER * freq_bin))) / PAGE_LEN\n",
    "\n",
    "def refineGuess(guess, signal):\n",
    "    def loss(x):\n",
    "        if x < 0:\n",
    "            return 0\n",
    "        return - sft(signal, x)\n",
    "    freq_bin, loss = blindDescend(loss, .01, .4, guess)\n",
    "    return freq_bin * SR / PAGE_LEN, - loss\n",
    "\n",
    "def widePlot(h = 3, w = 12):\n",
    "    plt.gcf().set_size_inches(w, h)\n",
    "\n",
    "    \n",
    "def spectro(signal, do_wide = True, trim = 130):\n",
    "    energy = np.abs(rfft(signal * HANN))\n",
    "    plt.plot(energy[:trim])\n",
    "    if do_wide:\n",
    "        widePlot()\n",
    "\n",
    "def concatSynth(synth, harmonics, n):\n",
    "    buffer = []\n",
    "    for i in range(n):\n",
    "        synth.eat(harmonics)\n",
    "        buffer.append(synth.mix())\n",
    "    return np.concatenate(buffer)\n",
    "\n",
    "def pitch2freq(pitch):\n",
    "    return np.exp((pitch + 36.37631656229591) * 0.0577622650466621)\n",
    "\n",
    "def freq2pitch(f):\n",
    "    return np.log(f) * 17.312340490667562 - 36.37631656229591"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy.interpolate import interp1d"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw, SR = librosa.load('range.wav')\n",
    "raw = raw[30000:-10000]\n",
    "raw = raw[:(raw.size // PAGE_LEN) * PAGE_LEN]\n",
    "print(SR)\n",
    "play(raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = raw[PAGE_LEN*31:PAGE_LEN*32] * HANN\n",
    "spec = np.log(np.abs(rfft(a)))\n",
    "plt.plot(spec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f0 = yin(a, SR, PAGE_LEN)\n",
    "tops = np.array([sft(a, f0 * i / SR * PAGE_LEN) * PAGE_LEN for i in range(20)])\n",
    "f = interp1d([f0 * i for i in range(20)], tops)\n",
    "freqs = np.linspace(0, SR/2, int(PAGE_LEN/2)+1)\n",
    "plt.plot(freqs, np.exp(spec))\n",
    "plt.plot(freqs[:300], [f(x) for x in freqs[:300]])\n",
    "[plt.axvline(f0 * i) for i in range(20)]\n",
    "# plt.axis([0, 6000, -14, 5])\n",
    "widePlot()\n",
    "plt.yscale('log')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "f0 = yin(a, SR, PAGE_LEN)\n",
    "tops = np.log([sft(a, f0 * i / SR * PAGE_LEN) * PAGE_LEN for i in range(20)])\n",
    "f = interp1d([f0 * i for i in range(20)], tops)\n",
    "freqs = np.linspace(0, SR/2, int(PAGE_LEN/2)+1)\n",
    "plt.plot(freqs, np.exp(spec))\n",
    "plt.plot(freqs[:300], np.exp([f(x) for x in freqs[:300]]))\n",
    "[plt.axvline(f0 * i) for i in range(20)]\n",
    "widePlot()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "freqs = np.linspace(0, SR/2, int(PAGE_LEN/2)+1)\n",
    "plt.plot(freqs, spec)\n",
    "[plt.axvline(f0 * i) for i in range(20)]\n",
    "widePlot()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getEnvelope(signal, len_signal):\n",
    "    f0 = yin(signal, SR, len_signal)\n",
    "    harmonics_f = np.arange(0, NYQUIST, f0)\n",
    "    harmonics_a = [sft(signal, f_bin) for f_bin in harmonics_f / SR * len_signal]\n",
    "    harmonics_a[0] = harmonics_a[1]\n",
    "    f = interp1d(harmonics_f, harmonics_a)\n",
    "    def envelope(x):\n",
    "        try:\n",
    "            return f(x)\n",
    "        except ValueError:\n",
    "            return 0\n",
    "    return envelope"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def video(width = 100):\n",
    "    cursor = 0\n",
    "    X = []\n",
    "    Y = []\n",
    "    C = []\n",
    "    y = 0\n",
    "    while cursor + PAGE_LEN < raw.size:\n",
    "        signal = raw[cursor : cursor + PAGE_LEN]\n",
    "        cursor += PAGE_LEN\n",
    "        y += 1\n",
    "        envelope = getEnvelope(signal, PAGE_LEN)\n",
    "        for x in range(0, 6000, 30):\n",
    "            X.append(x)\n",
    "            Y.append(y)\n",
    "            t = (np.log(envelope(x)) + 8) / (3 + 8)\n",
    "            t = min(1, max(0, t))\n",
    "            C.append(np.array([1-t, 1-t, t]))\n",
    "    plt.scatter(x = X, y = Y, c = C, s = 2)\n",
    "    widePlot(6, 10)\n",
    "video()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## reconstruction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def reconstruct(n_harmos = 8):\n",
    "    hs = HarmonicSynth(\n",
    "        n_harmos, SR, PAGE_LEN, DTYPE, STUPID_MATCH = True, \n",
    "        DO_SWIPE = True, CROSSFADE_RATIO = .3, \n",
    "    )\n",
    "    cursor = 0\n",
    "    buffer = []\n",
    "    while cursor + PAGE_LEN < raw.size:\n",
    "        signal = raw[cursor : cursor + PAGE_LEN]\n",
    "        cursor += PAGE_LEN\n",
    "        envelope = getEnvelope(signal, PAGE_LEN)\n",
    "        f0 = yin(signal, SR, PAGE_LEN)\n",
    "        harmonics_f = np.arange(1, n_harmos + 1) * f0\n",
    "        harmonics = [Harmonic(f, envelope(f)) for f in harmonics_f]\n",
    "        hs.eat(harmonics)\n",
    "        buffer.append(hs.mix())\n",
    "    return np.concatenate(buffer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "play(raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "play(reconstruct(8))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "play(reconstruct(80))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conclusion: 8 is enough"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## pitch transfer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pitchTransfer(freq = 440, n_harmos = 8):\n",
    "    hs = HarmonicSynth(\n",
    "        n_harmos, SR, PAGE_LEN, DTYPE, STUPID_MATCH = True, \n",
    "        DO_SWIPE = True, CROSSFADE_RATIO = .3, \n",
    "    )\n",
    "    cursor = 0\n",
    "    buffer = []\n",
    "    while cursor + PAGE_LEN < raw.size:\n",
    "        signal = raw[cursor : cursor + PAGE_LEN]\n",
    "        cursor += PAGE_LEN\n",
    "        envelope = getEnvelope(signal, PAGE_LEN)\n",
    "        f0 = freq\n",
    "        harmonics_f = np.arange(1, n_harmos + 1) * f0\n",
    "        harmonics = [Harmonic(f, envelope(f)) for f in harmonics_f]\n",
    "        hs.eat(harmonics)\n",
    "        buffer.append(hs.mix())\n",
    "    return np.concatenate(buffer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "play(pitchTransfer(330))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## is that you"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw, SR = librosa.load('isthatyou.wav')\n",
    "raw = raw[25000:-10000]\n",
    "raw = raw[:(raw.size // PAGE_LEN) * PAGE_LEN]\n",
    "print(SR)\n",
    "play(raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "play(reconstruct(60))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "play(pitchTransfer(220, 60))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conclusion: PAGE_LEN = 256; N_HARMONICS = 60"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## white noise for unvoiced"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "from scipy.stats import norm\n",
    "\n",
    "def whitenoise(size):\n",
    "    return norm.rvs(0, 1, size)\n",
    "def unvoiced():\n",
    "    spectrum = whitenoise(PAGE_LEN//2 + 1)\n",
    "#     spectrum = np.multiply(spectrum, scipy.signal.get_window('hann', PAGE_LEN//2+1, False))\n",
    "    return np.fft.irfft(spectrum)\n",
    "play(np.concatenate([unvoiced() for _ in range(20)]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def voiced():\n",
    "    spectrum = np.zeros((PAGE_LEN//2 + 1, ))\n",
    "    spectrum[round(220 / SR * PAGE_LEN)] = 1\n",
    "    spectrum[round(440 / SR * PAGE_LEN)] = .8\n",
    "    spectrum[round(660 / SR * PAGE_LEN)] = .95\n",
    "#     spectrum = np.multiply(spectrum, scipy.signal.get_window('hann', PAGE_LEN//2+1, False))\n",
    "    return np.fft.irfft(spectrum)\n",
    "play(np.concatenate([voiced() for _ in range(20)]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from collections import namedtuple\n",
    "\n",
    "Harmonic = namedtuple('Harmonic', ['freq', 'mag', 'phase'])\n",
    "TWO_PI_J = 2j * np.pi\n",
    "\n",
    "def ifftSynth(harmonics, SR, PAGE_LEN):\n",
    "    spectrum = np.zeros((PAGE_LEN // 2 + 1, ), dtype=np.complex)\n",
    "    phases = []\n",
    "    for freq, mag, phase in harmonics:\n",
    "        try:\n",
    "            freq_bin = round(freq / SR * PAGE_LEN)\n",
    "            spectrum[freq_bin] += mag * np.exp(phase * TWO_PI_J)\n",
    "        except IndexError:\n",
    "            phases.append(0)\n",
    "            continue\n",
    "        else:\n",
    "            phases.append((phase + PAGE_LEN / SR * (freq_bin * SR / PAGE_LEN)) % 1)\n",
    "    return np.fft.irfft(spectrum) * PAGE_LEN, phases\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "phases = [0, 0, 0]\n",
    "buffer = []\n",
    "for i in range(30):\n",
    "    signal, phases = ifftSynth([\n",
    "        Harmonic(310, 1, 0), \n",
    "#         Harmonic(400, .7, 0), \n",
    "#         Harmonic(600, .9, 0), \n",
    "    ], SR, PAGE_LEN)\n",
    "    buffer.append(signal)\n",
    "play(np.concatenate(buffer))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Damn. All mid-bin freqs divide the page perfectly! no need to phase"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for t in np.linspace(0, 1, 7):\n",
    "    spec = np.abs(rfft(sino(195 + t * 21, PAGE_LEN)))[6:16]\n",
    "    print(np.sum(spec ** 2))\n",
    "    plt.plot(spec ** 2, c=np.array([(1-t)**.5, t**.5, (1-t)**.5]))\n",
    "widePlot(5, 10)"
   ]
  }
 ],
 "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.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}