diff --git a/tests/test_wenzhi.py b/tests/test_wenzhi.py new file mode 100644 index 0000000000..618c3bc14e --- /dev/null +++ b/tests/test_wenzhi.py @@ -0,0 +1,98 @@ +"""Tests for Wenzhi quirks.""" + +import contextlib + +import pytest +from zigpy.zcl.clusters.security import IasZone + +import zhaquirks +import zhaquirks.wenzhi.mtd085_motion + +zhaquirks.setup() + + +@pytest.mark.parametrize( + "manufacturer,model", + [ + ("_TZ321C_fkzihax8", "TS0225"), + ("_TZ321C_4slreunp", "TS0225"), + ], +) +def test_mtd085_signature(assert_signature_matches_quirk, manufacturer, model): + """Test Wenzhi MTD085-ZB signature is matched to its quirk.""" + signature = { + "node_descriptor": "NodeDescriptor(logical_type=, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=, mac_capability_flags=, manufacturer_code=4098, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=, *allocate_address=True, *is_alternate_pan_coordinator=False, *is_coordinator=False, *is_end_device=False, *is_full_function_device=True, *is_mains_powered=True, *is_receiver_on_when_idle=True, *is_router=True, *is_security_capable=False)", + "endpoints": { + "1": { + "profile_id": 260, + "device_type": "0x0402", + "in_clusters": ["0x0000", "0x0004", "0x0005", "0x0500"], + "out_clusters": ["0x000a", "0x0019"], + }, + "242": { + "profile_id": 41440, + "device_type": "0x0061", + "in_clusters": [], + "out_clusters": ["0x0021"], + }, + }, + "manufacturer": manufacturer, + "model": model, + "class": "zhaquirks.wenzhi.mtd085_motion.WenzhiMTD085_ZB", + } + assert_signature_matches_quirk( + zhaquirks.wenzhi.mtd085_motion.WenzhiMTD085_ZB, signature + ) + + +@pytest.mark.parametrize( + "manufacturer,model", + [ + ("_TZ321C_fkzihax8", "TS0225"), + ("_TZ321C_4slreunp", "TS0225"), + ], +) +@pytest.mark.asyncio +async def test_mtd085_quirk_applies(zigpy_device_from_quirk, manufacturer, model): + """Test that the MTD085-ZB quirk is applied correctly.""" + # Create the quirk with the correct signature + quirked_device = zigpy_device_from_quirk( + zhaquirks.wenzhi.mtd085_motion.WenzhiMTD085_ZB + ) + + # Check that the quirk was applied + assert isinstance(quirked_device, zhaquirks.wenzhi.mtd085_motion.WenzhiMTD085_ZB) + + # Check endpoint 1 exists and has IAS Zone cluster + ep = quirked_device.endpoints[1] + assert ep is not None + assert IasZone.cluster_id in ep.in_clusters + + # Verify IAS Zone cluster is properly instantiated + ias_zone = ep.in_clusters[IasZone.cluster_id] + assert isinstance(ias_zone, IasZone) + + +@pytest.mark.parametrize( + "manufacturer,model", + [ + ("_TZ321C_fkzihax8", "TS0225"), + ("_TZ321C_4slreunp", "TS0225"), + ], +) +@pytest.mark.asyncio +async def test_mtd085_magic_packet(zigpy_device_from_quirk, manufacturer, model): + """Test that the magic packet configuration is callable without errors.""" + quirked_device = zigpy_device_from_quirk( + zhaquirks.wenzhi.mtd085_motion.WenzhiMTD085_ZB + ) + + # Test that apply_custom_configuration exists and can be called + assert hasattr(quirked_device, "apply_custom_configuration") + + # This should not raise an exception even if the read fails + # (the real device will respond, but the mock won't) + with contextlib.suppress(Exception): + # Exception is acceptable in test environment + # The real test is that the method exists and is properly defined + await quirked_device.apply_custom_configuration() diff --git a/zhaquirks/wenzhi/__init__.py b/zhaquirks/wenzhi/__init__.py new file mode 100644 index 0000000000..4368b674c8 --- /dev/null +++ b/zhaquirks/wenzhi/__init__.py @@ -0,0 +1 @@ +"""Wenzhi/LeapMMW devices.""" diff --git a/zhaquirks/wenzhi/mtd085_motion.py b/zhaquirks/wenzhi/mtd085_motion.py new file mode 100644 index 0000000000..351fd85923 --- /dev/null +++ b/zhaquirks/wenzhi/mtd085_motion.py @@ -0,0 +1,125 @@ +"""Wenzhi/LeapMMW MTD085-ZB mmWave radar presence sensor. + +This device operates in basic IAS Zone mode when paired directly with ZHA. +The advanced features (illuminance, distance measurement, configuration parameters) +require the Tuya MCU cluster (0xEF00) which is not exposed in this mode. + +Current functionality: +- Motion detection via IAS Zone (binary_sensor.motion) +- Occupancy clear when leaving the room +- Basic device info +- Uses Tuya enchantment spell to initialize proper occupancy reporting +- Periodic reporting configured to detect device disconnection faster +""" + +from zigpy.zcl.clusters.general import Basic, Groups, Scenes +from zigpy.zcl.clusters.security import IasZone + +from zhaquirks.const import ( + DEVICE_TYPE, + ENDPOINTS, + INPUT_CLUSTERS, + MODELS_INFO, + OUTPUT_CLUSTERS, + PROFILE_ID, +) +from zhaquirks.tuya import EnchantedDevice + + +class WenzhiMTD085_ZB(EnchantedDevice): + """Wenzhi MTD085-ZB 24GHz mmWave radar presence sensor. + + Basic IAS Zone motion sensor mode. + Uses Tuya enchantment spell to report occupancy state changes. + + Manufacturer: _TZ321C_fkzihax8 / _TZ321C_4slreunp + Model: TS0225 + """ + + async def apply_custom_configuration(self, *args, **kwargs): + """Apply custom configuration including IAS Zone reporting.""" + # First apply the Tuya enchantment spell from parent class + await super().apply_custom_configuration(*args, **kwargs) + + try: + # Configure periodic reporting on IAS Zone cluster + # This ensures the device reports regularly to maintain availability + # Home Assistant marks devices unavailable after missing several reports + # Using shorter intervals for faster disconnection detection + ias_zone_cluster = self.endpoints[1].in_clusters[IasZone.cluster_id] + await ias_zone_cluster.bind() + await ias_zone_cluster.configure_reporting( + IasZone.AttributeDefs.zone_status.id, + min_interval=10, # Report at least every 10 seconds on state changes + max_interval=300, # Report at most every 5 minutes even if no change + reportable_change=1, # Report on any zone status change + ) + except Exception as ex: + # Log the error but continue - device may not support reporting config + self.debug("Failed to configure IAS Zone reporting: %s", ex) + + signature = { + MODELS_INFO: [ + ("_TZ321C_fkzihax8", "TS0225"), + ("_TZ321C_4slreunp", "TS0225"), + ], + ENDPOINTS: { + # + 1: { + PROFILE_ID: 260, # ZHA profile + DEVICE_TYPE: 0x0402, # IAS_ZONE = 1026 decimal = 0x0402 hex + INPUT_CLUSTERS: [ + 0x0000, # Basic + 0x0004, # Groups + 0x0005, # Scenes + 0x0500, # IasZone + ], + OUTPUT_CLUSTERS: [ + 0x000A, # Time + 0x0019, # OTA + ], + }, + # + 242: { + PROFILE_ID: 0xA1E0, # Green Power profile + DEVICE_TYPE: 0x0061, # Proxy Basic + INPUT_CLUSTERS: [], + OUTPUT_CLUSTERS: [ + 0x0021, # Green Power + ], + }, + }, + } + + replacement = { + ENDPOINTS: { + 1: { + PROFILE_ID: 260, + DEVICE_TYPE: 0x0402, + INPUT_CLUSTERS: [ + Basic.cluster_id, + Groups.cluster_id, + Scenes.cluster_id, + IasZone.cluster_id, + ], + OUTPUT_CLUSTERS: [ + 0x000A, # Time + 0x0019, # OTA + ], + }, + 242: { + PROFILE_ID: 0xA1E0, + DEVICE_TYPE: 0x0061, + INPUT_CLUSTERS: [], + OUTPUT_CLUSTERS: [ + 0x0021, + ], + }, + }, + }