Soil Quality Detection Using AI and LoRaWAN - Seeed Sensecap A1101

Start G-code:

G90 ; use absolute coordinates
M83 ; extruder relative mode
G28 ; move X/Y/Z to min endstops
G1 Z2.0 F3000 ; lift nozzle a bit 
G92 E0 ; Reset Extruder
G1 X10.1 Y20 Z0.28 F5000.0 ; Move to start position
G1 X10.1 Y200.0 Z0.28 F1500.0 E15 ; Draw the first line
G1 X10.4 Y200.0 Z0.28 F5000.0 ; Move to side a little
G1 X10.4 Y20 Z0.28 F1500.0 E30 ; Draw the second line
G92 E0 ; zero the extruded length again 
G1 E-2 F500 ; Retract a little 
M117
G21 ; set units to millimeters
G90 ; use absolute coordinates
M82 ; use absolute distances for extrusion
G92 E0
M107

End G-code:

M104 S0 ; Extruder off 
M140 S0 ; Heatbed off 
M107    ; Fan off 
G91     ; relative positioning 
G1 E-5 F3000  
G1 Z+0.3 F3000 ; lift print head 
G28 X0  F3000
M84            ; disable stepper motors

// Connections
// Arduino Nano :
//                                SH1106 OLED Display (128x64)
// D11  --------------------------- SDA
// D13  --------------------------- SCK
// D8   --------------------------- RST
// D9   --------------------------- DC
// D10  --------------------------- CS
//                                Control Button (A)
// A0   --------------------------- +
//                                Control Button (B)
// A1   --------------------------- +
//                                Control Button (C)
// A2   --------------------------- +
//                                5mm Common Anode RGB LED
// D3   --------------------------- R
// D5   --------------------------- G
// D6   --------------------------- B

#include <SPI.h>
#include <Adafruit_GFX.h>
#include <Adafruit_SH1106.h>

#define OLED_MOSI      11  // MOSI (SDA)
#define OLED_CLK       13  // SCK
#define OLED_DC        9
#define OLED_CS        10
#define OLED_RESET     8
Adafruit_SH1106 display(OLED_MOSI, OLED_CLK, OLED_DC, OLED_RESET, OLED_CS);

String classes[] = {"Enriched", "Unsafe", "Toxic"};
int color_codes[3][3] = {{0,255,0}, {255,255,0}, {255,0,0}};

  Serial.begin(115200);

  display.begin(SH1106_SWITCHCAPVCC);
  display.display();
  delay(1000);

void home_screen(){
  display.clearDisplay();   
  display.drawBitmap((128 - 40), 20, _home, 40, 40, WHITE);
  display.setTextSize(1); 
  display.setTextColor(BLACK, WHITE);
  display.setCursor(10,5);
  display.println(" Select Label: ");
  display.setTextColor(WHITE);
  display.setCursor(10,25);
  display.println("A) Enriched");  
  display.setCursor(10,40);
  display.println("B) Unsafe");  
  display.setCursor(10,55);
  display.println("C) Toxic");  
  display.display();
  delay(100);
}

void data_screen(int i){
  display.clearDisplay(); 
  if(i==0) display.drawBitmap((128 - 40) / 2, 0, enriched, 40, 40, WHITE);
  if(i==1) display.drawBitmap((128 - 40) / 2, 0, unsafe, 40, 40, WHITE);
  if(i==2) display.drawBitmap((128 - 40) / 2, 0, toxic, 40, 40, WHITE);
  // Print:
  int str_x = classes[i].length() * 11;
  display.setTextSize(2); 
  display.setTextColor(WHITE);
  display.setCursor((128 - str_x) / 2, 48);
  display.println(classes[i]);
  display.display();
  adjustColor(color_codes[i][0], color_codes[i][1], color_codes[i][2]);
  delay(4000);
  adjustColor(255,0,255);
}

  if(!digitalRead(button_A)){ Serial.println("Label: Enriched"); data_screen(0); delay(2000); }
  if(!digitalRead(button_B)){ Serial.println("Label: Unsafe"); data_screen(1); delay(2000); }
  if(!digitalRead(button_C)){ Serial.println("Label: Toxic"); data_screen(2); delay(2000); }

static const unsigned char PROGMEM toxic [] = {
0x00, 0x00, 0x81, 0x00, 0x00, 0x00, 0x03, 0x00, 0xC0, 0x00, 0x00, 0x06, 0x00, 0x60, 0x00, 0x00,
0x0C, 0x00, 0x30, 0x00, 0x00, 0x18, 0x00, 0x18, 0x00, 0x00, 0x38, 0x00, 0x1C, 0x00, 0x00, 0x30,
0x00, 0x0C, 0x00, 0x00, 0x70, 0x00, 0x0E, 0x00, 0x00, 0x70, 0x00, 0x0E, 0x00, 0x00, 0x70, 0x00,
0x0E, 0x00, 0x00, 0x70, 0x7E, 0x0E, 0x00, 0x00, 0x71, 0xFF, 0x8E, 0x00, 0x00, 0x73, 0xFF, 0xCE,
0x00, 0x00, 0x7B, 0x81, 0xDE, 0x00, 0x00, 0xFD, 0x00, 0xBF, 0x00, 0x01, 0xFC, 0x00, 0x3F, 0x80,
0x07, 0xFF, 0x00, 0xFF, 0xE0, 0x0F, 0xFF, 0xC3, 0xFF, 0xF0, 0x0F, 0xFF, 0xFF, 0xFF, 0xF0, 0x1E,
0x07, 0xE7, 0xE0, 0x78, 0x18, 0x31, 0xC3, 0x8C, 0x18, 0x30, 0x30, 0xC3, 0x0C, 0x0C, 0x30, 0x30,
0x00, 0x0C, 0x0C, 0x20, 0x30, 0x24, 0x0C, 0x04, 0x60, 0x38, 0x3C, 0x1C, 0x06, 0x40, 0x38, 0x3C,
0x1C, 0x02, 0x40, 0x1C, 0x3C, 0x38, 0x02, 0x40, 0x1C, 0x3C, 0x38, 0x02, 0x40, 0x0F, 0x3C, 0xF0,
0x02, 0x00, 0x07, 0xBD, 0xE0, 0x00, 0x00, 0x03, 0xBD, 0xC0, 0x00, 0x00, 0x01, 0x7E, 0x80, 0x00,
0x00, 0x00, 0x7E, 0x00, 0x00, 0x00, 0x00, 0xFF, 0x00, 0x00, 0x08, 0x01, 0xFF, 0x80, 0x10, 0x04,
0x03, 0xFF, 0xC0, 0x20, 0x03, 0x9F, 0xE7, 0xF9, 0xC0, 0x00, 0xFF, 0x81, 0xFF, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 
};

...

display.clearDisplay(); 
display.drawBitmap((128 - 40) / 2, 0, toxic, 40, 40, WHITE);
display.display();

import cv2
import serial
from threading import Thread
from time import sleep
import os
from PIL import Image
from io import BytesIO
import usb1
import numpy as np
import datetime

class A1101_data_img_collection():
    def __init__(self):
        # Define the required settings to obtain information from SenseCAP A1101 LoRaWAN Vision AI Sensor.
        self.WEBUSB_JPEG = (0x2B2D2B2D)
        self.WEBUSB_TEXT = 0x0F100E12
        self.VendorId = 0x2886
        self.ProductId = [0x8060, 0x8061]
        self.img_buff = bytearray()
        self.buff_size = 0
        self.time_out = 1000
        # Get the connected USB device context.       
        self.context = usb1.USBContext()
        # Initialize serial communication with Arduino Nano to obtain the given commands.
        self.arduino_nano = serial.Serial("COM7", 115200, timeout=2000)
        # Initialize and test SenseCAP A1101 connection.
        self.get_rlease_device(False)
        self.disconnect()

    def read_data(self):
        # If SenseCAP A1101 is accessible:
        with self.handle.claimInterface(2):
            # Get the data endpoints.
            self.handle.setInterfaceAltSetting(2, 0)
            self.handle.controlRead(0x01 << 5, request=0x22, value=0x01, index=2, length=2048, timeout=self.time_out)
            # Get all transferred data objects.
            transfer_list = []
            for _ in range(1):
                transfer = self.handle.getTransfer()
                transfer.setBulk(usb1.ENDPOINT_IN | 2, 2048, callback=self.processReceivedData, timeout=self.time_out)
                transfer.submit()
                transfer_list.append(transfer)
            # Check for any submitted data object in the received data packet.
            while any(x.isSubmitted() for x in transfer_list):
                self.context.handleEvents()

    def processReceivedData(self, transfer):
        # If SenseCAP A1101 generates a data packet successfully, process the received information.
        if transfer.getStatus() != usb1.TRANSFER_COMPLETED:
            # transfer.close()
            return
        # Extract the captured image from the processed data packet.
        data = transfer.getBuffer()[:transfer.getActualLength()]
        self.convert_and_show_img(data)
        # Resubmit the data packet after processing to avoid errors.
        transfer.submit()

    def convert_and_show_img(self, data: bytearray):
        # Convert the received data packet.
        if (len(data) == 8) and (int.from_bytes(bytes(data[:4]), 'big') == self.WEBUSB_JPEG):
            self.buff_size = int.from_bytes(bytes(data[4:]), 'big')
            self.img_buff = bytearray()
        elif (len(data) == 8) and (int.from_bytes(bytes(data[:4]), 'big') == self.WEBUSB_TEXT):
            self.buff_size = int.from_bytes(bytes(data[4:]), 'big')
            self.img_buff = bytearray()
        else:
            self.img_buff = self.img_buff + data
        # If the received data packet is converted to an image buffer successfully, display the generated image on the screen.
        if self.buff_size == len(self.img_buff):
            try:
                img = Image.open(BytesIO(self.img_buff))
                img = np.array(img)
                cv2.imshow('A1101_data_img_collection', cv2.cvtColor(img,cv2.COLOR_RGB2BGR))
                # Stop the video stream if requested.
                if cv2.waitKey(1) != -1:
                    cv2.destroyAllWindows()
                    print("\nCamera Feed Stopped!")
                # Store the latest frame captured by SenseCAP A1101.
                self.latest_frame = img
            except:
                self.img_buff = bytearray()
                return

    def connect(self):
        # Connect to SenseCAP A1101.
        self.handle = self.get_rlease_device(True)
        if self.handle is None:
            print('\nSenseCAP A1101 not detected!')
            return False
        with self.handle.claimInterface(2):
            self.handle.setInterfaceAltSetting(2, 0)
            self.handle.controlRead(0x01 << 5, request=0x22, value=0x01, index=2, length=2048, timeout=self.time_out)
            print('\nSenseCAP A1101 detected successfully!')
        return True

    def disconnect(self):
        # Reset the SenseCAP A1101 connection.
        try:
            print('Resetting the device connection... ')
            with usb1.USBContext() as context:
                handle = context.getByVendorIDAndProductID(self.VendorId, self.d_ProductId, skip_on_error=False).open()
                handle.controlRead(0x01 << 5, request=0x22, value=0x00, index=2, length=2048, timeout=self.time_out)
                handle.close()
                print('Device connection has been reset successfully!')
            return True
        except:
            return False

    def get_rlease_device(self, get=True):
        # Establish the SenseCAP A1101 connection.
        print('*' * 50)
        print('Establishing connection...')
        # Get the device information. 
        for device in self.context.getDeviceIterator(skip_on_error=True):
            product_id = device.getProductID()
            vendor_id = device.getVendorID()
            device_addr = device.getDeviceAddress()
            bus = '->'.join(str(x) for x in ['Bus %03i' % (device.getBusNumber(),)] + device.getPortNumberList())
            # Check if there is a connected device.
            if(vendor_id == self.VendorId) and (product_id in self.ProductId):
                self.d_ProductId = product_id
                print('\r' + f'\033[4;31mID {vendor_id:04x}:{product_id:04x} {bus} Device {device_addr} \033[0m', end='')
                # Turn on or off SenseCAP A1101.
                if get:
                    return device.open()
                else:
                    device.close()
                    print('\r' + f'\033[4;31mID {vendor_id:04x}:{product_id:04x} {bus} Device {device_addr} CLOSED\033[0m', flush=True)

    def get_transferred_data_packets(self):
        # Obtain the transferred commands from Arduino Nano via serial communication, including fertilizer hazard classes (labels).
        if self.arduino_nano.in_waiting > 0:
            command = self.arduino_nano.readline().decode("utf-8", "ignore").rstrip()
            if(command.find("Enriched") >= 0):
                print("\nCapturing an image! Label: Enriched")
                self.save_img_sample("Enriched")
            if(command.find("Unsafe") >= 0):
                print("\nCapturing an image! Label: Unsafe")
                self.save_img_sample("Unsafe")
            if(command.find("Toxic") >= 0):
                print("\nCapturing an image! Label: Toxic")
                self.save_img_sample("Toxic")
        sleep(1)

    def save_img_sample(self, _class):    
        date = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = './samples/{}_IMG_{}.jpg'.format(_class, date)
        # If requested, save the recently captured image (latest frame) as a sample.
        cv2.imwrite(filename, self.latest_frame)
        print("\nSample Saved Successfully: " + filename)

soil = A1101_data_img_collection()

# Define and initialize threads.
def A1101_camera_feed():
    soil.collect_data()
        
def activate_received_commands():
    while True:
        soil.get_transferred_data_packets()

Thread(target=A1101_camera_feed).start()
Thread(target=activate_received_commands).start()

from twilio.rest import Client
import requests
import json
from time import sleep

twilio_account_sid = '<_SID_>'
twilio_auth_token = '<_TOKEN_>'
twilio_client = Client(twilio_account_sid, twilio_auth_token)

API_ID = '<_ID_>'
API_key = '<_KEY_>'

device_eui = "2CF7F1C04340004A"
measurement_id = "4175"
channel_index = "1"

host = "https://sensecap.seeed.cc/openapi/"

get_latest_result = "view_latest_telemetry_data?device_eui={}&measurement_id={}&channel_index={}".format(device_eui, measurement_id, channel_index)

def send_WhatsApp_message(_from, _to, _message):
    # Send the given message via WhatsApp to inform the user of the model detection results.
    twilio_message = twilio_client.messages.create(
      from_ = 'whatsapp:'+_from,
      body = _message,
      media_url = 'https://media-cdn.seeedstudio.com/media/catalog/product/cache/bb49d3ec4ee05b6f018e93f896b8a25d/1/0/101990962-a1101-first-new-10.17.jpg',
      to = 'whatsapp:'+_to
    )
    print("\nWhatsApp Message Sent: "+twilio_message.sid)

def transfer_latest_result():
    # Obtain the latest model detection result via the SenseCAP HTTP API and notify the user of the received information through WhatsApp.
    url = host + get_latest_result
    # Make an HTTP GET request to the SenseCAP Portal by utilizing the provided HTTP authentication credentials (username and password).
    res = requests.get(url, auth = (API_ID, API_key))
    # Decode the received JSON object.
    res = json.loads(res.text)
    detection_digit = res["data"][0]["points"][0]["measurement_value"]
    date = res["data"][0]["points"][0]["time"]
    # Convert the obtained result digits to the detected class and the precision score.
    detected_class = "Nothing!"
    precision = 0
    if(detection_digit > 0 and detection_digit < 1):
        detected_class = "Enriched"
        precision = detection_digit
    if(detection_digit > 1 and detection_digit < 2):
        detected_class = "Toxic"
        precision = detection_digit-1
    if(detection_digit > 2):
        detected_class = "Unsafe"
        precision = detection_digit-2
    # Create a WhatsApp message from the retrieved information.
    message = "📌 Latest Model Detection Result\n\n🕒 {}\n🌱 Class: {}\n💯 Precision: {}".format(date, detected_class, round(precision, 2))
    print(message)
    # Transmit the generated message to the user via WhatsApp.
    send_WhatsApp_message('+_____________', '+_____________', message)

while True:
    # Notify the user of the latest model detection result every 10 minutes.
    transfer_latest_result()
    sleep(60*10)