"""Tuya valve devices.""" import logging 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, PowerConfiguration, Scenes, Time, ) from zhaquirks.const import ( DEVICE_TYPE, ENDPOINTS, INPUT_CLUSTERS, MODELS_INFO, OUTPUT_CLUSTERS, PROFILE_ID, ) from zhaquirks.tuya import NoManufacturerCluster from zhaquirks.tuya.mcu import ( DPToAttributeMapping, TuyaAttributesCluster, TuyaDPType, TuyaLevelControl, TuyaMCUCluster, TuyaOnOff, TuyaOnOffNM, TuyaPowerConfigurationCluster, ) _LOGGER = logging.getLogger(__name__) class TuyaLevelControlNM(NoManufacturerCluster, TuyaLevelControl): """Tuya LevelControl cluster with NoManufacturerID.""" class TuyaValveManufCluster(TuyaMCUCluster): """Tuya valve manufacturer cluster.""" attributes = TuyaMCUCluster.attributes.copy() attributes.update( { 0xEF03: ("dp_3", t.uint32_t, True), 0xEF65: ("dp_101", t.uint32_t, True), 0xEF66: ("dp_102", t.uint32_t, True), # <-- new cluster attribute with fake IDs (102=0x66).Format= ID: ("name", type, True) } ) dp_to_attribute: Dict[int, DPToAttributeMapping] = { 2: DPToAttributeMapping( TuyaOnOffNM.ep_attribute, "on_off", dp_type=TuyaDPType.BOOL, ), 1: DPToAttributeMapping( TuyaLevelControlNM.ep_attribute, "current_level", dp_type=TuyaDPType.VALUE, converter=lambda x: (x * 255) // 100, dp_converter=lambda x: (x * 100) // 255, ), 3: DPToAttributeMapping( TuyaMCUCluster.ep_attribute, "dp_3", dp_type=TuyaDPType.VALUE, ), 101: DPToAttributeMapping( TuyaMCUCluster.ep_attribute, "dp_101", dp_type=TuyaDPType.VALUE, ), 102: DPToAttributeMapping( # <-- device DP102 TuyaMCUCluster.ep_attribute, # <-- reference to the cluster which has the attribute "dp_102", # <-- attribute "name" dp_type=TuyaDPType.VALUE, # <-- DP Type it is related to the attribute type ), 108: DPToAttributeMapping( TuyaPowerConfigurationCluster.ep_attribute, "battery_percentage_remaining", dp_type=TuyaDPType.VALUE, ), } data_point_handlers = { 2: "_dp_2_attr_update", 1: "_dp_2_attr_update", 3: "_dp_2_attr_update", 101: "_dp_2_attr_update", 102: "_dp_2_attr_update", 108: "_dp_2_attr_update", } async def write_attributes(self, attributes, manufacturer=None): """Overwrite to force manufacturer code.""" return await super().write_attributes( attributes, manufacturer=foundation.ZCLHeader.NO_MANUFACTURER_ID ) class TuyaValve(CustomDevice): """Tuya valve device.""" signature = { """MODELS_INFO: [("_TZE200_sh1btabb", "TS0601")],""" MODELS_INFO: [("_TZE200_sh1btabb")], ENDPOINTS: { # 1: { PROFILE_ID: zha.PROFILE_ID, DEVICE_TYPE: zha.DeviceType.SMART_PLUG, INPUT_CLUSTERS: [ Basic.cluster_id, Groups.cluster_id, Scenes.cluster_id, TuyaValveManufCluster.cluster_id, ], OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], } }, } replacement = { ENDPOINTS: { 1: { DEVICE_TYPE: zha.DeviceType.ON_OFF_LIGHT, INPUT_CLUSTERS: [ Basic.cluster_id, Groups.cluster_id, Scenes.cluster_id, TuyaValveManufCluster, TuyaOnOffNM, TuyaLevelControlNM, TuyaPowerConfigurationCluster, ], OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], } } }