Support Frient devices

zhaquirks/const.py

@@ -18,6 +18,7 @@
 ATTR_ID = "attr_id"
 ATTRIBUTE_ID = "attribute_id"
 ATTRIBUTE_NAME = "attribute_name"
+ATTRIBUTE_VALUE = "attribute_value"
 BUTTON = "button"
 BUTTON_1 = "button_1"
 BUTTON_2 = "button_2"

zhaquirks/develco/__init__.py

@@ -1,9 +1,11 @@
 """Quirks for Develco Products A/S."""
 
-from zigpy import types as t
+from typing import Final
+
 from zigpy.quirks import CustomCluster
+import zigpy.types as t
 from zigpy.zcl import foundation
-from zigpy.zcl.clusters.security import IasZone
+from zigpy.zcl.clusters.security import IasZone, ZoneStatus
 
 from zhaquirks import PowerConfigurationCluster
 
@@ -19,22 +21,19 @@ class DevelcoPowerConfiguration(PowerConfigurationCluster):
 
 
 class DevelcoIasZone(CustomCluster, IasZone):
-    """Custom IasZone for Develco."""
-
-    client_commands = {
-        0x00: foundation.ZCLCommandDef(
-            "status_change_notification",
-            {
-                "zone_status": IasZone.ZoneStatus,
-                "extended_status?": t.bitmap8,
+    """IAS Zone, patched to fix a bug with the status change notification command."""
+
+    class ClientCommandDefs(IasZone.ClientCommandDefs):
+        """IAS Zone command definitions."""
+
+        status_change_notification: Final = foundation.ZCLCommandDef(
+            id=0x00,
+            schema={
+                "zone_status": ZoneStatus,
+                "extended_status": t.bitmap8,
+                # These two should not be optional
                 "zone_id?": t.uint8_t,
                 "delay?": t.uint16_t,
             },
-            False,
-        ),
-        0x01: foundation.ZCLCommandDef(
-            "enroll",
-            {"zone_type": IasZone.ZoneType, "manufacturer_code": t.uint16_t},
-            False,
-        ),
-    }
+            direction=foundation.Direction.Client_to_Server,
+        )

zhaquirks/develco/air_quality.py

@@ -1,241 +1,80 @@
 """Develco Air Quality Sensor."""
 
 import logging
+from typing import Final
 
-from zigpy.profiles import zha
-from zigpy.quirks import CustomCluster, CustomDevice
+from zigpy.quirks import CustomCluster
+from zigpy.quirks.v2 import QuirkBuilder, SensorDeviceClass, SensorStateClass
+from zigpy.quirks.v2.homeassistant import CONCENTRATION_PARTS_PER_BILLION
 import zigpy.types as t
-from zigpy.zcl.clusters.general import (
-    Basic,
-    Identify,
-    OnOff,
-    Ota,
-    PollControl,
-    PowerConfiguration,
-    Scenes,
-    Time,
+from zigpy.zcl.foundation import (
+    ZCL_CLUSTER_REVISION_ATTR,
+    ZCL_REPORTING_STATUS_ATTR,
+    BaseAttributeDefs,
+    ZCLAttributeDef,
 )
-from zigpy.zcl.clusters.measurement import RelativeHumidity, TemperatureMeasurement
 
-from zhaquirks import Bus, LocalDataCluster
-from zhaquirks.const import (
-    DEVICE_TYPE,
-    ENDPOINTS,
-    INPUT_CLUSTERS,
-    MODELS_INFO,
-    OUTPUT_CLUSTERS,
-    PROFILE_ID,
-)
-from zhaquirks.develco import DEVELCO, DevelcoPowerConfiguration
-
-MANUFACTURER = 0x1015
-VOC_MEASURED_VALUE = 0x0000
-VOC_MIN_MEASURED_VALUE = 0x0001
-VOC_MAX_MEASURED_VALUE = 0x0002
-VOC_RESOLUTION = 0x0003
-
-VOC_REPORTED = "voc_reported"
-MIN_VOC_REPORTED = "min_voc_reported"
-MAX_VOC_REPORTED = "max_voc_reported"
-VOC_RESOLUTION_REPORTED = "voc_resolution_reported"
+from zhaquirks.develco import DevelcoPowerConfiguration
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class DevelcoVOCMeasurement(CustomCluster):
-    """Input Cluster to route manufacturer specific VOC cluster to actual VOC cluster."""
+    """Develco VOC cluster definition."""
 
     cluster_id = 0xFC03
     name = "VOC Level"
-    ep_attribute = "voc_level"
-    attributes = {
-        VOC_MEASURED_VALUE: ("measured_value", t.uint16_t, True),
-        VOC_MIN_MEASURED_VALUE: ("min_measured_value", t.uint16_t, True),
-        VOC_MAX_MEASURED_VALUE: ("max_measured_value", t.uint16_t, True),
-        VOC_RESOLUTION: ("resolution", t.uint16_t, True),
-    }
-    server_commands = {}
-    client_commands = {}
+    ep_attribute = "develco_voc_level"
 
-    def __init__(self, *args, **kwargs):
-        """Init."""
-        self._current_state = {}
-        super().__init__(*args, **kwargs)
-        self.endpoint.device.app_cluster = self
+    class AttributeDefs(BaseAttributeDefs):
+        """Attribute definitions, same as all the other `Measurement` clusters."""
 
-    def _update_attribute(self, attrid, value):
-        super()._update_attribute(attrid, value)
-        if attrid == VOC_MEASURED_VALUE and value is not None:
-            self.endpoint.device.voc_bus.listener_event(VOC_REPORTED, value)
-        if attrid == VOC_MIN_MEASURED_VALUE and value is not None:
-            self.endpoint.device.voc_bus.listener_event(MIN_VOC_REPORTED, value)
-        if attrid == VOC_MAX_MEASURED_VALUE and value is not None:
-            self.endpoint.device.voc_bus.listener_event(MAX_VOC_REPORTED, value)
-        if attrid == VOC_RESOLUTION and value is not None:
-            self.endpoint.device.voc_bus.listener_event(VOC_RESOLUTION_REPORTED, value)
-        _LOGGER.debug(
-            "%s Develco VOC : [%s]",
-            self.endpoint.device.ieee,
-            self._attr_cache,
+        measured_value: Final = ZCLAttributeDef(
+            id=0x0000,
+            type=t.uint16_t,  # In parts per billion
+            access="rp",
+            mandatory=True,
+            is_manufacturer_specific=True,
         )
-
-
-class DevelcoRelativeHumidity(CustomCluster, RelativeHumidity):
-    """Handles invalid values for Humidity."""
-
-    def _update_attribute(self, attrid, value):
-        # Drop values out of specified range (0-100% RH)
-        if 0 <= value <= 10000:
-            super()._update_attribute(attrid, value)
-        _LOGGER.debug(
-            "%s Develco Humidity : [%s]",
-            self.endpoint.device.ieee,
-            self._attr_cache,
+        min_measured_value: Final = ZCLAttributeDef(
+            id=0x0001,
+            type=t.uint16_t,
+            access="r",
+            mandatory=True,
+            is_manufacturer_specific=True,
         )
-
-
-class DevelcoTemperatureMeasurement(CustomCluster, TemperatureMeasurement):
-    """Handles invalid values for Temperature."""
-
-    def _update_attribute(self, attrid, value):
-        # Drop values out of specified range (0-50°C)
-        if 0 <= value <= 5000:
-            super()._update_attribute(attrid, value)
-        _LOGGER.debug(
-            "%s Develco Temperature : [%s]",
-            self.endpoint.device.ieee,
-            self._attr_cache,
+        max_measured_value: Final = ZCLAttributeDef(
+            id=0x0002,
+            type=t.uint16_t,
+            access="r",
+            mandatory=True,
+            is_manufacturer_specific=True,
+        )
+        tolerance: Final = ZCLAttributeDef(
+            id=0x0003,
+            type=t.uint16_t,
+            access="r",
+            is_manufacturer_specific=True,
         )
 
-
-class EmulatedVOCMeasurement(LocalDataCluster):
-    """VOC measurement cluster to receive reports from the Develco VOC cluster."""
-
-    cluster_id = 0x042E
-    name = "VOC Level"
-    ep_attribute = "voc_level"
-    attributes = {
-        VOC_MEASURED_VALUE: ("measured_value", t.uint16_t, True),
-        VOC_MIN_MEASURED_VALUE: ("min_measured_value", t.uint16_t, True),
-        VOC_MAX_MEASURED_VALUE: ("max_measured_value", t.uint16_t, True),
-        VOC_RESOLUTION: ("resolution", t.uint16_t, True),
-    }
-    MEASURED_VALUE_ID = 0x0000
-    MIN_MEASURED_VALUE_ID = 0x0001
-    MAX_MEASURED_VALUE_ID = 0x0002
-    RESOLUTION_ID = 0x0003
-
-    def __init__(self, *args, **kwargs):
-        """Init."""
-        super().__init__(*args, **kwargs)
-        self.endpoint.device.voc_bus.add_listener(self)
-
-    async def bind(self):
-        """Bind cluster."""
-        result = await self.endpoint.device.app_cluster.bind()
-        return result
-
-    async def write_attributes(self, attributes, manufacturer=None):
-        """Ignore write_attributes."""
-        return (0,)
-
-    def _update_attribute(self, attrid, value):
-        # Drop values out of specified range (0-60000 ppb)
-        if 0 <= value <= 60000:
-            # Convert ppb into mg/m³ approximation according to develco spec
-            value = value * 0.0000045
-            super()._update_attribute(attrid, value)
-
-    def voc_reported(self, value):
-        """VOC reported."""
-        self._update_attribute(self.MEASURED_VALUE_ID, value)
-
-    def min_voc_reported(self, value):
-        """Minimum Measured VOC reported."""
-        self._update_attribute(self.MIN_MEASURED_VALUE_ID, value)
-
-    def max_voc_reported(self, value):
-        """Maximum Measured VOC reported."""
-        self._update_attribute(self.MAX_MEASURED_VALUE_ID, value)
-
-    def voc_resolution_reported(self, value):
-        """VOC Resolution reported."""
-        self._update_attribute(self.RESOLUTION_ID, value)
-
-
-class AQSZB110(CustomDevice):
-    """Custom device Develco air quality sensor."""
-
-    manufacturer_id_override = MANUFACTURER
-
-    def __init__(self, *args, **kwargs):
-        """Init."""
-        self.voc_bus = Bus()
-        super().__init__(*args, **kwargs)
-
-    signature = {
-        # <SimpleDescriptor endpoint=1 profile=49353 device_type=1 device_version=1
-        # input_clusters=[3, 5, 6] output_clusters=[]>
-        # <SimpleDescriptor endpoint=38 profile=260 device_type=770 device_version=0
-        # input_clusters=[0, 1, 3, 32, 1026, 1029, 64515] output_clusters=[3, 10, 25]>
-        MODELS_INFO: [
-            (DEVELCO, "AQSZB-110"),
-            ("frient A/S", "AQSZB-110"),
-        ],
-        ENDPOINTS: {
-            1: {
-                PROFILE_ID: 0xC0C9,
-                DEVICE_TYPE: 1,
-                INPUT_CLUSTERS: [
-                    Identify.cluster_id,
-                    Scenes.cluster_id,
-                    OnOff.cluster_id,
-                ],
-                OUTPUT_CLUSTERS: [],
-            },
-            38: {
-                PROFILE_ID: zha.PROFILE_ID,
-                DEVICE_TYPE: zha.DeviceType.TEMPERATURE_SENSOR,
-                INPUT_CLUSTERS: [
-                    Basic.cluster_id,
-                    PowerConfiguration.cluster_id,
-                    Identify.cluster_id,
-                    PollControl.cluster_id,
-                    TemperatureMeasurement.cluster_id,
-                    RelativeHumidity.cluster_id,
-                    0xFC03,
-                ],
-                OUTPUT_CLUSTERS: [Identify.cluster_id, Time.cluster_id, Ota.cluster_id],
-            },
-        },
-    }
-
-    replacement = {
-        ENDPOINTS: {
-            1: {
-                PROFILE_ID: 0xC0C9,
-                DEVICE_TYPE: 1,
-                INPUT_CLUSTERS: [
-                    Identify.cluster_id,
-                    Scenes.cluster_id,
-                    OnOff.cluster_id,
-                ],
-                OUTPUT_CLUSTERS: [],
-            },
-            38: {
-                PROFILE_ID: zha.PROFILE_ID,
-                DEVICE_TYPE: zha.DeviceType.TEMPERATURE_SENSOR,
-                INPUT_CLUSTERS: [
-                    Basic.cluster_id,
-                    DevelcoPowerConfiguration,
-                    Identify.cluster_id,
-                    PollControl.cluster_id,
-                    DevelcoTemperatureMeasurement,
-                    DevelcoRelativeHumidity,
-                    DevelcoVOCMeasurement,
-                    EmulatedVOCMeasurement,
-                ],
-                OUTPUT_CLUSTERS: [Identify.cluster_id, Time.cluster_id, Ota.cluster_id],
-            },
-        },
-    }
+        cluster_revision: Final = ZCL_CLUSTER_REVISION_ATTR
+        reporting_status: Final = ZCL_REPORTING_STATUS_ATTR
+
+
+(
+    QuirkBuilder("frient A/S", "AQSZB-110")
+    .applies_to("Develco Products A/S", "AQSZB-110")
+    .replaces(DevelcoVOCMeasurement, endpoint_id=38)
+    .replaces(DevelcoPowerConfiguration, endpoint_id=38)
+    .sensor(
+        attribute_name=DevelcoVOCMeasurement.AttributeDefs.measured_value.name,
+        cluster_id=DevelcoVOCMeasurement.cluster_id,
+        endpoint_id=38,
+        device_class=SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS,
+        state_class=SensorStateClass.MEASUREMENT,
+        unit=CONCENTRATION_PARTS_PER_BILLION,
+        fallback_name="VOC Level",
+        unique_id_suffix="voc_level",
+    )
+    .add_to_registry()
+)

zhaquirks/develco/emi.py

@@ -0,0 +1,15 @@
+"""Frient Electricity Meter Interface."""
+
+from zigpy.quirks.v2 import QuirkBuilder
+
+(
+    QuirkBuilder("frient A/S", "EMIZB-151")
+    # These endpoints are duplicates and completely broken: each one is a "mirror" of
+    # endpoint 2 and will set up duplicate attribute reporting for every attribute, the
+    # attribute reports will instead be emitted from endpoint 2!
+    .prevent_default_entity_creation(endpoint_id=64)
+    .prevent_default_entity_creation(endpoint_id=65)
+    .prevent_default_entity_creation(endpoint_id=66)
+    .prevent_default_entity_creation(endpoint_id=67)
+    .add_to_registry()
+)