sensor.py

Go to the documentation of this file.

"""Support for Ebusd sensors."""
 
from __future__ import annotations
 
import datetime
import logging
 
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import Throttle
import homeassistant.util.dt as dt_util
 
from .const import DOMAIN
 
TIME_FRAME1_BEGIN = "time_frame1_begin"
TIME_FRAME1_END = "time_frame1_end"
TIME_FRAME2_BEGIN = "time_frame2_begin"
TIME_FRAME2_END = "time_frame2_end"
TIME_FRAME3_BEGIN = "time_frame3_begin"
TIME_FRAME3_END = "time_frame3_end"
MIN_TIME_BETWEEN_UPDATES = datetime.timedelta(seconds=15)
 
_LOGGER = logging.getLogger(__name__)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Ebus sensor."""
    if not discovery_info:
        return
    ebusd_api = hass.data[DOMAIN]
    monitored_conditions = discovery_info["monitored_conditions"]
    name = discovery_info["client_name"]
 
    add_entities(
        (
            EbusdSensor(ebusd_api, discovery_info["sensor_types"][condition], name)
            for condition in monitored_conditions
        ),
        True,
    )
 
 
class EbusdSensor(SensorEntity):
    """Ebusd component sensor methods definition."""
 
    def __init__(self, data, sensor, name):
        """Initialize the sensor."""
        self._state_state = None
        self._client_name_client_name = name
        (
            self._name,
            self._unit_of_measurement,
            self._icon,
            self._type_type,
            self._device_class,
        ) = sensor
        self.datadata = data
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return f"{self._client_name} {self._name}"
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state
 
    @property
    def extra_state_attributes(self):
        """Return the device state attributes."""
        if self._type_type == 1 and self._state_state is not None:
            schedule = {
                TIME_FRAME1_BEGIN: None,
                TIME_FRAME1_END: None,
                TIME_FRAME2_BEGIN: None,
                TIME_FRAME2_END: None,
                TIME_FRAME3_BEGIN: None,
                TIME_FRAME3_END: None,
            }
            time_frame = self._state_state.split(";")
            for index, item in enumerate(sorted(schedule.items())):
                if index < len(time_frame):
                    parsed = datetime.datetime.strptime(time_frame[index], "%H:%M")
                    parsed = parsed.replace(
                        dt_util.now().year, dt_util.now().month, dt_util.now().day
                    )
                    schedule[item[0]] = parsed.isoformat()
            return schedule
        return None
 
    @property
    def device_class(self):
        """Return the class of this device, from component DEVICE_CLASSES."""
        return self._device_class
 
    @property
    def icon(self):
        """Icon to use in the frontend, if any."""
        return self._icon
 
    @property
    def native_unit_of_measurement(self):
        """Return the unit of measurement."""
        return self._unit_of_measurement
 
    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self) -> None:
        """Fetch new state data for the sensor."""
        try:
            self.datadata.update(self._name, self._type_type)
            if self._name not in self.datadata.value:
                return
 
            self._state_state = self.datadata.value[self._name]
        except RuntimeError:
            _LOGGER.debug("EbusdData.update exception")