6 akser, StallGuard og basic web interface med ledningsdiagrammer tilføjet

web/server.py

@@ -0,0 +1,193 @@
+"""Web-interface til stepper-test. Kører på PC'en og taler med ESP32 over USB-serial.
+
+Start:  web/.venv/bin/python web/server.py [--port /dev/ttyUSB0]
+Åbn:    http://127.0.0.1:8000
+"""
+
+import argparse
+import asyncio
+import json
+import threading
+import time
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import serial
+import uvicorn
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.responses import FileResponse
+
+STATIC_DIR = Path(__file__).parent / "static"
+HEARTBEAT_S = 0.5  # firmwarens watchdog stopper efter 2 s uden data
+
+
+class SerialLink:
+    """Én seriel forbindelse til ESP32. Genforbinder automatisk hvis kablet ryger."""
+
+    def __init__(self, port: str, baud: int, on_message):
+        self.port = port
+        self.baud = baud
+        self.on_message = on_message  # kaldes fra læsetråden med en dict
+        self.ser: serial.Serial | None = None
+        self.lock = threading.Lock()
+        self.connected = False
+        self.error = ""
+
+    def start(self):
+        threading.Thread(target=self._run, daemon=True).start()
+
+    def _set_state(self, connected: bool, error: str = ""):
+        if connected != self.connected or error != self.error:
+            self.connected, self.error = connected, error
+            self.on_message({"type": "conn", "connected": connected, "port": self.port, "error": error})
+
+    def _open(self) -> serial.Serial:
+        ser = serial.Serial()
+        ser.port, ser.baudrate, ser.timeout = self.port, self.baud, 0.2
+        # Undgå at DTR/RTS resetter ESP32 når porten åbnes (virker på de fleste boards).
+        ser.dtr = False
+        ser.rts = False
+        ser.open()
+        return ser
+
+    def _run(self):
+        while True:
+            try:
+                ser = self._open()
+            except (serial.SerialException, OSError) as e:
+                self._set_state(False, str(e))
+                time.sleep(2)
+                continue
+            with self.lock:
+                self.ser = ser
+            self._set_state(True)
+            # Reset ved hver forbindelse, så ESP32 altid starter i kendt tilstand (driver disabled).
+            # Firmwaren sender selv boot-log og en status-linje.
+            self.reset()
+            try:
+                while True:
+                    raw = ser.readline()
+                    if raw:
+                        text = raw.decode("utf-8", errors="replace").rstrip("\r\n")
+                        if text:
+                            self.on_message({"type": "line", "text": text})
+            except (serial.SerialException, OSError) as e:
+                with self.lock:
+                    self.ser = None
+                try:
+                    ser.close()
+                except Exception:
+                    pass
+                self._set_state(False, str(e))
+                time.sleep(1)
+
+    def reset(self) -> bool:
+        """Genstart ESP32 via auto-reset-kredsløbet: RTS trækker EN lav, DTR=False holder IO0 høj."""
+        with self.lock:
+            if not self.ser:
+                return False
+            try:
+                self.ser.dtr = False
+                self.ser.rts = True
+                time.sleep(0.15)
+                self.ser.rts = False
+                return True
+            except (serial.SerialException, OSError):
+                return False
+
+    def write(self, line: str) -> bool:
+        with self.lock:
+            if not self.ser:
+                return False
+            try:
+                self.ser.write((line.strip() + "\n").encode())
+                return True
+            except (serial.SerialException, OSError):
+                return False
+
+
+clients: set[WebSocket] = set()
+loop: asyncio.AbstractEventLoop | None = None
+link: SerialLink | None = None
+
+
+async def broadcast(msg: dict):
+    data = json.dumps(msg)
+    for ws in list(clients):
+        try:
+            await ws.send_text(data)
+        except Exception:
+            clients.discard(ws)
+
+
+def on_serial_message(msg: dict):
+    # Kaldes fra læsetråden – flyt over i asyncio-loopet.
+    if loop:
+        asyncio.run_coroutine_threadsafe(broadcast(msg), loop)
+
+
+async def heartbeat():
+    # Kun heartbeat mens en browser er forbundet. Lukkes fanen, stopper firmwaren selv.
+    while True:
+        await asyncio.sleep(HEARTBEAT_S)
+        if clients and link:
+            link.write("hb")
+
+
+@asynccontextmanager
+async def lifespan(_app):
+    global loop
+    loop = asyncio.get_running_loop()
+    link.start()
+    hb = asyncio.create_task(heartbeat())
+    yield
+    hb.cancel()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.get("/")
+async def index():
+    return FileResponse(STATIC_DIR / "index.html")
+
+
+@app.websocket("/ws")
+async def ws_endpoint(ws: WebSocket):
+    await ws.accept()
+    clients.add(ws)
+    await ws.send_text(json.dumps(
+        {"type": "conn", "connected": link.connected, "port": link.port, "error": link.error}))
+    try:
+        while True:
+            cmd = (await ws.receive_text()).strip()
+            if cmd == "!reset":
+                ok = await asyncio.to_thread(link.reset)
+                await ws.send_text(json.dumps(
+                    {"type": "line", "text": "# ESP32 resettes" if ok else "err ikke forbundet til ESP32"}))
+            elif cmd and not link.write(cmd):
+                await ws.send_text(json.dumps({"type": "line", "text": "err ikke forbundet til ESP32"}))
+    except WebSocketDisconnect:
+        pass
+    finally:
+        clients.discard(ws)
+        if not clients:
+            link.write("stop")  # sidste fane lukket → blød stop med det samme
+
+
+def main():
+    global link
+    p = argparse.ArgumentParser(description="Stepper-test web-interface")
+    p.add_argument("--port", default="/dev/ttyUSB0", help="seriel port til ESP32")
+    p.add_argument("--baud", type=int, default=115200)
+    p.add_argument("--host", default="127.0.0.1")
+    p.add_argument("--http-port", type=int, default=8000)
+    args = p.parse_args()
+
+    link = SerialLink(args.port, args.baud, on_serial_message)
+    print(f"Åbn http://{args.host}:{args.http_port}  (ESP32 på {args.port})")
+    uvicorn.run(app, host=args.host, port=args.http_port, log_level="warning")
+
+
+if __name__ == "__main__":
+    main()