Fix Tuya BlitzWolf motion sensor (#3629)

* Fix Tuya BlitzWolf motion sensor using wrong cluster

* Add custom `TuyaMotionWithReset` cluster

This code is somewhat copied from `MotionWithReset` and related clusters from the base `zhaquirks` module.
However, that code is outdated and still uses `Bus` somewhat and listens to the IAS zone status change commands. We generally listen to IAS zone status attribute updates now. This is also needed for the Tuya DP conversion.

* Expand test to except automatic reset

Author: TheJulianJES

tests/test_tuya_motion.py

@@ -1,8 +1,11 @@
 """Tests for Tuya quirks."""
 
+import asyncio
+
 import pytest
 from zigpy.zcl import foundation
 from zigpy.zcl.clusters.measurement import OccupancySensing
+from zigpy.zcl.clusters.security import IasZone
 
 from tests.common import ClusterListener
 import zhaquirks
@@ -32,12 +35,10 @@
         ("_TZE200_ztc6ggyl", "TS0601", ZCL_TUYA_MOTION),
         ("_TZE204_ztc6ggyl", "TS0601", ZCL_TUYA_MOTION),
         ("_TZE204_ztqnh5cg", "TS0601", ZCL_TUYA_MOTION),
-        ("_TYST11_i5j6ifxj", "5j6ifxj", ZCL_TUYA_MOTION_V3),
-        ("_TYST11_7hfcudw5", "hfcudw5", ZCL_TUYA_MOTION_V3),
     ],
 )
-async def test_tuya_motion_quirk(zigpy_device_from_v2_quirk, model, manuf, occ_msg):
-    """Test Tuya Motion Quirks."""
+async def test_tuya_motion_quirk_occ(zigpy_device_from_v2_quirk, model, manuf, occ_msg):
+    """Test Tuya Motion Quirks using Occupancy cluster."""
     quirked_device = zigpy_device_from_v2_quirk(model, manuf)
     ep = quirked_device.endpoints[1]
 
@@ -62,3 +63,47 @@ async def test_tuya_motion_quirk(zigpy_device_from_v2_quirk, model, manuf, occ_m
         occupancy_listener.attribute_updates[0][1]
         == OccupancySensing.Occupancy.Occupied
     )
+
+
+@pytest.mark.parametrize(
+    "model,manuf,occ_msg",
+    [
+        ("_TYST11_i5j6ifxj", "5j6ifxj", ZCL_TUYA_MOTION_V3),
+        ("_TYST11_7hfcudw5", "hfcudw5", ZCL_TUYA_MOTION_V3),
+    ],
+)
+@pytest.mark.asyncio
+async def test_tuya_motion_quirk_ias(zigpy_device_from_v2_quirk, model, manuf, occ_msg):
+    """Test Tuya Motion Quirks using IasZone cluster."""
+    quirked_device = zigpy_device_from_v2_quirk(model, manuf)
+    ep = quirked_device.endpoints[1]
+
+    assert ep.tuya_manufacturer is not None
+    assert isinstance(ep.tuya_manufacturer, TuyaMCUCluster)
+
+    assert ep.ias_zone is not None
+    assert isinstance(ep.ias_zone, IasZone)
+
+    # lower reset_s of IasZone cluster
+    ep.ias_zone.reset_s = 0
+
+    ias_zone_listener = ClusterListener(ep.ias_zone)
+
+    hdr, data = ep.tuya_manufacturer.deserialize(occ_msg)
+    status = ep.tuya_manufacturer.handle_get_data(data.data)
+
+    assert status == foundation.Status.SUCCESS
+
+    zcl_zone_status_id = IasZone.AttributeDefs.zone_status.id
+
+    # check that the zone status is set to alarm_1
+    assert len(ias_zone_listener.attribute_updates) == 1
+    assert ias_zone_listener.attribute_updates[0][0] == zcl_zone_status_id
+    assert ias_zone_listener.attribute_updates[0][1] == IasZone.ZoneStatus.Alarm_1
+
+    await asyncio.sleep(0.01)
+
+    # check that the zone status is reset automatically
+    assert len(ias_zone_listener.attribute_updates) == 2
+    assert ias_zone_listener.attribute_updates[1][0] == zcl_zone_status_id
+    assert ias_zone_listener.attribute_updates[1][1] == 0

zhaquirks/tuya/ts0601_motion.py

@@ -1,12 +1,15 @@
 """BlitzWolf IS-3/Tuya motion rechargeable occupancy sensor."""
 
+import asyncio
 import math
+from typing import Any
 
 from zigpy.quirks.v2 import EntityType
 from zigpy.quirks.v2.homeassistant import UnitOfLength, UnitOfTime
 from zigpy.quirks.v2.homeassistant.sensor import SensorDeviceClass, SensorStateClass
 import zigpy.types as t
 from zigpy.zcl.clusters.measurement import IlluminanceMeasurement, OccupancySensing
+from zigpy.zcl.clusters.security import IasZone
 
 from zhaquirks.tuya import TuyaLocalCluster
 from zhaquirks.tuya.builder import TuyaQuirkBuilder
@@ -20,6 +23,40 @@ class TuyaOccupancySensing(OccupancySensing, TuyaLocalCluster):
     """Tuya local OccupancySensing cluster."""
 
 
+class TuyaMotionWithReset(IasZone, TuyaLocalCluster):
+    """Tuya local IAS motion cluster with reset."""
+
+    _CONSTANT_ATTRIBUTES = {
+        IasZone.AttributeDefs.zone_type.id: IasZone.ZoneType.Motion_Sensor
+    }
+    reset_s: int = 15
+
+    def __init__(self, *args, **kwargs):
+        """Init."""
+        super().__init__(*args, **kwargs)
+        self._loop = asyncio.get_running_loop()
+        self._timer_handle = None
+
+    def _turn_off(self) -> None:
+        """Reset IAS zone status."""
+        self._timer_handle = None
+        self.debug("%s - Resetting Tuya motion sensor", self.endpoint.device.ieee)
+        self._update_attribute(IasZone.AttributeDefs.zone_status.id, 0)
+
+    def _update_attribute(self, attrid: int | t.uint16_t, value: Any) -> None:
+        """Catch zone status updates and potentially schedule reset."""
+        if (
+            attrid == IasZone.AttributeDefs.zone_status.id
+            and value == IasZone.ZoneStatus.Alarm_1
+        ):
+            self.debug("%s - Received Tuya motion event", self.endpoint.device.ieee)
+            if self._timer_handle:
+                self._timer_handle.cancel()
+            self._timer_handle = self._loop.call_later(self.reset_s, self._turn_off)
+
+        super()._update_attribute(attrid, value)
+
+
 base_tuya_motion = (
     TuyaQuirkBuilder()
     .adds(TuyaOccupancySensing)
@@ -169,13 +206,11 @@ class TuyaOccupancySensing(OccupancySensing, TuyaLocalCluster):
 (
     TuyaQuirkBuilder("_TYST11_i5j6ifxj", "5j6ifxj")
     .applies_to("_TYST11_7hfcudw5", "hfcudw5")
-    .tuya_dp(
+    .tuya_ias(
         dp_id=3,
-        ep_attribute=TuyaOccupancySensing.ep_attribute,
-        attribute_name=OccupancySensing.AttributeDefs.occupancy.name,
-        converter=lambda x: x == 2,
+        ias_cfg=TuyaMotionWithReset,
+        converter=lambda x: IasZone.ZoneStatus.Alarm_1 if x == 2 else 0,
     )
-    .adds(TuyaOccupancySensing)
     .skip_configuration()
     .add_to_registry()
 )