I had for a long time a little project on my desk. A temperature display reading last outside temperature from the server and display it on the dual seven segment display I have build some time ago.
My type of display.
But now that summer is closing I realized that the thermometer I have on my desk is not working any more. I searched in the parts pile and I found a DHT22 that I keep as a spare for Station 1. I also had in stock a TM1637 display from previous project. So I build the thing.
Reads the temperature and humidity from DHT22 and posts them to the server (Station 3) and gets the last temperature posted from Station 1. So, it displays for 20 seconds the local (room) temperature and for 20 seconds the outside temperature. That is that the LEDs indicate.
The final product.
The real reason I used the TM1637 display is that for some reason my dual ssd stopped working and couldn’t find why. Maybe the reason that the first TM1637 I connected was fried also. Also managed to destroy a DHT22 and 2 LEDs too. Destructive day. Anyways all good at the end, fortunately I had spare parts.
The prototype in action
The finished device in it’s final location.
For anyone interested the source.
import board, time
import adafruit_dht
import digitalio
import os, sys
import adafruit_connection_manager
import wifi
import adafruit_requests
import gc
import asyncio
import math
import TM1637
##############################################################################################
class NoSsidAvailable(Exception):
pass
##############################################################################################
class HttpClient:
def __init__(self,my_wifi_ssids, password):
self.my_wifi_ssids = my_wifi_ssids
self.password = password
self.ssid = self.findSsidToConnect()
self.requests = None
#self.requests = self.setupRequests()
def findSsidToConnect(self):
print(" | Available WiFi networks:")
networks = []
for network in wifi.radio.start_scanning_networks():
#print(network.ssid, network.ssid in self.my_wifi_ssids)
if network.ssid in self.my_wifi_ssids:
networks.append(network)
#print(network.ssid, network.rssi, network.channel) # Debug print
wifi.radio.stop_scanning_networks()
networks = sorted(networks, key=lambda net: net.rssi, reverse=True)
for network in networks:
print("\t%s\t\tRSSI: %d\tChannel: %d" % (str(network.ssid, "utf-8"), network.rssi, network.channel))
if len(networks)>0:
return str(networks[0].ssid, "utf-8");
else:
raise NoSsidAvailable(f"None of your SSID(s) {self.my_wifi_ssids} is available",300)
def connectWifi(self):
retries = 0
while retries < 5:
if wifi.radio.connected:
print(f"Connected to {self.ssid}")
if self.requests==None:
self.requests = self.setupRequests()
return True # Already connected
print(f"\n | Connecting to {self.ssid}... (attempt {retries+1}/5)")
show_in_display(f"c{retries+1}")
retries += 1
try:
wifi.radio.connect(self.ssid, self.password)
if self.requests==None:
self.requests = self.setupRequests()
print("✅ Wifi!")
show_in_display("co")
return True
except OSError as e:
print(f"❌ OSError: {e}")
show_in_display("ce")
time.sleep(2) # Add delay between retries
return False # Failed to connect
def setupRequests(self):
# Initalize Wifi, Socket Pool, Request Session
pool = adafruit_connection_manager.get_radio_socketpool(wifi.radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)
requests = adafruit_requests.Session(pool, ssl_context)
return requests
def postData(self,url, json_data):
out = False
retries = 0
while retries<3:
try:
# Check if connected first, reconnect if needed
if self.connectWifi():
print(f" | POST To: {url} Payload: {json_data}")
show_in_display(f"p{retries+1}")
with self.requests.post(url, json=json_data, timeout=10) as response:
json_resp = response.json()
print(f" | API Response: {json_resp['message']}")
if json_resp['message']=="YES":
out = True
break
else:
print("Post failed, trying again")
else:
print("Not connected to wifi, will not post")
show_in_display(f"pc")
break
except Exception as error:
print(f" | HttpClient.postData => Exception {error}")
show_in_display(f"pe")
retries += 1
gc.collect()
time.sleep(2) # Wait before retry
print("Trying again to post...")
return out
def getData(self, url):
out = None
gc.collect()
retries = 0
while retries<3:
print(f"in getData {gc.mem_free()=}")
show_in_display(f"g{retries+1}")
try:
if self.connectWifi(): # in case wifi is disconnected, connect again
print(f" | GET: {url}")
with self.requests.get(url, timeout=10) as response:
json_resp = response.json()
if json_resp['message']=='YES':
out = json_resp
break
except Exception as error:
print(f" | HttpClient.getData => Exception [{error}]")
show_in_display("ge")
retries += 1
time.sleep(2) # Wait before retry
if retries<3:
print("Trying again to get...")
else:
print("error")
show_in_display("gf")
gc.collect()
return out
def getTempFromServer():
success = True
last_temp = STATION_STATE['current_out']
data = client.getData(STATION_SETTINGS['json_get_data_url'])
if data!=None:
temp = data['temp']
if temp!=None:
last_temp = int(round( float(temp),0))
else:
success = False
if success:
STATION_STATE['current_out'] = last_temp
######################################################################################
class Thermometer:
def __init__(self, pinNum):
self.dhtDevice = adafruit_dht.DHT22(pinNum)
self.getReadings()
def getReadings(self):
retries = 0
while True:
retries = retries+1
if retries>10:
break
try:
temperature_c = self.dhtDevice.temperature
humidity = self.dhtDevice.humidity
if temperature_c is not None and humidity is not None:
return {"t": temperature_c,"h": humidity}
except RuntimeError as error:
# Errors happen fairly often, DHT's are hard to read, just keep going
print(f" | Thermometer.getReadings Runtime error: {error}")
time.sleep(0.3)
continue
except Exception as error:
print(f" | Thermometer.getReadings Exception {error}")
time.sleep(0.3)
continue
return None
######################################################################################
class IndicatorLed:
def __init__(self, pin):
self.ledPin = digitalio.DigitalInOut(pin)
self.ledPin.direction = digitalio.Direction.OUTPUT
def on(self):
self.ledPin.value = True
def off(self):
self.ledPin.value = False
##############################################################################################
def displayValueToSSD(v):
temp = int(round( float(v),0))
s = "{:02d}C".format(temp)
width = 4
s = " " * (width - len(s)) + s
show_in_display(s)
##############################################################################################
def show_in_display(txt):
display.write([0, 0, 0, 0])
display.show(txt)
##############################################################################################
async def getRemote():
while True:
getTempFromServer()
print("Get Remote")
gc.collect()
await asyncio.sleep(STATION_SETTINGS['station_sleep_time'])
##############################################################################################
async def postLocal():
while True:
print("Post local")
json_data = {
"stid": STATION_SETTINGS['station_id'],
}
readings = STATION_STATE['dht'].getReadings()
if readings is not None:
json_data["ta"] = readings["t"]
json_data["ha"] = readings["h"]
STATION_STATE['current_in'] = readings['t']
resp = True
if 'ta' in json_data:
resp = client.postData(STATION_SETTINGS['json_post_url'], json_data)
if not resp:
print("Failed to post data")
print("Loop end")
gc.collect()
await asyncio.sleep(STATION_SETTINGS['station_sleep_time'])
##############################################################################################
async def displayValue():
while True:
print("#################################")
if STATION_STATE['show']=='in':
print(f"Local {STATION_STATE['current_in']}")
displayValueToSSD(STATION_STATE['current_in'])
STATION_STATE['show']='out'
STATION_STATE['indLed1'].on()
STATION_STATE['indLed2'].off()
else:
print(f"Remote {STATION_STATE['current_out']}")
displayValueToSSD(STATION_STATE['current_out'])
STATION_STATE['show']='in'
STATION_STATE['indLed1'].off()
STATION_STATE['indLed2'].on()
print("#################################")
await asyncio.sleep(20)
##############################################################################################
async def main():
tasks = []
tasks.append(asyncio.create_task(postLocal()))
tasks.append(asyncio.create_task(getRemote()))
tasks.append(asyncio.create_task(displayValue()))
await asyncio.gather(*tasks)
##############################################################################################
STATION_SETTINGS = {
'wifi_ssids' : os.getenv("WIFI_SSID").split(','),
'password' : os.getenv("NETWORK_PASSWORD") ,
'json_post_url': "https://homeweather.salix.me/v1/measurements/log",
'json_get_data_url' : "https://homeweather.salix.me/v1/appdata/locationLastTemp/loc_1",
'station_id' : os.getenv("STATION_ID"),
'station_sleep_time' : os.getenv("STATION_SLEEP_TIME"),
'board_type': 'C3'
}
#########################################################################
STATION_STATE = {
'dht': None,
'indLed1' : None,
'indLed2' : None,
'current_in' : 0,
'current_out' : 0,
'show': 'in',
}
board_type_name = os.uname().machine
print("Board: ", board_type_name)
print(STATION_SETTINGS)
display = TM1637.TM1637(board.IO1, board.IO2)
display.write([0, 0, 0, 0])
display.show('boot')
time.sleep(1)
display.write([0, 0, 0, 0])
time.sleep(1)
display.show('boot')
STATION_STATE['dht'] = Thermometer(board.IO21)
STATION_STATE['indLed1'] = IndicatorLed(board.IO20)
STATION_STATE['indLed2'] = IndicatorLed(board.IO10)
STATION_STATE['indLed1'].on()
STATION_STATE['indLed2'].on()
try:
client = HttpClient(STATION_SETTINGS['wifi_ssids'],STATION_SETTINGS['password'])
client.connectWifi()
except Exception as e:
print(f"Caught an error: {e}")
display.show("er")
sys.exit()
gc.enable()
readings = STATION_STATE['dht'].getReadings()
print(readings)
if readings is not None:
STATION_STATE['current_in'] = readings['t']
asyncio.run(main())A tech note. Esp32 C3 supermini has less resources compared to the ESP32 DevKitc V4 board I have, that code without garbage collection is not working. On DevKitc V4 runs smoothly. On C3 supermini HTTP requests fail, with strange timeout error message. If free memory is less than 50k there is trouble.
UPDATE: The solution after all was CircuitPython. My DevKitc V4 has installed 9.2 and Station 1 too. So I downgraded CircuitPython from 10.1 to 9.2 and in combination with garbage collector, runs without issues.
