coordinator.py

Go to the documentation of this file.

"""Coordinator for BMW."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
from bimmer_connected.account import MyBMWAccount
from bimmer_connected.api.regions import get_region_from_name
from bimmer_connected.models import (
    GPSPosition,
    MyBMWAPIError,
    MyBMWAuthError,
    MyBMWCaptchaMissingError,
)
from httpx import RequestError
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.ssl import get_default_context
 
from .const import CONF_GCID, CONF_READ_ONLY, CONF_REFRESH_TOKEN, DOMAIN, SCAN_INTERVALS
 
_LOGGER = logging.getLogger(__name__)
 
 
class BMWDataUpdateCoordinator(DataUpdateCoordinator[None]):
    """Class to manage fetching BMW data."""
 
    account: MyBMWAccount
 
    def __init__(self, hass: HomeAssistant, *, entry: ConfigEntry) -> None:
        """Initialize account-wide BMW data updater."""
        self.accountaccount = MyBMWAccount(
            entry.data[CONF_USERNAME],
            entry.data[CONF_PASSWORD],
            get_region_from_name(entry.data[CONF_REGION]),
            observer_position=GPSPosition(hass.config.latitude, hass.config.longitude),
            verify=get_default_context(),
        )
        self.read_onlyread_only = entry.options[CONF_READ_ONLY]
        self._entry_entry = entry
 
        if CONF_REFRESH_TOKEN in entry.data:
            self.accountaccount.set_refresh_token(
                refresh_token=entry.data[CONF_REFRESH_TOKEN],
                gcid=entry.data.get(CONF_GCID),
            )
 
        super().__init__(
            hass,
            _LOGGER,
            name=f"{DOMAIN}-{entry.data['username']}",
            update_interval=timedelta(seconds=SCAN_INTERVALS[entry.data[CONF_REGION]]),
        )
 
        # Default to false on init so _async_update_data logic works
        self.last_update_successlast_update_successlast_update_success = False
 
    async def _async_update_data(self) -> None:
        """Fetch data from BMW."""
        old_refresh_token = self.accountaccount.refresh_token
 
        try:
            await self.accountaccount.get_vehicles()
        except MyBMWCaptchaMissingError as err:
            # If a captcha is required (user/password login flow), always trigger the reauth flow
            raise ConfigEntryAuthFailed(
                translation_domain=DOMAIN,
                translation_key="missing_captcha",
            ) from err
        except MyBMWAuthError as err:
            # Allow one retry interval before raising AuthFailed to avoid flaky API issues
            if self.last_update_successlast_update_successlast_update_success:
                raise UpdateFailed(err) from err
            # Clear refresh token and trigger reauth if previous update failed as well
            self._update_config_entry_refresh_token_update_config_entry_refresh_token(None)
            raise ConfigEntryAuthFailed(err) from err
        except (MyBMWAPIError, RequestError) as err:
            raise UpdateFailed(err) from err
 
        if self.accountaccount.refresh_token != old_refresh_token:
            self._update_config_entry_refresh_token_update_config_entry_refresh_token(self.accountaccount.refresh_token)
 
    def _update_config_entry_refresh_token(self, refresh_token: str | None) -> None:
        """Update or delete the refresh_token in the Config Entry."""
        data = {
            **self._entry_entry.data,
            CONF_REFRESH_TOKEN: refresh_token,
        }
        if not refresh_token:
            data.pop(CONF_REFRESH_TOKEN)
        self.hasshass.config_entries.async_update_entry(self._entry_entry, data=data)