sensor.py

Go to the documentation of this file.

"""Support for iammeter via local API."""
 
from __future__ import annotations
 
from asyncio import timeout
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
 
from iammeter.client import IamMeter
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorDeviceClass,
    SensorEntity,
    SensorEntityDescription,
    SensorStateClass,
)
from homeassistant.const import (
    CONF_HOST,
    CONF_NAME,
    CONF_PORT,
    PERCENTAGE,
    Platform,
    UnitOfElectricCurrent,
    UnitOfElectricPotential,
    UnitOfEnergy,
    UnitOfFrequency,
    UnitOfPower,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import debounce, entity_registry as er, update_coordinator
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DEVICE_3080, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
DEFAULT_PORT = 80
DEFAULT_DEVICE_NAME = "IamMeter"
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_HOST): cv.string,
        vol.Optional(CONF_NAME, default=DEFAULT_DEVICE_NAME): cv.string,
        vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
    }
)
 
SCAN_INTERVAL = timedelta(seconds=30)
PLATFORM_TIMEOUT = 8
 
 
def _migrate_to_new_unique_id(
    hass: HomeAssistant, model: str, serial_number: str
) -> None:
    """Migrate old unique ids to new unique ids."""
    ent_reg = er.async_get(hass)
    name_list = [
        "Voltage",
        "Current",
        "Power",
        "ImportEnergy",
        "ExportGrid",
        "Frequency",
        "PF",
    ]
    phase_list = ["A", "B", "C", "NET"]
    id_phase_range = 1 if model == DEVICE_3080 else 4
    id_name_range = 5 if model == DEVICE_3080 else 7
    for row in range(id_phase_range):
        for idx in range(id_name_range):
            old_unique_id = f"{serial_number}-{row}-{idx}"
            new_unique_id = (
                f"{serial_number}_{name_list[idx]}"
                if model == DEVICE_3080
                else f"{serial_number}_{name_list[idx]}_{phase_list[row]}"
            )
            entity_id = ent_reg.async_get_entity_id(
                Platform.SENSOR, DOMAIN, old_unique_id
            )
            if entity_id is not None:
                try:
                    ent_reg.async_update_entity(entity_id, new_unique_id=new_unique_id)
                except ValueError:
                    _LOGGER.warning(
                        "Skip migration of id [%s] to [%s] because it already exists",
                        old_unique_id,
                        new_unique_id,
                    )
                else:
                    _LOGGER.debug(
                        "Migrating unique_id from [%s] to [%s]",
                        old_unique_id,
                        new_unique_id,
                    )
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Platform setup."""
    config_host = config[CONF_HOST]
    config_port = config[CONF_PORT]
    config_name = config[CONF_NAME]
    try:
        api = await hass.async_add_executor_job(
            IamMeter, config_host, config_port, config_name
        )
    except TimeoutError as err:
        _LOGGER.error("Device is not ready")
        raise PlatformNotReady from err
 
    async def async_update_data():
        try:
            async with timeout(PLATFORM_TIMEOUT):
                return await hass.async_add_executor_job(api.client.get_data)
        except TimeoutError as err:
            raise UpdateFailed from err
 
    coordinator = DataUpdateCoordinator(
        hass,
        _LOGGER,
        name=config_name,
        update_method=async_update_data,
        update_interval=SCAN_INTERVAL,
        request_refresh_debouncer=debounce.Debouncer(
            hass, _LOGGER, cooldown=0.3, immediate=True
        ),
    )
    await coordinator.async_refresh()
    model = coordinator.data["Model"]
    serial_number = coordinator.data["sn"]
    _migrate_to_new_unique_id(hass, model, serial_number)
    if model == DEVICE_3080:
        async_add_entities(
            IammeterSensor(coordinator, description)
            for description in SENSOR_TYPES_3080
        )
    else:  # DEVICE_3080T:
        async_add_entities(
            IammeterSensor(coordinator, description)
            for description in SENSOR_TYPES_3080T
        )
 
 
class IammeterSensor(update_coordinator.CoordinatorEntity, SensorEntity):
    """Representation of a Sensor."""
 
    entity_description: IammeterSensorEntityDescription
    _attr_has_entity_name = True
    _attr_name = None
 
    def __init__(
        self,
        coordinator: DataUpdateCoordinator,
        description: IammeterSensorEntityDescription,
    ) -> None:
        """Initialize the sensor."""
        super().__init__(coordinator)
        self.entity_descriptionentity_description = description
        self._attr_unique_id_attr_unique_id = f"{coordinator.data['sn']}_{description.key}"
        self._attr_device_info_attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, coordinator.data["sn"])},
            manufacturer="IamMeter",
            name=coordinator.name,
        )
 
    @property
    def native_value(self):
        """Return the native sensor value."""
        raw_attr = self.coordinator.data.get(self.entity_descriptionentity_description.key, None)
        if self.entity_descriptionentity_description.value:
            return self.entity_descriptionentity_description.value(raw_attr)
        return raw_attr
 
 
@dataclass(frozen=True)
class IammeterSensorEntityDescription(SensorEntityDescription):
    """Describes Iammeter sensor entity."""
 
    value: Callable[[float | int], float] | Callable[[datetime], datetime] | None = None
 
 
SENSOR_TYPES_3080: tuple[IammeterSensorEntityDescription, ...] = (
    IammeterSensorEntityDescription(
        key="Voltage",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Current",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.CURRENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Power",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfPower.WATT,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    IammeterSensorEntityDescription(
        key="ImportEnergy",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="ExportGrid",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
)
SENSOR_TYPES_3080T: tuple[IammeterSensorEntityDescription, ...] = (
    IammeterSensorEntityDescription(
        key="Voltage_A",
        translation_key="voltage_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Current_A",
        translation_key="current_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.CURRENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Power_A",
        translation_key="power_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfPower.WATT,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    IammeterSensorEntityDescription(
        key="ImportEnergy_A",
        translation_key="import_energy_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="ExportGrid_A",
        translation_key="export_grid_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="Frequency_A",
        translation_key="frequency_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfFrequency.HERTZ,
        device_class=SensorDeviceClass.FREQUENCY,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="PF_A",
        translation_key="pf_a",
        icon="mdi:solar-power",
        native_unit_of_measurement=PERCENTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.POWER_FACTOR,
        value=lambda value: value * 100,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Voltage_B",
        translation_key="voltage_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Current_B",
        translation_key="current_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.CURRENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Power_B",
        translation_key="power_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfPower.WATT,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    IammeterSensorEntityDescription(
        key="ImportEnergy_B",
        translation_key="import_energy_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="ExportGrid_B",
        translation_key="export_grid_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="Frequency_B",
        translation_key="frequency_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfFrequency.HERTZ,
        device_class=SensorDeviceClass.FREQUENCY,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="PF_B",
        translation_key="pf_b",
        icon="mdi:solar-power",
        native_unit_of_measurement=PERCENTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.POWER_FACTOR,
        value=lambda value: value * 100,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Voltage_C",
        translation_key="voltage_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Current_C",
        translation_key="current_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.CURRENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Power_C",
        translation_key="power_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfPower.WATT,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    IammeterSensorEntityDescription(
        key="ImportEnergy_C",
        translation_key="import_energy_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="ExportGrid_C",
        translation_key="export_grid_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    IammeterSensorEntityDescription(
        key="Frequency_C",
        translation_key="frequency_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfFrequency.HERTZ,
        device_class=SensorDeviceClass.FREQUENCY,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="PF_C",
        translation_key="pf_c",
        icon="mdi:solar-power",
        native_unit_of_measurement=PERCENTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.POWER_FACTOR,
        value=lambda value: value * 100,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Voltage_Net",
        translation_key="voltage_net",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Power_Net",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfPower.WATT,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="ImportEnergy_Net",
        translation_key="import_energy_net",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="ExportGrid_Net",
        translation_key="export_grid_net",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="Frequency_Net",
        translation_key="frequency_net",
        icon="mdi:solar-power",
        native_unit_of_measurement=UnitOfFrequency.HERTZ,
        device_class=SensorDeviceClass.FREQUENCY,
        state_class=SensorStateClass.MEASUREMENT,
        entity_registry_enabled_default=False,
    ),
    IammeterSensorEntityDescription(
        key="PF_Net",
        translation_key="pf_net",
        icon="mdi:solar-power",
        native_unit_of_measurement=PERCENTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        device_class=SensorDeviceClass.POWER_FACTOR,
        value=lambda value: value * 100,
        entity_registry_enabled_default=False,
    ),
)