[Device Support Request] Tongou TO-Q-SY2-163JZT Smart Circuit Breaker

[MnM001]

Problem description

When adding TO-Q-SY2-163JZT to ZHA is coming up as TS011F by _TZ3000_cayepv1a - however this seems to be another product - TO-Q-SY1.
TO-Q-SY1 can only do monitoring.

TO-Q-SY2-163JZT can do monitoring and more: MCB Smart Circuit Breaker Over Current Under Voltage Protection Power Metering 1-63A Remote Control Switch.

However none of the circuit breakers are exposed to ZHA.

It will be good if this can be fixed.

Solution description

TO-Q-SY2-163JZT needs to have the following in ZHA:

temperature protection
over current protection
over voltage protection
under voltage protection
high power protection
timing
and adjustable Functions for the above

as well functional real-time energy consumption monitoring (I say functional as current monitoring is showing wrong values - see
#2652 )

This unit is fully functional in zigbee2mqtt (https://www.zigbee2mqtt.io/devices/TO-Q-SY2-163JZT.html)

Screenshots/Video

Screenshots from zigbee2mqtt

Screenshots from ZHA

Device signature

Device signature

{
  "node_descriptor": "NodeDescriptor(logical_type=<LogicalType.Router: 1>, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=<FrequencyBand.Freq2400MHz: 8>, mac_capability_flags=<MACCapabilityFlags.FullFunctionDevice|MainsPowered|RxOnWhenIdle|AllocateAddress: 142>, manufacturer_code=4098, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=<DescriptorCapability.NONE: 0>, *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": "0x0104",
      "device_type": "0x010a",
      "input_clusters": [
        "0x0000",
        "0x0003",
        "0x0004",
        "0x0005",
        "0x0006",
        "0x0402",
        "0x0702",
        "0x0b04",
        "0xe000",
        "0xe001"
      ],
      "output_clusters": [
        "0x000a",
        "0x0019"
      ]
    },
    "242": {
      "profile_id": "0xa1e0",
      "device_type": "0x0061",
      "input_clusters": [],
      "output_clusters": [
        "0x0021"
      ]
    }
  },
  "manufacturer": "_TZ3000_cayepv1a",
  "model": "TS011F",
  "class": "zigpy.device.Device"
}

Diagnostic information

Diagnostic information

[Paste the diagnostic information here]

Logs

Logs

[Paste the logs here]

Custom quirk

Custom quirk

[Paste your custom quirk here]

Additional information

No response

[Milhauz666]

This is just a tweaked quirk reusing similar definition for _TZ3000_qeuvnohg, and my comment here.

Current and Summation seem to work correctly. The temperature does not work and other circuit breaker functions have not been added. It's definitely not perfect and I'd appreciate any advice.

"""TS011F Circuit Breaker - Tongou TO-Q-SY2-JZT."""

from typing import Any, Dict

from zigpy.profiles import zgp, zha
from zigpy.quirks import CustomDevice
from zigpy.zcl.clusters.general import (
    Basic,
    GreenPowerProxy,
    Groups,
    Identify,
    OnOff,
    Ota,
    Scenes,
    Time,
)
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.lightlink import LightLink
from zigpy.zcl.clusters.measurement import TemperatureMeasurement
from zigpy.zcl.clusters.smartenergy import Metering

from zhaquirks.const import (
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODEL,
    MODELS_INFO,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
)
from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF
from zhaquirks.tuya import (
    EnchantedDevice,
    TuyaLocalCluster,
    TuyaManufCluster,
    TuyaNewManufCluster,
    TuyaZB1888Cluster,
    TuyaZBE000Cluster,
    TuyaZBElectricalMeasurement,
    TuyaZBExternalSwitchTypeCluster,
    TuyaZBMeteringCluster,
    TuyaZBMeteringClusterWithUnit,
    TuyaZBOnOffAttributeCluster,
)

from zhaquirks.tuya.mcu import (
    DPToAttributeMapping,
    EnchantedDevice,
    TuyaMCUCluster,
    TuyaPowerConfigurationCluster,
)
from zhaquirks.tuya import TuyaDPType

class TuyaTemperatureMeasurement(TemperatureMeasurement, TuyaLocalCluster):
    """Tuya local TemperatureMeasurement cluster."""

class TemperatureHumidityManufCluster(TuyaMCUCluster):
    """Tuya Manufacturer Cluster with Temperature data point."""

    dp_to_attribute: Dict[int, DPToAttributeMapping] = {
        1: DPToAttributeMapping(
            TuyaTemperatureMeasurement.ep_attribute,
            "measured_value",
            converter=lambda x: x * 10,  # decidegree to centidegree
        ),
    }

    data_point_handlers = {
        1: "_dp_2_attr_update",
    }

class Plug_CB_Metering_v2(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
   
    signature = {
        MODEL: "TS011F",
        MODELS_INFO: [("_TZ3000_cayepv1a", "TS011F")],
        ENDPOINTS: {
            # <SimpleDescriptor endpoint=1 profile=260 device_type=266
            # device_version=1
            # input_clusters=[0, 3, 4, 5, 6, 1794, 2820, 1026, 57344, 57345]
            # output_clusters=[25, 10]>
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    OnOff.cluster_id,
                    Metering.cluster_id,
                    ElectricalMeasurement.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    TuyaZBE000Cluster.cluster_id,
                    TuyaZBExternalSwitchTypeCluster.cluster_id,
                ],
                OUTPUT_CLUSTERS: [Ota.cluster_id, Time.cluster_id],
            },
            # <SimpleDescriptor endpoint=242 profile=41440 device_type=97
            # device_version=1
            # input_clusters=[]
            # output_clusters=[33]>
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }
    replacement = {
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    TuyaZBOnOffAttributeCluster,
                    TuyaZBMeteringCluster,
                    TuyaZBElectricalMeasurement,
                    TuyaTemperatureMeasurement,
                    TuyaZBE000Cluster,
                    TuyaZBExternalSwitchTypeCluster,
                ],
                OUTPUT_CLUSTERS: [Ota.cluster_id, Time.cluster_id],
            },
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }

[neildsb]

Sorry to hijack a thread,
I have a TO-Q-SY2-163JZT in the post and I plan to use with ZHA, can I assume the default is off for the 'breaker' protection variables and the max amp is 64?
My primary use case is on/off (passing up-to 32A), anything else is a bonus!
Hopefully the parameters will the exposed in ZHA soon to configure and monitor :-)

[franortiz]

I am new to zha and zigpy, but after looking the Z2M code and some debug with the device i was able to create this quirk and control the thresholds and breakers
Basically i created some fake attributes that converts to custom command when you write it.
I dont know if this is the zha way but it seems to work

Please, test it and will try to make a proper PR if everything is OK

"""TS011F Circuit Breaker * Tongou TO-Q-SY2-JZT."""

from typing import Any, Optional, Union
import logging
import enum
from struct import (iter_unpack, pack)

from zigpy.profiles import zgp, zha
from zigpy.quirks import CustomDevice
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.general import (
    Basic,
    GreenPowerProxy,
    Groups,
    Identify,
    OnOff,
    Ota,
    Scenes,
    Time,
)
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.lightlink import LightLink
from zigpy.zcl.clusters.measurement import TemperatureMeasurement
from zigpy.zcl.clusters.smartenergy import Metering

from zhaquirks import LocalDataCluster
from zhaquirks.const import (
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODEL,
    MODELS_INFO,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
)
from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF
from zhaquirks.tuya import (
    EnchantedDevice,
    TuyaNewManufCluster,
    TuyaZB1888Cluster,
    TuyaZBE000Cluster,
    TuyaZBElectricalMeasurement,
    TuyaZBExternalSwitchTypeCluster,
    TuyaZBMeteringCluster,
    TuyaZBMeteringClusterWithUnit,
    TuyaZBOnOffAttributeCluster,
    TuyaLocalCluster,
)

_LOGGER = logging.getLogger("ts011f")

TUYA_OPTIONS_2_DATA = 0xE6
TUYA_OPTIONS_3_DATA = 0xE7

class Breaker(t.enum8):
    Off = 0x00
    On = 0x01

breakeable_threshold_attributes = {
    0xe605: ("temperature_breaker", Breaker),
    0xe685: ("temperature_threshold", t.uint16_t),
    0xe607: ("power_breaker", Breaker),
    0xe687: ("power_threshold", t.uint16_t),
    0xe701: ("over_current_breaker", Breaker),
    0xe781: ("over_current_threshold", t.uint16_t),
    0xe703: ("over_voltage_breaker", Breaker),
    0xe783: ("over_voltage_threshold", t.uint16_t),
    0xe704: ("under_voltage_breaker", Breaker),
    0xe784: ("under_voltage_threshold", t.uint16_t),
}

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"

    attributes = TuyaZBExternalSwitchTypeCluster.attributes.copy()
    attributes.update(breakeable_threshold_attributes)

    server_commands = TuyaZBExternalSwitchTypeCluster.server_commands.copy()
    server_commands.update(
        {
            TUYA_OPTIONS_2_DATA: foundation.ZCLCommandDef(
                "set_options_2",
                {"data?": t.SerializableBytes},
                False,
                is_manufacturer_specific=True,
            ),
            TUYA_OPTIONS_3_DATA: foundation.ZCLCommandDef(
                "set_options_3",
                {"data?": t.SerializableBytes},
                False,
                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_attr_ids = breakeable_threshold_attributes.keys()

        local = dict(filter(lambda a: a[0] in local_attr_ids, attributes.items()))
        remote = dict(filter(lambda a: a[0] not in local_attr_ids, attributes.items()))

        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 not attr_id 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 = [], []

        local_attr_ids = breakeable_threshold_attributes.keys()
        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 local_attr_ids:
                local.append(attrid)
            else:
                remote.append(attrid)


        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 Plug_CB_Metering_Threshold(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

    signature = {
        MODELS_INFO: [("_TZ3000_cayepv1a", "TS011F")],
        ENDPOINTS: {
            # <SimpleDescriptor endpoint=1 profile=260 device_type=266
            # device_version=1
            # input_clusters=[0, 3, 4, 5, 6, 1026, 1794, 2820, 57344, 57345]
            # output_clusters=[10, 25]>
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    OnOff.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    Metering.cluster_id,
                    ElectricalMeasurement.cluster_id,
                    TuyaZBE000Cluster.cluster_id,
                    TuyaZBExternalSwitchTypeCluster.cluster_id,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            # <SimpleDescriptor endpoint=242 profile=41440 device_type=97
            # device_version=1
            # input_clusters=[]
            # output_clusters=[33]>
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }
    replacement = {
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    TuyaZBOnOffAttributeCluster,
                    TuyaZBMeteringCluster,
                    TuyaZBElectricalMeasurement,
                    TuyaZBE000Cluster,
                    TuyaZBExternalSwitchTypeThresholdCluster,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }

[drudgebg]

I confirm that it works with my _TZ3000_cayepv1a.

I am new to zha and zigpy, but after looking the Z2M code and some debug with the device i was able to create this quirk and control the thresholds and breakers Basically i created some fake attributes that converts to custom command when you write it. I dont know if this is the zha way but it seems to work

Please, test it and will try to make a proper PR if everything is OK

"""TS011F Circuit Breaker * Tongou TO-Q-SY2-JZT."""

from typing import Any, Optional, Union
import logging
import enum
from struct import (iter_unpack, pack)

from zigpy.profiles import zgp, zha
from zigpy.quirks import CustomDevice
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.general import (
    Basic,
    GreenPowerProxy,
    Groups,
    Identify,
    OnOff,
    Ota,
    Scenes,
    Time,
)
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.lightlink import LightLink
from zigpy.zcl.clusters.measurement import TemperatureMeasurement
from zigpy.zcl.clusters.smartenergy import Metering

from zhaquirks import LocalDataCluster
from zhaquirks.const import (
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODEL,
    MODELS_INFO,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
)
from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF
from zhaquirks.tuya import (
    EnchantedDevice,
    TuyaNewManufCluster,
    TuyaZB1888Cluster,
    TuyaZBE000Cluster,
    TuyaZBElectricalMeasurement,
    TuyaZBExternalSwitchTypeCluster,
    TuyaZBMeteringCluster,
    TuyaZBMeteringClusterWithUnit,
    TuyaZBOnOffAttributeCluster,
    TuyaLocalCluster,
)

_LOGGER = logging.getLogger("ts011f")

TUYA_OPTIONS_2_DATA = 0xE6
TUYA_OPTIONS_3_DATA = 0xE7

class Breaker(t.enum8):
    Off = 0x00
    On = 0x01

breakeable_threshold_attributes = {
    0xe605: ("temperature_breaker", Breaker),
    0xe685: ("temperature_threshold", t.uint16_t),
    0xe607: ("power_breaker", Breaker),
    0xe687: ("power_threshold", t.uint16_t),
    0xe701: ("over_current_breaker", Breaker),
    0xe781: ("over_current_threshold", t.uint16_t),
    0xe703: ("over_voltage_breaker", Breaker),
    0xe783: ("over_voltage_threshold", t.uint16_t),
    0xe704: ("under_voltage_breaker", Breaker),
    0xe784: ("under_voltage_threshold", t.uint16_t),
}

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"

    attributes = TuyaZBExternalSwitchTypeCluster.attributes.copy()
    attributes.update(breakeable_threshold_attributes)

    server_commands = TuyaZBExternalSwitchTypeCluster.server_commands.copy()
    server_commands.update(
        {
            TUYA_OPTIONS_2_DATA: foundation.ZCLCommandDef(
                "set_options_2",
                {"data?": t.SerializableBytes},
                False,
                is_manufacturer_specific=True,
            ),
            TUYA_OPTIONS_3_DATA: foundation.ZCLCommandDef(
                "set_options_3",
                {"data?": t.SerializableBytes},
                False,
                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_attr_ids = breakeable_threshold_attributes.keys()

        local = dict(filter(lambda a: a[0] in local_attr_ids, attributes.items()))
        remote = dict(filter(lambda a: a[0] not in local_attr_ids, attributes.items()))

        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 not attr_id 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 = [], []

        local_attr_ids = breakeable_threshold_attributes.keys()
        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 local_attr_ids:
                local.append(attrid)
            else:
                remote.append(attrid)


        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 Plug_CB_Metering_Threshold(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

    signature = {
        MODELS_INFO: [("_TZ3000_cayepv1a", "TS011F")],
        ENDPOINTS: {
            # <SimpleDescriptor endpoint=1 profile=260 device_type=266
            # device_version=1
            # input_clusters=[0, 3, 4, 5, 6, 1026, 1794, 2820, 57344, 57345]
            # output_clusters=[10, 25]>
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    OnOff.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    Metering.cluster_id,
                    ElectricalMeasurement.cluster_id,
                    TuyaZBE000Cluster.cluster_id,
                    TuyaZBExternalSwitchTypeCluster.cluster_id,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            # <SimpleDescriptor endpoint=242 profile=41440 device_type=97
            # device_version=1
            # input_clusters=[]
            # output_clusters=[33]>
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }
    replacement = {
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    TuyaZBOnOffAttributeCluster,
                    TuyaZBMeteringCluster,
                    TuyaZBElectricalMeasurement,
                    TuyaZBE000Cluster,
                    TuyaZBExternalSwitchTypeThresholdCluster,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }

[franortiz]

Future improvements will be available in this gist

I added support for other compatible devices

[garawas]

I am new to zha and zigpy, but after looking the Z2M code and some debug with the device i was able to create this quirk and control the thresholds and breakers Basically i created some fake attributes that converts to custom command when you write it. I dont know if this is the zha way but it seems to work

Please, test it and will try to make a proper PR if everything is OK

"""TS011F Circuit Breaker * Tongou TO-Q-SY2-JZT."""

from typing import Any, Optional, Union
import logging
import enum
from struct import (iter_unpack, pack)

from zigpy.profiles import zgp, zha
from zigpy.quirks import CustomDevice
import zigpy.types as t
from zigpy.zcl import foundation
from zigpy.zcl.clusters.general import (
    Basic,
    GreenPowerProxy,
    Groups,
    Identify,
    OnOff,
    Ota,
    Scenes,
    Time,
)
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.lightlink import LightLink
from zigpy.zcl.clusters.measurement import TemperatureMeasurement
from zigpy.zcl.clusters.smartenergy import Metering

from zhaquirks import LocalDataCluster
from zhaquirks.const import (
    DEVICE_TYPE,
    ENDPOINTS,
    INPUT_CLUSTERS,
    MODEL,
    MODELS_INFO,
    OUTPUT_CLUSTERS,
    PROFILE_ID,
)
from zhaquirks.quirk_ids import TUYA_PLUG_ONOFF
from zhaquirks.tuya import (
    EnchantedDevice,
    TuyaNewManufCluster,
    TuyaZB1888Cluster,
    TuyaZBE000Cluster,
    TuyaZBElectricalMeasurement,
    TuyaZBExternalSwitchTypeCluster,
    TuyaZBMeteringCluster,
    TuyaZBMeteringClusterWithUnit,
    TuyaZBOnOffAttributeCluster,
    TuyaLocalCluster,
)

_LOGGER = logging.getLogger("ts011f")

TUYA_OPTIONS_2_DATA = 0xE6
TUYA_OPTIONS_3_DATA = 0xE7

class Breaker(t.enum8):
    Off = 0x00
    On = 0x01

breakeable_threshold_attributes = {
    0xe605: ("temperature_breaker", Breaker),
    0xe685: ("temperature_threshold", t.uint16_t),
    0xe607: ("power_breaker", Breaker),
    0xe687: ("power_threshold", t.uint16_t),
    0xe701: ("over_current_breaker", Breaker),
    0xe781: ("over_current_threshold", t.uint16_t),
    0xe703: ("over_voltage_breaker", Breaker),
    0xe783: ("over_voltage_threshold", t.uint16_t),
    0xe704: ("under_voltage_breaker", Breaker),
    0xe784: ("under_voltage_threshold", t.uint16_t),
}

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"

    attributes = TuyaZBExternalSwitchTypeCluster.attributes.copy()
    attributes.update(breakeable_threshold_attributes)

    server_commands = TuyaZBExternalSwitchTypeCluster.server_commands.copy()
    server_commands.update(
        {
            TUYA_OPTIONS_2_DATA: foundation.ZCLCommandDef(
                "set_options_2",
                {"data?": t.SerializableBytes},
                False,
                is_manufacturer_specific=True,
            ),
            TUYA_OPTIONS_3_DATA: foundation.ZCLCommandDef(
                "set_options_3",
                {"data?": t.SerializableBytes},
                False,
                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_attr_ids = breakeable_threshold_attributes.keys()

        local = dict(filter(lambda a: a[0] in local_attr_ids, attributes.items()))
        remote = dict(filter(lambda a: a[0] not in local_attr_ids, attributes.items()))

        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 not attr_id 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 = [], []

        local_attr_ids = breakeable_threshold_attributes.keys()
        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 local_attr_ids:
                local.append(attrid)
            else:
                remote.append(attrid)


        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 Plug_CB_Metering_Threshold(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

    signature = {
        MODELS_INFO: [("_TZ3000_cayepv1a", "TS011F")],
        ENDPOINTS: {
            # <SimpleDescriptor endpoint=1 profile=260 device_type=266
            # device_version=1
            # input_clusters=[0, 3, 4, 5, 6, 1026, 1794, 2820, 57344, 57345]
            # output_clusters=[10, 25]>
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    OnOff.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    Metering.cluster_id,
                    ElectricalMeasurement.cluster_id,
                    TuyaZBE000Cluster.cluster_id,
                    TuyaZBExternalSwitchTypeCluster.cluster_id,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            # <SimpleDescriptor endpoint=242 profile=41440 device_type=97
            # device_version=1
            # input_clusters=[]
            # output_clusters=[33]>
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }
    replacement = {
        ENDPOINTS: {
            1: {
                PROFILE_ID: zha.PROFILE_ID,
                DEVICE_TYPE: zha.DeviceType.ON_OFF_PLUG_IN_UNIT,
                INPUT_CLUSTERS: [
                    Basic.cluster_id,
                    Identify.cluster_id,
                    Groups.cluster_id,
                    Scenes.cluster_id,
                    TemperatureMeasurement.cluster_id,
                    TuyaZBOnOffAttributeCluster,
                    TuyaZBMeteringCluster,
                    TuyaZBElectricalMeasurement,
                    TuyaZBE000Cluster,
                    TuyaZBExternalSwitchTypeThresholdCluster,
                ],
                OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
            },
            242: {
                PROFILE_ID: zgp.PROFILE_ID,
                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
                INPUT_CLUSTERS: [],
                OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
            },
        },
    }

Thanks, I use ZHA with TO-Q-SY2-JZT, I have the same issue I can't set the intervention thresholds.
I saw and installed your work, but reading the value of attributes I always have None, to change the values ​​I have to use the attribute writing, or the commands?
The values ​​to pass are the following?

For example
over_current_breaker (id: 0xe701) = 0 disabled
over_current_breaker (id: 0xe701) = 1 Enabled

over_current_threshold (id: 0xe781) is value expressed in amps?

Thanks

[github-actions]

There hasn't been any activity on this issue recently. Due to the high number of incoming GitHub notifications, we have to clean some of the old issues, as many of them have already been resolved with the latest updates. Please make sure to update to the latest version and check if that solves the issue. Let us know if that works for you by adding a comment 👍 This issue has now been marked as stale and will be closed if no further activity occurs. Thank you for your contributions.

[franortiz]

Updated gist to support 2025.08 command direction warnings

[puddly]

@franortiz could you open a PR?

[franortiz]

@franortiz could you open a PR?

PR #4267 submitted
Need to fix pre-commit issues