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