.ipynb_checkpoints/anomaly-detection-tflite-conversion-checkpoint.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import tensorflow as tf\n",
    "from tensorflow.keras import models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print versions\n",
    "!python --version\n",
    "print('Numpy ' + np.__version__)\n",
    "print('TensorFlow ' + tf.__version__)\n",
    "print('Keras ' + tf.keras.__version__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Settings\n",
    "models_path = 'models'  # Where we can find the model files (relative path location)\n",
    "keras_model_name = 'fan_low_model'           # Will be given .h5 suffix\n",
    "tflite_model_name = 'fan_low_model'          # Will be given .tflite suffix\n",
    "c_model_name = 'fan_low_model'               # Will be given .h suffix\n",
    "\n",
    "\n",
    "# raw_scale = 100             # Multiply raw values to fit into integers\n",
    "# sensor_sample_rate = 200    # Hz\n",
    "# desired_sample_rate = 50    # Hz\n",
    "# sample_time = 0.64          # Time (sec) length of each sample\n",
    "# samples_per_file = 128      # Expected number of measurements in each file (truncate to this)\n",
    "\n",
    "# max_measurements = int(sample_time * sensor_sample_rate)\n",
    "# downsample_factor = int(samples_per_file / desired_sample_rate)\n",
    "# win_len = int(max_measurements / downsample_factor)\n",
    "\n",
    "# keras_model_name = 'fan_low_model'           # Will be given .h5 suffix\n",
    "# tflite_model_name = 'fan_low_model'          # Will be given .tflite suffix\n",
    "# c_model_name = 'fan_low_model'               # Will be given .h suffix\n",
    "# c_hann_name = 'hann_window'                  # Will be given .h suffix\n",
    "# sample_file_name = 'normal_anomaly_samples'  # Will be given .npz suffix\n",
    "# rep_dataset_name = 'normal_anomaly_test_set' # Will be given .npz suffix\n",
    "\n",
    "# print('Max measurements per file:', max_measurements)\n",
    "# print('Downsample factor:', downsample_factor)\n",
    "# print('Window length:', win_len)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load model\n",
    "model = models.load_model(keras_model_name + '.h5')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load test set\n",
    "npzfile = np.load(rep_dataset_name + '.npz')\n",
    "x_test = npzfile['x_test']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generator function that provides representative data samples for quantization\n",
    "def representative_dataset_gen():\n",
    "    for sample in x_test:\n",
    "        sample = np.expand_dims(sample.astype(np.float32), axis=0)\n",
    "        yield [sample]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert Keras model to a tflite model\n",
    "converter = tf.lite.TFLiteConverter.from_keras_model(model)\n",
    "#converter.optimizations = [tf.lite.Optimize.OPTIMIZE_FOR_SIZE]\n",
    "#converter.optimizations = [tf.lite.Optimize.DEFAULT]\n",
    "\n",
    "# Quantization settings\n",
    "#converter.representative_dataset = representative_dataset_gen\n",
    "\n",
    "# Convert and save\n",
    "tflite_model = converter.convert()\n",
    "open(tflite_model_name + '.tflite', 'wb').write(tflite_model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function: Convert some values into an array for C programming\n",
    "def create_c_lookup_table(array, var_type, var_name, line_limit=80):\n",
    "\n",
    "    c_str = ''\n",
    "\n",
    "    # Create header guard\n",
    "    c_str += '#ifndef ' + var_name.upper() + '_H\\n'\n",
    "    c_str += '#define ' + var_name.upper() + '_H\\n\\n'\n",
    "\n",
    "    # Add array length at top of file\n",
    "    c_str += 'const unsigned int ' + var_name + '_len = ' + str(len(array)) + ';\\n'\n",
    "\n",
    "    # Declare C variable\n",
    "    c_str += 'const ' + var_type + ' ' + var_name + '[] = {\\n'\n",
    "    \n",
    "    # Create string for the array\n",
    "    indent = '  '\n",
    "    array_str = indent\n",
    "    line_len = len(indent)\n",
    "    val_sep = ', '\n",
    "    for i, val in enumerate(array):\n",
    "\n",
    "        # Create a new line if string is over line limit\n",
    "        val_str = str(val)\n",
    "        if line_len + len(val_str) + len(val_sep) > line_limit:\n",
    "            array_str += '\\n' + indent\n",
    "            line_len = len(indent)\n",
    "\n",
    "        # Add value and separator\n",
    "        array_str += val_str\n",
    "        line_len += len(val_str)\n",
    "        if (i + 1) < len(array):\n",
    "            array_str += val_sep\n",
    "            line_len += len(val_sep)\n",
    "\n",
    "    # Add closing brace\n",
    "    c_str += array_str + '\\n};\\n\\n'\n",
    "\n",
    "    # Close out header guard\n",
    "    c_str += '#endif //' + var_name.upper() + '_H'\n",
    "\n",
    "    return c_str"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write TFLite model to a C source (or header) file\n",
    "with open(c_model_name + '.h', 'w') as file:\n",
    "    hex_array = [format(val, '#04x') for val in tflite_model]\n",
    "    file.write(create_c_lookup_table(hex_array, 'unsigned char', c_model_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a hanning window lookup table\n",
    "with open(c_hann_name + '.h', 'w') as file:\n",
    "    file.write(create_c_lookup_table(np.hanning(win_len), 'float', c_hann_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load samples for testing\n",
    "npzfile = np.load(sample_file_name + '.npz')\n",
    "normal_sample = npzfile['normal_sample']\n",
    "anomaly_sample = npzfile['anomaly_sample']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print out normal sample, truncated to 128 measurements\n",
    "print(\"X\")\n",
    "for i in normal_sample[0:128, 0]:\n",
    "    print(str(i), end=', ')\n",
    "print()\n",
    "print(\"Y\")\n",
    "for i in normal_sample[0:128, 1]:\n",
    "    print(str(i), end=', ')\n",
    "print()\n",
    "print(\"Z\")\n",
    "for i in normal_sample[0:128, 2]:\n",
    "    print(str(i), end=', ')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print out anomaly sample, truncated to 128 measurements\n",
    "print(\"X\")\n",
    "for i in anomaly_sample[0:128, 0]:\n",
    "    print(str(i), end=', ')\n",
    "print()\n",
    "print(\"Y\")\n",
    "for i in anomaly_sample[0:128, 1]:\n",
    "    print(str(i), end=', ')\n",
    "print()\n",
    "print(\"Z\")\n",
    "for i in anomaly_sample[0:128, 2]:\n",
    "    print(str(i), end=', ')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function: extract specified features (variances, MAD) from sample\n",
    "def extract_features(sample, max_measurements=0, scale=1):\n",
    "    \n",
    "    features = []\n",
    "    \n",
    "    # Truncate sample\n",
    "    if max_measurements == 0:\n",
    "        max_measurements = sample.shape[0]\n",
    "    sample = sample[0:max_measurements]\n",
    "    \n",
    "    # Scale sample\n",
    "    sample = scale * sample\n",
    "    \n",
    "    # Compute a windowed FFT of each axis in the sample (leave off DC)\n",
    "    sample = sample[::downsample_factor, :]  # Downsample\n",
    "    sample = np.floor(sample)                # Round down to int\n",
    "    hann_window = np.hanning(sample.shape[0])\n",
    "    for i, axis in enumerate(sample.T):\n",
    "        fft = abs(np.fft.rfft(axis * hann_window))\n",
    "        features.append(fft[1:])  # Leave off DC\n",
    "    \n",
    "    return np.floor(np.array(features).flatten())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute windowed FFT of each axis in normal sample\n",
    "normal_fft = extract_features(normal_sample, max_measurements, raw_scale)\n",
    "print(normal_fft)\n",
    "plt.plot(normal_fft)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute windowed FFT of each axis in anomaly sample\n",
    "anomaly_fft = extract_features(anomaly_sample, max_measurements, raw_scale)\n",
    "print(anomaly_fft)\n",
    "plt.plot(anomaly_fft)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copy in Arduino FFTs for normal sample\n",
    "arduino_fft_normal_x = [11, 1, 0, 2, 0, 0, 0, 1, 0, 0, 1, 1, 3, 1, 0, 0, 1, 0, 1, 0, 1, 0, 0, 2, 1, 1, 0, 1, 1, 2, 3, 0]\n",
    "arduino_fft_normal_y = [330, 3, 4, 13, 10, 1, 0, 1, 0, 0, 4, 24, 32, 8, 2, 3, 2, 1, 2, 0, 3, 1, 0, 2, 0, 1, 1, 1, 2, 5, 2, 2]\n",
    "arduino_fft_normal_z = [2493, 30, 9, 5, 0, 4, 3, 4, 2, 3, 7, 29, 42, 19, 0, 0, 1, 1, 0, 0, 5, 6, 1, 2, 2, 1, 2, 3, 3, 0, 1, 4]\n",
    "arduino_fft_anomaly_x = [54, 2, 1, 1, 0, 1, 0, 0, 1, 0, 1, 3, 3, 0, 0, 0, 1, 0, 0, 1, 1, 2, 1, 0, 0, 1, 2, 2, 1, 0, 4, 3]\n",
    "arduino_fft_anomaly_y = [325, 1, 4, 1, 1, 3, 1, 1, 1, 1, 0, 7, 10, 4, 0, 0, 1, 0, 0, 0, 0, 4, 2, 0, 0, 0, 1, 1, 0, 1, 0, 0]\n",
    "arduino_fft_anomaly_z = [2490, 22, 15, 3, 10, 3, 0, 3, 3, 4, 1, 17, 22, 6, 1, 0, 2, 2, 1, 1, 4, 5, 2, 1, 1, 3, 1, 1, 1, 0, 2, 1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create feature set from Arduino FFTs\n",
    "arduino_fft_normal = np.array([arduino_fft_normal_x, arduino_fft_normal_y, arduino_fft_normal_z]).flatten()\n",
    "arduino_fft_anomaly = np.array([arduino_fft_anomaly_x, arduino_fft_anomaly_y, arduino_fft_anomaly_z]).flatten()\n",
    "print(arduino_fft_normal.shape)\n",
    "plt.plot(arduino_fft_normal)\n",
    "plt.figure()\n",
    "plt.plot(arduino_fft_anomaly)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test arduino FFT normal in full model NN\n",
    "input_tensor = arduino_fft_normal.reshape(1, -1)\n",
    "print(input_tensor)\n",
    "predictions = model.predict(input_tensor)\n",
    "print(predictions)\n",
    "plt.plot(predictions[0])\n",
    "mse = np.mean(np.power(input_tensor - predictions, 2), axis=1)\n",
    "print(mse)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test arduino FFT anomaly in full model NN\n",
    "input_tensor = arduino_fft_anomaly.reshape(1, -1)\n",
    "print(input_tensor)\n",
    "predictions = model.predict(input_tensor)\n",
    "print(predictions)\n",
    "plt.plot(predictions[0])\n",
    "mse = np.mean(np.power(input_tensor - predictions, 2), axis=1)\n",
    "print(np.power(input_tensor - predictions, 2))\n",
    "print(mse)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Arduino NN output for normal sample\n",
    "arduino_feature_set = [11.00, 1.00, 0.00, 2.00, 0.00, 0.00, 0.00, 1.00, 0.00, 0.00, 1.00, 1.00, 3.00, 1.00, 0.00, 0.00, 1.00, 0.00, 1.00, 0.00, 1.00, 0.00, 0.00, 2.00, 1.00, 1.00, 0.00, 1.00, 1.00, 2.00, 3.00, 0.00, 330.00, 3.00, 4.00, 13.00, 10.00, 1.00, 0.00, 1.00, 0.00, 0.00, 4.00, 24.00, 32.00, 8.00, 2.00, 3.00, 2.00, 1.00, 2.00, 0.00, 3.00, 1.00, 0.00, 2.00, 0.00, 1.00, 1.00, 1.00, 2.00, 5.00, 2.00, 2.00, 2493.00, 30.00, 9.00, 5.00, 0.00, 4.00, 3.00, 4.00, 2.00, 3.00, 7.00, 29.00, 42.00, 19.00, 0.00, 0.00, 1.00, 1.00, 0.00, 0.00, 5.00, 6.00, 1.00, 2.00, 2.00, 1.00, 2.00, 3.00, 3.00, 0.00, 1.00, 4.00]\n",
    "arduino_predictions = [-0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, -0.00, 1.00, 1.00, 0.00, 1.00, 1.00, 2.00, 3.00, 0.00, 330.00, 3.00, 4.00, 13.00, 10.00, 1.00, 0.00, 1.00, 0.00, 0.00, 4.00, 24.00, 32.00, 8.00, 2.00, 3.00, 2.00, 1.00, 2.00, 0.00, 3.00, 1.00, 0.00, 2.00, 0.00, 1.00, 1.00, 1.00, 2.00, 5.00, 2.00, 2.00, 2493.00, 30.00, 9.00, 5.00, 0.00, 4.00, 3.00, 4.00, 2.00, 3.00, 7.00, 29.00, 42.00, 19.00, 0.00, 0.00, 1.00, 1.00, 0.00, 0.00, 5.00, 6.00, 1.00, 2.00, 2.00, 1.00, 2.00, 3.00, 3.00, 0.00, 1.00, 4.00]\n",
    "mse = np.mean(np.power(np.array(arduino_feature_set) - np.array(arduino_predictions), 2))\n",
    "print('MSE:', mse)\n",
    "\n",
    "plt.plot(arduino_feature_set)\n",
    "plt.title('Input feature set')\n",
    "plt.figure()\n",
    "plt.plot(arduino_predictions)\n",
    "plt.title('Predictions')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Arduino NN output for anomaly sample\n",
    "arduino_feature_set = [54.00, 2.00, 1.00, 1.00, 0.00, 1.00, 0.00, 0.00, 1.00, 0.00, 1.00, 3.00, 3.00, 0.00, 0.00, 0.00, 1.00, 0.00, 0.00, 1.00, 1.00, 2.00, 1.00, 0.00, 0.00, 1.00, 2.00, 2.00, 1.00, 0.00, 4.00, 3.00, 325.00, 1.00, 4.00, 1.00, 1.00, 3.00, 1.00, 1.00, 1.00, 1.00, 0.00, 7.00, 10.00, 4.00, 0.00, 0.00, 1.00, 0.00, 0.00, 0.00, 0.00, 4.00, 2.00, 0.00, 0.00, 0.00, 1.00, 1.00, 0.00, 1.00, 0.00, 0.00, 2490.00, 22.00, 15.00, 3.00, 10.00, 3.00, 0.00, 3.00, 3.00, 4.00, 1.00, 17.00, 22.00, 6.00, 1.00, 0.00, 2.00, 2.00, 1.00, 1.00, 4.00, 5.00, 2.00, 1.00, 1.00, 3.00, 1.00, 1.00, 1.00, 0.00, 2.00, 1.00] \n",
    "arduino_predictions = [0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 0.00, 1.00, 2.00, 2.00, 1.00, 0.00, 4.00, 3.00, 325.00, 1.00, 4.00, 1.00, 1.00, 3.00, 1.00, 1.00, 1.00, 1.00, 0.00, 7.00, 10.00, 4.00, 0.00, 0.00, 1.00, 0.00, 0.00, 0.00, 0.00, 4.00, 2.00, 0.00, 0.00, 0.00, 1.00, 1.00, 0.00, 1.00, 0.00, 0.00, 2490.00, 22.00, 15.00, 3.00, 10.00, 3.00, 0.00, 3.00, 3.00, 4.00, 1.00, 17.00, 22.00, 6.00, 1.00, 0.00, 2.00, 2.00, 1.00, 1.00, 4.00, 5.00, 2.00, 1.00, 1.00, 3.00, 1.00, 1.00, 1.00, 0.00, 2.00, 1.00] \n",
    "mse = np.mean(np.power(np.array(arduino_feature_set) - np.array(arduino_predictions), 2))\n",
    "print('MSE:', mse)\n",
    "\n",
    "plt.plot(arduino_feature_set)\n",
    "plt.title('Input feature set')\n",
    "plt.figure()\n",
    "plt.plot(arduino_predictions)\n",
    "plt.title('Predictions')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}