core/homeassistant/components/solaredge/sensor.py

"""Support for SolarEdge Monitoring API."""
from abc import abstractmethod
from datetime import date, datetime
import logging

from requests.exceptions import ConnectTimeout, HTTPError
import solaredge
from stringcase import snakecase

from homeassistant.const import CONF_API_KEY, DEVICE_CLASS_BATTERY, DEVICE_CLASS_POWER
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import (
    CoordinatorEntity,
    DataUpdateCoordinator,
    UpdateFailed,
)

from .const import (
    CONF_SITE_ID,
    DETAILS_UPDATE_DELAY,
    ENERGY_DETAILS_DELAY,
    INVENTORY_UPDATE_DELAY,
    OVERVIEW_UPDATE_DELAY,
    POWER_FLOW_UPDATE_DELAY,
    SENSOR_TYPES,
)

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(hass, entry, async_add_entities):
    """Add an solarEdge entry."""
    # Add the needed sensors to hass
    api = solaredge.Solaredge(entry.data[CONF_API_KEY])

    # Check if api can be reached and site is active
    try:
        response = await hass.async_add_executor_job(
            api.get_details, entry.data[CONF_SITE_ID]
        )
        if response["details"]["status"].lower() != "active":
            _LOGGER.error("SolarEdge site is not active")
            return
        _LOGGER.debug("Credentials correct and site is active")
    except KeyError as ex:
        _LOGGER.error("Missing details data in SolarEdge response")
        raise ConfigEntryNotReady from ex
    except (ConnectTimeout, HTTPError) as ex:
        _LOGGER.error("Could not retrieve details from SolarEdge API")
        raise ConfigEntryNotReady from ex

    sensor_factory = SolarEdgeSensorFactory(
        hass, entry.title, entry.data[CONF_SITE_ID], api
    )
    for service in sensor_factory.all_services:
        service.async_setup()
        await service.coordinator.async_refresh()

    entities = []
    for sensor_key in SENSOR_TYPES:
        sensor = sensor_factory.create_sensor(sensor_key)
        if sensor is not None:
            entities.append(sensor)
    async_add_entities(entities)


class SolarEdgeSensorFactory:
    """Factory which creates sensors based on the sensor_key."""

    def __init__(self, hass, platform_name, site_id, api):
        """Initialize the factory."""
        self.platform_name = platform_name

        details = SolarEdgeDetailsDataService(hass, api, site_id)
        overview = SolarEdgeOverviewDataService(hass, api, site_id)
        inventory = SolarEdgeInventoryDataService(hass, api, site_id)
        flow = SolarEdgePowerFlowDataService(hass, api, site_id)
        energy = SolarEdgeEnergyDetailsService(hass, api, site_id)

        self.all_services = (details, overview, inventory, flow, energy)

        self.services = {"site_details": (SolarEdgeDetailsSensor, details)}

        for key in [
            "lifetime_energy",
            "energy_this_year",
            "energy_this_month",
            "energy_today",
            "current_power",
        ]:
            self.services[key] = (SolarEdgeOverviewSensor, overview)

        for key in ["meters", "sensors", "gateways", "batteries", "inverters"]:
            self.services[key] = (SolarEdgeInventorySensor, inventory)

        for key in ["power_consumption", "solar_power", "grid_power", "storage_power"]:
            self.services[key] = (SolarEdgePowerFlowSensor, flow)

        for key in ["storage_level"]:
            self.services[key] = (SolarEdgeStorageLevelSensor, flow)

        for key in [
            "purchased_power",
            "production_power",
            "feedin_power",
            "consumption_power",
            "selfconsumption_power",
        ]:
            self.services[key] = (SolarEdgeEnergyDetailsSensor, energy)

    def create_sensor(self, sensor_key):
        """Create and return a sensor based on the sensor_key."""
        sensor_class, service = self.services[sensor_key]

        return sensor_class(self.platform_name, sensor_key, service)


class SolarEdgeSensor(CoordinatorEntity, Entity):
    """Abstract class for a solaredge sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the sensor."""
        super().__init__(data_service.coordinator)
        self.platform_name = platform_name
        self.sensor_key = sensor_key
        self.data_service = data_service

    @property
    def unit_of_measurement(self):
        """Return the unit of measurement."""
        return SENSOR_TYPES[self.sensor_key][2]

    @property
    def name(self):
        """Return the name."""
        return "{} ({})".format(self.platform_name, SENSOR_TYPES[self.sensor_key][1])

    @property
    def icon(self):
        """Return the sensor icon."""
        return SENSOR_TYPES[self.sensor_key][3]


class SolarEdgeOverviewSensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API overview sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the overview sensor."""
        super().__init__(platform_name, sensor_key, data_service)

        self._json_key = SENSOR_TYPES[self.sensor_key][0]

    @property
    def state(self):
        """Return the state of the sensor."""
        return self.data_service.data.get(self._json_key)


class SolarEdgeDetailsSensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API details sensor."""

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        return self.data_service.attributes

    @property
    def state(self):
        """Return the state of the sensor."""
        return self.data_service.data


class SolarEdgeInventorySensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API inventory sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the inventory sensor."""
        super().__init__(platform_name, sensor_key, data_service)

        self._json_key = SENSOR_TYPES[self.sensor_key][0]

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        return self.data_service.attributes.get(self._json_key)

    @property
    def state(self):
        """Return the state of the sensor."""
        return self.data_service.data.get(self._json_key)


class SolarEdgeEnergyDetailsSensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API power flow sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the power flow sensor."""
        super().__init__(platform_name, sensor_key, data_service)

        self._json_key = SENSOR_TYPES[self.sensor_key][0]

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        return self.data_service.attributes.get(self._json_key)

    @property
    def state(self):
        """Return the state of the sensor."""
        return self.data_service.data.get(self._json_key)

    @property
    def unit_of_measurement(self):
        """Return the unit of measurement."""
        return self.data_service.unit


class SolarEdgePowerFlowSensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API power flow sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the power flow sensor."""
        super().__init__(platform_name, sensor_key, data_service)

        self._json_key = SENSOR_TYPES[self.sensor_key][0]

    @property
    def device_class(self):
        """Device Class."""
        return DEVICE_CLASS_POWER

    @property
    def device_state_attributes(self):
        """Return the state attributes."""
        return self.data_service.attributes.get(self._json_key)

    @property
    def state(self):
        """Return the state of the sensor."""
        return self.data_service.data.get(self._json_key)

    @property
    def unit_of_measurement(self):
        """Return the unit of measurement."""
        return self.data_service.unit


class SolarEdgeStorageLevelSensor(SolarEdgeSensor):
    """Representation of an SolarEdge Monitoring API storage level sensor."""

    def __init__(self, platform_name, sensor_key, data_service):
        """Initialize the storage level sensor."""
        super().__init__(platform_name, sensor_key, data_service)

        self._json_key = SENSOR_TYPES[self.sensor_key][0]

    @property
    def device_class(self):
        """Return the device_class of the device."""
        return DEVICE_CLASS_BATTERY

    @property
    def state(self):
        """Return the state of the sensor."""
        attr = self.data_service.attributes.get(self._json_key)
        if attr and "soc" in attr:
            return attr["soc"]
        return None


class SolarEdgeDataService:
    """Get and update the latest data."""

    def __init__(self, hass, api, site_id):
        """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):
        """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):
        """Update interval."""

    @abstractmethod
    def update(self):
        """Update data in executor."""

    async def async_update_data(self):
        """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):
        """Update interval."""
        return OVERVIEW_UPDATE_DELAY

    def update(self):
        """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, api, site_id):
        """Initialize the details data service."""
        super().__init__(hass, api, site_id)

        self.data = None

    @property
    def update_interval(self):
        """Update interval."""
        return DETAILS_UPDATE_DELAY

    def update(self):
        """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):
        """Update interval."""
        return INVENTORY_UPDATE_DELAY

    def update(self):
        """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, api, site_id):
        """Initialize the power flow data service."""
        super().__init__(hass, api, site_id)

        self.unit = None

    @property
    def update_interval(self):
        """Update interval."""
        return ENERGY_DETAILS_DELAY

    def update(self):
        """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, api, site_id):
        """Initialize the power flow data service."""
        super().__init__(hass, api, site_id)

        self.unit = None

    @property
    def update_interval(self):
        """Update interval."""
        return POWER_FLOW_UPDATE_DELAY

    def update(self):
        """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
        )