02_Arduino_Nano_Inference.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "17d0dacf",
   "metadata": {},
   "source": [
    "<h1 style=\"text-align: center;\">Emotion Recognition Inference with Arduino Nano 33 BLE Sense</h1>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1cdd243c",
   "metadata": {},
   "source": [
    "## Introduction\n",
    "In this notebook, we demonstrate the deployment of the speech emotion recognition (SER) model, previously developed and converted into a TinyML model in [this notebook](https://github.com/Hannibal0420/Speech-Emotion-Recognition-TinyML/blob/main/01_TFLite_Model_Preparation.ipynb), with an Arduino Nano 33 BLE Sense microcontroller, a compact and energy-efficient board that is ideal for implementing real-time applications. The primary objective of this notebook is to showcase the process of processing audio data from a microphone connected to the board and running inferences on the TensorFlow Lite (TFLite) model designed to classify four emotions: Happy, Surprised, Neutral, and Unpleasant.\n",
    "\n",
    "We begin by setting up a serial connection with the Arduino and calibrating the microphone for reliable data collection. Next, we define asynchronous functions for processing the data and running the model. We also measure the model's inference time to evaluate its performance on the microcontroller.\n",
    "\n",
    "By the end of this notebook, you will have a better understanding of how to implement an SER model with a resource-constrained device, opening the door to applications in the fields of affective computing and IoT."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d1a6cc0e",
   "metadata": {},
   "source": [
    "## Setting Up\n",
    "1. Connect your Arduino 33 BLE Sense to your computer using a USB cable.\n",
    "2. Open the Arduino IDE on your computer and select the correct board and port under the \"Tools\" menu.\n",
    "3. Upload the [Data Streaming Code](https://github.com/Hannibal0420/Speech-Emotion-Recognition-TinyML/blob/main/data_streaming.ino) provided in the project repository to your Arduino board.\n",
    "4. After uploading, open the Serial Monitor in the Arduino IDE and make sure the baud rate is set to 38400.\n",
    "5. Press the reset button on the Arduino board to initialize it with the greeting message \"Welcome to the data streaming...\". If you see this message, congratulations! You can move on to the next step.\n",
    "\n",
    "> **Note:** <br>Sometimes the board may go to sleep. If you experience any issues, press the reset button on the Arduino board again before running the code below.\n",
    "\n",
    "## Importing Libraries\n",
    "Include the necessary libraries for audio processing, communication with the Arduino, and TinyML inference in real time:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ab700d19",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023-03-30 17:30:48.795792: I tensorflow/core/platform/cpu_feature_guard.cc:193] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 FMA\n",
      "To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/dev/cu.BLTH - n/a\n",
      "/dev/cu.Bluetooth-Incoming-Port - n/a\n",
      "/dev/cu.usbmodem14401 - Nano 33 BLE\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import time\n",
    "from serial.tools import list_ports\n",
    "\n",
    "import librosa\n",
    "import librosa.display\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import serial\n",
    "from IPython.display import Audio, display, HTML\n",
    "\n",
    "import tensorflow as tf\n",
    "from tensorflow import keras\n",
    "\n",
    "\n",
    "ports = list_ports.comports()\n",
    "for port in ports:\n",
    "    print(port)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c934ee6b",
   "metadata": {},
   "source": [
    "## Initialization\n",
    "In this section, the serial connection with the Arduino is established, and the microphone is calibrated to normalize the input data within a specific range for improved processing and feature extraction."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "953c728c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Type 'x' to Activate Arduino: x\n",
      "Success Activated!\n",
      "Open the Serial Monitor to check if it is working, then close it.\n"
     ]
    }
   ],
   "source": [
    "SERIAL_PORT = '/dev/cu.usbmodem14401'\n",
    "BAUD_RATE = 38400\n",
    "\n",
    "def arduino_activate():\n",
    "    try:\n",
    "        arduino = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1)\n",
    "        time.sleep(2)  # Allow time for connection to establish\n",
    "    except serial.SerialException as e:\n",
    "        print(f\"Failed to connect to serial port {SERIAL_PORT}: \\n{e}\")\n",
    "        return None\n",
    "\n",
    "    command = input(\"Type 'x' to Activate Arduino: \")\n",
    "    if command.lower() == 'x':\n",
    "        arduino.write(command.encode('utf-8'))\n",
    "        print(\"Success Activated!\")\n",
    "        print(\"Open the Serial Monitor to check if it is working, then close it.\")\n",
    "    return arduino\n",
    "\n",
    "\n",
    "arduino = arduino_activate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "19b4d4a5",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Calibrating: Please Speak to the Microphone\n",
      "Normalized signal from the range (-553.0, 839.0) to (-1, 1)\n"
     ]
    }
   ],
   "source": [
    "def arduino_read(buffer, buffer_size, overlapping, norm=(None, None), max_attempts=10):\n",
    "    buffer = np.roll(buffer, overlapping)\n",
    "    num_data = buffer_size - overlapping\n",
    "    for i in range(num_data):\n",
    "        decoded_data = ''\n",
    "        attempts = 0\n",
    "        while decoded_data == '':\n",
    "            arduino_data = arduino.readline()\n",
    "            decoded_data = arduino_data[:len(arduino_data)].decode(\"utf-8\").strip('\\r\\n')\n",
    "            attempts += 1\n",
    "            if attempts >= max_attempts:\n",
    "                print('Fail to Retrieve Data...')\n",
    "                break\n",
    "        if norm[0] is not None:\n",
    "            decoded_data = normalize(int(decoded_data), 1, -1, norm[0], norm[1])\n",
    "        try:\n",
    "            buffer[i+overlapping] = decoded_data\n",
    "        except ValueError:\n",
    "            print(\"Open the Serial Monitor to check if it is working. If it's not, press the reset button and rerun 'arduino_activate' funciton again.\")\n",
    "    return buffer\n",
    "\n",
    "def normalize(array, new_max, new_min, old_max, old_min):\n",
    "    array = (((array - old_min) * (new_max - new_min)) / (old_max - old_min)) + new_min\n",
    "    return array.astype(np.float16)\n",
    "\n",
    "\n",
    "# Calibrate the microphone\n",
    "print(\"Calibrating: Please Speak to the Microphone\")\n",
    "time.sleep(1)\n",
    "tuning_data = arduino_read(np.zeros(48000), 48000, 0)\n",
    "TUNING_MAX, TUNING_MIN = (max(tuning_data), min(tuning_data))\n",
    "print(f'Normalized signal from the range ({TUNING_MIN}, {TUNING_MAX}) to (-1, 1)')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1e6e42ed",
   "metadata": {},
   "source": [
    "## Running Inference\n",
    "In this section, we set up asynchronous functions for MFCC features extraction. Then, we initialize the TensorFlow Lite interpreter, run emotion recognition inference, and send results back to the Arduino."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "55103551",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[7.80900940e-04 3.43083171e+00 1.64198561e+01 4.99347121e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[0.99700433 0.8471093  2.38675348 1.45389081]\n",
      "Emotion: surprise\n",
      "\n",
      "[0.98819506 1.08457843 8.62430839 5.77661814]\n",
      "Emotion: surprise\n",
      "\n",
      "[2.12448905e-03 4.15599687e+00 1.75587903e+01 4.98649031e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[9.02240630e-04 4.06786727e+00 1.69497880e+01 4.99269247e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[6.84981374e-03 1.01755920e+01 4.90557821e+01 4.95792329e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[4.37238440e-03 6.23137486e+00 3.28392955e+01 4.97299433e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[5.65976952e-04 2.42617920e+00 1.16034399e+01 4.99531537e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[1.48905376e-02 1.65415776e+01 4.07523941e+01 4.91769612e+02]\n",
      "Emotion: unpleasant\n",
      "\n",
      "[1.49475290e-02 2.07651086e+01 6.74704323e+01 4.91336733e+02]\n",
      "Emotion: unpleasant\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<hr>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Inference time for 10 rounds: 13.163029193878174 seconds\n"
     ]
    }
   ],
   "source": [
    "BUFFER_SIZE = 24000\n",
    "OVERLAPPED = 512\n",
    "\n",
    "NUM_MFCC = 13\n",
    "N_FFT = 2048\n",
    "HOP_LENGTH = 512\n",
    "SAMPLE_RATE = 16000\n",
    "\n",
    "EMOTIONS = ['neutral', 'happy', 'surprise', 'unpleasant']\n",
    "COMMANDS = ['a', 'b', 'c', 'd']\n",
    "\n",
    "# Manually Calibrate Sensitivity of Recognition\n",
    "RECOG_MASK = np.array([1, 30000, 40000, 500])\n",
    "\n",
    "\n",
    "async def tflite_process_data():\n",
    "    data = np.zeros(BUFFER_SIZE)\n",
    "    data = arduino_read(data, BUFFER_SIZE, OVERLAPPED, norm=(TUNING_MAX, TUNING_MIN))\n",
    "    mfcc = librosa.feature.mfcc(y=data, sr=SAMPLE_RATE, n_mfcc=NUM_MFCC, n_fft=N_FFT, hop_length=HOP_LENGTH)\n",
    "    features = np.array([mfcc.T], dtype=np.float32)\n",
    "\n",
    "    # Model Input Shape = (None, None, 13)\n",
    "    interpreter.set_tensor(interpreter.get_input_details()[0]['index'], features)\n",
    "    interpreter.invoke()\n",
    "    prediction = interpreter.get_tensor(interpreter.get_output_details()[0]['index'])[0]\n",
    "    result = np.multiply(prediction, RECOG_MASK)\n",
    "    emotion = EMOTIONS[np.argmax(result)]\n",
    "    command = COMMANDS[np.argmax(result)]\n",
    "    arduino.write(command.encode('utf-8'))\n",
    "    print(result)\n",
    "    print(f'Emotion: {emotion}\\n')\n",
    "    \n",
    "async def tflite_run(rounds=10):\n",
    "    tasks = []\n",
    "    start_time = time.time()\n",
    "    for turn in range(rounds):\n",
    "        task = asyncio.create_task(tflite_process_data())\n",
    "        tasks.append(task)\n",
    "        time.sleep(0.6)\n",
    "    await asyncio.gather(*tasks)\n",
    "    display(HTML(\"<hr>\"))\n",
    "    print(f\"Inference time for {turn+1} rounds: {time.time() - start_time} seconds\") \n",
    "\n",
    "\n",
    "interpreter = tf.lite.Interpreter(model_path=\"output/Models/SER_quant.tflite\")\n",
    "interpreter.allocate_tensors()\n",
    "await tflite_run(rounds=10)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python (SER)",
   "language": "python",
   "name": "ser_env"
  },
  "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}