136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
# Minimaler Empfangs-Server: ESP8266 -> POST /api/messwerte -> Website (GET /api/messwerte)
|
|
# Nur Python-Bordmittel, kein pip install noetig. Start: python server.py [Port] (oder start.bat)
|
|
import json
|
|
import math
|
|
import re
|
|
import socket
|
|
import sys
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 3000
|
|
HIER = Path(__file__).resolve().parent
|
|
FRONTEND_DIR = HIER.parent / "frontend"
|
|
WIFI_CONFIG = HIER.parent / "firmware" / "dht11_lcd" / "wifi_config.h"
|
|
MAX_VERLAUF = 100
|
|
MAX_BODY_BYTES = 1024
|
|
|
|
# Erlaubte Felder des Datenvertrags (so sendet esp_wifi.h)
|
|
FELDER = ["temperatur", "luftfeuchte", "eco2", "tvoc"]
|
|
|
|
verlauf = deque(maxlen=MAX_VERLAUF) # { zeit: ISO-String, temperatur, luftfeuchte, eco2, tvoc }
|
|
|
|
MIME = {
|
|
".html": "text/html; charset=utf-8",
|
|
".js": "text/javascript; charset=utf-8",
|
|
".css": "text/css; charset=utf-8",
|
|
".svg": "image/svg+xml",
|
|
}
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def sende_json(self, status, obj):
|
|
daten = json.dumps(obj).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Content-Length", str(len(daten)))
|
|
self.end_headers()
|
|
self.wfile.write(daten)
|
|
|
|
def pfad(self):
|
|
return self.path.split("?", 1)[0]
|
|
|
|
def do_POST(self):
|
|
if self.pfad() != "/api/messwerte":
|
|
return self.sende_json(405, {"fehler": "Methode nicht erlaubt"})
|
|
|
|
laenge = int(self.headers.get("Content-Length") or 0)
|
|
if laenge > MAX_BODY_BYTES:
|
|
return self.sende_json(413, {"fehler": "Anfrage zu gross"})
|
|
try:
|
|
daten = json.loads(self.rfile.read(laenge))
|
|
except (ValueError, UnicodeDecodeError):
|
|
return self.sende_json(400, {"fehler": "Ungueltiges JSON"})
|
|
if not isinstance(daten, dict):
|
|
return self.sende_json(400, {"fehler": "JSON-Objekt erwartet"})
|
|
|
|
eintrag = {"zeit": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")}
|
|
for feld in FELDER:
|
|
# Fehlende Sensoren (z. B. CCS811 noch nicht bereit) duerfen fehlen/null sein
|
|
wert = daten.get(feld)
|
|
if wert is None:
|
|
eintrag[feld] = None
|
|
elif isinstance(wert, (int, float)) and not isinstance(wert, bool) and math.isfinite(wert):
|
|
eintrag[feld] = wert
|
|
else:
|
|
return self.sende_json(400, {"fehler": f"Feld '{feld}' muss eine Zahl sein"})
|
|
|
|
verlauf.append(eintrag)
|
|
print(datetime.now().strftime("%H:%M:%S"), self.client_address[0], json.dumps(eintrag), flush=True)
|
|
self.sende_json(200, {"ok": True})
|
|
|
|
def do_GET(self):
|
|
pfad = self.pfad()
|
|
if pfad == "/api/messwerte":
|
|
return self.sende_json(200, {"aktuell": verlauf[-1] if verlauf else None, "verlauf": list(verlauf)})
|
|
|
|
datei = (FRONTEND_DIR / ("index.html" if pfad == "/" else pfad.lstrip("/"))).resolve()
|
|
if FRONTEND_DIR not in datei.parents:
|
|
return self.sende_json(403, {"fehler": "Verboten"})
|
|
if not datei.is_file():
|
|
return self.sende_json(404, {"fehler": "Nicht gefunden"})
|
|
inhalt = datei.read_bytes()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", MIME.get(datei.suffix, "application/octet-stream"))
|
|
self.send_header("Content-Length", str(len(inhalt)))
|
|
self.end_headers()
|
|
self.wfile.write(inhalt)
|
|
|
|
def log_message(self, *args):
|
|
pass # keine Zeile pro Seitenabruf, nur empfangene Messwerte werden ausgegeben
|
|
|
|
|
|
def eigene_ips():
|
|
"""IPv4-Adressen dieses PCs (ohne 127.x), z. B. 172.20.10.x im iPhone-Hotspot."""
|
|
try:
|
|
infos = socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET)
|
|
except socket.gaierror:
|
|
return []
|
|
return sorted({info[4][0] for info in infos if not info[4][0].startswith("127.")})
|
|
|
|
|
|
def konfigurierte_server_ip():
|
|
try:
|
|
treffer = re.search(r'#define\s+SERVER_IP\s+"([^"]+)"', WIFI_CONFIG.read_text(encoding="utf-8"))
|
|
except OSError:
|
|
return None
|
|
return treffer.group(1) if treffer else None
|
|
|
|
|
|
def ip_hinweis():
|
|
ips = eigene_ips()
|
|
soll = konfigurierte_server_ip()
|
|
print("IP-Adressen dieses PCs:", ", ".join(ips) or "keine (nicht verbunden?)")
|
|
if soll is None:
|
|
return
|
|
if soll in ips:
|
|
print(f"OK: SERVER_IP in wifi_config.h ({soll}) passt zu diesem PC.")
|
|
else:
|
|
print(f"ACHTUNG: wifi_config.h sendet an {soll}, diese IP hat der PC gerade nicht!")
|
|
print(" -> Mit dem Hotspot verbinden bzw. SERVER_IP auf die Hotspot-IP oben aendern und neu flashen.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 0.0.0.0, damit der ESP8266 im WLAN/Hotspot den Server erreicht
|
|
server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
|
|
print(f"Server laeuft auf Port {PORT} (Website: http://localhost:{PORT})")
|
|
ip_hinweis()
|
|
print("Beenden mit Strg+C\n", flush=True)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|