1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-22 02:47:14 +00:00
Files
core/homeassistant/components/yolink/sensor.py
2026-01-12 10:33:49 +01:00

462 lines
16 KiB
Python

"""YoLink Sensor."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from yolink.const import (
ATTR_DEVICE_CO_SMOKE_SENSOR,
ATTR_DEVICE_DIMMER,
ATTR_DEVICE_DOOR_SENSOR,
ATTR_DEVICE_FINGER,
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_LOCK,
ATTR_DEVICE_LOCK_V2,
ATTR_DEVICE_MANIPULATOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
ATTR_DEVICE_MULTI_OUTLET,
ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER,
ATTR_DEVICE_OUTLET,
ATTR_DEVICE_POWER_FAILURE_ALARM,
ATTR_DEVICE_SIREN,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_SMOKE_ALARM,
ATTR_DEVICE_SOIL_TH_SENSOR,
ATTR_DEVICE_SPRINKLER,
ATTR_DEVICE_SPRINKLER_V2,
ATTR_DEVICE_SWITCH,
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_THERMOSTAT,
ATTR_DEVICE_VIBRATION_SENSOR,
ATTR_DEVICE_WATER_DEPTH_SENSOR,
ATTR_DEVICE_WATER_METER_CONTROLLER,
ATTR_GARAGE_DOOR_CONTROLLER,
)
from yolink.device import YoLinkDevice
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
UnitOfConductivity,
UnitOfEnergy,
UnitOfLength,
UnitOfPower,
UnitOfTemperature,
UnitOfVolume,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import percentage
from .const import (
DEV_MODEL_PLUG_YS6602_EC,
DEV_MODEL_PLUG_YS6602_UC,
DEV_MODEL_PLUG_YS6614_EC,
DEV_MODEL_PLUG_YS6614_UC,
DEV_MODEL_PLUG_YS6803_EC,
DEV_MODEL_PLUG_YS6803_UC,
DEV_MODEL_TH_SENSOR_YS8004_EC,
DEV_MODEL_TH_SENSOR_YS8004_UC,
DEV_MODEL_TH_SENSOR_YS8008_EC,
DEV_MODEL_TH_SENSOR_YS8008_UC,
DEV_MODEL_TH_SENSOR_YS8014_EC,
DEV_MODEL_TH_SENSOR_YS8014_UC,
DEV_MODEL_TH_SENSOR_YS8017_EC,
DEV_MODEL_TH_SENSOR_YS8017_UC,
DOMAIN,
)
from .coordinator import YoLinkCoordinator
from .entity import YoLinkEntity
@dataclass(frozen=True, kw_only=True)
class YoLinkSensorEntityDescription(SensorEntityDescription):
"""YoLink SensorEntityDescription."""
exists_fn: Callable[[YoLinkDevice], bool] = lambda _: True
should_update_entity: Callable = lambda state: True
value: Callable[[YoLinkDevice, dict], Any | None] = lambda device, state: state
SENSOR_DEVICE_TYPE = [
ATTR_DEVICE_DIMMER,
ATTR_DEVICE_DOOR_SENSOR,
ATTR_DEVICE_FINGER,
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_MULTI_OUTLET,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_OUTLET,
ATTR_DEVICE_POWER_FAILURE_ALARM,
ATTR_DEVICE_SIREN,
ATTR_DEVICE_SWITCH,
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_THERMOSTAT,
ATTR_DEVICE_VIBRATION_SENSOR,
ATTR_DEVICE_WATER_DEPTH_SENSOR,
ATTR_DEVICE_WATER_METER_CONTROLLER,
ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER,
ATTR_DEVICE_LOCK,
ATTR_DEVICE_LOCK_V2,
ATTR_DEVICE_MANIPULATOR,
ATTR_DEVICE_CO_SMOKE_SENSOR,
ATTR_GARAGE_DOOR_CONTROLLER,
ATTR_DEVICE_SOIL_TH_SENSOR,
ATTR_DEVICE_SMOKE_ALARM,
ATTR_DEVICE_SPRINKLER,
ATTR_DEVICE_SPRINKLER_V2,
ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
]
BATTERY_POWER_SENSOR = [
ATTR_DEVICE_DOOR_SENSOR,
ATTR_DEVICE_FINGER,
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_POWER_FAILURE_ALARM,
ATTR_DEVICE_SIREN,
ATTR_DEVICE_SMART_REMOTER,
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_VIBRATION_SENSOR,
ATTR_DEVICE_LOCK,
ATTR_DEVICE_LOCK_V2,
ATTR_DEVICE_MANIPULATOR,
ATTR_DEVICE_CO_SMOKE_SENSOR,
ATTR_DEVICE_WATER_DEPTH_SENSOR,
ATTR_DEVICE_WATER_METER_CONTROLLER,
ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER,
ATTR_DEVICE_SOIL_TH_SENSOR,
ATTR_DEVICE_SMOKE_ALARM,
ATTR_DEVICE_SPRINKLER_V2,
ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
]
MCU_DEV_TEMPERATURE_SENSOR = [
ATTR_DEVICE_LEAK_SENSOR,
ATTR_DEVICE_MOTION_SENSOR,
ATTR_DEVICE_CO_SMOKE_SENSOR,
ATTR_DEVICE_SMOKE_ALARM,
]
NONE_HUMIDITY_SENSOR_MODELS = [
DEV_MODEL_TH_SENSOR_YS8004_EC,
DEV_MODEL_TH_SENSOR_YS8004_UC,
DEV_MODEL_TH_SENSOR_YS8008_EC,
DEV_MODEL_TH_SENSOR_YS8008_UC,
DEV_MODEL_TH_SENSOR_YS8014_EC,
DEV_MODEL_TH_SENSOR_YS8014_UC,
DEV_MODEL_TH_SENSOR_YS8017_UC,
DEV_MODEL_TH_SENSOR_YS8017_EC,
]
POWER_SUPPORT_MODELS = [
DEV_MODEL_PLUG_YS6602_UC,
DEV_MODEL_PLUG_YS6602_EC,
DEV_MODEL_PLUG_YS6614_UC,
DEV_MODEL_PLUG_YS6614_EC,
DEV_MODEL_PLUG_YS6803_UC,
DEV_MODEL_PLUG_YS6803_EC,
]
def parse_data_battery(device: YoLinkDevice, data: dict) -> int | None:
"""Parse battery data."""
if (val := data.get("battery")) is None:
return None
if val > 0:
return percentage.ordered_list_item_to_percentage([1, 2, 3, 4], val)
return 0
def parse_data_volume(device: YoLinkDevice, data: dict) -> str | None:
"""Parse volume data."""
if (val := data.get("volume")) is None:
return None
volume_level = {1: "low", 2: "medium", 3: "high"}
return volume_level.get(val)
def parse_data_humidity(device: YoLinkDevice, data: dict) -> int | None:
"""Parse humidity data."""
if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR:
return (
state.get("humidity") if (state := data.get("state")) is not None else None
)
return data.get("humidity")
def parse_data_temperature(device: YoLinkDevice, data: dict) -> float | None:
"""Parse temperature data."""
if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR:
return (
state.get("temperature")
if (state := data.get("state")) is not None
else None
)
return data.get("temperature")
SENSOR_TYPES: tuple[YoLinkSensorEntityDescription, ...] = (
YoLinkSensorEntityDescription(
key="battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: device.device_type in BATTERY_POWER_SENSOR,
should_update_entity=lambda value: value is not None,
value=parse_data_battery,
),
YoLinkSensorEntityDescription(
key="humidity",
device_class=SensorDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: (
device.device_type
in [
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_SOIL_TH_SENSOR,
ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
]
and device.device_model_name not in NONE_HUMIDITY_SENSOR_MODELS
),
value=parse_data_humidity,
),
YoLinkSensorEntityDescription(
key="temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: (
device.device_type
in [
ATTR_DEVICE_TH_SENSOR,
ATTR_DEVICE_SOIL_TH_SENSOR,
ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR,
]
),
value=parse_data_temperature,
),
# mcu temperature
YoLinkSensorEntityDescription(
key="devTemperature",
translation_key="device_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
exists_fn=lambda device: device.device_type in MCU_DEV_TEMPERATURE_SENSOR,
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("devTemperature"),
),
YoLinkSensorEntityDescription(
key="loraInfo",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
should_update_entity=lambda value: value is not None,
value=lambda device, data: (
loraData.get("signal")
if (loraData := data.get("loraInfo")) is not None
else None
),
),
YoLinkSensorEntityDescription(
key="state",
translation_key="power_failure_alarm",
device_class=SensorDeviceClass.ENUM,
options=["normal", "alert", "off"],
exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM,
value=lambda device, data: data.get("state"),
),
YoLinkSensorEntityDescription(
key="mute",
translation_key="power_failure_alarm_mute",
device_class=SensorDeviceClass.ENUM,
options=["muted", "unmuted"],
exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM,
value=lambda device, data: "muted" if data.get("mute") is True else "unmuted",
),
YoLinkSensorEntityDescription(
key="sound",
translation_key="power_failure_alarm_volume",
device_class=SensorDeviceClass.ENUM,
options=["low", "medium", "high"],
exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM,
value=parse_data_volume,
),
YoLinkSensorEntityDescription(
key="beep",
translation_key="power_failure_alarm_beep",
device_class=SensorDeviceClass.ENUM,
options=["enabled", "disabled"],
exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM,
value=lambda device, data: (
"enabled" if data.get("beep") is True else "disabled"
),
),
YoLinkSensorEntityDescription(
key="waterDepth",
device_class=SensorDeviceClass.DISTANCE,
native_unit_of_measurement=UnitOfLength.METERS,
exists_fn=lambda device: device.device_type == ATTR_DEVICE_WATER_DEPTH_SENSOR,
value=lambda device, data: data.get("waterDepth"),
),
YoLinkSensorEntityDescription(
key="meter_reading",
translation_key="water_meter_reading",
device_class=SensorDeviceClass.WATER,
native_unit_of_measurement=UnitOfVolume.CUBIC_METERS,
state_class=SensorStateClass.TOTAL_INCREASING,
exists_fn=lambda device: (
device.device_type == ATTR_DEVICE_WATER_METER_CONTROLLER
),
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("meter_reading"),
),
YoLinkSensorEntityDescription(
key="meter_1_reading",
translation_key="water_meter_1_reading",
device_class=SensorDeviceClass.WATER,
native_unit_of_measurement=UnitOfVolume.CUBIC_METERS,
state_class=SensorStateClass.TOTAL_INCREASING,
exists_fn=lambda device: (
device.device_type == ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER
),
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("meter_1_reading"),
),
YoLinkSensorEntityDescription(
key="meter_2_reading",
translation_key="water_meter_2_reading",
device_class=SensorDeviceClass.WATER,
native_unit_of_measurement=UnitOfVolume.CUBIC_METERS,
state_class=SensorStateClass.TOTAL_INCREASING,
exists_fn=lambda device: (
device.device_type == ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER
),
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("meter_2_reading"),
),
YoLinkSensorEntityDescription(
key="power",
translation_key="current_power",
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: device.device_model_name in POWER_SUPPORT_MODELS,
should_update_entity=lambda value: value is not None,
value=lambda device, data: (
value / 10 if (value := data.get("power")) is not None else None
),
),
YoLinkSensorEntityDescription(
key="watt",
translation_key="power_consumption",
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
state_class=SensorStateClass.TOTAL,
exists_fn=lambda device: device.device_model_name in POWER_SUPPORT_MODELS,
should_update_entity=lambda value: value is not None,
value=lambda device, data: (
value / 100 if (value := data.get("watt")) is not None else None
),
),
YoLinkSensorEntityDescription(
key="conductivity",
device_class=SensorDeviceClass.CONDUCTIVITY,
native_unit_of_measurement=UnitOfConductivity.MICROSIEMENS_PER_CM,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: device.device_type in [ATTR_DEVICE_SOIL_TH_SENSOR],
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("conductivity"),
),
YoLinkSensorEntityDescription(
key="coreTemperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
exists_fn=lambda device: (
device.device_model_name
in [DEV_MODEL_PLUG_YS6614_EC, DEV_MODEL_PLUG_YS6614_UC]
),
should_update_entity=lambda value: value is not None,
value=lambda device, data: data.get("coreTemperature"),
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up YoLink Sensor from a config entry."""
device_coordinators = hass.data[DOMAIN][config_entry.entry_id].device_coordinators
sensor_device_coordinators = [
device_coordinator
for device_coordinator in device_coordinators.values()
if device_coordinator.device.device_type in SENSOR_DEVICE_TYPE
]
async_add_entities(
YoLinkSensorEntity(
config_entry,
sensor_device_coordinator,
description,
)
for sensor_device_coordinator in sensor_device_coordinators
for description in SENSOR_TYPES
if description.exists_fn(sensor_device_coordinator.device)
)
class YoLinkSensorEntity(YoLinkEntity, SensorEntity):
"""YoLink Sensor Entity."""
entity_description: YoLinkSensorEntityDescription
def __init__(
self,
config_entry: ConfigEntry,
coordinator: YoLinkCoordinator,
description: YoLinkSensorEntityDescription,
) -> None:
"""Init YoLink Sensor."""
super().__init__(config_entry, coordinator)
self.entity_description = description
self._attr_unique_id = (
f"{coordinator.device.device_id} {self.entity_description.key}"
)
@callback
def update_entity_state(self, state: dict) -> None:
"""Update HA Entity State."""
if (
attr_val := self.entity_description.value(
self.coordinator.device,
state,
)
) is None and self.entity_description.should_update_entity(attr_val) is False:
return
self._attr_native_value = attr_val
self.async_write_ha_state()
@property
def available(self) -> bool:
"""Return true is device is available."""
return super().available and self.coordinator.dev_online