coordinator.py

Go to the documentation of this file.

"""DataUpdateCoordinators for the Fronius integration."""
 
from __future__ import annotations
 
from abc import ABC, abstractmethod
from datetime import timedelta
from typing import TYPE_CHECKING, Any
 
from pyfronius import BadStatusError, FroniusError
 
from homeassistant.core import callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import (
    SOLAR_NET_ID_POWER_FLOW,
    SOLAR_NET_ID_SYSTEM,
    FroniusDeviceInfo,
    SolarNetId,
)
from .sensor import (
    INVERTER_ENTITY_DESCRIPTIONS,
    LOGGER_ENTITY_DESCRIPTIONS,
    METER_ENTITY_DESCRIPTIONS,
    OHMPILOT_ENTITY_DESCRIPTIONS,
    POWER_FLOW_ENTITY_DESCRIPTIONS,
    STORAGE_ENTITY_DESCRIPTIONS,
    FroniusSensorEntityDescription,
)
 
if TYPE_CHECKING:
    from . import FroniusSolarNet
    from .sensor import _FroniusSensorEntity
 
 
class FroniusCoordinatorBase(
    ABC, DataUpdateCoordinator[dict[SolarNetId, dict[str, Any]]]
):
    """Query Fronius endpoint and keep track of seen conditions."""
 
    default_interval: timedelta
    error_interval: timedelta
    valid_descriptions: list[FroniusSensorEntityDescription]
 
    MAX_FAILED_UPDATES = 3
 
    def __init__(self, *args: Any, solar_net: FroniusSolarNet, **kwargs: Any) -> None:
        """Set up the FroniusCoordinatorBase class."""
        self._failed_update_count_failed_update_count = 0
        self.solar_netsolar_net = solar_net
        # unregistered_descriptors are used to create entities in platform module
        self.unregistered_descriptors: dict[
            SolarNetId, list[FroniusSensorEntityDescription]
        ] = {}
        super().__init__(*args, update_interval=self.default_interval, **kwargs)
 
    @abstractmethod
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
 
    async def _async_update_data(self) -> dict[SolarNetId, Any]:
        """Fetch the latest data from the source."""
        async with self.solar_net.coordinator_lock:
            try:
                data = await self._update_method()
            except FroniusError as err:
                self._failed_update_count += 1
                if self._failed_update_count == self.MAX_FAILED_UPDATES:
                    self.update_intervalupdate_intervalupdate_intervalupdate_intervalupdate_interval = self.error_interval
                raise UpdateFailed(err) from err
 
            if self._failed_update_count_failed_update_count != 0:
                self._failed_update_count_failed_update_count = 0
                self.update_intervalupdate_intervalupdate_intervalupdate_intervalupdate_interval = self.default_interval
 
            for solar_net_id in data:
                if solar_net_id not in self.unregistered_descriptors:
                    # id seen for the first time
                    self.unregistered_descriptors[solar_net_id] = (
                        self.valid_descriptions.copy()
                    )
            return data
 
    @callback
    def add_entities_for_seen_keys[_FroniusEntityT: _FroniusSensorEntity](
        self,
        async_add_entities: AddEntitiesCallback,
        entity_constructor: type[_FroniusEntityT],
    ) -> None:
        """Add entities for received keys and registers listener for future seen keys.
 
        Called from a platforms `async_setup_entry`.
        """
 
        @callback
        def _add_entities_for_unregistered_descriptors() -> None:
            """Add entities for keys seen for the first time."""
            new_entities: list[_FroniusEntityT] = []
            for solar_net_id, device_data in self.datadata.items():
                remaining_unregistered_descriptors = []
                for description in self.unregistered_descriptors[solar_net_id]:
                    key = description.response_key or description.key
                    if key not in device_data:
                        remaining_unregistered_descriptors.append(description)
                        continue
                    if device_data[key]["value"] is None:
                        remaining_unregistered_descriptors.append(description)
                        continue
                    new_entities.append(
                        entity_constructor(
                            coordinator=self,
                            description=description,
                            solar_net_id=solar_net_id,
                        )
                    )
                self.unregistered_descriptors[solar_net_id] = (
                    remaining_unregistered_descriptors
                )
            async_add_entities(new_entities)
 
        _add_entities_for_unregistered_descriptors()
        self.solar_netsolar_net.config_entry.async_on_unload(
            self.async_add_listenerasync_add_listenerasync_add_listener(_add_entities_for_unregistered_descriptors)
        )
 
 
class FroniusInverterUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius device inverter endpoint and keep track of seen conditions."""
 
    default_interval = timedelta(minutes=1)
    error_interval = timedelta(minutes=10)
    valid_descriptions = INVERTER_ENTITY_DESCRIPTIONS
 
    SILENT_RETRIES = 3
 
    def __init__(
        self, *args: Any, inverter_info: FroniusDeviceInfo, **kwargs: Any
    ) -> None:
        """Set up a Fronius inverter device scope coordinator."""
        super().__init__(*args, **kwargs)
        self.inverter_infoinverter_info = inverter_info
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        # almost 1% of `current_inverter_data` requests on Symo devices result in
        # `BadStatusError Code: 8 - LNRequestTimeout` due to flaky internal
        # communication between the logger and the inverter.
        for silent_retry in range(self.SILENT_RETRIESSILENT_RETRIES):
            try:
                data = await self.solar_netsolar_net.fronius.current_inverter_data(
                    self.inverter_infoinverter_info.solar_net_id
                )
            except BadStatusError:
                if silent_retry == (self.SILENT_RETRIESSILENT_RETRIES - 1):
                    raise
                continue
            break
        # wrap a single devices data in a dict with solar_net_id key for
        # FroniusCoordinatorBase _async_update_data and add_entities_for_seen_keys
        return {self.inverter_infoinverter_info.solar_net_id: data}
 
 
class FroniusLoggerUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius logger info endpoint and keep track of seen conditions."""
 
    default_interval = timedelta(hours=1)
    error_interval = timedelta(hours=1)
    valid_descriptions = LOGGER_ENTITY_DESCRIPTIONS
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        data = await self.solar_netsolar_net.fronius.current_logger_info()
        return {SOLAR_NET_ID_SYSTEM: data}
 
 
class FroniusMeterUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius system meter endpoint and keep track of seen conditions."""
 
    default_interval = timedelta(minutes=1)
    error_interval = timedelta(minutes=10)
    valid_descriptions = METER_ENTITY_DESCRIPTIONS
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        data = await self.solar_netsolar_net.fronius.current_system_meter_data()
        return data["meters"]  # type: ignore[no-any-return]
 
 
class FroniusOhmpilotUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius Ohmpilots and keep track of seen conditions."""
 
    default_interval = timedelta(minutes=1)
    error_interval = timedelta(minutes=10)
    valid_descriptions = OHMPILOT_ENTITY_DESCRIPTIONS
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        data = await self.solar_netsolar_net.fronius.current_system_ohmpilot_data()
        return data["ohmpilots"]  # type: ignore[no-any-return]
 
 
class FroniusPowerFlowUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius power flow endpoint and keep track of seen conditions."""
 
    default_interval = timedelta(seconds=10)
    error_interval = timedelta(minutes=3)
    valid_descriptions = POWER_FLOW_ENTITY_DESCRIPTIONS
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        data = await self.solar_netsolar_net.fronius.current_power_flow()
        return {SOLAR_NET_ID_POWER_FLOW: data}
 
 
class FroniusStorageUpdateCoordinator(FroniusCoordinatorBase):
    """Query Fronius system storage endpoint and keep track of seen conditions."""
 
    default_interval = timedelta(minutes=1)
    error_interval = timedelta(minutes=10)
    valid_descriptions = STORAGE_ENTITY_DESCRIPTIONS
 
    async def _update_method(self) -> dict[SolarNetId, Any]:
        """Return data per solar net id from pyfronius."""
        data = await self.solar_netsolar_net.fronius.current_system_storage_data()
        return data["storages"]  # type: ignore[no-any-return]