diff --git a/homeassistant/components/yolink/binary_sensor.py b/homeassistant/components/yolink/binary_sensor.py index d57e942734e38e..cfec02ca3e2ce1 100644 --- a/homeassistant/components/yolink/binary_sensor.py +++ b/homeassistant/components/yolink/binary_sensor.py @@ -4,13 +4,13 @@ from collections.abc import Callable from dataclasses import dataclass -from typing import Any from yolink.const import ( ATTR_DEVICE_CO_SMOKE_SENSOR, ATTR_DEVICE_DOOR_SENSOR, ATTR_DEVICE_LEAK_SENSOR, ATTR_DEVICE_MOTION_SENSOR, + ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR, ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER, ATTR_DEVICE_SMOKE_ALARM, ATTR_DEVICE_VIBRATION_SENSOR, @@ -36,20 +36,21 @@ from .entity import YoLinkEntity -@dataclass(frozen=True) +@dataclass(frozen=True, kw_only=True) class YoLinkBinarySensorEntityDescription(BinarySensorEntityDescription): """YoLink BinarySensorEntityDescription.""" exists_fn: Callable[[YoLinkDevice], bool] = lambda _: True - state_key: str = "state" - value: Callable[[Any], bool | None] = lambda _: None - should_update_entity: Callable = lambda state: True + should_update_entity: Callable = lambda _: True + value: Callable[[YoLinkDevice, dict], bool | None] + is_available: Callable[[YoLinkDevice, dict], bool] = lambda _, __: True SENSOR_DEVICE_TYPE = [ ATTR_DEVICE_DOOR_SENSOR, ATTR_DEVICE_MOTION_SENSOR, ATTR_DEVICE_LEAK_SENSOR, + ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR, ATTR_DEVICE_VIBRATION_SENSOR, ATTR_DEVICE_CO_SMOKE_SENSOR, ATTR_DEVICE_WATER_METER_CONTROLLER, @@ -58,52 +59,95 @@ class YoLinkBinarySensorEntityDescription(BinarySensorEntityDescription): ] +def parse_data_leak_sensor_state(device: YoLinkDevice, data: dict) -> bool | None: + """Parse leak sensor state.""" + if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR: + if (state := data.get("state")) is None or ( + value := state.get("waterDetection") + ) is None: + return None + return ( + value == "normal" + if data.get("waterDetectionMode") == "WaterPeak" + else value == "leak" + ) + return data.get("state") in ["alert", "full"] + + +def is_leak_sensor_state_available(device: YoLinkDevice, data: dict) -> bool: + """Check leak sensor state availability.""" + if device.device_type == ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR: + if ( + (alarms := data.get("alarm")) is not None + and isinstance(alarms, list) + and "Alert.DetectorError" in alarms + ): + return False + if (alarms := data.get("alarmState")) is not None and alarms.get( + "detectorError" + ) is True: + return False + return True + + SENSOR_TYPES: tuple[YoLinkBinarySensorEntityDescription, ...] = ( YoLinkBinarySensorEntityDescription( key="door_state", device_class=BinarySensorDeviceClass.DOOR, - value=lambda value: value == "open" if value is not None else None, exists_fn=lambda device: device.device_type == ATTR_DEVICE_DOOR_SENSOR, + value=lambda device, data: ( + value == "open" if (value := data.get("state")) is not None else None + ), ), YoLinkBinarySensorEntityDescription( key="motion_state", device_class=BinarySensorDeviceClass.MOTION, - value=lambda value: value == "alert" if value is not None else None, exists_fn=lambda device: device.device_type == ATTR_DEVICE_MOTION_SENSOR, + value=lambda device, data: ( + value == "alert" if (value := data.get("state")) is not None else None + ), ), YoLinkBinarySensorEntityDescription( key="leak_state", device_class=BinarySensorDeviceClass.MOISTURE, - value=lambda value: value in ("alert", "full") if value is not None else None, - exists_fn=lambda device: device.device_type == ATTR_DEVICE_LEAK_SENSOR, + exists_fn=lambda device: ( + device.device_type + in [ATTR_DEVICE_LEAK_SENSOR, ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR] + ), + value=parse_data_leak_sensor_state, + is_available=is_leak_sensor_state_available, ), YoLinkBinarySensorEntityDescription( key="vibration_state", device_class=BinarySensorDeviceClass.VIBRATION, - value=lambda value: value == "alert" if value is not None else None, exists_fn=lambda device: device.device_type == ATTR_DEVICE_VIBRATION_SENSOR, + value=lambda device, data: ( + value == "alert" if (value := data.get("state")) is not None else None + ), ), YoLinkBinarySensorEntityDescription( key="co_detected", device_class=BinarySensorDeviceClass.CO, - value=lambda state: state.get("gasAlarm"), exists_fn=lambda device: device.device_type == ATTR_DEVICE_CO_SMOKE_SENSOR, + value=lambda device, data: ( + state.get("gasAlarm") if (state := data.get("state")) is not None else None + ), ), YoLinkBinarySensorEntityDescription( key="smoke_detected", device_class=BinarySensorDeviceClass.SMOKE, - value=lambda state: state.get("smokeAlarm") is True - or state.get("denseSmokeAlarm") is True, - exists_fn=lambda device: device.device_type - in [ATTR_DEVICE_CO_SMOKE_SENSOR, ATTR_DEVICE_SMOKE_ALARM], + exists_fn=lambda device: ( + device.device_type in [ATTR_DEVICE_CO_SMOKE_SENSOR, ATTR_DEVICE_SMOKE_ALARM] + ), + value=lambda device, data: ( + state.get("smokeAlarm") is True or state.get("denseSmokeAlarm") is True + if (state := data.get("state")) is not None + else None + ), ), YoLinkBinarySensorEntityDescription( key="pipe_leak_detected", - state_key="alarm", device_class=BinarySensorDeviceClass.MOISTURE, - value=lambda state: state.get("leak") if state is not None else None, - # This property will be lost during valve operation. - should_update_entity=lambda value: value is not None, exists_fn=lambda device: ( device.device_type in [ @@ -111,17 +155,26 @@ class YoLinkBinarySensorEntityDescription(BinarySensorEntityDescription): ATTR_DEVICE_MULTI_WATER_METER_CONTROLLER, ] ), + # This property will be lost during valve operation. + should_update_entity=lambda value: value is not None, + value=lambda device, data: ( + alarms.get("leak") if (alarms := data.get("alarm")) is not None else None + ), ), YoLinkBinarySensorEntityDescription( key="water_running", translation_key="water_running", - value=lambda state: state.get("waterFlowing") if state is not None else None, - should_update_entity=lambda value: value is not None, exists_fn=lambda device: ( device.device_type == ATTR_DEVICE_WATER_METER_CONTROLLER and device.device_model_name in [DEV_MODEL_WATER_METER_YS5018_EC, DEV_MODEL_WATER_METER_YS5018_UC] ), + should_update_entity=lambda value: value is not None, + value=lambda device, data: ( + state.get("waterFlowing") + if (state := data.get("state")) is not None + else None + ), ), ) @@ -167,15 +220,17 @@ def __init__( ) @callback - def update_entity_state(self, state: dict[str, Any]) -> None: + def update_entity_state(self, state: dict) -> None: """Update HA Entity State.""" if ( - _attr_val := self.entity_description.value( - state.get(self.entity_description.state_key) - ) + _attr_val := self.entity_description.value(self.coordinator.device, state) ) is None or self.entity_description.should_update_entity(_attr_val) is False: return - self._attr_is_on = _attr_val + _is_attr_available = self.entity_description.is_available( + self.coordinator.device, state + ) + self._attr_available = _is_attr_available + self._attr_is_on = _attr_val if _is_attr_available else None self.async_write_ha_state() @property diff --git a/homeassistant/components/yolink/sensor.py b/homeassistant/components/yolink/sensor.py index 3bb0e965eae5be..cf7cd2a16d9b79 100644 --- a/homeassistant/components/yolink/sensor.py +++ b/homeassistant/components/yolink/sensor.py @@ -4,6 +4,7 @@ from collections.abc import Callable from dataclasses import dataclass +from typing import Any from yolink.const import ( ATTR_DEVICE_CO_SMOKE_SENSOR, @@ -15,6 +16,7 @@ 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, @@ -84,7 +86,7 @@ class YoLinkSensorEntityDescription(SensorEntityDescription): exists_fn: Callable[[YoLinkDevice], bool] = lambda _: True should_update_entity: Callable = lambda state: True - value: Callable = lambda state: state + value: Callable[[YoLinkDevice, dict], Any | None] = lambda device, state: state SENSOR_DEVICE_TYPE = [ @@ -114,6 +116,7 @@ class YoLinkSensorEntityDescription(SensorEntityDescription): ATTR_DEVICE_SMOKE_ALARM, ATTR_DEVICE_SPRINKLER, ATTR_DEVICE_SPRINKLER_V2, + ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR, ] BATTERY_POWER_SENSOR = [ @@ -136,6 +139,7 @@ class YoLinkSensorEntityDescription(SensorEntityDescription): ATTR_DEVICE_SOIL_TH_SENSOR, ATTR_DEVICE_SMOKE_ALARM, ATTR_DEVICE_SPRINKLER_V2, + ATTR_DEVICE_MULTI_CAPS_LEAK_SENSOR, ] MCU_DEV_TEMPERATURE_SENSOR = [ @@ -166,32 +170,52 @@ class YoLinkSensorEntityDescription(SensorEntityDescription): ] -def cvt_battery(val: int | None) -> int | None: - """Convert battery to percentage.""" - if val is None: +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 cvt_volume(val: int | None) -> str | None: - """Convert volume to string.""" - if val is None: +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, state_class=SensorStateClass.MEASUREMENT, - value=cvt_battery, 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", @@ -199,17 +223,30 @@ def cvt_volume(val: int | None) -> str | None: 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] + 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], + 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( @@ -219,16 +256,21 @@ def cvt_volume(val: int | None) -> str | None: state_class=SensorStateClass.MEASUREMENT, 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, - value=lambda value: value.get("signal") if value is not None else None, 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", @@ -236,6 +278,7 @@ def cvt_volume(val: int | None) -> str | None: 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", @@ -243,7 +286,7 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.ENUM, options=["muted", "unmuted"], exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM, - value=lambda value: "muted" if value is True else "unmuted", + value=lambda device, data: "muted" if data.get("mute") is True else "unmuted", ), YoLinkSensorEntityDescription( key="sound", @@ -251,7 +294,7 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.ENUM, options=["low", "medium", "high"], exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM, - value=cvt_volume, + value=parse_data_volume, ), YoLinkSensorEntityDescription( key="beep", @@ -259,13 +302,16 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.ENUM, options=["enabled", "disabled"], exists_fn=lambda device: device.device_type == ATTR_DEVICE_POWER_FAILURE_ALARM, - value=lambda value: "enabled" if value is True else "disabled", + 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", @@ -273,10 +319,11 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.WATER, native_unit_of_measurement=UnitOfVolume.CUBIC_METERS, state_class=SensorStateClass.TOTAL_INCREASING, - should_update_entity=lambda value: value is not None, 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", @@ -284,10 +331,11 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.WATER, native_unit_of_measurement=UnitOfVolume.CUBIC_METERS, state_class=SensorStateClass.TOTAL_INCREASING, - should_update_entity=lambda value: value is not None, 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", @@ -295,10 +343,11 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.WATER, native_unit_of_measurement=UnitOfVolume.CUBIC_METERS, state_class=SensorStateClass.TOTAL_INCREASING, - should_update_entity=lambda value: value is not None, 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", @@ -306,9 +355,11 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.POWER, native_unit_of_measurement=UnitOfPower.WATT, state_class=SensorStateClass.MEASUREMENT, - should_update_entity=lambda value: value is not None, exists_fn=lambda device: device.device_model_name in POWER_SUPPORT_MODELS, - value=lambda value: value / 10 if value is not None else None, + 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", @@ -316,9 +367,11 @@ def cvt_volume(val: int | None) -> str | None: device_class=SensorDeviceClass.ENERGY, native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, state_class=SensorStateClass.TOTAL, - should_update_entity=lambda value: value is not None, exists_fn=lambda device: device.device_model_name in POWER_SUPPORT_MODELS, - value=lambda value: value / 100 if value is not None else None, + 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", @@ -327,15 +380,19 @@ def cvt_volume(val: int | None) -> str | None: 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], + 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"), ), ) @@ -387,7 +444,8 @@ def update_entity_state(self, state: dict) -> None: """Update HA Entity State.""" if ( attr_val := self.entity_description.value( - state.get(self.entity_description.key) + self.coordinator.device, + state, ) ) is None and self.entity_description.should_update_entity(attr_val) is False: return