Reduce websocket reconnect load
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
name: SillyHome Next
|
name: SillyHome Next
|
||||||
version: "1.7.1"
|
version: "1.7.2"
|
||||||
slug: sillyhome_next
|
slug: sillyhome_next
|
||||||
description: Lernt automatisch aus deinem Verhalten und steuert freigegebene Aktoren
|
description: Lernt automatisch aus deinem Verhalten und steuert freigegebene Aktoren
|
||||||
url: http://192.168.6.31:3000/pino/sillyhome-next
|
url: http://192.168.6.31:3000/pino/sillyhome-next
|
||||||
|
|||||||
44
app/main.py
44
app/main.py
@@ -117,7 +117,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|||||||
app = FastAPI(
|
app = FastAPI(
|
||||||
title="SillyHome Next API",
|
title="SillyHome Next API",
|
||||||
description="Lokales Smart-Home-Intelligenzsystem für Home Assistant.",
|
description="Lokales Smart-Home-Intelligenzsystem für Home Assistant.",
|
||||||
version="1.7.1",
|
version="1.7.2",
|
||||||
lifespan=lifespan,
|
lifespan=lifespan,
|
||||||
)
|
)
|
||||||
app.state.settings = load_settings()
|
app.state.settings = load_settings()
|
||||||
@@ -255,6 +255,9 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
|||||||
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)
|
ws_status = getattr(app.state, "ws_status", None)
|
||||||
|
reconnect_delay = 1.0
|
||||||
|
relevant_entity_ids: set[str] = set()
|
||||||
|
relevant_loaded_at = 0.0
|
||||||
while True:
|
while True:
|
||||||
if ws_status is not None:
|
if ws_status is not None:
|
||||||
ws_status.status = "connecting"
|
ws_status.status = "connecting"
|
||||||
@@ -283,6 +286,9 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
|||||||
|
|
||||||
logger.info("WebSocket-Verbindung zu Home Assistant hergestellt")
|
logger.info("WebSocket-Verbindung zu Home Assistant hergestellt")
|
||||||
state_cache = await asyncio.to_thread(_load_ha_state_cache, ha_reader)
|
state_cache = await asyncio.to_thread(_load_ha_state_cache, ha_reader)
|
||||||
|
relevant_entity_ids = await asyncio.to_thread(_relevant_entity_ids, store)
|
||||||
|
relevant_loaded_at = asyncio.get_running_loop().time()
|
||||||
|
reconnect_delay = 1.0
|
||||||
if ws_status is not None:
|
if ws_status is not None:
|
||||||
ws_status.status = "connected"
|
ws_status.status = "connected"
|
||||||
ws_status.error = None
|
ws_status.error = None
|
||||||
@@ -309,6 +315,12 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
|||||||
entity_id = event_data.get("entity_id")
|
entity_id = event_data.get("entity_id")
|
||||||
if not entity_id:
|
if not entity_id:
|
||||||
continue
|
continue
|
||||||
|
loop_time = asyncio.get_running_loop().time()
|
||||||
|
if loop_time - relevant_loaded_at >= 10:
|
||||||
|
relevant_entity_ids = await asyncio.to_thread(_relevant_entity_ids, store)
|
||||||
|
relevant_loaded_at = loop_time
|
||||||
|
if entity_id not in relevant_entity_ids:
|
||||||
|
continue
|
||||||
new_state = event_data.get("new_state")
|
new_state = event_data.get("new_state")
|
||||||
_update_ha_state_cache(state_cache, entity_id, new_state)
|
_update_ha_state_cache(state_cache, entity_id, new_state)
|
||||||
# Prüfe, ob Entity ein Aktor oder relevanter Kontext ist
|
# Prüfe, ob Entity ein Aktor oder relevanter Kontext ist
|
||||||
@@ -328,17 +340,24 @@ async def _ha_event_listener(app: FastAPI, client: HaClient) -> None:
|
|||||||
websockets.exceptions.InvalidStatus,
|
websockets.exceptions.InvalidStatus,
|
||||||
OSError,
|
OSError,
|
||||||
) as exc:
|
) as exc:
|
||||||
logger.warning("WebSocket-Verbindung unterbrochen: %s. Wiederholung in 1s...", exc)
|
delay = reconnect_delay
|
||||||
|
logger.warning(
|
||||||
|
"WebSocket-Verbindung unterbrochen: %s. Wiederholung in %.0fs...",
|
||||||
|
exc,
|
||||||
|
delay,
|
||||||
|
)
|
||||||
if ws_status is not None:
|
if ws_status is not None:
|
||||||
ws_status.status = "reconnecting"
|
ws_status.status = "reconnecting"
|
||||||
ws_status.error = str(exc)
|
ws_status.error = str(exc)
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(delay)
|
||||||
|
reconnect_delay = min(reconnect_delay * 2, 60.0)
|
||||||
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:
|
if ws_status is not None:
|
||||||
ws_status.status = "error"
|
ws_status.status = "error"
|
||||||
ws_status.error = str(exc)
|
ws_status.error = str(exc)
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(reconnect_delay)
|
||||||
|
reconnect_delay = min(reconnect_delay * 2, 60.0)
|
||||||
|
|
||||||
|
|
||||||
# Fallback: periodische Vorhersage falls Event-Stream ausfällt
|
# Fallback: periodische Vorhersage falls Event-Stream ausfällt
|
||||||
@@ -353,7 +372,7 @@ async def _fallback_prediction(app: FastAPI) -> None:
|
|||||||
await asyncio.sleep(
|
await asyncio.sleep(
|
||||||
app.state.settings.prediction_interval_seconds
|
app.state.settings.prediction_interval_seconds
|
||||||
if websocket_connected
|
if websocket_connected
|
||||||
else min(5, app.state.settings.prediction_interval_seconds)
|
else max(30, app.state.settings.prediction_interval_seconds)
|
||||||
)
|
)
|
||||||
# Nur ausführen, wenn WebSocket nicht verbunden ist
|
# Nur ausführen, wenn WebSocket nicht verbunden ist
|
||||||
ws_status = getattr(app.state, "ws_status", None)
|
ws_status = getattr(app.state, "ws_status", None)
|
||||||
@@ -389,15 +408,14 @@ def _update_ha_state_cache(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _is_relevant_state_change(store: ActuatorStore, entity_id: str) -> bool:
|
def _relevant_entity_ids(store: ActuatorStore) -> set[str]:
|
||||||
|
result: set[str] = set()
|
||||||
for record in store.list():
|
for record in store.list():
|
||||||
if record.actuator_entity_id == entity_id:
|
result.add(record.actuator_entity_id)
|
||||||
return True
|
if record.assignment.selected_numeric_entity_id:
|
||||||
if record.assignment.selected_numeric_entity_id == entity_id:
|
result.add(record.assignment.selected_numeric_entity_id)
|
||||||
return True
|
result.update(record.assignment.selected_context_entity_ids)
|
||||||
if entity_id in record.assignment.selected_context_entity_ids:
|
return result
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def _ha_entity_from_event(
|
def _ha_entity_from_event(
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "sillyhome-next"
|
name = "sillyhome-next"
|
||||||
version = "1.7.1"
|
version = "1.7.2"
|
||||||
description = "Lokales Smart-Home-Intelligenzsystem für Home Assistant"
|
description = "Lokales Smart-Home-Intelligenzsystem für Home Assistant"
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -126,6 +126,43 @@ def test_ha_event_listener_processes_state_change(tmp_path: Path) -> None:
|
|||||||
assert mock_app.state.ws_status.error is None
|
assert mock_app.state.ws_status.error is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_ha_event_listener_skips_unrelated_state_change(tmp_path: Path) -> None:
|
||||||
|
async def run_test() -> None:
|
||||||
|
fake_ws = _FakeWebSocket(
|
||||||
|
[
|
||||||
|
'{"type":"auth_required"}',
|
||||||
|
'{"type":"auth_ok"}',
|
||||||
|
(
|
||||||
|
'{"type":"event","event":{"event_type":"state_changed",'
|
||||||
|
'"data":{"entity_id":"sensor.unused","new_state":{"state":"on"}}}}'
|
||||||
|
),
|
||||||
|
asyncio.CancelledError(),
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch("websockets.connect", return_value=fake_ws):
|
||||||
|
try:
|
||||||
|
await _ha_event_listener(mock_app, mock_client)
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
mock_app = MagicMock()
|
||||||
|
mock_app.state.settings = MagicMock()
|
||||||
|
mock_app.state.settings.ha_url = "http://homeassistant:8123"
|
||||||
|
mock_app.state.settings.ha_token = "test-token"
|
||||||
|
mock_app.state.ws_status = MagicMock()
|
||||||
|
mock_engine = _RecordingBehaviorEngine(tmp_path)
|
||||||
|
mock_app.state.behavior_engine = mock_engine
|
||||||
|
mock_app.state.ha_reader = _FakeHaReader()
|
||||||
|
mock_store = ActuatorStore(tmp_path / "store")
|
||||||
|
mock_store.configure("light.test")
|
||||||
|
mock_app.state.actuator_store = mock_store
|
||||||
|
mock_client = MagicMock()
|
||||||
|
|
||||||
|
anyio.run(run_test)
|
||||||
|
assert mock_engine.state_changes == []
|
||||||
|
|
||||||
|
|
||||||
def test_lifespan_skips_event_listener_without_ha_config() -> None:
|
def test_lifespan_skips_event_listener_without_ha_config() -> None:
|
||||||
app = FastAPI()
|
app = FastAPI()
|
||||||
app.state.settings = MagicMock()
|
app.state.settings = MagicMock()
|
||||||
|
|||||||
Reference in New Issue
Block a user