Skip to content

Instantly share code, notes, and snippets.

@epenet

epenet/ts0601_TZE284_81yrt3lo.py

Last active November 28, 2025 13:21
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save epenet/39050e3f47f0a8ac4a3aaf526173d5d7 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save epenet/39050e3f47f0a8ac4a3aaf526173d5d7 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
ts0601_TZE284_81yrt3lo.py
"""Tuya Energy Meter."""
from __future__ import annotations
from typing import Any
from zigpy.quirks.v2.homeassistant import PERCENTAGE, EntityType, UnitOfTime
import zigpy.types as t
from zhaquirks.tuya import (
TuyaLocalCluster,
TuyaZBElectricalMeasurement,
TuyaZBMeteringClusterWithUnit,
)
from zhaquirks.tuya.builder import TuyaQuirkBuilder
from zhaquirks.tuya.mcu import TuyaMCUCluster
class ChannelEndpoint:
"""Constants for meter channel endpoint_id."""
A = 2
B = 3
@classmethod
def attr_suffix(cls, endpoint: int) -> str:
"""Return mapping."""
if endpoint == cls.A:
return "_a"
if endpoint == cls.B:
return "_b"
return None
class TuyaEnergyDirection(t.enum8):
"""Energy direction attribute type."""
Forward = 0x0
Reverse = 0x1
class EnergyDirectionHelper:
"""Apply Tuya EnergyDirection to ZCL power attributes."""
_DIRECTION_HANDLER_ATTRIBUTES: tuple[str] = ()
UNSIGNED_ATTR_SUFFIX = "_attr_unsigned"
def __init__(self, *args, **kwargs):
"""Init."""
self._energy_direction: TuyaEnergyDirection | None = None
super().__init__(*args, **kwargs)
def align_with_energy_direction(self, value: int | None) -> int | None:
"""Align the value with current energy_direction."""
if value and (
(self.energy_direction == TuyaEnergyDirection.Reverse and value > 0)
or (self.energy_direction == TuyaEnergyDirection.Forward and value < 0)
):
value = -value
return value
@property
def _mcu_cluster(self) -> TuyaMCUCluster | None:
"""Return the MCU cluster."""
return getattr(
self.endpoint.device.endpoints[1], TuyaMCUCluster.ep_attribute, None
)
@property
def energy_direction(self) -> TuyaEnergyDirection | None:
"""Return the channel energy direction."""
if not self._mcu_cluster:
return self._energy_direction
try:
return self._mcu_cluster.get(
"energy_direction"
+ ChannelEndpoint.attr_suffix(self.endpoint.endpoint_id)
)
except KeyError:
return self._energy_direction
def energy_direction_handler(self, attr_name: str, value) -> tuple[str, Any]:
"""Unsigned device values are aligned with the energy direction."""
if attr_name not in self._DIRECTION_HANDLER_ATTRIBUTES:
return attr_name, value
if attr_name.endswith(self.UNSIGNED_ATTR_SUFFIX):
attr_name = attr_name.removesuffix(self.UNSIGNED_ATTR_SUFFIX)
value = self.align_with_energy_direction(value)
if value is not None:
# _energy_direction used for virtual calculations on devices with native signed values
self._energy_direction = (
TuyaEnergyDirection.Reverse
if value < 0
else TuyaEnergyDirection.Forward
)
return attr_name, value
class TuyaElectricalMeasurement(
TuyaLocalCluster,
TuyaZBElectricalMeasurement,
):
"""ElectricalMeasurement cluster for Tuya energy meter devices.
Attribute units prior to cluster formatting:
Current: A * 1000 (milliampres)
Frequency: Hz * 100
Power: W * 10 (deciwatt)
Voltage: V * 10 (decivolt)
"""
AMPERE_MULTIPLIER = 1000
HZ_MULTIPLIER = 100
VOLT_MULTIPLIER = 10
WATT_MULTIPLIER = 10
_CONSTANT_ATTRIBUTES: dict[int, Any] = {
**TuyaZBElectricalMeasurement._CONSTANT_ATTRIBUTES, # imports current divisor of 1000
TuyaZBElectricalMeasurement.AttributeDefs.ac_frequency_divisor.id: 100, # 2 decimals
TuyaZBElectricalMeasurement.AttributeDefs.ac_frequency_multiplier.id: 1,
TuyaZBElectricalMeasurement.AttributeDefs.ac_power_divisor.id: 10, # 1 decimal
TuyaZBElectricalMeasurement.AttributeDefs.ac_power_multiplier.id: 1,
TuyaZBElectricalMeasurement.AttributeDefs.ac_voltage_divisor.id: 10, # 1 decimal
TuyaZBElectricalMeasurement.AttributeDefs.ac_voltage_multiplier.id: 1,
}
def __init__(self, *args, **kwargs):
"""Init a TuyaMeteringMain cluster."""
super().__init__(*args, **kwargs)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.apparent_power.id,
)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.active_power.id
)
class TuyaElectricalMeasurementMain(TuyaElectricalMeasurement):
"""Filter supported attributes for global channels."""
def __init__(self, *args, **kwargs):
"""Init a TuyaElectricalMeasurementMain cluster."""
super().__init__(*args, **kwargs)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.rms_current.id
)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.power_factor.id
)
class TuyaElectricalMeasurementChannel(TuyaElectricalMeasurement):
"""Filter supported attributes for individual channels."""
def __init__(self, *args, **kwargs):
"""Init a TuyaElectricalMeasurementChannel cluster."""
super().__init__(*args, **kwargs)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.rms_voltage.id
)
self.add_unsupported_attribute(
TuyaElectricalMeasurement.AttributeDefs.ac_frequency.id
)
class TuyaMetering(
EnergyDirectionHelper,
TuyaLocalCluster,
TuyaZBMeteringClusterWithUnit,
):
"""Metering cluster for Tuya energy meter devices."""
WATT_MULTIPLIER = 10
WATT_HOUR_MULTIPLIER = 1000
DECIWATT_HOUR_MULTIPLIER = 100
@staticmethod
def format(
whole_digits: int, dec_digits: int, suppress_leading_zeros: bool = True
) -> int:
"""Return the formatter value for summation and demand Metering attributes."""
assert 0 <= whole_digits <= 7, "must be within range of 0 to 7."
assert 0 <= dec_digits <= 7, "must be within range of 0 to 7."
return (suppress_leading_zeros << 6) | (whole_digits << 3) | dec_digits
_CONSTANT_ATTRIBUTES: dict[int, Any] = {
**TuyaZBMeteringClusterWithUnit._CONSTANT_ATTRIBUTES,
TuyaZBMeteringClusterWithUnit.AttributeDefs.status.id: 0x00,
TuyaZBMeteringClusterWithUnit.AttributeDefs.multiplier.id: 1,
TuyaZBMeteringClusterWithUnit.AttributeDefs.divisor.id: 10000, # 2 decimals for summation attributes
TuyaZBMeteringClusterWithUnit.AttributeDefs.summation_formatting.id: format(
whole_digits=7, dec_digits=2
),
TuyaZBMeteringClusterWithUnit.AttributeDefs.demand_formatting.id: format(
whole_digits=7, dec_digits=1
),
}
_DIRECTION_HANDLER_ATTRIBUTES: tuple[str] = (
TuyaZBMeteringClusterWithUnit.AttributeDefs.instantaneous_demand.name
+ EnergyDirectionHelper.UNSIGNED_ATTR_SUFFIX,
)
def update_attribute(self, attr_name: str, value):
"""Update the cluster attribute."""
attr_name, value = self.energy_direction_handler(attr_name, value)
super().update_attribute(attr_name, value)
(
TuyaQuirkBuilder("_TZE284_81yrt3lo", "TS0601")
.tuya_enchantment()
.adds_endpoint(ChannelEndpoint.A)
.adds_endpoint(ChannelEndpoint.B)
.adds(TuyaElectricalMeasurementMain)
.adds(TuyaElectricalMeasurementChannel, endpoint_id=ChannelEndpoint.A)
.adds(TuyaElectricalMeasurementChannel, endpoint_id=ChannelEndpoint.B)
.adds(TuyaMetering)
.adds(TuyaMetering, endpoint_id=ChannelEndpoint.A)
.adds(TuyaMetering, endpoint_id=ChannelEndpoint.B)
.tuya_dp(
dp_id=1, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_delivered.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
)
.tuya_dp(
dp_id=2, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_received.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
)
.tuya_dp(
dp_id=115, # W * 10 (deciwatt)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.instantaneous_demand.name,
)
.tuya_dp(
dp_id=111, # Hz * 100
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.ac_frequency.name,
)
.tuya_dp(
dp_id=106, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_delivered.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
endpoint_id=ChannelEndpoint.A,
)
.tuya_dp(
dp_id=108, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_delivered.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
endpoint_id=ChannelEndpoint.B,
)
.tuya_dp(
dp_id=107, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_received.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
endpoint_id=ChannelEndpoint.A,
)
.tuya_dp(
dp_id=109, # Wh * 10 (deciwatt/hour)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.current_summ_received.name,
converter=lambda x: x * TuyaMetering.DECIWATT_HOUR_MULTIPLIER,
endpoint_id=ChannelEndpoint.B,
)
.tuya_dp(
dp_id=101, # W * 10 (deciwatt)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.instantaneous_demand.name
+ EnergyDirectionHelper.UNSIGNED_ATTR_SUFFIX,
endpoint_id=ChannelEndpoint.A,
)
.tuya_dp(
dp_id=105, # W * 10 (deciwatt)
ep_attribute=TuyaMetering.ep_attribute,
attribute_name=TuyaMetering.AttributeDefs.instantaneous_demand.name
+ EnergyDirectionHelper.UNSIGNED_ATTR_SUFFIX,
endpoint_id=ChannelEndpoint.B,
)
.tuya_dp(
dp_id=110, # % (power factor)
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.power_factor.name,
endpoint_id=ChannelEndpoint.A,
)
.tuya_dp(
dp_id=121, # % (power factor)
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.power_factor.name,
endpoint_id=ChannelEndpoint.B,
)
.tuya_dp(
dp_id=113, # A * 1000 (milliampre)
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.rms_current.name,
endpoint_id=ChannelEndpoint.A,
)
.tuya_dp(
dp_id=114, # A * 1000 (milliampre)
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.rms_current.name,
endpoint_id=ChannelEndpoint.B,
)
.tuya_dp(
dp_id=112, # V * 10 (decivolt)
ep_attribute=TuyaElectricalMeasurement.ep_attribute,
attribute_name=TuyaElectricalMeasurement.AttributeDefs.rms_voltage.name,
)
.tuya_number(
dp_id=129, # seconds
attribute_name="reporting_interval",
type=t.uint32_t_be,
unit=UnitOfTime.SECONDS,
min_value=5,
max_value=60,
step=1,
translation_key="reporting_interval",
fallback_name="Reporting interval",
entity_type=EntityType.CONFIG,
)
.tuya_number(
dp_id=122, # % * 10 (1 decimal precision)
attribute_name="ac_frequency_coefficient",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_ac_frequency",
fallback_name="Calibrate AC frequency",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=119, # % * 10 (1 decimal precision)
attribute_name="current_summ_delivered_coefficient_a",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_summ_delivered_a",
fallback_name="Calibrate summation delivered A",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=125, # % * 10 (1 decimal precision)
attribute_name="current_summ_delivered_coefficient_b",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_summ_delivered_b",
fallback_name="Calibrate summation delivered B",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=127, # % * 10 (1 decimal precision)
attribute_name="current_summ_received_coefficient_a",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_summ_received_a",
fallback_name="Calibrate summation received A",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=128, # % * 10 (1 decimal precision)
attribute_name="current_summ_received_coefficient_b",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_summ_received_b",
fallback_name="Calibrate summation received B",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=118, # % * 10 (1 decimal precision)
attribute_name="instantaneous_demand_coefficient_a",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_instantaneous_demand_a",
fallback_name="Calibrate instantaneous demand A",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=124, # % * 10 (1 decimal precision)
attribute_name="instantaneous_demand_coefficient_b",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_instantaneous_demand_b",
fallback_name="Calibrate instantaneous demand B",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=117, # % * 10 (1 decimal precision)
attribute_name="rms_current_coefficient_a",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_current_a",
fallback_name="Calibrate current A",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=123, # % * 10 (1 decimal precision)
attribute_name="rms_current_coefficient_b",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_current_b",
fallback_name="Calibrate current B",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_number(
dp_id=116, # % * 10 (1 decimal precision)
attribute_name="rms_voltage_coefficient",
type=t.uint32_t_be,
unit=PERCENTAGE,
min_value=0,
max_value=2000,
step=0.1,
multiplier=0.1,
translation_key="calibrate_voltage",
fallback_name="Calibrate voltage",
entity_type=EntityType.CONFIG,
initially_disabled=True,
)
.tuya_dp_attribute(
dp_id=102, # 0=Forward/1=Reverse
attribute_name="energy_direction_a",
type=TuyaEnergyDirection,
converter=lambda x: TuyaEnergyDirection(x),
)
.tuya_dp_attribute(
dp_id=104, # 0=Forward/1=Reverse
attribute_name="energy_direction_b",
type=TuyaEnergyDirection,
converter=lambda x: TuyaEnergyDirection(x),
)
.add_to_registry()
)
@Montspy
Copy link

Montspy commented Nov 28, 2025 •
edited
Loading

Works great, thanks!
I only had to remove the unisgned suffix to get the instantaneous demand showing up. (Remove + EnergyDirectionHelper.UNSIGNED_ATTR_SUFFIX)
https://gist.github.com/epenet/39050e3f47f0a8ac4a3aaf526173d5d7#file-ts0601_tze284_81yrt3lo-py-L272-L285

@epenet
Copy link
Author

epenet commented Nov 28, 2025 •
edited
Loading

I'm glad it's working for you.
I need to get back to this now that translation_placeholders have been rolled out!
We should be able to do something much cleaner now.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment