iosullivan/CTA_ESP32
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Updates for asyncio logic example

Browse Source
This commit is contained in:
iosullivan committed 2026-10-11 18:30:26 +10:00
1 parent ab110013c0
commit 658adc9baf
1 file changed
+86 -117
Download Patch File Download Diff File Expand all files Collapse all files
examples/src/micropython-webserver/main.py
+86 -117
View File
@@ -1,6 +1,6 @@
from time import time, gmtime, mktime,sleep
import network, machine, ntptime, socket, json, requests, neopixel, asyncio
from microdot import Microdot
from microdot import Microdot, send_file
app = Microdot()
@@ -10,122 +10,47 @@ def time_from_iso(datetime_str):
time_u = mktime((yyyy, mm, dd, h, m, s, 0, 0))
return time_u
def start_ap():
# Sets up initial captive portal for configuring wifi and basic options
ap = network.WLAN(network.AP_IF)
ap.active(True)
@app.post('/save')
async def save(request):
# Handle form submission
response = request.form
ap.config(
essid="CTA-Setup",
authmode=network.AUTH_OPEN
)
print("AP active for initial setup: ", ap.active())
print("AP IP address:", ap.ifconfig()[0])
return ap
def parse_form_data(body):
"""Parse URL-encoded form data"""
data = {}
pairs = body.split("&")
for pair in pairs:
if "=" in pair:
key, value = pair.split("=", 1)
# URL decode
data[key] = value.replace("+", " ").replace("%40", "@")
return data
def save_config(data):
"""Save Wi-Fi credentials to config.json"""
if not data.get("SSID"):
if not response["SSID"]:
raise ValueError("SSID is required")
config = {
"SSID": data.get("SSID"),
"PASS": data.get("PASS", ""),
"APIKEY": data.get("APIKEY", ""),
"LStation": data.get("LStation", ""),
"RStation": data.get("RStation", "")
"SSID": response["SSID"],
"PASS": response["PASS"],
"APIKEY": response["APIKEY"],
"LStation": response["LStation"],
"RStation": response["RStation"]
}
with open("config.json", "w") as f:
json.dump(config, f)
print("Configuration saved:", config)
@app.post('/save')
async def save(request):
# Handle form submission
body = request.split("\r\n\r\n", 1)[1]
form_data = parse_form_data(body)
save_config(form_data)
response_body = """<!DOCTYPE html>
<html>
<head>
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Setup Complete</title>
<title>Initial Setup Complete</title>
<style>
body { font-family: sans-serif; padding: 16px; text-align: center; }
h2 { color: #28a745; }
</style>
</head>
<body>
<h2>Configuration saved</h2>
<p>Device will reboot and connect to Wi-Fi.</p>
<p><small>You can close this window.</small></p>
<h2>Configuration Saved</h2>
<p>Reboot device to attempt Wi-Fi connection.</p>
</body>
</html>"""
return response_body
return response_body, {'Content-Type': 'text/html'}
@app.get('/')
async def index(request):
try:
with open("portal.html") as f:
response_body = f.read()
except:
response_body = "<h1>Setup Page Not Found</h1><p>Please upload portal.html</p>"
response = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/html\r\n"
"Connection: close\r\n\r\n"
+ response_body
)
return response
def reconnect_wifi():
while not wlan.isconnected():
try:
wlan.connect(SSID, PASS)
max_wait = 10
while max_wait > 0:
if wlan.isconnected():
break
max_wait -= 1
print('Waiting for connection...')
asyncio.wait(1)
if wlan.isconnected():
print('Connected to Wi-Fi')
print('Network config:', wlan.ifconfig())
break
except Exception as e:
print('Connection error:', e)
print('Retrying connection...')
asyncio.wait(2)
# Synchronize time of RTC with NTP
last_ntp_sync = -3700 # Init as more than an hour ago
last_api_query = mktime([2000,1,1,0,0,0,0,0]) # Set last query default (will run on first run)
query_period = 120 # Query the API at most every 2 minutes
distance_on = 10 # Time from station before turning on LED
distance_half = 30 # Time from station before going 'half' on
stations = {}
return send_file('portal.html')
# Setup Neopixel
p = machine.Pin(20) # connected on pin 20
@@ -138,31 +63,61 @@ wled.write()
try:
with open("config.json", "r") as f:
config = json.load(f)
APIKEY = config["APIKEY"]
SSID = config["SSID"]
PASS = config["PASS"]
# Allocate first and second station in demo only
stations[0] = { 'id':config["LStation"], 'eta':[] }
stations[1] = { 'id':config["RStation"], 'eta':[] }
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
reconnect_wifi()
# need to make below multithreaded....
#start_http_server() # Still launch server to allow changing SSID and config
demo = False
while not wlan.isconnected():
try:
wlan.connect(SSID, PASS)
max_wait = 10
while max_wait > 0:
if wlan.isconnected():
break
max_wait -= 1
print('Waiting for connection...')
sleep(1)
if wlan.isconnected():
print('Connected to Wi-Fi: ', SSID)
print('Device IP Address:', wlan.ifconfig()[0])
break
except Exception as e:
print('Connection error:', e)
print('Retrying connection...')
sleep(2)
except:
# Failed to open file, launch AP mode
start_ap()
# Failed to open config file, launch AP mode and disable trains
demo = True
wlan = network.WLAN(network.AP_IF)
wlan.active(True)
# Halt waiting for config
while(True):
asyncio.wait(5)
wlan.config(
essid="CTA-Setup",
authmode=network.AUTH_OPEN
)
while(True):
# Synchronise NTP time if hasn't been done in 1 hour
print("AP active for initial setup: ", wlan.active())
print("AP IP address:", wlan.ifconfig()[0])
async def CTA_Stations(config):
# Allocate first and second station
stations = {}
stations[0] = { 'id':config["LStation"], 'eta':[] }
stations[1] = { 'id':config["RStation"], 'eta':[] }
APIKEY = config["APIKEY"]
# Synchronize time of RTC with NTP
last_ntp_sync = -3700 # Init as more than an hour ago
last_api_query = mktime([2000,1,1,0,0,0,0,0]) # Set last query default (will run on first run)
query_period = 120 # Query the API at most every 2 minutes
distance_on = 10 # Time from station before turning on LED
distance_half = 30 # Time from station before going 'half' on
while True:
if (time() - last_ntp_sync) > 3600:
reconnect_wifi()
#reconnect_wifi()
try:
ntptime.settime()
print("Time synchronized successfully")
@@ -172,15 +127,16 @@ while(True):
# Have we queried API in the query_period seconds?
if (time() - last_api_query) > query_period:
# Query updated API
#trains = (requests.get("https://lapi.transitchicago.com/api/1.0/ttpositions.aspx?key="+APIKEY+"&rt=Blue&outputType=JSON")).json()['ctatt']
# Query all stations
for station,dict in stations.items():
trains_s = (requests.get(f"https://lapi.transitchicago.com/api/1.0/ttarrivals.aspx?key={APIKEY}&mapid={dict['id']}&outputType=JSON")).json()['ctatt']
request = f"https://lapi.transitchicago.com/api/1.0/ttarrivals.aspx?key={APIKEY}&mapid={dict['id']}&outputType=JSON"
trains_s = (requests.get(request)).json()['ctatt']
# Timestamps are served in local Chicago time, convert to GMT for simplicity on this side.
#print(f"Time of poll is: "+str(gmtime(time_from_iso(trains_s['tmst'])+(5*60*60))))
# Zeroize all existing ETAs
stations[station]['eta'] = []
# Check if any trains, otherwise skip
if 'eta' in trains_s:
# Loop through all trains and update matching stations
for train in trains_s['eta']:
eta_time = time_from_iso(train['arrT'])+(5*60*60)
@@ -197,7 +153,11 @@ while(True):
if (eta - time()) < -distance_on: # Remove old trains
stations[station]['eta'].pop(index)
# Set LED state tracker to 0, if no trains, it'll stay 0.
# Set default
wled[station] = (0,0,0)
if len(value['eta']) > 0:
# Get closest train
eta = min(value['eta'])
time_from_now = eta - time()
print(f"Next train will arrive at {value['id']} in {str(time_from_now)} seconds")
@@ -205,11 +165,20 @@ while(True):
wled[station] = (255,0,0)
elif time_from_now < distance_half:
wled[station] = (255,100,0)
else:
wled[station] = (0,0,0)
# Once all LEDs updated
wled.write()
asyncio.wait(1)
await asyncio.sleep(1)
async def main():
server = asyncio.create_task(app.start_server(port=80))
# Only run if we have internet
if wlan.IF_STA:
asyncio.create_task(CTA_Stations(config))
await server
asyncio.run(main())