Created
October 19, 2025 15:00
-
-
Save andreibsk/fa9d68a417430bb4d00ff592345866de to your computer and use it in GitHub Desktop.
Custom ZHA quirks for Aqara Wall Outlet H2 EU and Wall Switch H1 EU
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from __future__ import annotations | |
| from enum import Enum | |
| from typing import Final | |
| from zigpy import types | |
| from zigpy.quirks import CustomCluster | |
| from zigpy.quirks.v2 import QuirkBuilder | |
| from zigpy.zcl.clusters.general import ( | |
| DeviceTemperature, | |
| OnOff, | |
| ) | |
| from zigpy.zcl.clusters.measurement import TemperatureMeasurement | |
| from zigpy.zcl.foundation import ( | |
| BaseAttributeDefs, | |
| ZCLAttributeDef, | |
| ) | |
| from zhaquirks.xiaomi import ( | |
| AnalogInputCluster, | |
| BasicCluster, | |
| ElectricalMeasurementCluster, | |
| MeteringCluster, | |
| ) | |
| from zigpy.quirks.v2.homeassistant import ( | |
| UnitOfPower, | |
| ) | |
| import logging | |
| from zigpy.zcl import foundation | |
| from collections.abc import Iterable, Iterator | |
| import logging | |
| from typing import Final | |
| from zigpy import types as t | |
| from zigpy.zcl import foundation | |
| from zigpy.zcl.clusters.general import ( | |
| DeviceTemperature, | |
| OnOff, | |
| ) | |
| from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement | |
| from zigpy.zcl.clusters.measurement import ( | |
| IlluminanceMeasurement, | |
| PressureMeasurement, | |
| RelativeHumidity, | |
| TemperatureMeasurement, | |
| ) | |
| from zigpy.zcl.clusters.security import IasZone | |
| from zigpy.zcl.clusters.smartenergy import Metering | |
| from zigpy.zcl.foundation import BaseAttributeDefs, ZCLAttributeDef | |
| from zhaquirks.const import ( | |
| ATTRIBUTE_ID, | |
| ATTRIBUTE_NAME, | |
| COMMAND_ATTRIBUTE_UPDATED, | |
| COMMAND_TRIPLE, | |
| UNKNOWN, | |
| VALUE, | |
| ZHA_SEND_EVENT, | |
| ) | |
| AQARA = "Aqara" | |
| BATTERY_LEVEL = "battery_level" | |
| BATTERY_PERCENTAGE_REMAINING = 0x0021 | |
| BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE = "battery_percentage" | |
| BATTERY_SIZE = "battery_size" | |
| BATTERY_SIZE_ATTR = 0x0031 | |
| BATTERY_QUANTITY_ATTR = 0x0033 | |
| BATTERY_VOLTAGE_MV = "battery_voltage_mV" | |
| HUMIDITY_MEASUREMENT = "humidity_measurement" | |
| LUMI = "LUMI" | |
| MODEL = 5 | |
| MOTION_SENSITIVITY = "motion_sensitivity" | |
| DETECTION_INTERVAL = "detection_interval" | |
| MOTION_TYPE = 0x000D | |
| OCCUPANCY_STATE = 0 | |
| PATH = "path" | |
| POWER = "power" | |
| CONSUMPTION = "consumption" | |
| VOLTAGE = "voltage" | |
| PRESSURE_MEASUREMENT = "pressure_measurement" | |
| PRESSURE_MEASUREMENT_PRECISION = "pressure_measurement_precision" | |
| STATE = "state" | |
| TEMPERATURE = "temperature" | |
| TEMPERATURE_MEASUREMENT = "temperature_measurement" | |
| TVOC_MEASUREMENT = "tvoc_measurement" | |
| POWER_OUTAGE_COUNT = "power_outage_count" | |
| PRESENCE_DETECTED = "presence_detected" | |
| PRESENCE_EVENT = "presence_event" | |
| MONITORING_MODE = "monitoring_mode" | |
| APPROACH_DISTANCE = "approach_distance" | |
| ILLUMINANCE_MEASUREMENT = "illuminance_measurement" | |
| SMOKE = "smoke" | |
| SMOKE_DENSITY = "smoke_density" | |
| SELF_TEST = "self_test" | |
| BUZZER_MANUAL_MUTE = "buzzer_manual_mute" | |
| HEARTBEAT_INDICATOR = "heartbeat_indicator" | |
| LINKAGE_ALARM = "linkage_alarm" | |
| LINKAGE_ALARM_STATE = "linkage_alarm_state" | |
| XIAOMI_AQARA_ATTRIBUTE = 0xFF01 | |
| XIAOMI_AQARA_ATTRIBUTE_E1 = 0x00F7 | |
| XIAOMI_ATTR_3 = "X-attrib-3" | |
| XIAOMI_ATTR_4 = "X-attrib-4" | |
| XIAOMI_ATTR_5 = "X-attrib-5" | |
| XIAOMI_ATTR_6 = "X-attrib-6" | |
| XIAOMI_MIJA_ATTRIBUTE = 0xFF02 | |
| _LOGGER = logging.getLogger(__name__) | |
| ################################################################################## | |
| # Start of modified clusters from internal code: | |
| ################################################################################## | |
| class XiaomiClusterMod(CustomCluster): | |
| """Xiaomi cluster implementation.""" | |
| def _iter_parse_attr_report( | |
| self, data: bytes | |
| ) -> Iterator[tuple[foundation.Attribute, bytes]]: | |
| """Yield all interpretations of the first attribute in a Xiaomi report.""" | |
| # Peek at the attribute report | |
| attr_id, data = t.uint16_t.deserialize(data) | |
| attr_type, data = t.uint8_t.deserialize(data) | |
| if ( | |
| attr_id | |
| not in ( | |
| XIAOMI_AQARA_ATTRIBUTE, | |
| XIAOMI_MIJA_ATTRIBUTE, | |
| XIAOMI_AQARA_ATTRIBUTE_E1, | |
| ) | |
| or attr_type != 0x42 # "Character String" | |
| ): | |
| # Assume other attributes are reported correctly | |
| data = attr_id.serialize() + attr_type.serialize() + data | |
| attribute, data = foundation.Attribute.deserialize(data) | |
| yield attribute, data | |
| return | |
| # Length of the "string" can be wrong | |
| val_len, data = t.uint8_t.deserialize(data) | |
| # Try every offset. Start with 0 to pass unbroken reports through. | |
| for offset in (0, -1, 1): | |
| fixed_len = val_len + offset | |
| if len(data) < fixed_len: | |
| continue | |
| val, final_data = data[:fixed_len], data[fixed_len:] | |
| attr_val = t.LVBytes(val) | |
| attr_type = 0x41 # The data type should be "Octet String" | |
| yield ( | |
| foundation.Attribute( | |
| attrid=attr_id, | |
| value=foundation.TypeValue(type=attr_type, value=attr_val), | |
| ), | |
| final_data, | |
| ) | |
| def _interpret_attr_reports( | |
| self, data: bytes | |
| ) -> Iterable[tuple[foundation.Attribute]]: | |
| """Yield all valid interprations of a Xiaomi attribute report.""" | |
| if not data: | |
| yield () | |
| return | |
| try: | |
| parsed = list(self._iter_parse_attr_report(data)) | |
| except (KeyError, ValueError): | |
| return | |
| for attr, remaining_data in parsed: | |
| for remaining_attrs in self._interpret_attr_reports(remaining_data): | |
| yield (attr,) + remaining_attrs | |
| def deserialize(self, data): | |
| """Deserialize cluster data.""" | |
| hdr, data = foundation.ZCLHeader.deserialize(data) | |
| # Only handle attribute reports differently | |
| if ( | |
| hdr.frame_control.frame_type != foundation.FrameType.GLOBAL_COMMAND | |
| or hdr.command_id != foundation.GeneralCommand.Report_Attributes | |
| ): | |
| return super().deserialize(hdr.serialize() + data) | |
| reports = list(self._interpret_attr_reports(data)) | |
| if not reports: | |
| _LOGGER.warning("Failed to parse Xiaomi attribute report: %r", data) | |
| return super().deserialize(hdr.serialize() + data) | |
| elif len(reports) > 1: | |
| _LOGGER.warning( | |
| "Xiaomi attribute report has multiple valid interpretations: %r", | |
| reports, | |
| ) | |
| fixed_data = b"".join(attr.serialize() for attr in reports[0]) | |
| return super().deserialize(hdr.serialize() + fixed_data) | |
| def _update_attribute(self, attrid, value): | |
| if attrid in (XIAOMI_AQARA_ATTRIBUTE, XIAOMI_AQARA_ATTRIBUTE_E1): | |
| attributes = self._parse_aqara_attributes(value) | |
| super()._update_attribute(attrid, value) | |
| if self.endpoint.device.model == "lumi.sensor_switch.aq2": | |
| if value == b"\x04!\xa8C\n!\x00\x00": | |
| self.listener_event(ZHA_SEND_EVENT, COMMAND_TRIPLE, []) | |
| elif attrid == XIAOMI_MIJA_ATTRIBUTE: | |
| attributes = self._parse_mija_attributes(value) | |
| else: | |
| super()._update_attribute(attrid, value) | |
| if attrid == MODEL: | |
| # 0x0005 = model attribute. | |
| # Xiaomi sensors send the model attribute when their reset button is | |
| # pressed quickly.""" | |
| if attrid in self.attributes: | |
| attribute_name = self.attributes[attrid].name | |
| else: | |
| attribute_name = UNKNOWN | |
| self.listener_event( | |
| ZHA_SEND_EVENT, | |
| COMMAND_ATTRIBUTE_UPDATED, | |
| { | |
| ATTRIBUTE_ID: attrid, | |
| ATTRIBUTE_NAME: attribute_name, | |
| VALUE: value, | |
| }, | |
| ) | |
| return | |
| _LOGGER.debug( | |
| "%s - Xiaomi attribute report. attribute_id: [%s] value: [%s]", | |
| self.endpoint.device.ieee, | |
| attrid, | |
| attributes, | |
| ) | |
| if BATTERY_VOLTAGE_MV in attributes: | |
| # many Xiaomi devices report this, but not all quirks implement the XiaomiPowerConfiguration cluster, | |
| # so we might error out if the method doesn't exist | |
| if hasattr(self.endpoint.power, "battery_reported") and callable( | |
| self.endpoint.power.battery_reported | |
| ): | |
| self.endpoint.power.battery_reported(attributes[BATTERY_VOLTAGE_MV]) | |
| else: | |
| # log a debug message if the cluster is not implemented | |
| _LOGGER.debug( | |
| "%s - Xiaomi battery voltage attribute received but XiaomiPowerConfiguration not used", | |
| self.endpoint.device.ieee, | |
| ) | |
| if TEMPERATURE_MEASUREMENT in attributes: | |
| self.endpoint.temperature.update_attribute( | |
| TemperatureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[TEMPERATURE_MEASUREMENT], | |
| ) | |
| if HUMIDITY_MEASUREMENT in attributes: | |
| self.endpoint.humidity.update_attribute( | |
| RelativeHumidity.AttributeDefs.measured_value.id, | |
| attributes[HUMIDITY_MEASUREMENT], | |
| ) | |
| if PRESSURE_MEASUREMENT in attributes: | |
| self.endpoint.pressure.update_attribute( | |
| PressureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[PRESSURE_MEASUREMENT], | |
| ) | |
| if PRESSURE_MEASUREMENT_PRECISION in attributes: | |
| self.endpoint.pressure.update_attribute( | |
| PressureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[PRESSURE_MEASUREMENT_PRECISION] / 100, | |
| ) | |
| if POWER in attributes: | |
| self.endpoint.electrical_measurement.update_attribute( | |
| ElectricalMeasurement.AttributeDefs.active_power.id, | |
| round(attributes[POWER] * 10), | |
| ) | |
| if CONSUMPTION in attributes: | |
| zcl_consumption = round(attributes[CONSUMPTION] * 1000) | |
| self.endpoint.smartenergy_metering.update_attribute( | |
| Metering.AttributeDefs.current_summ_delivered.id, zcl_consumption | |
| ) | |
| if VOLTAGE in attributes: | |
| self.endpoint.electrical_measurement.update_attribute( | |
| ElectricalMeasurement.AttributeDefs.rms_voltage.id, | |
| attributes[VOLTAGE] * 0.1, | |
| ) | |
| if ILLUMINANCE_MEASUREMENT in attributes: | |
| self.endpoint.illuminance.update_attribute( | |
| IlluminanceMeasurement.AttributeDefs.measured_value.id, | |
| attributes[ILLUMINANCE_MEASUREMENT], | |
| ) | |
| if TVOC_MEASUREMENT in attributes: | |
| self.endpoint.voc_level.update_attribute( | |
| 0x0000, attributes[TVOC_MEASUREMENT] | |
| ) | |
| if TEMPERATURE in attributes: | |
| if hasattr(self.endpoint, "device_temperature"): | |
| self.endpoint.device_temperature.update_attribute( | |
| DeviceTemperature.AttributeDefs.current_temperature.id, | |
| attributes[TEMPERATURE] * 100, | |
| ) | |
| if BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE in attributes: | |
| self.endpoint.power.battery_percent_reported( | |
| attributes[BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE] | |
| ) | |
| if SMOKE in attributes: | |
| self.endpoint.ias_zone.update_attribute( | |
| IasZone.AttributeDefs.zone_status.id, attributes[SMOKE] | |
| ) | |
| def _parse_aqara_attributes(self, value): | |
| """Parse non-standard attributes.""" | |
| attributes = {} | |
| attribute_names = { | |
| 1: BATTERY_VOLTAGE_MV, | |
| 3: TEMPERATURE, | |
| 4: XIAOMI_ATTR_4, | |
| 5: XIAOMI_ATTR_5, | |
| 6: XIAOMI_ATTR_6, | |
| 10: PATH, | |
| } | |
| if self.endpoint.device.model in [ | |
| "lumi.sensor_ht", | |
| "lumi.sens", | |
| "lumi.weather", | |
| "lumi.airmonitor.acn01", | |
| "lumi.sensor_ht.agl02", | |
| ]: | |
| # Temperature sensors send temperature/humidity/pressure updates through this | |
| # cluster instead of the respective clusters | |
| attribute_names.update( | |
| { | |
| 100: TEMPERATURE_MEASUREMENT, | |
| 101: HUMIDITY_MEASUREMENT, | |
| 102: ( | |
| TVOC_MEASUREMENT | |
| if self.endpoint.device.model == "lumi.airmonitor.acn01" | |
| else ( | |
| PRESSURE_MEASUREMENT_PRECISION | |
| if self.endpoint.device.model == "lumi.weather" | |
| else PRESSURE_MEASUREMENT | |
| ) | |
| ), | |
| } | |
| ) | |
| elif self.endpoint.device.model in [ | |
| "lumi.plug", | |
| "lumi.plug.aeu001", | |
| "lumi.plug.maus01", | |
| "lumi.plug.maeu01", | |
| "lumi.plug.mmeu01", | |
| "lumi.relay.c2acn01", | |
| "lumi.switch.n0agl1", | |
| "lumi.switch.n0acn2", | |
| ]: | |
| attribute_names.update({149: CONSUMPTION, 150: VOLTAGE, 152: POWER}) | |
| elif self.endpoint.device.model == "lumi.switch.agl011": | |
| attribute_names.update({150: VOLTAGE, 151: CONSUMPTION, 152: POWER}) | |
| elif self.endpoint.device.model == "lumi.sensor_motion.aq2": | |
| attribute_names.update({11: ILLUMINANCE_MEASUREMENT}) | |
| elif self.endpoint.device.model == "lumi.curtain.acn002": | |
| attribute_names.update({101: BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE}) | |
| elif self.endpoint.device.model in [ | |
| "lumi.motion.agl02", | |
| "lumi.motion.ac02", | |
| "lumi.motion.acn001", | |
| ]: | |
| attribute_names.update({101: ILLUMINANCE_MEASUREMENT}) | |
| if self.endpoint.device.model == "lumi.motion.ac02": | |
| attribute_names.update({105: DETECTION_INTERVAL}) | |
| attribute_names.update({106: MOTION_SENSITIVITY}) | |
| elif self.endpoint.device.model == "lumi.motion.agl04": | |
| attribute_names.update({102: DETECTION_INTERVAL}) | |
| attribute_names.update({105: MOTION_SENSITIVITY}) | |
| attribute_names.update({258: DETECTION_INTERVAL}) | |
| attribute_names.update({268: MOTION_SENSITIVITY}) | |
| elif self.endpoint.device.model == "lumi.motion.ac01": | |
| attribute_names.update({5: POWER_OUTAGE_COUNT}) | |
| attribute_names.update({101: PRESENCE_DETECTED}) | |
| attribute_names.update({102: PRESENCE_EVENT}) | |
| attribute_names.update({103: MONITORING_MODE}) | |
| attribute_names.update({105: APPROACH_DISTANCE}) | |
| attribute_names.update({268: MOTION_SENSITIVITY}) | |
| attribute_names.update({322: PRESENCE_DETECTED}) | |
| attribute_names.update({323: PRESENCE_EVENT}) | |
| attribute_names.update({324: MONITORING_MODE}) | |
| attribute_names.update({326: APPROACH_DISTANCE}) | |
| elif self.endpoint.device.model == "lumi.sensor_smoke.acn03": | |
| attribute_names.update({160: SMOKE}) | |
| attribute_names.update({161: SMOKE_DENSITY}) | |
| attribute_names.update({162: SELF_TEST}) | |
| attribute_names.update({163: BUZZER_MANUAL_MUTE}) | |
| attribute_names.update({164: HEARTBEAT_INDICATOR}) | |
| attribute_names.update({165: LINKAGE_ALARM}) | |
| result = {} | |
| # Some attribute reports end with a stray null byte | |
| while value not in (b"", b"\x00"): | |
| skey = int(value[0]) | |
| svalue, value = foundation.TypeValue.deserialize(value[1:]) | |
| result[skey] = svalue.value | |
| for item, val in result.items(): | |
| key = ( | |
| attribute_names[item] | |
| if item in attribute_names | |
| else "0xff01-" + str(item) | |
| ) | |
| attributes[key] = val | |
| return attributes | |
| def _parse_mija_attributes(self, value): | |
| """Parse non-standard attributes.""" | |
| attribute_names = ( | |
| STATE, | |
| BATTERY_VOLTAGE_MV, | |
| XIAOMI_ATTR_3, | |
| XIAOMI_ATTR_4, | |
| XIAOMI_ATTR_5, | |
| XIAOMI_ATTR_6, | |
| ) | |
| result = [] | |
| for attr_value in value: | |
| result.append(attr_value.value) | |
| attributes = dict(zip(attribute_names, result)) | |
| return attributes | |
| class XiaomiAqaraE1ClusterMod(XiaomiClusterMod): | |
| """Xiaomi mfg cluster implementation.""" | |
| cluster_id = 0xFCC0 | |
| ep_attribute = "opple_cluster" | |
| class AttributeDefs(BaseAttributeDefs): | |
| """Cluster attributes.""" | |
| ################################################################################## | |
| # End of modified clusters from internal code. | |
| ################################################################################## | |
| class AqaraPowerOutageMemoryEnum(types.uint8_t, Enum): | |
| """Power Outage Memory enum.""" | |
| On = 0x00 | |
| Previous = 0x01 | |
| Off = 0x02 | |
| Inverted = 0x03 | |
| class PlugAEU001Cluster(XiaomiAqaraE1ClusterMod): | |
| """Custom cluster for Aqara lumi plug AEU001.""" | |
| class AttributeDefs(BaseAttributeDefs): | |
| """Attribute definitions.""" | |
| button_lock: Final = ZCLAttributeDef( | |
| id=0x0200, type=types.uint8_t, access="rw", is_manufacturer_specific=True | |
| ) | |
| charging_protection: Final = ZCLAttributeDef( | |
| id=0x0202, type=types.Bool, access="rw", is_manufacturer_specific=True | |
| ) | |
| led_indicator: Final = ZCLAttributeDef( | |
| id=0x0203, type=types.Bool, access="rw", is_manufacturer_specific=True | |
| ) | |
| charging_limit: Final = ZCLAttributeDef( | |
| id=0x0206, type=types.Single, access="rw", is_manufacturer_specific=True | |
| ) | |
| overload_protection: Final = ZCLAttributeDef( | |
| id=0x020B, type=types.Single, access="rw", is_manufacturer_specific=True | |
| ) | |
| power_on_behavior: Final = ZCLAttributeDef( | |
| id=0x0517, | |
| type=AqaraPowerOutageMemoryEnum, | |
| access="rw", | |
| is_manufacturer_specific=True, | |
| ) | |
| class PlugAEU001MeteringCluster(MeteringCluster): | |
| """Custom cluster for Aqara lumi plug AEU001.""" | |
| def _update_attribute(self, attrid, value): | |
| if attrid == self.CURRENT_SUMM_DELIVERED_ID: | |
| current_value = self._attr_cache.get(attrid, 0) | |
| if value < current_value: | |
| _LOGGER.debug( | |
| "Ignoring attribute update for %s: new value %s is less than current value %s", | |
| attrid, | |
| value, | |
| current_value, | |
| ) | |
| return | |
| super()._update_attribute(attrid, value) | |
| ( | |
| QuirkBuilder("Aqara", "lumi.plug.aeu001") | |
| .friendly_name(model="Wall Outlet H2 EU", manufacturer="Aqara") | |
| .removes(TemperatureMeasurement.cluster_id) | |
| .adds(DeviceTemperature) | |
| .removes(OnOff.cluster_id, endpoint_id=2) | |
| .replaces(BasicCluster) | |
| .replaces(PlugAEU001MeteringCluster) | |
| .replaces(ElectricalMeasurementCluster) | |
| .replaces(PlugAEU001Cluster) | |
| .replaces(AnalogInputCluster, endpoint_id=21) | |
| .switch( | |
| attribute_name="button_lock", | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| force_inverted=True, | |
| translation_key="child_lock", | |
| fallback_name="Child lock", | |
| ) | |
| .enum( | |
| attribute_name="power_on_behavior", | |
| enum_class=AqaraPowerOutageMemoryEnum, | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| translation_key="power_on_behavior", | |
| fallback_name="Power on behavior", | |
| ) | |
| .number( | |
| attribute_name="overload_protection", | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| min_value=100, | |
| max_value=3840, | |
| unit=UnitOfPower.WATT, | |
| translation_key="overload_protection", | |
| fallback_name="Overload protection", | |
| ) | |
| .switch( | |
| attribute_name="led_indicator", | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| translation_key="led_indicator", | |
| fallback_name="LED indicator", | |
| ) | |
| .switch( | |
| attribute_name="charging_protection", | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| translation_key="charging_protection", | |
| fallback_name="Charging protection", | |
| ) | |
| .number( | |
| attribute_name="charging_limit", | |
| cluster_id=PlugAEU001Cluster.cluster_id, | |
| min_value=0.1, | |
| max_value=2, | |
| step=0.1, | |
| unit=UnitOfPower.WATT, | |
| translation_key="charging_limit", | |
| fallback_name="Charging limit", | |
| ) | |
| .add_to_registry() | |
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from __future__ import annotations | |
| import typing | |
| from zigpy.zcl.clusters.general import ( | |
| Alarms, | |
| DeviceTemperature, | |
| ) | |
| from zigpy.quirks.v2 import QuirkBuilder | |
| from zigpy.quirks import signature_matches | |
| from zhaquirks.xiaomi import LUMI | |
| from zhaquirks.xiaomi.aqara.opple_switch import ( | |
| OppleSwitchMode, | |
| OppleSwitchType, | |
| XiaomiOpple2ButtonSwitch1, | |
| XiaomiOpple2ButtonSwitch3, | |
| XiaomiOpple2ButtonSwitchBase, | |
| OppleOperationMode, | |
| OppleIndicatorLight, | |
| ) | |
| from zhaquirks.xiaomi import ( | |
| LUMI, | |
| AnalogInputCluster, | |
| BasicCluster, | |
| DeviceTemperatureCluster, | |
| ElectricalMeasurementCluster, | |
| MeteringCluster, | |
| OnOffCluster, | |
| ) | |
| from zhaquirks.xiaomi.aqara.opple_remote import MultistateInputCluster | |
| from zhaquirks.xiaomi.aqara.switch_h1_single import ( | |
| AqaraH1SingleRockerBase, | |
| AqaraH1SingleRockerSwitchWithNeutral, | |
| ) | |
| from zhaquirks.const import ( | |
| ATTR_ID, | |
| BUTTON, | |
| PRESS_TYPE, | |
| VALUE, | |
| ZHA_SEND_EVENT, | |
| ) | |
| from typing import Final | |
| from zigpy import types as t | |
| from zigpy.zcl.clusters.general import ( | |
| Alarms, | |
| DeviceTemperature, | |
| ) | |
| from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement | |
| from zigpy.zcl.clusters.smartenergy import Metering | |
| from zigpy.zcl.foundation import ZCLAttributeDef | |
| from zhaquirks.const import ( | |
| ATTR_ID, | |
| BUTTON, | |
| PRESS_TYPE, | |
| VALUE, | |
| ZHA_SEND_EVENT, | |
| ) | |
| from zhaquirks.xiaomi import ( | |
| LUMI, | |
| AnalogInputCluster, | |
| BasicCluster, | |
| DeviceTemperatureCluster, | |
| ElectricalMeasurementCluster, | |
| MeteringCluster, | |
| OnOffCluster, | |
| ) | |
| from zhaquirks.xiaomi.aqara.opple_remote import MultistateInputCluster | |
| from zigpy import types | |
| from zigpy.zdo.types import NodeDescriptor | |
| from zhaquirks import CustomCluster | |
| from zhaquirks.const import ( | |
| ATTR_ID, | |
| BUTTON, | |
| PRESS_TYPE, | |
| VALUE, | |
| ZHA_SEND_EVENT, | |
| ) | |
| from zhaquirks.xiaomi import ( | |
| LUMI, | |
| BasicCluster, | |
| ) | |
| PRESS_TYPES = {0: "hold", 1: "single", 2: "double", 3: "triple", 255: "release"} | |
| STATUS_TYPE_ATTR = 0x0055 # decimal = 85 | |
| COMMAND_1_SINGLE = "1_single" | |
| COMMAND_1_DOUBLE = "1_double" | |
| COMMAND_1_TRIPLE = "1_triple" | |
| COMMAND_1_HOLD = "1_hold" | |
| COMMAND_1_RELEASE = "1_release" | |
| COMMAND_2_SINGLE = "2_single" | |
| COMMAND_2_DOUBLE = "2_double" | |
| COMMAND_2_TRIPLE = "2_triple" | |
| COMMAND_2_HOLD = "2_hold" | |
| COMMAND_2_RELEASE = "2_release" | |
| COMMAND_3_SINGLE = "3_single" | |
| COMMAND_3_DOUBLE = "3_double" | |
| COMMAND_3_TRIPLE = "3_triple" | |
| COMMAND_3_HOLD = "3_hold" | |
| COMMAND_3_RELEASE = "3_release" | |
| COMMAND_4_SINGLE = "4_single" | |
| COMMAND_4_DOUBLE = "4_double" | |
| COMMAND_4_TRIPLE = "4_triple" | |
| COMMAND_4_HOLD = "4_hold" | |
| COMMAND_4_RELEASE = "4_release" | |
| COMMAND_5_SINGLE = "5_single" | |
| COMMAND_5_DOUBLE = "5_double" | |
| COMMAND_5_TRIPLE = "5_triple" | |
| COMMAND_5_HOLD = "5_hold" | |
| COMMAND_5_RELEASE = "5_release" | |
| COMMAND_6_SINGLE = "6_single" | |
| COMMAND_6_DOUBLE = "6_double" | |
| COMMAND_6_TRIPLE = "6_triple" | |
| COMMAND_6_HOLD = "6_hold" | |
| COMMAND_6_RELEASE = "6_release" | |
| OPPLE_MFG_CODE = 0x115F | |
| BOTH_BUTTONS = "both_buttons" | |
| PRESS_TYPES = {0: "hold", 1: "single", 2: "double", 3: "triple", 255: "release"} | |
| import logging | |
| _LOGGER = logging.getLogger(__name__) | |
| from collections.abc import Iterable, Iterator | |
| from typing import Final | |
| from zigpy import types as t | |
| from zigpy.quirks import CustomCluster | |
| from zigpy.zcl import foundation | |
| from zigpy.zcl.clusters.general import ( | |
| DeviceTemperature, | |
| ) | |
| from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement | |
| from zigpy.zcl.clusters.measurement import ( | |
| IlluminanceMeasurement, | |
| PressureMeasurement, | |
| RelativeHumidity, | |
| TemperatureMeasurement, | |
| ) | |
| from zigpy.zcl.clusters.security import IasZone | |
| from zigpy.zcl.clusters.smartenergy import Metering | |
| from zigpy.zcl.foundation import BaseAttributeDefs, ZCLAttributeDef | |
| from zigpy.zdo.types import NodeDescriptor | |
| from zhaquirks.const import ( | |
| ATTRIBUTE_ID, | |
| ATTRIBUTE_NAME, | |
| COMMAND_ATTRIBUTE_UPDATED, | |
| COMMAND_TRIPLE, | |
| UNKNOWN, | |
| VALUE, | |
| ZHA_SEND_EVENT, | |
| ) | |
| AQARA = "Aqara" | |
| BATTERY_LEVEL = "battery_level" | |
| BATTERY_PERCENTAGE_REMAINING = 0x0021 | |
| BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE = "battery_percentage" | |
| BATTERY_SIZE = "battery_size" | |
| BATTERY_SIZE_ATTR = 0x0031 | |
| BATTERY_QUANTITY_ATTR = 0x0033 | |
| BATTERY_VOLTAGE_MV = "battery_voltage_mV" | |
| HUMIDITY_MEASUREMENT = "humidity_measurement" | |
| LUMI = "LUMI" | |
| MODEL = 5 | |
| MOTION_SENSITIVITY = "motion_sensitivity" | |
| DETECTION_INTERVAL = "detection_interval" | |
| MOTION_TYPE = 0x000D | |
| OCCUPANCY_STATE = 0 | |
| PATH = "path" | |
| POWER = "power" | |
| CONSUMPTION = "consumption" | |
| VOLTAGE = "voltage" | |
| PRESSURE_MEASUREMENT = "pressure_measurement" | |
| PRESSURE_MEASUREMENT_PRECISION = "pressure_measurement_precision" | |
| STATE = "state" | |
| TEMPERATURE = "temperature" | |
| TEMPERATURE_MEASUREMENT = "temperature_measurement" | |
| TVOC_MEASUREMENT = "tvoc_measurement" | |
| POWER_OUTAGE_COUNT = "power_outage_count" | |
| PRESENCE_DETECTED = "presence_detected" | |
| PRESENCE_EVENT = "presence_event" | |
| MONITORING_MODE = "monitoring_mode" | |
| APPROACH_DISTANCE = "approach_distance" | |
| ILLUMINANCE_MEASUREMENT = "illuminance_measurement" | |
| SMOKE = "smoke" | |
| SMOKE_DENSITY = "smoke_density" | |
| SELF_TEST = "self_test" | |
| BUZZER_MANUAL_MUTE = "buzzer_manual_mute" | |
| HEARTBEAT_INDICATOR = "heartbeat_indicator" | |
| LINKAGE_ALARM = "linkage_alarm" | |
| LINKAGE_ALARM_STATE = "linkage_alarm_state" | |
| XIAOMI_AQARA_ATTRIBUTE = 0xFF01 | |
| XIAOMI_AQARA_ATTRIBUTE_E1 = 0x00F7 | |
| XIAOMI_ATTR_3 = "X-attrib-3" | |
| XIAOMI_ATTR_4 = "X-attrib-4" | |
| XIAOMI_ATTR_5 = "X-attrib-5" | |
| XIAOMI_ATTR_6 = "X-attrib-6" | |
| XIAOMI_MIJA_ATTRIBUTE = 0xFF02 | |
| XIAOMI_NODE_DESC = NodeDescriptor( | |
| byte1=2, | |
| byte2=64, | |
| mac_capability_flags=128, | |
| manufacturer_code=4151, | |
| maximum_buffer_size=127, | |
| maximum_incoming_transfer_size=100, | |
| server_mask=0, | |
| maximum_outgoing_transfer_size=100, | |
| descriptor_capability_field=0, | |
| ) | |
| # requires adding to attribute names in | |
| # /usr/local/lib/python3.13/site-packages/zhaquirks/xiaomi/__init__.py | |
| # elif self.endpoint.device.model in [ | |
| # "lumi.plug", | |
| # "lumi.plug.aeu001", # TODO: BSK included in PR | |
| # "lumi.plug.maus01", | |
| # "lumi.plug.maeu01", | |
| # "lumi.plug.mmeu01", | |
| # "lumi.relay.c2acn01", | |
| # "lumi.switch.n0agl1", | |
| # "lumi.switch.n0acn2", | |
| # "lumi.switch.n2aeu1", # TODO: BSK add to PR | |
| # "lumi.switch.n1aeu1", # TODO: BSK add to PR | |
| # ]: | |
| ################################################################################## | |
| # Start of modified clusters from internal code: | |
| ################################################################################## | |
| class XiaomiClusterMod(CustomCluster): | |
| """Xiaomi cluster implementation.""" | |
| def _iter_parse_attr_report( | |
| self, data: bytes | |
| ) -> Iterator[tuple[foundation.Attribute, bytes]]: | |
| """Yield all interpretations of the first attribute in a Xiaomi report.""" | |
| # Peek at the attribute report | |
| attr_id, data = t.uint16_t.deserialize(data) | |
| attr_type, data = t.uint8_t.deserialize(data) | |
| if ( | |
| attr_id | |
| not in ( | |
| XIAOMI_AQARA_ATTRIBUTE, | |
| XIAOMI_MIJA_ATTRIBUTE, | |
| XIAOMI_AQARA_ATTRIBUTE_E1, | |
| ) | |
| or attr_type != 0x42 # "Character String" | |
| ): | |
| # Assume other attributes are reported correctly | |
| data = attr_id.serialize() + attr_type.serialize() + data | |
| attribute, data = foundation.Attribute.deserialize(data) | |
| yield attribute, data | |
| return | |
| # Length of the "string" can be wrong | |
| val_len, data = t.uint8_t.deserialize(data) | |
| # Try every offset. Start with 0 to pass unbroken reports through. | |
| for offset in (0, -1, 1): | |
| fixed_len = val_len + offset | |
| if len(data) < fixed_len: | |
| continue | |
| val, final_data = data[:fixed_len], data[fixed_len:] | |
| attr_val = t.LVBytes(val) | |
| attr_type = 0x41 # The data type should be "Octet String" | |
| yield ( | |
| foundation.Attribute( | |
| attrid=attr_id, | |
| value=foundation.TypeValue(type=attr_type, value=attr_val), | |
| ), | |
| final_data, | |
| ) | |
| def _interpret_attr_reports( | |
| self, data: bytes | |
| ) -> Iterable[tuple[foundation.Attribute]]: | |
| """Yield all valid interprations of a Xiaomi attribute report.""" | |
| if not data: | |
| yield () | |
| return | |
| try: | |
| parsed = list(self._iter_parse_attr_report(data)) | |
| except (KeyError, ValueError): | |
| return | |
| for attr, remaining_data in parsed: | |
| for remaining_attrs in self._interpret_attr_reports(remaining_data): | |
| yield (attr,) + remaining_attrs | |
| def deserialize(self, data): | |
| """Deserialize cluster data.""" | |
| hdr, data = foundation.ZCLHeader.deserialize(data) | |
| # Only handle attribute reports differently | |
| if ( | |
| hdr.frame_control.frame_type != foundation.FrameType.GLOBAL_COMMAND | |
| or hdr.command_id != foundation.GeneralCommand.Report_Attributes | |
| ): | |
| return super().deserialize(hdr.serialize() + data) | |
| reports = list(self._interpret_attr_reports(data)) | |
| if not reports: | |
| _LOGGER.warning("Failed to parse Xiaomi attribute report: %r", data) | |
| return super().deserialize(hdr.serialize() + data) | |
| elif len(reports) > 1: | |
| _LOGGER.warning( | |
| "Xiaomi attribute report has multiple valid interpretations: %r", | |
| reports, | |
| ) | |
| fixed_data = b"".join(attr.serialize() for attr in reports[0]) | |
| return super().deserialize(hdr.serialize() + fixed_data) | |
| def _update_attribute(self, attrid, value): | |
| if attrid in (XIAOMI_AQARA_ATTRIBUTE, XIAOMI_AQARA_ATTRIBUTE_E1): | |
| attributes = self._parse_aqara_attributes(value) | |
| super()._update_attribute(attrid, value) | |
| if self.endpoint.device.model == "lumi.sensor_switch.aq2": | |
| if value == b"\x04!\xa8C\n!\x00\x00": | |
| self.listener_event(ZHA_SEND_EVENT, COMMAND_TRIPLE, []) | |
| elif attrid == XIAOMI_MIJA_ATTRIBUTE: | |
| attributes = self._parse_mija_attributes(value) | |
| else: | |
| super()._update_attribute(attrid, value) | |
| if attrid == MODEL: | |
| # 0x0005 = model attribute. | |
| # Xiaomi sensors send the model attribute when their reset button is | |
| # pressed quickly.""" | |
| if attrid in self.attributes: | |
| attribute_name = self.attributes[attrid].name | |
| else: | |
| attribute_name = UNKNOWN | |
| self.listener_event( | |
| ZHA_SEND_EVENT, | |
| COMMAND_ATTRIBUTE_UPDATED, | |
| { | |
| ATTRIBUTE_ID: attrid, | |
| ATTRIBUTE_NAME: attribute_name, | |
| VALUE: value, | |
| }, | |
| ) | |
| return | |
| _LOGGER.debug( | |
| "%s - Xiaomi attribute report. attribute_id: [%s] value: [%s]", | |
| self.endpoint.device.ieee, | |
| attrid, | |
| attributes, | |
| ) | |
| if BATTERY_VOLTAGE_MV in attributes: | |
| # many Xiaomi devices report this, but not all quirks implement the XiaomiPowerConfiguration cluster, | |
| # so we might error out if the method doesn't exist | |
| if hasattr(self.endpoint.power, "battery_reported") and callable( | |
| self.endpoint.power.battery_reported | |
| ): | |
| self.endpoint.power.battery_reported(attributes[BATTERY_VOLTAGE_MV]) | |
| else: | |
| # log a debug message if the cluster is not implemented | |
| _LOGGER.debug( | |
| "%s - Xiaomi battery voltage attribute received but XiaomiPowerConfiguration not used", | |
| self.endpoint.device.ieee, | |
| ) | |
| if TEMPERATURE_MEASUREMENT in attributes: | |
| self.endpoint.temperature.update_attribute( | |
| TemperatureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[TEMPERATURE_MEASUREMENT], | |
| ) | |
| if HUMIDITY_MEASUREMENT in attributes: | |
| self.endpoint.humidity.update_attribute( | |
| RelativeHumidity.AttributeDefs.measured_value.id, | |
| attributes[HUMIDITY_MEASUREMENT], | |
| ) | |
| if PRESSURE_MEASUREMENT in attributes: | |
| self.endpoint.pressure.update_attribute( | |
| PressureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[PRESSURE_MEASUREMENT], | |
| ) | |
| if PRESSURE_MEASUREMENT_PRECISION in attributes: | |
| self.endpoint.pressure.update_attribute( | |
| PressureMeasurement.AttributeDefs.measured_value.id, | |
| attributes[PRESSURE_MEASUREMENT_PRECISION] / 100, | |
| ) | |
| if POWER in attributes: | |
| self.endpoint.electrical_measurement.update_attribute( | |
| ElectricalMeasurement.AttributeDefs.active_power.id, | |
| round(attributes[POWER] * 10), | |
| ) | |
| if CONSUMPTION in attributes: | |
| zcl_consumption = round(attributes[CONSUMPTION] * 1000) | |
| self.endpoint.smartenergy_metering.update_attribute( | |
| Metering.AttributeDefs.current_summ_delivered.id, zcl_consumption | |
| ) | |
| if VOLTAGE in attributes: | |
| self.endpoint.electrical_measurement.update_attribute( | |
| ElectricalMeasurement.AttributeDefs.rms_voltage.id, | |
| attributes[VOLTAGE] * 0.1, | |
| ) | |
| if ILLUMINANCE_MEASUREMENT in attributes: | |
| self.endpoint.illuminance.update_attribute( | |
| IlluminanceMeasurement.AttributeDefs.measured_value.id, | |
| attributes[ILLUMINANCE_MEASUREMENT], | |
| ) | |
| if TVOC_MEASUREMENT in attributes: | |
| self.endpoint.voc_level.update_attribute( | |
| 0x0000, attributes[TVOC_MEASUREMENT] | |
| ) | |
| if TEMPERATURE in attributes: | |
| if hasattr(self.endpoint, "device_temperature"): | |
| self.endpoint.device_temperature.update_attribute( | |
| DeviceTemperature.AttributeDefs.current_temperature.id, | |
| attributes[TEMPERATURE] * 100, | |
| ) | |
| if BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE in attributes: | |
| self.endpoint.power.battery_percent_reported( | |
| attributes[BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE] | |
| ) | |
| if SMOKE in attributes: | |
| self.endpoint.ias_zone.update_attribute( | |
| IasZone.AttributeDefs.zone_status.id, attributes[SMOKE] | |
| ) | |
| def _parse_aqara_attributes(self, value): | |
| """Parse non-standard attributes.""" | |
| attributes = {} | |
| attribute_names = { | |
| 1: BATTERY_VOLTAGE_MV, | |
| 3: TEMPERATURE, | |
| 4: XIAOMI_ATTR_4, | |
| 5: XIAOMI_ATTR_5, | |
| 6: XIAOMI_ATTR_6, | |
| 10: PATH, | |
| } | |
| if self.endpoint.device.model in [ | |
| "lumi.sensor_ht", | |
| "lumi.sens", | |
| "lumi.weather", | |
| "lumi.airmonitor.acn01", | |
| "lumi.sensor_ht.agl02", | |
| ]: | |
| # Temperature sensors send temperature/humidity/pressure updates through this | |
| # cluster instead of the respective clusters | |
| attribute_names.update( | |
| { | |
| 100: TEMPERATURE_MEASUREMENT, | |
| 101: HUMIDITY_MEASUREMENT, | |
| 102: ( | |
| TVOC_MEASUREMENT | |
| if self.endpoint.device.model == "lumi.airmonitor.acn01" | |
| else ( | |
| PRESSURE_MEASUREMENT_PRECISION | |
| if self.endpoint.device.model == "lumi.weather" | |
| else PRESSURE_MEASUREMENT | |
| ) | |
| ), | |
| } | |
| ) | |
| elif self.endpoint.device.model in [ | |
| "lumi.plug", | |
| "lumi.plug.aeu001", | |
| "lumi.plug.maus01", | |
| "lumi.plug.maeu01", | |
| "lumi.plug.mmeu01", | |
| "lumi.relay.c2acn01", | |
| "lumi.switch.n0agl1", | |
| "lumi.switch.n0acn2", | |
| "lumi.switch.n2aeu1", # TODO: BSK add to PR | |
| "lumi.switch.n1aeu1", # TODO: BSK add to PR | |
| ]: | |
| attribute_names.update({149: CONSUMPTION, 150: VOLTAGE, 152: POWER}) | |
| elif self.endpoint.device.model == "lumi.switch.agl011": | |
| attribute_names.update({150: VOLTAGE, 151: CONSUMPTION, 152: POWER}) | |
| elif self.endpoint.device.model == "lumi.sensor_motion.aq2": | |
| attribute_names.update({11: ILLUMINANCE_MEASUREMENT}) | |
| elif self.endpoint.device.model == "lumi.curtain.acn002": | |
| attribute_names.update({101: BATTERY_PERCENTAGE_REMAINING_ATTRIBUTE}) | |
| elif self.endpoint.device.model in [ | |
| "lumi.motion.agl02", | |
| "lumi.motion.ac02", | |
| "lumi.motion.acn001", | |
| ]: | |
| attribute_names.update({101: ILLUMINANCE_MEASUREMENT}) | |
| if self.endpoint.device.model == "lumi.motion.ac02": | |
| attribute_names.update({105: DETECTION_INTERVAL}) | |
| attribute_names.update({106: MOTION_SENSITIVITY}) | |
| elif self.endpoint.device.model == "lumi.motion.agl04": | |
| attribute_names.update({102: DETECTION_INTERVAL}) | |
| attribute_names.update({105: MOTION_SENSITIVITY}) | |
| attribute_names.update({258: DETECTION_INTERVAL}) | |
| attribute_names.update({268: MOTION_SENSITIVITY}) | |
| elif self.endpoint.device.model == "lumi.motion.ac01": | |
| attribute_names.update({5: POWER_OUTAGE_COUNT}) | |
| attribute_names.update({101: PRESENCE_DETECTED}) | |
| attribute_names.update({102: PRESENCE_EVENT}) | |
| attribute_names.update({103: MONITORING_MODE}) | |
| attribute_names.update({105: APPROACH_DISTANCE}) | |
| attribute_names.update({268: MOTION_SENSITIVITY}) | |
| attribute_names.update({322: PRESENCE_DETECTED}) | |
| attribute_names.update({323: PRESENCE_EVENT}) | |
| attribute_names.update({324: MONITORING_MODE}) | |
| attribute_names.update({326: APPROACH_DISTANCE}) | |
| elif self.endpoint.device.model == "lumi.sensor_smoke.acn03": | |
| attribute_names.update({160: SMOKE}) | |
| attribute_names.update({161: SMOKE_DENSITY}) | |
| attribute_names.update({162: SELF_TEST}) | |
| attribute_names.update({163: BUZZER_MANUAL_MUTE}) | |
| attribute_names.update({164: HEARTBEAT_INDICATOR}) | |
| attribute_names.update({165: LINKAGE_ALARM}) | |
| result = {} | |
| # Some attribute reports end with a stray null byte | |
| while value not in (b"", b"\x00"): | |
| skey = int(value[0]) | |
| svalue, value = foundation.TypeValue.deserialize(value[1:]) | |
| result[skey] = svalue.value | |
| for item, val in result.items(): | |
| key = ( | |
| attribute_names[item] | |
| if item in attribute_names | |
| else "0xff01-" + str(item) | |
| ) | |
| attributes[key] = val | |
| return attributes | |
| def _parse_mija_attributes(self, value): | |
| """Parse non-standard attributes.""" | |
| attribute_names = ( | |
| STATE, | |
| BATTERY_VOLTAGE_MV, | |
| XIAOMI_ATTR_3, | |
| XIAOMI_ATTR_4, | |
| XIAOMI_ATTR_5, | |
| XIAOMI_ATTR_6, | |
| ) | |
| result = [] | |
| for attr_value in value: | |
| result.append(attr_value.value) | |
| attributes = dict(zip(attribute_names, result)) | |
| return attributes | |
| class IterableMemberMeta(type): | |
| def __iter__(cls) -> typing.Iterator[typing.Any]: | |
| for name in dir(cls): | |
| if not name.startswith("_"): | |
| yield getattr(cls, name) | |
| class BaseAttributeDefs(metaclass=IterableMemberMeta): | |
| pass | |
| class XiaomiAqaraE1ClusterMod(XiaomiClusterMod): | |
| """Xiaomi mfg cluster implementation.""" | |
| cluster_id = 0xFCC0 | |
| ep_attribute = "opple_cluster" | |
| class AttributeDefs(BaseAttributeDefs): | |
| """Cluster attributes.""" | |
| class OppleClusterMod(XiaomiAqaraE1ClusterMod): | |
| """Opple cluster.""" | |
| attributes = { | |
| 0x0009: ("mode", types.uint8_t, True), | |
| } | |
| attr_config = {0x0009: 0x01} | |
| def __init__(self, *args, **kwargs): | |
| """Init.""" | |
| self._current_state = None | |
| super().__init__(*args, **kwargs) | |
| async def bind(self): | |
| """Bind cluster.""" | |
| result = await super().bind() | |
| await self.write_attributes(self.attr_config, manufacturer=OPPLE_MFG_CODE) | |
| return result | |
| class OppleSwitchClusterMod(OppleClusterMod): | |
| """Xiaomi mfg cluster implementation.""" | |
| class AttributeDefs(OppleClusterMod.AttributeDefs): | |
| """Attribute definitions.""" | |
| power_outage_count: Final = ZCLAttributeDef( | |
| id=0x0002, type=t.uint8_t, is_manufacturer_specific=True | |
| ) | |
| switch_type: Final = ZCLAttributeDef( | |
| id=0x000A, type=OppleSwitchType, is_manufacturer_specific=True | |
| ) | |
| reverse_indicator_light: Final = ZCLAttributeDef( | |
| id=0x00F0, type=OppleIndicatorLight, is_manufacturer_specific=True | |
| ) | |
| switch_mode: Final = ZCLAttributeDef( | |
| id=0x0125, type=OppleSwitchMode, is_manufacturer_specific=True | |
| ) | |
| operation_mode: Final = ZCLAttributeDef( | |
| id=0x0200, type=OppleOperationMode, is_manufacturer_specific=True | |
| ) | |
| power_outage_memory: Final = ZCLAttributeDef( | |
| id=0x0201, type=t.Bool, is_manufacturer_specific=True | |
| ) | |
| auto_off: Final = ZCLAttributeDef( | |
| id=0x0202, type=t.Bool, is_manufacturer_specific=True | |
| ) | |
| do_not_disturb: Final = ZCLAttributeDef( | |
| id=0x0203, type=t.Bool, is_manufacturer_specific=True | |
| ) | |
| def _update_attribute(self, attrid, value): | |
| super()._update_attribute(attrid, value) | |
| if attrid == 0x00FC: | |
| self._current_state = PRESS_TYPES.get(value) | |
| event_args = { | |
| BUTTON: self.endpoint.endpoint_id, | |
| PRESS_TYPE: self._current_state, | |
| ATTR_ID: attrid, | |
| VALUE: value, | |
| } | |
| action = f"{self.endpoint.endpoint_id}_{self._current_state}" | |
| self.listener_event(ZHA_SEND_EVENT, action, event_args) | |
| # show something in the sensor in HA | |
| super()._update_attribute(0, action) | |
| ################################################################################## | |
| # End of modified clusters from internal code. | |
| ################################################################################## | |
| ( | |
| QuirkBuilder(LUMI, "lumi.switch.n2aeu1") | |
| .friendly_name( | |
| model="Wall Switch H1 EU Double (With Neutral)", manufacturer="Aqara" | |
| ) | |
| .filter(signature_matches(XiaomiOpple2ButtonSwitch1.signature)) | |
| .replace_cluster_occurrences(BasicCluster) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(OnOffCluster) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(OppleSwitchClusterMod) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(MultistateInputCluster) # endpoints: 1, 2, 41, 42, 51 | |
| .replace_cluster_occurrences(AnalogInputCluster) # endpoints: 21 | |
| .replaces(DeviceTemperatureCluster) | |
| .adds(Alarms) | |
| .adds(MeteringCluster) | |
| .adds(ElectricalMeasurementCluster) | |
| .device_automation_triggers(XiaomiOpple2ButtonSwitchBase.device_automation_triggers) | |
| .enum( | |
| attribute_name="operation_mode", | |
| enum_class=OppleOperationMode, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="operation_mode_left", | |
| fallback_name="Operation mode (left)", | |
| ) | |
| .enum( | |
| attribute_name="operation_mode", | |
| enum_class=OppleOperationMode, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="operation_mode_right", | |
| fallback_name="Operation mode (right)", | |
| ) | |
| .switch( | |
| attribute_name="power_outage_memory", | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="power_outage_memory_left", | |
| fallback_name="Power outage memory (left)", | |
| ) | |
| .switch( | |
| attribute_name="power_outage_memory", | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="power_outage_memory_right", | |
| fallback_name="Power outage memory (right)", | |
| ) | |
| .enum( | |
| attribute_name="reverse_indicator_light", | |
| enum_class=OppleIndicatorLight, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="reverse_indicator_light_left", | |
| fallback_name="Reverse indicator light (left)", | |
| ) | |
| .enum( | |
| attribute_name="reverse_indicator_light", | |
| enum_class=OppleIndicatorLight, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="reverse_indicator_light_right", | |
| fallback_name="Reverse indicator light (right)", | |
| ) | |
| .add_to_registry() | |
| ) | |
| ( | |
| QuirkBuilder(LUMI, "lumi.switch.n2aeu1") | |
| .friendly_name( | |
| model="Wall Switch H1 EU Double (With Neutral)", manufacturer="Aqara" | |
| ) | |
| .filter(signature_matches(XiaomiOpple2ButtonSwitch3.signature)) | |
| .replace_cluster_occurrences(BasicCluster) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(OnOffCluster) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(OppleSwitchClusterMod) # endpoints: 1, 2 | |
| .replace_cluster_occurrences(MultistateInputCluster) # endpoints: 1, 2, 41, 42, 51 | |
| .replaces(DeviceTemperatureCluster) | |
| .adds(Alarms) | |
| .adds(MeteringCluster) | |
| .adds(ElectricalMeasurementCluster) | |
| .adds(OppleSwitchClusterMod) | |
| .device_automation_triggers(XiaomiOpple2ButtonSwitchBase.device_automation_triggers) | |
| .enum( | |
| attribute_name="operation_mode", | |
| enum_class=OppleOperationMode, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="operation_mode_left", | |
| fallback_name="Operation mode (left)", | |
| ) | |
| .enum( | |
| attribute_name="operation_mode", | |
| enum_class=OppleOperationMode, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="operation_mode_right", | |
| fallback_name="Operation mode (right)", | |
| ) | |
| .switch( | |
| attribute_name="power_outage_memory", | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="power_outage_memory_left", | |
| fallback_name="Power outage memory (left)", | |
| ) | |
| .switch( | |
| attribute_name="power_outage_memory", | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="power_outage_memory_right", | |
| fallback_name="Power outage memory (right)", | |
| ) | |
| .enum( | |
| attribute_name="reverse_indicator_light", | |
| enum_class=OppleIndicatorLight, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="reverse_indicator_light_left", | |
| fallback_name="Reverse indicator light (left)", | |
| ) | |
| .enum( | |
| attribute_name="reverse_indicator_light", | |
| enum_class=OppleIndicatorLight, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=2, | |
| translation_key="reverse_indicator_light_right", | |
| fallback_name="Reverse indicator light (right)", | |
| ) | |
| .add_to_registry() | |
| ) | |
| ( | |
| QuirkBuilder(LUMI, "lumi.switch.n1aeu1") | |
| .friendly_name( | |
| model="Wall Switch H1 EU Single (With Neutral)", manufacturer="Aqara" | |
| ) | |
| .filter(signature_matches(AqaraH1SingleRockerSwitchWithNeutral.signature)) | |
| .replaces(BasicCluster) | |
| .replaces(OnOffCluster) | |
| .replaces(MultistateInputCluster) | |
| .adds(MeteringCluster) | |
| .adds(ElectricalMeasurementCluster) | |
| .replaces(OppleSwitchClusterMod) | |
| .replaces(AnalogInputCluster, endpoint_id=21) | |
| .replaces(MultistateInputCluster, endpoint_id=41) | |
| .device_automation_triggers(AqaraH1SingleRockerBase.device_automation_triggers) | |
| .enum( | |
| attribute_name="operation_mode", | |
| enum_class=OppleOperationMode, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="operation_mode", | |
| fallback_name="Operation mode", | |
| ) | |
| .switch( | |
| attribute_name="power_outage_memory", | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="power_outage_memory", | |
| fallback_name="Power outage memory", | |
| ) | |
| .enum( | |
| attribute_name="reverse_indicator_light", | |
| enum_class=OppleIndicatorLight, | |
| cluster_id=OppleSwitchClusterMod.cluster_id, | |
| endpoint_id=1, | |
| translation_key="reverse_indicator_light", | |
| fallback_name="Reverse indicator light", | |
| ) | |
| .add_to_registry() | |
| ) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment