diff --git a/.coveragerc b/.coveragerc index 9357f6d9972..5eab1aab00e 100644 --- a/.coveragerc +++ b/.coveragerc @@ -944,6 +944,7 @@ omit = homeassistant/components/snmp/* homeassistant/components/sochain/sensor.py homeassistant/components/solaredge/__init__.py + homeassistant/components/solaredge/coordinator.py homeassistant/components/solaredge/sensor.py homeassistant/components/solaredge_local/sensor.py homeassistant/components/solarlog/* diff --git a/homeassistant/components/solaredge/coordinator.py b/homeassistant/components/solaredge/coordinator.py new file mode 100644 index 00000000000..b2fe27db808 --- /dev/null +++ b/homeassistant/components/solaredge/coordinator.py @@ -0,0 +1,280 @@ +"""Provides the data update coordinators for SolarEdge.""" +from __future__ import annotations + +from abc import abstractmethod +from datetime import date, datetime, timedelta + +from solaredge 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: + """Get and update the latest data.""" + + def __init__(self, hass: HomeAssistant, api: Solaredge, site_id: str) -> None: + """Initialize the data object.""" + self.api = api + self.site_id = site_id + + self.data = {} + self.attributes = {} + + self.hass = hass + self.coordinator = None + + @callback + def async_setup(self) -> None: + """Coordinator creation.""" + self.coordinator = DataUpdateCoordinator( + self.hass, + LOGGER, + name=str(self), + update_method=self.async_update_data, + update_interval=self.update_interval, + ) + + @property + @abstractmethod + def update_interval(self) -> timedelta: + """Update interval.""" + + @abstractmethod + def update(self) -> None: + """Update data in executor.""" + + async def async_update_data(self) -> None: + """Update data.""" + await self.hass.async_add_executor_job(self.update) + + +class SolarEdgeOverviewDataService(SolarEdgeDataService): + """Get and update the latest overview data.""" + + @property + def update_interval(self) -> timedelta: + """Update interval.""" + return OVERVIEW_UPDATE_DELAY + + def update(self) -> None: + """Update the data from the SolarEdge Monitoring API.""" + try: + data = self.api.get_overview(self.site_id) + overview = data["overview"] + except KeyError as ex: + raise UpdateFailed("Missing overview data, skipping update") from ex + + self.data = {} + + for key, value in overview.items(): + if key in ["lifeTimeData", "lastYearData", "lastMonthData", "lastDayData"]: + data = value["energy"] + elif key in ["currentPower"]: + data = value["power"] + else: + data = value + self.data[key] = data + + LOGGER.debug("Updated SolarEdge overview: %s", self.data) + + +class SolarEdgeDetailsDataService(SolarEdgeDataService): + """Get and update the latest details data.""" + + def __init__(self, hass: HomeAssistant, api: Solaredge, site_id: str) -> None: + """Initialize the details data service.""" + super().__init__(hass, api, site_id) + + self.data = None + + @property + def update_interval(self) -> timedelta: + """Update interval.""" + return DETAILS_UPDATE_DELAY + + def update(self) -> None: + """Update the data from the SolarEdge Monitoring API.""" + + try: + data = self.api.get_details(self.site_id) + details = data["details"] + except KeyError as ex: + raise UpdateFailed("Missing details data, skipping update") from ex + + self.data = None + self.attributes = {} + + for key, value in details.items(): + key = snakecase(key) + + if key in ["primary_module"]: + for module_key, module_value in value.items(): + self.attributes[snakecase(module_key)] = module_value + elif key in [ + "peak_power", + "type", + "name", + "last_update_time", + "installation_date", + ]: + self.attributes[key] = value + elif key == "status": + self.data = value + + LOGGER.debug("Updated SolarEdge details: %s, %s", self.data, self.attributes) + + +class SolarEdgeInventoryDataService(SolarEdgeDataService): + """Get and update the latest inventory data.""" + + @property + def update_interval(self) -> timedelta: + """Update interval.""" + return INVENTORY_UPDATE_DELAY + + def update(self) -> None: + """Update the data from the SolarEdge Monitoring API.""" + try: + data = self.api.get_inventory(self.site_id) + inventory = data["Inventory"] + except KeyError as ex: + raise UpdateFailed("Missing inventory data, skipping update") from ex + + self.data = {} + self.attributes = {} + + for key, value in inventory.items(): + self.data[key] = len(value) + self.attributes[key] = {key: value} + + LOGGER.debug("Updated SolarEdge inventory: %s, %s", self.data, self.attributes) + + +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.unit = None + + @property + def update_interval(self) -> timedelta: + """Update interval.""" + return ENERGY_DETAILS_DELAY + + def update(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 = self.api.get_energy_details( + self.site_id, + midnight, + now.strftime("%Y-%m-%d %H:%M:%S"), + meters=None, + 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.data = {} + self.attributes = {} + self.unit = 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.data[meter["type"]] = meter["values"][0]["value"] + self.attributes[meter["type"]] = {"date": meter["values"][0]["date"]} + + LOGGER.debug( + "Updated SolarEdge energy details: %s, %s", self.data, self.attributes + ) + + +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.unit = None + + @property + def update_interval(self) -> timedelta: + """Update interval.""" + return POWER_FLOW_UPDATE_DELAY + + def update(self) -> None: + """Update the data from the SolarEdge Monitoring API.""" + try: + data = self.api.get_current_power_flow(self.site_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.data = {} + self.attributes = {} + self.unit = power_flow["unit"] + + for key, value in power_flow.items(): + if key in ["LOAD", "PV", "GRID", "STORAGE"]: + self.data[key] = value["currentPower"] + self.attributes[key] = {"status": value["status"]} + + if key in ["GRID"]: + export = key.lower() in power_to + self.data[key] *= -1 if export else 1 + self.attributes[key]["flow"] = "export" if export else "import" + + if key in ["STORAGE"]: + charge = key.lower() in power_to + self.data[key] *= -1 if charge else 1 + self.attributes[key]["flow"] = "charge" if charge else "discharge" + self.attributes[key]["soc"] = value["chargeLevel"] + + LOGGER.debug("Updated SolarEdge power flow: %s, %s", self.data, self.attributes) diff --git a/homeassistant/components/solaredge/sensor.py b/homeassistant/components/solaredge/sensor.py index 75bb25f722d..5cd644f5006 100644 --- a/homeassistant/components/solaredge/sensor.py +++ b/homeassistant/components/solaredge/sensor.py @@ -1,35 +1,25 @@ """Support for SolarEdge Monitoring API.""" from __future__ import annotations -from abc import abstractmethod -from datetime import date, datetime, timedelta from typing import Any from solaredge import Solaredge -from stringcase import snakecase from homeassistant.components.sensor import SensorEntity from homeassistant.config_entries import ConfigEntry from homeassistant.const import DEVICE_CLASS_BATTERY, DEVICE_CLASS_POWER -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from homeassistant.helpers.update_coordinator import ( - CoordinatorEntity, - DataUpdateCoordinator, - UpdateFailed, -) +from homeassistant.helpers.update_coordinator import CoordinatorEntity -from .const import ( - CONF_SITE_ID, - DATA_API_CLIENT, - DETAILS_UPDATE_DELAY, - DOMAIN, - ENERGY_DETAILS_DELAY, - INVENTORY_UPDATE_DELAY, - LOGGER, - OVERVIEW_UPDATE_DELAY, - POWER_FLOW_UPDATE_DELAY, - SENSOR_TYPES, +from .const import CONF_SITE_ID, DATA_API_CLIENT, DOMAIN, SENSOR_TYPES +from .coordinator import ( + SolarEdgeDataService, + SolarEdgeDetailsDataService, + SolarEdgeEnergyDetailsService, + SolarEdgeInventoryDataService, + SolarEdgeOverviewDataService, + SolarEdgePowerFlowDataService, ) @@ -249,263 +239,3 @@ class SolarEdgeStorageLevelSensor(SolarEdgeSensor): if attr and "soc" in attr: return attr["soc"] return None - - -class SolarEdgeDataService: - """Get and update the latest data.""" - - def __init__(self, hass: HomeAssistant, api: Solaredge, site_id: str) -> None: - """Initialize the data object.""" - self.api = api - self.site_id = site_id - - self.data = {} - self.attributes = {} - - self.hass = hass - self.coordinator = None - - @callback - def async_setup(self) -> None: - """Coordinator creation.""" - self.coordinator = DataUpdateCoordinator( - self.hass, - LOGGER, - name=str(self), - update_method=self.async_update_data, - update_interval=self.update_interval, - ) - - @property - @abstractmethod - def update_interval(self) -> timedelta: - """Update interval.""" - - @abstractmethod - def update(self) -> None: - """Update data in executor.""" - - async def async_update_data(self) -> None: - """Update data.""" - await self.hass.async_add_executor_job(self.update) - - -class SolarEdgeOverviewDataService(SolarEdgeDataService): - """Get and update the latest overview data.""" - - @property - def update_interval(self) -> timedelta: - """Update interval.""" - return OVERVIEW_UPDATE_DELAY - - def update(self) -> None: - """Update the data from the SolarEdge Monitoring API.""" - try: - data = self.api.get_overview(self.site_id) - overview = data["overview"] - except KeyError as ex: - raise UpdateFailed("Missing overview data, skipping update") from ex - - self.data = {} - - for key, value in overview.items(): - if key in ["lifeTimeData", "lastYearData", "lastMonthData", "lastDayData"]: - data = value["energy"] - elif key in ["currentPower"]: - data = value["power"] - else: - data = value - self.data[key] = data - - LOGGER.debug("Updated SolarEdge overview: %s", self.data) - - -class SolarEdgeDetailsDataService(SolarEdgeDataService): - """Get and update the latest details data.""" - - def __init__(self, hass: HomeAssistant, api: Solaredge, site_id: str) -> None: - """Initialize the details data service.""" - super().__init__(hass, api, site_id) - - self.data = None - - @property - def update_interval(self) -> timedelta: - """Update interval.""" - return DETAILS_UPDATE_DELAY - - def update(self) -> None: - """Update the data from the SolarEdge Monitoring API.""" - - try: - data = self.api.get_details(self.site_id) - details = data["details"] - except KeyError as ex: - raise UpdateFailed("Missing details data, skipping update") from ex - - self.data = None - self.attributes = {} - - for key, value in details.items(): - key = snakecase(key) - - if key in ["primary_module"]: - for module_key, module_value in value.items(): - self.attributes[snakecase(module_key)] = module_value - elif key in [ - "peak_power", - "type", - "name", - "last_update_time", - "installation_date", - ]: - self.attributes[key] = value - elif key == "status": - self.data = value - - LOGGER.debug("Updated SolarEdge details: %s, %s", self.data, self.attributes) - - -class SolarEdgeInventoryDataService(SolarEdgeDataService): - """Get and update the latest inventory data.""" - - @property - def update_interval(self) -> timedelta: - """Update interval.""" - return INVENTORY_UPDATE_DELAY - - def update(self) -> None: - """Update the data from the SolarEdge Monitoring API.""" - try: - data = self.api.get_inventory(self.site_id) - inventory = data["Inventory"] - except KeyError as ex: - raise UpdateFailed("Missing inventory data, skipping update") from ex - - self.data = {} - self.attributes = {} - - for key, value in inventory.items(): - self.data[key] = len(value) - self.attributes[key] = {key: value} - - LOGGER.debug("Updated SolarEdge inventory: %s, %s", self.data, self.attributes) - - -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.unit = None - - @property - def update_interval(self) -> timedelta: - """Update interval.""" - return ENERGY_DETAILS_DELAY - - def update(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 = self.api.get_energy_details( - self.site_id, - midnight, - now.strftime("%Y-%m-%d %H:%M:%S"), - meters=None, - 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.data = {} - self.attributes = {} - self.unit = 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.data[meter["type"]] = meter["values"][0]["value"] - self.attributes[meter["type"]] = {"date": meter["values"][0]["date"]} - - LOGGER.debug( - "Updated SolarEdge energy details: %s, %s", self.data, self.attributes - ) - - -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.unit = None - - @property - def update_interval(self) -> timedelta: - """Update interval.""" - return POWER_FLOW_UPDATE_DELAY - - def update(self) -> None: - """Update the data from the SolarEdge Monitoring API.""" - try: - data = self.api.get_current_power_flow(self.site_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.data = {} - self.attributes = {} - self.unit = power_flow["unit"] - - for key, value in power_flow.items(): - if key in ["LOAD", "PV", "GRID", "STORAGE"]: - self.data[key] = value["currentPower"] - self.attributes[key] = {"status": value["status"]} - - if key in ["GRID"]: - export = key.lower() in power_to - self.data[key] *= -1 if export else 1 - self.attributes[key]["flow"] = "export" if export else "import" - - if key in ["STORAGE"]: - charge = key.lower() in power_to - self.data[key] *= -1 if charge else 1 - self.attributes[key]["flow"] = "charge" if charge else "discharge" - self.attributes[key]["soc"] = value["chargeLevel"] - - LOGGER.debug("Updated SolarEdge power flow: %s, %s", self.data, self.attributes)