coordinator.py

Go to the documentation of this file.

"""Provides the data update coordinators for SolarEdge."""
 
from __future__ import annotations
 
from abc import ABC, abstractmethod
from datetime import date, datetime, timedelta
from typing import Any
 
from aiosolaredge import SolarEdge
from stringcase import snakecase
 
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import (
    DETAILS_UPDATE_DELAY,
    ENERGY_DETAILS_DELAY,
    INVENTORY_UPDATE_DELAY,
    LOGGER,
    OVERVIEW_UPDATE_DELAY,
    POWER_FLOW_UPDATE_DELAY,
)
 
 
class SolarEdgeDataService(ABC):
    """Get and update the latest data."""
 
    coordinator: DataUpdateCoordinator[None]
 
    def __init__(self, hass: HomeAssistant, api: SolarEdge, site_id: str) -> None:
        """Initialize the data object."""
        self.apiapi = api
        self.site_idsite_id = site_id
 
        self.data: dict[str, Any] = {}
        self.attributes: dict[str, Any] = {}
 
        self.hasshass = hass
 
    @callback
    def async_setup(self) -> None:
        """Coordinator creation."""
        self.coordinatorcoordinator = DataUpdateCoordinator(
            self.hasshass,
            LOGGER,
            name=str(self),
            update_method=self.async_update_dataasync_update_data,
            update_interval=self.update_intervalupdate_interval,
        )
 
    @property
    @abstractmethod
    def update_interval(self) -> timedelta:
        """Update interval."""
 
    @abstractmethod
    async def async_update_data(self) -> None:
        """Update data."""
 
 
class SolarEdgeOverviewDataService(SolarEdgeDataService):
    """Get and update the latest overview data."""
 
    @property
    def update_interval(self) -> timedelta:
        """Update interval."""
        return OVERVIEW_UPDATE_DELAY
 
    async def async_update_data(self) -> None:
        """Update the data from the SolarEdge Monitoring API."""
        try:
            data = await self.apiapi.get_overview(self.site_idsite_id)
            overview = data["overview"]
        except KeyError as ex:
            raise UpdateFailed("Missing overview data, skipping update") from ex
 
        self.datadata = {}
 
        energy_keys = ["lifeTimeData", "lastYearData", "lastMonthData", "lastDayData"]
        for key, value in overview.items():
            if key in energy_keys:
                data = value["energy"]
            elif key in ["currentPower"]:
                data = value["power"]
            else:
                data = value
            self.datadata[key] = data
 
        # Sanity check the energy values. SolarEdge API sometimes report "lifetimedata" of zero,
        # while values for last Year, Month and Day energy are still OK.
        # See https://github.com/home-assistant/core/issues/59285 .
        if set(energy_keys).issubset(self.datadata.keys()):
            for index, key in enumerate(energy_keys, start=1):
                # All coming values in list should be larger than the current value.
                if any(self.datadata[k] > self.datadata[key] for k in energy_keys[index:]):
                    LOGGER.warning(
                        "Ignoring invalid energy value %s for %s", self.datadata[key], key
                    )
                    self.datadata.pop(key)
 
        LOGGER.debug("Updated SolarEdge overview: %s", self.datadata)
 
 
class SolarEdgeDetailsDataService(SolarEdgeDataService):
    """Get and update the latest details data."""
 
    @property
    def update_interval(self) -> timedelta:
        """Update interval."""
        return DETAILS_UPDATE_DELAY
 
    async def async_update_data(self) -> None:
        """Update the data from the SolarEdge Monitoring API."""
 
        try:
            data = await self.apiapi.get_details(self.site_idsite_id)
            details = data["details"]
        except KeyError as ex:
            raise UpdateFailed("Missing details data, skipping update") from ex
 
        self.datadata = {}
        self.attributesattributes = {}
 
        for key, value in details.items():
            key = snakecase(key)
 
            if key in ["primary_module"]:
                for module_key, module_value in value.items():
                    self.attributesattributes[snakecase(module_key)] = module_value
            elif key in [
                "peak_power",
                "type",
                "name",
                "last_update_time",
                "installation_date",
            ]:
                self.attributesattributes[key] = value
            elif key == "status":
                self.datadata["status"] = value
 
        LOGGER.debug(
            "Updated SolarEdge details: %s, %s",
            self.datadata.get("status"),
            self.attributesattributes,
        )
 
 
class SolarEdgeInventoryDataService(SolarEdgeDataService):
    """Get and update the latest inventory data."""
 
    @property
    def update_interval(self) -> timedelta:
        """Update interval."""
        return INVENTORY_UPDATE_DELAY
 
    async def async_update_data(self) -> None:
        """Update the data from the SolarEdge Monitoring API."""
        try:
            data = await self.apiapi.get_inventory(self.site_idsite_id)
            inventory = data["Inventory"]
        except KeyError as ex:
            raise UpdateFailed("Missing inventory data, skipping update") from ex
 
        self.datadata = {}
        self.attributesattributes = {}
 
        for key, value in inventory.items():
            self.datadata[key] = len(value)
            self.attributesattributes[key] = {key: value}
 
        LOGGER.debug("Updated SolarEdge inventory: %s, %s", self.datadata, self.attributesattributes)
 
 
class SolarEdgeEnergyDetailsService(SolarEdgeDataService):
    """Get and update the latest power flow data."""
 
    def __init__(self, hass: HomeAssistant, api: SolarEdge, site_id: str) -> None:
        """Initialize the power flow data service."""
        super().__init__(hass, api, site_id)
 
        self.unitunit = None
 
    @property
    def update_interval(self) -> timedelta:
        """Update interval."""
        return ENERGY_DETAILS_DELAY
 
    async def async_update_data(self) -> None:
        """Update the data from the SolarEdge Monitoring API."""
        try:
            now = datetime.now()
            today = date.today()
            midnight = datetime.combine(today, datetime.min.time())
            data = await self.apiapi.get_energy_details(
                self.site_idsite_id,
                midnight,
                now,
                time_unit="DAY",
            )
            energy_details = data["energyDetails"]
        except KeyError as ex:
            raise UpdateFailed("Missing power flow data, skipping update") from ex
 
        if "meters" not in energy_details:
            LOGGER.debug(
                "Missing meters in energy details data. Assuming site does not have any"
            )
            return
 
        self.datadata = {}
        self.attributesattributes = {}
        self.unitunit = energy_details["unit"]
 
        for meter in energy_details["meters"]:
            if "type" not in meter or "values" not in meter:
                continue
            if meter["type"] not in [
                "Production",
                "SelfConsumption",
                "FeedIn",
                "Purchased",
                "Consumption",
            ]:
                continue
            if len(meter["values"][0]) == 2:
                self.datadata[meter["type"]] = meter["values"][0]["value"]
                self.attributesattributes[meter["type"]] = {"date": meter["values"][0]["date"]}
 
        LOGGER.debug(
            "Updated SolarEdge energy details: %s, %s", self.datadata, self.attributesattributes
        )
 
 
class SolarEdgePowerFlowDataService(SolarEdgeDataService):
    """Get and update the latest power flow data."""
 
    def __init__(self, hass: HomeAssistant, api: SolarEdge, site_id: str) -> None:
        """Initialize the power flow data service."""
        super().__init__(hass, api, site_id)
 
        self.unitunit = None
 
    @property
    def update_interval(self) -> timedelta:
        """Update interval."""
        return POWER_FLOW_UPDATE_DELAY
 
    async def async_update_data(self) -> None:
        """Update the data from the SolarEdge Monitoring API."""
        try:
            data = await self.apiapi.get_current_power_flow(self.site_idsite_id)
            power_flow = data["siteCurrentPowerFlow"]
        except KeyError as ex:
            raise UpdateFailed("Missing power flow data, skipping update") from ex
 
        power_from = []
        power_to = []
 
        if "connections" not in power_flow:
            LOGGER.debug(
                "Missing connections in power flow data. Assuming site does not"
                " have any"
            )
            return
 
        for connection in power_flow["connections"]:
            power_from.append(connection["from"].lower())
            power_to.append(connection["to"].lower())
 
        self.datadata = {}
        self.attributesattributes = {}
        self.unitunit = power_flow["unit"]
 
        for key, value in power_flow.items():
            if key in ["LOAD", "PV", "GRID", "STORAGE"]:
                self.datadata[key] = value.get("currentPower")
                self.attributesattributes[key] = {"status": value["status"]}
 
            if key in ["GRID"]:
                export = key.lower() in power_to
                if self.datadata[key]:
                    self.datadata[key] *= -1 if export else 1
                self.attributesattributes[key]["flow"] = "export" if export else "import"
 
            if key in ["STORAGE"]:
                charge = key.lower() in power_to
                if self.datadata[key]:
                    self.datadata[key] *= -1 if charge else 1
                self.attributesattributes[key]["flow"] = "charge" if charge else "discharge"
                self.attributesattributes[key]["soc"] = value["chargeLevel"]
 
        LOGGER.debug("Updated SolarEdge power flow: %s, %s", self.datadata, self.attributesattributes)