sensor.py

Go to the documentation of this file.

"""Sensor component for LaCrosse View."""
 
from __future__ import annotations
 
from collections.abc import Callable
from dataclasses import dataclass, replace
import logging
 
from lacrosse_view import Sensor
 
from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorEntityDescription,
    SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    DEGREE,
    PERCENTAGE,
    UnitOfPrecipitationDepth,
    UnitOfPressure,
    UnitOfSpeed,
    UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
    CoordinatorEntity,
    DataUpdateCoordinator,
)
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
@dataclass(frozen=True, kw_only=True)
class LaCrosseSensorEntityDescription(SensorEntityDescription):
    """Description for LaCrosse View sensor."""
 
    value_fn: Callable[[Sensor, str], float | int | str | None]
 
 
def get_value(sensor: Sensor, field: str) -> float | int | str | None:
    """Get the value of a sensor field."""
    field_data = sensor.data.get(field)
    if field_data is None:
        return None
    value = field_data["values"][-1]["s"]
    try:
        value = float(value)
    except ValueError:
        return str(value)  # handle non-numericals
    return int(value) if value.is_integer() else value
 
 
PARALLEL_UPDATES = 0
SENSOR_DESCRIPTIONS = {
    "Temperature": LaCrosseSensorEntityDescription(
        key="Temperature",
        device_class=SensorDeviceClass.TEMPERATURE,
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
    ),
    "Humidity": LaCrosseSensorEntityDescription(
        key="Humidity",
        device_class=SensorDeviceClass.HUMIDITY,
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        native_unit_of_measurement=PERCENTAGE,
    ),
    "HeatIndex": LaCrosseSensorEntityDescription(
        key="HeatIndex",
        translation_key="heat_index",
        device_class=SensorDeviceClass.TEMPERATURE,
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
    ),
    "WindSpeed": LaCrosseSensorEntityDescription(
        key="WindSpeed",
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
        device_class=SensorDeviceClass.WIND_SPEED,
    ),
    "Rain": LaCrosseSensorEntityDescription(
        key="Rain",
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        native_unit_of_measurement=UnitOfPrecipitationDepth.MILLIMETERS,
        device_class=SensorDeviceClass.PRECIPITATION,
    ),
    "WindHeading": LaCrosseSensorEntityDescription(
        key="WindHeading",
        translation_key="wind_heading",
        value_fn=get_value,
        native_unit_of_measurement=DEGREE,
    ),
    "WetDry": LaCrosseSensorEntityDescription(
        key="WetDry",
        translation_key="wet_dry",
        value_fn=get_value,
    ),
    "Flex": LaCrosseSensorEntityDescription(
        key="Flex",
        translation_key="flex",
        value_fn=get_value,
    ),
    "BarometricPressure": LaCrosseSensorEntityDescription(
        key="BarometricPressure",
        translation_key="barometric_pressure",
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        device_class=SensorDeviceClass.ATMOSPHERIC_PRESSURE,
        native_unit_of_measurement=UnitOfPressure.HPA,
    ),
    "FeelsLike": LaCrosseSensorEntityDescription(
        key="FeelsLike",
        translation_key="feels_like",
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        device_class=SensorDeviceClass.TEMPERATURE,
        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
    ),
    "WindChill": LaCrosseSensorEntityDescription(
        key="WindChill",
        translation_key="wind_chill",
        state_class=SensorStateClass.MEASUREMENT,
        value_fn=get_value,
        device_class=SensorDeviceClass.TEMPERATURE,
        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
    ),
}
# map of API returned unit of measurement strings to their corresponding unit of measurement
UNIT_OF_MEASUREMENT_MAP = {
    "degrees_celsius": UnitOfTemperature.CELSIUS,
    "degrees_fahrenheit": UnitOfTemperature.FAHRENHEIT,
    "inches": UnitOfPrecipitationDepth.INCHES,
    "millimeters": UnitOfPrecipitationDepth.MILLIMETERS,
    "kilometers_per_hour": UnitOfSpeed.KILOMETERS_PER_HOUR,
    "miles_per_hour": UnitOfSpeed.MILES_PER_HOUR,
}
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up LaCrosse View from a config entry."""
    coordinator: DataUpdateCoordinator[list[Sensor]] = hass.data[DOMAIN][
        entry.entry_id
    ]["coordinator"]
    sensors: list[Sensor] = coordinator.data
 
    sensor_list = []
    for i, sensor in enumerate(sensors):
        for field in sensor.sensor_field_names:
            description = SENSOR_DESCRIPTIONS.get(field)
            if description is None:
                message = (
                    f"Unsupported sensor field: {field}\nPlease create an issue on "
                    "GitHub."
                    " https://github.com/home-assistant/core/issues/new?assignees="
                    "&labels=&template=bug_report.yml&integration_name=LaCrosse%20View"
                    "&integration_link="
                    "https://www.home-assistant.io/integrations/lacrosse_view/"
                    "&additional_information="
                    f"Field:%20{field}%0ASensor%20Model:%20{sensor.model}"
                    f"&title=LaCrosse%20View%20Unsupported%20sensor%20field:%20{field}"
                )
 
                _LOGGER.warning(message)
                continue
 
            # if the API returns a different unit of measurement from the description, update it
            if sensor.data.get(field) is not None:
                native_unit_of_measurement = UNIT_OF_MEASUREMENT_MAP.get(
                    sensor.data[field].get("unit")
                )
 
                if native_unit_of_measurement is not None:
                    description = replace(
                        description,
                        native_unit_of_measurement=native_unit_of_measurement,
                    )
 
            sensor_list.append(
                LaCrosseViewSensor(
                    coordinator=coordinator,
                    description=description,
                    sensor=sensor,
                    index=i,
                )
            )
 
    async_add_entities(sensor_list)
 
 
class LaCrosseViewSensor(
    CoordinatorEntity[DataUpdateCoordinator[list[Sensor]]], SensorEntity
):
    """LaCrosse View sensor."""
 
    entity_description: LaCrosseSensorEntityDescription
    _attr_has_entity_name = True
 
    def __init__(
        self,
        description: LaCrosseSensorEntityDescription,
        coordinator: DataUpdateCoordinator[list[Sensor]],
        sensor: Sensor,
        index: int,
    ) -> None:
        """Initialize."""
        super().__init__(coordinator)
 
        self.entity_descriptionentity_description = description
        self._attr_unique_id_attr_unique_id = f"{sensor.sensor_id}-{description.key}"
        self._attr_device_info_attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, sensor.sensor_id)},
            name=sensor.name,
            manufacturer="LaCrosse Technology",
            model=sensor.model,
            via_device=(DOMAIN, sensor.location.id),
        )
        self.indexindex = index
 
    @property
    def native_value(self) -> int | float | str | None:
        """Return the sensor value."""
        return self.entity_descriptionentity_description.value_fn(
            self.coordinator.data[self.indexindex], self.entity_descriptionentity_description.key
        )
 
    @property
    def available(self) -> bool:
        """Return True if entity is available."""
        return (
            super().available
            and self.entity_descriptionentity_description.key in self.coordinator.data[self.indexindex].data
        )