# two oled htu+bmp+bme
from machine import Pin, I2C, WDT, Timer
import network
import socket
import bme280
import time
import utime
import math
import framebuf #oled text scroll
import gc # Import garbage collection module
import ntptime # Added: Import NTP time synchronization module
# Import OLED display and font libraries
from oled import Write, GFX, SSD1306_I2C
from oled.fonts import ubuntu_mono_15, ubuntu_mono_20 # oled fonts
from ssd1306 import SSD1306_I2C # oled ssd1306
# --- Configuration ---
# WiFi Credentials
WIFI_SSID = "SSID" # <--- REPLACE WITH YOUR WIFI SSID
WIFI_PASSWORD = "PASSWORD" # <--- REPLACE WITH YOUR WIFI PASSWORD
# Timezone Adjustment (in hours from UTC)
# Example: -4 for EDT (Eastern Daylight Time), 1 for CET (Central European Time)
TIMEZONE_OFFSET_HOURS = -4 # <--- ADJUST THIS TO YOUR TIMEZONE OFFSET
# HTU21D I2C address
HTU_ADDRESS = 64
STATUS_BITS_MASK = 0xFFFC
# OLED and Sensor I2C setups
# I2C(0) for internal sensor (HTU21D/BME1) and left OLED
# SDA: GP16, SCL: GP17
i2c0 = I2C(0, sda=Pin(16), scl=Pin(17), freq=400000)
# I2C(1) for external sensor (BME2) and right OLED
# SDA: GP14, SCL: GP15
i2c1 = I2C(1, sda=Pin(14), scl=Pin(15), freq=400000)
# --- Hardware Initialization ---
led = Pin("LED", Pin.OUT)
led.on() # Turn on onboard LED initially
time.sleep(1)
led.off() # Turn off after a short delay
# OLED Displays
# These lines will now use the SSD1306_I2C that is effectively imported last (from ssd1306.py)
oled1 = SSD1306_I2C(128, 64, i2c0) # Left display, internal readings
oled2 = SSD1306_I2C(128, 64, i2c1) # Right display, external readings
# BME280 Sensors
bme1 = bme280.BME280(i2c=i2c0) # BME1 (internal) on i2c0
bme2 = bme280.BME280(i2c=i2c1) # BME2 (external) on i2c1
# Watchdog Timer
wdt = WDT(timeout=8000) # 8 seconds timeout
wdt.feed(
# Periodic watchdog refresh (every 4 seconds, under the 8s timeout)
_wdt_timer = Timer()
_wdt_timer.init(period=4000, mode=Timer.PERIODIC, callback=lambda t: wdt.feed())
# --- Global Sensor Data Variables (updated in main loop) ---
temp_htu = 0.0
hum_htu = 0.0
pressure1 = 0.0 # From BME1
temp1 = 0.0 # From BME1
hum1 = 0.0 # From BME1
temp2 = 0.0 # From BME2
hum2 = 0.0 # From BME2
pressure2 = 0.0 # From BME2
# --- OLED Display Helper Functions ---
def oled1_clear(): # left display. internal readings
oled1.fill_rect(0, 0, 128, 64, 0)
oled1.show()
def oled2_clear(): # right display, external readings
oled2.fill_rect(0, 0, 128, 64, 0)
oled2.show()
# Initial OLED test display
def initial_oled_test():
write20_1 = Write(oled1, ubuntu_mono_20)
write20_1.text("***************", 0, 0)
write20_1.text("***************", 0, 20)
write20_1.text("***************", 0, 40)
oled1.show()
write20_2 = Write(oled2, ubuntu_mono_20)
write20_2.text("***************", 0, 0)
write20_2.text("***************", 0, 20)
write20_2.text("***************", 0, 40)
oled2.show()
wdt.feed()
time.sleep(2) # Keep the sleep for visibility
wdt.feed()
oled1_clear()
oled2_clear()
# Call initial test
initial_oled_test()
# --- HTU21D Reading Functions ---
def read_h(i2c_obj, addr, status_b_m):
i2c_obj.writeto(addr, b'\xF5') # Trigger humidity measurement
utime.sleep_ms(29) # Wait for it to finish (29ms max)
data = i2c_obj.readfrom(addr, 2) # Get the 2 byte result
adjusted = (data[0] << 8) + data[1] # convert to 16 bit value
adjusted &= status_b_m # zero the status bits STATUS_BITS_MASK
adjusted *= 125 # scale
adjusted /= 1 << 16 # divide by 2^16
adjusted -= 6 # subtract 6
return adjusted
def read_t(i2c_obj, addr, status_b_m):
i2c_obj.writeto(addr, b'\xE3') # Trigger temp measurement
utime.sleep_ms(29) # Wait for it to finish (29ms max)
data1 = i2c_obj.readfrom(addr, 2) # Get the 2 byte result
adjusted = (data1[0] << 8) + data1[1] # convert to 16 bit value
adjusted &= status_b_m # zero the status bits STATUS_BITS_MASK
adjusted *= 175.72 # scale
adjusted /= 1 << 16 # divide by 2^16
adjusted -= 46.85 # subtract 46.85
return adjusted
# --- Wi-Fi Setup ---
def connect_wifi():
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(WIFI_SSID, WIFI_PASSWORD)
max_attempts = 20
attempts = 0
while not wlan.isconnected() and attempts < max_attempts:
print(f"Connecting to WiFi... ({attempts+1}/{max_attempts})")
oled1.fill(0)
oled1.text(f"WiFi Connect...", 0, 0)
oled1.text(f"Attempt {attempts+1}", 0, 20)
oled1.show()
oled2.fill(0)
oled2.text(f"WiFi Connect...", 0, 0)
oled2.text(f"Attempt {attempts+1}", 0, 20)
oled2.show()
wdt.feed()
time.sleep(1)
attempts += 1
if wlan.isconnected():
ip_address = wlan.ifconfig()[0]
print(f"WiFi Connected! IP: {ip_address}")
oled1.fill(0)
oled1.text("WiFi Connected!", 0, 0)
oled1.text(f"IP: {ip_address}", 0, 20)
oled1.show()
oled2.fill(0)
oled2.text("WiFi Connected!", 0, 0)
oled2.text(f"IP: {ip_address}", 0, 20)
oled2.show()
wdt.feed()
time.sleep(2)
# --- Synchronize Time via NTP ---
try:
print("Synchronizing time with NTP...")
ntptime.settime() # This will set the RTC to UTC. not implemented
print("Time synchronized.")
# You can verify time here: print(utime.localtime())
except OSError as e:
print(f"Error synchronizing time with NTP: {e}")
oled1.text("NTP Error!", 0, 40)
oled1.show()
oled2.text("NTP Error!", 0, 40)
oled2.show()
time.sleep(1) # Show error briefly
return ip_address
else:
print("Failed to connect to WiFi.")
oled1.fill(0)
oled1.text("WiFi Failed!", 0, 0)
oled1.show()
oled2.fill(0)
oled2.text("WiFi Failed!", 0, 0)
oled2.show()
return None
# Connect to WiFi initially
station_ip = connect_wifi()
# Create socket for web server (only if IP is available, otherwise it will be re-attempted in loop)
s = None
if station_ip:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', 80))
s.listen(5)
s.setblocking(False) # Set socket to non-blocking
print(f"Web server listening on http://{station_ip}/")
else:
print("Initial WiFi connection failed. Will keep trying in main loop.")
# --- Web Server Handlers ---
def serve_html():
global temp_htu, hum_htu, pressure1, temp2, hum2, pressure2 # Access global sensor data
# Get current time for "Last updated" using utime.localtime() directly
# Added timezone adjustment using TIMEZONE_OFFSET_HOURS
utc_timestamp = utime.time() # Get current UTC timestamp
local_timestamp = utc_timestamp + (TIMEZONE_OFFSET_HOURS * 3600) # Convert hours to seconds
current_time_tuple = utime.localtime(local_timestamp) # Convert adjusted timestamp to local time tuple
current_hour = current_time_tuple[3]
current_minute = current_time_tuple[4]
current_second = current_time_tuple[5]
html_response = f"""
Pico Sensor Data
Raspberry Pi Pico W Sensor Readings
Last updated: {current_hour:02}:{current_minute:02}:{current_second:02}
Inside
Temperature: {temp_htu:.1f}°C
Humidity: {hum_htu:.1f}%
Pressure: {pressure1:.1f}mmHg
Outside
Temperature: {temp2:.1f}°C
Humidity: {hum2:.1f}%
Pressure: {pressure2:.1f}mmHg
"""
return html_response
def serve_json():
global temp_htu, hum_htu, pressure1, temp2, hum2, pressure2 # Access global sensor data
json_response = f"""{{
"bme1": {{
"temperature": {temp_htu:.2f},
"humidity": {hum_htu:.2f},
"pressure_mmhg": {pressure1:.2f}
}},
"bme2": {{
"temperature": {temp2:.2f},
"humidity": {hum2:.2f},
"pressure_mmhg": {pressure2:.2f}
}}
}}"""
return json_response
# --- Main Loop ---
oled_shift = 0
oled_shift_down = 0
while True:
wdt.feed()
wlan = network.WLAN(network.STA_IF)
# Check and re-establish WiFi connection if lost
if not wlan.isconnected():
print("WiFi disconnected. Attempting to reconnect...")
# Call connect_wifi, which handles multiple attempts internally
station_ip = connect_wifi()
# If reconnected successfully AND the socket is not yet created or is invalid
if wlan.isconnected() and (s is None or not isinstance(s, socket.socket)):
try:
# Close existing socket if it's somehow invalid but not None
if s is not None:
try:
s.close()
except OSError:
pass # Ignore error if socket already closed/invalid
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', 80))
s.listen(5)
s.setblocking(False)
print(f"Web server re-started on http://{station_ip}/")
except Exception as e:
print(f"Error re-creating socket: {e}")
s = None # Ensure s is None if creation fails
elif not wlan.isconnected(): # If connect_wifi() failed to connect after its attempts
print("Failed to reconnect to WiFi. Web server will be inactive.")
s = None # Ensure socket is considered inactive if WiFi is not connected
# --- Read Sensors ---
# HTU21D on i2c0
temp_htu = read_t(i2c0, HTU_ADDRESS, STATUS_BITS_MASK)
hum_htu = read_h(i2c0, HTU_ADDRESS, STATUS_BITS_MASK)
# BME1 (on i2c0, same as HTU21D
#
#
#.
bme1_values = bme1.values
pressure1 = float(bme1_values[1]) * 0.75006 # Convert hPa to mmHg
# BME2 (external) on i2c1
bme2_values = bme2.values
temp2 = float(bme2_values[0][:-1]) # Remove 'C' and convert
hum2 = float(bme2_values[2][:-1]) # Remove '%' and convert
pressure2 = float(bme2_values[1]) * 0.75006 # Convert hPa to mmHg
# --- Adjustments ---
temp2 = temp2 + 2 # adjust temp value
hum_htu = hum_htu + 2
hum2 = hum2 - 1
pressure2 = pressure2 + 1
# --- Update OLEDs ---
oled1_clear()
write20_1 = Write(oled1, ubuntu_mono_20)
write20_1.text(f"TMP: {int(temp_htu):.0f}C", 0 + oled_shift, 0 + oled_shift_down)
write20_1.text(f"HUM: {int(hum_htu):.0f}%", 0 + oled_shift, 20 + oled_shift_down)
write20_1.text(f"PRE: {int(pressure1):.0f}", 0 + oled_shift, 40 + oled_shift_down)
oled1.show()
oled2_clear()
write20_2 = Write(oled2, ubuntu_mono_20)
write20_2.text(f"TMP: {int(temp2):.0f}C", 0 + oled_shift, 0 + oled_shift_down)
write20_2.text(f"HUM: {int(hum2):.0f}%", 0 + oled_shift, 20 + oled_shift_down)
write20_2.text(f"PRE: {int(pressure2):.0f}", 0 + oled_shift, 40 + oled_shift_down)
oled2.show()
# --- OLED Text Scrolling Logic ---
if oled_shift < 39:
oled_shift = oled_shift + 1
else:
oled_shift = 0
oled_shift_down = oled_shift_down + 1
if oled_shift_down == 9:
oled_shift_down = 0
# --- Handle Web Server Clients (non-blocking) ---
# Only try to handle clients if WiFi is connected and socket is open
if wlan.isconnected() and s is not None:
try:
conn, addr = s.accept()
conn.settimeout(3.0) # Set a timeout for client connection
print(f"Got a connection from {addr}")
request = conn.recv(1024)
request = request.decode('utf-8')
print(f"Request: {request}")
# Parse request to determine path
request_lines = request.split('\r\n')
if len(request_lines) > 0:
first_line = request_lines[0]
path = first_line.split(' ')[1] # Get the path from "GET /path HTTP/1.1"
# Handle both /data and /data/ for JSON
if path == "/data" or path == "/data/":
response_content = serve_json()
conn.send('HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n')
conn.send(response_content)
else: # Default to HTML for any other path
response_content = serve_html()
conn.send('HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nConnection: close\r\n\r\n')
conn.send(response_content)
else:
conn.send('HTTP/1.1 400 Bad Request\r\n\r\n') # Respond to empty request
conn.close() # Explicitly close the connection after serving
print("Connection closed")
gc.collect() # Run garbage collection after serving a request
except OSError as e:
if e.args[0] == 11: # errno 11 is EAGAIN, means no incoming connection
pass # No client connected, continue loop
else:
print(f"Socket error: {e}")
else:
# If no WiFi or socket not ready, print a message (optional)
# print("Skipping web server operations (WiFi not connected or socket not ready).")
pass # Do nothing if no WiFi, will attempt reconnect at start of next loop
# --- Loop Delay ---
led.value(not led.value()) # Toggle LED to show activity
wdt.feed()
time.sleep(3) # Wait 5 seconds before next sensor reading and OLED update
wdt.feed()
gc.collect() # Run garbage collection regularly in the main loop