ha-xiaoxiang-bms/custom_components/xiaoxiang_bms/coordinator.py

"""DataUpdateCoordinator for the Xiaoxiang Smart BMS."""
from __future__ import annotations

import asyncio
import logging
from datetime import timedelta

from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .bluetooth_handler import BmsBluetoothHandler
from .const import CMD_CELL, CMD_GENERAL, CMD_VERSION, DOMAIN

_LOGGER = logging.getLogger(__name__)

# Only mark sensors unavailable after this many *consecutive* failed polls.
# Transient BLE misses (device not in cache, ESPHome proxy busy, etc.) return
# the last known data instead so the UI doesn't oscillate.
_FAILURES_BEFORE_UNAVAILABLE = 3


class BmsCoordinator(DataUpdateCoordinator[dict]):
    """Polls the BMS over BLE and distributes data to all sensor entities.

    Uses a connect → read → disconnect pattern on every poll so the BMS's
    single BLE connection slot is free between updates (mobile app access).
    """

    def __init__(
        self,
        hass: HomeAssistant,
        address: str,
        poll_interval: int,
        name: str = "Xiaoxiang Smart BMS",
    ) -> None:
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=poll_interval),
        )
        self.address = address
        self._device_name = name
        self._poll_timeout = max(poll_interval - 3, 10)
        self._handler = BmsBluetoothHandler(address)
        self.hw_version: str | None = None
        self._consecutive_failures = 0

    # ------------------------------------------------------------------
    # Device info — shared by sensor, binary_sensor, number platforms
    # ------------------------------------------------------------------

    @property
    def device_info(self) -> DeviceInfo:
        return DeviceInfo(
            identifiers={(DOMAIN, self.address)},
            name=self._device_name,
            manufacturer="Xiaoxiang",
            model=self.hw_version or "Smart BMS",
        )

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def async_setup(self) -> None:
        """No-op — no persistent connection to establish."""

    async def async_teardown(self) -> None:
        """No-op — each poll disconnects itself."""

    # ------------------------------------------------------------------
    # Poll
    # ------------------------------------------------------------------

    def _handle_failure(self, reason: str) -> dict:
        """On a transient failure, return cached data up to the threshold.

        Only raises UpdateFailed (→ sensors go unavailable) after
        _FAILURES_BEFORE_UNAVAILABLE consecutive misses.
        """
        self._consecutive_failures += 1
        if self._consecutive_failures <= _FAILURES_BEFORE_UNAVAILABLE and self.data:
            _LOGGER.debug(
                "BMS poll failed (%d/%d), keeping last known data: %s",
                self._consecutive_failures,
                _FAILURES_BEFORE_UNAVAILABLE,
                reason,
            )
            return self.data
        raise UpdateFailed(reason)

    async def _async_update_data(self) -> dict:
        """Connect to the BMS, fetch all data, disconnect."""

        # The BMS may not be in the scanner cache immediately after a disconnect.
        # Wait up to 2 s for an advertisement before giving up.
        device = async_ble_device_from_address(self.hass, self.address, connectable=True)
        if device is None:
            await asyncio.sleep(2.0)
            device = async_ble_device_from_address(
                self.hass, self.address, connectable=True
            )
        if device is None:
            return self._handle_failure(
                f"BMS ({self.address}) not reachable — check Bluetooth adapter / proxy"
            )

        commands = [CMD_GENERAL, CMD_CELL]
        if self.hw_version is None:
            commands.append(CMD_VERSION)

        try:
            responses = await asyncio.wait_for(
                self._handler.poll(device, commands),
                timeout=self._poll_timeout,
            )
        except asyncio.TimeoutError:
            return self._handle_failure(
                f"BMS poll timed out after {self._poll_timeout}s"
            )
        except Exception as exc:
            return self._handle_failure(f"BMS poll failed: {exc}")

        general_frame, cell_frame = responses[0], responses[1]

        if general_frame is None:
            return self._handle_failure("No response to general info request (0x03)")
        if cell_frame is None:
            return self._handle_failure("No response to cell info request (0x04)")

        # Successful poll — reset failure counter
        self._consecutive_failures = 0

        if self.hw_version is None and len(responses) > 2 and responses[2]:
            self.hw_version = BmsBluetoothHandler.parse_version(responses[2])
            _LOGGER.debug("BMS hardware version: %s", self.hw_version)

        data = BmsBluetoothHandler.parse_general_info(general_frame)
        data.update(BmsBluetoothHandler.parse_cell_info(cell_frame))

        data["power"] = round(data["voltage"] * data["current"], 2)
        data["energy_stored"] = round(data["voltage"] * data["residual_capacity"] / 1000, 3)

        if data["cell_voltages"]:
            v_max = max(data["cell_voltages"])
            v_min = min(data["cell_voltages"])
            data["cell_delta"] = round((v_max - v_min) * 1000, 1)
        else:
            data["cell_delta"] = None

        _LOGGER.debug(
            "BMS data: %.2fV  %.2fA  %d%%  %.2fAh  %.3fkWh  %d cells",
            data["voltage"], data["current"], data["state_of_charge"],
            data["residual_capacity"], data["energy_stored"],
            len(data["cell_voltages"]),
        )
        return data