So I got it running with a bare bones HTML webpage to display the sensor data without any CSS voodoo or other bells and whistles. I wanted to run the web server on one core and the main code on the other but it seems like the uasyncio module in the firmware on the micropython website is missing a few key functions, so I just added the web server as another async function and it all seems to work well.
With the web server active the limitations of the power supply on the dev board have become apparent. Now the LCD is showing a dip in voltage which is happening in sync with the webpage update cycle which shows a barely noticeable drop in luminosity at each update.
Maybe I could smooth that out with a small cap across 5V and ground pins but a better solution would probably be to use an external power supply. I can live with it as it’s barely noticeable but it could annoy some people.
This is the last incarnation of the code with a web server and web page.
import machine
import time
import BME280
from neopixel import NeoPixel
import uasyncio as asyncio
from lcd_api import LcdApi
from i2c_lcd import I2cLcd
import network
import socket
I2C_ADDR = 0x27
totalRows = 2
totalColumns = 16
# Define SDA and SCL pins
SDA_PIN = 8 # ESP32 S3
SCL_PIN = 9 # ESP32 S3
# Initialise I2C
i2c = machine.I2C(0, scl=machine.Pin(SCL_PIN), sda=machine.Pin(SDA_PIN))
# Initialise BME280
bme = BME280.BME280(i2c=i2c)
# Initialise lcd
lcd = I2cLcd(i2c, I2C_ADDR, totalRows, totalColumns)
# Initialise NeoPixel
NEOPIXEL_PIN = 48
num_pixels = 1 # Number of NeoPixels
np = NeoPixel(machine.Pin(NEOPIXEL_PIN), num_pixels)
# Variables to track pressure and timing
previous_pressure = None
pressure_drop_threshold_1hPa = 1
pressure_drop_threshold_2hPa = 2
pressure_drop_threshold_3hPa = 3
check_interval = 600 # 10 minutes in seconds
display_interval = 30 # 30 seconds for displaying data
lcd_update_interval = 10 # 10 seconds between lcd updates
led_timeout = 3600 # 1 hour in seconds
pressure_readings = []
flash_active = False
# Shared data structure for sensor readings
sensor_data = {
'temperature': None,
'pressure': None,
'humidity': None
}
# Function to start the web server
async def web_server():
addr_info = socket.getaddrinfo('0.0.0.0', 80)
for addr in addr_info:
try:
s = socket.socket()
s.bind(addr[4]) # Use addr[4] which is the (ip, port) tuple
s.listen(1)
print("Web server running on http://<your-esp32-ip>")
break
except OSError as e:
print(f"Failed to bind to {addr[4]}: {e}")
while True:
cl, addr = s.accept()
# print('Client connected from', addr)
request = cl.recv(1024)
if request:
# print("Request:", request)
await asyncio.sleep(0.1) # Wait briefly before checking again
# Generate HTML response
html = f"""
<!DOCTYPE html>
<html>
<head>
<center>
<title>Sensor Data</title>
<meta charset="UTF-8">
<meta http-equiv="refresh" content="10">
</head>
<body>
<h1>Sensor Data</h1>
<p>Temperature: {sensor_data['temperature']:.1f} °C</p>
<p>Humidity: {sensor_data['humidity']:.1f} %</p>
<p>Pressure: {sensor_data['pressure']:.1f} hPa</p>
</body>
</html>
"""
cl.send('HTTP/1.0 200 OK\r\nContent-type: text/html\r\n\r\n')
cl.send(html)
await asyncio.sleep(0.1) # Wait briefly before checking again
else:
print("No request received.")
cl.close() # Ensure the client connection is closed
# Function to fade the NeoPixel LED in cycles
async def fade_colours():
while True:
if flash_active: # Check if a flasmicropython variablesh is active
await asyncio.sleep(0.1) # Wait briefly before checking again
continue # Skip fading if flashing is active
# Fade from Red to Yellow
for r in range(256):
np[0] = (r, 255 - r, 0) # Fade from Red to Yellow
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Yellow to Green
for g in range(256):
np[0] = (255 - g, g, 0) # Fade from Yellow to Green
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Green to Cyan
for b in range(256):
np[0] = (0, 255 - b, b) # Fade from Green to Cyan
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Cyan to Blue
for r in range(256):
np[0] = (0, r, 255 - r) # Fade from Cyan to Blue
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Blue to Magenta
for g in range(256):
np[0] = (0, g, 255 - g) # Fade from Blue to Magenta
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Magenta to Red
for r in range(256):
np[0] = (255 - r, 0, 255) # Fade from Magenta to Red
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Function to flash the NeoPixel LED in blue until led_timeout is reached
async def flash_blue():
global flash_active
flash_active = True # Set the flag to indicate flashing is active
start_time = time.time() # Record the start time
while time.time() - start_time < led_timeout: # Continue until led_timeout is reached
np[0] = (0, 0, 255) # Set colour to blue
np.write()
await asyncio.sleep(0.5) # Flash on
np[0] = (0, 0, 0) # Turn off
np.write()
await asyncio.sleep(0.5) # Flash off
flash_active = False # Reset the flag when done
# Function to flash the NeoPixel LED in green
async def flash_green():
global flash_active
flash_active = True # Set the flag to indicate flashing is active
start_time = time.time() # Record the start time
while time.time() - start_time < led_timeout: # Continue until led_timeout is reached
np[0] = (0, 255, 0) # Set colour to blue
np.write()
await asyncio.sleep(0.5) # Flash on
np[0] = (0, 0, 0) # Turn off
np.write()
await asyncio.sleep(0.5) # Flash off
flash_active = False # Reset the flag when done
# Function to flash the NeoPixel LED in red
async def flash_red():
global flash_active
flash_active = True # Set the flag to indicate flashing is active
start_time = time.time() # Record the start time
while time.time() - start_time < led_timeout: # Continue until led_timeout is reached
np[0] = (255, 0, 0) # Set colour to blue
np.write()
await asyncio.sleep(0.5) # Flash on
np[0] = (0, 0, 0) # Turn off
np.write()
await asyncio.sleep(0.5) # Flash off
flash_active = False # Reset the flag when done
# Function to read sensor data
async def read_sensor_data():
global previous_pressure, pressure_readings, sensor_data # Declare as global
while True:
try:
temperature = bme.read_temperature() / 100 # Convert to Celsius
pressure = bme.read_pressure() / 25600 # Convert to hPa
humidity = bme.read_humidity() / 1024 # Convert to percentage
# Update shared sensor data
sensor_data['temperature'] = temperature
sensor_data['pressure'] = pressure
sensor_data['humidity'] = humidity
# Log the sensor data in CSV format
print(f"{temperature},{humidity},{pressure}") # Send data in CSV format
# Store the last four pressure readings
pressure_readings.append(pressure)
if len(pressure_readings) > 4:
pressure_readings.pop(0) # Keep only the last four readings
# Calculate the average of the last four readings
if len(pressure_readings) == 4:
average_pressure = sum(pressure_readings) / len(pressure_readings)
pressure_drop = previous_pressure - average_pressure if previous_pressure is not None else 0
# Debugging print statements
print(f"Pressure Drop: {pressure_drop:.2f} hPa")
print(f"Current Pressure: {pressure:.2f} hPa")
print(f"Average Pressure: {average_pressure:.2f} hPa")
# Trigger LED functions based on the average pressure drop
if pressure_drop > pressure_drop_threshold_3hPa:
await flash_red()
elif pressure_drop > pressure_drop_threshold_2hPa:
await flash_green()
elif pressure_drop > pressure_drop_threshold_1hPa:
await flash_blue()
previous_pressure = pressure
except Exception as e:
print("Error reading sensor data:", e)
await asyncio.sleep(check_interval) # Wait for the check interval before the next read
# Function to display data
async def display_data():
while True:
try:
# Use shared sensor data
print("Temperature: {:.1f} °C".format(sensor_data['temperature']))
print("Humidity: {:.1f} %".format(sensor_data['humidity']))
print("Pressure: {:.1f} hPa".format(sensor_data['pressure']))
except Exception as e:
print("Error displaying data:", e)
await asyncio.sleep(display_interval) # Wait for 30 seconds before the next display
# Function to update the LCD with sensor data
async def update_lcd():
last_update_time = time.time()
display_duration = 5 # Duration to display each reading
display_state = 'temperature' # Initial state
while True:
try:
current_time = time.time()
if current_time - last_update_time >= display_duration:
lcd.clear()
if display_state == 'temperature':
lcd.move_to(0, 0)
lcd.putstr("Temp: {:.1f}C".format(sensor_data['temperature']))
lcd.move_to(0, 1)
lcd.putstr("Humidity: {:.1f}%".format(sensor_data['humidity']))
display_state = 'pressure' # Switch to pressure next
else:
lcd.move_to(0, 0)
lcd.putstr("Temp: {:.1f}C".format(sensor_data['temperature']))
lcd.move_to(0, 1)
lcd.putstr("Barom: {:.1f}hPa".format(sensor_data['pressure']))
display_state = 'temperature' # Switch back to temperature next
last_update_time = current_time # Reset the timer
await asyncio.sleep(0.1) # Yield control to other tasks
except Exception as e:
print("Error updating LCD:", e)
# Main function to run tasks concurrently
async def main():
# Connect to Wi-Fi (make sure to replace with your credentials)
ssid = 'Your wifi network name'
password = 'Your wifi password'
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(ssid, password)
while not wlan.isconnected():
print("Connecting to WiFi...")
await asyncio.sleep(1)
print("Connected to WiFi:", wlan.ifconfig())
await asyncio.gather(
read_sensor_data(),
display_data(),
update_lcd(),
fade_colours(),
web_server() # Start the web server
)
# Run the main function
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Program stopped.")
The code sans the web server for a standalone device. Anyone dabbling in asynchronous coding for other projects might find it useful.
import machine
import time
import BME280
from neopixel import NeoPixel
import uasyncio as asyncio
from lcd_api import LcdApi
from i2c_lcd import I2cLcd
I2C_ADDR = 0x27
totalRows = 2
totalColumns = 16
# Define SDA and SCL pins
SDA_PIN = 8 # ESP32 S3
SCL_PIN = 9 # ESP32 S3
# Initialise I2C
i2c = machine.I2C(0, scl=machine.Pin(SCL_PIN), sda=machine.Pin(SDA_PIN))
# Initialise BME280
bme = BME280.BME280(i2c=i2c)
# Initialise lcd
lcd = I2cLcd(i2c, I2C_ADDR, totalRows, totalColumns)
# Initialise NeoPixel
NEOPIXEL_PIN = 48
num_pixels = 1 # Number of NeoPixels
np = NeoPixel(machine.Pin(NEOPIXEL_PIN), num_pixels)
# Variables to track pressure and timing
previous_pressure = None
pressure_drop_threshold_1hPa = 1 #(changed for led flash test)
pressure_drop_threshold_2hPa = 2 #(changed for led flash test)
pressure_drop_threshold_3hPa = 3 #(changed for led flash test)
check_interval = 600 # 10 minutes in seconds
display_interval = 30 # 30 seconds for displaying data
lcd_update_interval = 10 # 10 seconds between lcd updates
led_timeout = 3600 # 1 hour in seconds
pressure_readings = []
flash_active = False
# Shared data structure for sensor readings
sensor_data = {
'temperature': None,
'pressure': None,
'humidity': None
}
# Function to fade the NeoPixel LED in cycles
async def fade_colours():
while True:
if flash_active: # Check if a flasmicropython variablesh is active
await asyncio.sleep(0.1) # Wait briefly before checking again
continue # Skip fading if flashing is active
# Fade from Red to Yellow
for r in range(256):
np[0] = (r, 255 - r, 0) # Fade from Red to Yellow
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Yellow to Green
for g in range(256):
np[0] = (255 - g, g, 0) # Fade from Yellow to Green
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Green to Cyan
for b in range(256):
np[0] = (0, 255 - b, b) # Fade from Green to Cyan
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Cyan to Blue
for r in range(256):
np[0] = (0, r, 255 - r) # Fade from Cyan to Blue
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Blue to Magenta
for g in range(256):
np[0] = (0, 255 - g, g) # Fade from Blue to Magenta
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Fade from Magenta to Red
for r in range(256):
np[0] = (255 - r, 0, 255) # Fade from Magenta to Red
np.write()
await asyncio.sleep(0.01) # Allow other tasks to run
# Function to flash the NeoPixel LED in blue until led_timeout is reached
async def flash_blue():
global flash_active, current_flash_color
flash_active = True
current_flash_color = "Blue" # Set the current flash color
start_time = time.time()
while time.time() - start_time < led_timeout:
np[0] = (0, 0, 255)
np.write()
await asyncio.sleep(0.5)
np[0] = (0, 0, 0)
np.write()
await asyncio.sleep(0.5)
flash_active = False
current_flash_color = None # Reset the current flash color
# Function to flash the NeoPixel LED in green
async def flash_green():
global flash_active, current_flash_color
flash_active = True
current_flash_color = "Green" # Set the current flash color
start_time = time.time()
while time.time() - start_time < led_timeout:
np[0] = (0, 255, 0)
np.write()
await asyncio.sleep(0.5)
np[0] = (0, 0, 0)
np.write()
await asyncio.sleep(0.5)
flash_active = False
current_flash_color = None # Reset the current flash color
# Function to flash the NeoPixel LED in red
async def flash_red():
global flash_active, current_flash_color
flash_active = True
current_flash_color = "Red" # Set the current flash color
start_time = time.time()
while time.time() - start_time < led_timeout:
np[0] = (255, 0, 0)
np.write()
await asyncio.sleep(0.5)
np[0] = (0, 0, 0)
np.write()
await asyncio.sleep(0.5)
flash_active = False
current_flash_color = None # Reset the current flash color
# Function to read sensor data
async def read_sensor_data():
global previous_pressure, pressure_readings, sensor_data # Declare as global
while True:
try:
temperature = bme.read_temperature() / 100 # Convert to Celsius
pressure = bme.read_pressure() / 25600 # Convert to hPa
humidity = bme.read_humidity() / 1024 # Convert to percentage
# Update shared sensor data
sensor_data['temperature'] = temperature
sensor_data['pressure'] = pressure
sensor_data['humidity'] = humidity
# Log the sensor data in CSV format
print(f"{temperature},{humidity},{pressure}") # Send data in CSV format
# Store the last four pressure readings
pressure_readings.append(pressure)
if len(pressure_readings) > 4:
pressure_readings.pop(0) # Keep only the last four readings
# Calculate the average of the last four readings
if len(pressure_readings) == 4:
average_pressure = sum(pressure_readings) / len(pressure_readings)
pressure_drop = previous_pressure - average_pressure if previous_pressure is not None else 0
# Debugging print statements
print(f"Pressure Drop: {pressure_drop:.2f} hPa")
print(f"Current Pressure: {pressure:.2f} hPa")
print(f"Average Pressure: {average_pressure:.2f} hPa")
# Trigger LED functions based on the average pressure drop
if pressure_drop > pressure_drop_threshold_3hPa:
await flash_red()
elif pressure_drop > pressure_drop_threshold_2hPa:
await flash_green()
elif pressure_drop > pressure_drop_threshold_1hPa:
await flash_blue()
previous_pressure = pressure
except Exception as e:
print("Error reading sensor data:", e)
await asyncio.sleep(check_interval) # Wait for the check interval before the next read
# Function to display data
async def display_data():
while True:
try:
# Use shared sensor data
print("Temperature: {:.1f} °C".format(sensor_data['temperature']))
print("Humidity: {:.1f} %".format(sensor_data['humidity']))
print("Pressure: {:.1f} hPa".format(sensor_data['pressure']))
except Exception as e:
print("Error displaying data:", e)
await asyncio.sleep(display_interval) # Wait for 30 seconds before the next display
# Function to update lcd
async def update_lcd():
last_update_time = time.time()
display_duration = 5 # Duration to display each reading
display_state = 'temperature' # Initial state
while True:
try:
current_time = time.time()
if current_time - last_update_time >= display_duration:
lcd.clear()
if display_state == 'temperature':
lcd.move_to(0, 0)
lcd.putstr("Temp: {:.1f}C".format(sensor_data['temperature']))
lcd.move_to(0, 1)
lcd.putstr("Humidity: {:.1f}%".format(sensor_data['humidity']))
display_state = 'pressure' # Switch to pressure next
else:
lcd.move_to(0, 0)
lcd.putstr("Temp: {:.1f}C".format(sensor_data['temperature']))
lcd.move_to(0, 1)
lcd.putstr("Barom: {:.1f}hPa".format(sensor_data['pressure']))
display_state = 'temperature' # Switch back to temperature next
last_update_time = current_time # Reset the timer
await asyncio.sleep(0.1) # Yield control to other tasks
except Exception as e:
print("Error updating LCD:", e)
# Main function to run tasks concurrently
async def main():
# Start reading sensor data, displaying data, update lcd and fade colours concurrently
await asyncio.gather(
read_sensor_data(),
display_data(),
update_lcd(),
asyncio.create_task(fade_colours())# # Start fading colours in parallel
)
# Run the main function
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Program stopped.")
