-
Notifications
You must be signed in to change notification settings - Fork 681
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Device Support Request] TS0601 by _TZE204_wbhaespm 3 phase power meter din #3282
Comments
I've made some progress. I tinkered a bit and checked how the implementations of other three-phase meters look. I think I'm now mapping the entire data from the three phases + summation delivered. Unfortunately, the sensors in the ZHA device panel only show data from the first phase (even though the data is transmitted and read by ZHA for all three phases; rms_voltage - as a sensor, only shows phase 1, rms_voltage_ph_b - hidden but reads data, similarly with current and power for the other phases). Additionally, the sum of W for the respective phases goes to total_active_power (it's calculated correctly but doesn't display as a sensor). There are still some other mysteries to solve like switch control or these attributes: [zhaquirks.tuya] [0xf438:1:0xef00] Received value [0, 0, 0, 0] for attribute 0x020d (command 0x0002)
[zhaquirks.tuya] [0xf438:1:0xef00] Received value [0] for attribute 0x0509 (command 0x0002)
[zhaquirks.tuya] [0xf438:1:0xef00] Received value [0, 0, 2, 126] for attribute 0x0201 (command 0x0002) #summation deliverd
[zhaquirks.tuya] [0xf438:1:0xef00] Received value [48, 48, 48, 48, 48, 48, 48, 48, 48, 48] for attribute 0x0313 (command 0x0002) This is far from an ideal solution but it allows for further development. I would appreciate any guidance. Especially how to make sensors out of recivied data. Custom quirk (very messy!)"""Tuya Din Power Meter."""
from zigpy.profiles import zha
import zigpy.types as t
import logging
from zigpy.zcl.clusters.general import Basic, Groups, Ota, Scenes, Time, GreenPowerProxy
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.smartenergy import Metering
from zhaquirks import Bus, LocalDataCluster
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.tuya import TuyaManufClusterAttributes, TuyaOnOff
from zigpy.quirks import CustomDevice
_LOGGER = logging.getLogger(__name__)
# Tuya attribute IDs
TUYA_TOTAL_ENERGY_ATTR = 0x0201
#TUYA_CURRENT_ATTR = 0x0212
#TUYA_POWER_ATTR = 0x0267
#TUYA_VOLTAGE_ATTR = 0x0214
RTX_TOTAL_ACTIVE_POWER_ATTR = 0x0304
RTX_TOTAL_CURRENT = 0x0212
RTX_AVG_VOLTAGE = 0x0214
RTX_VCP_P1 = 0x0006
RTX_VCP_P2 = 0x0007
RTX_VCP_P3 = 0x0008
#SWITCH_EVENT = "switch_event"
class TuyaManufClusterDinPower(TuyaManufClusterAttributes):
"""Manufacturer Specific Cluster of the Tuya Power Meter device."""
attributes = {
TUYA_TOTAL_ENERGY_ATTR: ("energy", t.uint32_t, True),
RTX_TOTAL_CURRENT: ("current", t.uint32_t, True),
RTX_AVG_VOLTAGE: ("voltage", t.uint32_t, True),
RTX_TOTAL_ACTIVE_POWER_ATTR: ("power", t.uint32_t, True),
RTX_VCP_P1: ("vcp_ph_1", t.data64, True),
RTX_VCP_P2: ("vcp_ph_2", t.data64, True),
RTX_VCP_P3: ("vcp_ph_3", t.data64, True),
}
def _update_attribute(self, attrid, value):
super()._update_attribute(attrid, value)
if attrid == TUYA_TOTAL_ENERGY_ATTR:
self.endpoint.smartenergy_metering.energy_reported(value / 100)
elif attrid in [RTX_VCP_P1, RTX_VCP_P2, RTX_VCP_P3]:
_LOGGER.debug(f"Received phase data: attrid={attrid}, value={value}")
# Handling voltage and current for each phase
power = int.from_bytes(value[0:3], byteorder="little")
current = int.from_bytes(value[3:6], byteorder="little")
voltage = int.from_bytes(value[6:8], byteorder="little")
phase_index = attrid - RTX_VCP_P1 # Determine phase index (0, 1, or 2)
# Update individual phase values
self.endpoint.electrical_measurement.vcp_reported(voltage, current, power, phase_index)
#elif attrid == TUYA_DIN_SWITCH_ATTR:
# self.endpoint.device.switch_bus.listener_event(
# SWITCH_EVENT, self.endpoint.endpoint_id, value
# )
class TuyaPowerMeasurement(LocalDataCluster, ElectricalMeasurement):
"""Custom class for power, voltage, and current measurement."""
cluster_id = ElectricalMeasurement.cluster_id
AC_VOLTAGE_MULTIPLIER = 0x0600
AC_VOLTAGE_DIVISOR = 0x0601
AC_CURRENT_MULTIPLIER = 0x0602
AC_CURRENT_DIVISOR = 0x0603
AC_FREQ_MULTIPLIER = 0x0400
AC_FREQ_DIVISOR = 0x0401
_CONSTANT_ATTRIBUTES = {
AC_VOLTAGE_MULTIPLIER: 1,
AC_VOLTAGE_DIVISOR: 10,
AC_CURRENT_MULTIPLIER: 1,
AC_CURRENT_DIVISOR: 1000,
AC_FREQ_MULTIPLIER: 1,
AC_FREQ_DIVISOR: 100,
}
phase_attributes = [
{
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current.id,
"power": ElectricalMeasurement.AttributeDefs.active_power.id,
},
{
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage_ph_b.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current_ph_b.id,
"power": ElectricalMeasurement.AttributeDefs.active_power_ph_b.id,
},
{
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage_ph_c.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current_ph_c.id,
"power": ElectricalMeasurement.AttributeDefs.active_power_ph_c.id,
},
]
def vcp_reported(self, voltage, current, power, phase=0):
"""Voltage, current, power reported."""
try:
if phase < 0 or phase > 2:
_LOGGER.error(f"Invalid phase index: {phase}")
return
_LOGGER.debug(f"Received data for phase {phase+1}: Voltage={voltage}, Current={current}, Power={power}")
# Update the corresponding attributes
self._update_attribute(self.phase_attributes[phase]["voltage"], voltage)
self._update_attribute(self.phase_attributes[phase]["current"], current)
self._update_attribute(self.phase_attributes[phase]["power"], power)
# Calculate the total active power across all phases
self.update_total_active_power()
except Exception as e:
_LOGGER.error(f"Error updating attributes for phase {phase+1}: {str(e)}")
def update_total_active_power(self):
"""Calculate and update the total active power across all phases."""
try:
total_power = 0
for phase in range(3):
power_attr = self.phase_attributes[phase]["power"]
phase_power = self.get(power_attr, 0)
total_power += phase_power
_LOGGER.debug(f"Total active power across all phases: {total_power}W")
# Update the total power attribute
self._update_attribute(RTX_TOTAL_ACTIVE_POWER_ATTR, total_power)
except Exception as e:
_LOGGER.error(f"Error calculating total active power: {str(e)}")
class TuyaElectricalMeasurement(LocalDataCluster, Metering):
"""Custom class for total energy measurement."""
cluster_id = Metering.cluster_id
CURRENT_ID = 0x0000
POWER_WATT = 0x0000
"""Setting unit of measurement."""
_CONSTANT_ATTRIBUTES = {0x0300: POWER_WATT}
def energy_reported(self, value):
"""Summation Energy reported."""
self._update_attribute(self.CURRENT_ID, value)
class RTXC63PowerMeter3Phase(CustomDevice):
"""RTXC63 Power Meter Device"""
def __init__(self, *args, **kwargs):
"""Init device."""
self.switch_bus = Bus()
super().__init__(*args, **kwargs)
signature = {
MODELS_INFO: [
("_TZE204_wbhaespm", "TS0601"),
],
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterAttributes.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242: {
PROFILE_ID: 0xA1E0,
DEVICE_TYPE: 0x0061,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterDinPower,
TuyaPowerMeasurement,
TuyaElectricalMeasurement,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242: {
PROFILE_ID: 0xA1E0,
DEVICE_TYPE: 0x0061,
INPUT_CLUSTERS: [
TuyaOnOff,
],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
}
} |
I was able to build the basic functionality I was aiming for, but I am pausing work on this quirk. However, there are still many features and data from the device that remain unhandled, and the ones I managed to implement could be done better. In this quirk, I've combined power, current, and voltage readings across all three phases to provide a single entity for each metric. The power values from all phases are summed and reflected in the main The original data for phase A has been overwritten with these combined values, so entities now represent the overall 3-phase performance rather than just phase A. This gives more accurate and holistic readings for power, current, and voltage while keeping the energy summation unchanged. Worth to mention separate VCP data for all three phases is accessible by 'manage device' option but not as entities. Quirk provides total power (sum), total current (sum), and average voltage, all while preserving the original energy summation behavior. This implementation improves monitoring of the 3-phase system by giving a complete view rather than per-phase / phase A data. Any further development (like on/off switch or mystery attributes mentioned in previous comment) will be very welcome. I.e I'm positive that over-current threshold can be set by configuration. Custom quirk (read comment above!)from zigpy.profiles import zha
import zigpy.types as t
from zigpy.quirks import CustomDevice
from zigpy.zcl.clusters.general import Basic, Groups, Ota, Scenes, Time, GreenPowerProxy
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
from zigpy.zcl.clusters.smartenergy import Metering
import logging
from zhaquirks import Bus, LocalDataCluster
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.tuya import TuyaManufClusterAttributes
_LOGGER = logging.getLogger(__name__)
TUYA_TOTAL_ENERGY_ATTR = 0x0201
RTX_VCP_P1 = 0x0006
RTX_VCP_P2 = 0x0007
RTX_VCP_P3 = 0x0008
class TuyaManufClusterDinPower(TuyaManufClusterAttributes):
attributes = {
TUYA_TOTAL_ENERGY_ATTR: ("energy", t.uint32_t, True),
RTX_VCP_P1: ("vcp_ph_1", t.data64, True),
RTX_VCP_P2: ("vcp_ph_2", t.data64, True),
RTX_VCP_P3: ("vcp_ph_3", t.data64, True),
}
def _update_attribute(self, attrid, value):
super()._update_attribute(attrid, value)
if attrid == TUYA_TOTAL_ENERGY_ATTR:
self.endpoint.smartenergy_metering.energy_reported(value / 100)
elif attrid in [RTX_VCP_P1, RTX_VCP_P2, RTX_VCP_P3]:
_LOGGER.debug(f"Received phase data: attrid={attrid}, value={value}")
power = int.from_bytes(value[0:3], byteorder="little")
current = int.from_bytes(value[3:6], byteorder="little")
voltage = int.from_bytes(value[6:8], byteorder="little")
phase_index = attrid - RTX_VCP_P1
self.endpoint.electrical_measurement.vcp_reported(voltage, current, power, phase_index)
class TuyaPowerMeasurement(LocalDataCluster, ElectricalMeasurement):
cluster_id = ElectricalMeasurement.cluster_id
phase_attributes = [
{ # Phase 1
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current.id,
"power": ElectricalMeasurement.AttributeDefs.active_power.id,
},
{ # Phase 2
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage_ph_b.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current_ph_b.id,
"power": ElectricalMeasurement.AttributeDefs.active_power_ph_b.id,
},
{ # Phase 3
"voltage": ElectricalMeasurement.AttributeDefs.rms_voltage_ph_c.id,
"current": ElectricalMeasurement.AttributeDefs.rms_current_ph_c.id,
"power": ElectricalMeasurement.AttributeDefs.active_power_ph_c.id,
},
]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.phase_power = [0, 0, 0]
self.phase_current = [0, 0, 0]
self.phase_voltage = [0, 0, 0]
def vcp_reported(self, voltage, current, power, phase=0):
try:
if phase < 0 or phase > 2:
_LOGGER.error(f"Invalid phase index: {phase}")
return
self._update_attribute(self.phase_attributes[phase]["voltage"], voltage)
self._update_attribute(self.phase_attributes[phase]["current"], current)
self._update_attribute(self.phase_attributes[phase]["power"], power)
self.phase_power[phase] = power
self.phase_current[phase] = current
self.phase_voltage[phase] = voltage
total_power = sum(self.phase_power)
_LOGGER.debug(f"Total active power across all phases: {total_power}W")
self._update_attribute(ElectricalMeasurement.AttributeDefs.active_power.id, total_power)
total_current = sum(self.phase_current)
_LOGGER.debug(f"Total current across all phases: {total_current}mA")
self._update_attribute(ElectricalMeasurement.AttributeDefs.rms_current.id, total_current)
average_voltage = sum(self.phase_voltage) / 3.0
_LOGGER.debug(f"Average voltage across all phases: {average_voltage}V")
self._update_attribute(ElectricalMeasurement.AttributeDefs.rms_voltage.id, round(average_voltage))
except Exception as e:
_LOGGER.error(f"Error updating attributes for phase {phase+1}: {str(e)}")
class TuyaElectricalMeasurement(LocalDataCluster, Metering):
cluster_id = Metering.cluster_id
def energy_reported(self, value):
self._update_attribute(Metering.AttributeDefs.current_summ_delivered.id, value)
class TuyaPowerMeter3Phase(CustomDevice):
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=4417, maximum_buffer_size=66, maximum_incoming_transfer_size=66, server_mask=10752, maximum_outgoing_transfer_size=66,
#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": "0x0051",
# "input_clusters": [
# "0x0000",
# "0x0004",
# "0x0005",
# "0x0702",
# "0x0b04",
# "0xef00"
# ],
# "output_clusters": [
# "0x000a",
# "0x0019"
# ]
# },
# "242": {
# "profile_id": "0xa1e0",
# "device_type": "0x0061",
# "input_clusters": [
# "0x0006"
# ],
# "output_clusters": [
# "0x0021"
# ]
# }
MODELS_INFO: [("_TZE204_wbhaespm", "TS0601")],
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterAttributes.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242: {
PROFILE_ID: 0xA1E0,
DEVICE_TYPE: 0x0061,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterDinPower,
TuyaPowerMeasurement,
TuyaElectricalMeasurement,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242: {
PROFILE_ID: 0xA1E0,
DEVICE_TYPE: 0x0061,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
}
} |
Thank you for this, highly appreciated. |
Problem description
https://allegro.pl/oferta/bezpiecznik-3-fazy-regulowany-1-do-63a-zigbee-tuya-licznik-miernik-energii-16097353966
The device is a 3-phase circuit breaker (63A) with power metering, mounted on a DIN rail. It is presented as a Tuya device under the RTX brand, model C63. The device is recognized by ZHA, but no entities are registered. I was able to bind the power summation delivery entity with a custom quirk and identify attributes for each of the three phases (including amps, volts, and watts). However, I am currently stuck with correctly binding that data to ZHA.
Solution description
Bind data for all of three phases to ZHA.
Screenshots/Video
Screenshots/Video
Device signature
Device signature
Diagnostic information
Diagnostic information
Logs
Logs
I'm positive that three last attributes (0x0006 to 0x0008) are 3 phases. The data in each array can be decoded to sane values.returns (V, mA, W) ((241.3, 743, 147), (238.60000000000002, 634, 125), (234.3, 397, 82))
Debug logs shows that data is incoming regularly:
Custom quirk
Custom quirk
In this quirk, I attempted to parse and bind data for all three phases, but the relevant attributes in ZHA (e.g., rms_power, rms_current, rms_current_ph_b, etc.) always show a value of None. Please excuse the overall structure of the quirk; this is my first attempt at creating a custom quirk.Additional information
All suggestions on how to bind the data in ZHA would be greatly appreciated.
The text was updated successfully, but these errors were encountered: