Speed up HA event processing
This commit is contained in:
27
app/main.py
27
app/main.py
@@ -105,7 +105,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
app = FastAPI(
|
||||
title="SillyHome Next API",
|
||||
description="Lokales Smart-Home-Intelligenzsystem für Home Assistant.",
|
||||
version="0.7.16",
|
||||
version="0.7.17",
|
||||
lifespan=lifespan,
|
||||
)
|
||||
app.state.settings = load_settings()
|
||||
@@ -211,7 +211,11 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
||||
if ws_status is not None:
|
||||
ws_status.status = "connecting"
|
||||
try:
|
||||
async with websockets.connect(ws_url, ping_interval=None) as websocket:
|
||||
async with websockets.connect(
|
||||
ws_url,
|
||||
ping_interval=20,
|
||||
ping_timeout=10,
|
||||
) as websocket:
|
||||
auth_required_msg = await websocket.recv()
|
||||
auth_required_data = json.loads(auth_required_msg)
|
||||
if auth_required_data.get("type") != "auth_required":
|
||||
@@ -263,6 +267,8 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
||||
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(
|
||||
@@ -280,17 +286,17 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
||||
websockets.exceptions.InvalidStatus,
|
||||
OSError,
|
||||
) as exc:
|
||||
logger.warning("WebSocket-Verbindung unterbrochen: %s. Wiederholung in 5s...", 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(5)
|
||||
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(5)
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
# Fallback: periodische Vorhersage falls Event-Stream ausfällt
|
||||
@@ -341,6 +347,17 @@ def _update_ha_state_cache(
|
||||
)
|
||||
|
||||
|
||||
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],
|
||||
|
||||
Reference in New Issue
Block a user