lippert-onecontrol/custom_components/lippert_onecontrol/coordinator.py

import asyncio
import logging
from homeassistant.core import HomeAssistant
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator

from .client.onecontrol_client import EventType, OneControlClient, MovementState
from .const import DOMAIN, RECONNECT_DELAYS, KEEPALIVE_INTERVAL, MAX_KEEPALIVE_FAILURES

_LOGGER = logging.getLogger(__name__)


class OneControlCoordinator(DataUpdateCoordinator):
    """Manages the BLE connection and pushes state to HA entities."""

    def __init__(self, hass: HomeAssistant, address: str) -> None:
        super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=None)
        self.address = address
        self._switch_states: dict[int, bool] = {}
        self._tank_levels:   dict[int, int]  = {}
        self._battery_v:     float | None    = None
        self._client = OneControlClient(
            address,
            state_callback=self._on_state,
            disconnect_callback=self._on_disconnect,
        )
        self._keepalive_task: asyncio.Task | None = None
        self._reconnect_task: asyncio.Task | None = None

    async def async_connect(self) -> None:
        self._refresh_ble_device()
        await self._client.connect()
        await self._sync_state()
        self._start_keepalive()

    async def _sync_state(self) -> None:
        """Force an immediate state poll so HA reflects panel reality."""
        try:
            await self._client.get_devices()
            await asyncio.sleep(1)  # give notifications time to arrive
            await self._client.get_devices()  # second poll for reliability
        except Exception as e:
            _LOGGER.warning("Initial state sync failed: %s", e)

    def _refresh_ble_device(self) -> None:
        """Get the latest BLE device reference from HA's scanner."""
        ble_device = async_ble_device_from_address(self.hass, self.address)
        if ble_device:
            self._client.set_ble_device(ble_device)
        else:
            _LOGGER.warning("BLE device %s not found by HA scanner — using raw address", self.address)

    async def async_disconnect(self) -> None:
        self._cancel_tasks()
        await self._client.disconnect()

    # ------------------------------------------------------------------
    # State accessors
    # ------------------------------------------------------------------

    @property
    def is_connected(self) -> bool:
        return self._client.is_connected

    def get_switch_state(self, device_id: int) -> bool | None:
        return self._switch_states.get(device_id)

    def get_tank_level(self, device_id: int) -> int | None:
        return self._tank_levels.get(device_id)

    @property
    def battery_voltage(self) -> float | None:
        return self._battery_v

    # ------------------------------------------------------------------
    # Commands
    # ------------------------------------------------------------------

    async def async_set_switch(self, device_id: int, on: bool) -> None:
        await self._client.set_switch(device_id, on)

    async def async_control_movement(self, device_id: int, state: MovementState) -> None:
        await self._client.control_movement(device_id, state)

    # ------------------------------------------------------------------
    # State callback (called from BLE notification thread)
    # ------------------------------------------------------------------

    def _on_state(self, event_type: EventType, data: dict) -> None:
        changed = False

        if event_type == EventType.SWITCH:
            dev_id = data["device_id"]
            on     = data["on"]
            if self._switch_states.get(dev_id) != on:
                self._switch_states[dev_id] = on
                changed = True

        elif event_type == EventType.TANK:
            dev_id = data["device_id"]
            pct    = data["pct"]
            if self._tank_levels.get(dev_id) != pct:
                self._tank_levels[dev_id] = pct
                changed = True

        elif event_type == EventType.BATTERY:
            v = data["voltage"]
            if self._battery_v != v:
                self._battery_v = v
                changed = True

        if changed:
            self.hass.loop.call_soon_threadsafe(
                lambda: self.hass.async_create_task(self._async_notify())
            )

    async def _async_notify(self) -> None:
        # push-based: we call this manually instead of relying on update_interval
        self.async_set_updated_data(None)

    # ------------------------------------------------------------------
    # Keepalive
    # ------------------------------------------------------------------

    def _start_keepalive(self) -> None:
        self._cancel_tasks()
        self._keepalive_task = self.hass.async_create_task(self._keepalive_loop())

    async def _keepalive_loop(self) -> None:
        consecutive_failures = 0
        while True:
            await asyncio.sleep(KEEPALIVE_INTERVAL)
            if self._client.is_connected:
                try:
                    await self._client.get_devices()
                    consecutive_failures = 0
                except Exception as e:
                    consecutive_failures += 1
                    _LOGGER.warning(
                        "Keepalive failed (%d/%d): %s",
                        consecutive_failures, MAX_KEEPALIVE_FAILURES, e,
                    )
                    if consecutive_failures >= MAX_KEEPALIVE_FAILURES:
                        _LOGGER.error(
                            "Keepalive failed %d times — connection is dead, forcing reconnect",
                            consecutive_failures,
                        )
                        try:
                            await self._client.disconnect()
                        except Exception:
                            pass
                        self._on_disconnect()
                        return

    # ------------------------------------------------------------------
    # Reconnect
    # ------------------------------------------------------------------

    def _on_disconnect(self) -> None:
        _LOGGER.warning("OneControl disconnected — scheduling reconnect")
        self._cancel_tasks()
        self._reconnect_task = self.hass.async_create_task(self._reconnect_loop())

    async def _reconnect_loop(self) -> None:
        delays = RECONNECT_DELAYS
        for i, delay in enumerate(delays):
            _LOGGER.info("Reconnecting in %ds (attempt %d)...", delay, i + 1)
            await asyncio.sleep(delay)
            try:
                self._refresh_ble_device()
                await self._client.connect()
                _LOGGER.info("OneControl reconnected")
                self._start_keepalive()
                return
            except Exception as e:
                _LOGGER.warning("Reconnect attempt %d failed: %s", i + 1, e)

        # Keep retrying at max interval indefinitely — RV may be off for hours
        while True:
            await asyncio.sleep(delays[-1])
            try:
                self._refresh_ble_device()
                await self._client.connect()
                _LOGGER.info("OneControl reconnected")
                self._start_keepalive()
                return
            except Exception as e:
                _LOGGER.warning("Reconnect failed: %s", e)

    def _cancel_tasks(self) -> None:
        for task in (self._keepalive_task, self._reconnect_task):
            if task and not task.done():
                task.cancel()
        self._keepalive_task = None
        self._reconnect_task = None

    async def _async_update_data(self):
        return None