denis/esp32-uart-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

second commit

Browse Source
This commit is contained in:
denis
2026-03-11 23:03:02 +01:00
parent 3bd07d9bb4
commit 948a778f42
6 changed files with 691 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
umod/__init__.py
View File
umod/uaio_http_server.py
+411
View File
@@ -0,0 +1,411 @@
# -*- coding: utf-8 -*-
# uaio_http_server.py
# GNU GENERAL PUBLIC LICENSE: (c) DD miraceti.net
import uasyncio as asyncio # @UnresolvedImport
import hashlib # @UnresolvedImport
import ubinascii # @UnresolvedImport
import os
'''
AsyncHttpServer est un serveur HTTP asynchrone 100 % MicroPython.
Autonome et sans dépendances externes, conçu pour ESP32.
Il gère le parsing HTTP, le routage GET/POST, le service de fichiers statiques en streaming et les WebSockets.
Respecte les contraintes mémoire et appuie uniquement sur uasyncio et les bibliothèques natives (hashlib, ubinascii, ujson).
'''
MIME = {
'.txt': 'text/plain',
'.html': 'text/html; charset=utf-8',
'.css': 'text/css',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
'.gif': 'image/gif',
'.svg': 'image/svg+xml',
'.ico': 'image/x-icon',
'.json': 'application/json',
'.js': 'application/javascript',
}
EPOCH_2000_UNIX_TIME = 946684800
def toUnixTimestamp(totalseconds):
return totalseconds + EPOCH_2000_UNIX_TIME
def to2000Timestamp(totalseconds):
return totalseconds - EPOCH_2000_UNIX_TIME
def base64_encode(data):
return ubinascii.b2a_base64(data).strip()
def base64_decode(data):
return ubinascii.a2b_base64(data)
class AsyncHttpServer:
'''
--------------------------
Exemple minimal d’usage :
--------------------------
server = AsyncHttpServer()
@server.get("/hello")
async def hello(server, reader, writer):
await server.send_response(writer, b"Hello GET")
@server.post("/echo")
async def echo(server, reader, writer, body):
await server.send_response(writer, body, mime="application/json")
async def on_ws_message(server, msg, writer):
await server.ws_send_text(writer, msg)
server.ws_handler = on_ws_message
server.run()
-------------------------------------------------------------------
Client HTML + JS (pour tester WebSocket + API + fichiers statiques)
-------------------------------------------------------------------
<!DOCTYPE html>
<html>
<body>
<h1>Serveur Micropython 100% autonome</h1>
<button id="g">Test GET</button>
<button id="p">Test POST</button>
<button id="w">WS Send</button>
<pre id="log"></pre>
<script>
const out = t => document.getElementById("log").textContent += t+"\n";
// GET
document.getElementById("g").onclick = async () => {
let r = await fetch("/hello");
out(await r.text());
};
// POST
document.getElementById("p").onclick = async () => {
let r = await fetch("/echo", {
method: "POST",
body: JSON.stringify({x: 1})
});
out(await r.text());
};
// WS
let ws = new WebSocket(`ws://${location.host}/ws`);
ws.onmessage = e => out("WS → " + e.data);
document.getElementById("w").onclick = () => ws.send("ping");
</script>
</body>
</html>
'''
def __init__(self, host="0.0.0.0", port=80, www="www"):
self.host = host
self.port = port
self.www = www
self.routes_get = {}
self.routes_post = {}
self.ws_routes = {}
self.ws_clients = []
self._server = None
# -------------------------
# Routing API
# -------------------------
def get(self, path):
def decorator(func):
self.routes_get[path] = func
return func
return decorator
def post(self, path):
def decorator(func):
self.routes_post[path] = func
return func
return decorator
def websocket(self, path):
def decorator(func):
self.ws_routes[path] = func
return func
return decorator
# -------------------------
# Serveur principal
# -------------------------
async def start(self):
self._server = await asyncio.start_server(self.handle_client, self.host, self.port)
print(f"HTTP Serveur {self.host}:{self.port} vient de démarrer!...")
while True:
await asyncio.sleep(1)
# -------------------------
# Lecture HTTP minimaliste
# -------------------------
def ensure_bytes(self, body):
if body is None:
return b""
if isinstance(body, bytes):
return body
if isinstance(body, str):
return body.encode("utf-8")
# Si c'est un dict, liste, JSON…
return str(body).encode("utf-8")
def get_mime(self, path):
for ext, mt in MIME.items():
if path.lower().endswith(ext):
return mt
return 'application/octet-stream'
def file_exists(self, path):
try:
os.stat(path)
return True
except:
return False
async def read_headers(self, reader):
headers = {}
while True:
line = await reader.readline()
if not line:
break
line = line.decode().strip()
if line == "":
break # fin des headers
if ":" in line:
k, v = line.split(":", 1)
headers[k.lower()] = v.strip()
return headers
async def read_body(self, reader, headers):
length = int(headers.get("content-length", "0"))
if length > 0:
return await reader.read(length)
return b""
# -------------------------
# Réponse HTTP standard
# -------------------------
async def send_response(self, writer, body=b'', status="200 OK", content_type="text/plain"):
size = len(body)
await writer.awrite(
"HTTP/1.1 " + status + "\r\n"
"Content-Type: " + content_type + "\r\n"
"Content-Length: " + str(size) + "\r\n"
"Connection: close\r\n"
"\r\n"
)
await writer.drain()
if body:
body = self.ensure_bytes(body)
await writer.awrite(body)
await writer.drain()
# -------------------------
# Client handler dispatcher
# -------------------------
async def handle_client(self, reader, writer):
try:
first = await reader.readline()
if not first:
await writer.aclose()
return
request_line = first.decode().strip()
method, path, proto = request_line.split() # @UnusedVariable
headers = await self.read_headers(reader)
# POST ?
if method == "POST" and path in self.routes_post:
body = await self.read_body(reader, headers)
await self.routes_post[path](self, writer, headers, body)
# GET API ?
elif method == "GET" and path in self.routes_get:
await self.routes_get[path](self, writer, headers)
# websocket ?
elif headers.get('upgrade', '').lower() == 'websocket' and path in self.ws_routes:
await self.handle_websocket(reader, writer, headers, path)
else:
await self.serve_static(writer, path)
except Exception as e:
print("Server error:", e)
await self.send_response(writer, "500 Internal Server Error")
finally:
try:
await writer.aclose()
except:
pass
# ---------------------------
# Fichiers statiques resolver
# ---------------------------
async def serve_static(self, writer, path):
if path == "/":
path = "/index.html"
full = self.www + path
print("serve static =>", full)
try:
if not self.file_exists(full):
raise Exception("File Not found")
size = os.stat(full)[6]
mime = self.get_mime(full)
header = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: " + mime + "\r\n"
"Content-Length: " + str(size) + "\r\n"
#"Cache-Control: no-cache, no-store, must-revalidate\r\n"
#"Pragma: no-cache\r\n"
#"Expires: 0\r\n"
"Connection: close\r\n"
"\r\n"
)
await writer.awrite(header)
await writer.drain()
with open(full, "rb") as f:
while True:
chunk = f.read(1024)
if not chunk:
break
await writer.awrite(chunk)
await writer.drain()
except Exception as e:
print("serve_static error =>", full, e)
await self.send_response(writer, "404 File Not found")
# --- compute Sec-WebSocket-Accept ---
def ws_accept_key(self, key):
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11".encode()
key_bytes = key.encode()
sha1 = hashlib.sha1(key_bytes + GUID).digest()
ws_key = ubinascii.b2a_base64(sha1).strip().decode()
return ws_key
# --- send one WebSocket TEXT frame ---
async def ws_send_text(self, writer, text):
payload = text.encode()
await writer.awrite(b"\x81") # FIN + opcode TEXT
length = len(payload)
if length < 126:
await writer.awrite(bytes([length]))
else:
await writer.awrite(b"\x7E" + bytes([(length >> 8) & 0xFF, length & 0xFF]))
await writer.awrite(payload)
await writer.drain()
# --- receive a single frame ---
async def ws_recv(self, reader):
header = await reader.readexactly(2)
if not header:
return None
opcode = header[0] & 0x0F
masked = header[1] & 0x80
length = header[1] & 0x7F
if opcode == 0x8:
return None
if length == 126:
ext = await reader.readexactly(2)
length = (ext[0] << 8) | ext[1]
elif length == 127:
return None
mask = await reader.readexactly(4) if masked else None
data = await reader.readexactly(length)
if masked:
data = bytes(b ^ mask[i % 4] for i, b in enumerate(data))
return data.decode()
# -------------------------
# Websocket
# -------------------------
async def handle_websocket(self, reader, writer, headers, path):
try:
ws_key = headers["sec-websocket-key"]
if not ws_key:
await writer.awrite("HTTP/1.1 400 Bad Request\r\n\r\n")
await writer.drain()
await writer.aclose()
return
accept = self.ws_accept_key(ws_key)
response = (
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: " + accept +"\r\n"
"\r\n"
)
await writer.awrite(response)
await writer.drain()
#print(response)
# Register this client
self.ws_clients.append(writer)
# Main recv loop
while True:
try:
msg = await self.ws_recv(reader)
if msg is None:
break
callback = self.ws_routes[path]
if callback:
# APPEL DU CALLBACK UTILISATEUR (si défini)
# ws_routes[path] doit être une coroutine ou fonction acceptant (server, msg, writer)
asyncio.create_task(callback(self, msg, writer))
else:
await self.ws_send_text(writer, msg)
except:
pass
except Exception:
pass
finally:
# retirer le client s'il est présent
try:
if writer in self.ws_clients:
self.ws_clients.remove(writer)
except:
pass
try:
await writer.aclose()
except:
pass
umod/wifi_manager.py
+201
View File
@@ -0,0 +1,201 @@
## -*- coding: utf-8 -*-
# wifi_manager.py
# GNU GENERAL PUBLIC LICENSE: (c) DD miraceti.net
import utime as time # @UnresolvedImport
import ubinascii # @UnresolvedImport
import network # @UnresolvedImport
'''
Management de la WiFi pour ESP32 (MicroPython).
Usage:
SSID = "ssid_8chars"
PASSWORD = "ChangeMeSoon"
if __name__ == "__main__":
try:
WiFi = WifiManager(mode='AP')
wifi = WiFi(SSID, PASSWORD)
wifi.start()
n=0
while True:
time.sleep(5.0)
print(n, wifi.status())
n += 1
except KeyboardInterrupt:
pass
finally:
wifi.stop()
print("Arrêt")
'''
class WifiBase:
ssid = None
password = None
channel = 1
hidden = False
max_clients = 4
authmode = 3
_iface= None
_mode='AP'
wifi_delay = 1.5
def start(self):
pass
def stop(self):
pass
def status(self):
pass
def restart(self):
if_was = self._iface.active()
self._iface.active(False)
time.sleep(self.wifi_delay)
if if_was:
self._iface.active(True)
return self.status()
def get_ip(self):
return self._iface.ifconfig()[0] if self._iface.active() else None
def get_ssid(self):
return self._iface.config('essid') if self._iface.active() else None
def get_rssi(self):
return self._iface.status('rssi') if self._iface.isconnected() else None
def get_iface(self):
return self._mode
class APManager(WifiBase):
"""
Gestion WiFi pour ESP32 (MicroPython).
Arguments (tous optionnels selon le mode) :
ssid (str) : nom du réseau AP ou SSID à rejoindre
password (str) : mot de passe AP ou STA (min 8 chars si authmode != 0)
channel (int) : canal pour AP (1-13)
hidden (bool) : AP caché ou non
max_clients (int) : nombre max de clients en AP
authmode (int) : 0=OPEN,1=WEP,2=WPAPSK,3=WPAPSK/SAE,4=SAEs
timeout_connect: délai pour connecter
"""
def __init__(self, ssid=None, password=None, channel=1, hidden=False,
max_clients=4, authmode=3, timeout_connect=15):
self.ssid = ssid or b"ESP32-" + ubinascii.hexlify(self._iface.config('mac')[-3:]).decode()
self.password = password or ''
self.channel = channel if 1 <= channel <= 13 else 1
self.hidden = bool(hidden)
self.max_clients = max_clients if max_clients > 0 else 4
self.authmode = authmode
self.timeout_connect = timeout_connect
self._iface= network.WLAN(network.AP_IF)
self._mode = 'AP'
def start(self):
if not self._iface.active():
self._iface.active(True)
cfg_auth = 0 if not self.password else self.authmode
try:
self._iface.config(essid=self.ssid, password=self.password, authmode=cfg_auth,
channel=self.channel, hidden=self.hidden, max_clients=self.max_clients)
except Exception as e:
print("start error", e)
return False
time.sleep(self.wifi_delay)
print("Démarrage de la wifi:", self.status())
return True
def stop(self):
if self._iface.active():
self._iface.active(False)
print("Déconnexion de la wifi:", self.status())
def status(self):
return {
'iface': self.get_iface(),
'active': self._iface.active(),
'ssid': self.get_ssid(),
'ip': self.get_ip(),
}
class STAManager(WifiBase):
"""
Gestion WiFi pour ESP32 (MicroPython).
Arguments (tous optionnels selon le mode) :
ssid (str) : nom du réseau AP ou SSID à rejoindre
password (str) : mot de passe AP ou STA (min 8 chars si authmode != 0)
authmode (int) : 0=OPEN,1=WEP,2=WPAPSK,3=WPAPSK/SAE,4=SAEs
timeout_connect: délai pour connecter
"""
def __init__(self, ssid=None, password=None, authmode=3, timeout_connect=10):
self.ssid = ssid
self.password = password
self.authmode = authmode
self.timeout_connect = timeout_connect
self._iface= network.WLAN(network.STA_IF)
self._mode = 'STA'
def start(self):
try:
if not self.ssid:
raise ValueError("SSID requis pour le mode STA")
if self.password and len(self.password) < 8:
raise ValueError("Mot de passe WPA/WPA2 doit faire >=8 caractères")
#self._iface.disconnect()
self._iface.active(False)
time.sleep(self.wifi_delay)
self._iface.active(True)
self._iface.connect(self.ssid, self.password)
t0 = time.time()
wait, n = 0.1, 0
print('Tentative de connexion au réseau...')
while not self._iface.isconnected():
time.sleep(wait)
wait = min(0.5, wait + 0.05)
n+=1
print(n, '> Essai de connexion...', 'status>', self._iface.status())
if (time.time() - t0) > self.timeout_connect:
raise Exception('Timeout connexion atteint') # @UndefinedVariable
time.sleep(self.wifi_delay)
print("Démarrage de la wifi:", self.status())
return True
except Exception as e:
print(f"STA connect() raised: {e}")
return False
def stop(self):
if self._iface.isconnected():
try:
self._iface.disconnect()
except Exception:
pass
print("Déconnexion de la wifi", self.status())
def status(self):
return {
'iface': self.get_iface(),
'active': self._iface.active(),
'connected': self._iface.isconnected(),
'ip': self.get_ip(),
'rssi': self.get_rssi()
}
def WifiManager(mode='AP'):
return APManager if mode == 'AP' else STAManager
www/favicon.ico
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 4.2 KiB

www/index.html
+71
View File
@@ -0,0 +1,71 @@
<!doctype html>
<html lang="fr">
<head>
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no">
<meta charset="utf-8">
<title>Uart controller for esp32</title>
<link rel="icon" href="favicon.ico" sizes="32x32">
<style>
:root {
--bg: #222;
--panel: #14171b;
--danger: #ff3b30;
--text: #e6eef6;
}
html, body {
height:100%; margin:0;
background:var(--bg);
color:var(--text);
font-family: Inter,system-ui, Segoe UI, Roboto, Helvetica,Arial;
}
</style>
</head>
<body>
<div class="app">
<header class="app-header"></header>
<div class="content"></div>
<footer class="footer"> <div class="error"></div></footer>
</div>
<script src="tools.js"></script>
<script>
let ws = null;
const retryDelay = 1000; // 1 seconde
function sendCommand(payload) {
try {
if (!ws || ws.readyState !== WebSocket.OPEN)
return;
ws.send(JSON.stringify(payload));
} catch (e) {}
}
function connect() {
ws = new WebSocket("ws://" + location.host + "/ws");
ws.onopen = () => {
sendCommand({action: 'status'});
};
ws.onclose = (e) => {
console.log("WS closed", e.code, e.reason);
setTimeout(connect, retryDelay);
};
ws.onerror = (e) => {
console.log("WS error", e);
const data = JSON.parse(e.data);
s('.error').textContent = data.error;
ws.close();
};
ws.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
//console.log("Réponse JSON : ", data);
} catch (e) {
//console.log("Message error: ", e);
s('.error').textContent = e;
}
};
}
connect();
</script>
</body>
</html>
www/tools.js
+8
View File
@@ -0,0 +1,8 @@
/*
** tools.js
** GNU GENERAL PUBLIC LICENSE: (c) DD miraceti.net
*/
let s = function(sel) { return document.querySelector(sel); };
let sId = function(sel) { return document.getElementById(sel); };
let isJsonObject = function(value) { return value !== undefined && value !== null && typeof value === 'object'; }
let to_date = function(unixtimestamp) { return new Date(parseInt(unixtimestamp)*1000).toLocaleString(); }