From 51d23e0a9afc1ebe29512f4cf02e91ceb1ff9ec2 Mon Sep 17 00:00:00 2001 From: Otto Date: Sun, 14 Jun 2026 17:55:44 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20WebSocket-Healthcheck=20und=20Status-Tr?= =?UTF-8?q?acking\n\n-=20F=C3=BCgt=20=5FWsStatus-Klasse=20hinzu,=20die=20d?= =?UTF-8?q?en=20aktuellen=20Verbindungsstatus=20verfolgt\n-=20Neuer=20Endp?= =?UTF-8?q?oint=20/health/websocket=20gibt=20Status=20zur=C3=BCck=20(conne?= =?UTF-8?q?cted/connecting/error)\n-=20Event-Listener=20aktualisiert=20den?= =?UTF-8?q?=20Status=20bei=20allen=20Zustands=C3=A4nderungen\n-=20Fallback?= =?UTF-8?q?-Task=20wird=20korrekt=20im=20lifespan=20verwaltet\n-=20Bessere?= =?UTF-8?q?=20Fehlerbehandlung=20und=20Statusmeldungen\n\nImproves=20obser?= =?UTF-8?q?vability=20of=20the=20event-based=20architecture.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 67 ++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 64 insertions(+), 3 deletions(-) diff --git a/app/main.py b/app/main.py index fb8f54a..4b7df19 100644 --- a/app/main.py +++ b/app/main.py @@ -25,6 +25,17 @@ from backend.routes.ml import init_ml_routes logger = logging.getLogger(__name__) +class _WsStatus: + """Einfacher Status-Tracker für den WebSocket-Listener. + + Wird als Attribut an app.state gehängt und enthält: + - status: "disconnected" | "connecting" | "connected" | "error" + - error: str | None (Fehlermeldung bei status=error) + """ + def __init__(self) -> None: + self.status: str = "disconnected" + self.error: str | None = None + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: @@ -32,6 +43,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: client: HaClient | None = None reconcile_task: asyncio.Task[None] | None = None event_listener_task: asyncio.Task[None] | None = None + fallback_task: asyncio.Task[None] | None = None app.state.registry = ModelRegistry(settings.model_store) app.state.actuator_store = ActuatorStore(settings.actuator_store) if hasattr(app.state, "ha_reader"): @@ -59,11 +71,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: store=app.state.actuator_store, settings=settings, ) + app.state.ws_status = _WsStatus() await asyncio.to_thread(app.state.actuator_service.reconcile_all, "startup") await asyncio.to_thread(app.state.behavior_engine.train_all) await asyncio.to_thread(app.state.behavior_engine.evaluate_all) reconcile_task = asyncio.create_task(_periodic_reconciliation(app)) event_listener_task = asyncio.create_task(_ha_event_listener(app, client)) + fallback_task = asyncio.create_task(_fallback_prediction(app)) try: yield finally: @@ -75,6 +89,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: event_listener_task.cancel() with suppress(asyncio.CancelledError): await event_listener_task + if fallback_task is not None: + fallback_task.cancel() + with suppress(asyncio.CancelledError): + await fallback_task if client is not None: client.close() @@ -99,6 +117,19 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") def health() -> dict[str, str]: return {"status": "ok"} +@app.get("/health/websocket") +def websocket_health() -> dict[str, object]: + """Gibt den aktuellen Status des WebSocket-Listeners zurück. + + Antwort: + - status: "disconnected" | "connecting" | "connected" | "error" + - error: str | None (nur bei status=error) + """ + ws_status = getattr(app.state, "ws_status", None) + if ws_status is None: + return {"status": "unavailable", "error": "WebSocket-Listener nicht initialisiert"} + return {"status": ws_status.status, "error": ws_status.error} + @app.get("/") def root() -> FileResponse: @@ -127,21 +158,34 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None: store = app.state.actuator_store if not isinstance(engine, BehaviorEngine) or not isinstance(store, ActuatorStore): logger.error("BehaviorEngine oder ActuatorStore nicht initialisiert") + ws_status = getattr(app.state, "ws_status", None) + if ws_status is not None: + ws_status.status = "error" + ws_status.error = "BehaviorEngine oder ActuatorStore nicht initialisiert" return ha_url = str(settings.ha_url).rstrip("/") ws_url = ha_url.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket" auth_token = cast(str, settings.ha_token) + ws_status = getattr(app.state, "ws_status", None) + if ws_status is not None: + ws_status.status = "connecting" while True: try: async with websockets.connect(ws_url, extra_headers={ "Authorization": f"Bearer {auth_token}" }) as websocket: logger.info("WebSocket-Verbindung zu Home Assistant hergestellt") + if ws_status is not None: + ws_status.status = "connected" + ws_status.error = None # Authentifizierung durchführen auth_msg = await websocket.recv() auth_data = json.loads(auth_msg) if auth_data.get("type") != "auth_ok": logger.error("WebSocket-Authentifizierung fehlgeschlagen") + if ws_status is not None: + ws_status.status = "error" + ws_status.error = "Authentifizierung fehlgeschlagen" await asyncio.sleep(5) continue # Auf alle State Changes subscriben @@ -171,16 +215,33 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None: logger.exception("Fehler bei Event-Verarbeitung: %s", exc) except (websockets.exceptions.ConnectionClosed, OSError) as exc: logger.warning("WebSocket-Verbindung unterbrochen: %s. Wiederholung in 5s...", exc) + if ws_status is not None: + ws_status.status = "reconnecting" + ws_status.error = str(exc) await asyncio.sleep(5) except Exception as exc: logger.exception("Unerwarteter Fehler im Event-Listener: %s", exc) + if ws_status is not None: + ws_status.status = "error" + ws_status.error = str(exc) await asyncio.sleep(5) # Fallback: periodische Vorhersage falls Event-Stream ausfällt async def _fallback_prediction(app: FastAPI) -> None: + """Periodische Vorhersage als Fallback, wenn WebSocket-Listener nicht verbunden ist. + + Dies verhindert kompletten Ausfall der Vorhersagen bei Netzwerkproblemen. + """ while True: await asyncio.sleep(app.state.settings.prediction_interval_seconds) - engine = getattr(app.state, "behavior_engine", None) - if isinstance(engine, BehaviorEngine): - await asyncio.to_thread(engine.evaluate_all) + # Nur ausführen, wenn WebSocket nicht verbunden ist + ws_status = getattr(app.state, "ws_status", None) + if ws_status is None or ws_status.status != "connected": + engine = getattr(app.state, "behavior_engine", None) + if isinstance(engine, BehaviorEngine): + logger.debug( + "Fallback-Vorhersage aktiv (WebSocket-Status: %s)", + ws_status.status if ws_status else "unavailable", + ) + await asyncio.to_thread(engine.evaluate_all)