zha-device-handlers
zha-device-handlers copied to clipboard
Published
20 hours ago •
zigpy
zigpy
[Device Support Request] TS0601 _TZE284_81yrt3lo Tuya Smart Life Dual channel energy meter
Problem description
Would like help in adding support for this small 2 channel AC power meter. Purchased from Aliexpress.
Solution description
Able to see the power readings for the two channels
Screenshots/Video
Screenshots/Video
[Paste/upload your media here]
Device signature
Device signature
{
"node_descriptor": {
"logical_type": 1,
"complex_descriptor_available": 0,
"user_descriptor_available": 0,
"reserved": 0,
"aps_flags": 0,
"frequency_band": 8,
"mac_capability_flags": 142,
"manufacturer_code": 4417,
"maximum_buffer_size": 66,
"maximum_incoming_transfer_size": 66,
"server_mask": 10752,
"maximum_outgoing_transfer_size": 66,
"descriptor_capability_field": 0
},
"endpoints": {
"1": {
"profile_id": "0x0104",
"device_type": "0x0051",
"input_clusters": [
"0x0000",
"0x0004",
"0x0005",
"0xed00",
"0xef00"
],
"output_clusters": [
"0x000a",
"0x0019"
]
},
"242": {
"profile_id": "0xa1e0",
"device_type": "0x0061",
"input_clusters": [],
"output_clusters": [
"0x0021"
]
}
},
"manufacturer": "_TZE284_81yrt3lo",
"model": "TS0601",
"class": "zigpy.device.Device"
}
Diagnostic information
Diagnostic information
{
"home_assistant": {
"installation_type": "Home Assistant OS",
"version": "2024.12.5",
"dev": false,
"hassio": true,
"virtualenv": false,
"python_version": "3.13.0",
"docker": true,
"arch": "aarch64",
"timezone": "Australia/Sydney",
"os_name": "Linux",
"os_version": "6.6.62-haos-raspi",
"supervisor": "2024.12.0",
"host_os": "Home Assistant OS 14.1",
"docker_version": "27.2.0",
"chassis": "embedded",
"run_as_root": true
},
"custom_components": {
"hacs": {
"documentation": "https://hacs.xyz/docs/configuration/start",
"version": "2.0.1",
"requirements": [
"aiogithubapi>=22.10.1"
]
},
"waste_collection_schedule": {
"documentation": "https://github.com/mampfes/hacs_waste_collection_schedule#readme",
"version": "2.5.0",
"requirements": [
"icalendar",
"recurring_ical_events",
"icalevents>=0.1.26,!=0.1.28",
"beautifulsoup4",
"lxml",
"pycryptodome"
]
}
},
"integration_manifest": {
"domain": "zha",
"name": "Zigbee Home Automation",
"after_dependencies": [
"hassio",
"onboarding",
"usb"
],
"codeowners": [
"dmulcahey",
"adminiuga",
"puddly",
"TheJulianJES"
],
"config_flow": true,
"dependencies": [
"file_upload"
],
"documentation": "https://www.home-assistant.io/integrations/zha",
"iot_class": "local_polling",
"loggers": [
"aiosqlite",
"bellows",
"crccheck",
"pure_pcapy3",
"zhaquirks",
"zigpy",
"zigpy_deconz",
"zigpy_xbee",
"zigpy_zigate",
"zigpy_znp",
"zha",
"universal_silabs_flasher"
],
"requirements": [
"universal-silabs-flasher==0.0.25",
"zha==0.0.42"
],
"usb": [
{
"vid": "10C4",
"pid": "EA60",
"description": "*2652*",
"known_devices": [
"slae.sh cc2652rb stick"
]
},
{
"vid": "10C4",
"pid": "EA60",
"description": "*slzb-07*",
"known_devices": [
"smlight slzb-07"
]
},
{
"vid": "1A86",
"pid": "55D4",
"description": "*sonoff*plus*",
"known_devices": [
"sonoff zigbee dongle plus v2"
]
},
{
"vid": "10C4",
"pid": "EA60",
"description": "*sonoff*plus*",
"known_devices": [
"sonoff zigbee dongle plus"
]
},
{
"vid": "10C4",
"pid": "EA60",
"description": "*tubeszb*",
"known_devices": [
"TubesZB Coordinator"
]
},
{
"vid": "1A86",
"pid": "7523",
"description": "*tubeszb*",
"known_devices": [
"TubesZB Coordinator"
]
},
{
"vid": "1A86",
"pid": "7523",
"description": "*zigstar*",
"known_devices": [
"ZigStar Coordinators"
]
},
{
"vid": "1CF1",
"pid": "0030",
"description": "*conbee*",
"known_devices": [
"Conbee II"
]
},
{
"vid": "0403",
"pid": "6015",
"description": "*conbee*",
"known_devices": [
"Conbee III"
]
},
{
"vid": "10C4",
"pid": "8A2A",
"description": "*zigbee*",
"known_devices": [
"Nortek HUSBZB-1"
]
},
{
"vid": "0403",
"pid": "6015",
"description": "*zigate*",
"known_devices": [
"ZiGate+"
]
},
{
"vid": "10C4",
"pid": "EA60",
"description": "*zigate*",
"known_devices": [
"ZiGate"
]
},
{
"vid": "10C4",
"pid": "8B34",
"description": "*bv 2010/10*",
"known_devices": [
"Bitron Video AV2010/10"
]
}
],
"zeroconf": [
{
"type": "_esphomelib._tcp.local.",
"name": "tube*"
},
{
"type": "_zigate-zigbee-gateway._tcp.local.",
"name": "*zigate*"
},
{
"type": "_zigstar_gw._tcp.local.",
"name": "*zigstar*"
},
{
"type": "_uzg-01._tcp.local.",
"name": "uzg-01*"
},
{
"type": "_slzb-06._tcp.local.",
"name": "slzb-06*"
},
{
"type": "_xzg._tcp.local.",
"name": "xzg*"
},
{
"type": "_czc._tcp.local.",
"name": "czc*"
},
{
"type": "_zigbee-coordinator._tcp.local.",
"name": "*"
}
],
"is_built_in": true,
"overwrites_built_in": false
},
"setup_times": {
"null": {
"setup": 0.0001411839621141553
},
"1e3e986c701bf2ab7361cb8960ad6f16": {
"config_entry_setup": 23.60823148302734
}
},
"data": {
"ieee": "**REDACTED**",
"nwk": 12546,
"manufacturer": "_TZE284_81yrt3lo",
"model": "TS0601",
"name": "_TZE284_81yrt3lo TS0601",
"quirk_applied": false,
"quirk_class": "zigpy.device.Device",
"quirk_id": null,
"manufacturer_code": 4417,
"power_source": "Mains",
"lqi": 255,
"rssi": -80,
"last_seen": "2024-12-31T11:34:04",
"available": true,
"device_type": "Router",
"signature": {
"node_descriptor": {
"logical_type": 1,
"complex_descriptor_available": 0,
"user_descriptor_available": 0,
"reserved": 0,
"aps_flags": 0,
"frequency_band": 8,
"mac_capability_flags": 142,
"manufacturer_code": 4417,
"maximum_buffer_size": 66,
"maximum_incoming_transfer_size": 66,
"server_mask": 10752,
"maximum_outgoing_transfer_size": 66,
"descriptor_capability_field": 0
},
"endpoints": {
"1": {
"profile_id": "0x0104",
"device_type": "0x0051",
"input_clusters": [
"0x0000",
"0x0004",
"0x0005",
"0xed00",
"0xef00"
],
"output_clusters": [
"0x000a",
"0x0019"
]
},
"242": {
"profile_id": "0xa1e0",
"device_type": "0x0061",
"input_clusters": [],
"output_clusters": [
"0x0021"
]
}
},
"manufacturer": "_TZE284_81yrt3lo",
"model": "TS0601"
},
"active_coordinator": false,
"entities": [
{
"entity_id": "update.back_ac_firmware",
"name": "_TZE284_81yrt3lo TS0601"
}
],
"neighbors": [
{
"device_type": "Router",
"rx_on_when_idle": "On",
"relationship": "Parent",
"extended_pan_id": "**REDACTED**",
"ieee": "**REDACTED**",
"nwk": "0x399E",
"permit_joining": "Unknown",
"depth": "3",
"lqi": "45"
},
{
"device_type": "Router",
"rx_on_when_idle": "On",
"relationship": "Sibling",
"extended_pan_id": "**REDACTED**",
"ieee": "**REDACTED**",
"nwk": "0x8CF3",
"permit_joining": "Unknown",
"depth": "3",
"lqi": "30"
}
],
"routes": [],
"endpoint_names": [
{
"name": "SMART_PLUG"
},
{
"name": "PROXY_BASIC"
}
],
"user_given_name": "Back AC",
"device_reg_id": "7693de6f89a8265645ded656338c7ad0",
"area_id": "energy_generation",
"cluster_details": {
"1": {
"device_type": {
"name": "SMART_PLUG",
"id": 81
},
"profile_id": 260,
"in_clusters": {
"0x0004": {
"endpoint_attribute": "groups",
"attributes": {
"0xfffd": {
"attribute": "ZCLAttributeDef(id=0xFFFD, name='cluster_revision', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0000": {
"attribute": "ZCLAttributeDef(id=0x0000, name='name_support', type=<flag 'NameSupport'>, zcl_type=<DataTypeId.map8: 24>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0xfffe": {
"attribute": "ZCLAttributeDef(id=0xFFFE, name='reporting_status', type=<enum 'AttributeReportingStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
}
},
"unsupported_attributes": []
},
"0x0005": {
"endpoint_attribute": "scenes",
"attributes": {
"0xfffd": {
"attribute": "ZCLAttributeDef(id=0xFFFD, name='cluster_revision', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0000": {
"attribute": "ZCLAttributeDef(id=0x0000, name='count', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0002": {
"attribute": "ZCLAttributeDef(id=0x0002, name='current_group', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0001": {
"attribute": "ZCLAttributeDef(id=0x0001, name='current_scene', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0005": {
"attribute": "ZCLAttributeDef(id=0x0005, name='last_configured_by', type=<class 'zigpy.types.named.EUI64'>, zcl_type=<DataTypeId.EUI64: 240>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0004": {
"attribute": "ZCLAttributeDef(id=0x0004, name='name_support', type=<flag 'NameSupport'>, zcl_type=<DataTypeId.map8: 24>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0xfffe": {
"attribute": "ZCLAttributeDef(id=0xFFFE, name='reporting_status', type=<enum 'AttributeReportingStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0003": {
"attribute": "ZCLAttributeDef(id=0x0003, name='scene_valid', type=<enum 'Bool'>, zcl_type=<DataTypeId.bool_: 16>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
}
},
"unsupported_attributes": []
},
"0xef00": {
"endpoint_attribute": null,
"attributes": {},
"unsupported_attributes": []
},
"0x0000": {
"endpoint_attribute": "basic",
"attributes": {
"0x0013": {
"attribute": "ZCLAttributeDef(id=0x0013, name='alarm_mask', type=<flag 'AlarmMask'>, zcl_type=<DataTypeId.map8: 24>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0001": {
"attribute": "ZCLAttributeDef(id=0x0001, name='app_version', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": 78
},
"0xfffd": {
"attribute": "ZCLAttributeDef(id=0xFFFD, name='cluster_revision', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0006": {
"attribute": "ZCLAttributeDef(id=0x0006, name='date_code', type=<class 'zigpy.types.basic.LimitedCharString.<locals>.LimitedCharString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0012": {
"attribute": "ZCLAttributeDef(id=0x0012, name='device_enabled', type=<enum 'Bool'>, zcl_type=<DataTypeId.bool_: 16>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0014": {
"attribute": "ZCLAttributeDef(id=0x0014, name='disable_local_config', type=<flag 'DisableLocalConfig'>, zcl_type=<DataTypeId.map8: 24>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0008": {
"attribute": "ZCLAttributeDef(id=0x0008, name='generic_device_class', type=<enum 'GenericDeviceClass'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0009": {
"attribute": "ZCLAttributeDef(id=0x0009, name='generic_device_type', type=<enum 'GenericLightingDeviceType'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0003": {
"attribute": "ZCLAttributeDef(id=0x0003, name='hw_version', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0010": {
"attribute": "ZCLAttributeDef(id=0x0010, name='location_desc', type=<class 'zigpy.types.basic.LimitedCharString.<locals>.LimitedCharString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0004": {
"attribute": "ZCLAttributeDef(id=0x0004, name='manufacturer', type=<class 'zigpy.types.basic.LimitedCharString.<locals>.LimitedCharString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": "_TZE284_81yrt3lo"
},
"0x000c": {
"attribute": "ZCLAttributeDef(id=0x000C, name='manufacturer_version_details', type=<class 'zigpy.types.basic.CharacterString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0005": {
"attribute": "ZCLAttributeDef(id=0x0005, name='model', type=<class 'zigpy.types.basic.LimitedCharString.<locals>.LimitedCharString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": "TS0601"
},
"0x0011": {
"attribute": "ZCLAttributeDef(id=0x0011, name='physical_env', type=<enum 'PhysicalEnvironment'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0007": {
"attribute": "ZCLAttributeDef(id=0x0007, name='power_source', type=<enum 'PowerSource'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x000a": {
"attribute": "ZCLAttributeDef(id=0x000A, name='product_code', type=<class 'zigpy.types.basic.LVBytes'>, zcl_type=<DataTypeId.octstr: 65>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x000e": {
"attribute": "ZCLAttributeDef(id=0x000E, name='product_label', type=<class 'zigpy.types.basic.CharacterString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x000b": {
"attribute": "ZCLAttributeDef(id=0x000B, name='product_url', type=<class 'zigpy.types.basic.CharacterString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0xfffe": {
"attribute": "ZCLAttributeDef(id=0xFFFE, name='reporting_status', type=<enum 'AttributeReportingStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x000d": {
"attribute": "ZCLAttributeDef(id=0x000D, name='serial_number', type=<class 'zigpy.types.basic.CharacterString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0002": {
"attribute": "ZCLAttributeDef(id=0x0002, name='stack_version', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x4000": {
"attribute": "ZCLAttributeDef(id=0x4000, name='sw_build_id', type=<class 'zigpy.types.basic.CharacterString'>, zcl_type=<DataTypeId.string: 66>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0000": {
"attribute": "ZCLAttributeDef(id=0x0000, name='zcl_version', type=<class 'zigpy.types.basic.uint8_t'>, zcl_type=<DataTypeId.uint8: 32>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
}
},
"unsupported_attributes": []
},
"0xed00": {
"endpoint_attribute": null,
"attributes": {},
"unsupported_attributes": []
}
},
"out_clusters": {
"0x0019": {
"endpoint_attribute": "ota",
"attributes": {
"0xfffd": {
"attribute": "ZCLAttributeDef(id=0xFFFD, name='cluster_revision', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0002": {
"attribute": "ZCLAttributeDef(id=0x0002, name='current_file_version', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": 78
},
"0x0003": {
"attribute": "ZCLAttributeDef(id=0x0003, name='current_zigbee_stack_version', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0004": {
"attribute": "ZCLAttributeDef(id=0x0004, name='downloaded_file_version', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0005": {
"attribute": "ZCLAttributeDef(id=0x0005, name='downloaded_zigbee_stack_version', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0001": {
"attribute": "ZCLAttributeDef(id=0x0001, name='file_offset', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x000a": {
"attribute": "ZCLAttributeDef(id=0x000A, name='image_stamp', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0008": {
"attribute": "ZCLAttributeDef(id=0x0008, name='image_type_id', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0006": {
"attribute": "ZCLAttributeDef(id=0x0006, name='image_upgrade_status', type=<enum 'ImageUpgradeStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0007": {
"attribute": "ZCLAttributeDef(id=0x0007, name='manufacturer_id', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0009": {
"attribute": "ZCLAttributeDef(id=0x0009, name='minimum_block_req_delay', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0xfffe": {
"attribute": "ZCLAttributeDef(id=0xFFFE, name='reporting_status', type=<enum 'AttributeReportingStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x000b": {
"attribute": "ZCLAttributeDef(id=0x000B, name='upgrade_activation_policy', type=<enum 'UpgradeActivationPolicy'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0000": {
"attribute": "ZCLAttributeDef(id=0x0000, name='upgrade_server_id', type=<class 'zigpy.types.named.EUI64'>, zcl_type=<DataTypeId.EUI64: 240>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x000c": {
"attribute": "ZCLAttributeDef(id=0x000C, name='upgrade_timeout_policy', type=<enum 'UpgradeTimeoutPolicy'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
}
},
"unsupported_attributes": []
},
"0x000a": {
"endpoint_attribute": "time",
"attributes": {
"0xfffd": {
"attribute": "ZCLAttributeDef(id=0xFFFD, name='cluster_revision', type=<class 'zigpy.types.basic.uint16_t'>, zcl_type=<DataTypeId.uint16: 33>, access=<ZCLAttributeAccess.Read: 1>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0004": {
"attribute": "ZCLAttributeDef(id=0x0004, name='dst_end', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0005": {
"attribute": "ZCLAttributeDef(id=0x0005, name='dst_shift', type=<class 'zigpy.types.basic.int32s'>, zcl_type=<DataTypeId.int32: 43>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0003": {
"attribute": "ZCLAttributeDef(id=0x0003, name='dst_start', type=<class 'zigpy.types.basic.uint32_t'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0008": {
"attribute": "ZCLAttributeDef(id=0x0008, name='last_set_time', type=<class 'zigpy.types.named.UTCTime'>, zcl_type=<DataTypeId.UTC: 226>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0007": {
"attribute": "ZCLAttributeDef(id=0x0007, name='local_time', type=<class 'zigpy.types.named.LocalTime'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0xfffe": {
"attribute": "ZCLAttributeDef(id=0xFFFE, name='reporting_status', type=<enum 'AttributeReportingStatus'>, zcl_type=<DataTypeId.enum8: 48>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0006": {
"attribute": "ZCLAttributeDef(id=0x0006, name='standard_time', type=<class 'zigpy.types.named.StandardTime'>, zcl_type=<DataTypeId.uint32: 35>, access=<ZCLAttributeAccess.Read: 1>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0000": {
"attribute": "ZCLAttributeDef(id=0x0000, name='time', type=<class 'zigpy.types.named.UTCTime'>, zcl_type=<DataTypeId.UTC: 226>, access=<ZCLAttributeAccess.Read|Write_Optional: 5>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0001": {
"attribute": "ZCLAttributeDef(id=0x0001, name='time_status', type=<flag 'TimeStatus'>, zcl_type=<DataTypeId.map8: 24>, access=<ZCLAttributeAccess.Read|Write_Optional: 5>, mandatory=True, is_manufacturer_specific=False)",
"value": null
},
"0x0002": {
"attribute": "ZCLAttributeDef(id=0x0002, name='time_zone', type=<class 'zigpy.types.basic.int32s'>, zcl_type=<DataTypeId.int32: 43>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
},
"0x0009": {
"attribute": "ZCLAttributeDef(id=0x0009, name='valid_until_time', type=<class 'zigpy.types.named.UTCTime'>, zcl_type=<DataTypeId.UTC: 226>, access=<ZCLAttributeAccess.Read|Write: 3>, mandatory=False, is_manufacturer_specific=False)",
"value": null
}
},
"unsupported_attributes": []
}
}
},
"242": {
"device_type": {
"name": "PROXY_BASIC",
"id": 97
},
"profile_id": 41440,
"in_clusters": {},
"out_clusters": {
"0x0021": {
"endpoint_attribute": "green_power",
"attributes": {},
"unsupported_attributes": []
}
}
}
}
}
}
Logs
Logs
Custom quirk
Custom quirk
Additional information
Happy to test quirks, help debug etc. New to custom quirks for Zigbee, but not new to python & programming. Can't find a good guide for this so filing the request while I look more.
Hello, I’m experiencing the same issue with the recognition of this meter. I believe the problem lies with the incorrect identifier _TZE284_81yrt3lo. I have an identical meter recognized as _TZE204_81yrt3lo, and it is successfully integrated into Zigbee2MQTT, working without any issues. Could the solution be to adjust the external converter for the _TZE204_81yrt3lo meter to support this new identifier?
Hi, I have the same problem but it is when integrating into ZHA. ZHA discovers the device, configures it but it has no entities or sensors. It detects that it is _TZE284_81yrt3lo and does not apply the quirk. I have other devices with quirks applied but this one is not applied. I have adapted a quirk for the _TZE204_81yrt3lo by changing the model to the _TZE284_81yrt3lo but it has not worked. `"""Tuya Din Power Meter.""" from typing import Dict
from zigpy.profiles import zha from zigpy.quirks import CustomDevice import zigpy.types as t from zigpy.zcl import foundation from zigpy.zcl.clusters.general import Basic, Groups, Ota, Scenes, Time from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement from zigpy.zcl.clusters.smartenergy import Metering
from zhaquirks import LocalDataCluster from zhaquirks.const import ( DEVICE_TYPE, ENDPOINTS, INPUT_CLUSTERS, MODELS_INFO, OUTPUT_CLUSTERS, PROFILE_ID, ) from zhaquirks.tuya import TuyaLocalCluster from zhaquirks.tuya.mcu import ( DPToAttributeMapping, EnchantedDevice, TuyaMCUCluster, TuyaOnOff, )
class TuyaPowerMeasurement(TuyaLocalCluster, ElectricalMeasurement): """Custom class for power, voltage and current measurement."""
AC_CURRENT_MULTIPLIER = 0x0602
AC_CURRENT_DIVISOR = 0x0603
_CONSTANT_ATTRIBUTES = {AC_CURRENT_MULTIPLIER: 1, AC_CURRENT_DIVISOR: 1000}
class TuyaElectricalMeasurement(TuyaLocalCluster, Metering): """Custom class for total energy measurement."""
POWER_WATT = 0x0000
_CONSTANT_ATTRIBUTES = {
0x0300: POWER_WATT, # unit_of_measure
0x0302: 1000, # divisor
}
class DinPowerManufCluster(TuyaMCUCluster): """Tuya Manufacturer Cluster with din power datapoints."""
class TuyaConnectionStatus(t.Struct):
"""Tuya request data."""
tsn: t.uint8_t
status: t.LVBytes
client_commands = TuyaMCUCluster.client_commands.copy()
client_commands.update(
{
0x25: foundation.ZCLCommandDef(
"mcu_connection_status",
{"payload": TuyaConnectionStatus},
True,
is_manufacturer_specific=True,
),
}
)
server_commands = TuyaMCUCluster.server_commands.copy()
server_commands.update(
{
0x25: foundation.ZCLCommandDef(
"mcu_connection_status_rsp",
{"payload": TuyaConnectionStatus},
False,
is_manufacturer_specific=True,
),
}
)
def handle_mcu_connection_status(
self, payload: TuyaConnectionStatus
) -> foundation.Status:
"""Handle gateway connection status requests (0x25)."""
payload_rsp = DinPowerManufCluster.TuyaConnectionStatus()
payload_rsp.tsn = payload.tsn
payload_rsp.status = b"\x01" # 0x00 not connected to internet | 0x01 connected to internet | 0x02 time out
self.create_catching_task(
super().command(0x25, payload_rsp, expect_reply=False)
)
return foundation.Status.SUCCESS
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
0x01: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"current_summ_delivered",
),
0x06: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
("rms_current", "rms_voltage"),
converter=lambda x: (x >> 16, (x & 0x0000FFFF) / 10),
),
0x10: DPToAttributeMapping(
TuyaOnOff.ep_attribute,
"on_off",
),
0x66: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"current_summ_received",
),
0x67: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"active_power",
),
0x69: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"ac_frequency",
),
0x6D: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"total_reactive_power",
),
0x6E: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"reactive_power",
),
0x6F: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"power_factor",
),
}
data_point_handlers = {
0x01: "_dp_2_attr_update",
0x06: "_dp_2_attr_update",
0x10: "_dp_2_attr_update",
0x66: "_dp_2_attr_update",
0x67: "_dp_2_attr_update",
0x69: "_dp_2_attr_update",
0x6D: "_dp_2_attr_update",
0x6E: "_dp_2_attr_update",
0x6F: "_dp_2_attr_update",
}
class TuyaManufClusterDinPower(DinPowerManufCluster): """Manufacturer Specific Cluster of the Tuya Power Meter device."""
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
17: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"current_summ_delivered",
),
18: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"rms_current",
),
19: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"active_power",
converter=lambda x: x // 10,
),
20: DPToAttributeMapping(
TuyaPowerMeasurement.ep_attribute,
"rms_voltage",
converter=lambda x: x // 10,
),
}
data_point_handlers = {
17: "_dp_2_attr_update",
18: "_dp_2_attr_update",
19: "_dp_2_attr_update",
20: "_dp_2_attr_update",
}
class TuyaPowerMeter(EnchantedDevice): """Tuya power meter device."""
signature = {
# "node_descriptor": "<NodeDescriptor byte1=1 byte2=64 mac_capability_flags=142 manufacturer_code=4098
# maximum_buffer_size=82 maximum_incoming_transfer_size=82 server_mask=11264
# maximum_outgoing_transfer_size=82 descriptor_capability_field=0>",
# device_version=1
# input_clusters=[0x0000, 0x0004, 0x0005, 0xef00]
# output_clusters=[0x000a, 0x0019]
MODELS_INFO: [
("_TZE284_cjbofhxw", "TS0601"),
],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterDinPower.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.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],
}
}
}
class HikingPowerMeter(CustomDevice): """Hiking Power Meter Device - DDS238-2."""
signature = {
# "node_descriptor": "<NodeDescriptor byte1=1 byte2=64 mac_capability_flags=142 manufacturer_code=4098
# maximum_buffer_size=82 maximum_incoming_transfer_size=82 server_mask=11264
# maximum_outgoing_transfer_size=82 descriptor_capability_field=0>",
# device_version=1
# input_clusters=[0x0000, 0x0004, 0x0005, 0xef00]
# output_clusters=[0x000a, 0x0019]
MODELS_INFO: [
("_TZE200_bkkmqmyo", "TS0601"),
("ffffffffffffffff", "TS0601"),
],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
DinPowerManufCluster.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.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,
DinPowerManufCluster,
TuyaElectricalMeasurement,
TuyaPowerMeasurement,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
16: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
TuyaOnOff,
],
OUTPUT_CLUSTERS: [],
},
}
}`
Hi, I switched from ZHA to Z2M some time ago. Somehow, I find it more user-friendly. Regarding the converter for this meter in Z2M, I managed to put something together, and it now recognizes the device, and the measurements seem correct. If there are any possible improvements, I’d greatly appreciate them. As for the quirk in ZHA, I’m not sure. This is external converter that works for me.
const exposes = require('zigbee-herdsman-converters/lib/exposes');
const tuya = require('zigbee-herdsman-converters/lib/tuya');
const e = exposes.presets;
const ea = exposes.access;
const definition = {
fingerprint: [
{modelID: 'TS0601', manufacturerName: '_TZE284_81yrt3lo'},
],
model: 'TS0601_Energy_Meter',
vendor: 'Tuya',
description: 'Energy meter with 80A clamp',
fromZigbee: [tuya.fz.datapoints],
toZigbee: [],
exposes: [ //only report data
e.ac_frequency(), // Frequency in Hz
exposes.numeric('total_power_A', ea.STATE).withUnit('W').withDescription('Total power A'),
exposes.numeric('total_power_B', ea.STATE).withUnit('W').withDescription('Total power B'),
exposes.numeric('total_power_AB', ea.STATE).withUnit('W').withDescription('Total power A'),
exposes.numeric('voltage', ea.STATE).withUnit('V').withDescription('Voltage'),
exposes.numeric('current_A', ea.STATE).withUnit('A').withDescription('Current A'),
exposes.numeric('current_B', ea.STATE).withUnit('A').withDescription('Current B'),
exposes.numeric('power_factor_A', ea.STATE).withUnit('%').withDescription('Instantaneous measured pow>
exposes.numeric('power_factor_B', ea.STATE).withUnit('%').withDescription('Instantaneous measured pow>
exposes.numeric('power_direction A', ea.STATE).withDescription('Power direction A 0/1 for forward/rev>
exposes.numeric('power_direction B', ea.STATE).withDescription('Power direction B 0/1 for forward/rev>
exposes.numeric('energy_forward_A', ea.STATE).withUnit('kWh').withDescription('Total energy A forward>
exposes.numeric('energy_forward_B', ea.STATE).withUnit('kWh').withDescription('Total energy B forward>
exposes.numeric('energy_reverse_A', ea.STATE).withUnit('kWh').withDescription('Total energy A reverse>
exposes.numeric('energy_reverse_B', ea.STATE).withUnit('kWh').withDescription('Total energy B reverse>
exposes.numeric('update_frequency', ea.STATE).withUnit('sec').withDescription('Update frequency'),
],
meta: {
tuyaDatapoints: [//only report data
[111, 'ac_frequency', tuya.valueConverter.divideBy100],
[101, 'total_power_A', tuya.valueConverter.divideBy10],
[105, 'total_power_B', tuya.valueConverter.divideBy10],
[115, 'total_power_AB', tuya.valueConverter.divideBy10],
[112, 'voltage', tuya.valueConverter.divideBy10],
[113, 'current_A', tuya.valueConverter.divideBy1000],
[114, 'current_B', tuya.valueConverter.divideBy1000],
[110, 'power_factor_A', tuya.valueConverter.divideBy100],
[121, 'power_factor_B', tuya.valueConverter.divideBy100],
[102, 'power_direction A', tuya.valueConverter.raw],
[104, 'power_direction B', tuya.valueConverter.raw],
[106, 'energy_forward_A', tuya.valueConverter.divideBy100],
[108, 'energy_forward_B', tuya.valueConverter.divideBy100],
[107, 'energy_reverse_A', tuya.valueConverter.divideBy100],
[109, 'energy_reverse_B', tuya.valueConverter.divideBy100],
[129, 'update_frequency' , tuya.valueConverter.raw],
],
},
};
module.exports = definition;
Hi, I have the same problem and I would like to try your converter but with the new version 2.0 I don't know where to put the js file. I use the z2m addon from Home Assistant, can you help me?
Thanks
Hi, I was also on version 2.0, but during one of my attempts, I reverted to version 1.42. Now, I’m not sure what will happen when I update. I think there shouldn’t be any issues. Try this procedure:
- Create a Directory for External Converters
- Access your terminal (via SSH) or use the File Editor in Home Assistant.
- Create the directory if it doesn’t already exist:
mkdir -p /config/zigbee2mqtt/data/external_converters/
- Create a New External Converter File
- In the /config/zigbee2mqtt/data/external_converters/ directory, create a new file:
nano /config/zigbee2mqtt/data/external_converters/Tuya_TZE284_81yrt3lo.js
- Paste the code for your device.
const exposes = require('zigbee-herdsman-converters/lib/exposes');
const tuya = require('zigbee-herdsman-converters/lib/tuya');
const e = exposes.presets;
const ea = exposes.access;
const definition = {
fingerprint: [
{modelID: 'TS0601', manufacturerName: '_TZE284_81yrt3lo'},
],
model: 'PJ-1203A',
vendor: 'Tuya',
description: 'Energy meter with 80A clamp',
fromZigbee: [tuya.fz.datapoints],
toZigbee: [tuya.tz.datapoints],
onEvent: tuya.onEventSetTime,
configure: tuya.configureMagicPacket,
exposes: [ //only report data
e.ac_frequency(), // Frequency in Hz
exposes.numeric('total_power_A', ea.STATE).withUnit('W').withDescription('Total power A'),
exposes.numeric('total_power_B', ea.STATE).withUnit('W').withDescription('Total power B'),
exposes.numeric('total_power_AB', ea.STATE).withUnit('W').withDescription('Total power A'),
exposes.numeric('voltage', ea.STATE).withUnit('V').withDescription('Voltage'),
exposes.numeric('current_A', ea.STATE).withUnit('A').withDescription('Current A'),
exposes.numeric('current_B', ea.STATE).withUnit('A').withDescription('Current B'),
exposes.numeric('power_factor_A', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor A'),
exposes.numeric('power_factor_B', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor B'),
exposes.numeric('power_direction A', ea.STATE).withDescription('Power direction A 0/1 for forward/reverse'),
exposes.numeric('power_direction B', ea.STATE).withDescription('Power direction B 0/1 for forward/reverse'),
exposes.numeric('energy_forward_A', ea.STATE).withUnit('kWh').withDescription('Total energy A forward'),
exposes.numeric('energy_forward_B', ea.STATE).withUnit('kWh').withDescription('Total energy B forward'),
exposes.numeric('energy_reverse_A', ea.STATE).withUnit('kWh').withDescription('Total energy A reverse'),
exposes.numeric('energy_reverse_B', ea.STATE).withUnit('kWh').withDescription('Total energy B reverse'),
exposes.numeric('update_frequency', ea.STATE).withUnit('sec').withDescription('Update frequency'),
],
meta: {
tuyaDatapoints: [//only report data
[111, 'ac_frequency', tuya.valueConverter.divideBy100],
[101, 'total_power_A', tuya.valueConverter.divideBy10],
[105, 'total_power_B', tuya.valueConverter.divideBy10],
[115, 'total_power_AB', tuya.valueConverter.divideBy10],
[112, 'voltage', tuya.valueConverter.divideBy10],
[113, 'current_A', tuya.valueConverter.divideBy1000],
[114, 'current_B', tuya.valueConverter.divideBy1000],
[110, 'power_factor_A', tuya.valueConverter.divideBy100],
[121, 'power_factor_B', tuya.valueConverter.divideBy100],
[102, 'power_direction A', tuya.valueConverter.raw],
[104, 'power_direction B', tuya.valueConverter.raw],
[106, 'energy_forward_A', tuya.valueConverter.divideBy100],
[108, 'energy_forward_B', tuya.valueConverter.divideBy100],
[107, 'energy_reverse_A', tuya.valueConverter.divideBy100],
[109, 'energy_reverse_B', tuya.valueConverter.divideBy100],
[129, 'update_frequency' , tuya.valueConverter.raw],
],
},
};
module.exports = definition;
- Save the file: Press CTRL+O, confirm with Enter, then CTRL+X
- Add the Converter Path to configuration.yaml
- Open the configuration.yaml file:
nano /config/zigbee2mqtt/configuration.yaml
- Add or update the external_converters section to include your new file:
external_converters:
- data/external_converters/Tuya_TZE284_81yrt3lo.js
- Save the changes.
- Restart Zigbee2MQTT
It finally worked! Thank you very much. In my case the directory is \config\zigbee2mqtt\external_converters, without \data.
Hello, I have also installed this module, but I am using ZHA. Could someone share the quirk to make it report correctly with the data? Thank you.
Hello, I just wanted to thank you. Also, I think in next Z2M version realeasing 1 feb, they gonna add official support
Hello, do you have any suggestions for solving the issue with the device "_TZE284_81yrt3lo" using ZHA in Home Assistant?
Currently, it is not supported by the quirk available here: https://github.com/jeverley/zha-device-handlers/blob/ts0601_energy_meter_devices/zhaquirks/tuya/ts0601_energy_meter.py
If it helps, I have the device and am available to capture data and debug logs.
Thank you very much!
Hi, I was also on version 2.0, but during one of my attempts, I reverted to version 1.42. Now, I’m not sure what will happen when I update. I think there shouldn’t be any issues. Try this procedure:
- Create a Directory for External Converters
- Access your terminal (via SSH) or use the File Editor in Home Assistant.
- Create the directory if it doesn’t already exist:
mkdir -p /config/zigbee2mqtt/data/external_converters/
- Create a New External Converter File
- In the /config/zigbee2mqtt/data/external_converters/ directory, create a new file:
nano /config/zigbee2mqtt/data/external_converters/Tuya_TZE284_81yrt3lo.js
- Paste the code for your device.
const exposes = require('zigbee-herdsman-converters/lib/exposes'); const tuya = require('zigbee-herdsman-converters/lib/tuya'); const e = exposes.presets; const ea = exposes.access; const definition = { fingerprint: [ {modelID: 'TS0601', manufacturerName: '_TZE284_81yrt3lo'}, ], model: 'PJ-1203A', vendor: 'Tuya', description: 'Energy meter with 80A clamp', fromZigbee: [tuya.fz.datapoints], toZigbee: [tuya.tz.datapoints], onEvent: tuya.onEventSetTime, configure: tuya.configureMagicPacket, exposes: [ //only report data e.ac_frequency(), // Frequency in Hz exposes.numeric('total_power_A', ea.STATE).withUnit('W').withDescription('Total power A'), exposes.numeric('total_power_B', ea.STATE).withUnit('W').withDescription('Total power B'), exposes.numeric('total_power_AB', ea.STATE).withUnit('W').withDescription('Total power A'), exposes.numeric('voltage', ea.STATE).withUnit('V').withDescription('Voltage'), exposes.numeric('current_A', ea.STATE).withUnit('A').withDescription('Current A'), exposes.numeric('current_B', ea.STATE).withUnit('A').withDescription('Current B'), exposes.numeric('power_factor_A', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor A'), exposes.numeric('power_factor_B', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor B'), exposes.numeric('power_direction A', ea.STATE).withDescription('Power direction A 0/1 for forward/reverse'), exposes.numeric('power_direction B', ea.STATE).withDescription('Power direction B 0/1 for forward/reverse'), exposes.numeric('energy_forward_A', ea.STATE).withUnit('kWh').withDescription('Total energy A forward'), exposes.numeric('energy_forward_B', ea.STATE).withUnit('kWh').withDescription('Total energy B forward'), exposes.numeric('energy_reverse_A', ea.STATE).withUnit('kWh').withDescription('Total energy A reverse'), exposes.numeric('energy_reverse_B', ea.STATE).withUnit('kWh').withDescription('Total energy B reverse'), exposes.numeric('update_frequency', ea.STATE).withUnit('sec').withDescription('Update frequency'), ], meta: { tuyaDatapoints: [//only report data [111, 'ac_frequency', tuya.valueConverter.divideBy100], [101, 'total_power_A', tuya.valueConverter.divideBy10], [105, 'total_power_B', tuya.valueConverter.divideBy10], [115, 'total_power_AB', tuya.valueConverter.divideBy10], [112, 'voltage', tuya.valueConverter.divideBy10], [113, 'current_A', tuya.valueConverter.divideBy1000], [114, 'current_B', tuya.valueConverter.divideBy1000], [110, 'power_factor_A', tuya.valueConverter.divideBy100], [121, 'power_factor_B', tuya.valueConverter.divideBy100], [102, 'power_direction A', tuya.valueConverter.raw], [104, 'power_direction B', tuya.valueConverter.raw], [106, 'energy_forward_A', tuya.valueConverter.divideBy100], [108, 'energy_forward_B', tuya.valueConverter.divideBy100], [107, 'energy_reverse_A', tuya.valueConverter.divideBy100], [109, 'energy_reverse_B', tuya.valueConverter.divideBy100], [129, 'update_frequency' , tuya.valueConverter.raw], ], }, }; module.exports = definition;
- Save the file: Press CTRL+O, confirm with Enter, then CTRL+X
- Add the Converter Path to configuration.yaml
- Open the configuration.yaml file:
nano /config/zigbee2mqtt/configuration.yaml
- Add or update the external_converters section to include your new file:
external_converters: - data/external_converters/Tuya_TZE284_81yrt3lo.js
- Save the changes.
- Restart Zigbee2MQTT
it works great! thanks!
Hello, same here, any news on support of this device through ZHA? Thank you!
Hi, I was also on version 2.0, but during one of my attempts, I reverted to version 1.42. Now, I’m not sure what will happen when I update. I think there shouldn’t be any issues. Try this procedure:
- Create a Directory for External Converters
- Access your terminal (via SSH) or use the File Editor in Home Assistant.
- Create the directory if it doesn’t already exist:
mkdir -p /config/zigbee2mqtt/data/external_converters/
- Create a New External Converter File
- In the /config/zigbee2mqtt/data/external_converters/ directory, create a new file:
nano /config/zigbee2mqtt/data/external_converters/Tuya_TZE284_81yrt3lo.js
- Paste the code for your device.
const exposes = require('zigbee-herdsman-converters/lib/exposes'); const tuya = require('zigbee-herdsman-converters/lib/tuya'); const e = exposes.presets; const ea = exposes.access; const definition = { fingerprint: [ {modelID: 'TS0601', manufacturerName: '_TZE284_81yrt3lo'}, ], model: 'PJ-1203A', vendor: 'Tuya', description: 'Energy meter with 80A clamp', fromZigbee: [tuya.fz.datapoints], toZigbee: [tuya.tz.datapoints], onEvent: tuya.onEventSetTime, configure: tuya.configureMagicPacket, exposes: [ //only report data e.ac_frequency(), // Frequency in Hz exposes.numeric('total_power_A', ea.STATE).withUnit('W').withDescription('Total power A'), exposes.numeric('total_power_B', ea.STATE).withUnit('W').withDescription('Total power B'), exposes.numeric('total_power_AB', ea.STATE).withUnit('W').withDescription('Total power A'), exposes.numeric('voltage', ea.STATE).withUnit('V').withDescription('Voltage'), exposes.numeric('current_A', ea.STATE).withUnit('A').withDescription('Current A'), exposes.numeric('current_B', ea.STATE).withUnit('A').withDescription('Current B'), exposes.numeric('power_factor_A', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor A'), exposes.numeric('power_factor_B', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor B'), exposes.numeric('power_direction A', ea.STATE).withDescription('Power direction A 0/1 for forward/reverse'), exposes.numeric('power_direction B', ea.STATE).withDescription('Power direction B 0/1 for forward/reverse'), exposes.numeric('energy_forward_A', ea.STATE).withUnit('kWh').withDescription('Total energy A forward'), exposes.numeric('energy_forward_B', ea.STATE).withUnit('kWh').withDescription('Total energy B forward'), exposes.numeric('energy_reverse_A', ea.STATE).withUnit('kWh').withDescription('Total energy A reverse'), exposes.numeric('energy_reverse_B', ea.STATE).withUnit('kWh').withDescription('Total energy B reverse'), exposes.numeric('update_frequency', ea.STATE).withUnit('sec').withDescription('Update frequency'), ], meta: { tuyaDatapoints: [//only report data [111, 'ac_frequency', tuya.valueConverter.divideBy100], [101, 'total_power_A', tuya.valueConverter.divideBy10], [105, 'total_power_B', tuya.valueConverter.divideBy10], [115, 'total_power_AB', tuya.valueConverter.divideBy10], [112, 'voltage', tuya.valueConverter.divideBy10], [113, 'current_A', tuya.valueConverter.divideBy1000], [114, 'current_B', tuya.valueConverter.divideBy1000], [110, 'power_factor_A', tuya.valueConverter.divideBy100], [121, 'power_factor_B', tuya.valueConverter.divideBy100], [102, 'power_direction A', tuya.valueConverter.raw], [104, 'power_direction B', tuya.valueConverter.raw], [106, 'energy_forward_A', tuya.valueConverter.divideBy100], [108, 'energy_forward_B', tuya.valueConverter.divideBy100], [107, 'energy_reverse_A', tuya.valueConverter.divideBy100], [109, 'energy_reverse_B', tuya.valueConverter.divideBy100], [129, 'update_frequency' , tuya.valueConverter.raw], ], }, }; module.exports = definition;
- Save the file: Press CTRL+O, confirm with Enter, then CTRL+X
- Add the Converter Path to configuration.yaml
- Open the configuration.yaml file:
nano /config/zigbee2mqtt/configuration.yaml
- Add or update the external_converters section to include your new file:
external_converters: - data/external_converters/Tuya_TZE284_81yrt3lo.js
- Save the changes.
- Restart Zigbee2MQTT
Works like a charm, thank you very much Mr. Z2M Wizard :)
Hello, do you have any suggestions for solving the issue with the device "_TZE284_81yrt3lo" using ZHA in Home Assistant?
Currently, it is not supported by the quirk available here: https://github.com/jeverley/zha-device-handlers/blob/ts0601_energy_meter_devices/zhaquirks/tuya/ts0601_energy_meter.py
If it helps, I have the device and am available to capture data and debug logs.
Thank you very much!
I'm a beginner so I'm sure I have made some mistakes, but I have added a _TZE284_81yrt3lo version to the bottom of @jeverley quirk and it now works perfectly, I hope this helps everyone.
"""Tuya Energy Meter."""
from enum import Enum
from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
from zigpy.profiles import zgp, zha
from zigpy.quirks import CustomDevice
import zigpy.types as t
from zigpy.zcl.clusters.general import Basic, GreenPowerProxy, Groups, Ota, Scenes, Time
from zigpy.zcl.foundation import ZCLAttributeDef
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.tuya import (
NoManufacturerCluster,
TuyaLocalCluster,
TuyaZBElectricalMeasurement,
TuyaZBMeteringClusterWithUnit,
)
from zhaquirks.tuya.mcu import DPToAttributeMapping, TuyaMCUCluster
# from zigpy.zcl.clusters.homeautomation import MeasurementType
# Manufacturer cluster identifiers for device signatures
EARU_MANUFACTURER_CLUSTER_ID = 0xFF66
# Offset of 512 (0x200) for transating DP ID to Attribute ID
# Attribute IDs don't need to match every device's specific values
DP_ATTR_OFFSET = 512
# Power direction acttributes
POWER_FLOW = 102 + DP_ATTR_OFFSET # PowerFlow (0: forward, 1: reverse)
POWER_FLOW_B = 104 + DP_ATTR_OFFSET # PowerFlow (0: forward, 1: reverse)
# Calibration attributes
AC_FREQUENCY_COEF = 122 + DP_ATTR_OFFSET # uint32_t_be
CURRENT_SUMM_DELIVERED_COEF = 119 + DP_ATTR_OFFSET # uint32_t_be
CURRENT_SUMM_DELIVERED_COEF_B = 125 + DP_ATTR_OFFSET # uint32_t_be
CURRENT_SUMM_RECEIVED_COEF = 127 + DP_ATTR_OFFSET # uint32_t_be
CURRENT_SUMM_RECEIVED_COEF_B = 128 + DP_ATTR_OFFSET # uint32_t_be
INSTANTANEOUS_DEMAND_COEF = 118 + DP_ATTR_OFFSET # uint32_t_be
INSTANTANEOUS_DEMAND_COEF_B = 124 + DP_ATTR_OFFSET # uint32_t_be
RMS_CURRENT_COEF = 117 + DP_ATTR_OFFSET # uint32_t_be
RMS_CURRENT_COEF_B = 123 + DP_ATTR_OFFSET # uint32_t_be
RMS_VOLTAGE_COEF = 116 + DP_ATTR_OFFSET # uint32_t_be
# Device configuration attributes
UPDATE_PERIOD = 129 + DP_ATTR_OFFSET # uint32_t_be (3-60 seconds supported)
# Local configuration attributes
CHANNEL_CONFIGURATION = 0x5000
SUPPRESS_REVERSE_FLOW = 0x5010
SUPPRESS_REVERSE_FLOW_B = 0x5011
POWER_FLOW_PREEMPT = 0x5020
# Suffix for device attributes which need power flow direction applied
UNSIGNED_POWER_ATTR_SUFFIX = "_attr_unsigned"
# Default Tuya MCU cluster endpoint_id
TUYA_MCU_ENDPOINT_ID = 1
def is_type_uint(attr_type: Type) -> bool:
"""True if the specified attribute type is an unsigned integer."""
return issubclass(attr_type, t.uint_t)
class Channel(str, Enum):
"""Meter channels."""
A = "a"
B = "b"
AB = "ab"
@classmethod
def attr_with_channel(cls, attr_name: str, channel=None) -> str:
"""Returns the attr_name with channel suffix."""
assert channel is None or channel in cls, "Invalid channel."
if channel and channel != cls.A:
attr_name = attr_name + "_ch_" + channel
return attr_name
class ChannelConfiguration(t.enum8):
"""Enums for for all energy meter configurations."""
NONE = 0x00
A_PLUS_B = 0x01
A_MINUS_B = 0x02
GRID_PLUS_PRODUCTION = 0x03
CONSUMPTION_MINUS_PRODUCTION = 0x04
class ChannelConfiguration_1CH(t.enum8):
"""Enums for 1 channel energy meter configuration."""
NONE = ChannelConfiguration.NONE
DEFAULT = NONE
class ChannelConfiguration_1CHB(t.enum8):
"""Enums for 1 channel bidirectional energy meter configuration."""
NONE = ChannelConfiguration.NONE
DEFAULT = NONE
class ChannelConfiguration_2CH(t.enum8):
"""Enums for 2 channel energy meter configuration."""
A_PLUS_B = ChannelConfiguration.A_PLUS_B
A_MINUS_B = ChannelConfiguration.A_MINUS_B
CONSUMPTION_MINUS_PRODUCTION = ChannelConfiguration.CONSUMPTION_MINUS_PRODUCTION
DEFAULT = CONSUMPTION_MINUS_PRODUCTION
class ChannelConfiguration_2CHB(t.enum8):
"""Enums for 2 channel bidirectional energy meter configuration."""
A_PLUS_B = ChannelConfiguration.A_PLUS_B
A_MINUS_B = ChannelConfiguration.A_MINUS_B
GRID_PLUS_PRODUCTION = ChannelConfiguration.GRID_PLUS_PRODUCTION
CONSUMPTION_MINUS_PRODUCTION = ChannelConfiguration.CONSUMPTION_MINUS_PRODUCTION
DEFAULT = GRID_PLUS_PRODUCTION
class MeasurementType(
t.bitmap32
): # Would like to import this from zigpy.zcl.clusters.homeautomation, but its offset is currently incorrect
"""Defines the measurement type bits for the ElectricalMeasurement cluster."""
Active_measurement_AC = 1 << 0
Reactive_measurement_AC = 1 << 1
Apparent_measurement_AC = 1 << 2
Phase_A_measurement = 1 << 3
Phase_B_measurement = 1 << 4
Phase_C_measurement = 1 << 5
DC_measurement = 1 << 6
Harmonics_measurement = 1 << 7
Power_quality_measurement = 1 << 8
class Metering:
"""Functions for use with the ZCL Metering cluster."""
@staticmethod
def format(
int_digits: int, dec_digits: int, suppress_leading_zeros: bool = True
) -> int:
"""Returns the formatter value for summation and demand Metering attributes."""
assert 0 <= int_digits <= 7, "int_digits must be within range of 0 to 7."
assert 0 <= dec_digits <= 7, "dec_digits must be within range of 0 to 7."
return (suppress_leading_zeros << 6) | (int_digits << 3) | dec_digits
class PowerFlow(t.enum1):
"""Indicates power flow direction."""
FORWARD = 0x0
REVERSE = 0x1
@classmethod
def align_value(cls, value: int, power_flow=None) -> int:
"""Aligns the value with the power_flow direction."""
if (
power_flow == cls.REVERSE
and value > 0
or power_flow == cls.FORWARD
and value < 0
):
value = -value
return value
class TuyaPowerPhase:
"""Extracts values from Tuya power phase datapoints."""
@staticmethod
def variant_1(value) -> Tuple[t.uint_t, t.uint_t]:
voltage = value[14] | value[13] << 8
current = value[12] | value[11] << 8
return voltage, current
@staticmethod
def variant_2(value) -> Tuple[t.uint_t, t.uint_t, int]:
voltage = value[1] | value[0] << 8
current = value[4] | value[3] << 8
power = value[7] | value[6] << 8
return voltage, current, power * 10
@staticmethod
def variant_3(value) -> Tuple[t.uint_t, t.uint_t, int]:
voltage = (value[0] << 8) | value[1]
current = (value[2] << 16) | (value[3] << 8) | value[4]
power = (value[5] << 16) | (value[6] << 8) | value[7]
return voltage, current, power * 10
class PowerCalculation:
"""Methods for calculating power values."""
@staticmethod
def active_power_from_apparent_power_power_factor_and_power_flow(
apparent_power: Optional[t.uint_t],
power_factor: Optional[t.int_t],
power_flow: Optional[PowerFlow] = None,
) -> Optional[t.int_t]:
if apparent_power is None or power_factor is None:
return
power_factor *= 0.01
return round(apparent_power * abs(power_factor) * (-1 if power_flow else 1))
@staticmethod
def apparent_power_from_active_power_and_power_factor(
active_power: Optional[t.int_t], power_factor: Optional[t.int_t]
) -> Optional[t.uint_t]:
if active_power is None or power_factor is None:
return
power_factor *= 0.01
return round(abs(active_power) / abs(power_factor))
@staticmethod
def apparent_power_from_rms_current_and_rms_voltage(
rms_current: Optional[t.uint_t],
rms_voltage: Optional[t.uint_t],
ac_current_divisor: int = 1,
ac_current_multiplier: int = 1,
ac_voltage_divisor: int = 1,
ac_voltage_multiplier: int = 1,
ac_power_divisor: int = 1,
ac_power_multiplier: int = 1,
) -> Optional[t.uint_t]:
if rms_current is None or rms_voltage is None:
return
return round(
(rms_current * ac_current_multiplier / ac_current_divisor)
* (rms_voltage * ac_voltage_multiplier / ac_voltage_divisor)
* ac_power_divisor
/ ac_power_multiplier
)
@staticmethod
def reactive_power_from_apparent_power_and_power_factor(
apparent_power: Optional[t.uint_t], power_factor: Optional[t.int_t]
) -> Optional[t.int_t]:
if apparent_power is None or power_factor is None:
return
power_factor *= 0.01
return round(
(apparent_power * (1 - power_factor**2) ** 0.5)
* (-1 if power_factor < 0 else 1)
)
class LocalClusterAttributes:
"""Methods for handling local configuration attributes on device."""
_ATTRIBUTE_DEFAULTS: Dict[int, Any] = {}
_LOCAL_ATTRIBUTES: Tuple[int] = ()
def _attr_default(
self, attrid: Union[str, int], default: Optional[Any] = None
) -> Optional[Any]:
"""Returns an attribute's default value."""
attr_def = self.find_attribute(attrid)
return self._ATTRIBUTE_DEFAULTS.get(
attr_def.id, getattr(attr_def.type, "DEFAULT", default)
)
def _format_attr_value(self, attrid: Union[str, int], value: Any) -> Optional[Any]:
"""Used to format the input the input value with the attribute's type."""
try:
attr_def = self.find_attribute(attrid)
value = attr_def.type(value)
return value
except KeyError:
self.error("%s is not a valid attribute id", attrid)
except ValueError as e:
self.error(
"Failed to convert attribute %s from %s (%s) to type %s: %s",
attr_def.id,
value,
type(value),
attr_def.type,
e,
)
return
def get(self, key: Union[int, str], default: Optional[Any] = None) -> Optional[Any]:
"""Get cached attribute value and fall back to its device/type default if defined."""
value = super().get(key, default)
if value is None:
value = self._attr_default(key, default)
return value
async def read_attributes(self, attributes, *args, **kwargs):
"""Handle reads to local configuration attributes."""
success, failure = await super().read_attributes(attributes, *args, **kwargs)
for attrid in set(self._LOCAL_ATTRIBUTES).intersection(set(attributes)):
if attrid not in success:
default = self._attr_default(attrid)
if default is None:
continue
success[attrid] = default
failure.pop(attrid, None)
if success[attrid] not in (None, ""):
success[attrid] = self.attributes[attrid].type(success[attrid])
return success, failure
async def write_attributes(self, attributes, *args, **kwargs):
"""Handle writes to local configuration attributes."""
local_attributes = {}
for attrid in set(self._LOCAL_ATTRIBUTES).intersection(set(attributes)):
value = attributes.pop(attrid)
if value in (None, ""):
local_attributes[attrid] = None
continue
value = self._format_attr_value(attrid, value)
if value is not None:
local_attributes[attrid] = value
await TuyaLocalCluster.write_attributes(self, local_attributes, *args, **kwargs)
return await super().write_attributes(attributes, *args, **kwargs)
class TuyaEnergyMeterManufCluster(
LocalClusterAttributes, NoManufacturerCluster, TuyaMCUCluster
):
"""Manufactuter cluster for Tuya energy meter devices."""
_CHANNEL_CONFIGURATION_ATTRIBUTES: Dict[Type, Tuple[int]] = {
ChannelConfiguration_1CHB: (SUPPRESS_REVERSE_FLOW,),
ChannelConfiguration_2CHB: (
POWER_FLOW_PREEMPT,
SUPPRESS_REVERSE_FLOW,
SUPPRESS_REVERSE_FLOW_B,
),
}
_LOCAL_ATTRIBUTES: Tuple[int] = (
CHANNEL_CONFIGURATION,
POWER_FLOW_PREEMPT,
SUPPRESS_REVERSE_FLOW,
SUPPRESS_REVERSE_FLOW_B,
)
attributes: Dict[int, ZCLAttributeDef] = {
AC_FREQUENCY_COEF: ("ac_frequency_coefficient", t.uint32_t_be, True),
CURRENT_SUMM_DELIVERED_COEF: (
"current_summ_delivered_coefficient",
t.uint32_t_be,
True,
),
CURRENT_SUMM_DELIVERED_COEF_B: (
"current_summ_delivered_coefficient_ch_b",
t.uint32_t_be,
True,
),
CURRENT_SUMM_RECEIVED_COEF: (
"current_summ_received_coefficient",
t.uint32_t_be,
True,
),
CURRENT_SUMM_RECEIVED_COEF_B: (
"current_summ_received_coefficient_ch_b",
t.uint32_t_be,
True,
),
INSTANTANEOUS_DEMAND_COEF: (
"instantaneous_demand_coefficient",
t.uint32_t_be,
True,
),
INSTANTANEOUS_DEMAND_COEF_B: (
"instantaneous_demand_coefficient_ch_b",
t.uint32_t_be,
True,
),
POWER_FLOW: ("power_flow", PowerFlow, True),
POWER_FLOW_B: ("power_flow_ch_b", PowerFlow, True),
RMS_CURRENT_COEF: ("rms_current_coefficient", t.uint32_t_be, True),
RMS_CURRENT_COEF_B: (
"rms_current_coefficient_ch_b",
t.uint32_t_be,
True,
),
RMS_VOLTAGE_COEF: ("rms_voltage_coefficient", t.uint32_t_be, True),
CHANNEL_CONFIGURATION: (
"channel_configuration",
ChannelConfiguration,
True,
),
UPDATE_PERIOD: ("update_period", t.uint32_t_be, True),
POWER_FLOW_PREEMPT: ("power_flow_preempt", t.Bool, True),
SUPPRESS_REVERSE_FLOW: ("suppress_reverse_flow", t.Bool, True),
SUPPRESS_REVERSE_FLOW_B: ("suppress_reverse_flow_ch_b", t.Bool, True),
}
def get_optional(
self, key: Union[int, str], default: Optional[Any] = None
) -> Optional[Any]:
"""Returns the provided default value or None if an attribute is undefined."""
try:
return self.get(key, default)
except KeyError:
return default
def __init_subclass__(cls, configuration_type: Type) -> None:
"""Init cluster subclass."""
cls.attributes = {**TuyaMCUCluster.attributes}
cls._populate_mapped_attributes_lookup(cls)
cls._setup_channel_config_attributes(cls, configuration_type)
cls._setup_device_attributes(cls)
super().__init_subclass__()
def _populate_mapped_attributes_lookup(cls) -> None:
"""Stores a tuple for each cluster attribute mapped from MCU data points."""
cls.mapped_attributes: Tuple[Tuple[str, str, int]] = tuple(
(dp_map.ep_attribute, attr_name, dp_map.endpoint_id or TUYA_MCU_ENDPOINT_ID)
for dp_map in cls.dp_to_attribute.values()
for attr_name in (
dp_map.attribute_name
if isinstance(dp_map.attribute_name, tuple)
else (dp_map.attribute_name,)
)
)
def _setup_channel_config_attributes(cls, configuration_type: Type) -> None:
"""Setup local attributes for the device channel configuration type."""
config_type_attr = TuyaEnergyMeterManufCluster.attributes[CHANNEL_CONFIGURATION]
cls.attributes[CHANNEL_CONFIGURATION] = (
config_type_attr.name,
configuration_type,
config_type_attr.is_manufacturer_specific,
)
config_attr = cls._CHANNEL_CONFIGURATION_ATTRIBUTES.get(configuration_type, ())
for attrid in config_attr:
cls.attributes[attrid] = TuyaEnergyMeterManufCluster.attributes[attrid]
def _setup_device_attributes(cls) -> None:
"""Setup manufacturer cluster attributes for mapped device data points."""
attr_name_to_id: Dict[str, int] = {
attr[0] if isinstance(attr, tuple) else attr.name: attrid
for attrid, attr in TuyaEnergyMeterManufCluster.attributes.items()
}
for ep_attribute, attr_name, endpoint_id in cls.mapped_attributes:
if ep_attribute != cls.ep_attribute:
continue
assert (
endpoint_id == 1
), "Check endpoint_id of TuyaEnergyMeterManufCluster dp_to_attribute."
attrid = attr_name_to_id.get(attr_name)
if attrid is not None:
cls.attributes[attrid] = TuyaEnergyMeterManufCluster.attributes[attrid]
class EnergyMeterChannel:
"""Methods and properties for energy meter channel clusters."""
_ENDPOINT_TO_CHANNEL: Dict[Tuple[Type, int], Channel] = {
(ChannelConfiguration_1CH, 1): Channel.A,
(ChannelConfiguration_1CHB, 1): Channel.A,
(ChannelConfiguration_2CH, 1): Channel.A,
(ChannelConfiguration_2CH, 2): Channel.B,
(ChannelConfiguration_2CH, 3): Channel.AB,
(ChannelConfiguration_2CHB, 1): Channel.A,
(ChannelConfiguration_2CHB, 2): Channel.B,
(ChannelConfiguration_2CHB, 3): Channel.AB,
}
_EXTENSIVE_ATTRIBUTES: Tuple[str] = ()
_INTENSIVE_ATTRIBUTES: Tuple[str] = ()
_CUMULATIVE_FORWARD_ATTRIBUTES: Tuple[str] = ()
_CUMULATIVE_REVERSE_ATTRIBUTES: Tuple[str] = ()
_INVERSE_ATTRIBUTES: Dict[str, str] = {}
def __init__(self, *args, **kwargs):
"""Init."""
self._CHANNEL_TO_ENDPOINT: Dict[Tuple[Type, Channel], int] = {
(k[0], v): k[1] for k, v in self._ENDPOINT_TO_CHANNEL.items()
}
self._INVERSE_ATTRIBUTES.update(
{v: k for k, v in dict(self._INVERSE_ATTRIBUTES).items()}
)
self._CUMULATIVE_ATTRIBUTES = (
self._CUMULATIVE_FORWARD_ATTRIBUTES + self._CUMULATIVE_REVERSE_ATTRIBUTES
)
super().__init__(*args, **kwargs)
@property
def channel(self) -> Optional[str]:
"""Returns the cluster's channel."""
return self._ENDPOINT_TO_CHANNEL.get(
(self.channel_configuration_type, self.endpoint.endpoint_id), None
)
@property
def channel_configuration(self) -> Optional[ChannelConfiguration]:
"""Returns the device's current channel configuration."""
return self.manufacturer_cluster.get("channel_configuration")
@property
def channel_configuration_type(self) -> Type:
"""Returns the device's channel configuration type."""
return self.manufacturer_cluster.AttributeDefs.channel_configuration.type
@property
def manufacturer_cluster(self) -> TuyaEnergyMeterManufCluster:
"""Returns the device's manufacturer cluster."""
return getattr(
self.endpoint.device.endpoints[TUYA_MCU_ENDPOINT_ID],
TuyaEnergyMeterManufCluster.ep_attribute,
)
def attr_present(
self,
*attr_names: str,
ep_attribute: Optional[str] = None,
endpoint_id: Optional[int] = None,
) -> bool:
"""Returns True if any of the specified attributes are provided by the device."""
ep_attribute = ep_attribute or self.ep_attribute
endpoint_id = endpoint_id or self.endpoint.endpoint_id
return any(
attr in self.manufacturer_cluster.mapped_attributes
for attr in tuple(
(ep_attribute, attr_name, endpoint_id) for attr_name in attr_names
)
)
def attr_type(self, attr_name: str) -> Type:
"""Returns the type of the specified attribute."""
return getattr(self.AttributeDefs, attr_name).type
def get_cluster(
self,
channel_or_endpoint_id: Union[Channel, int],
ep_attribute: Optional[str] = None,
):
"""Returns the device cluster for the given channel or endpoint."""
if channel_or_endpoint_id in Channel:
channel_or_endpoint_id = self._CHANNEL_TO_ENDPOINT.get(
(self.channel_configuration_type, channel_or_endpoint_id), None
)
assert channel_or_endpoint_id is not None, "Invalid channel_or_endpoint_id."
return getattr(
self.endpoint.device.endpoints[channel_or_endpoint_id],
ep_attribute or self.ep_attribute,
)
def update_calculated_attribute(self, attr_name: str, calculated_value) -> None:
"""Updates the specified attribute if the calculated value is valid."""
if calculated_value is None:
return
self.update_attribute(attr_name, calculated_value)
class EnergyMeterPowerFlow(EnergyMeterChannel):
"""Methods and properties for handling power flow on Tuya energy meter devices."""
@property
def power_flow(self) -> Optional[PowerFlow]:
"""Returns the channel's current power flow direction."""
return self.manufacturer_cluster.get_optional(
Channel.attr_with_channel("power_flow", self.channel)
)
@power_flow.setter
def power_flow(self, value: PowerFlow) -> None:
"""Updates the channel's power flow direction."""
self.manufacturer_cluster.update_attribute(
Channel.attr_with_channel("power_flow", self.channel), value
)
@property
def suppress_reverse_flow(self) -> bool:
"""Returns True if suppress_reverse_flow is enabled for the channel."""
return self.manufacturer_cluster.get_optional(
Channel.attr_with_channel("suppress_reverse_flow", self.channel), False
)
def _align_unsigned_attribute_with_power_flow(
self, attr_name: str, value
) -> Tuple[str, Any]:
"""Attributes marked as unsigned are aligned with the current power flow direction."""
if attr_name.endswith(UNSIGNED_POWER_ATTR_SUFFIX):
attr_name = attr_name.removesuffix(UNSIGNED_POWER_ATTR_SUFFIX)
value = PowerFlow.align_value(value, self.power_flow)
return attr_name, value
def _suppress_reverse_power_flow(self, attr_name: str, value) -> Optional[Any]:
"""Returns 0 if suppress_reverse_flow is enabled for the channel and power flow is reverse."""
if self.suppress_reverse_flow and (
attr_name in self._EXTENSIVE_ATTRIBUTES
and self.power_flow == PowerFlow.REVERSE
or attr_name in self._CUMULATIVE_REVERSE_ATTRIBUTES
):
value = 0
return value
def power_flow_handler(self, attr_name: str, value) -> Tuple[str, Any]:
"""Orchestrates processing of directional attributes."""
attr_name, value = self._align_unsigned_attribute_with_power_flow(
attr_name, value
)
value = self._suppress_reverse_power_flow(attr_name, value)
return attr_name, value
class PowerFlowPreemptConfiguration:
"""Contains the parameters for preempting power_flow direction."""
def __init__(
self,
source_channels: tuple = (),
trigger_channel: Optional[Channel] = None,
preempt_method: Optional[Callable] = None,
) -> None:
self.source_channels = source_channels
self.trigger_channel = trigger_channel
self.preempt_method = preempt_method
class PowerFlowPreempt(EnergyMeterPowerFlow, EnergyMeterChannel):
"""Logic for preempting delayed power flow direction change on 2 channel devices."""
HOLD = "hold"
PREEMPT = "preempt"
RELEASE = "release"
@property
def power_flow_preempt(self) -> bool:
"""Returns True if power_flow_preempt is enabled for the device."""
return self.manufacturer_cluster.get_optional("power_flow_preempt", False)
def __init__(self, *args, **kwargs):
"""Init."""
self._preempt_values: Dict[str, Optional[int]] = {}
super().__init__(*args, **kwargs)
def _preempt_grid_plus_production(self, attr_name: str) -> None:
"""Power flow preempt method for grid_plus_production configured devices."""
cluster_a = self.get_cluster(Channel.A)
cluster_b = self.get_cluster(Channel.B)
value_a = cluster_a._get_preempt_value(attr_name)
value_b = cluster_b._get_preempt_value(attr_name)
if None in (value_a, value_b):
return
cluster_a.power_flow = (
PowerFlow.FORWARD
if cluster_a.power_flow == PowerFlow.REVERSE and abs(value_a) > abs(value_b)
else cluster_a.power_flow
)
cluster_b.power_flow = (
PowerFlow.FORWARD
if cluster_b.power_flow == PowerFlow.REVERSE and abs(value_b) > abs(value_a)
else cluster_b.power_flow
)
_PREEMPT_CONFIGURATION: Dict[
ChannelConfiguration, PowerFlowPreemptConfiguration
] = {
ChannelConfiguration.GRID_PLUS_PRODUCTION: PowerFlowPreemptConfiguration(
(Channel.A, Channel.B),
Channel.B,
_preempt_grid_plus_production,
),
}
def _preempt_action(
self, attr_name: str, value: int, trigger_channel: Channel
) -> str:
"""Returns the action for the power flow preempt handler."""
if self.channel == trigger_channel:
return self.PREEMPT
if self._get_preempt_value(attr_name) != value:
return self.HOLD
return self.RELEASE
def _get_preempt_value(self, attr_name: str) -> Optional[int]:
"""Retrieves the value which was held for consideration in the preempt method."""
return self._preempt_values.get(attr_name, None)
def _store_preempt_value(self, attr_name: str, value: Optional[int]) -> None:
"""Stores the value for consideration in the preempt method."""
self._preempt_values[attr_name] = value
def _release_preempt_values(
self, attr_name: str, source_channels: Tuple[Channel], trigger_channel: Channel
) -> None:
"""Releases held values to update the cluster attributes following the preempt method."""
for channel in source_channels:
cluster = self.get_cluster(channel)
if channel != trigger_channel:
value = cluster._get_preempt_value(attr_name)
if value is not None:
cluster.update_attribute(attr_name, value)
cluster._store_preempt_value(attr_name, None)
def power_flow_preempt_handler(self, attr_name: str, value) -> Optional[str]:
"""Compensates for delay in reported power flow direction."""
if (
not self.power_flow_preempt
or attr_name.removesuffix(UNSIGNED_POWER_ATTR_SUFFIX)
not in self._EXTENSIVE_ATTRIBUTES
or not self.attr_present(attr_name)
):
return
config = self._PREEMPT_CONFIGURATION.get(
self.channel_configuration, PowerFlowPreemptConfiguration()
)
if not config.preempt_method or self.channel not in config.source_channels:
return
action = self._preempt_action(attr_name, value, config.trigger_channel)
if action != self.RELEASE:
self._store_preempt_value(attr_name, value)
if action != self.PREEMPT:
return action
config.preempt_method(self, attr_name)
self._release_preempt_values(
attr_name, config.source_channels, config.trigger_channel
)
return action
class VirtualChannelConfiguration:
"""Contains the parameters for updating a virtual channel."""
def __init__(
self,
virtual_channel: Optional[Channel] = None,
source_channels: tuple = (),
trigger_channel: Optional[Channel] = None,
discrete_method: Optional[Callable] = None,
cumulative_method: Optional[Callable] = None,
) -> None:
self.virtual_channel = virtual_channel
self.source_channels = source_channels
self.trigger_channel = trigger_channel
self.discrete_method = discrete_method
self.cumulative_method = cumulative_method
class VirtualChannel(EnergyMeterPowerFlow, EnergyMeterChannel):
"""Methods and properties for updating virtual energy meter channel attributes."""
@property
def virtual_channel(self) -> Optional[Channel]:
"""Returns the virtual channel for the current configuration."""
return self._VIRTUAL_CHANNEL_CONFIGURATION.get(
self.channel_configuration,
VirtualChannelConfiguration(),
).virtual_channel
def __init__(self, *args, **kwargs):
"""Init."""
self._virtual_channel_stored_values: Dict[str, Dict[str, int]] = {}
super().__init__(*args, **kwargs)
def _a_plus_b(self, attr_name: str) -> Optional[int]:
"""Method for calculating virtual channel values in a_plus_b configuration types."""
cluster_a = self.get_cluster(Channel.A)
cluster_b = self.get_cluster(Channel.B)
value_a = cluster_a.get(attr_name)
value_b = cluster_b.get(attr_name)
if None in (value_a, value_b):
return
if attr_name in self._EXTENSIVE_ATTRIBUTES and is_type_uint(
self.attr_type(attr_name)
):
value_a = PowerFlow.align_value(value_a, cluster_a.power_flow)
value_b = PowerFlow.align_value(value_b, cluster_b.power_flow)
return value_a + value_b
def _a_minus_b(self, attr_name: str) -> Optional[int]:
"""Method for calculating virtual channel values in a_minus_b configuration types."""
cluster_a = self.get_cluster(Channel.A)
cluster_b = self.get_cluster(Channel.B)
value_a = cluster_a.get(attr_name)
value_b = cluster_b.get(attr_name)
if None in (value_a, value_b):
return
if attr_name in self._EXTENSIVE_ATTRIBUTES and is_type_uint(
self.attr_type(attr_name)
):
value_a = PowerFlow.align_value(value_a, cluster_a.power_flow)
value_b = PowerFlow.align_value(value_b, cluster_b.power_flow)
return value_a - value_b
def _cumulative_grid_plus_production(self, attr_name: str) -> Optional[t.uint_t]:
"""Method for calculating cumulative virtual channel values in grid_plus_production configuration."""
if attr_name in self._CUMULATIVE_REVERSE_ATTRIBUTES:
return 0
inv_attr_name = self._INVERSE_ATTRIBUTES.get(attr_name, None)
assert (
inv_attr_name is not None
), "An inverse attribute must be defined for cumulative values."
cluster_a = self.get_cluster(Channel.A)
cluster_b = self.get_cluster(Channel.B)
value_a = cluster_a.get(attr_name)
value_a_inv = cluster_a.get(inv_attr_name)
value_b = cluster_b.get(attr_name)
value_b_inv = cluster_b.get(inv_attr_name)
if None in (value_a, value_a_inv, value_b, value_b_inv):
return
return (value_a + value_b) - (value_a_inv + value_b_inv)
def _cumulative_consumption_minus_production(
self, attr_name: str
) -> Optional[t.uint_t]:
"""Method for calculating cumulative virtual channel values in consumption_minus_production configuration."""
inv_attr_name = self._INVERSE_ATTRIBUTES.get(attr_name, None)
assert (
inv_attr_name is not None
), "An inverse attribute must be defined for cumulative values."
cluster_a = self.get_cluster(Channel.A)
cluster_b = self.get_cluster(Channel.B)
cluster_ab = self.get_cluster(Channel.AB)
value_a = cluster_a.get(attr_name)
value_a_inv = cluster_a.get(inv_attr_name)
value_b = cluster_b.get(attr_name)
value_b_inv = cluster_b.get(inv_attr_name)
value_ab = cluster_ab.get(attr_name, 0)
value_a_prev = cluster_a._get_previous_value(attr_name)
value_a_inv_prev = cluster_a._get_previous_value(inv_attr_name, attr_name)
value_b_prev = cluster_a._get_previous_value(attr_name)
value_b_inv_prev = cluster_b._get_previous_value(inv_attr_name, attr_name)
cluster_a._store_current_value(attr_name)
cluster_a._store_current_value(inv_attr_name, attr_name)
cluster_b._store_current_value(attr_name)
cluster_b._store_current_value(inv_attr_name, attr_name)
if None in (value_a, value_a_inv, value_b, value_b_inv):
return
delta = (value_a - value_a_prev) - (value_b - value_b_prev)
delta_inv = (value_a_inv - value_a_inv_prev) - (value_b_inv - value_b_inv_prev)
return (
value_ab + (delta if delta > 0 else 0) - (delta_inv if delta_inv < 0 else 0)
)
_VIRTUAL_CHANNEL_CONFIGURATION: Dict[
ChannelConfiguration, VirtualChannelConfiguration
] = {
ChannelConfiguration.A_PLUS_B: VirtualChannelConfiguration(
Channel.AB,
(Channel.A, Channel.B),
Channel.B,
_a_plus_b,
_a_plus_b,
),
ChannelConfiguration.A_MINUS_B: VirtualChannelConfiguration(
Channel.AB,
(Channel.A, Channel.B),
Channel.B,
_a_minus_b,
_a_minus_b,
),
ChannelConfiguration.GRID_PLUS_PRODUCTION: VirtualChannelConfiguration(
Channel.AB,
(Channel.A, Channel.B),
Channel.B,
_a_plus_b,
_cumulative_grid_plus_production,
),
ChannelConfiguration.CONSUMPTION_MINUS_PRODUCTION: VirtualChannelConfiguration(
Channel.AB,
(Channel.A, Channel.B),
Channel.B,
_a_minus_b,
_cumulative_consumption_minus_production,
),
}
def _get_previous_value(
self, attr_name: str, child_key: Optional[str] = None
) -> Optional[int]:
"""Returns the stored value of the attribute."""
child_key = child_key if child_key else attr_name
if attr_name in self._virtual_channel_stored_values:
return self._virtual_channel_stored_values[attr_name].get(
child_key, self._virtual_channel_stored_values[attr_name][attr_name]
)
else:
return self.get(attr_name)
def _store_current_value(
self, attr_name: str, child_key: Optional[str] = None
) -> None:
"""Stores the current value of the attribute."""
child_key = child_key if child_key else attr_name
value = self.get(attr_name)
if attr_name in self._virtual_channel_stored_values:
self._virtual_channel_stored_values[attr_name][child_key] = value
else:
self._virtual_channel_stored_values[attr_name] = {child_key: value}
def virtual_channel_initial_values(self, attr_name: str, value):
"""Retains the initial attribute value for use in delta calculations."""
if (
attr_name in self._CUMULATIVE_ATTRIBUTES
and ChannelConfiguration.CONSUMPTION_MINUS_PRODUCTION
in self.channel_configuration_type
and attr_name not in self._virtual_channel_stored_values
):
self._store_current_value(attr_name)
def virtual_channel_handler(self, attr_name: str) -> None:
"""Handles updates to a virtual energy meter channel."""
config = self._VIRTUAL_CHANNEL_CONFIGURATION.get(
self.channel_configuration,
VirtualChannelConfiguration(),
)
if (
self.channel not in config.source_channels
or self.channel != config.trigger_channel
and attr_name not in self._CUMULATIVE_ATTRIBUTES
):
return
method = None
if attr_name in self._EXTENSIVE_ATTRIBUTES:
method = config.discrete_method
elif attr_name in self._CUMULATIVE_ATTRIBUTES:
method = config.cumulative_method
if not method:
return
virtual_value = method(self, attr_name)
if virtual_value is None:
return
virtual_cluster = self.get_cluster(config.virtual_channel)
virtual_cluster.update_attribute(attr_name, virtual_value)
class TuyaElectricalMeasurement(
VirtualChannel,
PowerFlowPreempt,
EnergyMeterPowerFlow,
EnergyMeterChannel,
TuyaLocalCluster,
TuyaZBElectricalMeasurement,
):
"""ElectricalMeasurement cluster for Tuya energy meter devices."""
_CONSTANT_ATTRIBUTES: Dict[int, Any] = {
**TuyaZBElectricalMeasurement._CONSTANT_ATTRIBUTES,
TuyaZBElectricalMeasurement.AttributeDefs.ac_frequency_divisor.id: 100,
TuyaZBElectricalMeasurement.AttributeDefs.ac_frequency_multiplier.id: 1,
TuyaZBElectricalMeasurement.AttributeDefs.ac_power_divisor.id: 10,
TuyaZBElectricalMeasurement.AttributeDefs.ac_power_multiplier.id: 1,
TuyaZBElectricalMeasurement.AttributeDefs.ac_voltage_divisor.id: 10,
TuyaZBElectricalMeasurement.AttributeDefs.ac_voltage_multiplier.id: 1,
}
_ATTRIBUTE_MEASUREMENT_TYPES: Dict[str, MeasurementType] = {
"active_power": MeasurementType.Active_measurement_AC
| MeasurementType.Phase_A_measurement,
"active_power_ph_b": MeasurementType.Active_measurement_AC
| MeasurementType.Phase_B_measurement,
"active_power_ph_c": MeasurementType.Active_measurement_AC
| MeasurementType.Phase_C_measurement,
"reactive_power": MeasurementType.Reactive_measurement_AC
| MeasurementType.Phase_A_measurement,
"reactive_power_ph_b": MeasurementType.Reactive_measurement_AC
| MeasurementType.Phase_B_measurement,
"reactive_power_ph_c": MeasurementType.Reactive_measurement_AC
| MeasurementType.Phase_C_measurement,
"apparent_power": MeasurementType.Apparent_measurement_AC
| MeasurementType.Phase_A_measurement,
"apparent_power_ph_b": MeasurementType.Apparent_measurement_AC
| MeasurementType.Phase_B_measurement,
"apparent_power_ph_c": MeasurementType.Apparent_measurement_AC
| MeasurementType.Phase_C_measurement,
}
_EXTENSIVE_ATTRIBUTES: Tuple[str] = (
"active_power",
"apparent_power",
"reactive_power",
"rms_current",
)
_INTENSIVE_ATTRIBUTES: Tuple[str] = ("rms_voltage",)
def calculated_attributes(self, attr_name: str, value) -> None:
"""Calculates attributes that are not reported by the device."""
if (
self.channel == self.virtual_channel
): # Attributes are not calculated for the virtual channel.
return
if attr_name == "apparent_power" and not self.attr_present("active_power"):
self.update_calculated_attribute(
"active_power",
PowerCalculation.active_power_from_apparent_power_power_factor_and_power_flow(
value, self.get("power_factor"), self.power_flow
),
)
if attr_name == "apparent_power" and not self.attr_present("reactive_power"):
self.update_calculated_attribute(
"reactive_power",
PowerCalculation.reactive_power_from_apparent_power_and_power_factor(
value, self.get("power_factor")
),
)
if attr_name == "active_power" and not self.attr_present(
"apparent_power", "rms_current"
):
self.update_calculated_attribute(
"apparent_power",
PowerCalculation.apparent_power_from_active_power_and_power_factor(
value, self.get("power_factor")
),
)
if attr_name == "rms_current" and not self.attr_present("apparent_power"):
self.update_calculated_attribute(
"apparent_power",
PowerCalculation.apparent_power_from_rms_current_and_rms_voltage(
value,
self.get("rms_voltage")
or self.get_cluster(Channel.A).get("rms_voltage"),
self.get("ac_current_divisor", 1),
self.get("ac_current_multiplier", 1),
self.get("ac_voltage_divisor", 1),
self.get("ac_voltage_multiplier", 1),
self.get("ac_power_divisor", 1),
self.get("ac_power_multiplier", 1),
),
)
def update_attribute(self, attr_name: str, value) -> None:
"""Updates the cluster attribute."""
if self.power_flow_preempt_handler(attr_name, value) == PowerFlowPreempt.HOLD:
return
attr_name, value = self.power_flow_handler(attr_name, value)
self.update_measurement_type(attr_name)
self.calculated_attributes(attr_name, value)
self.virtual_channel_initial_values(attr_name, value)
super().update_attribute(attr_name, value)
self.virtual_channel_handler(attr_name)
def update_measurement_type(self, attr_name: str) -> None:
"""Derives the measurement type from reported attributes."""
if attr_name not in self._ATTRIBUTE_MEASUREMENT_TYPES:
return
measurement_type = 0
for measurement, mask in self._ATTRIBUTE_MEASUREMENT_TYPES.items():
if measurement == attr_name or self.get(measurement) is not None:
measurement_type |= mask
super().update_attribute("measurement_type", measurement_type)
class TuyaMetering(
VirtualChannel,
PowerFlowPreempt,
EnergyMeterPowerFlow,
EnergyMeterChannel,
TuyaLocalCluster,
TuyaZBMeteringClusterWithUnit,
):
"""Metering cluster for Tuya energy meter devices."""
_CONSTANT_ATTRIBUTES: Dict[int, Any] = {
**TuyaZBMeteringClusterWithUnit._CONSTANT_ATTRIBUTES,
TuyaZBMeteringClusterWithUnit.AttributeDefs.status.id: 0x00,
TuyaZBMeteringClusterWithUnit.AttributeDefs.multiplier.id: 1,
TuyaZBMeteringClusterWithUnit.AttributeDefs.divisor.id: 10000, # 1 decimal place after conversion from kW to W
TuyaZBMeteringClusterWithUnit.AttributeDefs.summation_formatting.id: Metering.format(
7, 2, True
),
TuyaZBMeteringClusterWithUnit.AttributeDefs.demand_formatting.id: Metering.format(
7, 1, True
),
}
_EXTENSIVE_ATTRIBUTES: Tuple[str] = ("instantaneous_demand",)
_CUMULATIVE_FORWARD_ATTRIBUTES: Tuple[str] = ("current_summ_delivered",)
_CUMULATIVE_REVERSE_ATTRIBUTES: Tuple[str] = ("current_summ_received",)
_INVERSE_ATTRIBUTES: Dict[str, str] = {
"current_summ_delivered": "current_summ_received",
}
def update_attribute(self, attr_name: str, value) -> None:
"""Updates the cluster attribute."""
if self.power_flow_preempt_handler(attr_name, value) == PowerFlowPreempt.HOLD:
return
attr_name, value = self.power_flow_handler(attr_name, value)
self.virtual_channel_initial_values(attr_name, value)
super().update_attribute(attr_name, value)
self.virtual_channel_handler(attr_name)
class TuyaEnergyMeterManufCluster_1CH(
TuyaEnergyMeterManufCluster, configuration_type=ChannelConfiguration_1CH
):
"""Tuya 1 channel energy meter manufacturer cluster."""
TUYA_DP_CURRENT_SUMM_DELIVERED = 101
TUYA_DP_INSTANTANEOUS_DEMAND_UINT = 19
TUYA_DP_RMS_CURRENT = 18
TUYA_DP_RMS_VOLTAGE = 20
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
TUYA_DP_CURRENT_SUMM_DELIVERED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
converter=lambda x: x * 10,
),
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand",
),
TUYA_DP_RMS_CURRENT: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_current",
),
TUYA_DP_RMS_VOLTAGE: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_voltage",
),
}
data_point_handlers = {
TUYA_DP_CURRENT_SUMM_DELIVERED: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT: "_dp_2_attr_update",
TUYA_DP_RMS_VOLTAGE: "_dp_2_attr_update",
}
class TuyaEnergyMeterManufCluster_1CHB(
TuyaEnergyMeterManufCluster,
configuration_type=ChannelConfiguration_1CHB,
):
"""Tuya 1 channel bidirectional energy meter manufacturer cluster."""
TUYA_DP_CURRENT_SUMM_DELIVERED = 1
TUYA_DP_CURRENT_SUMM_RECEIVED = 2
TUYA_DP_INSTANTANEOUS_DEMAND_UINT = 101
TUYA_DP_POWER_FLOW = 102
TUYA_DP_POWER_PHASE = 6
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
TUYA_DP_CURRENT_SUMM_DELIVERED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_RECEIVED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_received",
converter=lambda x: x * 100,
),
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand" + UNSIGNED_POWER_ATTR_SUFFIX,
converter=lambda x: x * 10,
),
TUYA_DP_POWER_FLOW: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"power_flow",
converter=lambda x: PowerFlow(x),
),
TUYA_DP_POWER_PHASE: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
(
"rms_voltage",
"rms_current",
"active_power" + UNSIGNED_POWER_ATTR_SUFFIX,
),
converter=lambda x: TuyaPowerPhase.variant_3(x),
),
}
data_point_handlers = {
TUYA_DP_CURRENT_SUMM_DELIVERED: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: "_dp_2_attr_update",
TUYA_DP_POWER_FLOW: "_dp_2_attr_update",
TUYA_DP_POWER_PHASE: "_dp_2_attr_update",
}
class TuyaEnergyMeterManufCluster_2CHB_MatSeePlus(
TuyaEnergyMeterManufCluster, configuration_type=ChannelConfiguration_2CHB
):
"""MatSee Plus Tuya 2 channel bidirectional energy meter manufacturer cluster."""
_ATTRIBUTE_DEFAULTS: Dict[int, Any] = {
POWER_FLOW_PREEMPT: True,
}
TUYA_DP_AC_FREQUENCY = 111
TUYA_DP_AC_FREQUENCY_COEF = 122
TUYA_DP_CURRENT_SUMM_DELIVERED = 106
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF = 119
TUYA_DP_CURRENT_SUMM_DELIVERED_B = 108
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF_B = 125
TUYA_DP_CURRENT_SUMM_RECEIVED = 107
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF = 127
TUYA_DP_CURRENT_SUMM_RECEIVED_B = 109
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF_B = 128
TUYA_DP_INSTANTANEOUS_DEMAND_UINT = 101
TUYA_DP_INSTANTANEOUS_DEMAND_UINT_B = 105
TUYA_DP_INSTANTANEOUS_DEMAND_COEF = 118
TUYA_DP_INSTANTANEOUS_DEMAND_COEF_B = 124
TUYA_DP_POWER_FACTOR = 110
TUYA_DP_POWER_FACTOR_B = 121
TUYA_DP_POWER_FLOW = 102
TUYA_DP_POWER_FLOW_B = 104
TUYA_DP_UPDATE_PERIOD = 129
TUYA_DP_RMS_CURRENT = 113
TUYA_DP_RMS_CURRENT_COEF = 117
TUYA_DP_RMS_CURRENT_B = 114
TUYA_DP_RMS_CURRENT_COEF_B = 123
TUYA_DP_RMS_VOLTAGE = 112
TUYA_DP_RMS_VOLTAGE_COEF = 116
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
TUYA_DP_AC_FREQUENCY: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"ac_frequency",
),
TUYA_DP_AC_FREQUENCY_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"ac_frequency_coefficient",
),
TUYA_DP_CURRENT_SUMM_DELIVERED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_DELIVERED_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
endpoint_id=2,
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"current_summ_delivered_coefficient",
),
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"current_summ_delivered_coefficient_ch_b",
),
TUYA_DP_CURRENT_SUMM_RECEIVED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_received",
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_RECEIVED_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_received",
endpoint_id=2,
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"current_summ_received_coefficient",
),
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"current_summ_received_coefficient_ch_b",
),
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand" + UNSIGNED_POWER_ATTR_SUFFIX,
),
TUYA_DP_INSTANTANEOUS_DEMAND_UINT_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand" + UNSIGNED_POWER_ATTR_SUFFIX,
endpoint_id=2,
),
TUYA_DP_INSTANTANEOUS_DEMAND_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"instantaneous_demand_coefficient",
),
TUYA_DP_INSTANTANEOUS_DEMAND_COEF_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"instantaneous_demand_coefficient_ch_b",
),
TUYA_DP_POWER_FACTOR: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"power_factor",
),
TUYA_DP_POWER_FACTOR_B: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"power_factor",
endpoint_id=2,
),
TUYA_DP_POWER_FLOW: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"power_flow",
converter=lambda x: PowerFlow(x),
),
TUYA_DP_POWER_FLOW_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"power_flow_ch_b",
converter=lambda x: PowerFlow(x),
),
TUYA_DP_RMS_CURRENT: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_current",
),
TUYA_DP_RMS_CURRENT_B: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_current",
endpoint_id=2,
),
TUYA_DP_RMS_CURRENT_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"rms_current_coefficient",
),
TUYA_DP_RMS_CURRENT_COEF_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"rms_current_coefficient_ch_b",
),
TUYA_DP_RMS_VOLTAGE: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_voltage",
),
TUYA_DP_RMS_VOLTAGE_COEF: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"rms_voltage_coefficient",
),
TUYA_DP_UPDATE_PERIOD: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"update_period",
),
}
data_point_handlers = {
TUYA_DP_AC_FREQUENCY: "_dp_2_attr_update",
TUYA_DP_AC_FREQUENCY_COEF: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED_B: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED_COEF_B: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED_B: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED_COEF_B: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_UINT: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_UINT_B: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_COEF: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_COEF_B: "_dp_2_attr_update",
TUYA_DP_POWER_FACTOR: "_dp_2_attr_update",
TUYA_DP_POWER_FACTOR_B: "_dp_2_attr_update",
TUYA_DP_POWER_FLOW: "_dp_2_attr_update",
TUYA_DP_POWER_FLOW_B: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT_B: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT_COEF: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT_COEF_B: "_dp_2_attr_update",
TUYA_DP_RMS_VOLTAGE: "_dp_2_attr_update",
TUYA_DP_RMS_VOLTAGE_COEF: "_dp_2_attr_update",
TUYA_DP_UPDATE_PERIOD: "_dp_2_attr_update",
}
class TuyaEnergyMeterManufCluster_2CHB_EARU(
TuyaEnergyMeterManufCluster, configuration_type=ChannelConfiguration_2CHB
):
"""EARU Tuya 2 channel bidirectional energy meter manufacturer cluster."""
TUYA_DP_AC_FREQUENCY = 113
TUYA_DP_CURRENT_SUMM_DELIVERED = 101
TUYA_DP_CURRENT_SUMM_DELIVERED_B = 103
TUYA_DP_CURRENT_SUMM_RECEIVED = 102
TUYA_DP_CURRENT_SUMM_RECEIVED_B = 104
TUYA_DP_INSTANTANEOUS_DEMAND = 108
TUYA_DP_INSTANTANEOUS_DEMAND_B = 111
TUYA_DP_POWER_FACTOR = 109
TUYA_DP_POWER_FACTOR_B = 112
TUYA_DP_POWER_FLOW = 114
TUYA_DP_POWER_FLOW_B = 115
TUYA_DP_UPDATE_PERIOD = 116
TUYA_DP_RMS_CURRENT = 107
TUYA_DP_RMS_CURRENT_B = 110
TUYA_DP_RMS_VOLTAGE = 106
dp_to_attribute: Dict[int, DPToAttributeMapping] = {
TUYA_DP_AC_FREQUENCY: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"ac_frequency",
),
TUYA_DP_CURRENT_SUMM_DELIVERED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_DELIVERED_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_delivered",
endpoint_id=2,
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_RECEIVED: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_received",
converter=lambda x: x * 100,
),
TUYA_DP_CURRENT_SUMM_RECEIVED_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"current_summ_received",
endpoint_id=2,
converter=lambda x: x * 100,
),
TUYA_DP_INSTANTANEOUS_DEMAND: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand",
),
TUYA_DP_INSTANTANEOUS_DEMAND_B: DPToAttributeMapping(
TuyaMetering.ep_attribute,
"instantaneous_demand",
endpoint_id=2,
),
TUYA_DP_POWER_FACTOR: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"power_factor",
),
TUYA_DP_POWER_FACTOR_B: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"power_factor",
endpoint_id=2,
),
TUYA_DP_POWER_FLOW: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"power_flow",
converter=lambda x: PowerFlow(x),
),
TUYA_DP_POWER_FLOW_B: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"power_flow_ch_b",
converter=lambda x: PowerFlow(x),
),
TUYA_DP_RMS_CURRENT: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_current",
),
TUYA_DP_RMS_CURRENT_B: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_current",
endpoint_id=2,
),
TUYA_DP_RMS_VOLTAGE: DPToAttributeMapping(
TuyaElectricalMeasurement.ep_attribute,
"rms_voltage",
),
TUYA_DP_UPDATE_PERIOD: DPToAttributeMapping(
TuyaEnergyMeterManufCluster.ep_attribute,
"update_period",
),
}
data_point_handlers = {
TUYA_DP_AC_FREQUENCY: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_DELIVERED_B: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED: "_dp_2_attr_update",
TUYA_DP_CURRENT_SUMM_RECEIVED_B: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND: "_dp_2_attr_update",
TUYA_DP_INSTANTANEOUS_DEMAND_B: "_dp_2_attr_update",
TUYA_DP_POWER_FACTOR: "_dp_2_attr_update",
TUYA_DP_POWER_FACTOR_B: "_dp_2_attr_update",
TUYA_DP_POWER_FLOW: "_dp_2_attr_update",
TUYA_DP_POWER_FLOW_B: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT: "_dp_2_attr_update",
TUYA_DP_RMS_CURRENT_B: "_dp_2_attr_update",
TUYA_DP_RMS_VOLTAGE: "_dp_2_attr_update",
TUYA_DP_UPDATE_PERIOD: "_dp_2_attr_update",
}
class TuyaEnergyMeter_1CH(CustomDevice):
"""Tuya PJ-MGW1203 1 channel energy meter."""
signature = {
MODELS_INFO: [("_TZE204_cjbofhxw", "TS0601")],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaMCUCluster.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
}
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaEnergyMeterManufCluster_1CH,
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
}
}
}
class TuyaEnergyMeter_1CHB(CustomDevice):
"""Tuya bidirectional 1 channel energy meter with Zigbee Green Power."""
signature = {
MODELS_INFO: [("_TZE204_ac0fhfiq", "TS0601")],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaMCUCluster.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
# <SimpleDescriptor endpoint=242 profile=41440 device_type=97
# 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.METER_INTERFACE,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaEnergyMeterManufCluster_1CHB,
TuyaElectricalMeasurement,
TuyaMetering,
],
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],
},
}
}
class TuyaEnergyMeter_2CHB_EARU(CustomDevice):
"""EARU Tuya PC311-Z-TY bidirectional 2 channel energy meter."""
signature = {
MODELS_INFO: [("_TZE200_rks0sgb7", "TS0601")],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters: [0, 4, 5, 61184, 65382]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaMCUCluster.cluster_id,
EARU_MANUFACTURER_CLUSTER_ID,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaEnergyMeterManufCluster_2CHB_EARU,
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
2: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
3: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
}
}
class TuyaEnergyMeter_2CHB_MatSeePlus(CustomDevice):
"""MatSee Plus Tuya PJ-1203A 2 channel bidirectional energy meter with Zigbee Green Power."""
signature = {
MODELS_INFO: [("_TZE204_81yrt3lo", "TS0601")],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaMCUCluster.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
# <SimpleDescriptor endpoint=242 profile=41440 device_type=97
# 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.METER_INTERFACE,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaEnergyMeterManufCluster_2CHB_MatSeePlus,
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
2: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
3: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
242: {
PROFILE_ID: zgp.PROFILE_ID,
DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
}
}
class TuyaEnergyMeter_2CHB_MatSeePlus2(CustomDevice):
"""MatSee Plus Tuya PJ-1203A v2 2 channel bidirectional energy meter with Zigbee Green Power."""
signature = {
MODELS_INFO: [("_TZE284_81yrt3lo", "TS0601")],
ENDPOINTS: {
# <SimpleDescriptor endpoint=1 profile=260 device_type=51
# device_version=1
# input_clusters=[0, 4, 5, 61184]
# output_clusters=[10, 25]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaMCUCluster.cluster_id,
0xed00
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
# <SimpleDescriptor endpoint=242 profile=41440 device_type=97
# 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.METER_INTERFACE,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaEnergyMeterManufCluster_2CHB_MatSeePlus,
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
2: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
3: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.METER_INTERFACE,
INPUT_CLUSTERS: [
TuyaElectricalMeasurement,
TuyaMetering,
],
OUTPUT_CLUSTERS: [],
},
242: {
PROFILE_ID: zgp.PROFILE_ID,
DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
}
}
Here is a converter that work for me, it allows to modify the update frequency to other frequencies :
` const exposes = require('zigbee-herdsman-converters/lib/exposes'); const tuya = require('zigbee-herdsman-converters/lib/tuya'); const e = exposes.presets; const ea = exposes.access;
const definition = { fingerprint: [ {modelID: 'TS0601', manufacturerName: '_TZE284_81yrt3lo'}, ], model: 'TS0601_Energy_Meter', vendor: 'Tuya', description: 'Energy meter with 80A clamp', fromZigbee: [tuya.fz.datapoints], toZigbee: [ { key: ['update_frequency'], convertSet: async (entity, key, value, meta) => { await tuya.sendDataPointValue(entity, 129, value); }, }, ], exposes: [ e.ac_frequency(), // Frequency in Hz exposes.numeric('total_power_A', ea.STATE).withUnit('W').withDescription('Total power A'), exposes.numeric('total_power_B', ea.STATE).withUnit('W').withDescription('Total power B'), exposes.numeric('total_power_AB', ea.STATE).withUnit('W').withDescription('Total power AB'), exposes.numeric('voltage', ea.STATE).withUnit('V').withDescription('Voltage'), exposes.numeric('current_A', ea.STATE).withUnit('A').withDescription('Current A'), exposes.numeric('current_B', ea.STATE).withUnit('A').withDescription('Current B'), exposes.numeric('power_factor_A', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor A'), exposes.numeric('power_factor_B', ea.STATE).withUnit('%').withDescription('Instantaneous measured power factor B'), exposes.numeric('power_direction_A', ea.STATE).withDescription('Power direction A (0: forward, 1: reverse)'), exposes.numeric('power_direction_B', ea.STATE).withDescription('Power direction B (0: forward, 1: reverse)'), exposes.numeric('energy_forward_A', ea.STATE).withUnit('kWh').withDescription('Total energy A forward'), exposes.numeric('energy_forward_B', ea.STATE).withUnit('kWh').withDescription('Total energy B forward'), exposes.numeric('energy_reverse_A', ea.STATE).withUnit('kWh').withDescription('Total energy A reverse'), exposes.numeric('energy_reverse_B', ea.STATE).withUnit('kWh').withDescription('Total energy B reverse'), exposes.numeric('update_frequency', ea.STATE_SET).withUnit('sec').withDescription('Update frequency'), ], meta: { tuyaDatapoints: [ [111, 'ac_frequency', tuya.valueConverter.divideBy100], [101, 'total_power_A', tuya.valueConverter.divideBy10], [105, 'total_power_B', tuya.valueConverter.divideBy10], [115, 'total_power_AB', tuya.valueConverter.divideBy10], [112, 'voltage', tuya.valueConverter.divideBy10], [113, 'current_A', tuya.valueConverter.divideBy1000], [114, 'current_B', tuya.valueConverter.divideBy1000], [110, 'power_factor_A', tuya.valueConverter.divideBy100], [121, 'power_factor_B', tuya.valueConverter.divideBy100], [102, 'power_direction_A', tuya.valueConverter.raw], [104, 'power_direction_B', tuya.valueConverter.raw], [106, 'energy_forward_A', tuya.valueConverter.divideBy100], [108, 'energy_forward_B', tuya.valueConverter.divideBy100], [107, 'energy_reverse_A', tuya.valueConverter.divideBy100], [109, 'energy_reverse_B', tuya.valueConverter.divideBy100], [129, 'update_frequency', tuya.valueConverter.raw], ], }, };
module.exports = definition; `
Hello,
I configure this device automatically TS0601 from vendor TZE284_81yrt3lo (supported from 1feb). I get a state from the device after pairing, but after, no more state seems to be received. Time-stamp of state stay unchanged.
Any idea to help me ?
+1 same problem as @dorianmartinez31 ZHA (SONOFF ZIGBEE 3.0 USB DONGLE)
https://es.aliexpress.com/item/1005007670042835.html?gatewayAdapt=glo2esp MODEL BOX: PJ-1203A TS0601 _TZE284_81yrt3lo
HOME ASSISTANT Core: 2025.3.1 Supervisor: 2025.03.2 Operating System: 14.2 Frontend: 20250306.0
Any idea to help me ?
So what's the working quirk for this _TZE284_81yrt3lo, please? I tried editing the _TZE204_81yrt3lo but no luck. I am using ZHA.
The custom quirk v1 in comment https://github.com/zigpy/zha-device-handlers/issues/3658#issuecomment-2643002772 should work with ZHA.
The v2 quirk I'm working on the linked pull request above is work in progress and won't work on the main releases until some other dependencies have been merged.
Edit:
Are you able to share your device's signature if you're still seeing an issue after trying the above?
You'll need to remove any custom quirks for the device and reload ZHA to ensure it shows the device's original 'unquirked' signature.
Got it working. Thanks a bunch. Will compare to my clamp meter and come back with the result.
La peculiaridad personalizada v1 en el comentario #3658 (comentario) debería funcionar con ZHA.
La peculiaridad v2 en la que estoy trabajando, la solicitud de extracción vinculada anterior, es un trabajo en progreso y no funcionará en las versiones principales hasta que se hayan fusionado algunas otras dependencias.
Editar:
¿Puedes compartir la firma de tu dispositivo si sigues teniendo problemas después de intentar lo anterior?
Tendrás que eliminar cualquier peculiaridad personalizada del dispositivo y volver a cargar ZHA para asegurarte de que muestra la firma "no peculiar" original del dispositivo.
Working!! Although from what I've seen, you'll need to give it time to start seeing the readings... it's not instantaneous, thank you very much.
One small detail: what would I have to change to see the values in kW instead of VA?
Good job!!
Hi, I've bought 2 modules (TS0601 by _TZE284_81yrt3lo) on Aliexpress, last week, and installed today. As said here, ZHA recognized the devices, but without any sensors. I've searched and found this quirk, and also the method to install a custom quirk. It worked instantaneously after the restat of HA, but only for one channel. The second (and the third, I think calculated), shows "unknown" on all readings. How can I obtain the second channel ?
I'm new to HA, and have one of these devices.
I'm using ZHA will these devices get added in an update at some point, if so I'm happy to wait, or will I have to install a quirk?
Thanks
La peculiaridad personalizada v1 en el comentario #3658 (comentario) debería funcionar con ZHA. La peculiaridad v2 en la que estoy trabajando, la solicitud de extracción vinculada anterior, es un trabajo en progreso y no funcionará en las versiones principales hasta que se hayan fusionado algunas otras dependencias. Editar: ¿Puedes compartir la firma de tu dispositivo si sigues teniendo problemas después de intentar lo anterior? Tendrás que eliminar cualquier peculiaridad personalizada del dispositivo y volver a cargar ZHA para asegurarte de que muestra la firma "no peculiar" original del dispositivo.
Working!! Although from what I've seen, you'll need to give it time to start seeing the readings... it's not instantaneous, thank you very much.
One small detail: what would I have to change to see the values in kW instead of VA?
Good job!!
The device works for me. In my case, I only use one of the two available channels.
But... How can I modify the detection interval for status values? Let me explain...
Today, when I tried to automate a NOTICE (via Telegram) for a NO VOLTAGE alert after 1 minute of detection, it didn't work.
IT DOESN'T WORK, probably because the voltage detection interval is predefined for a longer time. I see on the panel that the status change isn't instantaneous, as it is with other types of devices. This is IMPORTANT, especially when you want to automate power FAILURE warnings so you can take action, or to be notified that power is back on.
Where can I change it so that it measures every 1 minute, for example? Although the ideal would be INSTANT, like other common devices.
The custom quirk v1 in comment #3658 (comment) should work with ZHA. Are you able to share your device's signature if you're still seeing an issue after trying the above?
@jeverley thank you for your huge work. In case you need a feedback. I have no issues with wolsty7's quirk v1 based on yours. Not sure about correct current direction values in all cases though (i need to check it later). Other than that, everything works for both TZE284_81yrt3lo and TZE204_81yrt3lo (both channels and the third virtual summarizing).
Thanks for this quick. Actually It have added sensors to _TZE204_81yrt3lo, but they are all stuck as "unknown". Any idea how to fix this?
The v1 quirk from @jeverley adapted by @wolsty7 works for my two _TZE284_81yrt3lo devices and adds basic support!
NOTE: After adding the quirk and restarting HA (maybe restarting ZHA would have been enough), I added my first device, but it only showed sensor entities after another restart. The same applied when adding the second device... Strange enough.
Now, both clamp sensors are steadily reporting, as well as the virtual third channel. But there are some tripled sensors, like Voltage that always show Unknown for the second and third sensor of its type. I guess that some of those sensors should have never been tripled in the first place (see screenshots here for comparison). I am fine with disabling those entities for the moment.
There are only sensor entities, but no configuration entities, but one finds a huge amount of zigbee attributes that can be changed manually through the manage zigbee device dialog. But I could not get my head around most of them, at least in the cluster TuyaEnergyMeterManufCluster_2CHB_MatSeePlus I was able to change the update_period and suppress_reverse_flow successfully, and had minor success with channel_configuration, power_flow_preempt and power_flow (looking up the necessary enumerations in the quirk). But the last three ones did not seem to make any changes in the actual numbers... same solution like above: I disabled all virtual/third channels, since I actually have no need for them (right now).
Probably with v2 of the quirk from @jeverley this will be solved, but there was no progress for a few weeks, so the above information might help others in the meantime.
@LEANOSPA: I assume you misunderstand the sensors. The unit VA is used for apparent power which is not what you probably want. There is another sensor power with the correct unit W.
Despite all that I don't fully understand the reported numbers: all four clamps only report 0W for power very seldom. They mostly show either a small positive or even negative numbers, even if no power is drawn at all. Only when some power is drawn, then the wattage seems to be realistic. I also read that nearby cables can induce wrong readings, what are others doing about this? With suppress_reverse_flow = true I was able to zero out negative numbers, but I am certain it shouldn't detect reverse flow in the first place...
Does anybody know of a good article about those (cheap china) clamp meters? How they are correctly setup/configured/calibrated?
I tried the method, restarting homeassistant, but no clue, now the devices are empty, and on homeassistant I got this
2025-04-07 18:36:22.224 ERROR (SyncWorker_0) [zhaquirks] Unexpected exception importing custom quirk 'ts0601_energy_meter'
Traceback (most recent call last):
File "/usr/local/lib/python3.13/site-packages/zhaquirks/__init__.py", line 479, in setup
spec.loader.exec_module(module)
~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^
File "<frozen importlib._bootstrap_external>", line 1026, in exec_module
File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
File "/config/custom_zha_quirks/ts0601_energy_meter.py", line 598, in <module>
.tuya_dp_attribute(
~~~~~~~~~~~~~~~~~^
dp_id=102,
^^^^^^^^^^
...<2 lines>...
converter=lambda x: TuyaEnergyDirection(x),
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/usr/local/lib/python3.13/site-packages/zhaquirks/tuya/builder/__init__.py", line 579, in tuya_dp_attribute
self.tuya_dp(
~~~~~~~~~~~~^
dp_id=dp_id,
^^^^^^^^^^^^
...<5 lines>...
dp_handler=dp_handler,
^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/usr/local/lib/python3.13/site-packages/zhaquirks/tuya/builder/__init__.py", line 528, in tuya_dp
self.tuya_dp_multi(
~~~~~~~~~~~~~~~~~~^
dp_id,
^^^^^^
...<9 lines>...
dp_handler,
^^^^^^^^^^^
)
^
File "/usr/local/lib/python3.13/site-packages/zhaquirks/tuya/builder/__init__.py", line 552, in tuya_dp_multi
raise ValueError(f"DP {dp_id} is already mapped.")
ValueError: DP 102 is already mapped.
I didn't tried to fix it yet.