Environmental Monitoring – No Screen
Saves a CSV file. Use Thonny to retrieve data.
import machine
import utime
import time
from machine import Pin, ADC, UART
from pimoroni import RGBLED, Button
from breakout_bme68x import BreakoutBME68X, STATUS_HEATER_STABLE
from pimoroni_i2c import PimoroniI2C
from breakout_ltr559 import BreakoutLTR559
# from pms5003 import PMS5003
"""
This basic example shows how to read from all the sensors on Enviro+.
Prints results to the REPL and saves data as a CSV
"""
# Set time
rtc=machine.RTC()
# Setup buttons
button_a = Button(12, invert=True)
button_b = Button(13, invert=True)
button_x = Button(14, invert=True)
button_y = Button(15, invert=True)
# change this to adjust temperature compensation
TEMPERATURE_OFFSET = 3
# set up the LED
led = RGBLED(6, 7, 10, invert=True)
led.set_rgb(0, 0, 255)
# set up the Pico's I2C
PINS_BREAKOUT_GARDEN = {"sda": 4, "scl": 5}
i2c = PimoroniI2C(**PINS_BREAKOUT_GARDEN)
# set up BME688 and LTR559 sensors
bme = BreakoutBME68X(i2c, address=0x77)
ltr = BreakoutLTR559(i2c)
# setup analog channel for microphone
MIC_PIN = 26
mic = ADC(Pin(26))
# Button a state
a = False
while True:
if button_a.is_pressed:
a = True
led.set_rgb(0, 255, 0)
print("Button A pressed!")
# Open file
timestamp=rtc.datetime()
fname ="%04d%02d%02d-%02d%02d%02d.csv"%(timestamp[0:3] + timestamp[4:7])
file=open(fname,"w")
file.write("Time,Temperature (°C),Humidity (%), Pressure (hPa), Gas, Lux, Mic")
# Run program
while not button_b.is_pressed:
# read BME688
temperature, pressure, humidity, gas, status, _, _ = bme.read()
heater = "Stable" if status & STATUS_HEATER_STABLE else "Unstable"
# correct temperature and humidity using an offset
corrected_temperature = temperature - TEMPERATURE_OFFSET
dewpoint = temperature - ((100 - humidity) / 5)
corrected_humidity = 100 - (5 * (corrected_temperature - dewpoint))
# read LTR559
ltr_reading = ltr.get_reading()
lux = ltr_reading[BreakoutLTR559.LUX]
prox = ltr_reading[BreakoutLTR559.PROXIMITY]
# read mic
mic_reading = mic.read_u16()
if heater == "Stable" and ltr_reading is not None:
timestamp=rtc.datetime()
print(timestamp)
timestring="%04d-%02d-%02d %02d:%02d:%02d"%(timestamp[0:3] + timestamp[4:7])
led.set_rgb(0, 255, 0)
print(f"""\n
Time = {timestring}
Temperature = {corrected_temperature} °C
Humidity = {corrected_humidity} %
Pressure = {pressure/100} hPa
Gas = {gas}
Lux = {lux}
Mic = {mic_reading}
""")
csvline = f"""\n{timestring},{corrected_temperature},{corrected_humidity},{pressure/100},{gas},{lux},{mic_reading}"""
file.write(csvline)
else:
# light up the LED red if there's a problem with the BME688 or LTR559 sensor readings
led.set_rgb(255, 0, 0)
time.sleep(1.0)
# Button B has been pressed
print("Button B pressed")
led.set_rgb(0, 0, 255)
# Close file
if a:
file.close()
a = False
Code language: Python (python)With Screen, Network and Time Server
- Picographics library
- ntptime
- Requires
network_manager.pyandWIFI_CONFIG.pyfrom the micropython/examples/common directory.
Save the above on the pico.
ntptime.py
import time
try:
import usocket as socket
except:
import socket
try:
import ustruct as struct
except:
import struct
# The NTP host can be configured at runtime by doing: ntptime.host = 'myhost.org'
host = "pool.ntp.org"
# The NTP socket timeout can be configured at runtime by doing: ntptime.timeout = 2
timeout = 1
def ttime():
NTP_QUERY = bytearray(48)
NTP_QUERY[0] = 0x1B
addr = socket.getaddrinfo(host, 123)[0][-1]
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.settimeout(timeout)
res = s.sendto(NTP_QUERY, addr)
msg = s.recv(48)
finally:
s.close()
val = struct.unpack("!I", msg[40:44])[0]
EPOCH_YEAR = time.gmtime(0)[0]
if EPOCH_YEAR == 2000:
# (date(2000, 1, 1) - date(1900, 1, 1)).days * 24*60*60
NTP_DELTA = 3155673600
elif EPOCH_YEAR == 1970:
# (date(1970, 1, 1) - date(1900, 1, 1)).days * 24*60*60
NTP_DELTA = 2208988800
else:
raise Exception("Unsupported epoch: {}".format(EPOCH_YEAR))
return val - NTP_DELTA
# There's currently no timezone support in MicroPython, and the RTC is set in UTC time.
# Add 11 for AEDT
def settime():
t = ttime()
import machine
tm = time.gmtime(t)
machine.RTC().datetime((tm[0], tm[1], tm[2], tm[6] + 1, tm[3], tm[4], tm[5], 0))
Code language: Python (python)network_manager.py
import rp2
import network
import machine
import uasyncio
class NetworkManager:
_ifname = ("Client", "Access Point")
def __init__(self, country="GB", client_timeout=60, access_point_timeout=5, status_handler=None, error_handler=None):
rp2.country(country)
self._ap_if = network.WLAN(network.AP_IF)
self._sta_if = network.WLAN(network.STA_IF)
self._mode = network.STA_IF
self._client_timeout = client_timeout
self._access_point_timeout = access_point_timeout
self._status_handler = status_handler
self._error_handler = error_handler
self.UID = ("{:02X}" * 8).format(*machine.unique_id())
def isconnected(self):
return self._sta_if.isconnected() or self._ap_if.isconnected()
def config(self, var):
if self._sta_if.active():
return self._sta_if.config(var)
else:
if var == "password":
return self.UID
return self._ap_if.config(var)
def mode(self):
if self._sta_if.isconnected():
return self._ifname[0]
if self._ap_if.isconnected():
return self._ifname[1]
return None
def ifaddress(self):
if self._sta_if.isconnected():
return self._sta_if.ifconfig()[0]
if self._ap_if.isconnected():
return self._ap_if.ifconfig()[0]
return '0.0.0.0'
def disconnect(self):
if self._sta_if.isconnected():
self._sta_if.disconnect()
if self._ap_if.isconnected():
self._ap_if.disconnect()
async def wait(self, mode):
while not self.isconnected():
self._handle_status(mode, None)
await uasyncio.sleep_ms(1000)
def _handle_status(self, mode, status):
if callable(self._status_handler):
self._status_handler(self._ifname[mode], status, self.ifaddress())
def _handle_error(self, mode, msg):
if callable(self._error_handler):
if self._error_handler(self._ifname[mode], msg):
return
raise RuntimeError(msg)
async def client(self, ssid, psk):
if self._sta_if.isconnected():
self._handle_status(network.STA_IF, True)
return
self._ap_if.disconnect()
self._ap_if.active(False)
self._sta_if.active(True)
self._sta_if.config(pm=0xa11140)
self._sta_if.connect(ssid, psk)
try:
await uasyncio.wait_for(self.wait(network.STA_IF), self._client_timeout)
self._handle_status(network.STA_IF, True)
except uasyncio.TimeoutError:
self._sta_if.active(False)
self._handle_status(network.STA_IF, False)
self._handle_error(network.STA_IF, "WIFI Client Failed")
async def access_point(self):
if self._ap_if.isconnected():
self._handle_status(network.AP_IF, True)
return
self._sta_if.disconnect()
self._sta_if.active(False)
self._ap_if.ifconfig(("10.10.1.1", "255.255.255.0", "10.10.1.1", "10.10.1.1"))
self._ap_if.config(password=self.UID)
self._ap_if.active(True)
try:
await uasyncio.wait_for(self.wait(network.AP_IF), self._access_point_timeout)
self._handle_status(network.AP_IF, True)
except uasyncio.TimeoutError:
self._sta_if.active(False)
self._handle_status(network.AP_IF, False)
self._handle_error(network.AP_IF, "WIFI Client Failed")
Code language: Python (python)WIFI_CONFIG.py
Replace the values below with the appropriate values.
SSID = "WIFI SSID"
PSK = "WIFI PASSWORD "
COUNTRY = "COUNTRY CODE"
Code language: Python (python)main.py
17/3/2024: Updated to only write average data (microphone maximum) once a minute rather than every second.
import machine
import time
import ntptime
import WIFI_CONFIG
import random
from machine import Pin, ADC, UART
from picographics import PicoGraphics, DISPLAY_ENVIRO_PLUS
from pimoroni import RGBLED, Button
from breakout_bme68x import BreakoutBME68X, STATUS_HEATER_STABLE
from pimoroni_i2c import PimoroniI2C
from breakout_ltr559 import BreakoutLTR559
from network_manager import NetworkManager
import uasyncio
# from pms5003 import PMS5003
"""
This basic example shows how to read from all the sensors on Enviro+.
Prints results to the REPL and saves data as a CSV
"""
# set up wifi
def status_handler(mode, status, ip):
display.set_font("bitmap8")
display.set_pen(BLACK)
display.clear()
display.set_pen(WHITE)
display.text("Network: {}".format(WIFI_CONFIG.SSID), 10, 10, scale=2)
print("Network: {}".format(WIFI_CONFIG.SSID))
status_text = "Connecting..."
if status is not None:
if status:
status_text = "Connection successful!"
else:
status_text = "Connection failed!"
display.text(status_text, 10, 40, scale=2)
print(status_text)
print("IP: {}".format(ip))
display.text("IP: {}".format(ip), 10, 60, scale=2)
display.update()
display.set_font("sans")
# Time on display
def display_time():
colours = [RED, GREEN, CYAN, MAGENTA, YELLOW, BLUE, WHITE]
timestamp=rtc.datetime()
timestring="%04d-%02d-%02d %02d:%02d:%02d"%(timestamp[0:3] + timestamp[4:7])
textheight = random.randint(25, 180)
textwidth = random.randint(0, 30)
colour_choice = random.choice(colours)
display.set_font("bitmap8")
display.set_pen(BLACK)
display.clear()
display.set_pen(colour_choice)
display.text("Press A to start", textwidth, textheight, WIDTH, scale=2)
display.text(f"{timestring}", textwidth, textheight + 20, WIDTH, scale=2)
display.update()
time.sleep(1)
# Set time
rtc=machine.RTC()
# Setup buttons
button_a = Button(12, invert=True)
button_b = Button(13, invert=True)
button_x = Button(14, invert=True)
button_y = Button(15, invert=True)
# Setup screen
# set up the display
display = PicoGraphics(display=DISPLAY_ENVIRO_PLUS)
BRIGHTNESS = 0.8
display.set_backlight(BRIGHTNESS)
# some constants we'll use for drawing
WHITE = display.create_pen(255, 255, 255)
BLACK = display.create_pen(0, 0, 0)
RED = display.create_pen(255, 0, 0)
GREEN = display.create_pen(0, 255, 0)
CYAN = display.create_pen(0, 255, 255)
MAGENTA = display.create_pen(200, 0, 200)
YELLOW = display.create_pen(200, 200, 0)
BLUE = display.create_pen(0, 0, 200)
FFT_COLOUR = display.create_pen(255, 0, 255)
GREY = display.create_pen(75, 75, 75)
WIDTH, HEIGHT = display.get_bounds()
display.set_font("sans")
# change this to adjust temperature compensation - doesn't seem to require this.
TEMPERATURE_OFFSET = 5
# set up the LED
led = RGBLED(6, 7, 10, invert=True)
led.set_rgb(0, 0, 255)
# set up the Pico's I2C
PINS_BREAKOUT_GARDEN = {"sda": 4, "scl": 5}
i2c = PimoroniI2C(**PINS_BREAKOUT_GARDEN)
# set up BME688 and LTR559 sensors
bme = BreakoutBME68X(i2c, address=0x77)
ltr = BreakoutLTR559(i2c)
# setup analog channel for microphone
MIC_PIN = 26
mic = ADC(Pin(26))
network_manager = NetworkManager(WIFI_CONFIG.COUNTRY, status_handler=status_handler)
# connect to wifi
uasyncio.get_event_loop().run_until_complete(network_manager.client(WIFI_CONFIG.SSID, WIFI_CONFIG.PSK))
# Set time
ntptime.settime()
# Disconnect network
network_manager.disconnect()
# Button a state
a = False
# Second counter
secs = 0
while True:
display_time()
if button_a.is_pressed:
a = True
led.set_rgb(0, 255, 0)
display.set_backlight(BRIGHTNESS)
display.set_font("sans")
display.set_pen(RED)
display.text("waiting for sensors", 0, 20, WIDTH, scale=1)
display.update()
# Setup measures
temperature_sum = 0
humidity_sum = 0
lux_sum = 0
pressure_sum = 0
gas_sum = 0
mic_max = 0
# Open file
timestamp=rtc.datetime()
fname ="%04d%02d%02d-%02d%02d%02d.csv"%(timestamp[0:3] + timestamp[4:7])
file=open(fname,"w")
file.write("Time,Temperature (°C),Humidity (%), Pressure (hPa), Gas, Lux, Mic")
# Run program
while not button_b.is_pressed:
if button_y.is_pressed:
# Switch backlight off
display.set_backlight(0)
elif button_x.is_pressed:
display.set_backlight(BRIGHTNESS)
# read BME688
temperature, pressure, humidity, gas, status, _, _ = bme.read()
heater = "Stable" if status & STATUS_HEATER_STABLE else "Unstable"
# correct temperature and humidity using an offset
corrected_temperature = temperature - TEMPERATURE_OFFSET
dewpoint = temperature - ((100 - humidity) / 5)
corrected_humidity = 100 - (5 * (corrected_temperature - dewpoint))
temperature_sum = temperature_sum + corrected_temperature
pressure_sum = pressure_sum + pressure
humidity_sum = humidity_sum + corrected_humidity
gas_sum = gas_sum + gas
# read LTR559
ltr_reading = ltr.get_reading()
lux = ltr_reading[BreakoutLTR559.LUX]
prox = ltr_reading[BreakoutLTR559.PROXIMITY]
lux_sum = lux_sum + lux
# read mic
mic_reading = mic.read_u16()
if mic_reading > mic_max:
mic_max = mic_reading
if heater == "Stable" and ltr_reading is not None:
timestamp=rtc.datetime()
print(timestamp)
timestring="%04d-%02d-%02d %02d:%02d:%02d"%(timestamp[0:3] + timestamp[4:7])
led.set_rgb(0, 255, 0)
print(f"""\n
Time = {timestring}
Temperature = {corrected_temperature} °C
Humidity = {corrected_humidity} %
Pressure = {pressure/100} hPa
Gas = {gas}
Lux = {lux}
Mic = {mic_reading}
""")
if secs > 59:
avg_temp = temperature_sum / secs
avg_humidity = humidity_sum / secs
avg_pressure = pressure_sum / secs
avg_lux = lux_sum / secs
avg_gas = gas_sum / secs
# Write to csv file
csvline = f"""\n{timestring},{avg_temp},{avg_humidity},{avg_pressure/100},{avg_gas},{avg_lux},{mic_max}"""
file.write(csvline)
secs = 0
mic_max = 0
temperature_sum = 0
humidity_sum = 0
lux_sum = 0
pressure_sum = 0
gas_sum = 0
mic_max = 0
# Display on screen
display.set_pen(BLACK)
display.clear()
display.set_pen(WHITE)
display.text(f"Temp: {corrected_temperature:.0f} °C", 0, 25, WIDTH, scale=1)
display.text(f"Humd: {corrected_humidity:.0f}%", 0, 55, WIDTH, scale=1)
display.text(f"Prs: {pressure/100:.0f}hPa", 0, 85, WIDTH, scale=1)
display.text(f"Gas: {gas}", 0, 115, WIDTH, scale=1)
display.text(f"Lux: {lux:.0f} lux", 0, 145, WIDTH, scale=1)
display.text(f"Mic: {mic_reading - 32000}", 0, 175, WIDTH, scale=1)
display.update()
else:
# light up the LED red if there's a problem with the BME688 or LTR559 sensor readings
led.set_rgb(255, 0, 0)
display.set_pen(BLACK)
display.clear()
display.set_backlight(BRIGHTNESS)
# Increment seconds
secs = secs + 1
# Measure every second.
time.sleep(1.0)
# Button B has been pressed
print("Button B pressed")
led.set_rgb(0, 0, 255)
display_time()
# Close file
if a:
file.close()
a = False
secs = 0
Code language: Python (python)Usage
Press A to start collecting data and B to stop. X and Y switch on and off the screen backlight.
Retrieve the data files off the Pico using Thonny and delete off the Pico to save storage space.