Once you've finished setting up your QT Py ESP32-S3 with CircuitPython, you can access the code and necessary libraries by downloading the Project Bundle.
To do this, click on the Download Project Bundle button in the window below. It will download to your computer as a zipped folder.
# SPDX-FileCopyrightText: 2023 Liz Clark for Adafruit Industries
#
# SPDX-License-Identifier: MIT
import os
import ssl
import time
import wifi
import socketpool
import board
import busio
import microcontroller
import adafruit_requests
import neopixel
from adafruit_ticks import ticks_ms, ticks_add, ticks_diff
from adafruit_io.adafruit_io import IO_HTTP
pixel_pin = board.NEOPIXEL
num_pixels = 1
pixels = neopixel.NeoPixel(pixel_pin, num_pixels, brightness=0.05, auto_write=False)
pixels.fill((255, 255, 0))
pixels.show()
try:
wifi.radio.connect(os.getenv('CIRCUITPY_WIFI_SSID'), os.getenv('CIRCUITPY_WIFI_PASSWORD'))
except Exception as e: # pylint: disable=broad-except
pixels.fill((100, 100, 100))
pixels.show()
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
aio_username = os.getenv("ADAFRUIT_IO_USERNAME")
aio_key = os.getenv("ADAFRUIT_IO_KEY")
try:
pool = socketpool.SocketPool(wifi.radio)
requests = adafruit_requests.Session(pool, ssl.create_default_context())
# Initialize an Adafruit IO HTTP API object
io = IO_HTTP(aio_username, aio_key, requests)
print("connected to io")
except Exception as e: # pylint: disable=broad-except
pixels.fill((100, 100, 100))
pixels.show()
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
try:
# get feed
ikea_pm25 = io.get_feed("ikeapm25")
except Exception: # pylint: disable=broad-except
# if no feed exists, create one
ikea_pm25 = io.create_new_feed("ikeapm25")
uart = busio.UART(board.TX, board.RX, baudrate=9600)
measurements = [0, 0, 0, 0, 0]
measurement_idx = 0
def valid_header(d):
headerValid = (d[0] == 0x16 and d[1] == 0x11 and d[2] == 0x0B)
# debug
# if not headerValid:
# print("msg without header")
return headerValid
start_read = False
clock = ticks_ms()
pixels.fill((0, 255, 0))
pixels.show()
io_time = 60000
while True:
try:
data = uart.read(32) # read up to 32 bytes
#print(data) # this is a bytearray type
time.sleep(0.01)
if ticks_diff(ticks_ms(), clock) >= io_time:
pixels.fill((0, 0, 255))
pixels.show()
io_data = measurements[0]
if io_data != 0:
io.send_data(ikea_pm25["key"], io_data)
print(f"sent {io_data} to {ikea_pm25['key']} feed")
time.sleep(1)
clock = ticks_add(clock, io_time)
pixels.fill((0, 0, 0))
pixels.show()
if data is not None:
v = valid_header(data)
if v is True:
measurement_idx = 0
start_read = True
if start_read is True:
pixels.fill((255, 0, 0))
pixels.show()
pm25 = (data[5] << 8) | data[6]
measurements[measurement_idx] = pm25
if measurement_idx == 4:
start_read = False
measurement_idx = (measurement_idx + 1) % 5
print(pm25)
print(measurements)
else:
pixels.fill((0, 255, 0))
pixels.show()
except Exception as e: # pylint: disable=broad-except
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
Upload the Code and Libraries to the QT Py ESP32-S3
After downloading the Project Bundle, plug your QT Py ESP32-S3 into the computer's USB port with a known good USB data+power cable. You should see a new flash drive appear in the computer's File Explorer or Finder (depending on your operating system) called CIRCUITPY. Unzip the folder and copy the following items to the QT Py ESP32-S3's CIRCUITPY drive.
- lib folder
- code.py
Your QT Py ESP32-S3 CIRCUITPY drive should look like this after copying the lib folder and the code.py file.
Add Your settings.toml File
As of CircuitPython 8, there is support for Environment Variables. These Environmental Variables are stored in a settings.toml file. Similar to secrets.py, the settings.toml file separates your sensitive information from your main code.py file. Add your settings.toml file as described in the Create Your settings.toml File page earlier in this guide. You'll need to include your CIRCUITPY_WIFI_SSID, CIRCUITPY_WIFI_PASSWORD, ADAFRUIT_IO_USERNAME and ADAFRUIT_IO_KEY in the file.
CIRCUITPY_WIFI_SSID = "your-wifi-ssid-here" CIRCUITPY_WIFI_PASSWORD = "your-wifi-password-here" ADAFRUIT_IO_USERNAME = "your-Adafruit-IO-username-here" ADAFRUIT_IO_KEY = "your-Adafruit-IO-key-here"
How the CircuitPython Code Works
First, the onboard NeoPixel is set up. Since you won't be able to see the serial monitor while the QT Py is inside the Vindriktning, the NeoPixel color will act as a status indicator.
pixel_pin = board.NEOPIXEL num_pixels = 1 pixels = neopixel.NeoPixel(pixel_pin, num_pixels, brightness=0.05, auto_write=False) pixels.fill((255, 255, 0)) pixels.show()
Connections
Then, the QT Py connects to WiFi and Adafruit IO. Both of these processes are wrapped in try/except statements to retry if any connection errors occur.
try:
wifi.radio.connect(os.getenv('CIRCUITPY_WIFI_SSID'), os.getenv('CIRCUITPY_WIFI_PASSWORD'))
except Exception as e:
pixels.fill((100, 100, 100))
pixels.show()
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
aio_username = os.getenv('ADAFRUIT_IO_USERNAME')
aio_key = os.getenv('ADAFRUIT_IO_KEY')
try:
pool = socketpool.SocketPool(wifi.radio)
requests = adafruit_requests.Session(pool, ssl.create_default_context())
# Initialize an Adafruit IO HTTP API object
io = IO_HTTP(aio_username, aio_key, requests)
print("connected to io")
except Exception as e:
pixels.fill((100, 100, 100))
pixels.show()
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
PM1006 Setup
The PM1006 sensor communicates over UART. Each packet of data from the PM1006 contains five measurements. You can tell the beginning of a packet based on the header values at the beginning of the UART data: 0x16, 0x11 and 0x0B. The valid_header function is used in the loop to determine if a new reading is coming in from the sensor.
uart = busio.UART(board.TX, board.RX, baudrate=9600)
measurements = [0, 0, 0, 0, 0]
measurement_idx = 0
def valid_header(d):
headerValid = (d[0] == 0x16 and d[1] == 0x11 and d[2] == 0x0B)
# debug
# if not headerValid:
# print("msg without header")
return headerValid
The Loop
In the loop, every minute, the last PM2.5 data reading from the PM1006 sensor is logged to Adafruit IO. While data is being sent to Adafruit IO, the NeoPixel is blue.
...
data = uart.read(32) # read up to 32 bytes
#print(data) # this is a bytearray type
time.sleep(0.01)
if ticks_diff(ticks_ms(), clock) >= io_time:
pixels.fill((0, 0, 255))
pixels.show()
io_data = measurements[0]
if io_data != 0:
io.send_data(ikea_pm25["key"], io_data)
print(f"sent {io_data} to {ikea_pm25["key"]} feed")
time.sleep(1)
clock = ticks_add(clock, io_time)
pixels.fill((0, 0, 0))
pixels.show()
Parsing Data
If data is read over UART, then the valid_header function determines if its the start of a new sensor reading array from the PM1006. If a new data series is being read, the NeoPixel turns red and the data is logged to the measurements array.
...
if data is not None:
v = valid_header(data)
if v is True:
measurement_idx = 0
start_read = True
if start_read is True:
pixels.fill((255, 0, 0))
pixels.show()
pm25 = (data[5] << 8) | data[6]
measurements[measurement_idx] = pm25
if measurement_idx == 4:
start_read = False
measurement_idx = (measurement_idx + 1) % 5
print(pm25)
print(measurements)
Try, Try Again
When the QT Py is idle, the onboard NeoPixel is green. The entire loop is wrapped in a try/except statement so that the QT Py will reset in case of any errors.
try:
...
else:
pixels.fill((0, 255, 0))
pixels.show()
except Exception as e:
print("Error:\n", str(e))
print("Resetting microcontroller in 5 seconds")
time.sleep(5)
microcontroller.reset()
Text editor powered by tinymce.
