sensor.py

Go to the documentation of this file.

"""Support for hydrological data from the Fed. Office for the Environment."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
from swisshydrodata import SwissHydroData
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorEntity,
)
from homeassistant.const import CONF_MONITORED_CONDITIONS
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__)
 
ATTR_MAX_24H = "max-24h"
ATTR_MEAN_24H = "mean-24h"
ATTR_MIN_24H = "min-24h"
ATTR_STATION = "station"
ATTR_STATION_UPDATE = "station_update"
ATTR_WATER_BODY = "water_body"
ATTR_WATER_BODY_TYPE = "water_body_type"
 
CONF_STATION = "station"
 
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
 
SENSOR_DISCHARGE = "discharge"
SENSOR_LEVEL = "level"
SENSOR_TEMPERATURE = "temperature"
 
CONDITIONS = {
    SENSOR_DISCHARGE: "mdi:waves",
    SENSOR_LEVEL: "mdi:zodiac-aquarius",
    SENSOR_TEMPERATURE: "mdi:oil-temperature",
}
 
CONDITION_DETAILS = [
    ATTR_MAX_24H,
    ATTR_MEAN_24H,
    ATTR_MIN_24H,
]
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_STATION): vol.Coerce(int),
        vol.Optional(CONF_MONITORED_CONDITIONS, default=[SENSOR_TEMPERATURE]): vol.All(
            cv.ensure_list, [vol.In(CONDITIONS)]
        ),
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Swiss hydrological sensor."""
    station = config[CONF_STATION]
    monitored_conditions = config[CONF_MONITORED_CONDITIONS]
 
    hydro_data = HydrologicalData(station)
    hydro_data.update()
 
    if hydro_data.data is None:
        _LOGGER.error("The station doesn't exists: %s", station)
        return
 
    add_entities(
        (
            SwissHydrologicalDataSensor(hydro_data, station, condition)
            for condition in monitored_conditions
        ),
        True,
    )
 
 
class SwissHydrologicalDataSensor(SensorEntity):
    """Implementation of a Swiss hydrological sensor."""
 
    _attr_attribution = (
        "Data provided by the Swiss Federal Office for the Environment FOEN"
    )
 
    def __init__(self, hydro_data, station, condition):
        """Initialize the Swiss hydrological sensor."""
        self.hydro_datahydro_data = hydro_data
        self._condition_condition = condition
        self._data_data = self._state_state = self._unit_of_measurement_unit_of_measurement = None
        self._icon_icon = CONDITIONS[condition]
        self._station_station = station
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return f"{self._data['water-body-name']} {self._condition}"
 
    @property
    def unique_id(self) -> str:
        """Return a unique, friendly identifier for this entity."""
        return f"{self._station}_{self._condition}"
 
    @property
    def native_unit_of_measurement(self):
        """Return the unit of measurement of this entity, if any."""
        if self._state_state is not None:
            return self.hydro_datahydro_data.data["parameters"][self._condition_condition]["unit"]
        return None
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        if isinstance(self._state_state, (int, float)):
            return round(self._state_state, 2)
        return None
 
    @property
    def extra_state_attributes(self):
        """Return the device state attributes."""
        attrs = {}
 
        if not self._data_data:
            return attrs
 
        attrs[ATTR_WATER_BODY_TYPE] = self._data_data["water-body-type"]
        attrs[ATTR_STATION] = self._data_data["name"]
        attrs[ATTR_STATION_UPDATE] = self._data_data["parameters"][self._condition_condition][
            "datetime"
        ]
 
        for entry in CONDITION_DETAILS:
            attrs[entry.replace("-", "_")] = self._data_data["parameters"][self._condition_condition][
                entry
            ]
 
        return attrs
 
    @property
    def icon(self):
        """Icon to use in the frontend."""
        return self._icon_icon
 
    def update(self) -> None:
        """Get the latest data and update the state."""
        self.hydro_datahydro_data.update()
        self._data_data = self.hydro_datahydro_data.data
 
        if self._data_data is None:
            self._state_state = None
        else:
            self._state_state = self._data_data["parameters"][self._condition_condition]["value"]
 
 
class HydrologicalData:
    """The Class for handling the data retrieval."""
 
    def __init__(self, station):
        """Initialize the data object."""
        self.stationstation = station
        self.datadata = None
 
    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        """Get the latest data."""
 
        shd = SwissHydroData()
        self.datadata = shd.get_station(self.stationstation)