sensor.py

Go to the documentation of this file.

"""Sensor platform for Ista EcoTrend integration."""
 
from __future__ import annotations
 
import asyncio
from dataclasses import dataclass
import datetime
from enum import StrEnum
import logging
 
from homeassistant.components.recorder.models.statistics import (
    StatisticData,
    StatisticMetaData,
)
from homeassistant.components.recorder.statistics import (
    async_add_external_statistics,
    get_instance,
    get_last_statistics,
)
from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorEntityDescription,
    SensorStateClass,
)
from homeassistant.const import UnitOfEnergy, UnitOfVolume
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import (
    DeviceEntry,
    DeviceEntryType,
    DeviceInfo,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
from . import IstaConfigEntry
from .const import DOMAIN
from .coordinator import IstaCoordinator
from .util import IstaConsumptionType, IstaValueType, get_native_value, get_statistics
 
_LOGGER = logging.getLogger(__name__)
 
 
@dataclass(kw_only=True, frozen=True)
class IstaSensorEntityDescription(SensorEntityDescription):
    """Ista EcoTrend Sensor Description."""
 
    consumption_type: IstaConsumptionType
    value_type: IstaValueType | None = None
 
 
class IstaSensorEntity(StrEnum):
    """Ista EcoTrend Entities."""
 
    HEATING = "heating"
    HEATING_ENERGY = "heating_energy"
    HEATING_COST = "heating_cost"
 
    HOT_WATER = "hot_water"
    HOT_WATER_ENERGY = "hot_water_energy"
    HOT_WATER_COST = "hot_water_cost"
 
    WATER = "water"
    WATER_COST = "water_cost"
 
 
SENSOR_DESCRIPTIONS: tuple[IstaSensorEntityDescription, ...] = (
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HEATING,
        translation_key=IstaSensorEntity.HEATING,
        suggested_display_precision=0,
        consumption_type=IstaConsumptionType.HEATING,
        native_unit_of_measurement="units",
        state_class=SensorStateClass.TOTAL,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HEATING_ENERGY,
        translation_key=IstaSensorEntity.HEATING_ENERGY,
        device_class=SensorDeviceClass.ENERGY,
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=1,
        consumption_type=IstaConsumptionType.HEATING,
        value_type=IstaValueType.ENERGY,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HEATING_COST,
        translation_key=IstaSensorEntity.HEATING_COST,
        device_class=SensorDeviceClass.MONETARY,
        native_unit_of_measurement="EUR",
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=0,
        entity_registry_enabled_default=False,
        consumption_type=IstaConsumptionType.HEATING,
        value_type=IstaValueType.COSTS,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HOT_WATER,
        translation_key=IstaSensorEntity.HOT_WATER,
        device_class=SensorDeviceClass.WATER,
        native_unit_of_measurement=UnitOfVolume.CUBIC_METERS,
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=1,
        consumption_type=IstaConsumptionType.HOT_WATER,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HOT_WATER_ENERGY,
        translation_key=IstaSensorEntity.HOT_WATER_ENERGY,
        device_class=SensorDeviceClass.ENERGY,
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=1,
        consumption_type=IstaConsumptionType.HOT_WATER,
        value_type=IstaValueType.ENERGY,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.HOT_WATER_COST,
        translation_key=IstaSensorEntity.HOT_WATER_COST,
        device_class=SensorDeviceClass.MONETARY,
        native_unit_of_measurement="EUR",
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=0,
        entity_registry_enabled_default=False,
        consumption_type=IstaConsumptionType.HOT_WATER,
        value_type=IstaValueType.COSTS,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.WATER,
        translation_key=IstaSensorEntity.WATER,
        device_class=SensorDeviceClass.WATER,
        native_unit_of_measurement=UnitOfVolume.CUBIC_METERS,
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=1,
        entity_registry_enabled_default=False,
        consumption_type=IstaConsumptionType.WATER,
    ),
    IstaSensorEntityDescription(
        key=IstaSensorEntity.WATER_COST,
        translation_key=IstaSensorEntity.WATER_COST,
        device_class=SensorDeviceClass.MONETARY,
        native_unit_of_measurement="EUR",
        state_class=SensorStateClass.TOTAL,
        suggested_display_precision=0,
        entity_registry_enabled_default=False,
        consumption_type=IstaConsumptionType.WATER,
        value_type=IstaValueType.COSTS,
    ),
)
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: IstaConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the ista EcoTrend sensors."""
 
    coordinator = config_entry.runtime_data
 
    async_add_entities(
        IstaSensor(coordinator, description, consumption_unit)
        for description in SENSOR_DESCRIPTIONS
        for consumption_unit in coordinator.data
    )
 
 
class IstaSensor(CoordinatorEntity[IstaCoordinator], SensorEntity):
    """Ista EcoTrend sensor."""
 
    entity_description: IstaSensorEntityDescription
    _attr_has_entity_name = True
    device_entry: DeviceEntry
 
    def __init__(
        self,
        coordinator: IstaCoordinator,
        entity_description: IstaSensorEntityDescription,
        consumption_unit: str,
    ) -> None:
        """Initialize the ista EcoTrend sensor."""
        super().__init__(coordinator)
        self.consumption_unitconsumption_unit = consumption_unit
        self.entity_descriptionentity_description = entity_description
        self._attr_unique_id_attr_unique_id = f"{consumption_unit}_{entity_description.key}"
        self._attr_device_info_attr_device_info = DeviceInfo(
            entry_type=DeviceEntryType.SERVICE,
            manufacturer="ista SE",
            model="ista EcoTrend",
            name=f"{coordinator.details[consumption_unit]["address"]["street"]} "
            f"{coordinator.details[consumption_unit]["address"]["houseNumber"]}".strip(),
            configuration_url="https://ecotrend.ista.de/",
            identifiers={(DOMAIN, consumption_unit)},
        )
 
    @property
    def native_value(self) -> StateType:
        """Return the state of the device."""
 
        return get_native_value(
            data=self.coordinator.data[self.consumption_unitconsumption_unit],
            consumption_type=self.entity_descriptionentity_description.consumption_type,
            value_type=self.entity_descriptionentity_description.value_type,
        )
 
    async def async_added_to_hass(self) -> None:
        """When added to hass."""
        # perform initial statistics import when sensor is added, otherwise it would take
        # 1 day when _handle_coordinator_update is triggered for the first time.
        await self.update_statisticsupdate_statistics()
        await super().async_added_to_hass()
 
    def _handle_coordinator_update(self) -> None:
        """Handle coordinator update."""
        asyncio.run_coroutine_threadsafe(self.update_statisticsupdate_statistics(), self.hasshasshass.loop)
 
    async def update_statistics(self) -> None:
        """Import ista EcoTrend historical statistics."""
 
        # Remember the statistic_id that was initially created
        # in case the entity gets renamed, because we cannot
        # change the statistic_id
        name = self.coordinator.config_entry.options.get(
            f"lts_{self.entity_description.key}_{self.consumption_unit}"
        )
        if not name:
            name = self.entity_identity_id.removeprefix("sensor.")
            self.hasshasshass.config_entries.async_update_entry(
                entry=self.coordinator.config_entry,
                options={
                    **self.coordinator.config_entry.options,
                    f"lts_{self.entity_description.key}_{self.consumption_unit}": name,
                },
            )
 
        statistic_id = f"{DOMAIN}:{name}"
        statistics_sum = 0.0
        statistics_since = None
 
        last_stats = await get_instance(self.hasshasshass).async_add_executor_job(
            get_last_statistics,
            self.hasshasshass,
            1,
            statistic_id,
            False,
            {"sum"},
        )
 
        _LOGGER.debug("Last statistics: %s", last_stats)
 
        if last_stats:
            statistics_sum = last_stats[statistic_id][0].get("sum") or 0.0
            statistics_since = datetime.datetime.fromtimestamp(
                last_stats[statistic_id][0].get("end") or 0, tz=datetime.UTC
            ) + datetime.timedelta(days=1)
 
        if monthly_consumptions := get_statistics(
            self.coordinator.data[self.consumption_unitconsumption_unit],
            self.entity_descriptionentity_description.consumption_type,
            self.entity_descriptionentity_description.value_type,
        ):
            statistics: list[StatisticData] = [
                {
                    "start": consumptions["date"],
                    "state": consumptions["value"],
                    "sum": (statistics_sum := statistics_sum + consumptions["value"]),
                }
                for consumptions in monthly_consumptions
                if statistics_since is None or consumptions["date"] > statistics_since
            ]
 
            metadata: StatisticMetaData = {
                "has_mean": False,
                "has_sum": True,
                "name": f"{self.device_entry.name} {self.name}",
                "source": DOMAIN,
                "statistic_id": statistic_id,
                "unit_of_measurement": self.entity_descriptionentity_description.native_unit_of_measurement,
            }
            if statistics:
                _LOGGER.debug("Insert statistics: %s %s", metadata, statistics)
                async_add_external_statistics(self.hasshasshass, metadata, statistics)