Smart Alert Using Blues and Machine Learning

1. Introduction

There are several reasons that motivated me to do this project; 1) The first is to make a useful device for people with visual impairment to detect a possible fire in their home while they rest, sleep or are in another room; 2) this device can also be portable and easy to use to detect any alloholic substance in the user's glass; 3) I personally want to design a device using machine learning with blues and edge impulse, since there is little information or tutorials about it.

2. Data Collection

Below I show you the schematic diagram of the device that I will use for data collection, and that will later be used in training the model with Edge Impulse.

Code

Below I show you the code for data collection with Arduino IDE (data_collection.ino)

// Author: Guillermo Perez Guillen

#include <Wire.h>

// Constants
#define SAMPLING_FREQ_HZ    100       // Sampling frequency (Hz)
#define SAMPLING_PERIOD_MS  1000 / SAMPLING_FREQ_HZ   // Sampling period (ms)
#define NUM_SAMPLES         100       // 100 samples at 100 Hz is 1 sec window

int analogPin1 = PA3; //A0 MQ-135   
int analogPin2 = PC3; //A2 MQ-7   
int val1 = 0;           
int val2 = 0;           

void setup() {
  // Enable button pin
  pinMode(USER_BTN, INPUT_PULLUP); //BTN pin
  pinMode(LED_BUILTIN, OUTPUT); //LED pin

  // Start serial
  Serial1.setRx(PA10);                       //Redefine required Serial1 RX pin
  Serial1.setTx(PA9);                       //Redefine required Serial1 TX pin    
  Serial1.begin(115200);
}

void loop() {
  uint8_t len = 0;
  uint8_t addr = 0;
  uint8_t i;

  unsigned long timestamp;
  unsigned long start_timestamp;

  // Wait for button press
  while (digitalRead(USER_BTN) == 1);

  // Turn on LED to show we're recording
  digitalWrite(LED_BUILTIN, LOW);

  // Print header
  Serial1.println("timestamp,MQ-135,MQ-7");
 
  // Record samples in buffer
  start_timestamp = millis();
  for (int i = 0; i < NUM_SAMPLES; i++) {

    // Take timestamp so we can hit our target frequency
    timestamp = millis();

    // Read and print values
    Serial1.print(timestamp - start_timestamp);
    Serial1.print(",");
    val1 = analogRead(analogPin1);     // read the input pin1
    Serial1.print(val1);             // debug value MQ-135
    Serial1.print(",");
    val2 = analogRead(analogPin2);     // read the input pin2
    Serial1.println(val2);             // debug value MQ-7
        
    // Wait just long enough for our sampling period
    while (millis() < timestamp + SAMPLING_PERIOD_MS);
  }

  // Print empty line to transmit termination of recording
  Serial1.println();

  // Turn off LED to show we're done
  digitalWrite(LED_BUILTIN, HIGH);

  // Make sure the button has been released for a few milliseconds
  while (digitalRead(USER_BTN) == 0);
  delay(100);
}

Below I show you the code for data collection with Python. I ran the file in a Wondows 10 environment at the command prompt (serial-data-collect-csv.py)

import argparse
import os
import uuid

# Third-party libraries
import serial
import serial.tools.list_ports

# Settings
DEFAULT_BAUD = 115200       # Must match transmitting program baud rate
DEFAULT_LABEL = "_unknown"  # Label prepended to all CSV files

# Create a file with unique filename and write CSV data to it
def write_csv(data, dir, label):

    # Keep trying if the file exists
    exists = True
    while exists:

        # Generate unique ID for file (last 12 characters from uuid4 method)
        uid = str(uuid.uuid4())[-12:]
        filename = label + "." + uid + ".csv"
        
        # Create and write to file if it does not exist
        out_path = os.path.join(dir, filename)
        if not os.path.exists(out_path):
            exists = False
            try:
                with open(out_path, 'w') as file:
                    file.write(data)
                print("Data written to:", out_path)
            except IOError as e:
                print("ERROR", e)
                return    

# Command line arguments
parser = argparse.ArgumentParser(description="Serial Data Collection CSV")
parser.add_argument('-p',
                    '--port',
                    dest='port',
                    type=str,
                    required=True,
                    help="Serial port to connect to")
parser.add_argument('-b',
                    '--baud',
                    dest='baud',
                    type=int,
                    default=DEFAULT_BAUD,
                    help="Baud rate (default = " + str(DEFAULT_BAUD) + ")")
parser.add_argument('-d',
                    '--directory',
                    dest='directory',
                    type=str,
                    default=".",
                    help="Output directory for files (default = .)")
parser.add_argument('-l',
                    '--label',
                    dest='label',
                    type=str,
                    default=DEFAULT_LABEL,
                    help="Label for files (default = " + DEFAULT_LABEL + ")")
                    
# Print out available serial ports
print()
print("Available serial ports:")
available_ports = serial.tools.list_ports.comports()
for port, desc, hwid in sorted(available_ports):
    print("  {} : {} [{}]".format(port, desc, hwid))
    
# Parse arguments
args = parser.parse_args()
port = args.port
baud = args.baud
out_dir = args.directory
label = args.label

# Configure serial port
ser = serial.Serial()
ser.port = port
ser.baudrate = baud

# Attempt to connect to the serial port
try:
    ser.open()
except Exception as e:
    print("ERROR:", e)
    exit()
print()
print("Connected to {} at a baud rate of {}".format(port, baud))
print("Press 'ctrl+c' to exit")

# Serial receive buffer
rx_buf = b''

# Make output directory
try:
    os.makedirs(out_dir)
except FileExistsError:
    pass

# Loop forever (unless ctrl+c is captured)
try:
    while True:
        
        # Read bytes from serial port
        if ser.in_waiting > 0:
            while(ser.in_waiting):
                
                # Read bytes
                rx_buf += ser.read()
                
                # Look for an empty line
                if rx_buf[-4:] == b'\r\n\r\n':

                    # Strip extra newlines (convert \r\n to \n)
                    buf_str = rx_buf.decode('utf-8').strip()
                    buf_str = buf_str.replace('\r', '')

                    # Write contents to file
                    write_csv(buf_str, out_dir, label)
                    rx_buf = b''

# Look for keyboard interrupt (ctrl+c)
except KeyboardInterrupt:
    pass

# Close serial port
print("Closing serial port")
ser.close()

3. Assembling the Device

First, I have used a 80x80x60 mm plastic box.

The MQ-135 and MQ-7 sensors are placed at the bottom of the box. For these purposes you have to make two large holes and glue the sensors with silicone.

The Blue Swan board is placed on the front of the box. For these purposes I have made the holes to insert the pins and I have glued the board with a few drops of silicone.

On the left side of the case I have inserted the USB to TTL device as shown in the image below.

Finally, below I show the placement of an LED diode and a buzzer, which will be used in the Inference section.

4. Edge Impulse

For the use of the Blues Swan plate with Edge Impulse I have found few tutorials. You can find the official documentation at the following links:

• https://docs.edgeimpulse.com/docs/edge-ai-hardware/mcu/blues-wireless-swan
• https://dev.blues.io/guides-and-tutorials/building-edge-ml-applications/blues-swan/

On the other hand, I have been inspired by the project made with Arduino published at this link, since I intend to do something similar but with the Blues Swan board and the MQ-135 and MQ-7 gas sensors:

• https://www.edgeimpulse.com/blog/ai-managed-crops-irrigation/

Edge Impulse is a leading development platform for machine learning on edge devices, free for developers and trusted by enterprises. Head over to Edge Impulse, and create your account and login here: https://edgeimpulse.com/

Once logged in you will be taken to the project selection/creation page. In my case I have created the smart-alert project.

Dataset

In this example I will use data collection of alcohol (180 items), gas (180 items) and smoke (180 items). The total items are 540, and please use a ratio of 80% of the items for training and 20% for testing.

Create Impulse

• Now we are going to create our neural network and train our model.
• Head to the Create Impulse tab. Next click Add Raw Data block and select input axes.
• Now click Add learning block and select Classification, and type a name (in my case is NN-Classifier).
• Finally click Save impulse.

Transfer Learning

• Click on the Save parameters button to save the parameters.
• If you are not automatically redirected to the Generate features tab, click on the Raw Data tab and then click on Generate features button.
• Your data should be nicely clustered and there should be as little mixing of the classes as possible. You should inspect the clusters and look for any data that is clustered incorrectly. If you find any data out of place, you can relabel or remove it.

Training

• Now we are going to train our model. Click on the NN-Classifier tab, then click Data augmentation. In my case I have added 150 training cycles and a dropout layer (0.1), then Start training.
• Once training has completed, you will see the results displayed at the bottom of the page. Here we see that I have 93.1 % of accuracy and 0.21 of loss. Lets test our model and see how it works on our test data.

Model Testing

Head over to the Model testing tab where you will see all of the unseen test data available. Click on the Classify all wait a moment. In my case I obtained an accuracy of 86.11%.

Deployment

Now we will deploy the software directly to the Blues Swan board. To do this simply download the library as shown below. In my case I have downloaded for Arduino.

5. Inference

Below I show you the schematic diagram of the device that I will use for model inference.

Code

Below I show you the code for inference with Arduino IDE (alert_inference.ino)

// Author: Guillermo Perez Guillen 

#include <smart-alert_inferencing.h>
#include <Wire.h>

static uint8_t recv_cmd[8] = {};

// Settings
#define THRESHOLD           0.8       // Threshold for performing action 0.8

// Constants
#define SAMPLING_FREQ_HZ    100       // Sampling frequency (Hz)
#define SAMPLING_PERIOD_MS  1000 / SAMPLING_FREQ_HZ   // Sampling period (ms) 1000
#define NUM_CHANNELS        EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME // 2 channels
#define NUM_READINGS        EI_CLASSIFIER_RAW_SAMPLE_COUNT      // 100 readings
#define NUM_CLASSES         EI_CLASSIFIER_LABEL_COUNT           // 3 classes

#define led1 D10 //PA4
#define led2 D11 //PA7

// Means and standard deviations from our dataset curation
static const float means[] = {0.4686, -0.4075, 8.3669, 0.0717, 4.7533, -9.9816};
static const float std_devs[] = {2.9989, 7.0776, 6.8269, 60.9333, 101.3666, 109.0392};

int analogPin1 = PA3; //A0    
int analogPin2 = PC3; //A2    
int val1 = 0;           
int val2 = 0;           

float test0 = 0.0;
float test1 = 0.0;
float test2 = 0.0;

float myArray[3];

void setup() {  
  // Enable button pin
  pinMode(USER_BTN, INPUT_PULLUP); //BTN pin
  pinMode(LED_BUILTIN, OUTPUT); //LED pin

  pinMode(led1, OUTPUT); 
  pinMode(led2, OUTPUT);
    
  // Start serial
  Serial1.setRx(PA10);                       //Redefine required Serial1 RX pin
  Serial1.setTx(PA9);                       //Redefine required Serial1 TX pin    
  Serial1.begin(115200);
}

void loop() {  
  uint8_t len = 0;
  uint8_t addr = 0;
  uint8_t i;
  //uint32_t val = 0;
  
  unsigned long timestamp;
  ei_impulse_result_t result;
  int err;
  float input_buf[EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE];
  signal_t signal;

  // Wait for button press
  while (digitalRead(USER_BTN) == 1);

  // Turn on LED to show we're recording
  digitalWrite(LED_BUILTIN, LOW);

  // Record samples in buffer
  for (int i = 0; i < NUM_READINGS; i++) {

    // Take timestamp so we can hit our target frequency
    timestamp = millis();

    // Get raw readings from the gas sensor  
    val1 = analogRead(analogPin1);
    int mq135 = val1;
    val2 = analogRead(analogPin2);
    int mq7 = val2;

    // Perform standardization on each reading
    // Use the values from means[] and std_devs[]
    mq135 = (mq135 - means[0]) / std_devs[0];
    mq7 = (mq7 - means[1]) / std_devs[1];

    // Fill input_buf with the standardized readings. Recall tha the order
    // is [mq135, mq7 ...]
    input_buf[(NUM_CHANNELS * i) + 0] = mq135;
    input_buf[(NUM_CHANNELS * i) + 1] = mq7;

    // Wait just long enough for our sampling period
    while (millis() < timestamp + SAMPLING_PERIOD_MS);
  }

  // Turn off LED to show we're done recording
  digitalWrite(LED_BUILTIN, HIGH);

  // Turn the raw buffer into a signal for inference
  err = numpy::signal_from_buffer(input_buf, 
                                  EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE, 
                                  &signal);
  if (err != 0) {
    Serial1.print("ERROR: Failed to create signal from buffer: ");
    Serial1.println(err);
    return;
  }

  // Run the impulse
  err = run_classifier(&signal, &result, false);
  if (err != 0) {
    Serial1.print("ERROR: Failed to run classifier: ");
    Serial1.println(err);
    return;
  }

  // Print the results
  Serial1.println("Predictions");
  for (int i = 0; i < EI_CLASSIFIER_LABEL_COUNT; i++) {
    Serial1.print("  ");
    Serial1.print(result.classification[i].label);
    Serial1.print(": ");
    Serial1.println(result.classification[i].value);
    myArray[i] = (result.classification[i].value);
  }
    
  if (myArray[0] > 0.65) {         //ALCOHOL
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
    delay(500);
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);    
  }
  if (myArray[1] > 0.65) {         //GAS
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
  }
  if (myArray[2] > 0.65) {         //SMOKE
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
    delay(500);
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
  delay(500);
    digitalWrite(led1, HIGH);
    digitalWrite(led2, HIGH);
    delay(500);
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
  }
  else {
    digitalWrite(led1, LOW);
    digitalWrite(led2, LOW);
  }
  
  // Make sure the button has been released for a few milliseconds
  while (digitalRead(USER_BTN) == 0);
  delay(100);
}

6. Test

Below I show you the video of the first test with the device and where we can verify the precision of the artificial intelligence model.

Finally, below I show you a test done with the device and a battery to demonstrate its use without the need to connect it to the PC.

7. Conclusion

• This is the first time that I have used the Blues Swan board for an artificial intelligence application and I am satisfied because I have learned that it is possible to implement it despite not having a tutorial as a reference.
• Now the visually impaired person can use their device to detect alcoholic substances in their cups.
• The user can also activate their alarm to detect smoke caused by a fire.
• This is just a test device and can be improved with the following ideas: try other sensors to detect organic substances such as coffee or rotten food, etc. You can even add a voice that alerts the user instead of beeps emitted by a buzzer.