diff --git a/homeassistant/components/zha/binary_sensor.py b/homeassistant/components/zha/binary_sensor.py index 730748d74ae..954c60fa895 100644 --- a/homeassistant/components/zha/binary_sensor.py +++ b/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 diff --git a/homeassistant/components/zha/button.py b/homeassistant/components/zha/button.py index f130936df02..9f241795267 100644 --- a/homeassistant/components/zha/button.py +++ b/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 diff --git a/homeassistant/components/zha/number.py b/homeassistant/components/zha/number.py index 22d086891ca..bece4bc894d 100644 --- a/homeassistant/components/zha/number.py +++ b/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" diff --git a/homeassistant/components/zha/sensor.py b/homeassistant/components/zha/sensor.py index 249034ef068..3e3017f6fa9 100644 --- a/homeassistant/components/zha/sensor.py +++ b/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 diff --git a/tests/components/zha/test_button.py b/tests/components/zha/test_button.py index 762d2d46e54..f692528203f 100644 --- a/tests/components/zha/test_button.py +++ b/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})