import network
import socket
import time
from machine import ADC, Pin
# =========================
# WLAN-Zugangsdaten
# =========================
SSID = "raspi"
PASSWORD = "abcd1234"
# =========================
# ADC-Einstellung
# =========================
adc_pin = ADC(Pin(4)) # ADC-Pin, z.B. GPIO34
adc_pin.atten(ADC.ATTN_11DB) # Bereich ungefähr 0...3,3 V
adc_pin.width(ADC.WIDTH_12BIT) # Werte 0...4095
# =========================
# Messspeicher
# =========================
MAX_N = 2000 # Sicherheitsgrenze
N = 300 # Standardwert
werte = [0] * N
# =========================
# WLAN verbinden
# =========================
def wlan_verbinden():
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if not wlan.isconnected():
print("Verbinde mit WLAN ...")
wlan.connect(SSID, PASSWORD)
for i in range(20):
if wlan.isconnected():
break
print(".", end="")
time.sleep(1)
print()
if wlan.isconnected():
ip = wlan.ifconfig()[0]
print("WLAN verbunden")
print("IP-Adresse:", ip)
return ip
else:
print("Keine WLAN-Verbindung")
return None
# =========================
# einfache URL-Decodierung
# =========================
def url_decode(text):
text = text.replace("+", " ")
result = ""
i = 0
while i < len(text):
if text[i] == "%" and i + 2 < len(text):
try:
result += chr(int(text[i+1:i+3], 16))
i += 3
except:
result += text[i]
i += 1
else:
result += text[i]
i += 1
return result
# =========================
# Query-Parameter aus URL holen
# Beispiel: /messen?n=500
# =========================
def hole_parameter(pfad, name):
suchtext = name + "="
pos = pfad.find(suchtext)
if pos == -1:
return None
start = pos + len(suchtext)
ende = pfad.find("&", start)
if ende == -1:
ende = len(pfad)
return url_decode(pfad[start:ende])
# =========================
# Messung durchführen
# =========================
def messen(anzahl):
global N, werte
if anzahl < 1:
anzahl = 1
if anzahl > MAX_N:
anzahl = MAX_N
N = anzahl
werte = [0] * N
print("Messung startet, N =", N)
t_start = time.ticks_us()
for i in range(N):
werte[i] = adc_pin.read()
t_ende = time.ticks_us()
messzeit_us = time.ticks_diff(t_ende, t_start)
print("Messung abgeschlossen")
print("Messzeit:", messzeit_us, "us")
return messzeit_us
# =========================
# Messwerte als CSV
# =========================
def werte_als_csv():
text = ""
for i in range(N):
text += str(i) + ";" + str(werte[i]) + "\n"
return text
# =========================
# HTML-Seite
# =========================
def html_seite():
return """
ESP32 Web-Speicheroszi
"""
# =========================
# HTTP-Antwort senden
# =========================
def sende_antwort(client, content, content_type="text/plain; charset=utf-8"):
header = "HTTP/1.1 200 OK\r\n"
header += "Content-Type: " + content_type + "\r\n"
header += "Connection: close\r\n\r\n"
client.send(header.encode("utf-8"))
if isinstance(content, str):
client.send(content.encode("utf-8"))
else:
client.send(content)
# =========================
# Webserver
# =========================
def webserver_starten():
ip = wlan_verbinden()
if ip is None:
print("Webserver nicht gestartet.")
return
addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(addr)
s.listen(1)
print("Webserver läuft")
print("Browser öffnen: http://" + ip)
while True:
client = None
try:
client, addr = s.accept()
request = client.recv(1024).decode("utf-8")
# Erste Zeile der HTTP-Anfrage holen
erste_zeile = request.split("\r\n")[0]
teile = erste_zeile.split(" ")
if len(teile) < 2:
client.close()
continue
pfad = teile[1]
print("Anfrage:", pfad)
# -------------------------
# Messung starten
# -------------------------
if pfad.startswith("/messen"):
n_text = hole_parameter(pfad, "n")
try:
anzahl = int(n_text)
except:
anzahl = N
messzeit_us = messen(anzahl)
antwort = "Messung abgeschlossen: "
antwort += str(N) + " Werte, "
antwort += str(messzeit_us) + " us"
sende_antwort(client, antwort)
# -------------------------
# Messwerte senden
# -------------------------
elif pfad.startswith("/werte"):
csv = werte_als_csv()
sende_antwort(client, csv, "text/plain; charset=utf-8")
# -------------------------
# Hauptseite senden
# -------------------------
else:
seite = html_seite()
sende_antwort(client, seite, "text/html; charset=utf-8")
client.close()
except Exception as e:
print("Fehler:", e)
if client:
try:
client.close()
except:
pass
# =========================
# Start
# =========================
webserver_starten()