# Minimaler Empfangs-Server: ESP8266 -> POST /api/messwerte -> Website/Unity-Zwilling (GET /api/messwerte) # Der Unity-Zwilling meldet seine Steuerentscheidung per POST /api/steuerung zurueck (GET liefert sie). # 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 } # Steuerbefehle des Unity-Zwillings, z. B. {"window":"open","blind":1.0,"heat":0,"vent":0,"light":0.4,"beamer":0,"szenario":"Hitze"} STEUER_TEXTFELDER = {"window", "szenario"} STEUER_ZAHLFELDER = {"blind", "heat", "vent", "light", "beamer"} steuerung = None 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 lies_json_objekt(self): """Liest den Body als JSON-Objekt. Bei Fehlern wird die Antwort gesendet und None geliefert.""" laenge = int(self.headers.get("Content-Length") or 0) if laenge > MAX_BODY_BYTES: self.sende_json(413, {"fehler": "Anfrage zu gross"}) return None try: daten = json.loads(self.rfile.read(laenge)) except (ValueError, UnicodeDecodeError): self.sende_json(400, {"fehler": "Ungueltiges JSON"}) return None if not isinstance(daten, dict): self.sende_json(400, {"fehler": "JSON-Objekt erwartet"}) return None return daten def do_POST(self): if self.pfad() == "/api/steuerung": return self.steuerung_empfangen() if self.pfad() != "/api/messwerte": return self.sende_json(405, {"fehler": "Methode nicht erlaubt"}) daten = self.lies_json_objekt() if daten is None: return 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 steuerung_empfangen(self): global steuerung daten = self.lies_json_objekt() if daten is None: return eintrag = {"zeit": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")} for feld, wert in daten.items(): if feld in STEUER_TEXTFELDER and isinstance(wert, str): eintrag[feld] = wert[:40] elif feld in STEUER_ZAHLFELDER and isinstance(wert, (int, float)) and not isinstance(wert, bool) and math.isfinite(wert): eintrag[feld] = wert steuerung = eintrag print(datetime.now().strftime("%H:%M:%S"), "Unity-Steuerung", json.dumps(eintrag, ensure_ascii=False), 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)}) if pfad == "/api/steuerung": return self.sende_json(200, {"aktuell": steuerung}) 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