sensor.py

Go to the documentation of this file.

"""Linky Atome."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
from pyatome.client import AtomeClient, PyAtomeError
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.const import (
    CONF_NAME,
    CONF_PASSWORD,
    CONF_USERNAME,
    UnitOfEnergy,
    UnitOfPower,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import Throttle
 
_LOGGER = logging.getLogger(__name__)
 
DEFAULT_NAME = "atome"
 
LIVE_SCAN_INTERVAL = timedelta(seconds=30)
DAILY_SCAN_INTERVAL = timedelta(seconds=150)
WEEKLY_SCAN_INTERVAL = timedelta(hours=1)
MONTHLY_SCAN_INTERVAL = timedelta(hours=1)
YEARLY_SCAN_INTERVAL = timedelta(days=1)
 
LIVE_NAME = "Atome Live Power"
DAILY_NAME = "Atome Daily"
WEEKLY_NAME = "Atome Weekly"
MONTHLY_NAME = "Atome Monthly"
YEARLY_NAME = "Atome Yearly"
 
LIVE_TYPE = "live"
DAILY_TYPE = "day"
WEEKLY_TYPE = "week"
MONTHLY_TYPE = "month"
YEARLY_TYPE = "year"
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_USERNAME): cv.string,
        vol.Required(CONF_PASSWORD): cv.string,
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Atome sensor."""
    username = config[CONF_USERNAME]
    password = config[CONF_PASSWORD]
 
    try:
        atome_client = AtomeClient(username, password)
        atome_client.login()
    except PyAtomeError as exp:
        _LOGGER.error(exp)
        return
 
    data = AtomeData(atome_client)
 
    sensors = []
    sensors.append(AtomeSensor(data, LIVE_NAME, LIVE_TYPE))
    sensors.append(AtomeSensor(data, DAILY_NAME, DAILY_TYPE))
    sensors.append(AtomeSensor(data, WEEKLY_NAME, WEEKLY_TYPE))
    sensors.append(AtomeSensor(data, MONTHLY_NAME, MONTHLY_TYPE))
    sensors.append(AtomeSensor(data, YEARLY_NAME, YEARLY_TYPE))
 
    add_entities(sensors, True)
 
 
class AtomeData:
    """Stores data retrieved from Neurio sensor."""
 
    def __init__(self, client: AtomeClient) -> None:
        """Initialize the data."""
        self.atome_clientatome_client = client
        self._live_power_live_power = None
        self._subscribed_power_subscribed_power = None
        self._is_connected_is_connected = None
        self._day_usage_day_usage = None
        self._day_price_day_price = None
        self._week_usage_week_usage = None
        self._week_price_week_price = None
        self._month_usage_month_usage = None
        self._month_price_month_price = None
        self._year_usage_year_usage = None
        self._year_price_year_price = None
 
    @property
    def live_power(self):
        """Return latest active power value."""
        return self._live_power_live_power
 
    @property
    def subscribed_power(self):
        """Return latest active power value."""
        return self._subscribed_power_subscribed_power
 
    @property
    def is_connected(self):
        """Return latest active power value."""
        return self._is_connected_is_connected
 
    def _retrieve_live(self):
        values = self.atome_clientatome_client.get_live()
        if (
            values.get("last")
            and values.get("subscribed")
            and (values.get("isConnected") is not None)
        ):
            self._live_power_live_power = values["last"]
            self._subscribed_power_subscribed_power = values["subscribed"]
            self._is_connected_is_connected = values["isConnected"]
            _LOGGER.debug(
                "Updating Atome live data. Got: %d, isConnected: %s, subscribed: %d",
                self._live_power_live_power,
                self._is_connected_is_connected,
                self._subscribed_power_subscribed_power,
            )
            return True
 
        _LOGGER.error("Live Data : Missing last value in values: %s", values)
        return False
 
    @Throttle(LIVE_SCAN_INTERVAL)
    def update_live_usage(self):
        """Return current power value."""
        if not self._retrieve_live_retrieve_live():
            _LOGGER.debug("Perform Reconnect during live request")
            self.atome_clientatome_client.login()
            self._retrieve_live_retrieve_live()
 
    def _retrieve_period_usage(self, period_type):
        """Return current daily/weekly/monthly/yearly power usage."""
        values = self.atome_clientatome_client.get_consumption(period_type)
        if values.get("total") and values.get("price"):
            period_usage = values["total"] / 1000
            period_price = values["price"]
            _LOGGER.debug("Updating Atome %s data. Got: %d", period_type, period_usage)
            return True, period_usage, period_price
 
        _LOGGER.error("%s : Missing last value in values: %s", period_type, values)
        return False, None, None
 
    def _retrieve_period_usage_with_retry(self, period_type):
        """Return current daily/weekly/monthly/yearly power usage with one retry."""
        (
            retrieve_success,
            period_usage,
            period_price,
        ) = self._retrieve_period_usage_retrieve_period_usage(period_type)
        if not retrieve_success:
            _LOGGER.debug("Perform Reconnect during %s", period_type)
            self.atome_clientatome_client.login()
            (
                retrieve_success,
                period_usage,
                period_price,
            ) = self._retrieve_period_usage_retrieve_period_usage(period_type)
        return (period_usage, period_price)
 
    @property
    def day_usage(self):
        """Return latest daily usage value."""
        return self._day_usage_day_usage
 
    @property
    def day_price(self):
        """Return latest daily usage value."""
        return self._day_price_day_price
 
    @Throttle(DAILY_SCAN_INTERVAL)
    def update_day_usage(self):
        """Return current daily power usage."""
        (
            self._day_usage_day_usage,
            self._day_price_day_price,
        ) = self._retrieve_period_usage_with_retry_retrieve_period_usage_with_retry(DAILY_TYPE)
 
    @property
    def week_usage(self):
        """Return latest weekly usage value."""
        return self._week_usage_week_usage
 
    @property
    def week_price(self):
        """Return latest weekly usage value."""
        return self._week_price_week_price
 
    @Throttle(WEEKLY_SCAN_INTERVAL)
    def update_week_usage(self):
        """Return current weekly power usage."""
        (
            self._week_usage_week_usage,
            self._week_price_week_price,
        ) = self._retrieve_period_usage_with_retry_retrieve_period_usage_with_retry(WEEKLY_TYPE)
 
    @property
    def month_usage(self):
        """Return latest monthly usage value."""
        return self._month_usage_month_usage
 
    @property
    def month_price(self):
        """Return latest monthly usage value."""
        return self._month_price_month_price
 
    @Throttle(MONTHLY_SCAN_INTERVAL)
    def update_month_usage(self):
        """Return current monthly power usage."""
        (
            self._month_usage_month_usage,
            self._month_price_month_price,
        ) = self._retrieve_period_usage_with_retry_retrieve_period_usage_with_retry(MONTHLY_TYPE)
 
    @property
    def year_usage(self):
        """Return latest yearly usage value."""
        return self._year_usage_year_usage
 
    @property
    def year_price(self):
        """Return latest yearly usage value."""
        return self._year_price_year_price
 
    @Throttle(YEARLY_SCAN_INTERVAL)
    def update_year_usage(self):
        """Return current yearly power usage."""
        (
            self._year_usage_year_usage,
            self._year_price_year_price,
        ) = self._retrieve_period_usage_with_retry_retrieve_period_usage_with_retry(YEARLY_TYPE)
 
 
class AtomeSensor(SensorEntity):
    """Representation of a sensor entity for Atome."""
 
    def __init__(self, data, name, sensor_type):
        """Initialize the sensor."""
        self._attr_name_attr_name = name
        self._data_data = data
 
        self._sensor_type_sensor_type = sensor_type
 
        if sensor_type == LIVE_TYPE:
            self._attr_device_class_attr_device_class = SensorDeviceClass.POWER
            self._attr_native_unit_of_measurement_attr_native_unit_of_measurement = UnitOfPower.WATT
            self._attr_state_class_attr_state_class = SensorStateClass.MEASUREMENT
        else:
            self._attr_device_class_attr_device_class = SensorDeviceClass.ENERGY
            self._attr_native_unit_of_measurement_attr_native_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
            self._attr_state_class_attr_state_class = SensorStateClass.TOTAL_INCREASING
 
    def update(self) -> None:
        """Update device state."""
        update_function = getattr(self._data_data, f"update_{self._sensor_type}_usage")
        update_function()
 
        if self._sensor_type_sensor_type == LIVE_TYPE:
            self._attr_native_value_attr_native_value = self._data_data.live_power
            self._attr_extra_state_attributes_attr_extra_state_attributes = {
                "subscribed_power": self._data_data.subscribed_power,
                "is_connected": self._data_data.is_connected,
            }
        else:
            self._attr_native_value_attr_native_value = getattr(self._data_data, f"{self._sensor_type}_usage")
            self._attr_extra_state_attributes_attr_extra_state_attributes = {
                "price": getattr(self._data_data, f"{self._sensor_type}_price")
            }