diff --git a/zhaquirks/tuya/ts011f_breaker.py b/zhaquirks/tuya/ts011f_breaker.py new file mode 100644 index 0000000000..c5ac7f195e --- /dev/null +++ b/zhaquirks/tuya/ts011f_breaker.py @@ -0,0 +1,372 @@ +"""TS011F Circuit Breaker * Tongou TO-Q-SY2-JZT.""" + +import logging +from struct import iter_unpack, pack +from typing import Any, Optional, Union + +from zigpy.quirks.v2 import CustomDeviceV2, QuirkBuilder +from zigpy.quirks.v2.homeassistant import ( + UnitOfElectricCurrent, + UnitOfElectricPotential, + UnitOfPower, + UnitOfTemperature, +) +from zigpy.quirks.v2.homeassistant.number import NumberDeviceClass +import zigpy.types as t +from zigpy.zcl import foundation + +from zhaquirks import LocalDataCluster +from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF +from zhaquirks.tuya import ( + EnchantedDevice, + TuyaZBE000Cluster, + TuyaZBElectricalMeasurement, + TuyaZBExternalSwitchTypeCluster, + TuyaZBMeteringCluster, + TuyaZBOnOffAttributeCluster, +) + +_LOGGER = logging.getLogger("ts011f_breaker") + +TUYA_OPTIONS_2_DATA = 0xE6 +TUYA_OPTIONS_3_DATA = 0xE7 + + +class Breaker(t.enum8): + Off = 0x00 + On = 0x01 + + +class TuyaZBExternalSwitchTypeThresholdCluster( + LocalDataCluster, TuyaZBExternalSwitchTypeCluster +): + """Tuya External Switch Type With Threshold Cluster.""" + + name = "Tuya External Switch Type With Threshold Cluster" + ep_attribute = "tuya_external_switch_type_threshold" + + class AttributeDefs(TuyaZBExternalSwitchTypeCluster.AttributeDefs): + """Attribute definitions.""" + + temperature_breaker = foundation.ZCLAttributeDef( + id=0xE605, + type=Breaker, + is_manufacturer_specific=True, + zcl_type=foundation.DataTypeId.uint8, + ) + temperature_threshold = foundation.ZCLAttributeDef( + id=0xE685, type=t.uint16_t, is_manufacturer_specific=True + ) + power_breaker = foundation.ZCLAttributeDef( + id=0xE607, + type=Breaker, + is_manufacturer_specific=True, + zcl_type=foundation.DataTypeId.uint8, + ) + power_threshold = foundation.ZCLAttributeDef( + id=0xE687, type=t.uint16_t, is_manufacturer_specific=True + ) + over_current_breaker = foundation.ZCLAttributeDef( + id=0xE701, + type=Breaker, + is_manufacturer_specific=True, + zcl_type=foundation.DataTypeId.uint8, + ) + over_current_threshold = foundation.ZCLAttributeDef( + id=0xE781, type=t.uint16_t, is_manufacturer_specific=True + ) + over_voltage_breaker = foundation.ZCLAttributeDef( + id=0xE703, + type=Breaker, + is_manufacturer_specific=True, + zcl_type=foundation.DataTypeId.uint8, + ) + over_voltage_threshold = foundation.ZCLAttributeDef( + id=0xE783, type=t.uint16_t, is_manufacturer_specific=True + ) + under_voltage_breaker = foundation.ZCLAttributeDef( + id=0xE704, + type=Breaker, + is_manufacturer_specific=True, + zcl_type=foundation.DataTypeId.uint8, + ) + under_voltage_threshold = foundation.ZCLAttributeDef( + id=0xE784, type=t.uint16_t, is_manufacturer_specific=True + ) + + class ServerCommandDefs(TuyaZBExternalSwitchTypeCluster.ServerCommandDefs): + """Server command definitions.""" + + set_options_2 = foundation.ZCLCommandDef( + TUYA_OPTIONS_2_DATA, + {"data?": t.SerializableBytes}, + is_manufacturer_specific=True, + ) + set_options_3 = foundation.ZCLCommandDef( + TUYA_OPTIONS_3_DATA, + {"data?": t.SerializableBytes}, + is_manufacturer_specific=True, + ) + + def handle_cluster_request( + self, + hdr: foundation.ZCLHeader, + args: tuple, + *, + dst_addressing: Optional[ + Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK] + ] = None, + ) -> None: + """Handle cluster request.""" + data = args + + _LOGGER.debug( + "[0x%04x:%s:0x%04x] Received value %s " + "for attribute 0x%04x (command 0x%04x)", + self.endpoint.device.nwk, + self.endpoint.endpoint_id, + self.cluster_id, + repr(data), + hdr.command_id, + hdr.command_id, + ) + + if hdr.command_id in (TUYA_OPTIONS_2_DATA, TUYA_OPTIONS_3_DATA): + for attr_id, breaker, threshold in iter_unpack(">bbH", data): + self._update_attribute((hdr.command_id << 8) + attr_id, breaker) + self._update_attribute( + (hdr.command_id << 8) + 0x80 + attr_id, threshold + ) + + super().handle_cluster_request(hdr, args, dst_addressing=dst_addressing) + + async def write_attributes(self, attributes, manufacturer=None): + """Defer attributes writing to the set_options_* command.""" + + local, remote = {}, {} + + remote_attr_ids = list( + map( + lambda attrDef: attrDef.id, + TuyaZBExternalSwitchTypeCluster.AttributeDefs, + ) + ) + + for key, value in attributes.items(): + if ( + key in TuyaZBExternalSwitchTypeCluster.AttributeDefs + or key in remote_attr_ids + ): + remote[key] = value + else: + local[key] = value + + _LOGGER.debug( + "write_attributes attrs: %s local: %s remote %s", + repr(attributes), + repr(local), + repr(remote), + ) + + if local: + records = self._write_attr_records(local) + + _LOGGER.debug("write_attributes records: %s ", repr(records)) + + command_attributes = {TUYA_OPTIONS_2_DATA: {}, TUYA_OPTIONS_3_DATA: {}} + + for attribute in records: + attr_id = attribute.attrid + command_id = attr_id >> 8 + comp_attr_id = attr_id ^ 0x80 + if attr_id not in command_attributes[command_id]: + if comp_attr_id in local: + comp_attr = next( + filter(lambda a: a.id == comp_attr_id, records), None + ) + comp_value = comp_attr.value.value + else: + comp_value = self.get(comp_attr_id) + + if comp_value != None: + command_attributes[command_id][attr_id & 0x7F] = { + ((attr_id & 0x80) >> 7): attribute.value.value, + ((comp_attr_id & 0x80) >> 7): comp_value, + } + + for command_id, command_attribute in command_attributes.items(): + if command_attribute: + data = bytearray(b"") + for attr_id, values in command_attribute.items(): + data.extend(pack(">bbH", attr_id, values[0], values[1])) + + await super().command(command_id, data) + + if remote: + await TuyaZBExternalSwitchTypeCluster.write_attributes( + self, remote, manufacturer + ) + + return [[foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)]] + + async def read_attributes( + self, + attributes: list[int | str], + allow_cache: bool = False, + only_cache: bool = False, + manufacturer: int | t.uint16_t | None = None, + ) -> Any: + local_success, local_failure = {}, {} + remote_success, remote_failure = {}, {} + local, remote = [], [] + + remote_attr_ids = list( + map( + lambda attrDef: attrDef.id, + TuyaZBExternalSwitchTypeCluster.AttributeDefs, + ) + ) + + for attribute in attributes: + if isinstance(attribute, str): + attrid = self.attributes_by_name[attribute].id + else: + # Allow reading attributes that aren't defined + attrid = attribute + + if attrid in remote_attr_ids: + remote.append(attrid) + else: + local.append(attrid) + + _LOGGER.debug( + "read_attributes attrs: %s local: %s remote %s", + repr(attributes), + repr(local), + repr(remote), + ) + + if local: + local_success, local_failure = await LocalDataCluster.read_attributes( + self, local, allow_cache, only_cache, manufacturer + ) + + if remote: + ( + remote_success, + remote_failure, + ) = await TuyaZBExternalSwitchTypeCluster.read_attributes( + self, remote, allow_cache, only_cache, manufacturer + ) + + return local_success | remote_success, local_failure | remote_failure + + +class CB_Metering_Threshold(CustomDeviceV2, EnchantedDevice): + """Circuit breaker with monitoring, e.g. Tongou TO-Q-SY2-JZT. First one using this definition was _TZ3000_cayepv1a.""" + + quirk_id = TUYA_PLUG_ONOFF + + +( + QuirkBuilder("_TZ3000_cayepv1a", "TS011F") + .also_applies_to("_TZ3000_lepzuhto", "TS011F") + .also_applies_to("_TZ3000_qystbcjg", "TS011F") + .device_class(CB_Metering_Threshold) + .replaces(TuyaZBOnOffAttributeCluster) + .replaces(TuyaZBMeteringCluster) + .replaces(TuyaZBElectricalMeasurement) + .replaces(TuyaZBE000Cluster) + .replaces(TuyaZBExternalSwitchTypeThresholdCluster) + .switch( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.temperature_breaker.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + fallback_name="Temperature Breaker", + translation_key=TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.temperature_breaker.name, + ) + .number( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.temperature_threshold.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + min_value=40, + max_value=100, + unit=UnitOfTemperature.CELSIUS, + mode="box", + fallback_name="Temperature Threshold", + device_class=NumberDeviceClass.TEMPERATURE, + ) + .switch( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.power_breaker.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + fallback_name="Power Breaker", + translation_key=TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.power_breaker.name, + ) + .number( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.power_threshold.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + min_value=1, + max_value=26, + unit=UnitOfPower.KILO_WATT, + mode="box", + fallback_name="Power Threshold", + device_class=NumberDeviceClass.POWER, + ) + .switch( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_current_breaker.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + fallback_name="Over Current Breaker", + translation_key=TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_current_breaker.name, + ) + .number( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_current_threshold.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + min_value=1, + max_value=63, + unit=UnitOfElectricCurrent.AMPERE, + mode="box", + fallback_name="Over Current Threshold", + device_class=NumberDeviceClass.CURRENT, + ) + .switch( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_voltage_breaker.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + fallback_name="Over Voltage Breaker", + translation_key=TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_voltage_breaker.name, + ) + .number( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.over_voltage_threshold.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + min_value=230, + max_value=265, + unit=UnitOfElectricPotential.VOLT, + mode="box", + fallback_name="Over Voltage Threshold", + device_class=NumberDeviceClass.VOLTAGE, + ) + .switch( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.under_voltage_breaker.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + fallback_name="Under Voltage Breaker", + translation_key=TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.under_voltage_breaker.name, + ) + .number( + TuyaZBExternalSwitchTypeThresholdCluster.AttributeDefs.under_voltage_threshold.name, + TuyaZBExternalSwitchTypeThresholdCluster.cluster_id, + attribute_initialized_from_cache=False, + min_value=75, + max_value=240, + unit=UnitOfElectricPotential.VOLT, + mode="box", + fallback_name="Under Voltage Threshold", + device_class=NumberDeviceClass.VOLTAGE, + ) + .add_to_registry() +)