coordinator.py

Go to the documentation of this file.

"""Provides the DataUpdateCoordinator."""
 
from __future__ import annotations
 
from datetime import timedelta
 
from automower_ble.mower import Mower
from bleak import BleakError
from bleak_retry_connector import close_stale_connections_by_address
 
from homeassistant.components import bluetooth
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DOMAIN, LOGGER
 
SCAN_INTERVAL = timedelta(seconds=60)
 
 
class HusqvarnaCoordinator(DataUpdateCoordinator[dict[str, bytes]]):
    """Class to manage fetching data."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        mower: Mower,
        address: str,
        channel_id: str,
        model: str,
    ) -> None:
        """Initialize global data updater."""
        super().__init__(
            hass=hass,
            logger=LOGGER,
            name=DOMAIN,
            update_interval=SCAN_INTERVAL,
        )
        self.addressaddress = address
        self.channel_idchannel_id = channel_id
        self.modelmodel = model
        self.mowermower = mower
 
    async def async_shutdown(self) -> None:
        """Shutdown coordinator and any connection."""
        LOGGER.debug("Shutdown")
        await super().async_shutdown()
        if self.mowermower.is_connected():
            await self.mowermower.disconnect()
 
    async def _async_find_device(self):
        LOGGER.debug("Trying to reconnect")
        await close_stale_connections_by_address(self.addressaddress)
 
        device = bluetooth.async_ble_device_from_address(
            self.hasshass, self.addressaddress, connectable=True
        )
 
        try:
            if not await self.mowermower.connect(device):
                raise UpdateFailed("Failed to connect")
        except BleakError as err:
            raise UpdateFailed("Failed to connect") from err
 
    async def _async_update_data(self) -> dict[str, bytes]:
        """Poll the device."""
        LOGGER.debug("Polling device")
 
        data: dict[str, bytes] = {}
 
        try:
            if not self.mowermower.is_connected():
                await self._async_find_device_async_find_device()
        except BleakError as err:
            raise UpdateFailed("Failed to connect") from err
 
        try:
            data["battery_level"] = await self.mowermower.battery_level()
            LOGGER.debug("battery_level" + str(data["battery_level"]))
            if data["battery_level"] is None:
                await self._async_find_device_async_find_device()
                raise UpdateFailed("Error getting data from device")
 
            data["activity"] = await self.mowermower.mower_activity()
            LOGGER.debug("activity:" + str(data["activity"]))
            if data["activity"] is None:
                await self._async_find_device_async_find_device()
                raise UpdateFailed("Error getting data from device")
 
            data["state"] = await self.mowermower.mower_state()
            LOGGER.debug("state:" + str(data["state"]))
            if data["state"] is None:
                await self._async_find_device_async_find_device()
                raise UpdateFailed("Error getting data from device")
 
        except BleakError as err:
            LOGGER.error("Error getting data from device")
            await self._async_find_device_async_find_device()
            raise UpdateFailed("Error getting data from device") from err
 
        return data