feat: WebSocket-Healthcheck und Status-Tracking\n\n- Fügt _WsStatus-Klasse hinzu, die den aktuellen Verbindungsstatus verfolgt\n- Neuer Endpoint /health/websocket gibt Status zurück (connected/connecting/error)\n- Event-Listener aktualisiert den Status bei allen Zustandsänderungen\n- Fallback-Task wird korrekt im lifespan verwaltet\n- Bessere Fehlerbehandlung und Statusmeldungen\n\nImproves observability of the event-based architecture.
Some checks failed
quality / test (3.11) (push) Has been cancelled
quality / test (3.13) (push) Has been cancelled

This commit is contained in:
2026-06-14 17:55:44 +02:00
parent c8f491ba1a
commit 51d23e0a9a

View File

@@ -25,6 +25,17 @@ from backend.routes.ml import init_ml_routes
logger = logging.getLogger(__name__) 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 @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncIterator[None]:
@@ -32,6 +43,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
client: HaClient | None = None client: HaClient | None = None
reconcile_task: asyncio.Task[None] | None = None reconcile_task: asyncio.Task[None] | None = None
event_listener_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.registry = ModelRegistry(settings.model_store)
app.state.actuator_store = ActuatorStore(settings.actuator_store) app.state.actuator_store = ActuatorStore(settings.actuator_store)
if hasattr(app.state, "ha_reader"): if hasattr(app.state, "ha_reader"):
@@ -59,11 +71,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
store=app.state.actuator_store, store=app.state.actuator_store,
settings=settings, settings=settings,
) )
app.state.ws_status = _WsStatus()
await asyncio.to_thread(app.state.actuator_service.reconcile_all, "startup") 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.train_all)
await asyncio.to_thread(app.state.behavior_engine.evaluate_all) await asyncio.to_thread(app.state.behavior_engine.evaluate_all)
reconcile_task = asyncio.create_task(_periodic_reconciliation(app)) reconcile_task = asyncio.create_task(_periodic_reconciliation(app))
event_listener_task = asyncio.create_task(_ha_event_listener(app, client)) event_listener_task = asyncio.create_task(_ha_event_listener(app, client))
fallback_task = asyncio.create_task(_fallback_prediction(app))
try: try:
yield yield
finally: finally:
@@ -75,6 +89,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
event_listener_task.cancel() event_listener_task.cancel()
with suppress(asyncio.CancelledError): with suppress(asyncio.CancelledError):
await event_listener_task 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: if client is not None:
client.close() client.close()
@@ -99,6 +117,19 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
def health() -> dict[str, str]: def health() -> dict[str, str]:
return {"status": "ok"} 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("/") @app.get("/")
def root() -> FileResponse: def root() -> FileResponse:
@@ -127,21 +158,34 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
store = app.state.actuator_store store = app.state.actuator_store
if not isinstance(engine, BehaviorEngine) or not isinstance(store, ActuatorStore): if not isinstance(engine, BehaviorEngine) or not isinstance(store, ActuatorStore):
logger.error("BehaviorEngine oder ActuatorStore nicht initialisiert") 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 return
ha_url = str(settings.ha_url).rstrip("/") ha_url = str(settings.ha_url).rstrip("/")
ws_url = ha_url.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket" ws_url = ha_url.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket"
auth_token = cast(str, settings.ha_token) 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: while True:
try: try:
async with websockets.connect(ws_url, extra_headers={ async with websockets.connect(ws_url, extra_headers={
"Authorization": f"Bearer {auth_token}" "Authorization": f"Bearer {auth_token}"
}) as websocket: }) as websocket:
logger.info("WebSocket-Verbindung zu Home Assistant hergestellt") 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 # Authentifizierung durchführen
auth_msg = await websocket.recv() auth_msg = await websocket.recv()
auth_data = json.loads(auth_msg) auth_data = json.loads(auth_msg)
if auth_data.get("type") != "auth_ok": if auth_data.get("type") != "auth_ok":
logger.error("WebSocket-Authentifizierung fehlgeschlagen") logger.error("WebSocket-Authentifizierung fehlgeschlagen")
if ws_status is not None:
ws_status.status = "error"
ws_status.error = "Authentifizierung fehlgeschlagen"
await asyncio.sleep(5) await asyncio.sleep(5)
continue continue
# Auf alle State Changes subscriben # 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) logger.exception("Fehler bei Event-Verarbeitung: %s", exc)
except (websockets.exceptions.ConnectionClosed, OSError) as exc: except (websockets.exceptions.ConnectionClosed, OSError) as exc:
logger.warning("WebSocket-Verbindung unterbrochen: %s. Wiederholung in 5s...", 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) await asyncio.sleep(5)
except Exception as exc: except Exception as exc:
logger.exception("Unerwarteter Fehler im Event-Listener: %s", 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) await asyncio.sleep(5)
# Fallback: periodische Vorhersage falls Event-Stream ausfällt # Fallback: periodische Vorhersage falls Event-Stream ausfällt
async def _fallback_prediction(app: FastAPI) -> None: 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: while True:
await asyncio.sleep(app.state.settings.prediction_interval_seconds) await asyncio.sleep(app.state.settings.prediction_interval_seconds)
engine = getattr(app.state, "behavior_engine", None) # Nur ausführen, wenn WebSocket nicht verbunden ist
if isinstance(engine, BehaviorEngine): ws_status = getattr(app.state, "ws_status", None)
await asyncio.to_thread(engine.evaluate_all) 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)