111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
"""Data coordinator for MB-Secure Bridge API v1."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, override
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryAuthFailed
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
|
|
|
from .api import (
|
|
BridgeAuthenticationError,
|
|
BridgeClient,
|
|
BridgeConnectionError,
|
|
BridgeResponseError,
|
|
BridgeVersionError,
|
|
)
|
|
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
|
|
from .models import BridgeDataError, BridgeInfo, BridgeSnapshot
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class MBSecureCoordinator(DataUpdateCoordinator[BridgeSnapshot]):
|
|
"""Coordinate snapshots and push events from one local bridge."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry[MBSecureCoordinator],
|
|
client: BridgeClient,
|
|
) -> None:
|
|
"""Initialize the coordinator."""
|
|
super().__init__(
|
|
hass,
|
|
logger=_LOGGER,
|
|
config_entry=config_entry,
|
|
name=DOMAIN,
|
|
update_interval=DEFAULT_SCAN_INTERVAL,
|
|
always_update=False,
|
|
)
|
|
self.client = client
|
|
self._entry: ConfigEntry[Any] = config_entry
|
|
self.info: BridgeInfo | None = None
|
|
self._event_task: asyncio.Task[None] | None = None
|
|
|
|
@override
|
|
async def _async_update_data(self) -> BridgeSnapshot:
|
|
"""Fetch a full snapshot for startup and periodic reconciliation."""
|
|
try:
|
|
info = await self.client.async_get_info()
|
|
if info.bridge_id is None:
|
|
raise BridgeDataError("bridge installation ID is missing")
|
|
if info.bridge_id != self._entry.unique_id:
|
|
raise BridgeDataError("bridge installation ID has changed")
|
|
self.info = info
|
|
return await self.client.async_get_snapshot()
|
|
except BridgeAuthenticationError as err:
|
|
raise ConfigEntryAuthFailed("Bridge authentication failed") from err
|
|
except (
|
|
BridgeConnectionError,
|
|
BridgeResponseError,
|
|
BridgeVersionError,
|
|
BridgeDataError,
|
|
) as err:
|
|
raise UpdateFailed(f"Bridge update failed: {err}") from err
|
|
|
|
def async_start_event_listener(self) -> None:
|
|
"""Start the config-entry-managed event listener."""
|
|
if self._event_task is not None:
|
|
return
|
|
self._event_task = self._entry.async_create_background_task(
|
|
self.hass,
|
|
self._async_event_loop(),
|
|
f"{DOMAIN}-events-{self._entry.entry_id}",
|
|
)
|
|
|
|
async def async_shutdown(self) -> None:
|
|
"""Stop background event handling."""
|
|
if self._event_task is None:
|
|
return
|
|
self._event_task.cancel()
|
|
await asyncio.gather(self._event_task, return_exceptions=True)
|
|
self._event_task = None
|
|
|
|
async def _async_event_loop(self) -> None:
|
|
"""Reconnect to events and reconcile gaps with a snapshot."""
|
|
reconnect_delay = 1
|
|
while True:
|
|
try:
|
|
async for event in self.client.async_events():
|
|
reconnect_delay = 1
|
|
updated = self.data.apply_event(event)
|
|
if updated is None:
|
|
await self.async_request_refresh()
|
|
else:
|
|
self.async_set_updated_data(updated)
|
|
except BridgeAuthenticationError:
|
|
self._entry.async_start_reauth(self.hass)
|
|
return
|
|
except (BridgeConnectionError, BridgeResponseError, BridgeDataError):
|
|
_LOGGER.debug(
|
|
"Bridge event stream disconnected; retrying in %s seconds",
|
|
reconnect_delay,
|
|
)
|
|
await asyncio.sleep(reconnect_delay)
|
|
reconnect_delay = min(reconnect_delay * 2, 60)
|