coordinator.py

Go to the documentation of this file.

"""The enphase_envoy component."""
 
from __future__ import annotations
 
import contextlib
import datetime
from datetime import timedelta
import logging
from typing import Any
 
from pyenphase import Envoy, EnvoyError, EnvoyTokenAuth
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
import homeassistant.util.dt as dt_util
 
from .const import INVALID_AUTH_ERRORS
 
SCAN_INTERVAL = timedelta(seconds=60)
 
TOKEN_REFRESH_CHECK_INTERVAL = timedelta(days=1)
STALE_TOKEN_THRESHOLD = timedelta(days=30).total_seconds()
NOTIFICATION_ID = "enphase_envoy_notification"
 
_LOGGER = logging.getLogger(__name__)
 
 
type EnphaseConfigEntry = ConfigEntry[EnphaseUpdateCoordinator]
 
 
class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """DataUpdateCoordinator to gather data from any envoy."""
 
    envoy_serial_number: str
    envoy_firmware: str
 
    def __init__(
        self, hass: HomeAssistant, envoy: Envoy, entry: EnphaseConfigEntry
    ) -> None:
        """Initialize DataUpdateCoordinator for the envoy."""
        self.envoyenvoy = envoy
        entry_data = entry.data
        self.entryentry = entry
        self.usernameusername = entry_data[CONF_USERNAME]
        self.passwordpassword = entry_data[CONF_PASSWORD]
        self._setup_complete_setup_complete = False
        self.envoy_firmwareenvoy_firmware = ""
        self._cancel_token_refresh_cancel_token_refresh: CALLBACK_TYPE | None = None
        super().__init__(
            hass,
            _LOGGER,
            name=entry_data[CONF_NAME],
            update_interval=SCAN_INTERVAL,
            always_update=False,
        )
 
    @callback
    def _async_refresh_token_if_needed(self, now: datetime.datetime) -> None:
        """Proactively refresh token if its stale in case cloud services goes down."""
        assert isinstance(self.envoyenvoy.auth, EnvoyTokenAuth)
        expire_time = self.envoyenvoy.auth.expire_timestamp
        remain = expire_time - now.timestamp()
        fresh = remain > STALE_TOKEN_THRESHOLD
        name = self.namename
        _LOGGER.debug("%s: %s seconds remaining on token fresh=%s", name, remain, fresh)
        if not fresh:
            self.hasshass.async_create_background_task(
                self._async_try_refresh_token_async_try_refresh_token(), "{name} token refresh"
            )
 
    async def _async_try_refresh_token(self) -> None:
        """Try to refresh token."""
        assert isinstance(self.envoyenvoy.auth, EnvoyTokenAuth)
        _LOGGER.debug("%s: Trying to refresh token", self.namename)
        try:
            await self.envoyenvoy.auth.refresh()
        except EnvoyError as err:
            # If we can't refresh the token, we try again later
            # If the token actually ends up expiring, we'll
            # re-authenticate with username/password and get a new token
            # or log an error if that fails
            _LOGGER.debug("%s: Error refreshing token: %s", err, self.namename)
            return
        self._async_update_saved_token_async_update_saved_token()
 
    @callback
    def _async_mark_setup_complete(self) -> None:
        """Mark setup as complete and setup token refresh if needed."""
        self._setup_complete_setup_complete = True
        self.async_cancel_token_refreshasync_cancel_token_refresh()
        if not isinstance(self.envoyenvoy.auth, EnvoyTokenAuth):
            return
        self._cancel_token_refresh_cancel_token_refresh = async_track_time_interval(
            self.hasshass,
            self._async_refresh_token_if_needed_async_refresh_token_if_needed,
            TOKEN_REFRESH_CHECK_INTERVAL,
            cancel_on_shutdown=True,
        )
 
    async def _async_setup_and_authenticate(self) -> None:
        """Set up and authenticate with the envoy."""
        envoy = self.envoyenvoy
        await envoy.setup()
        assert envoy.serial_number is not None
        self.envoy_serial_numberenvoy_serial_number = envoy.serial_number
        if token := self.entryentry.data.get(CONF_TOKEN):
            with contextlib.suppress(*INVALID_AUTH_ERRORS):
                # Always set the username and password
                # so we can refresh the token if needed
                await envoy.authenticate(
                    username=self.usernameusername, password=self.passwordpassword, token=token
                )
                # The token is valid, but we still want
                # to refresh it if it's stale right away
                self._async_refresh_token_if_needed_async_refresh_token_if_needed(dt_util.utcnow())
                return
            # token likely expired or firmware changed
            # so we fall through to authenticate with
            # username/password
        await self.envoyenvoy.authenticate(username=self.usernameusername, password=self.passwordpassword)
        # Password auth succeeded, so we can update the token
        # if we are using EnvoyTokenAuth
        self._async_update_saved_token_async_update_saved_token()
 
    def _async_update_saved_token(self) -> None:
        """Update saved token in config entry."""
        envoy = self.envoyenvoy
        if not isinstance(envoy.auth, EnvoyTokenAuth):
            return
        # update token in config entry so we can
        # startup without hitting the Cloud API
        # as long as the token is valid
        _LOGGER.debug("%s: Updating token in config entry from auth", self.namename)
        self.hasshass.config_entries.async_update_entry(
            self.entryentry,
            data={
                **self.entryentry.data,
                CONF_TOKEN: envoy.auth.token,
            },
        )
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Fetch all device and sensor data from api."""
        envoy = self.envoyenvoy
        for tries in range(2):
            try:
                if not self._setup_complete_setup_complete:
                    await self._async_setup_and_authenticate_async_setup_and_authenticate()
                    self._async_mark_setup_complete_async_mark_setup_complete()
                # dump all received data in debug mode to assist troubleshooting
                envoy_data = await envoy.update()
            except INVALID_AUTH_ERRORS as err:
                if self._setup_complete_setup_complete and tries == 0:
                    # token likely expired or firmware changed, try to re-authenticate
                    self._setup_complete_setup_complete = False
                    continue
                raise ConfigEntryAuthFailed from err
            except EnvoyError as err:
                raise UpdateFailed(f"Error communicating with API: {err}") from err
 
            # if we have a firmware version from previous setup, compare to current one
            # when envoy gets new firmware there will be an authentication failure
            # which results in getting fw version again, if so reload the integration.
            if (current_firmware := self.envoy_firmwareenvoy_firmware) and current_firmware != (
                new_firmware := envoy.firmware
            ):
                _LOGGER.warning(
                    "Envoy firmware changed from: %s to: %s, reloading enphase envoy integration",
                    current_firmware,
                    new_firmware,
                )
                # reload the integration to get all established again
                self.hasshass.async_create_task(
                    self.hasshass.config_entries.async_reload(self.entryentry.entry_id)
                )
            # remember firmware version for next time
            self.envoy_firmwareenvoy_firmware = envoy.firmware
            _LOGGER.debug("Envoy data: %s", envoy_data)
            return envoy_data.raw
 
        raise RuntimeError("Unreachable code in _async_update_data")  # pragma: no cover
 
    @callback
    def async_cancel_token_refresh(self) -> None:
        """Cancel token refresh."""
        if self._cancel_token_refresh_cancel_token_refresh:
            self._cancel_token_refresh_cancel_token_refresh()
            self._cancel_token_refresh_cancel_token_refresh = None