ZHA Add entities for Lidl water valve quirk (#72307)

* init

* added timer number entity

* added write attribute button entity

* fixed missed errors

* minor changes & fixed failing test

* removed icon

* unit and icons

homeassistant/components/zha/binary_sensor.py

@@ -171,3 +171,16 @@ class IASZone(BinarySensor):
         value = await self._channel.get_attribute_value("zone_status")
         if value is not None:
             self._state = value & 3
+
+
+@MULTI_MATCH(
+    channel_names="tuya_manufacturer",
+    manufacturers={
+        "_TZE200_htnnfasr",
+    },
+)
+class FrostLock(BinarySensor, id_suffix="frost_lock"):
+    """ZHA BinarySensor."""
+
+    SENSOR_ATTR = "frost_lock"
+    _attr_device_class: BinarySensorDeviceClass = BinarySensorDeviceClass.LOCK

homeassistant/components/zha/button.py

@@ -6,6 +6,9 @@ import functools
 import logging
 from typing import Any
 
+import zigpy.exceptions
+from zigpy.zcl.foundation import Status
+
 from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
@@ -21,6 +24,9 @@ from .core.typing import ChannelType, ZhaDeviceType
 from .entity import ZhaEntity
 
 MULTI_MATCH = functools.partial(ZHA_ENTITIES.multipass_match, Platform.BUTTON)
+CONFIG_DIAGNOSTIC_MATCH = functools.partial(
+    ZHA_ENTITIES.config_diagnostic_match, Platform.BUTTON
+)
 DEFAULT_DURATION = 5  # seconds
 
 _LOGGER = logging.getLogger(__name__)
@@ -103,3 +109,50 @@ class ZHAIdentifyButton(ZHAButton):
         """Return the arguments to use in the command."""
 
         return [DEFAULT_DURATION]
+
+
+class ZHAAttributeButton(ZhaEntity, ButtonEntity):
+    """Defines a ZHA button, which stes value to an attribute."""
+
+    _attribute_name: str = None
+    _attribute_value: Any = None
+
+    def __init__(
+        self,
+        unique_id: str,
+        zha_device: ZhaDeviceType,
+        channels: list[ChannelType],
+        **kwargs,
+    ) -> None:
+        """Init this button."""
+        super().__init__(unique_id, zha_device, channels, **kwargs)
+        self._channel: ChannelType = channels[0]
+
+    async def async_press(self) -> None:
+        """Write attribute with defined value."""
+        try:
+            result = await self._channel.cluster.write_attributes(
+                {self._attribute_name: self._attribute_value}
+            )
+        except zigpy.exceptions.ZigbeeException as ex:
+            self.error("Could not set value: %s", ex)
+            return
+        if not isinstance(result, Exception) and all(
+            record.status == Status.SUCCESS for record in result[0]
+        ):
+            self.async_write_ha_state()
+
+
+@CONFIG_DIAGNOSTIC_MATCH(
+    channel_names="tuya_manufacturer",
+    manufacturers={
+        "_TZE200_htnnfasr",
+    },
+)
+class FrostLockResetButton(ZHAAttributeButton, id_suffix="reset_frost_lock"):
+    """Defines a ZHA identify button."""
+
+    _attribute_name = "frost_lock_reset"
+    _attribute_value = 0
+    _attr_device_class = ButtonDeviceClass.RESTART
+    _attr_entity_category = EntityCategory.CONFIG

homeassistant/components/zha/number.py

@@ -495,3 +495,20 @@ class StartUpCurrentLevelConfigurationEntity(
     _attr_min_value: float = 0x00
     _attr_max_value: float = 0xFF
     _zcl_attribute: str = "start_up_current_level"
+
+
+@CONFIG_DIAGNOSTIC_MATCH(
+    channel_names="tuya_manufacturer",
+    manufacturers={
+        "_TZE200_htnnfasr",
+    },
+)
+class TimerDurationMinutes(ZHANumberConfigurationEntity, id_suffix="timer_duration"):
+    """Representation of a ZHA timer duration configuration entity."""
+
+    _attr_entity_category = EntityCategory.CONFIG
+    _attr_icon: str = ICONS[14]
+    _attr_min_value: float = 0x00
+    _attr_max_value: float = 0x257
+    _attr_unit_of_measurement: str | None = UNITS[72]
+    _zcl_attribute: str = "timer_duration"

homeassistant/components/zha/sensor.py

@@ -27,6 +27,7 @@ from homeassistant.const import (
     PRESSURE_HPA,
     TEMP_CELSIUS,
     TIME_HOURS,
+    TIME_MINUTES,
     TIME_SECONDS,
     VOLUME_CUBIC_FEET,
     VOLUME_CUBIC_METERS,
@@ -754,3 +755,18 @@ class RSSISensor(Sensor, id_suffix="rssi"):
 @MULTI_MATCH(channel_names=CHANNEL_BASIC)
 class LQISensor(RSSISensor, id_suffix="lqi"):
     """LQI sensor for a device."""
+
+
+@MULTI_MATCH(
+    channel_names="tuya_manufacturer",
+    manufacturers={
+        "_TZE200_htnnfasr",
+    },
+)
+class TimeLeft(Sensor, id_suffix="time_left"):
+    """Sensor that displays time left value."""
+
+    SENSOR_ATTR = "timer_time_left"
+    _attr_device_class: SensorDeviceClass = SensorDeviceClass.DURATION
+    _attr_icon = "mdi:timer"
+    _unit = TIME_MINUTES

tests/components/zha/test_button.py

@@ -1,11 +1,22 @@
 """Test ZHA button."""
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 from freezegun import freeze_time
 import pytest
+from zhaquirks.const import (
+    DEVICE_TYPE,
+    ENDPOINTS,
+    INPUT_CLUSTERS,
+    OUTPUT_CLUSTERS,
+    PROFILE_ID,
+)
 from zigpy.const import SIG_EP_PROFILE
+from zigpy.exceptions import ZigbeeException
 import zigpy.profiles.zha as zha
+from zigpy.quirks import CustomCluster, CustomDevice
+import zigpy.types as t
 import zigpy.zcl.clusters.general as general
+from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
 import zigpy.zcl.clusters.security as security
 import zigpy.zcl.foundation as zcl_f
 
@@ -14,6 +25,7 @@ from homeassistant.components.button.const import SERVICE_PRESS
 from homeassistant.const import (
     ATTR_DEVICE_CLASS,
     ATTR_ENTITY_ID,
+    ENTITY_CATEGORY_CONFIG,
     ENTITY_CATEGORY_DIAGNOSTIC,
     STATE_UNKNOWN,
 )
@@ -48,6 +60,49 @@ async def contact_sensor(hass, zigpy_device_mock, zha_device_joined_restored):
     return zha_device, zigpy_device.endpoints[1].identify
 
 
+class FrostLockQuirk(CustomDevice):
+    """Quirk with frost lock attribute."""
+
+    class TuyaManufCluster(CustomCluster, ManufacturerSpecificCluster):
+        """Tuya manufacturer specific cluster."""
+
+        cluster_id = 0xEF00
+        ep_attribute = "tuya_manufacturer"
+
+        attributes = {0xEF01: ("frost_lock_reset", t.Bool)}
+
+    replacement = {
+        ENDPOINTS: {
+            1: {
+                PROFILE_ID: zha.PROFILE_ID,
+                DEVICE_TYPE: zha.DeviceType.ON_OFF_SWITCH,
+                INPUT_CLUSTERS: [general.Basic.cluster_id, TuyaManufCluster],
+                OUTPUT_CLUSTERS: [],
+            },
+        }
+    }
+
+
+@pytest.fixture
+async def tuya_water_valve(hass, zigpy_device_mock, zha_device_joined_restored):
+    """Tuya Water Valve fixture."""
+
+    zigpy_device = zigpy_device_mock(
+        {
+            1: {
+                SIG_EP_INPUT: [general.Basic.cluster_id],
+                SIG_EP_OUTPUT: [],
+                SIG_EP_TYPE: zha.DeviceType.ON_OFF_SWITCH,
+            }
+        },
+        manufacturer="_TZE200_htnnfasr",
+        quirk=FrostLockQuirk,
+    )
+
+    zha_device = await zha_device_joined_restored(zigpy_device)
+    return zha_device, zigpy_device.endpoints[1].tuya_manufacturer
+
+
 @freeze_time("2021-11-04 17:37:00", tz_offset=-1)
 async def test_button(hass, contact_sensor):
     """Test zha button platform."""
@@ -87,3 +142,52 @@ async def test_button(hass, contact_sensor):
     assert state
     assert state.state == "2021-11-04T16:37:00+00:00"
     assert state.attributes[ATTR_DEVICE_CLASS] == ButtonDeviceClass.UPDATE
+
+
+async def test_frost_unlock(hass, tuya_water_valve):
+    """Test custom frost unlock zha button."""
+
+    entity_registry = er.async_get(hass)
+    zha_device, cluster = tuya_water_valve
+    assert cluster is not None
+    entity_id = await find_entity_id(DOMAIN, zha_device, hass)
+    assert entity_id is not None
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes[ATTR_DEVICE_CLASS] == ButtonDeviceClass.RESTART
+
+    entry = entity_registry.async_get(entity_id)
+    assert entry
+    assert entry.entity_category == ENTITY_CATEGORY_CONFIG
+
+    with patch(
+        "zigpy.zcl.Cluster.request",
+        return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        assert len(cluster.write_attributes.mock_calls) == 1
+        assert cluster.write_attributes.call_args == call({"frost_lock_reset": 0})
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.attributes[ATTR_DEVICE_CLASS] == ButtonDeviceClass.RESTART
+
+    cluster.write_attributes.reset_mock()
+    cluster.write_attributes.side_effect = ZigbeeException
+
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+    assert len(cluster.write_attributes.mock_calls) == 1
+    assert cluster.write_attributes.call_args == call({"frost_lock_reset": 0})