Custom quirk for "DIYRuZ_AirSense"

yatsik · January 2, 2022, 4:59pm

Hi. My cusrom quirk for DIYRuZ_Airsense does not work after HA update.

Does anybody can help?

HW: DIGI Zigbee is used as coordinator.
SW: HA version 2021.12.7

HA Log:

(MainThread) [zhaquirks] Loading custom quirks module sls
MainThread) [zhaquirks] Loading custom quirks module sls.airqmon
...
(MainThread) [zigpy.appdb] [0xbf0d:1:0x0000] Attribute id: 4 value: SLS
(MainThread) [zigpy.appdb] [0xbf0d:1:0x0000] Attribute id: 5 value: AirQMon
(MainThread) [zigpy.appdb] [0xffdc:1:0x0000] Attribute id: 4 value: eWeLink
(MainThread) [zigpy.appdb] [0xffdc:1:0x0000] Attribute id: 5 value: TH01
(MainThread) [zigpy.appdb] [0xf283:1:0x0000] Attribute id: 5 value: lumi.plug
(MainThread) [zigpy.appdb] [0xf283:1:0x0000] Attribute id: 4 value: LUMI
(MainThread) [zigpy.appdb] [0x334a:1:0x0000] Attribute id: 4 value: LUMI
(MainThread) [zigpy.appdb] [0x334a:1:0x0000] Attribute id: 5 value: lumi.sensor_wleak.aq1
....
(MainThread) [zigpy.quirks.registry] Checking quirks for SLS AirQMon (00:xx:xx:xx:xx:xx:f1:ec)
(MainThread) [zigpy.quirks.registry] Considering <class sls.airqmon.AirQMon>
...
(MainThread) [zigpy_xbee.api] _handle_explicit_rx: (00:xx:xx:xx:xx:xx:f1:ec, 0xbf0d, 1, 1026, 0, b180101000086)
(MainThread) [zigpy.zcl] [0xbf0d:1:0x0402] ZCL deserialize: <ZCLHeader frame_control=<FrameControl frame_type=GLOBAL_COMMAND manufacturer_specific=False is_reply=True disable_default_response=True> manufacturer=None tsn=1 command_id=Command.Read_Attributes_rsp>
(MainThread) [homeassistant.components.zha.core.channels.base] [0xBF0D:1:0x0402]: finished channel initialization
(MainThread) [homeassistant.components.zha.core.channels.base] [0xBF0D:1:0x0402]: async_initialize stage succeeded
(MainThread) [homeassistant.components.zha.core.channels.base] [0xBF0D:1:0x0000]: async_initialize stage succeeded
(MainThread) [zigpy_xbee.api] Frame received: tx_status
(MainThread) [zigpy_xbee.api] tx_explicit to 0xbf0d: TXStatus.SUCCESS after 0 tries. Discovery Status: DiscoveryStatus.SUCCESS, Frame #18
(MainThread) [zigpy_xbee.api] Frame received: tx_status
(MainThread) [zigpy_xbee.api] tx_explicit to 0x2b10: TXStatus.SUCCESS after 0 tries. Discovery Status: DiscoveryStatus.SUCCESS, Frame #19
(MainThread) [zigpy_xbee.api] Frame received: explicit_rx_indicator
(MainThread) [zigpy_xbee.api] _handle_explicit_rx: (00:xx:xx:xx:xx:xx:f1:ec, 0xbf0d, 2, 1037, 0, b180201000086)
(MainThread) [zigpy.zcl] [0xbf0d:2:0x040d] ZCL deserialize: <ZCLHeader frame_control=<FrameControl frame_type=GLOBAL_COMMAND manufacturer_specific=False is_reply=True disable_default_response=True> manufacturer=None tsn=2 command_id=Command.Read_Attributes_rsp>
(MainThread) [zigpy.appdb] Error handling _unsupported_attribute_added event with (00:xx:xx:xx:xx:xx:f1:ec, 1, 1026, 0) params: FOREIGN KEY constraint failed
(MainThread) [zigpy_xbee.api] Frame received: explicit_rx_indicator
(MainThread) [zigpy_xbee.api] _handle_explicit_rx: (00:xx:xx:xx:xx:xx:95:75, 0x2b10, 1, 6, 0, b1803010000001000)
(MainThread) [zigpy.zcl] [0x2b10:1:0x0006] ZCL deserialize: <ZCLHeader frame_control=<FrameControl frame_type=GLOBAL_COMMAND manufacturer_specific=False is_reply=True disable_default_response=True> manufacturer=None tsn=3 command_id=Command.Read_Attributes_rsp>
...
(MainThread) [zigpy.appdb] Error handling _unsupported_attribute_added event with (00:xx:xx:xx:xx:xx:f1:ec, 2, 1037, 0) params: FOREIGN KEY constraint failed
...
(MainThread) [zigpy_xbee.api] _handle_explicit_rx: (00:xx:xx:xx:xx:xx:f1:ec, 0xbf0d, 1, 12, 0, b009a0a5500390000e8416f0018021c00420143)
(MainThread) [zigpy.zcl] [0xbf0d:1:0x000c] ZCL deserialize: <ZCLHeader frame_control=<FrameControl frame_type=GLOBAL_COMMAND manufacturer_specific=False is_reply=False disable_default_response=False> manufacturer=None tsn=154 command_id=Command.Report_Attributes>
(MainThread) [zigpy.zcl] [0xbf0d:1:0x000c] ZCL request 0x000a: [[Attribute(attrid=85, value=<TypeValue type=Single, value=29.0>), Attribute(attrid=111, value=<TypeValue type=bitmap8, value=bitmap8.2>), Attribute(attrid=28, value=<TypeValue type=CharacterString, value=C>)]]
(MainThread) [zigpy.zcl] [0xbf0d:1:0x000c] Attribute report received: present_value=29.0, status_flags=2, description=C
(MainThread) [zigpy_xbee.zigbee.application] Zigbee request tsn #154: b189a0b0a00
(MainThread) [zigpy_xbee.api] Command tx_explicit (00:xx:xx:xx:xx:xx:f1:ec, 0xBF0D, 1, 1, 12, 260, 0, 0, b\x18\x9a\x0b\n\x00)
(MainThread) [zigpy_xbee.api] Frame received: tx_status
(MainThread) [zigpy_xbee.api] tx_explicit to 0xbf0d: TXStatus.SUCCESS after 0 tries. Discovery Status: DiscoveryStatus.SUCCESS, Frame #44
(MainThread) [zigpy_xbee.api] Frame received: explicit_rx_indicator
(MainThread) [zigpy_xbee.api] _handle_explicit_rx: (00:xx:xx:xx:xx:xx:f1:ec, 0xbf0d, 1, 12, 0, b009b0a550039000026446f0018a91c00420370706d)
(MainThread) [zigpy.zcl] [0xbf0d:2:0x000c] ZCL deserialize: <ZCLHeader frame_control=<FrameControl frame_type=GLOBAL_COMMAND manufacturer_specific=False is_reply=False disable_default_response=False> manufacturer=None tsn=155 command_id=Command.Report_Attributes>
(MainThread) [zigpy.zcl] [0xbf0d:2:0x000c] ZCL request 0x000a: [[Attribute(attrid=85, value=<TypeValue type=Single, value=664.0>), Attribute(attrid=111, value=<TypeValue type=bitmap8, value=bitmap8.128|32|8|1>), Attribute(attrid=28, value=<TypeValue type=CharacterString, value=ppm>)]]
(MainThread) [zigpy.zcl] [0xbf0d:2:0x000c] Attribute report received: present_value=664.0, status_flags=169, description=ppm
(MainThread) [zigpy_xbee.zigbee.application] Zigbee request tsn #155: b189b0b0a00

yatsik · January 2, 2022, 5:05pm

"""SLS common components for custom device handlers."""
#__init__.py
import logging


from zigpy import types as t

from zigpy.quirks import CustomCluster, CustomDevice
from zigpy.zcl.clusters.general import (
    AnalogInput,
    Basic
)

from zigpy.zcl.clusters.measurement import (
    TemperatureMeasurement,
    CarbonDioxideConcentration
)

from zhaquirks import (
    Bus,
    QuickInitDevice
)


import zigpy.quirks as quirks

TEMPERATURE = "temperature"
TEMPERATURE_MEASUREMENT = "temperature_measurement"
TEMPERATURE_REPORTED = "temperature_reported"

CARBON_MONOXIDE = "carbon_monoxide"
CARBON_MONOXIDE_MEASUREMENT = "carbon_monoxide_measurement"
CARBON_MONOXIDE_REPORTED = "carbon_monoxide_reported"

CARBON_DIOXIDE = "carbon_dioxide"
CARBON_DIOXIDE_MEASUREMENT = "carbon_dioxide_measurement"
CARBON_DIOXIDE_REPORTED = "carbon_dioxide_reported"

#logname = 'slslogfile.log'
#logging.basicConfig(filename=logname,
#                            filemode='a',
#                            format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
#                            datefmt='%H:%M:%S',
#                            level=logging.DEBUG)


_LOGGER = logging.getLogger(__name__)

class SLSCustomDevice(CustomDevice):
    """Custom device representing SLS devices."""

    def __init__(self, *args, **kwargs):
        """Init."""
        super().__init__(*args, **kwargs)

class SLSQuickInitDevice(SLSCustomDevice, QuickInitDevice):
    """SLS devices eligible for QuickInit."""
    
class AnalogInputTemperatureCluster(CustomCluster, AnalogInput):
    """Analog input cluster, used to temperature measurement."""
    
    cluster_id = AnalogInput.cluster_id
    ATTR_ID = 85

    def __init__(self, *args, **kwargs):
        """Init."""
        self._current_state = {}
        super().__init__(*args, **kwargs)
        _LOGGER.debug("init")
        print("AnalogInputTemperatureCluster: init")

    def _update_attribute(self, attrid, value):
        super()._update_attribute(attrid, value)
        print('Before: AnalogInputTemperatureCluster: Analog claster _update_attribute: {0} -> {1}'.format(TEMPERATURE_REPORTED, value))
        if (value is not None) and (attrid == self.ATTR_ID):
            self.endpoint.device.temperature_bus.listener_event(TEMPERATURE_REPORTED, value)
            _LOGGER.debug('Analog claster _update_attribute: {0} -> {1}'.format(TEMPERATURE_REPORTED, value))
            print('After: AnalogInputTemperatureCluster: Analog claster _update_attribute: {0} -> {1}'.format(TEMPERATURE_REPORTED, value))

class AnalogInputCarbonDioCluster(CustomCluster, AnalogInput):
    """Analog input cluster, used for carbon monixide measurement"""

    cluster_id = AnalogInput.cluster_id
    ATTR_ID = 85

    def __init__(self, *args, **kwargs):
        """Init."""
        self._current_state = {}
        super().__init__(*args, **kwargs)

    def _update_attribute(self, attrid, value):
        super()._update_attribute(attrid, value)
        
        if (value is not None) and (attrid == self.ATTR_ID):
            self.endpoint.device.carbon_dioxide_bus.listener_event(CARBON_DIOXIDE_REPORTED, value)
            #_LOGGER.debug('Analog claster _update_attribute : {0} -> {1}'.format(CARBON_DIOXIDE_REPORTED, value))

class TemperatureMeasurementCluster(CustomCluster, TemperatureMeasurement):
    """Temperature cluster that filters out invalid temperature readings."""

    cluster_id = TemperatureMeasurement.cluster_id
    ATTR_ID = 0x0000
   
    def __init__(self, *args, **kwargs):
        """Init."""
        super().__init__(*args, **kwargs)
        self.endpoint.device.temperature_bus.add_listener(self)

    def _update_attribute(self, attrid, value):
        # drop values above and below documented range for this sensor
        if value is not None and attrid == self.ATTR_ID:
            super()._update_attribute(self.ATTR_ID, value)
            #_LOGGER.debug("TemperatureMeasurementCluster:_update_attribute {0}-{1}".format(self.ATTR_ID, value))
            #print("TemperatureMeasurementCluster:_update_attribute {0}-{1}".format(self.ATTR_ID, value))

    def temperature_reported(self, value):
        """Temperature reported."""
        self._update_attribute(self.ATTR_ID, value)
        #_LOGGER.debug("TemperatureMeasurementCluster:temperature_reported {0}-{1}".format(self.ATTR_ID, value))
        #print("TemperatureMeasurementCluster:temperature_reported {0}-{1}".format(self.ATTR_ID, value))

class CarbonDioxideMeasurementCluster(CustomCluster, CarbonDioxideConcentration):
    """Carbon Dioxide that filters out invalid temperature readings."""

    cluster_id = CarbonDioxideConcentration.cluster_id
    ATTR_ID = 0x0000
    
    def __init__(self, *args, **kwargs):
        """Init."""
        super().__init__(*args, **kwargs)
        self.endpoint.device.carbon_dioxide_bus.add_listener(self)

    def _update_attribute(self, attrid, value):
        # drop values above and below documented range for this sensor
        # value is in ppm
        if value is not None and attrid == self.ATTR_ID:
            super()._update_attribute(self.ATTR_ID, value)
            #_LOGGER.debug("_update_attribute {0}-{1}".format(attrid, value))
            
    def carbon_dioxide_reported(self, value):
        """Carbon dioxide reported."""
        self._update_attribute(self.ATTR_ID, value)
        #_LOGGER.debug("YY: _update_attribute {0}-{1}".format(self.ATTR_ID, value))

yatsik · January 2, 2022, 5:06pm

#airqmon.py
import logging

from zigpy.profiles import zha

from zigpy.quirks import CustomCluster, CustomDevice
from zigpy.zcl.clusters.general import Basic, Identify, OnOff, PowerConfiguration
from zigpy.zcl.clusters.measurement import TemperatureMeasurement, CarbonDioxideConcentration

from zhaquirks import Bus

from zhaquirks.const import (
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODELS_INFO,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
    SKIP_CONFIGURATION
)

from sls import (
    TemperatureMeasurementCluster, 
    AnalogInputTemperatureCluster, 
    CarbonDioxideMeasurementCluster, 
    AnalogInputCarbonDioCluster, 
    SLSQuickInitDevice)

_LOGGER = logging.getLogger(__name__)

class AirQMon(SLSQuickInitDevice):
    """Custom device representing devices."""

    def __init__(self, *args, **kwargs):
        """Init."""
        
        self.carbon_dioxide_bus = Bus()
        self.temperature_bus = Bus()
        super().__init__(*args, **kwargs)
        _LOGGER.debug("AirQMon: init")
        print("AirQMon: init")
 

    signature = {
        # <SimpleDescriptor endpoint=1 profile=260 device_type=0 device_version=0
        # input_clusters=[0, 3, 1] output_clusters=[6, 3]>
        MODELS_INFO: [("SLS", "AirQMon")],
        ENDPOINTS: {
            1: {
                PROFILE_ID: 260,
                DEVICE_TYPE: 0x0100,
                INPUT_CLUSTERS: [
                    0x0000,
                    0x000c,
                    0x0b05,
                ],
                OUTPUT_CLUSTERS: [
                    0x0000,
        	        0x0014,
                    0x0b05,
                ],
            },
            2: {
                PROFILE_ID: 260,
                DEVICE_TYPE: 0x0100,
                INPUT_CLUSTERS: [
                    0x000c,
                ],
                OUTPUT_CLUSTERS: [
        	        0x0014,
                ],
            },
        },
    }
    
    replacement = {
        #SKIP_CONFIGURATION: True,
        MODELS_INFO: [("SLS", "AirQMon")],
        ENDPOINTS: {
            1: {
                PROFILE_ID: 260,
                DEVICE_TYPE: zha.DeviceType.TEMPERATURE_SENSOR,
                INPUT_CLUSTERS: [
                    0x0000,
                    AnalogInputTemperatureCluster,
                    TemperatureMeasurementCluster,
                    0x0b05,
                ],
                OUTPUT_CLUSTERS: [
                    0x0000,
        	        0x0014,
                    0x0b05,
                    
                ],
            },
            2: {
                PROFILE_ID: 260,
                #DEVICE_TYPE: CarbonMonoxideConcentration.cluster_id,
                DEVICE_TYPE: zha.DeviceType.SIMPLE_SENSOR,
                INPUT_CLUSTERS: [
                    AnalogInputCarbonDioCluster,
                    CarbonDioxideMeasurementCluster,
                ],
                OUTPUT_CLUSTERS: [
        	    0x0014,
                ],
            },
        },
    }