sensor.py

Go to the documentation of this file.

"""Representation of Z-Wave sensors."""
 
from __future__ import annotations
 
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any
 
import voluptuous as vol
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.const import CommandClass
from zwave_js_server.const.command_class.meter import (
    RESET_METER_OPTION_TARGET_VALUE,
    RESET_METER_OPTION_TYPE,
)
from zwave_js_server.exceptions import BaseZwaveJSServerError
from zwave_js_server.model.controller import Controller
from zwave_js_server.model.controller.statistics import ControllerStatistics
from zwave_js_server.model.driver import Driver
from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.node.statistics import NodeStatistics
from zwave_js_server.util.command_class.meter import get_meter_type
 
from homeassistant.components.sensor import (
    DOMAIN as SENSOR_DOMAIN,
    SensorDeviceClass,
    SensorEntity,
    SensorEntityDescription,
    SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONCENTRATION_PARTS_PER_MILLION,
    LIGHT_LUX,
    PERCENTAGE,
    SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
    UV_INDEX,
    EntityCategory,
    UnitOfElectricCurrent,
    UnitOfElectricPotential,
    UnitOfEnergy,
    UnitOfPower,
    UnitOfPressure,
    UnitOfTemperature,
    UnitOfTime,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_platform
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import UNDEFINED, StateType
 
from .binary_sensor import is_valid_notification_binary_sensor
from .const import (
    ATTR_METER_TYPE,
    ATTR_METER_TYPE_NAME,
    ATTR_VALUE,
    DATA_CLIENT,
    DOMAIN,
    ENTITY_DESC_KEY_BATTERY,
    ENTITY_DESC_KEY_CO,
    ENTITY_DESC_KEY_CO2,
    ENTITY_DESC_KEY_CURRENT,
    ENTITY_DESC_KEY_ENERGY_MEASUREMENT,
    ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
    ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
    ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
    ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
    ENTITY_DESC_KEY_ENERGY_TOTAL_INCREASING,
    ENTITY_DESC_KEY_HUMIDITY,
    ENTITY_DESC_KEY_ILLUMINANCE,
    ENTITY_DESC_KEY_MEASUREMENT,
    ENTITY_DESC_KEY_POWER,
    ENTITY_DESC_KEY_POWER_FACTOR,
    ENTITY_DESC_KEY_PRESSURE,
    ENTITY_DESC_KEY_SIGNAL_STRENGTH,
    ENTITY_DESC_KEY_TARGET_TEMPERATURE,
    ENTITY_DESC_KEY_TEMPERATURE,
    ENTITY_DESC_KEY_TOTAL_INCREASING,
    ENTITY_DESC_KEY_UV_INDEX,
    ENTITY_DESC_KEY_VOLTAGE,
    LOGGER,
    SERVICE_RESET_METER,
)
from .discovery import ZwaveDiscoveryInfo
from .discovery_data_template import (
    NumericSensorDataTemplate,
    NumericSensorDataTemplateData,
)
from .entity import ZWaveBaseEntity
from .helpers import get_device_info, get_valueless_base_unique_id
from .migrate import async_migrate_statistics_sensors
 
PARALLEL_UPDATES = 0
 
 
# These descriptions should include device class.
ENTITY_DESCRIPTION_KEY_DEVICE_CLASS_MAP: dict[
    tuple[str, str], SensorEntityDescription
] = {
    (ENTITY_DESC_KEY_BATTERY, PERCENTAGE): SensorEntityDescription(
        key=ENTITY_DESC_KEY_BATTERY,
        device_class=SensorDeviceClass.BATTERY,
        entity_category=EntityCategory.DIAGNOSTIC,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=PERCENTAGE,
    ),
    (ENTITY_DESC_KEY_CURRENT, UnitOfElectricCurrent.AMPERE): SensorEntityDescription(
        key=ENTITY_DESC_KEY_CURRENT,
        device_class=SensorDeviceClass.CURRENT,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
    ),
    (ENTITY_DESC_KEY_VOLTAGE, UnitOfElectricPotential.VOLT): SensorEntityDescription(
        key=ENTITY_DESC_KEY_VOLTAGE,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfElectricPotential.VOLT,
        suggested_display_precision=0,
    ),
    (
        ENTITY_DESC_KEY_VOLTAGE,
        UnitOfElectricPotential.MILLIVOLT,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_VOLTAGE,
        device_class=SensorDeviceClass.VOLTAGE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
    ),
    (
        ENTITY_DESC_KEY_ENERGY_TOTAL_INCREASING,
        UnitOfEnergy.KILO_WATT_HOUR,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_TOTAL_INCREASING,
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
        native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
    ),
    (ENTITY_DESC_KEY_POWER, UnitOfPower.WATT): SensorEntityDescription(
        key=ENTITY_DESC_KEY_POWER,
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPower.WATT,
    ),
    (ENTITY_DESC_KEY_POWER_FACTOR, PERCENTAGE): SensorEntityDescription(
        key=ENTITY_DESC_KEY_POWER_FACTOR,
        device_class=SensorDeviceClass.POWER_FACTOR,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=PERCENTAGE,
    ),
    (ENTITY_DESC_KEY_CO, CONCENTRATION_PARTS_PER_MILLION): SensorEntityDescription(
        key=ENTITY_DESC_KEY_CO,
        device_class=SensorDeviceClass.CO,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
    ),
    (ENTITY_DESC_KEY_CO2, CONCENTRATION_PARTS_PER_MILLION): SensorEntityDescription(
        key=ENTITY_DESC_KEY_CO2,
        device_class=SensorDeviceClass.CO2,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
    ),
    (ENTITY_DESC_KEY_HUMIDITY, PERCENTAGE): SensorEntityDescription(
        key=ENTITY_DESC_KEY_HUMIDITY,
        device_class=SensorDeviceClass.HUMIDITY,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=PERCENTAGE,
    ),
    (ENTITY_DESC_KEY_ILLUMINANCE, LIGHT_LUX): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ILLUMINANCE,
        device_class=SensorDeviceClass.ILLUMINANCE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=LIGHT_LUX,
    ),
    (ENTITY_DESC_KEY_PRESSURE, UnitOfPressure.KPA): SensorEntityDescription(
        key=ENTITY_DESC_KEY_PRESSURE,
        device_class=SensorDeviceClass.PRESSURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPressure.KPA,
    ),
    (ENTITY_DESC_KEY_PRESSURE, UnitOfPressure.PSI): SensorEntityDescription(
        key=ENTITY_DESC_KEY_PRESSURE,
        device_class=SensorDeviceClass.PRESSURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPressure.PSI,
    ),
    (ENTITY_DESC_KEY_PRESSURE, UnitOfPressure.INHG): SensorEntityDescription(
        key=ENTITY_DESC_KEY_PRESSURE,
        device_class=SensorDeviceClass.PRESSURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPressure.INHG,
    ),
    (ENTITY_DESC_KEY_PRESSURE, UnitOfPressure.MMHG): SensorEntityDescription(
        key=ENTITY_DESC_KEY_PRESSURE,
        device_class=SensorDeviceClass.PRESSURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPressure.MMHG,
    ),
    (
        ENTITY_DESC_KEY_SIGNAL_STRENGTH,
        SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_SIGNAL_STRENGTH,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        entity_category=EntityCategory.DIAGNOSTIC,
        entity_registry_enabled_default=False,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
    ),
    (ENTITY_DESC_KEY_TEMPERATURE, UnitOfTemperature.CELSIUS): SensorEntityDescription(
        key=ENTITY_DESC_KEY_TEMPERATURE,
        device_class=SensorDeviceClass.TEMPERATURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
    ),
    (
        ENTITY_DESC_KEY_TEMPERATURE,
        UnitOfTemperature.FAHRENHEIT,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_TEMPERATURE,
        device_class=SensorDeviceClass.TEMPERATURE,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
    ),
    (
        ENTITY_DESC_KEY_TARGET_TEMPERATURE,
        UnitOfTemperature.CELSIUS,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_TARGET_TEMPERATURE,
        device_class=SensorDeviceClass.TEMPERATURE,
        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
    ),
    (
        ENTITY_DESC_KEY_TARGET_TEMPERATURE,
        UnitOfTemperature.FAHRENHEIT,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_TARGET_TEMPERATURE,
        device_class=SensorDeviceClass.TEMPERATURE,
        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
    ),
    (
        ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
        UnitOfTime.SECONDS,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
        name="Energy production time",
        device_class=SensorDeviceClass.DURATION,
        native_unit_of_measurement=UnitOfTime.SECONDS,
    ),
    (ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME, UnitOfTime.HOURS): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_PRODUCTION_TIME,
        device_class=SensorDeviceClass.DURATION,
        native_unit_of_measurement=UnitOfTime.HOURS,
    ),
    (
        ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
        UnitOfEnergy.WATT_HOUR,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_PRODUCTION_TODAY,
        name="Energy production today",
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
    ),
    (
        ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
        UnitOfEnergy.WATT_HOUR,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_PRODUCTION_TOTAL,
        name="Energy production total",
        device_class=SensorDeviceClass.ENERGY,
        state_class=SensorStateClass.TOTAL_INCREASING,
        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
    ),
    (
        ENTITY_DESC_KEY_ENERGY_PRODUCTION_POWER,
        UnitOfPower.WATT,
    ): SensorEntityDescription(
        key=ENTITY_DESC_KEY_POWER,
        name="Energy production power",
        device_class=SensorDeviceClass.POWER,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UnitOfPower.WATT,
    ),
}
 
# These descriptions are without device class.
ENTITY_DESCRIPTION_KEY_MAP = {
    ENTITY_DESC_KEY_CO: SensorEntityDescription(
        key=ENTITY_DESC_KEY_CO,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_ENERGY_MEASUREMENT: SensorEntityDescription(
        key=ENTITY_DESC_KEY_ENERGY_MEASUREMENT,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_HUMIDITY: SensorEntityDescription(
        key=ENTITY_DESC_KEY_HUMIDITY,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_ILLUMINANCE: SensorEntityDescription(
        key=ENTITY_DESC_KEY_ILLUMINANCE,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_POWER_FACTOR: SensorEntityDescription(
        key=ENTITY_DESC_KEY_POWER_FACTOR,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_SIGNAL_STRENGTH: SensorEntityDescription(
        key=ENTITY_DESC_KEY_SIGNAL_STRENGTH,
        entity_category=EntityCategory.DIAGNOSTIC,
        entity_registry_enabled_default=False,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_MEASUREMENT: SensorEntityDescription(
        key=ENTITY_DESC_KEY_MEASUREMENT,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ENTITY_DESC_KEY_TOTAL_INCREASING: SensorEntityDescription(
        key=ENTITY_DESC_KEY_TOTAL_INCREASING,
        state_class=SensorStateClass.TOTAL_INCREASING,
    ),
    ENTITY_DESC_KEY_UV_INDEX: SensorEntityDescription(
        key=ENTITY_DESC_KEY_UV_INDEX,
        state_class=SensorStateClass.MEASUREMENT,
        native_unit_of_measurement=UV_INDEX,
    ),
}
 
 
def convert_nested_attr(
    statistics: ControllerStatistics | NodeStatistics, key: str
) -> Any:
    """Convert a string that represents a nested attr to a value."""
    data = statistics
    for _key in key.split("."):
        if data is None:
            return None  # type: ignore[unreachable]
        data = getattr(data, _key)
    return data
 
 
@dataclass(frozen=True, kw_only=True)
class ZWaveJSStatisticsSensorEntityDescription(SensorEntityDescription):
    """Class to represent a Z-Wave JS statistics sensor entity description."""
 
    convert: Callable[[ControllerStatistics | NodeStatistics, str], Any] = getattr
    entity_registry_enabled_default: bool = False
 
 
# Controller statistics descriptions
ENTITY_DESCRIPTION_CONTROLLER_STATISTICS_LIST = [
    ZWaveJSStatisticsSensorEntityDescription(
        key="messages_tx",
        translation_key="successful_messages",
        translation_placeholders={"direction": "TX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="messages_rx",
        translation_key="successful_messages",
        translation_placeholders={"direction": "RX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="messages_dropped_tx",
        translation_key="messages_dropped",
        translation_placeholders={"direction": "TX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="messages_dropped_rx",
        translation_key="messages_dropped",
        translation_placeholders={"direction": "RX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="nak", translation_key="nak", state_class=SensorStateClass.TOTAL
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="can", translation_key="can", state_class=SensorStateClass.TOTAL
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="timeout_ack",
        translation_key="timeout_ack",
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="timeout_response",
        translation_key="timeout_response",
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="timeout_callback",
        translation_key="timeout_callback",
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_0.average",
        translation_key="average_background_rssi",
        translation_placeholders={"channel": "0"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        convert=convert_nested_attr,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_0.current",
        translation_key="current_background_rssi",
        translation_placeholders={"channel": "0"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        state_class=SensorStateClass.MEASUREMENT,
        convert=convert_nested_attr,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_1.average",
        translation_key="average_background_rssi",
        translation_placeholders={"channel": "1"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        convert=convert_nested_attr,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_1.current",
        translation_key="current_background_rssi",
        translation_placeholders={"channel": "1"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        state_class=SensorStateClass.MEASUREMENT,
        convert=convert_nested_attr,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_2.average",
        translation_key="average_background_rssi",
        translation_placeholders={"channel": "2"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        convert=convert_nested_attr,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="background_rssi.channel_2.current",
        translation_key="current_background_rssi",
        translation_placeholders={"channel": "2"},
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        state_class=SensorStateClass.MEASUREMENT,
        convert=convert_nested_attr,
    ),
]
 
CONTROLLER_STATISTICS_KEY_MAP: dict[str, str] = {
    "messages_tx": "messagesTX",
    "messages_rx": "messagesRX",
    "messages_dropped_tx": "messagesDroppedTX",
    "messages_dropped_rx": "messagesDroppedRX",
    "nak": "NAK",
    "can": "CAN",
    "timeout_ack": "timeoutAck",
    "timeout_response": "timeoutResponse",
    "timeout_callback": "timeoutCallback",
    "background_rssi.channel_0.average": "backgroundRSSI.channel0.average",
    "background_rssi.channel_0.current": "backgroundRSSI.channel0.current",
    "background_rssi.channel_1.average": "backgroundRSSI.channel1.average",
    "background_rssi.channel_1.current": "backgroundRSSI.channel1.current",
    "background_rssi.channel_2.average": "backgroundRSSI.channel2.average",
    "background_rssi.channel_2.current": "backgroundRSSI.channel2.current",
}
 
# Node statistics descriptions
ENTITY_DESCRIPTION_NODE_STATISTICS_LIST = [
    ZWaveJSStatisticsSensorEntityDescription(
        key="commands_rx",
        translation_key="successful_commands",
        translation_placeholders={"direction": "RX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="commands_tx",
        translation_key="successful_commands",
        translation_placeholders={"direction": "TX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="commands_dropped_rx",
        translation_key="commands_dropped",
        translation_placeholders={"direction": "RX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="commands_dropped_tx",
        translation_key="commands_dropped",
        translation_placeholders={"direction": "TX"},
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="timeout_response",
        translation_key="timeout_response",
        state_class=SensorStateClass.TOTAL,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="rtt",
        translation_key="rtt",
        native_unit_of_measurement=UnitOfTime.MILLISECONDS,
        device_class=SensorDeviceClass.DURATION,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="rssi",
        translation_key="rssi",
        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
        state_class=SensorStateClass.MEASUREMENT,
    ),
    ZWaveJSStatisticsSensorEntityDescription(
        key="last_seen",
        translation_key="last_seen",
        device_class=SensorDeviceClass.TIMESTAMP,
        entity_registry_enabled_default=True,
    ),
]
 
NODE_STATISTICS_KEY_MAP: dict[str, str] = {
    "commands_rx": "commandsRX",
    "commands_tx": "commandsTX",
    "commands_dropped_rx": "commandsDroppedRX",
    "commands_dropped_tx": "commandsDroppedTX",
    "timeout_response": "timeoutResponse",
    "rtt": "rtt",
    "rssi": "rssi",
    "last_seen": "lastSeen",
}
 
 
def get_entity_description(
    data: NumericSensorDataTemplateData,
) -> SensorEntityDescription:
    """Return the entity description for the given data."""
    data_description_key = data.entity_description_key or ""
    data_unit = data.unit_of_measurement or ""
    return ENTITY_DESCRIPTION_KEY_DEVICE_CLASS_MAP.get(
        (data_description_key, data_unit),
        ENTITY_DESCRIPTION_KEY_MAP.get(
            data_description_key,
            SensorEntityDescription(
                key="base_sensor", native_unit_of_measurement=data.unit_of_measurement
            ),
        ),
    )
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up Z-Wave sensor from config entry."""
    client: ZwaveClient = config_entry.runtime_data[DATA_CLIENT]
    driver = client.driver
    assert driver is not None  # Driver is ready before platforms are loaded.
 
    @callback
    def async_add_sensor(info: ZwaveDiscoveryInfo) -> None:
        """Add Z-Wave Sensor."""
        entities: list[ZWaveBaseEntity] = []
 
        if info.platform_data:
            data: NumericSensorDataTemplateData = info.platform_data
        else:
            data = NumericSensorDataTemplateData()
 
        entity_description = get_entity_description(data)
 
        if info.platform_hint == "numeric_sensor":
            entities.append(
                ZWaveNumericSensor(
                    config_entry,
                    driver,
                    info,
                    entity_description,
                    data.unit_of_measurement,
                )
            )
        elif info.platform_hint == "notification":
            # prevent duplicate entities for values that are already represented as binary sensors
            if is_valid_notification_binary_sensor(info):
                return
            entities.append(
                ZWaveListSensor(config_entry, driver, info, entity_description)
            )
        elif info.platform_hint == "config_parameter":
            entities.append(
                ZWaveConfigParameterSensor(
                    config_entry, driver, info, entity_description
                )
            )
        elif info.platform_hint == "meter":
            entities.append(
                ZWaveMeterSensor(config_entry, driver, info, entity_description)
            )
        else:
            entities.append(ZwaveSensor(config_entry, driver, info, entity_description))
 
        async_add_entities(entities)
 
    @callback
    def async_add_controller_status_sensor() -> None:
        """Add controller status sensor."""
        async_add_entities([ZWaveControllerStatusSensor(config_entry, driver)])
 
    @callback
    def async_add_node_status_sensor(node: ZwaveNode) -> None:
        """Add node status sensor."""
        async_add_entities([ZWaveNodeStatusSensor(config_entry, driver, node)])
 
    @callback
    def async_add_statistics_sensors(node: ZwaveNode) -> None:
        """Add statistics sensors."""
        async_migrate_statistics_sensors(
            hass,
            driver,
            node,
            CONTROLLER_STATISTICS_KEY_MAP
            if driver.controller.own_node == node
            else NODE_STATISTICS_KEY_MAP,
        )
        async_add_entities(
            [
                ZWaveStatisticsSensor(
                    config_entry,
                    driver,
                    driver.controller if driver.controller.own_node == node else node,
                    entity_description,
                )
                for entity_description in (
                    ENTITY_DESCRIPTION_CONTROLLER_STATISTICS_LIST
                    if driver.controller.own_node == node
                    else ENTITY_DESCRIPTION_NODE_STATISTICS_LIST
                )
            ]
        )
 
    config_entry.async_on_unload(
        async_dispatcher_connect(
            hass,
            f"{DOMAIN}_{config_entry.entry_id}_add_{SENSOR_DOMAIN}",
            async_add_sensor,
        )
    )
 
    config_entry.async_on_unload(
        async_dispatcher_connect(
            hass,
            f"{DOMAIN}_{config_entry.entry_id}_add_controller_status_sensor",
            async_add_controller_status_sensor,
        )
    )
 
    config_entry.async_on_unload(
        async_dispatcher_connect(
            hass,
            f"{DOMAIN}_{config_entry.entry_id}_add_node_status_sensor",
            async_add_node_status_sensor,
        )
    )
 
    config_entry.async_on_unload(
        async_dispatcher_connect(
            hass,
            f"{DOMAIN}_{config_entry.entry_id}_add_statistics_sensors",
            async_add_statistics_sensors,
        )
    )
 
    platform = entity_platform.async_get_current_platform()
    platform.async_register_entity_service(
        SERVICE_RESET_METER,
        {
            vol.Optional(ATTR_METER_TYPE): vol.Coerce(int),
            vol.Optional(ATTR_VALUE): vol.Coerce(int),
        },
        "async_reset_meter",
    )
 
 
class ZwaveSensor(ZWaveBaseEntity, SensorEntity):
    """Basic Representation of a Z-Wave sensor."""
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        driver: Driver,
        info: ZwaveDiscoveryInfo,
        entity_description: SensorEntityDescription,
        unit_of_measurement: str | None = None,
    ) -> None:
        """Initialize a ZWaveSensorBase entity."""
        self.entity_descriptionentity_description = entity_description
        super().__init__(config_entry, driver, info)
        self._attr_native_unit_of_measurement_attr_native_unit_of_measurement = unit_of_measurement
 
        # Entity class attributes
        self._attr_force_update_attr_force_update = True
        if not entity_description.name or entity_description.name is UNDEFINED:
            self._attr_name_attr_name_attr_name = self.generate_namegenerate_name(include_value_name=True)
 
    @property
    def native_value(self) -> StateType:
        """Return state of the sensor."""
        key = str(self.infoinfo.primary_value.value)
        if key not in self.infoinfo.primary_value.metadata.states:
            return self.infoinfo.primary_value.value
        return str(self.infoinfo.primary_value.metadata.states[key])
 
    @property
    def native_unit_of_measurement(self) -> str | None:
        """Return unit of measurement the value is expressed in."""
        if (unit := super().native_unit_of_measurement) is not None:
            return unit
        if self.infoinfo.primary_value.metadata.unit is None:
            return None
        return str(self.infoinfo.primary_value.metadata.unit)
 
 
class ZWaveNumericSensor(ZwaveSensor):
    """Representation of a Z-Wave Numeric sensor."""
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        driver: Driver,
        info: ZwaveDiscoveryInfo,
        entity_description: SensorEntityDescription,
        unit_of_measurement: str | None = None,
    ) -> None:
        """Initialize a ZWaveBasicSensor entity."""
        super().__init__(
            config_entry, driver, info, entity_description, unit_of_measurement
        )
        if self.infoinfo.primary_value.command_class == CommandClass.BASIC:
            self._attr_name_attr_name_attr_name_attr_name = self.generate_namegenerate_name(
                include_value_name=True, alternate_value_name="Basic"
            )
 
    @callback
    def on_value_update(self) -> None:
        """Handle scale changes for this value on value updated event."""
        data = NumericSensorDataTemplate().resolve_data(self.infoinfo.primary_value)
        self.entity_descriptionentity_descriptionentity_description = get_entity_description(data)
        self._attr_native_unit_of_measurement_attr_native_unit_of_measurement_attr_native_unit_of_measurement = data.unit_of_measurement
 
    @property
    def native_value(self) -> float:
        """Return state of the sensor."""
        if self.infoinfo.primary_value.value is None:
            return 0
        return float(self.infoinfo.primary_value.value)
 
 
class ZWaveMeterSensor(ZWaveNumericSensor):
    """Representation of a Z-Wave Meter CC sensor."""
 
    @property
    def extra_state_attributes(self) -> Mapping[str, int | str] | None:
        """Return extra state attributes."""
        meter_type = get_meter_type(self.infoinfo.primary_value)
        return {
            ATTR_METER_TYPE: meter_type.value,
            ATTR_METER_TYPE_NAME: meter_type.name,
        }
 
    async def async_reset_meter(
        self, meter_type: int | None = None, value: int | None = None
    ) -> None:
        """Reset meter(s) on device."""
        node = self.infoinfo.node
        endpoint = self.infoinfo.primary_value.endpoint or 0
        options = {}
        if meter_type is not None:
            options[RESET_METER_OPTION_TYPE] = meter_type
        if value is not None:
            options[RESET_METER_OPTION_TARGET_VALUE] = value
        args = [options] if options else []
        try:
            await node.endpoints[endpoint].async_invoke_cc_api(
                CommandClass.METER, "reset", *args, wait_for_result=False
            )
        except BaseZwaveJSServerError as err:
            raise HomeAssistantError(
                f"Failed to reset meters on node {node} endpoint {endpoint}: {err}"
            ) from err
        LOGGER.debug(
            "Meters on node %s endpoint %s reset with the following options: %s",
            node,
            endpoint,
            options,
        )
 
 
class ZWaveListSensor(ZwaveSensor):
    """Representation of a Z-Wave Numeric sensor with multiple states."""
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        driver: Driver,
        info: ZwaveDiscoveryInfo,
        entity_description: SensorEntityDescription,
        unit_of_measurement: str | None = None,
    ) -> None:
        """Initialize a ZWaveListSensor entity."""
        super().__init__(
            config_entry, driver, info, entity_description, unit_of_measurement
        )
 
        # Entity class attributes
        # Notification sensors have the following name mapping (variables are property
        # keys, name is property)
        # https://github.com/zwave-js/node-zwave-js/blob/master/packages/config/config/notifications.json
        self._attr_name_attr_name_attr_name_attr_name = self.generate_namegenerate_name(
            alternate_value_name=self.infoinfo.primary_value.property_name,
            additional_info=[self.infoinfo.primary_value.property_key_name],
        )
        if self.infoinfo.primary_value.metadata.states:
            self._attr_device_class_attr_device_class = SensorDeviceClass.ENUM
            self._attr_options_attr_options = list(info.primary_value.metadata.states.values())
 
    @property
    def extra_state_attributes(self) -> dict[str, str] | None:
        """Return the device specific state attributes."""
        if (value := self.infoinfo.primary_value.value) is None:
            return None
        # add the value's int value as property for multi-value (list) items
        return {ATTR_VALUE: value}
 
 
class ZWaveConfigParameterSensor(ZWaveListSensor):
    """Representation of a Z-Wave config parameter sensor."""
 
    _attr_entity_category = EntityCategory.DIAGNOSTIC
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        driver: Driver,
        info: ZwaveDiscoveryInfo,
        entity_description: SensorEntityDescription,
        unit_of_measurement: str | None = None,
    ) -> None:
        """Initialize a ZWaveConfigParameterSensor entity."""
        super().__init__(
            config_entry, driver, info, entity_description, unit_of_measurement
        )
 
        property_key_name = self.infoinfo.primary_value.property_key_name
        # Entity class attributes
        self._attr_name_attr_name_attr_name_attr_name_attr_name = self.generate_namegenerate_name(
            alternate_value_name=self.infoinfo.primary_value.property_name,
            additional_info=[property_key_name] if property_key_name else None,
        )
 
    @property
    def extra_state_attributes(self) -> dict[str, str] | None:
        """Return the device specific state attributes."""
        if (value := self.infoinfo.primary_value.value) is None:
            return None
        # add the value's int value as property for multi-value (list) items
        return {ATTR_VALUE: value}
 
 
class ZWaveNodeStatusSensor(SensorEntity):
    """Representation of a node status sensor."""
 
    _attr_should_poll = False
    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_has_entity_name = True
    _attr_translation_key = "node_status"
 
    def __init__(
        self, config_entry: ConfigEntry, driver: Driver, node: ZwaveNode
    ) -> None:
        """Initialize a generic Z-Wave device entity."""
        self.config_entryconfig_entry = config_entry
        self.nodenode = node
 
        # Entity class attributes
        self._base_unique_id_base_unique_id = get_valueless_base_unique_id(driver, node)
        self._attr_unique_id_attr_unique_id = f"{self._base_unique_id}.node_status"
        # device may not be precreated in main handler yet
        self._attr_device_info_attr_device_info = get_device_info(driver, node)
 
    async def async_poll_value(self, _: bool) -> None:
        """Poll a value."""
        # We log an error instead of raising an exception because this service call occurs
        # in a separate task since it is called via the dispatcher and we don't want to
        # raise the exception in that separate task because it is confusing to the user.
        LOGGER.error(
            "There is no value to refresh for this entity so the zwave_js.refresh_value"
            " service won't work for it"
        )
 
    @callback
    def _status_changed(self, _: dict) -> None:
        """Call when status event is received."""
        self._attr_native_value_attr_native_value = self.nodenode.status.name.lower()
        self.async_write_ha_stateasync_write_ha_state()
 
    async def async_added_to_hass(self) -> None:
        """Call when entity is added."""
        # Add value_changed callbacks.
        for evt in ("wake up", "sleep", "dead", "alive"):
            self.async_on_removeasync_on_remove(self.nodenode.on(evt, self._status_changed_status_changed))
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self.unique_id}_poll_value",
                self.async_poll_valueasync_poll_value,
            )
        )
        # we don't listen for `remove_entity_on_ready_node` signal because this entity
        # is created when the node is added which occurs before ready. It only needs to
        # be removed if the node is removed from the network.
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self._base_unique_id}_remove_entity",
                self.async_removeasync_remove,
            )
        )
        self._attr_native_value_attr_native_value: str = self.nodenode.status.name.lower()
        self.async_write_ha_stateasync_write_ha_state()
 
 
class ZWaveControllerStatusSensor(SensorEntity):
    """Representation of a controller status sensor."""
 
    _attr_should_poll = False
    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_has_entity_name = True
    _attr_translation_key = "controller_status"
 
    def __init__(self, config_entry: ConfigEntry, driver: Driver) -> None:
        """Initialize a generic Z-Wave device entity."""
        self.config_entryconfig_entry = config_entry
        self.controllercontroller = driver.controller
        node = self.controllercontroller.own_node
        assert node
 
        # Entity class attributes
        self._base_unique_id_base_unique_id = get_valueless_base_unique_id(driver, node)
        self._attr_unique_id_attr_unique_id = f"{self._base_unique_id}.controller_status"
        # device may not be precreated in main handler yet
        self._attr_device_info_attr_device_info = get_device_info(driver, node)
 
    async def async_poll_value(self, _: bool) -> None:
        """Poll a value."""
        # We log an error instead of raising an exception because this service call occurs
        # in a separate task since it is called via the dispatcher and we don't want to
        # raise the exception in that separate task because it is confusing to the user.
        LOGGER.error(
            "There is no value to refresh for this entity so the zwave_js.refresh_value"
            " service won't work for it"
        )
 
    @callback
    def _status_changed(self, _: dict) -> None:
        """Call when status event is received."""
        self._attr_native_value_attr_native_value = self.controllercontroller.status.name.lower()
        self.async_write_ha_stateasync_write_ha_state()
 
    async def async_added_to_hass(self) -> None:
        """Call when entity is added."""
        # Add value_changed callbacks.
        self.async_on_removeasync_on_remove(self.controllercontroller.on("status changed", self._status_changed_status_changed))
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self.unique_id}_poll_value",
                self.async_poll_valueasync_poll_value,
            )
        )
        # we don't listen for `remove_entity_on_ready_node` signal because this is not
        # a regular node
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self._base_unique_id}_remove_entity",
                self.async_removeasync_remove,
            )
        )
        self._attr_native_value_attr_native_value: str = self.controllercontroller.status.name.lower()
 
 
class ZWaveStatisticsSensor(SensorEntity):
    """Representation of a node/controller statistics sensor."""
 
    entity_description: ZWaveJSStatisticsSensorEntityDescription
    _attr_should_poll = False
    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_has_entity_name = True
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        driver: Driver,
        statistics_src: ZwaveNode | Controller,
        description: ZWaveJSStatisticsSensorEntityDescription,
    ) -> None:
        """Initialize a Z-Wave statistics entity."""
        self.entity_descriptionentity_description = description
        self.config_entryconfig_entry = config_entry
        self.statistics_srcstatistics_src = statistics_src
        node = (
            statistics_src.own_node
            if isinstance(statistics_src, Controller)
            else statistics_src
        )
        assert node
 
        # Entity class attributes
        self._base_unique_id_base_unique_id = get_valueless_base_unique_id(driver, node)
        self._attr_unique_id_attr_unique_id = f"{self._base_unique_id}.statistics_{description.key}"
        # device may not be precreated in main handler yet
        self._attr_device_info_attr_device_info = get_device_info(driver, node)
 
    async def async_poll_value(self, _: bool) -> None:
        """Poll a value."""
        # We log an error instead of raising an exception because this service call occurs
        # in a separate task since it is called via the dispatcher and we don't want to
        # raise the exception in that separate task because it is confusing to the user.
        LOGGER.error(
            "There is no value to refresh for this entity so the zwave_js.refresh_value"
            " service won't work for it"
        )
 
    @callback
    def statistics_updated(self, event_data: dict) -> None:
        """Call when statistics updated event is received."""
        self._attr_native_value_attr_native_value = self.entity_descriptionentity_description.convert(
            event_data["statistics_updated"], self.entity_descriptionentity_description.key
        )
        self.async_write_ha_stateasync_write_ha_state()
 
    async def async_added_to_hass(self) -> None:
        """Call when entity is added."""
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self.unique_id}_poll_value",
                self.async_poll_valueasync_poll_value,
            )
        )
        self.async_on_removeasync_on_remove(
            async_dispatcher_connect(
                self.hasshass,
                f"{DOMAIN}_{self._base_unique_id}_remove_entity",
                self.async_removeasync_remove,
            )
        )
        self.async_on_removeasync_on_remove(
            self.statistics_srcstatistics_src.on("statistics updated", self.statistics_updatedstatistics_updated)
        )
 
        # Set initial state
        self._attr_native_value_attr_native_value = self.entity_descriptionentity_description.convert(
            self.statistics_srcstatistics_src.statistics, self.entity_descriptionentity_description.key
        )