keras_demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "## Example curve prediction script\n",
    "In this notebook we will try to predict the function coefficients."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "%matplotlib widget\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import tensorflow as tf\n",
    "\n",
    "from tensorflow import keras\n",
    "\n",
    "from sklearn.model_selection import train_test_split\n",
    "from matplotlib import pyplot as plt\n",
    "from sklearn import preprocessing\n",
    "\n",
    "from generate_TF import GenerateTF, get_freq\n",
    "from scipy.optimize import curve_fit\n",
    "from prettytable import PrettyTable\n",
    "\n",
    "from pytorchClassifiers import get_keras_nn, plot_history\n",
    "\n",
    "avg_pool1d = keras.layers.AveragePooling1D\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "# Load the data\n",
    "df = pd.read_pickle('./data/tf-ampl-response-82000-noise0.1.pkl')\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "# Visualize the data\n",
    "for index, row in df.sample(n=3).iterrows():\n",
    "    y = row['amplitude']\n",
    "    x = get_freq(row['fmax'], row['np'])\n",
    "    plt.plot(x, y)\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "# Extract the target variables\n",
    "phase = df.pop('phase')\n",
    "gain = df.pop('gain')\n",
    "\n",
    "# All fmax, np should be equal\n",
    "fmax = df.iloc[0].fmax\n",
    "NP = df.iloc[0].np\n",
    "df.drop(columns=['fmax', 'np'], inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "# target_orig is the vector with the originale phase, gain labels\n",
    "target_orig = np.array((phase, gain), dtype=np.float32).T\n",
    "\n",
    "target_scaler = preprocessing.StandardScaler().fit(target_orig)\n",
    "# target is scaled, better for training\n",
    "target = target_scaler.transform(target_orig)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load the models\n",
    "model1 = keras.models.load_model('models/keras/regression/phase_best')\n",
    "model2 = keras.models.load_model('models/keras/regression/gain_best')\n",
    "print(model1.summary())\n",
    "print(model2.summary())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "def curve_fit_deluxe(func, freq, sample, trim_edges=0, kernel_size=1, stride=1, **kwargs):\n",
    "    # center crop sample\n",
    "    if trim_edges > 0:\n",
    "        freq, sample = freq[trim_edges:-trim_edges], sample[trim_edges:-trim_edges]\n",
    "    # prepare the shapes for avg_pooling\n",
    "    freq = freq.reshape(1, -1, 1)\n",
    "    sample = sample.reshape(1, -1, 1)\n",
    "    # perform average pooling\n",
    "    freq = avg_pool1d(pool_size=kernel_size, strides=stride)(freq).numpy().flatten()\n",
    "    sample = avg_pool1d(pool_size=kernel_size, strides=stride)(sample).numpy().flatten()\n",
    "    # pass to curve_fit\n",
    "    return curve_fit(func, freq, sample, **kwargs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def classify(phase, gain):\n",
    "    phase, gain = float(phase), float(gain)\n",
    "\n",
    "    # table = ['Param', 'True', 'NeuralNet', 'Optimizer']\n",
    "    table = []\n",
    "    \n",
    "    gen_tf_noise = GenerateTF(fb_attn_index=3, noise_amplitude=0.1, with_noise=True)\n",
    "    x = gen_tf_noise.frequency\n",
    "\n",
    "    gen_tf_no_noise = GenerateTF(fb_attn_index=3, with_noise=False)\n",
    "\n",
    "    # I get one input with given phase, gain from X_validate or y_validate\n",
    "    ampl, resp = gen_tf_noise(x, phase, gain, return_response=True)\n",
    "    # ampl_scaled = torch.tensor(data_scaler.transform([ampl]), dtype=torch.float32)\n",
    "    resp = np.array([resp.real, resp.imag]).T.reshape(1, -1, 2)\n",
    "\n",
    "    # make predictions with the NN\n",
    "    pred_phase = model1(resp,training=False).numpy()\n",
    "    pred_gain = model2(resp, training=False).numpy()\n",
    "    pred_scaled = np.array([pred_phase, pred_gain]).reshape(1, -1)\n",
    "\n",
    "    # Descale them to get original values\n",
    "    pred_descaled = target_scaler.inverse_transform(pred_scaled)\n",
    "    pred_phase, pred_gain = float(pred_descaled[0, 0]), float(pred_descaled[0, 1])\n",
    "\n",
    "    # Get optimizers results\n",
    "    (opt_phase, opt_gain), _ = curve_fit_deluxe(gen_tf_no_noise, x, ampl, trim_edges=130,\n",
    "                                                kernel_size=4, stride=1,\n",
    "                                                bounds=([-20, 0.001], [20, 0.004]), method='trf')\n",
    "\n",
    "    # I add them to the table\n",
    "\n",
    "    table.append(['phase', np.round(phase, 2), np.round(pred_phase, 2), np.round(opt_phase, 2)])\n",
    "    table.append(['gain', np.round(gain, 4), np.round(pred_gain, 4), np.round(opt_gain, 4)])\n",
    "\n",
    "    # I plot them\n",
    "    fig = plt.figure()\n",
    "    p = plt.plot(x, gen_tf_no_noise(x, phase, gain), label=f'True', ls='-', color='black')\n",
    "    plt.plot(x, gen_tf_noise(x, phase, gain), label=f'True, with Noise', ls='-', color='black', alpha=0.5)\n",
    "    plt.plot(x, gen_tf_no_noise(x, pred_phase, pred_gain), label=f'NeuralNet',\n",
    "             ls='--', color='tab:orange')\n",
    "    plt.plot(x, gen_tf_no_noise(x, opt_phase, opt_gain), label=f'Optimizer',\n",
    "             ls=':', color='tab:green')\n",
    "    plt.xlabel('Frequency')\n",
    "    plt.ylabel('Amplitude')\n",
    "    plt.legend()\n",
    "    # print(table.get_string())\n",
    "    return fig, table\n",
    "\n",
    "fig, table = classify(-15, 0.002)\n",
    "print(table)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import gradio as gr\n",
    "\n",
    "# print(gain_encoder.classes_.astype(str).tolist())\n",
    "# build the gradio interface\n",
    "slider = gr.Slider(label=f'Phase', minimum=-20.,\n",
    "                   maximum=20., step=1.)\n",
    "dropdown = gr.Dropdown(label='Gain',\n",
    "                       choices=np.round(np.geomspace(\n",
    "                           0.001, 0.004, num=20, endpoint=True), 4).tolist(),\n",
    "                       type=\"value\")\n",
    "interface = gr.Interface(fn=classify,\n",
    "                         inputs=[slider,\n",
    "                                 dropdown,\n",
    "                                 ],\n",
    "                         outputs=[\n",
    "                             gr.Plot(label='Graphical Evaluation'),\n",
    "                             gr.DataFrame(label='Numerical Evaluation',\n",
    "                                          headers=['Param', 'True',\n",
    "                                                   'NeuralNet', 'Optimizer'],\n",
    "                                          datatype=['str', 'number',\n",
    "                                                    'number', 'number'],\n",
    "                                          row_count=2, col_count=(4, 'fixed')\n",
    "                                          )\n",
    "                         ],\n",
    "                         interpretation=\"default\",\n",
    "                         )\n",
    "\n",
    "interface.launch(debug=False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.7 ('base')",
   "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.9.7"
  },
  "vscode": {
   "interpreter": {
    "hash": "79ab8fd01a8cec42884b8b2a5d7fb4751c5402d97e9e61d151ed5c6a6352873c"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}