452 lines
18 KiB
Python
452 lines
18 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from contextlib import asynccontextmanager, suppress
|
|
from collections.abc import AsyncIterator
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import websockets
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app.actuators.cache_db import DashboardCache
|
|
from app.actuators.lifecycle import ActuatorReconciliationService
|
|
from app.actuators.store import ActuatorStore
|
|
from app.api.v1.actuators import router as actuators_router
|
|
from app.api.v1.entities import router as entities_router
|
|
from app.behavior.engine import BehaviorEngine
|
|
from app.config import load_settings
|
|
from app.core.exception_handlers import register_exception_handlers
|
|
from app.ha.client import HaClient, HaClientSettings
|
|
from app.ha.discovery import discover_entities
|
|
from app.ha.models import HaEntitySummary
|
|
from app.ha.reader import HaReader
|
|
from app.ml.registry.model_registry import ModelRegistry
|
|
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" | "reconnecting" | "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]:
|
|
settings = app.state.settings
|
|
client: HaClient | None = None
|
|
startup_task: asyncio.Task[None] | None = None
|
|
reconcile_task: asyncio.Task[None] | None = None
|
|
event_listener_task: asyncio.Task[None] | None = None
|
|
fallback_task: asyncio.Task[None] | None = None
|
|
cache_refresh_task: asyncio.Task[None] | None = None
|
|
app.state.registry = ModelRegistry(settings.model_store)
|
|
app.state.actuator_store = ActuatorStore(settings.actuator_store)
|
|
app.state.dashboard_cache = DashboardCache(
|
|
Path(settings.actuator_store).resolve() / "dashboard_cache.sqlite3"
|
|
)
|
|
if hasattr(app.state, "ha_reader"):
|
|
del app.state.ha_reader
|
|
if hasattr(app.state, "actuator_service"):
|
|
del app.state.actuator_service
|
|
if hasattr(app.state, "behavior_engine"):
|
|
del app.state.behavior_engine
|
|
if settings.ha_configured:
|
|
client = HaClient(
|
|
settings=HaClientSettings(
|
|
url=cast(str, settings.ha_url),
|
|
token=cast(str, settings.ha_token),
|
|
timeout_seconds=settings.ha_timeout_seconds,
|
|
)
|
|
)
|
|
app.state.ha_reader = HaReader(client=client)
|
|
app.state.actuator_service = ActuatorReconciliationService(
|
|
ha_reader=app.state.ha_reader,
|
|
store=app.state.actuator_store,
|
|
registry=app.state.registry,
|
|
settings=settings,
|
|
)
|
|
app.state.behavior_engine = BehaviorEngine(
|
|
ha_reader=app.state.ha_reader,
|
|
store=app.state.actuator_store,
|
|
settings=settings,
|
|
)
|
|
app.state.ws_status = _WsStatus()
|
|
startup_task = asyncio.create_task(_startup_reconciliation(app))
|
|
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))
|
|
cache_refresh_task = asyncio.create_task(_periodic_dashboard_cache_refresh(app))
|
|
try:
|
|
yield
|
|
finally:
|
|
if startup_task is not None:
|
|
startup_task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await startup_task
|
|
if reconcile_task is not None:
|
|
reconcile_task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await reconcile_task
|
|
if event_listener_task is not 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 cache_refresh_task is not None:
|
|
cache_refresh_task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await cache_refresh_task
|
|
if client is not None:
|
|
client.close()
|
|
|
|
|
|
app = FastAPI(
|
|
title="SillyHome Next API",
|
|
description="Lokales Smart-Home-Intelligenzsystem für Home Assistant.",
|
|
version="1.5.4",
|
|
lifespan=lifespan,
|
|
)
|
|
app.state.settings = load_settings()
|
|
register_exception_handlers(app)
|
|
app.include_router(entities_router)
|
|
app.include_router(actuators_router)
|
|
init_ml_routes(app, model_store=app.state.settings.model_store)
|
|
|
|
STATIC_DIR = Path(__file__).with_name("static")
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
|
|
|
|
@app.get("/health")
|
|
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" | "reconnecting" | "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:
|
|
return FileResponse(
|
|
STATIC_DIR / "index.html",
|
|
headers={"Cache-Control": "no-store, max-age=0"},
|
|
)
|
|
|
|
|
|
async def _periodic_reconciliation(app: FastAPI) -> None:
|
|
while True:
|
|
await asyncio.sleep(app.state.settings.reconcile_interval_seconds)
|
|
service = getattr(app.state, "actuator_service", None)
|
|
if not isinstance(service, ActuatorReconciliationService):
|
|
continue
|
|
try:
|
|
await asyncio.to_thread(service.reconcile_all, "scheduled")
|
|
engine = getattr(app.state, "behavior_engine", None)
|
|
if isinstance(engine, BehaviorEngine):
|
|
await asyncio.to_thread(engine.train_all)
|
|
except Exception:
|
|
logger.exception("Geplante Reconciliation fehlgeschlagen; nächster Lauf versucht es erneut.")
|
|
|
|
|
|
async def _periodic_dashboard_cache_refresh(app: FastAPI) -> None:
|
|
await asyncio.sleep(2)
|
|
while True:
|
|
await _refresh_dashboard_cache(app, trigger="scheduled")
|
|
await asyncio.sleep(app.state.settings.dashboard_cache_refresh_seconds)
|
|
|
|
|
|
async def _refresh_dashboard_cache(app: FastAPI, *, trigger: str) -> None:
|
|
ha_reader = getattr(app.state, "ha_reader", None)
|
|
cache = getattr(app.state, "dashboard_cache", None)
|
|
if not isinstance(ha_reader, HaReader) or not isinstance(cache, DashboardCache):
|
|
return
|
|
try:
|
|
entities = await asyncio.to_thread(ha_reader.read_entities)
|
|
groups = _discovery_group_payload(list(entities))
|
|
await asyncio.to_thread(
|
|
cache.save_entities_payload,
|
|
entities=list(entities),
|
|
discovery_groups=groups,
|
|
)
|
|
logger.info("Dashboard-Cache aktualisiert (%s): %d Entities", trigger, len(entities))
|
|
except Exception as exc:
|
|
logger.warning("Dashboard-Cache konnte nicht aktualisiert werden (%s): %s", trigger, exc)
|
|
|
|
|
|
def _discovery_group_payload(entities: list[HaEntitySummary]) -> list[dict[str, object]]:
|
|
group_counts: dict[tuple[str, str], int] = {}
|
|
for entity in discover_entities(entities):
|
|
key = (entity.category, entity.role.value)
|
|
group_counts[key] = group_counts.get(key, 0) + 1
|
|
return [
|
|
{"category": category, "role": role, "count": count}
|
|
for (category, role), count in sorted(group_counts.items())
|
|
]
|
|
|
|
|
|
async def _startup_reconciliation(app: FastAPI) -> None:
|
|
delay_seconds = 5
|
|
while True:
|
|
service = getattr(app.state, "actuator_service", None)
|
|
engine = getattr(app.state, "behavior_engine", None)
|
|
if not isinstance(service, ActuatorReconciliationService) or not isinstance(
|
|
engine,
|
|
BehaviorEngine,
|
|
):
|
|
return
|
|
try:
|
|
await asyncio.to_thread(service.reconcile_all, "startup")
|
|
await asyncio.to_thread(engine.train_all)
|
|
await asyncio.to_thread(engine.evaluate_all)
|
|
logger.info("Startup-Reconciliation erfolgreich abgeschlossen.")
|
|
return
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Startup-Reconciliation verschoben: %s. Neuer Versuch in %ss.",
|
|
exc,
|
|
delay_seconds,
|
|
)
|
|
await asyncio.sleep(delay_seconds)
|
|
delay_seconds = min(delay_seconds * 2, 60)
|
|
|
|
|
|
async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
|
"""Hört auf Home-Assistant-Websocket-Events und löst sofortige Vorhersagen aus."""
|
|
settings = app.state.settings
|
|
engine = app.state.behavior_engine
|
|
ha_reader = getattr(app.state, "ha_reader", None)
|
|
store = app.state.actuator_store
|
|
if (
|
|
not isinstance(engine, BehaviorEngine)
|
|
or not isinstance(store, ActuatorStore)
|
|
or not isinstance(ha_reader, HaReader)
|
|
):
|
|
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
|
|
state_cache: dict[str, HaEntitySummary] = {}
|
|
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)
|
|
while True:
|
|
if ws_status is not None:
|
|
ws_status.status = "connecting"
|
|
try:
|
|
async with websockets.connect(
|
|
ws_url,
|
|
ping_interval=30,
|
|
ping_timeout=30,
|
|
) as websocket:
|
|
auth_required_msg = await websocket.recv()
|
|
auth_required_data = json.loads(auth_required_msg)
|
|
if auth_required_data.get("type") != "auth_required":
|
|
logger.error("Unerwartete WebSocket-Authentifizierungsaufforderung")
|
|
if ws_status is not None:
|
|
ws_status.status = "error"
|
|
ws_status.error = "Unerwartete Authentifizierungsaufforderung"
|
|
await asyncio.sleep(5)
|
|
continue
|
|
|
|
await websocket.send(json.dumps({"type": "auth", "access_token": auth_token}))
|
|
auth_result_msg = await websocket.recv()
|
|
auth_result_data = json.loads(auth_result_msg)
|
|
if auth_result_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
|
|
|
|
logger.info("WebSocket-Verbindung zu Home Assistant hergestellt")
|
|
state_cache = await asyncio.to_thread(_load_ha_state_cache, ha_reader)
|
|
if ws_status is not None:
|
|
ws_status.status = "connected"
|
|
ws_status.error = None
|
|
# Auf alle State Changes subscriben
|
|
subscribe_msg = {
|
|
"id": 1,
|
|
"type": "subscribe_events",
|
|
"event_type": "state_changed"
|
|
}
|
|
await websocket.send(json.dumps(subscribe_msg))
|
|
while True:
|
|
message = await websocket.recv()
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") != "event":
|
|
continue
|
|
event = data.get("event", {})
|
|
if event.get("event_type") != "state_changed":
|
|
continue
|
|
event_data = event.get("data", {})
|
|
if not isinstance(event_data, dict):
|
|
logger.warning("State-Changed-Event ohne gültige Daten empfangen")
|
|
continue
|
|
entity_id = event_data.get("entity_id")
|
|
if not entity_id:
|
|
continue
|
|
new_state = event_data.get("new_state")
|
|
_update_ha_state_cache(state_cache, entity_id, new_state)
|
|
if not _is_relevant_state_change(store, str(entity_id)):
|
|
continue
|
|
# Prüfe, ob Entity ein Aktor oder relevanter Kontext ist
|
|
# Sofortige Vorhersage für betroffene Aktoren auslösen
|
|
await asyncio.to_thread(
|
|
engine.handle_state_change,
|
|
entity_id,
|
|
new_state,
|
|
current_entities=list(state_cache.values()),
|
|
)
|
|
except json.JSONDecodeError:
|
|
logger.warning("Ungültige JSON-Nachricht von HA-WebSocket")
|
|
except Exception as exc:
|
|
logger.exception("Fehler bei Event-Verarbeitung: %s", exc)
|
|
except (
|
|
websockets.exceptions.ConnectionClosed,
|
|
websockets.exceptions.InvalidStatus,
|
|
OSError,
|
|
) as exc:
|
|
logger.warning("WebSocket-Verbindung unterbrochen: %s. Wiederholung in 1s...", exc)
|
|
if ws_status is not None:
|
|
ws_status.status = "reconnecting"
|
|
ws_status.error = str(exc)
|
|
await asyncio.sleep(1)
|
|
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(1)
|
|
|
|
|
|
# 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:
|
|
ws_status = getattr(app.state, "ws_status", None)
|
|
websocket_connected = ws_status is not None and ws_status.status == "connected"
|
|
await asyncio.sleep(
|
|
app.state.settings.prediction_interval_seconds
|
|
if websocket_connected
|
|
else min(5, app.state.settings.prediction_interval_seconds)
|
|
)
|
|
# 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",
|
|
)
|
|
try:
|
|
await asyncio.to_thread(engine.evaluate_all)
|
|
except Exception:
|
|
logger.exception("Fallback-Vorhersage fehlgeschlagen.")
|
|
|
|
|
|
def _load_ha_state_cache(reader: HaReader) -> dict[str, HaEntitySummary]:
|
|
return {entity.entity_id: entity for entity in reader.read_entities()}
|
|
|
|
|
|
def _update_ha_state_cache(
|
|
state_cache: dict[str, HaEntitySummary],
|
|
entity_id: str,
|
|
new_state: object,
|
|
) -> None:
|
|
if not isinstance(new_state, dict):
|
|
state_cache.pop(entity_id, None)
|
|
return
|
|
state_cache[entity_id] = _ha_entity_from_event(
|
|
entity_id,
|
|
new_state,
|
|
state_cache.get(entity_id),
|
|
)
|
|
|
|
|
|
def _is_relevant_state_change(store: ActuatorStore, entity_id: str) -> bool:
|
|
for record in store.list():
|
|
if record.actuator_entity_id == entity_id:
|
|
return True
|
|
if record.assignment.selected_numeric_entity_id == entity_id:
|
|
return True
|
|
if entity_id in record.assignment.selected_context_entity_ids:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _ha_entity_from_event(
|
|
entity_id: str,
|
|
new_state: dict[str, object],
|
|
previous: HaEntitySummary | None,
|
|
) -> HaEntitySummary:
|
|
attributes = new_state.get("attributes")
|
|
attr = attributes if isinstance(attributes, dict) else {}
|
|
state_class = _optional_event_string(attr.get("state_class"))
|
|
device_class = _optional_event_string(attr.get("device_class"))
|
|
unit_of_measurement = _optional_event_string(attr.get("unit_of_measurement"))
|
|
friendly_name = _optional_event_string(attr.get("friendly_name"))
|
|
return HaEntitySummary(
|
|
entity_id=entity_id,
|
|
domain=entity_id.split(".", 1)[0],
|
|
state=_optional_event_string(new_state.get("state")),
|
|
last_changed=_event_datetime(new_state.get("last_changed"))
|
|
or _event_datetime(new_state.get("last_updated")),
|
|
state_class=state_class or (previous.state_class if previous else None),
|
|
device_class=device_class or (previous.device_class if previous else None),
|
|
unit_of_measurement=unit_of_measurement
|
|
or (previous.unit_of_measurement if previous else None),
|
|
friendly_name=friendly_name or (previous.friendly_name if previous else None),
|
|
area_id=previous.area_id if previous else None,
|
|
area_name=previous.area_name if previous else None,
|
|
device_id=previous.device_id if previous else None,
|
|
device_name=previous.device_name if previous else None,
|
|
)
|
|
|
|
|
|
def _optional_event_string(value: object) -> str | None:
|
|
return value if isinstance(value, str) else None
|
|
|
|
|
|
def _event_datetime(value: object) -> datetime | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed
|