Add support for YS7914

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,70 +59,122 @@ 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 [
                 ATTR_DEVICE_WATER_METER_CONTROLLER,
                 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