From d489d4f6b1ac329b83a11f8560f59a76a48522a9 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Wed, 17 Apr 2024 09:12:35 +1000 Subject: [PATCH 01/14] Add Tuya cover `_TZE200_eevqq1uv` Similar to other Tuya window cover devices, but reports multiple tuya data points in set_data_response, so this implementation is based on the newer TuyaMCUCluster rather than TuyaManufCluster. Sold as Zemismart ZM25R3 roller blind motor. --- tests/test_tuya_cover.py | 189 ++++++++++++++++++++++++++++++++- zhaquirks/tuya/__init__.py | 31 ++++-- zhaquirks/tuya/mcu/__init__.py | 169 ++++++++++++++++++++++++++++- zhaquirks/tuya/ts0601_cover.py | 58 ++++++++++ zhaquirks/tuya/ts0601_siren.py | 7 +- 5 files changed, 441 insertions(+), 13 deletions(-) diff --git a/tests/test_tuya_cover.py b/tests/test_tuya_cover.py index 8c28a5c3cf..2e59ee40a8 100644 --- a/tests/test_tuya_cover.py +++ b/tests/test_tuya_cover.py @@ -1,6 +1,168 @@ """Test units for Tuya covers.""" +from unittest import mock + +import pytest +from zigpy.zcl import foundation + +from tests.common import ClusterListener, wait_for_zigpy_tasks +import zhaquirks +from zhaquirks.tuya.ts0601_cover import ( + TuyaCover0601MultipleDataPoints, + TuyaMoesCover0601, +) + +zhaquirks.setup() + +@pytest.mark.parametrize("command, expected_frame", + ( + # Window cover open, close, stop commands are 0, 1 & 2 respectively + # DP 1 should be 0 for open, 2 for close, 1 to stop + ( + 0x00, + b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x00', + ), + ( + 0x01, + b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x02', + ), + ( + 0x02, + b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x01', + ), + ), +) +async def test_cover_move_commands(zigpy_device_from_quirk, command, expected_frame): + """Test executing cluster move commands for tuya cover (that supports multiple data points).""" + + device = zigpy_device_from_quirk( + zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints + ) + tuya_cluster = device.endpoints[1].tuya_manufacturer + tuya_listener = ClusterListener(tuya_cluster) + cover_cluster = device.endpoints[1].window_covering + + assert len(tuya_listener.cluster_commands) == 0 + assert len(tuya_listener.attribute_updates) == 0 + + with mock.patch.object( + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + ) as m1: + rsp = await cover_cluster.command(command) + + await wait_for_zigpy_tasks() + m1.assert_called_with( + 0xef00, + 1, + expected_frame, + expect_reply=True, + command_id=0, + ) + assert rsp.status == foundation.Status.SUCCESS + +async def test_cover_set_position_command(zigpy_device_from_quirk): + """Test executing cluster command to set the position for tuya cover (that supports multiple data points).""" + + device = zigpy_device_from_quirk( + zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints + ) + tuya_cluster = device.endpoints[1].tuya_manufacturer + tuya_listener = ClusterListener(tuya_cluster) + cover_cluster = device.endpoints[1].window_covering + + assert len(tuya_listener.cluster_commands) == 0 + assert len(tuya_listener.attribute_updates) == 0 + + with mock.patch.object( + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + ) as m1: + rsp = await cover_cluster.command(5, 20) + + await wait_for_zigpy_tasks() + m1.assert_called_with( + 0xef00, + 1, + b'\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x50', + expect_reply=True, + command_id=0, + ) + assert rsp.status == foundation.Status.SUCCESS + +@pytest.mark.parametrize( + "frame, cluster, attributes", + ( + ( # TuyaDatapointData(dp=3, data=TuyaData(dp_type=, function=0, raw=b'\x00\x00\x00\x14', *payload=20)) + b"\x09\x00\x02\x00\x00\x03\x02\x00\x04\x00\x00\x00\x14", + "window_covering", + # current_position_lift_percentage (blind reports % closed, cluster attribute expects % open) + {0x0008: 80}, + ), + ( # TuyaDatapointData(dp=13, data=TuyaData(dp_type=, function=0, raw=b'\x00\x00\x00\\', *payload=92)) + b"\x09\x00\x02\x00\x00\x0d\x02\x00\x04\x00\x00\x00\x5c", + "power", + # battery_percentage_remaining (attribute expects 2x real percentage) + {0x0021: 184}, + ), + ), +) +async def test_cover_report_values(zigpy_device_from_quirk, frame, cluster, attributes): + """Test receiving single attributes from tuya cover (that supports multiple data points).""" + + cover_dev = zigpy_device_from_quirk( + zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints + ) + tuya_cluster = cover_dev.endpoints[1].tuya_manufacturer + target_cluster = getattr(cover_dev.endpoints[1], cluster) + tuya_listener = ClusterListener(target_cluster) + + assert len(tuya_listener.cluster_commands) == 0 + assert len(tuya_listener.attribute_updates) == 0 + + hdr, args = tuya_cluster.deserialize(frame) + tuya_cluster.handle_message(hdr, args) + + assert tuya_listener.attribute_updates == list(attributes.items()) + + +async def test_cover_report_multiple_values(zigpy_device_from_quirk): + """Test receiving multiple attributes from tuya cover (that supports multiple data points).""" + + # A real packet with multiple Tuya data points 1,7,3,5 & 13 + frame = b"\x09\x00\x02\x00\x00\x01\x04\x00\x01\x01\x07\x04\x00\x01\x01\x03\x02\x00\x04\x00\x00\x00\x14\x05\x04\x00\x01\x01\x0d\x02\x00\x04\x00\x00\x00\x5c" + blind_open_pct_id = 0x08 + blind_open_pct_expected = ( + 80 # (blind reports % closed, cluster attribute expects % open) + ) + battery_pct_id = ( + 0x21 # PowerConfiguration.AttributeDefs.battery_percentage_remaining.id + ) + battery_pct_expected = 92 * 2 # (attribute expects 2x real percentage) + + device = zigpy_device_from_quirk( + zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints + ) + tuya_cluster = device.endpoints[1].tuya_manufacturer + cover_cluster = device.endpoints[1].window_covering + cover_listener = ClusterListener(cover_cluster) + power_cluster = device.endpoints[1].power + power_listener = ClusterListener(power_cluster) + + assert len(cover_listener.cluster_commands) == 0 + assert len(cover_listener.attribute_updates) == 0 + assert len(power_listener.cluster_commands) == 0 + assert len(power_listener.attribute_updates) == 0 + + hdr, args = tuya_cluster.deserialize(frame) + tuya_cluster.handle_message(hdr, args) + + # blind cover will have an attribute update for motor direction, I only care that it + # includes the new position + assert len(cover_listener.attribute_updates) < 3 + assert ( + blind_open_pct_id, + blind_open_pct_expected, + ) in cover_listener.attribute_updates + assert power_listener.attribute_updates == [(battery_pct_id, battery_pct_expected)] -from zhaquirks.tuya.ts0601_cover import TuyaMoesCover0601 def test_ts601_moes_signature(assert_signature_matches_quirk): @@ -20,3 +182,28 @@ def test_ts601_moes_signature(assert_signature_matches_quirk): "class": "zigpy.device.Device", } assert_signature_matches_quirk(TuyaMoesCover0601, signature) + + +def test_cover_signature_ZemiSmart_ZM25R3_TuyaCover0601MultipleDataPoints( + assert_signature_matches_quirk, +): + """Test Zemismart ZM25R3 matches TuyaCover0601MultipleDataPoints.""" + + signature = { + "node_descriptor": "NodeDescriptor(logical_type=, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=, mac_capability_flags=, manufacturer_code=4098, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=, *allocate_address=True, *is_alternate_pan_coordinator=False, *is_coordinator=False, *is_end_device=True, *is_full_function_device=False, *is_mains_powered=False, *is_receiver_on_when_idle=False, *is_router=False, *is_security_capable=False)", + "endpoints": { + "1": { + "profile_id": 260, + "device_type": "0x0051", + "in_clusters": ["0x0000", "0x0004", "0x0005", "0xef00"], + "out_clusters": ["0x000a", "0x0019"], + } + }, + "manufacturer": "_TZE200_eevqq1uv", + "model": "TS0601", + "class": "zigpy.device.Device", + } + + assert_signature_matches_quirk( + TuyaCover0601MultipleDataPoints, signature + ) diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index 347a6c711d..59c1681af3 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -103,7 +103,12 @@ COVER_EVENT = "cover_event" ATTR_COVER_POSITION = 0x0008 ATTR_COVER_DIRECTION = 0x8001 +ATTR_COVER_DIRECTION_NAME = "motor_direction" ATTR_COVER_INVERTED = 0x8002 +ATTR_COVER_INVERTED_NAME = "cover_inverted" +ATTR_COVER_LIFTPERCENT_NAME = "current_position_lift_percentage" +ATTR_COVER_LIFTPERCENT_CONTROL = 0x8003 +ATTR_COVER_LIFTPERCENT_CONTROL_NAME = "current_position_lift_percentage_control" # --------------------------------------------------------- # TUYA Switch Custom Values @@ -1127,7 +1132,11 @@ class TuyaZB1888Cluster(CustomCluster): # Tuya Window Cover Implementation class TuyaManufacturerWindowCover(TuyaManufCluster): - """Manufacturer Specific Cluster for cover device.""" + """Manufacturer Specific Cluster for cover device. + + Consider using TuyaNewManufClusterForWindowCover instead, it's based on a newer TuyaMCUCluster + base class. + """ def handle_cluster_request( self, @@ -1182,7 +1191,7 @@ def handle_cluster_request( tuya_payload.data[1], # Check this ) elif hdr.command_id == TUYA_SET_TIME: - """Time event call super""" + # Time event call super super().handle_cluster_request(hdr, args, dst_addressing=dst_addressing) else: _LOGGER.debug( @@ -1195,12 +1204,16 @@ def handle_cluster_request( class TuyaWindowCoverControl(LocalDataCluster, WindowCovering): - """Manufacturer Specific Cluster of Device cover.""" + """Window Covering cluster for cover device. + + Consider using TuyaNewWindowCoverControl instead, it's compatible with the newer + TuyaMCUCluster implementation method. + """ # Add additional attributes for direction attributes = WindowCovering.attributes.copy() - attributes.update({ATTR_COVER_DIRECTION: ("motor_direction", t.Bool)}) - attributes.update({ATTR_COVER_INVERTED: ("cover_inverted", t.Bool)}) + attributes.update({ATTR_COVER_DIRECTION: (ATTR_COVER_DIRECTION_NAME, t.enum8)}) + attributes.update({ATTR_COVER_INVERTED: (ATTR_COVER_INVERTED_NAME, t.Bool)}) def __init__(self, *args, **kwargs): """Initialize instance.""" @@ -1264,7 +1277,7 @@ def command( tuya_payload.tsn = tsn if tsn else 0 tuya_payload.command_id = TUYA_DP_TYPE_VALUE + TUYA_DP_ID_PERCENT_CONTROL tuya_payload.function = 0 - """Check direction and correct value""" + # Check direction and correct value invert_attr = self._attr_cache.get(ATTR_COVER_INVERTED) == 1 invert = ( not invert_attr @@ -1314,7 +1327,11 @@ class TuyaWindowCover(CustomDevice): """Tuya Window cover device.""" # For most tuya devices 0 = Up/Open, 1 = Stop, 2 = Down/Close - tuya_cover_command = {0x0000: 0x0000, 0x0001: 0x0002, 0x0002: 0x0001} + tuya_cover_command = { + WINDOW_COVER_COMMAND_UPOPEN: 0x0000, + WINDOW_COVER_COMMAND_DOWNCLOSE: 0x0002, + WINDOW_COVER_COMMAND_STOP: 0x0001 + } # For all covers which need their position inverted by default # Default (False) is 100 = open, 0 = closed; When True use 0 = open, 100 = closed instead diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index e9d73d3333..d66745c820 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -3,20 +3,33 @@ from collections.abc import Callable import dataclasses import datetime +import logging from typing import Any, Optional, Union import zigpy.types as t from zigpy.zcl import foundation -from zigpy.zcl.clusters.general import LevelControl, OnOff +from zigpy.zcl.clusters.closures import WindowCovering +from zigpy.zcl.clusters.general import LevelControl, OnOff, PowerConfiguration from zhaquirks import Bus, DoublingPowerConfigurationCluster # add EnchantedDevice import for custom quirks backwards compatibility from zhaquirks.tuya import ( + ATTR_COVER_DIRECTION, + ATTR_COVER_DIRECTION_NAME, + ATTR_COVER_INVERTED, + ATTR_COVER_INVERTED_NAME, + ATTR_COVER_LIFTPERCENT_CONTROL, + ATTR_COVER_LIFTPERCENT_CONTROL_NAME, + ATTR_COVER_LIFTPERCENT_NAME, TUYA_MCU_COMMAND, TUYA_MCU_VERSION_RSP, TUYA_SET_DATA, TUYA_SET_TIME, + WINDOW_COVER_COMMAND_DOWNCLOSE, + WINDOW_COVER_COMMAND_LIFTPERCENT, + WINDOW_COVER_COMMAND_STOP, + WINDOW_COVER_COMMAND_UPOPEN, EnchantedDevice, # noqa: F401 NoManufacturerCluster, PowerOnState, @@ -33,6 +46,7 @@ # manufacturer commands TUYA_MCU_CONNECTION_STATUS = 0x25 +_LOGGER = logging.getLogger(__name__) @dataclasses.dataclass class DPToAttributeMapping: @@ -369,6 +383,7 @@ async def command( manufacturer: Optional[Union[int, t.uint16_t]] = None, expect_reply: bool = True, tsn: Optional[Union[int, t.uint8_t]] = None, + **_kwargs: Any ): """Override the default Cluster command.""" @@ -530,7 +545,7 @@ class MoesSwitchManufCluster(TuyaOnOffManufCluster): 14: DPToAttributeMapping( TuyaMCUCluster.ep_attribute, "power_on_state", - converter=lambda x: PowerOnState(x), + converter=PowerOnState, ) } ) @@ -539,7 +554,7 @@ class MoesSwitchManufCluster(TuyaOnOffManufCluster): 15: DPToAttributeMapping( TuyaMCUCluster.ep_attribute, "backlight_mode", - converter=lambda x: MoesBacklight(x), + converter=MoesBacklight, ), } ) @@ -549,6 +564,154 @@ class MoesSwitchManufCluster(TuyaOnOffManufCluster): data_point_handlers.update({15: "_dp_2_attr_update"}) +class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): + """Manufacturer Specific Cluster for cover device (based on new TuyaMCUCluster). + + I.e. Uses newer TuyaMCUCluster mechanism for translations between cluster attributes and + Tuya data points. + """ + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + + self.dp_to_attribute: dict[int, DPToAttributeMapping] = { + 1: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_DIRECTION_NAME, + # make sure this is sent as a enum, regardless that tuya_cover_command probably + # contains ints. + None, + t.enum8 + ), + 2: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_LIFTPERCENT_CONTROL_NAME, + self.convert_lift_percent, + self.convert_lift_percent + ), + 3: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_LIFTPERCENT_NAME, + self.convert_lift_percent, + self.convert_lift_percent + ), + 13: DPToAttributeMapping( + PowerConfiguration.ep_attribute, + PowerConfiguration.AttributeDefs.battery_percentage_remaining.name + # Tuya report real percent, zigbee expects value*2, but + # TuyaPowerConfigurationCluster will convert it + ), + } + + data_point_handlers = { + 1: "_dp_2_attr_update", + 2: "_dp_2_attr_update", + 3: "_dp_2_attr_update", + # Define DPs 5 & 7 to avoid warning logs, but I'm not sure what they are so just ignore + # updates from them + 5: "ignore_update", + 7: "ignore_update", + 13: "_dp_2_attr_update", + } + + def convert_lift_percent(self, input_value: int): + """Convert/invert lift percent when needed. + + HA shows % open. The zigbee cluster value is called 'lift_percent' but seems to need to + be % closed. This logic follows the convention of other Tuya covers, inverting the value + by default and reversing that invert if tuya_cover_inverted_by_default or the cluster + invert attribute is set. (This seems strange to me, but it's better to be consistent.) + """ + + invert_attr = self.endpoint.window_covering._attr_cache.get(ATTR_COVER_INVERTED) == 1 + invert = ( + not invert_attr + if self.endpoint.device.tuya_cover_inverted_by_default + else invert_attr + ) + return input_value if invert else 100 - input_value + + def ignore_update(self, _datapoint: TuyaDatapointData) -> None: + """Process (and ignore) some data point updates.""" + return None + + +class TuyaNewWindowCoverControl(TuyaLocalCluster, WindowCovering): + """Tuya Window Cover Cluster, based on new TuyaNewManufClusterForWindowCover. + + Derive from TuyaLocalCluster to disable attribute writes, in a way that's compatible with + TuyaNewManufClusterForWindowCover (not TuyaManufacturerWindowCover.) + """ + + attributes = WindowCovering.attributes.copy() + # cover direction and lift percent control attributes are not very useful to HA, but needed to + # allow TuyaMCUCluster to map to the data point. + attributes.update({ATTR_COVER_DIRECTION: (ATTR_COVER_DIRECTION_NAME, t.enum8)}) + attributes.update({ATTR_COVER_INVERTED: (ATTR_COVER_INVERTED_NAME, t.Bool)}) + attributes.update({ATTR_COVER_LIFTPERCENT_CONTROL: (ATTR_COVER_LIFTPERCENT_CONTROL_NAME, t.uint16_t)}) + + async def command( + self, + command_id: Union[foundation.GeneralCommand, int, t.uint8_t], + *args, + manufacturer: Optional[Union[int, t.uint16_t]] = None, + expect_reply: bool = True, + tsn: Optional[Union[int, t.uint8_t]] = None, + **_kwargs: Any + ): + """Override the default Cluster command.""" + + _LOGGER.debug( + "Sending Tuya Cluster Command... Cluster Command is %x, Arguments are %s", + command_id, + args, + ) + + # Open Close or Stop commands + if command_id in ( + WINDOW_COVER_COMMAND_UPOPEN, + WINDOW_COVER_COMMAND_DOWNCLOSE, + WINDOW_COVER_COMMAND_STOP, + ): + cluster_data = TuyaClusterData( + endpoint_id=self.endpoint.endpoint_id, + cluster_name=self.ep_attribute, + cluster_attr=ATTR_COVER_DIRECTION_NAME, + # Map from zigbee command to tuya DP value + attr_value=self.endpoint.device.tuya_cover_command[command_id], + expect_reply=expect_reply, + manufacturer=manufacturer, + ) + self.endpoint.device.command_bus.listener_event( + TUYA_MCU_COMMAND, + cluster_data, + ) + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + elif command_id == WINDOW_COVER_COMMAND_LIFTPERCENT: + cluster_data = TuyaClusterData( + endpoint_id=self.endpoint.endpoint_id, + cluster_name=self.ep_attribute, + cluster_attr=ATTR_COVER_LIFTPERCENT_CONTROL_NAME, + attr_value=args[0], + expect_reply=expect_reply, + manufacturer=manufacturer, + ) + self.endpoint.device.command_bus.listener_event( + TUYA_MCU_COMMAND, + cluster_data, + ) + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + _LOGGER.warning("Unsupported command_id: %s", command_id) + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND) + + class TuyaLevelControl(LevelControl, TuyaLocalCluster): """Tuya MCU Level cluster for dimmable device.""" diff --git a/zhaquirks/tuya/ts0601_cover.py b/zhaquirks/tuya/ts0601_cover.py index 57f76a43f4..d1fb298834 100644 --- a/zhaquirks/tuya/ts0601_cover.py +++ b/zhaquirks/tuya/ts0601_cover.py @@ -17,6 +17,11 @@ TuyaWindowCover, TuyaWindowCoverControl, ) +from zhaquirks.tuya.mcu import ( + TuyaNewManufClusterForWindowCover, + TuyaNewWindowCoverControl, + TuyaPowerConfigurationCluster, +) class TuyaZemismartSmartCover0601(TuyaWindowCover): @@ -619,3 +624,56 @@ class TuyaCloneCover0601(TuyaWindowCover): } } } + + +class TuyaCover0601MultipleDataPoints(TuyaWindowCover): + """Tuya window cover device. + + This variant supports: + - multiple data points included in tuya set_data_response. + - non-inverted control inputs, + - battery percentage remaining + + Most/all the quirks above are based on TuyaManufacturerWindowCover that only decodes + ONE attribute from the Tuya set_data_response packet. This quirk is based on + TuyaNewManufClusterForWindowCover which can handle multiple updates in one zigby frame. + """ + + signature = { + MODELS_INFO: [ + ("_TZE200_eevqq1uv", "TS0601"), # Zemismart ZM25R3 roller blind motor + ], + # SimpleDescriptor(endpoint=1, profile=260, device_type=81, device_version=1, + # input_clusters=[0, 4, 5, 61184], + # output_clusters=[25, 10]) + ENDPOINTS: { + 1: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.SMART_PLUG, + INPUT_CLUSTERS: [ + Basic.cluster_id, + Groups.cluster_id, + Scenes.cluster_id, + TuyaNewManufClusterForWindowCover.cluster_id, + ], + OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], + } + }, + } + + replacement = { + ENDPOINTS: { + 1: { + DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE, + INPUT_CLUSTERS: [ + Basic.cluster_id, + Groups.cluster_id, + Scenes.cluster_id, + TuyaNewWindowCoverControl, + TuyaPowerConfigurationCluster, + TuyaNewManufClusterForWindowCover, + ], + OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], + } + } + } diff --git a/zhaquirks/tuya/ts0601_siren.py b/zhaquirks/tuya/ts0601_siren.py index 4ce637fde9..501813f88f 100644 --- a/zhaquirks/tuya/ts0601_siren.py +++ b/zhaquirks/tuya/ts0601_siren.py @@ -27,9 +27,12 @@ OUTPUT_CLUSTERS, PROFILE_ID, ) -from zhaquirks.tuya import TuyaManufCluster, TuyaManufClusterAttributes -from zhaquirks.tuya.mcu import ( +from zhaquirks.tuya import ( TUYA_MCU_COMMAND, + TuyaManufCluster, + TuyaManufClusterAttributes, +) +from zhaquirks.tuya.mcu import ( DPToAttributeMapping, TuyaAttributesCluster, TuyaClusterData, From d54593972274e2b530b7523ffbcad2099c4ca525 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Thu, 18 Apr 2024 16:45:58 +1000 Subject: [PATCH 02/14] Add unit test for tuya cover unsupported command to increase test coverage --- tests/test_tuya_cover.py | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/tests/test_tuya_cover.py b/tests/test_tuya_cover.py index 2e59ee40a8..a6f3434c1c 100644 --- a/tests/test_tuya_cover.py +++ b/tests/test_tuya_cover.py @@ -16,7 +16,8 @@ @pytest.mark.parametrize("command, expected_frame", ( # Window cover open, close, stop commands are 0, 1 & 2 respectively - # DP 1 should be 0 for open, 2 for close, 1 to stop + # Expected frame is a set_value command for data point 1, with an enum value of 0 for open, + # 2 for close, 1 to stop ( 0x00, b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x00', @@ -75,18 +76,43 @@ async def test_cover_set_position_command(zigpy_device_from_quirk): with mock.patch.object( tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS ) as m1: + # command #5 is go_to_lift_percentage (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id) rsp = await cover_cluster.command(5, 20) await wait_for_zigpy_tasks() m1.assert_called_with( 0xef00, 1, + # expect a frame to set data point id 4 to a int value of 80 (100-20%) b'\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x50', expect_reply=True, command_id=0, ) assert rsp.status == foundation.Status.SUCCESS +async def test_cover_unknown_command(zigpy_device_from_quirk): + """Test executing unexpected cluster command returns an unsupported status.""" + + device = zigpy_device_from_quirk( + zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints + ) + tuya_cluster = device.endpoints[1].tuya_manufacturer + tuya_listener = ClusterListener(tuya_cluster) + cover_cluster = device.endpoints[1].window_covering + + assert len(tuya_listener.cluster_commands) == 0 + assert len(tuya_listener.attribute_updates) == 0 + + with mock.patch.object( + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + ) as m1: + # send a command, use the max (uint8) value as an example unsupported command id + rsp = await cover_cluster.command(0xFF) + + await wait_for_zigpy_tasks() + m1.assert_not_called() + assert rsp.status == foundation.Status.UNSUP_CLUSTER_COMMAND + @pytest.mark.parametrize( "frame, cluster, attributes", ( From cdae82c2869fb1a91dbf2234842eee7fe13aa831 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Sat, 27 Apr 2024 10:26:32 +1000 Subject: [PATCH 03/14] Change to quirk v2, add support for blind direction and limit settings --- tests/test_tuya_cover.py | 274 +++++++++++++++++++++---------- zhaquirks/tuya/__init__.py | 74 ++++++--- zhaquirks/tuya/mcu/__init__.py | 288 +++++++++++++++++++++++++-------- zhaquirks/tuya/ts0601_cover.py | 133 +++++++++------ 4 files changed, 539 insertions(+), 230 deletions(-) diff --git a/tests/test_tuya_cover.py b/tests/test_tuya_cover.py index a6f3434c1c..d894d9f038 100644 --- a/tests/test_tuya_cover.py +++ b/tests/test_tuya_cover.py @@ -1,4 +1,5 @@ """Test units for Tuya covers.""" + from unittest import mock import pytest @@ -6,38 +7,87 @@ from tests.common import ClusterListener, wait_for_zigpy_tasks import zhaquirks -from zhaquirks.tuya.ts0601_cover import ( - TuyaCover0601MultipleDataPoints, - TuyaMoesCover0601, +from zhaquirks.tuya import ATTR_COVER_DIRECTION_SETTING, ATTR_COVER_MAIN_CONTROL +from zhaquirks.tuya.mcu import ( + CoverCommandStepDirection, + CoverMotorStatus, + CoverSettingLimitOperation, + CoverSettingMotorDirection, ) +from zhaquirks.tuya.ts0601_cover import TuyaMoesCover0601 zhaquirks.setup() -@pytest.mark.parametrize("command, expected_frame", + +@pytest.mark.parametrize( + "command, args, kwargs, expected_frame", ( # Window cover open, close, stop commands are 0, 1 & 2 respectively # Expected frame is a set_value command for data point 1, with an enum value of 0 for open, # 2 for close, 1 to stop + (0x00, [], {}, b"\x01\x01\x00\x00\x01\x01\x04\x00\x01\x00"), + (0x01, [], {}, b"\x01\x01\x00\x00\x01\x01\x04\x00\x01\x02"), + (0x02, [], {}, b"\x01\x01\x00\x00\x01\x01\x04\x00\x01\x01"), + # command #5 is go_to_lift_percentage (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id) + # expect a frame to set data point id 4 to a int value of 80 (100-20%) + (0x05, [20], {}, b"\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x50"), + # small step open + ( + 0xF0, + [], + {"direction": CoverCommandStepDirection.Open}, + b"\x01\x01\x00\x00\x01\x14\x04\x00\x01\x00", + ), + # small step close + ( + 0xF0, + [], + {"direction": CoverCommandStepDirection.Close}, + b"\x01\x01\x00\x00\x01\x14\x04\x00\x01\x01", + ), + # open limit set ( - 0x00, - b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x00', + 0xF1, + [], + {"operation": CoverSettingLimitOperation.SetOpen}, + b"\x01\x01\x00\x00\x01\x10\x04\x00\x01\x00", ), + # close limit set ( - 0x01, - b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x02', + 0xF1, + [], + {"operation": CoverSettingLimitOperation.SetClose}, + b"\x01\x01\x00\x00\x01\x10\x04\x00\x01\x01", ), + # clear open limit clear ( - 0x02, - b'\x01\x01\x00\x00\x01\x01\x04\x00\x01\x01', + 0xF1, + [], + {"operation": CoverSettingLimitOperation.ClearOpen}, + b"\x01\x01\x00\x00\x01\x10\x04\x00\x01\x02", + ), + # clear close limit clear + ( + 0xF1, + [], + {"operation": CoverSettingLimitOperation.ClearClose}, + b"\x01\x01\x00\x00\x01\x10\x04\x00\x01\x03", + ), + # clear both limits + ( + 0xF1, + [], + {"operation": CoverSettingLimitOperation.ClearBoth}, + b"\x01\x01\x00\x00\x01\x10\x04\x00\x01\x04", ), ), ) -async def test_cover_move_commands(zigpy_device_from_quirk, command, expected_frame): +async def test_cover_move_commands( + zigpy_device_from_v2_quirk, command, args, kwargs, expected_frame +): """Test executing cluster move commands for tuya cover (that supports multiple data points).""" - device = zigpy_device_from_quirk( - zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints - ) + device = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") tuya_cluster = device.endpoints[1].tuya_manufacturer tuya_listener = ClusterListener(tuya_cluster) cover_cluster = device.endpoints[1].window_covering @@ -48,11 +98,11 @@ async def test_cover_move_commands(zigpy_device_from_quirk, command, expected_fr with mock.patch.object( tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS ) as m1: - rsp = await cover_cluster.command(command) + rsp = await cover_cluster.command(command, *args, **kwargs) await wait_for_zigpy_tasks() m1.assert_called_with( - 0xef00, + 0xEF00, 1, expected_frame, expect_reply=True, @@ -60,42 +110,11 @@ async def test_cover_move_commands(zigpy_device_from_quirk, command, expected_fr ) assert rsp.status == foundation.Status.SUCCESS -async def test_cover_set_position_command(zigpy_device_from_quirk): - """Test executing cluster command to set the position for tuya cover (that supports multiple data points).""" - - device = zigpy_device_from_quirk( - zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints - ) - tuya_cluster = device.endpoints[1].tuya_manufacturer - tuya_listener = ClusterListener(tuya_cluster) - cover_cluster = device.endpoints[1].window_covering - assert len(tuya_listener.cluster_commands) == 0 - assert len(tuya_listener.attribute_updates) == 0 - - with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS - ) as m1: - # command #5 is go_to_lift_percentage (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id) - rsp = await cover_cluster.command(5, 20) - - await wait_for_zigpy_tasks() - m1.assert_called_with( - 0xef00, - 1, - # expect a frame to set data point id 4 to a int value of 80 (100-20%) - b'\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x50', - expect_reply=True, - command_id=0, - ) - assert rsp.status == foundation.Status.SUCCESS - -async def test_cover_unknown_command(zigpy_device_from_quirk): +async def test_cover_unknown_command(zigpy_device_from_v2_quirk): """Test executing unexpected cluster command returns an unsupported status.""" - device = zigpy_device_from_quirk( - zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints - ) + device = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") tuya_cluster = device.endpoints[1].tuya_manufacturer tuya_listener = ClusterListener(tuya_cluster) cover_cluster = device.endpoints[1].window_covering @@ -113,6 +132,7 @@ async def test_cover_unknown_command(zigpy_device_from_quirk): m1.assert_not_called() assert rsp.status == foundation.Status.UNSUP_CLUSTER_COMMAND + @pytest.mark.parametrize( "frame, cluster, attributes", ( @@ -130,12 +150,12 @@ async def test_cover_unknown_command(zigpy_device_from_quirk): ), ), ) -async def test_cover_report_values(zigpy_device_from_quirk, frame, cluster, attributes): +async def test_cover_report_values( + zigpy_device_from_v2_quirk, frame, cluster, attributes +): """Test receiving single attributes from tuya cover (that supports multiple data points).""" - cover_dev = zigpy_device_from_quirk( - zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints - ) + cover_dev = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") tuya_cluster = cover_dev.endpoints[1].tuya_manufacturer target_cluster = getattr(cover_dev.endpoints[1], cluster) tuya_listener = ClusterListener(target_cluster) @@ -149,11 +169,14 @@ async def test_cover_report_values(zigpy_device_from_quirk, frame, cluster, attr assert tuya_listener.attribute_updates == list(attributes.items()) -async def test_cover_report_multiple_values(zigpy_device_from_quirk): +async def test_cover_report_multiple_values(zigpy_device_from_v2_quirk): """Test receiving multiple attributes from tuya cover (that supports multiple data points).""" - # A real packet with multiple Tuya data points 1,7,3,5 & 13 + # A real packet with multiple Tuya data points 1,7,3,5 & 13 (motor status, unknown, position, + # direction, battery) frame = b"\x09\x00\x02\x00\x00\x01\x04\x00\x01\x01\x07\x04\x00\x01\x01\x03\x02\x00\x04\x00\x00\x00\x14\x05\x04\x00\x01\x01\x0d\x02\x00\x04\x00\x00\x00\x5c" + motor_status_id = ATTR_COVER_MAIN_CONTROL + motor_status_expected = CoverMotorStatus.Stopped blind_open_pct_id = 0x08 blind_open_pct_expected = ( 80 # (blind reports % closed, cluster attribute expects % open) @@ -161,11 +184,11 @@ async def test_cover_report_multiple_values(zigpy_device_from_quirk): battery_pct_id = ( 0x21 # PowerConfiguration.AttributeDefs.battery_percentage_remaining.id ) + motor_direction_id = ATTR_COVER_DIRECTION_SETTING + motor_direction_expected = CoverSettingMotorDirection.Backward battery_pct_expected = 92 * 2 # (attribute expects 2x real percentage) - device = zigpy_device_from_quirk( - zhaquirks.tuya.ts0601_cover.TuyaCover0601MultipleDataPoints - ) + device = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") tuya_cluster = device.endpoints[1].tuya_manufacturer cover_cluster = device.endpoints[1].window_covering cover_listener = ClusterListener(cover_cluster) @@ -180,16 +203,120 @@ async def test_cover_report_multiple_values(zigpy_device_from_quirk): hdr, args = tuya_cluster.deserialize(frame) tuya_cluster.handle_message(hdr, args) - # blind cover will have an attribute update for motor direction, I only care that it - # includes the new position - assert len(cover_listener.attribute_updates) < 3 + assert ( + motor_status_id, + motor_status_expected, + ) in cover_listener.attribute_updates assert ( blind_open_pct_id, blind_open_pct_expected, ) in cover_listener.attribute_updates + assert ( + motor_direction_id, + motor_direction_expected, + ) in cover_listener.attribute_updates assert power_listener.attribute_updates == [(battery_pct_id, battery_pct_expected)] +@pytest.mark.parametrize( + "name, value, expected_frame", + ( + ( + "motor_direction", + CoverSettingMotorDirection.Backward, + b"\x01\x01\x00\x00\x01\x05\x04\x00\x01\x01", + ), + ), +) +async def test_cover_attributes_set( + zigpy_device_from_v2_quirk, name, value, expected_frame +): + """Test expected commands are sent when setting attributes of a tuya cover (that supports multiple data points).""" + + device = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") + tuya_cluster = device.endpoints[1].tuya_manufacturer + cover_cluster = device.endpoints[1].window_covering + + with mock.patch.object( + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + ) as m1: + write_results = await cover_cluster.write_attributes({name: value}) + + await wait_for_zigpy_tasks() + m1.assert_called_with( + 0xEF00, + 1, + expected_frame, + expect_reply=False, + command_id=0, + ) + assert write_results == [ + [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)] + ] + + +@pytest.mark.parametrize( + "inverted, received_frame, expected_received_value, expected_sent_frame", + ( + # When the invert attribute is false, the value is 100-x + ( + False, + b"\x09\x00\x02\x00\x00\x03\x02\x00\x04\x00\x00\x00\x0a", + 90, + b"\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x0a", + ), + # When inverted the value is sent and received unmodified (relative to the attribute/command + # value) + ( + True, + b"\x09\x00\x02\x00\x00\x03\x02\x00\x04\x00\x00\x00\x0a", + 10, + b"\x01\x01\x00\x00\x01\x02\x02\x00\x04\x00\x00\x00\x0a", + ), + ), +) +async def test_cover_invert( + zigpy_device_from_v2_quirk, + inverted, + received_frame, + expected_received_value, + expected_sent_frame, +): + """Test tuya cover position properly honours inverted attribute when sending and receiving.""" + + device = zigpy_device_from_v2_quirk("_TZE200_eevqq1uv", "TS0601") + tuya_cluster = device.endpoints[1].tuya_manufacturer + cover_cluster = device.endpoints[1].window_covering + cover_listener = ClusterListener(cover_cluster) + + # set the invert attribute to the value to be tested + await cover_cluster.write_attributes({"cover_inverted": inverted}) + + # assert we get the value we expect when processing the received frame + blind_open_pct_id = 0x08 + hdr, args = tuya_cluster.deserialize(received_frame) + tuya_cluster.handle_message(hdr, args) + assert ( + blind_open_pct_id, + expected_received_value, + ) in cover_listener.attribute_updates + + # Now send a command to set that value and assert we send the frame we expect + with mock.patch.object( + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + ) as m1: + rsp = await cover_cluster.command(0x05, expected_received_value) + + await wait_for_zigpy_tasks() + m1.assert_called_with( + 0xEF00, + 1, + expected_sent_frame, + expect_reply=True, + command_id=0, + ) + assert rsp.status == foundation.Status.SUCCESS + def test_ts601_moes_signature(assert_signature_matches_quirk): """Test TS0121 cover signature is matched to its quirk.""" @@ -208,28 +335,3 @@ def test_ts601_moes_signature(assert_signature_matches_quirk): "class": "zigpy.device.Device", } assert_signature_matches_quirk(TuyaMoesCover0601, signature) - - -def test_cover_signature_ZemiSmart_ZM25R3_TuyaCover0601MultipleDataPoints( - assert_signature_matches_quirk, -): - """Test Zemismart ZM25R3 matches TuyaCover0601MultipleDataPoints.""" - - signature = { - "node_descriptor": "NodeDescriptor(logical_type=, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=, mac_capability_flags=, manufacturer_code=4098, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=, *allocate_address=True, *is_alternate_pan_coordinator=False, *is_coordinator=False, *is_end_device=True, *is_full_function_device=False, *is_mains_powered=False, *is_receiver_on_when_idle=False, *is_router=False, *is_security_capable=False)", - "endpoints": { - "1": { - "profile_id": 260, - "device_type": "0x0051", - "in_clusters": ["0x0000", "0x0004", "0x0005", "0xef00"], - "out_clusters": ["0x000a", "0x0019"], - } - }, - "manufacturer": "_TZE200_eevqq1uv", - "model": "TS0601", - "class": "zigpy.device.Device", - } - - assert_signature_matches_quirk( - TuyaCover0601MultipleDataPoints, signature - ) diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index 59c1681af3..df3959fdea 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -51,12 +51,14 @@ LEVEL_EVENT = "level_event" TUYA_MCU_COMMAND = "tuya_mcu_command" +TUYA_MCU_SET_DATA = "_tuya_mcu_set_data" # Rotating for remotes STOP = "stop" # To constants # --------------------------------------------------------- # Value for dp_type +# https://developer.tuya.com/en/docs/iot/tuya-zigbee-universal-docking-access-standard?id=K9ik6zvofpzql # --------------------------------------------------------- # ID Name Description # --------------------------------------------------------- @@ -73,22 +75,29 @@ TUYA_DP_TYPE_ENUM = 0x0400 TUYA_DP_TYPE_FAULT = 0x0500 # --------------------------------------------------------- -# Value for dp_identifier (These are device specific) +# Value for dp_identifier. These are device type and potentially device specific. +# The ones we use here appear to be consistent for all covers we support. +# https://developer.tuya.com/en/docs/iot/f?id=K9gf46o5mtfyc # --------------------------------------------------------- # ID Name Type Description # --------------------------------------------------------- # 0x01 control enum open, stop, close, continue # 0x02 percent_control value 0-100% control # 0x03 percent_state value Report from motor about current percentage -# 0x04 control_back enum Configures motor direction (untested) -# 0x05 work_state enum Motor Direction Setting +# 0x05 control_back_mode enum Configures motor direction # 0x06 situation_set enum Configures if 100% equals to fully closed or fully open (untested) # 0x07 fault bitmap Anything but 0 means something went wrong (untested) +# 13 fault value Battery charge percentage +# 16 border enum set open limit, set close limit, clear open, clear close, clear both +# 20 click control enum move up/open a small step, move down/close TUYA_DP_ID_CONTROL = 0x01 TUYA_DP_ID_PERCENT_CONTROL = 0x02 TUYA_DP_ID_PERCENT_STATE = 0x03 -TUYA_DP_ID_DIRECTION_CHANGE = 0x05 +TUYA_DP_ID_DIRECTION_SETTING = 0x05 TUYA_DP_ID_COVER_INVERTED = 0x06 +TUYA_DP_ID_BATTERY_PERCENT = 13 +TUYA_DP_ID_LIMIT_SETTINGS = 16 +TUYA_DP_ID_SMALL_STEP = 20 # --------------------------------------------------------- # Window Cover Server Commands # --------------------------------------------------------- @@ -97,18 +106,31 @@ WINDOW_COVER_COMMAND_STOP = 0x0002 WINDOW_COVER_COMMAND_LIFTPERCENT = 0x0005 WINDOW_COVER_COMMAND_CUSTOM = 0x0006 + +# TODO - What are appropriate ids for a custom commands we introduce that aren't part +# of the zigbee spec, nor tuya manufacturer specific extensions? +# I've used a high uint8 number to try to reduce the chance of conflicts with future +# zigbee/tuya changes. (I believe it needs to be a uint8 to pass to ZclCommandDef, +# despite other ids being declared as 16 bits.) +WINDOW_COVER_COMMAND_SMALL_STEP = 0xF0 +WINDOW_COVER_COMMAND_SMALL_STEP_NAME = "move_small_step" +WINDOW_COVER_COMMAND_UPDATE_LIMITS = 0xF1 +WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME = "update_limits" + # --------------------------------------------------------- # TUYA Cover Custom Values # --------------------------------------------------------- COVER_EVENT = "cover_event" ATTR_COVER_POSITION = 0x0008 -ATTR_COVER_DIRECTION = 0x8001 -ATTR_COVER_DIRECTION_NAME = "motor_direction" -ATTR_COVER_INVERTED = 0x8002 -ATTR_COVER_INVERTED_NAME = "cover_inverted" ATTR_COVER_LIFTPERCENT_NAME = "current_position_lift_percentage" -ATTR_COVER_LIFTPERCENT_CONTROL = 0x8003 -ATTR_COVER_LIFTPERCENT_CONTROL_NAME = "current_position_lift_percentage_control" +ATTR_COVER_MAIN_CONTROL = 0x8000 +ATTR_COVER_MAIN_CONTROL_NAME = "motor_status" +ATTR_COVER_DIRECTION_SETTING = 0x8001 +ATTR_COVER_DIRECTION_SETTING_NAME = "motor_direction" +# Note: I'd like to rename this to lift_percent_inverted (since it's a little ambiguous with +# motor_direction,) but I'm not sure if that will lose the value for existing instances +ATTR_COVER_INVERTED_SETTING = 0x8002 +ATTR_COVER_INVERTED_SETTING_NAME = "cover_inverted" # --------------------------------------------------------- # TUYA Switch Custom Values @@ -213,7 +235,7 @@ def __new__(cls, *args, **kwargs): """Disable copy constructor.""" return super().__new__(cls) - def __init__(self, value=None, function=0, *args, **kwargs): + def __init__(self, value=None, function=0, *_args, **_kwargs): """Convert from a zigpy typed value to a tuya data payload.""" self.function = function @@ -327,7 +349,13 @@ class TuyaManufCluster(CustomCluster): set_time_local_offset = None class Command(t.Struct): - """Tuya manufacturer cluster command.""" + """Tuya manufacturer cluster command. + + Note: This doesn't seem to match the way tuya define their packet structure. (e.g. command + should be 1 byte, followed by a list of data points.) It may have changed over time but + TuyaCommand is a better match for their newer devices. + https://developer.tuya.com/en/docs/iot/tuya-zigbee-universal-docking-access-standard?id=K9ik6zvofpzql + """ status: t.uint8_t tsn: t.uint8_t @@ -653,7 +681,7 @@ def handle_cluster_request( tuya_payload.data[1], ) elif hdr.command_id == TUYA_SET_TIME: - """Time event call super""" + # Time event call super _LOGGER.debug("TUYA_SET_TIME --> hdr: %s, args: %s", hdr, args) super().handle_cluster_request(hdr, args, dst_addressing=dst_addressing) else: @@ -704,7 +732,7 @@ def state_change(self, value): self._update_attribute(self.attributes_by_name["running_state"].id, state) # pylint: disable=R0201 - def map_attribute(self, attribute, value): + def map_attribute(self, _attribute, _value): """Map standardized attribute value to dict of manufacturer values.""" return {} @@ -812,7 +840,7 @@ def child_lock_change(self, mode): self._update_attribute(self.attributes_by_name["keypad_lockout"].id, lockout) - def map_attribute(self, attribute, value): + def map_attribute(self, _attribute, _value): """Map standardized attribute value to dict of manufacturer values.""" return {} @@ -1175,11 +1203,11 @@ def handle_cluster_request( ) elif ( tuya_payload.command_id - == TUYA_DP_TYPE_ENUM + TUYA_DP_ID_DIRECTION_CHANGE + == TUYA_DP_TYPE_ENUM + TUYA_DP_ID_DIRECTION_SETTING ): self.endpoint.device.cover_bus.listener_event( COVER_EVENT, - ATTR_COVER_DIRECTION, + ATTR_COVER_DIRECTION_SETTING, tuya_payload.data[1], ) elif ( @@ -1187,7 +1215,7 @@ def handle_cluster_request( ): self.endpoint.device.cover_bus.listener_event( COVER_EVENT, - ATTR_COVER_INVERTED, + ATTR_COVER_INVERTED_SETTING, tuya_payload.data[1], # Check this ) elif hdr.command_id == TUYA_SET_TIME: @@ -1212,8 +1240,8 @@ class TuyaWindowCoverControl(LocalDataCluster, WindowCovering): # Add additional attributes for direction attributes = WindowCovering.attributes.copy() - attributes.update({ATTR_COVER_DIRECTION: (ATTR_COVER_DIRECTION_NAME, t.enum8)}) - attributes.update({ATTR_COVER_INVERTED: (ATTR_COVER_INVERTED_NAME, t.Bool)}) + attributes.update({ATTR_COVER_DIRECTION_SETTING: (ATTR_COVER_DIRECTION_SETTING_NAME, t.enum8)}) + attributes.update({ATTR_COVER_INVERTED_SETTING: (ATTR_COVER_INVERTED_SETTING_NAME, t.Bool)}) def __init__(self, *args, **kwargs): """Initialize instance.""" @@ -1223,7 +1251,7 @@ def __init__(self, *args, **kwargs): def cover_event(self, attribute, value): """Event listener for cover events.""" if attribute == ATTR_COVER_POSITION: - invert_attr = self._attr_cache.get(ATTR_COVER_INVERTED) == 1 + invert_attr = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 invert = ( not invert_attr if self.endpoint.device.tuya_cover_inverted_by_default @@ -1278,7 +1306,7 @@ def command( tuya_payload.command_id = TUYA_DP_TYPE_VALUE + TUYA_DP_ID_PERCENT_CONTROL tuya_payload.function = 0 # Check direction and correct value - invert_attr = self._attr_cache.get(ATTR_COVER_INVERTED) == 1 + invert_attr = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 invert = ( not invert_attr if self.endpoint.device.tuya_cover_inverted_by_default @@ -1618,7 +1646,7 @@ def handle_get_data(self, command: TuyaCommand) -> foundation.Status: handle_set_data_response = handle_get_data handle_active_status_report = handle_get_data - def handle_set_time_request(self, payload: t.uint16_t) -> foundation.Status: + def handle_set_time_request(self, _payload: t.uint16_t) -> foundation.Status: """Handle Time set request.""" return foundation.Status.SUCCESS diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index d66745c820..4922fc6278 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -15,25 +15,38 @@ # add EnchantedDevice import for custom quirks backwards compatibility from zhaquirks.tuya import ( - ATTR_COVER_DIRECTION, - ATTR_COVER_DIRECTION_NAME, - ATTR_COVER_INVERTED, - ATTR_COVER_INVERTED_NAME, - ATTR_COVER_LIFTPERCENT_CONTROL, - ATTR_COVER_LIFTPERCENT_CONTROL_NAME, + ATTR_COVER_DIRECTION_SETTING, + ATTR_COVER_DIRECTION_SETTING_NAME, + ATTR_COVER_INVERTED_SETTING, + ATTR_COVER_INVERTED_SETTING_NAME, ATTR_COVER_LIFTPERCENT_NAME, + ATTR_COVER_MAIN_CONTROL, + ATTR_COVER_MAIN_CONTROL_NAME, + TUYA_DP_ID_BATTERY_PERCENT, + TUYA_DP_ID_CONTROL, + TUYA_DP_ID_DIRECTION_SETTING, + TUYA_DP_ID_LIMIT_SETTINGS, + TUYA_DP_ID_PERCENT_CONTROL, + TUYA_DP_ID_PERCENT_STATE, + TUYA_DP_ID_SMALL_STEP, TUYA_MCU_COMMAND, + TUYA_MCU_SET_DATA, TUYA_MCU_VERSION_RSP, TUYA_SET_DATA, TUYA_SET_TIME, WINDOW_COVER_COMMAND_DOWNCLOSE, WINDOW_COVER_COMMAND_LIFTPERCENT, + WINDOW_COVER_COMMAND_SMALL_STEP, + WINDOW_COVER_COMMAND_SMALL_STEP_NAME, WINDOW_COVER_COMMAND_STOP, + WINDOW_COVER_COMMAND_UPDATE_LIMITS, + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, WINDOW_COVER_COMMAND_UPOPEN, EnchantedDevice, # noqa: F401 NoManufacturerCluster, PowerOnState, TuyaCommand, + TuyaData, TuyaDatapointData, TuyaLocalCluster, TuyaNewManufCluster, @@ -92,6 +105,37 @@ class MoesBacklight(t.enum8): light_when_off = 0x02 freeze = 0x03 +class CoverCommandStepDirection(t.enum8): + """Window cover step command direction enum.""" + + Open = 0 + Close = 1 + + +class CoverMotorStatus(t.enum8): + """Window cover motor states enum.""" + + Opening = 0 + Stopped = 1 + Closing = 2 + + +class CoverSettingMotorDirection(t.enum8): + """Window cover motor direction configuration enum.""" + + Forward = 0 + Backward = 1 + + +class CoverSettingLimitOperation(t.enum8): + """Window cover limits item to set / clear.""" + + SetOpen = 0 + SetClose = 1 + ClearOpen = 2 + ClearClose = 3 + ClearBoth = 4 + class TuyaPowerConfigurationCluster( TuyaLocalCluster, DoublingPowerConfigurationCluster @@ -299,6 +343,33 @@ def tuya_mcu_command(self, cluster_data: TuyaClusterData): cluster = getattr(endpoint, cluster_data.cluster_name) cluster.update_attribute(cluster_data.cluster_attr, cluster_data.attr_value) + def _tuya_mcu_set_data( + self, + datapoints: list[TuyaDatapointData], + manufacturer: Optional[Union[int, t.uint16_t]] = None, + expect_reply: bool = True, + ): + # TODO - merge with tuya_mcu_command, or just rename to TUYA_MCU_SET_CLUSTER_DATA & TUYA_MCU_SET_DATAPOINTS + self.debug("tuya_mcu_set_data: datapoints=%s", datapoints) + + if len(datapoints) == 0: + self.warning("no datapoints for tuya_mcu_set_data") + return + + cmd_payload = TuyaCommand() + cmd_payload.status = 0 + cmd_payload.tsn = self.endpoint.device.application.get_sequence() + cmd_payload.datapoints = datapoints + + self.create_catching_task( + self.command( + TUYA_SET_DATA, + cmd_payload, + manufacturer=manufacturer, + expect_reply=expect_reply, + ) + ) + def get_dp_mapping( self, endpoint_id: int, attribute_name: str ) -> Optional[tuple[int, DPToAttributeMapping]]: @@ -571,73 +642,62 @@ class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): Tuya data points. """ + # TODO - work out if I can do this as a class function, yet still access the invert attribute value def __init__(self, *args, **kwargs): """Init.""" super().__init__(*args, **kwargs) self.dp_to_attribute: dict[int, DPToAttributeMapping] = { - 1: DPToAttributeMapping( + TUYA_DP_ID_CONTROL: DPToAttributeMapping( TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_DIRECTION_NAME, - # make sure this is sent as a enum, regardless that tuya_cover_command probably - # contains ints. - None, - t.enum8 + ATTR_COVER_MAIN_CONTROL_NAME, + # Converting raw ints to a type allows the attributes UI show meaningful values + CoverMotorStatus, + CoverMotorStatus, ), - 2: DPToAttributeMapping( + TUYA_DP_ID_PERCENT_STATE: DPToAttributeMapping( TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_LIFTPERCENT_CONTROL_NAME, - self.convert_lift_percent, - self.convert_lift_percent + ATTR_COVER_LIFTPERCENT_NAME, + self._convert_lift_percent, + self._convert_lift_percent, ), - 3: DPToAttributeMapping( + TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_LIFTPERCENT_NAME, - self.convert_lift_percent, - self.convert_lift_percent + ATTR_COVER_DIRECTION_SETTING_NAME, + CoverSettingMotorDirection, + CoverSettingMotorDirection, ), - 13: DPToAttributeMapping( + TUYA_DP_ID_BATTERY_PERCENT: DPToAttributeMapping( PowerConfiguration.ep_attribute, - PowerConfiguration.AttributeDefs.battery_percentage_remaining.name + PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, # Tuya report real percent, zigbee expects value*2, but # TuyaPowerConfigurationCluster will convert it ), } data_point_handlers = { - 1: "_dp_2_attr_update", - 2: "_dp_2_attr_update", - 3: "_dp_2_attr_update", - # Define DPs 5 & 7 to avoid warning logs, but I'm not sure what they are so just ignore - # updates from them - 5: "ignore_update", + TUYA_DP_ID_CONTROL: "_dp_2_attr_update", + TUYA_DP_ID_PERCENT_STATE: "_dp_2_attr_update", + TUYA_DP_ID_DIRECTION_SETTING: "_dp_2_attr_update", + TUYA_DP_ID_BATTERY_PERCENT: "_dp_2_attr_update", + # Ignore updates from data points that are used as write-only commands to the device, we + # don't need attributes to display their values, but they're they're echoed back in + # get_data and would otherwise log debug messages. + TUYA_DP_ID_PERCENT_CONTROL: "ignore_update", + # I don't know what 7 is, but it's part of set_data_response 7: "ignore_update", - 13: "_dp_2_attr_update", + TUYA_DP_ID_LIMIT_SETTINGS: "ignore_update", + TUYA_DP_ID_SMALL_STEP: "ignore_update", } - def convert_lift_percent(self, input_value: int): - """Convert/invert lift percent when needed. - - HA shows % open. The zigbee cluster value is called 'lift_percent' but seems to need to - be % closed. This logic follows the convention of other Tuya covers, inverting the value - by default and reversing that invert if tuya_cover_inverted_by_default or the cluster - invert attribute is set. (This seems strange to me, but it's better to be consistent.) - """ - - invert_attr = self.endpoint.window_covering._attr_cache.get(ATTR_COVER_INVERTED) == 1 - invert = ( - not invert_attr - if self.endpoint.device.tuya_cover_inverted_by_default - else invert_attr - ) - return input_value if invert else 100 - input_value + def _convert_lift_percent(self, input_value: int): + return self.endpoint.window_covering.convert_lift_percent(input_value) def ignore_update(self, _datapoint: TuyaDatapointData) -> None: """Process (and ignore) some data point updates.""" return None - -class TuyaNewWindowCoverControl(TuyaLocalCluster, WindowCovering): +class TuyaNewWindowCoverControl(TuyaAttributesCluster, WindowCovering): """Tuya Window Cover Cluster, based on new TuyaNewManufClusterForWindowCover. Derive from TuyaLocalCluster to disable attribute writes, in a way that's compatible with @@ -645,11 +705,53 @@ class TuyaNewWindowCoverControl(TuyaLocalCluster, WindowCovering): """ attributes = WindowCovering.attributes.copy() - # cover direction and lift percent control attributes are not very useful to HA, but needed to - # allow TuyaMCUCluster to map to the data point. - attributes.update({ATTR_COVER_DIRECTION: (ATTR_COVER_DIRECTION_NAME, t.enum8)}) - attributes.update({ATTR_COVER_INVERTED: (ATTR_COVER_INVERTED_NAME, t.Bool)}) - attributes.update({ATTR_COVER_LIFTPERCENT_CONTROL: (ATTR_COVER_LIFTPERCENT_CONTROL_NAME, t.uint16_t)}) + # main control attribute is logically write-only, only used by commands and not very useful + # to HA, but it's return in a set_data and set_data_response packet so I've mapped it to + # an attribute. + attributes.update( + { + ATTR_COVER_MAIN_CONTROL: (ATTR_COVER_MAIN_CONTROL_NAME, t.enum8), + ATTR_COVER_INVERTED_SETTING: ( + ATTR_COVER_INVERTED_SETTING_NAME, + t.Bool, + ), + ATTR_COVER_DIRECTION_SETTING: ( + ATTR_COVER_DIRECTION_SETTING_NAME, + CoverSettingMotorDirection, + ), + } + ) + + server_commands = WindowCovering.server_commands.copy() + server_commands.update( + { + WINDOW_COVER_COMMAND_SMALL_STEP: foundation.ZCLCommandDef( + WINDOW_COVER_COMMAND_SMALL_STEP, + {"direction": CoverCommandStepDirection}, + foundation.Direction.Client_to_Server, + is_manufacturer_specific=True, + name=WINDOW_COVER_COMMAND_SMALL_STEP_NAME, + ), + WINDOW_COVER_COMMAND_UPDATE_LIMITS: foundation.ZCLCommandDef( + WINDOW_COVER_COMMAND_UPDATE_LIMITS, + {"operation": CoverSettingLimitOperation}, + foundation.Direction.Client_to_Server, + is_manufacturer_specific=True, + name=WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, + ), + } + ) + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + + # For most tuya devices Up/Open = 0, Stop = 1, Down/Close = 2 + self.tuya_cover_command = { + WINDOW_COVER_COMMAND_UPOPEN: 0x0000, + WINDOW_COVER_COMMAND_DOWNCLOSE: 0x0002, + WINDOW_COVER_COMMAND_STOP: 0x0001, + } async def command( self, @@ -658,14 +760,15 @@ async def command( manufacturer: Optional[Union[int, t.uint16_t]] = None, expect_reply: bool = True, tsn: Optional[Union[int, t.uint8_t]] = None, - **_kwargs: Any + **kwargs: Any, ): """Override the default Cluster command.""" _LOGGER.debug( - "Sending Tuya Cluster Command... Cluster Command is %x, Arguments are %s", + "Sending Tuya Cluster Command... Cluster Command is %x, args=%s, kwargs=%s", command_id, args, + kwargs, ) # Open Close or Stop commands @@ -677,9 +780,9 @@ async def command( cluster_data = TuyaClusterData( endpoint_id=self.endpoint.endpoint_id, cluster_name=self.ep_attribute, - cluster_attr=ATTR_COVER_DIRECTION_NAME, + cluster_attr=ATTR_COVER_MAIN_CONTROL_NAME, # Map from zigbee command to tuya DP value - attr_value=self.endpoint.device.tuya_cover_command[command_id], + attr_value=self.tuya_cover_command[command_id], expect_reply=expect_reply, manufacturer=manufacturer, ) @@ -687,30 +790,75 @@ async def command( TUYA_MCU_COMMAND, cluster_data, ) - return foundation.GENERAL_COMMANDS[ - foundation.GeneralCommand.Default_Response - ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + return self._default_response(command_id) + # TODO - refactor command to DP mapping into something similar to _dp_2_attr_update & remove send_tuya_set_data_command or move it to TuyaMCUCluster elif command_id == WINDOW_COVER_COMMAND_LIFTPERCENT: - cluster_data = TuyaClusterData( - endpoint_id=self.endpoint.endpoint_id, - cluster_name=self.ep_attribute, - cluster_attr=ATTR_COVER_LIFTPERCENT_CONTROL_NAME, - attr_value=args[0], + self.send_tuya_set_data_command( + TUYA_DP_ID_PERCENT_CONTROL, + self.convert_lift_percent(args[0]), expect_reply=expect_reply, manufacturer=manufacturer, ) - self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, - cluster_data, + return self._default_response(command_id) + elif command_id == WINDOW_COVER_COMMAND_SMALL_STEP: + self.send_tuya_set_data_command( + TUYA_DP_ID_SMALL_STEP, + kwargs["direction"], + expect_reply=expect_reply, + manufacturer=manufacturer, ) - return foundation.GENERAL_COMMANDS[ - foundation.GeneralCommand.Default_Response - ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + return self._default_response(command_id) + elif command_id == WINDOW_COVER_COMMAND_UPDATE_LIMITS: + self.send_tuya_set_data_command( + TUYA_DP_ID_LIMIT_SETTINGS, + kwargs["operation"], + expect_reply=expect_reply, + manufacturer=manufacturer, + ) + return self._default_response(command_id) _LOGGER.warning("Unsupported command_id: %s", command_id) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND) + def send_tuya_set_data_command( + self, + dp: t.uint8_t, + data: TuyaData, + manufacturer: Optional[Union[int, t.uint16_t]] = None, + expect_reply: bool = True, + ): + """Send a set_data for a Tuya data point value (via the mcu cluster).""" + + datapoints = [TuyaDatapointData(dp, data)] + self.debug("Sending TUYA_MCU_SET_DATA: %s", datapoints) + + self.endpoint.device.command_bus.listener_event( + TUYA_MCU_SET_DATA, datapoints, manufacturer, expect_reply + ) + + # TODO - move this to TuyaMCUCluster + def _default_response( + self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] + ): + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + + def convert_lift_percent(self, input_value: int): + """Convert/invert lift percent when needed. + + HA shows % open. The zigbee cluster value is called 'lift_percent' but seems to need to + be % closed. This logic follows the convention of other Tuya covers, inverting the value + by default, unless the cluster invert attribute is set. (This seems strange to me, but it's + better to be consistent.) + + It's safe to use the same calculation converting motor position to zigbee attribute value + and attribute value to motor position command. + """ + + invert = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 + return input_value if invert else 100 - input_value class TuyaLevelControl(LevelControl, TuyaLocalCluster): """Tuya MCU Level cluster for dimmable device.""" diff --git a/zhaquirks/tuya/ts0601_cover.py b/zhaquirks/tuya/ts0601_cover.py index d1fb298834..239a868c82 100644 --- a/zhaquirks/tuya/ts0601_cover.py +++ b/zhaquirks/tuya/ts0601_cover.py @@ -1,6 +1,11 @@ """Tuya based cover and blinds.""" +import logging + from zigpy.profiles import zha +from zigpy.quirks import _DEVICE_REGISTRY +from zigpy.quirks.v2 import add_to_registry_v2 +from zigpy.quirks.v2.homeassistant import EntityType from zigpy.zcl.clusters.general import Basic, Groups, Identify, OnOff, Ota, Scenes, Time from zhaquirks.const import ( @@ -12,17 +17,25 @@ PROFILE_ID, ) from zhaquirks.tuya import ( + ATTR_COVER_DIRECTION_SETTING_NAME, + WINDOW_COVER_COMMAND_SMALL_STEP_NAME, + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaManufacturerWindowCover, TuyaManufCluster, TuyaWindowCover, TuyaWindowCoverControl, ) from zhaquirks.tuya.mcu import ( + CoverCommandStepDirection, + CoverSettingLimitOperation, + CoverSettingMotorDirection, TuyaNewManufClusterForWindowCover, TuyaNewWindowCoverControl, TuyaPowerConfigurationCluster, ) +_LOGGER = logging.getLogger(__name__) + class TuyaZemismartSmartCover0601(TuyaWindowCover): """Tuya Zemismart blind cover motor.""" @@ -626,54 +639,72 @@ class TuyaCloneCover0601(TuyaWindowCover): } -class TuyaCover0601MultipleDataPoints(TuyaWindowCover): - """Tuya window cover device. - - This variant supports: - - multiple data points included in tuya set_data_response. - - non-inverted control inputs, - - battery percentage remaining - - Most/all the quirks above are based on TuyaManufacturerWindowCover that only decodes - ONE attribute from the Tuya set_data_response packet. This quirk is based on - TuyaNewManufClusterForWindowCover which can handle multiple updates in one zigby frame. - """ - - signature = { - MODELS_INFO: [ - ("_TZE200_eevqq1uv", "TS0601"), # Zemismart ZM25R3 roller blind motor - ], - # SimpleDescriptor(endpoint=1, profile=260, device_type=81, device_version=1, - # input_clusters=[0, 4, 5, 61184], - # output_clusters=[25, 10]) - ENDPOINTS: { - 1: { - PROFILE_ID: zha.PROFILE_ID, - DEVICE_TYPE: zha.DeviceType.SMART_PLUG, - INPUT_CLUSTERS: [ - Basic.cluster_id, - Groups.cluster_id, - Scenes.cluster_id, - TuyaNewManufClusterForWindowCover.cluster_id, - ], - OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], - } - }, - } - - replacement = { - ENDPOINTS: { - 1: { - DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE, - INPUT_CLUSTERS: [ - Basic.cluster_id, - Groups.cluster_id, - Scenes.cluster_id, - TuyaNewWindowCoverControl, - TuyaPowerConfigurationCluster, - TuyaNewManufClusterForWindowCover, - ], - OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id], - } - } - } +def register_v2_quirks() -> None: + """Register v2 Quirks.""" + + # Tuya window cover device. + # + # This variant supports: + # - multiple data points included in tuya set_data_response. + # - non-inverted control inputs, + # - battery percentage remaining + # + # Most/all the quirks above are based on TuyaManufacturerWindowCover that only decodes + # ONE attribute from the Tuya set_data_response packet. This quirk is based on + # TuyaNewManufClusterForWindowCover which can handle multiple updates in one zigby frame. + # + # "_TZE200_eevqq1uv", "TS0601" - Zemismart ZM25R3 roller blind motor + add_to_registry_v2("_TZE200_68nvbio9", "TS0601" + ).also_applies_to("_TZE200_eevqq1uv", "TS0601" + ).replaces(TuyaNewManufClusterForWindowCover + ).adds(TuyaNewWindowCoverControl + ).adds(TuyaPowerConfigurationCluster + ).command_button( + WINDOW_COVER_COMMAND_SMALL_STEP_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"direction": CoverCommandStepDirection.Open}, + entity_type=EntityType.STANDARD, + translation_key="small_step_open", + ).command_button( + WINDOW_COVER_COMMAND_SMALL_STEP_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"direction": CoverCommandStepDirection.Close}, + entity_type=EntityType.STANDARD, + translation_key="small_step_close", + ).enum( + ATTR_COVER_DIRECTION_SETTING_NAME, + CoverSettingMotorDirection, + TuyaNewWindowCoverControl.cluster_id, + translation_key="motor_direction", + ).command_button( + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"operation": CoverSettingLimitOperation.SetOpen}, + translation_key="set_open_limit", + ).command_button( + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"operation": CoverSettingLimitOperation.SetClose}, + translation_key="set_close_limit", + ).command_button( + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"operation": CoverSettingLimitOperation.ClearOpen}, + translation_key="clear_open_limit", + ).command_button( + WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, + TuyaNewWindowCoverControl.cluster_id, + None, + {"operation": CoverSettingLimitOperation.ClearClose}, + translation_key="clear_close_limit", + ) + + +( + register_v2_quirks() +) From 45144e383b162d3541ce464a6570c313766dca53 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Thu, 2 May 2024 22:53:51 +1000 Subject: [PATCH 04/14] Rename TUYA_MCU_SET_DATA to TUYA_MCU_SET_DATAPOINTS --- zhaquirks/tuya/__init__.py | 2 +- zhaquirks/tuya/mcu/__init__.py | 15 ++++++++------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index df3959fdea..0bdb525a0b 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -51,7 +51,7 @@ LEVEL_EVENT = "level_event" TUYA_MCU_COMMAND = "tuya_mcu_command" -TUYA_MCU_SET_DATA = "_tuya_mcu_set_data" +TUYA_MCU_SET_DATAPOINTS = "tuya_mcu_set_datapoints" # Rotating for remotes STOP = "stop" # To constants diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 4922fc6278..2e61e9733f 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -30,7 +30,7 @@ TUYA_DP_ID_PERCENT_STATE, TUYA_DP_ID_SMALL_STEP, TUYA_MCU_COMMAND, - TUYA_MCU_SET_DATA, + TUYA_MCU_SET_DATAPOINTS, TUYA_MCU_VERSION_RSP, TUYA_SET_DATA, TUYA_SET_TIME, @@ -343,17 +343,18 @@ def tuya_mcu_command(self, cluster_data: TuyaClusterData): cluster = getattr(endpoint, cluster_data.cluster_name) cluster.update_attribute(cluster_data.cluster_attr, cluster_data.attr_value) - def _tuya_mcu_set_data( + def tuya_mcu_set_datapoints( self, datapoints: list[TuyaDatapointData], manufacturer: Optional[Union[int, t.uint16_t]] = None, expect_reply: bool = True, ): - # TODO - merge with tuya_mcu_command, or just rename to TUYA_MCU_SET_CLUSTER_DATA & TUYA_MCU_SET_DATAPOINTS - self.debug("tuya_mcu_set_data: datapoints=%s", datapoints) + """Tuya MCU listener to send/set tuya datapoint values.""" + + self.debug("tuya_mcu_set_datapoints: datapoints=%s", datapoints) if len(datapoints) == 0: - self.warning("no datapoints for tuya_mcu_set_data") + self.warning("no datapoints for tuya_mcu_set_datapoints") return cmd_payload = TuyaCommand() @@ -831,10 +832,10 @@ def send_tuya_set_data_command( """Send a set_data for a Tuya data point value (via the mcu cluster).""" datapoints = [TuyaDatapointData(dp, data)] - self.debug("Sending TUYA_MCU_SET_DATA: %s", datapoints) + self.debug("Sending TUYA_MCU_SET_DATAPOINTS: %s", datapoints) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_SET_DATA, datapoints, manufacturer, expect_reply + TUYA_MCU_SET_DATAPOINTS, datapoints, manufacturer, expect_reply ) # TODO - move this to TuyaMCUCluster From ee99882a0737802c477f47c82bd96808d49eedf5 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Thu, 2 May 2024 23:17:06 +1000 Subject: [PATCH 05/14] Add TUYA_MCU_SET_CLUSTER_DATA replacing most calls to TUYA_MCU_COMMAND TuyaManufCluster has a TUYA_MCU_COMMAND listener that is passed a command to send. Many existing calls to TUYA_MCU_COMMAND were calling the TuyaMCUCluster variant which takes a TuyaClusterData structure. I've renamed those calls to TUYA_MCU_SET_CLUSTER_DATA so sit alongside TUYA_MCU_SET_DATAPOINTS and distinguish the data type they accept. --- tests/test_tuya_mcu.py | 4 ++-- zhaquirks/tuya/__init__.py | 1 + zhaquirks/tuya/mcu/__init__.py | 21 ++++++++++++--------- zhaquirks/tuya/ts0601_rcbo.py | 6 +++--- zhaquirks/tuya/ts0601_siren.py | 4 ++-- 5 files changed, 20 insertions(+), 16 deletions(-) diff --git a/tests/test_tuya_mcu.py b/tests/test_tuya_mcu.py index 615dffcea9..70853b9ff5 100644 --- a/tests/test_tuya_mcu.py +++ b/tests/test_tuya_mcu.py @@ -276,14 +276,14 @@ async def test_tuya_methods(zigpy_device_from_quirk, quirk): assert not result_2 with mock.patch.object(tuya_cluster, "create_catching_task") as m1: - tuya_cluster.tuya_mcu_command(tcd_2) + tuya_cluster.tuya_mcu_set_cluster_data(tcd_2) # no DP resolution will not call TUYA_SET_DATA command m1.assert_not_called() result_3 = await dimmer2_cluster.command(0x0006) assert result_3.status == foundation.Status.UNSUP_CLUSTER_COMMAND - with mock.patch.object(tuya_cluster, "tuya_mcu_command") as m1: + with mock.patch.object(tuya_cluster, "tuya_mcu_set_cluster_data") as m1: rsp = await switch1_cluster.command(0x0001) m1.assert_called_once_with(tcd_switch1_on) diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index 0bdb525a0b..8719d3b192 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -51,6 +51,7 @@ LEVEL_EVENT = "level_event" TUYA_MCU_COMMAND = "tuya_mcu_command" +TUYA_MCU_SET_CLUSTER_DATA = "tuya_mcu_set_cluster_data" TUYA_MCU_SET_DATAPOINTS = "tuya_mcu_set_datapoints" # Rotating for remotes diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 2e61e9733f..ebf0036540 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -29,7 +29,7 @@ TUYA_DP_ID_PERCENT_CONTROL, TUYA_DP_ID_PERCENT_STATE, TUYA_DP_ID_SMALL_STEP, - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, TUYA_MCU_SET_DATAPOINTS, TUYA_MCU_VERSION_RSP, TUYA_SET_DATA, @@ -175,7 +175,7 @@ async def write_attributes(self, attributes, manufacturer=None): manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) @@ -312,11 +312,14 @@ def from_cluster_data(self, data: TuyaClusterData) -> Optional[TuyaCommand]: tuya_commands.append(cmd_payload) return tuya_commands - def tuya_mcu_command(self, cluster_data: TuyaClusterData): - """Tuya MCU command listener. Only manufacturer endpoint must listen to MCU commands.""" + def tuya_mcu_set_cluster_data(self, cluster_data: TuyaClusterData): + """Tuya MCU listener to send/set tuya data points from cluster attributes. + + Only manufacturer endpoint must listen to MCU commands. + """ self.debug( - "tuya_mcu_command: cluster_data=%s", + "tuya_mcu_set_cluster_data: cluster_data=%s", cluster_data, ) @@ -476,7 +479,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return foundation.GENERAL_COMMANDS[ @@ -788,7 +791,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return self._default_response(command_id) @@ -902,7 +905,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) @@ -923,7 +926,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return foundation.GENERAL_COMMANDS[ diff --git a/zhaquirks/tuya/ts0601_rcbo.py b/zhaquirks/tuya/ts0601_rcbo.py index b1b2b82b55..acd6fc24ab 100644 --- a/zhaquirks/tuya/ts0601_rcbo.py +++ b/zhaquirks/tuya/ts0601_rcbo.py @@ -25,7 +25,7 @@ OUTPUT_CLUSTERS, PROFILE_ID, ) -from zhaquirks.tuya import TUYA_MCU_COMMAND, AttributeWithMask, PowerOnState +from zhaquirks.tuya import TUYA_MCU_SET_CLUSTER_DATA, AttributeWithMask, PowerOnState from zhaquirks.tuya.mcu import ( DPToAttributeMapping, TuyaAttributesCluster, @@ -185,7 +185,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return foundation.GENERAL_COMMANDS[ @@ -319,7 +319,7 @@ async def command( manufacturer=manufacturer, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return foundation.GENERAL_COMMANDS[ diff --git a/zhaquirks/tuya/ts0601_siren.py b/zhaquirks/tuya/ts0601_siren.py index 501813f88f..b1ab468fc7 100644 --- a/zhaquirks/tuya/ts0601_siren.py +++ b/zhaquirks/tuya/ts0601_siren.py @@ -28,7 +28,7 @@ PROFILE_ID, ) from zhaquirks.tuya import ( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, TuyaManufCluster, TuyaManufClusterAttributes, ) @@ -300,7 +300,7 @@ async def command( manufacturer=foundation.ZCLHeader.NO_MANUFACTURER_ID, ) self.endpoint.device.command_bus.listener_event( - TUYA_MCU_COMMAND, + TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) return foundation.GENERAL_COMMANDS[ From 34a508d4114a191eab71c53cc5cc8572f298d338 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Fri, 3 May 2024 06:55:15 +1000 Subject: [PATCH 06/14] Add TuyaCommandCluster to process commands based on map to tuya data points --- zhaquirks/tuya/mcu/__init__.py | 178 ++++++++++++++++++++++----------- 1 file changed, 122 insertions(+), 56 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index ebf0036540..3a1920aa4b 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -6,6 +6,7 @@ import logging from typing import Any, Optional, Union +from zigpy.quirks import CustomCluster import zigpy.types as t from zigpy.zcl import foundation from zigpy.zcl.clusters.closures import WindowCovering @@ -86,6 +87,14 @@ class DPToAttributeMapping: endpoint_id: Optional[int] = None +@dataclasses.dataclass +class CommandToDPValueMapping: + """Container for command id to datapoint value mapping.""" + + dp: t.uint8_t + value_source: Union[int, str, Callable[..., TuyaData]] + + class TuyaClusterData(t.Struct): """Tuya cluster data.""" @@ -182,6 +191,90 @@ async def write_attributes(self, attributes, manufacturer=None): return [[foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)]] +class TuyaCommandCluster(CustomCluster): + """A tuya-based cluster that accepts zigbee commands and maps them to data point updates. + + Derived classed only need define a map and value converter to enable processing of commands + into data point updates, sent to the tuya mcu cluster to send a set data command to the device. + """ + + command_to_dp: dict[Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping] = { + } + + async def command( + self, + command_id: Union[foundation.GeneralCommand, int, t.uint8_t], + *args, + manufacturer: Optional[Union[int, t.uint16_t]] = None, + expect_reply: bool = True, + _tsn: Optional[Union[int, t.uint8_t]] = None, + **kwargs: Any, + ): + """Process any commands that are mapped to data points.""" + _LOGGER.debug( + "Processing command to dp mappings for Cluster Command. Command is %x, args=%s, kwargs=%s", + command_id, + args, + kwargs, + ) + + # if there's a map for this command to a data points, call the map value function and send + # the new value to the MCU cluster to send to the device + command_map = self.command_to_dp.get(command_id, None) + if command_map is not None: + # command_map.value_source can refer to a numbered or named parameter or lambda + if isinstance(command_map.value_source, int): + value = args[command_map.value_source] + elif isinstance(command_map.value_source, str): + value = kwargs[command_map.value_source] + else: + value = command_map.value_source(*args, **kwargs) + + self.send_tuya_set_datapoints_command(command_map.dp, value, expect_reply=expect_reply, + manufacturer=manufacturer) + return self.default_response(command_id) + + _LOGGER.warning("Unsupported command_id: %s", command_id) + return self.unsupported_response(command_id) + + + def send_tuya_set_datapoints_command( + self, + dp: t.uint8_t, + data: TuyaData, + manufacturer: Optional[Union[int, t.uint16_t]] = None, + expect_reply: bool = True, + ): + """Send a set_data for a Tuya data point value (via the mcu cluster).""" + + datapoints = [TuyaDatapointData(dp, data)] + self.debug("Sending TUYA_MCU_SET_DATAPOINTS: %s", datapoints) + + self.endpoint.device.command_bus.listener_event( + TUYA_MCU_SET_DATAPOINTS, datapoints, manufacturer, expect_reply + ) + + + def default_response( + self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] + ): + """Return a default success response for a given command.""" + + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.SUCCESS) + + + def unsupported_response( + self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] + ): + """Return an 'unsupported' response for a given command.""" + + return foundation.GENERAL_COMMANDS[ + foundation.GeneralCommand.Default_Response + ].schema(command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND) + + class TuyaMCUCluster(TuyaAttributesCluster, TuyaNewManufCluster): """Manufacturer specific cluster for sending Tuya MCU commands.""" @@ -701,7 +794,8 @@ def ignore_update(self, _datapoint: TuyaDatapointData) -> None: """Process (and ignore) some data point updates.""" return None -class TuyaNewWindowCoverControl(TuyaAttributesCluster, WindowCovering): + +class TuyaNewWindowCoverControl(TuyaAttributesCluster, TuyaCommandCluster, WindowCovering): """Tuya Window Cover Cluster, based on new TuyaNewManufClusterForWindowCover. Derive from TuyaLocalCluster to disable attribute writes, in a way that's compatible with @@ -746,16 +840,17 @@ class TuyaNewWindowCoverControl(TuyaAttributesCluster, WindowCovering): } ) - def __init__(self, *args, **kwargs): - """Init.""" - super().__init__(*args, **kwargs) + command_to_dp: dict[Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping] = { + WINDOW_COVER_COMMAND_SMALL_STEP: CommandToDPValueMapping(TUYA_DP_ID_SMALL_STEP, "direction"), + WINDOW_COVER_COMMAND_UPDATE_LIMITS: CommandToDPValueMapping(TUYA_DP_ID_LIMIT_SETTINGS, "operation"), + } - # For most tuya devices Up/Open = 0, Stop = 1, Down/Close = 2 - self.tuya_cover_command = { - WINDOW_COVER_COMMAND_UPOPEN: 0x0000, - WINDOW_COVER_COMMAND_DOWNCLOSE: 0x0002, - WINDOW_COVER_COMMAND_STOP: 0x0001, - } + # For most tuya devices Up/Open = 0, Stop = 1, Down/Close = 2 + tuya_cover_command = { + WINDOW_COVER_COMMAND_UPOPEN: 0x0000, + WINDOW_COVER_COMMAND_DOWNCLOSE: 0x0002, + WINDOW_COVER_COMMAND_STOP: 0x0001, + } async def command( self, @@ -775,7 +870,10 @@ async def command( kwargs, ) + # Custom command processing first # Open Close or Stop commands + # TODO - consider using command to dp mapping for these (knowing that the attribute will be + # updated) when the device echos the dp values. if command_id in ( WINDOW_COVER_COMMAND_UPOPEN, WINDOW_COVER_COMMAND_DOWNCLOSE, @@ -794,60 +892,27 @@ async def command( TUYA_MCU_SET_CLUSTER_DATA, cluster_data, ) - return self._default_response(command_id) - # TODO - refactor command to DP mapping into something similar to _dp_2_attr_update & remove send_tuya_set_data_command or move it to TuyaMCUCluster + return self.default_response(command_id) + # TODO - Use command to dp mapping for lift percent, but need a way to call convert_lift_percent elif command_id == WINDOW_COVER_COMMAND_LIFTPERCENT: - self.send_tuya_set_data_command( + self.send_tuya_set_datapoints_command( TUYA_DP_ID_PERCENT_CONTROL, self.convert_lift_percent(args[0]), expect_reply=expect_reply, manufacturer=manufacturer, ) - return self._default_response(command_id) - elif command_id == WINDOW_COVER_COMMAND_SMALL_STEP: - self.send_tuya_set_data_command( - TUYA_DP_ID_SMALL_STEP, - kwargs["direction"], - expect_reply=expect_reply, - manufacturer=manufacturer, - ) - return self._default_response(command_id) - elif command_id == WINDOW_COVER_COMMAND_UPDATE_LIMITS: - self.send_tuya_set_data_command( - TUYA_DP_ID_LIMIT_SETTINGS, - kwargs["operation"], - expect_reply=expect_reply, - manufacturer=manufacturer, - ) - return self._default_response(command_id) - _LOGGER.warning("Unsupported command_id: %s", command_id) - return foundation.GENERAL_COMMANDS[ - foundation.GeneralCommand.Default_Response - ].schema(command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND) - - def send_tuya_set_data_command( - self, - dp: t.uint8_t, - data: TuyaData, - manufacturer: Optional[Union[int, t.uint16_t]] = None, - expect_reply: bool = True, - ): - """Send a set_data for a Tuya data point value (via the mcu cluster).""" - - datapoints = [TuyaDatapointData(dp, data)] - self.debug("Sending TUYA_MCU_SET_DATAPOINTS: %s", datapoints) + return self.default_response(command_id) - self.endpoint.device.command_bus.listener_event( - TUYA_MCU_SET_DATAPOINTS, datapoints, manufacturer, expect_reply - ) + # now let TuyaCommandCluster handle any remaining commands mapped to data points + return await super().command( + command_id, + *args, + manufacturer=manufacturer, + expect_reply=expect_reply, + tsn=tsn, + **kwargs + ) - # TODO - move this to TuyaMCUCluster - def _default_response( - self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] - ): - return foundation.GENERAL_COMMANDS[ - foundation.GeneralCommand.Default_Response - ].schema(command_id=command_id, status=foundation.Status.SUCCESS) def convert_lift_percent(self, input_value: int): """Convert/invert lift percent when needed. @@ -864,6 +929,7 @@ def convert_lift_percent(self, input_value: int): invert = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 return input_value if invert else 100 - input_value + class TuyaLevelControl(LevelControl, TuyaLocalCluster): """Tuya MCU Level cluster for dimmable device.""" From 785356c7f5bf97b6c2876988c85948f5009f1085 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Sat, 11 May 2024 06:53:26 +1000 Subject: [PATCH 07/14] Reorder tuya window cluster classes --- zhaquirks/tuya/mcu/__init__.py | 126 ++++++++++++++++----------------- 1 file changed, 63 insertions(+), 63 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 3a1920aa4b..875f9246a0 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -732,69 +732,6 @@ class MoesSwitchManufCluster(TuyaOnOffManufCluster): data_point_handlers.update({15: "_dp_2_attr_update"}) -class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): - """Manufacturer Specific Cluster for cover device (based on new TuyaMCUCluster). - - I.e. Uses newer TuyaMCUCluster mechanism for translations between cluster attributes and - Tuya data points. - """ - - # TODO - work out if I can do this as a class function, yet still access the invert attribute value - def __init__(self, *args, **kwargs): - """Init.""" - super().__init__(*args, **kwargs) - - self.dp_to_attribute: dict[int, DPToAttributeMapping] = { - TUYA_DP_ID_CONTROL: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_MAIN_CONTROL_NAME, - # Converting raw ints to a type allows the attributes UI show meaningful values - CoverMotorStatus, - CoverMotorStatus, - ), - TUYA_DP_ID_PERCENT_STATE: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_LIFTPERCENT_NAME, - self._convert_lift_percent, - self._convert_lift_percent, - ), - TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_DIRECTION_SETTING_NAME, - CoverSettingMotorDirection, - CoverSettingMotorDirection, - ), - TUYA_DP_ID_BATTERY_PERCENT: DPToAttributeMapping( - PowerConfiguration.ep_attribute, - PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, - # Tuya report real percent, zigbee expects value*2, but - # TuyaPowerConfigurationCluster will convert it - ), - } - - data_point_handlers = { - TUYA_DP_ID_CONTROL: "_dp_2_attr_update", - TUYA_DP_ID_PERCENT_STATE: "_dp_2_attr_update", - TUYA_DP_ID_DIRECTION_SETTING: "_dp_2_attr_update", - TUYA_DP_ID_BATTERY_PERCENT: "_dp_2_attr_update", - # Ignore updates from data points that are used as write-only commands to the device, we - # don't need attributes to display their values, but they're they're echoed back in - # get_data and would otherwise log debug messages. - TUYA_DP_ID_PERCENT_CONTROL: "ignore_update", - # I don't know what 7 is, but it's part of set_data_response - 7: "ignore_update", - TUYA_DP_ID_LIMIT_SETTINGS: "ignore_update", - TUYA_DP_ID_SMALL_STEP: "ignore_update", - } - - def _convert_lift_percent(self, input_value: int): - return self.endpoint.window_covering.convert_lift_percent(input_value) - - def ignore_update(self, _datapoint: TuyaDatapointData) -> None: - """Process (and ignore) some data point updates.""" - return None - - class TuyaNewWindowCoverControl(TuyaAttributesCluster, TuyaCommandCluster, WindowCovering): """Tuya Window Cover Cluster, based on new TuyaNewManufClusterForWindowCover. @@ -930,6 +867,69 @@ def convert_lift_percent(self, input_value: int): return input_value if invert else 100 - input_value +class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): + """Manufacturer Specific Cluster for cover device (based on new TuyaMCUCluster). + + I.e. Uses newer TuyaMCUCluster mechanism for translations between cluster attributes and + Tuya data points. + """ + + # TODO - work out if I can do this as a class function, yet still access the invert attribute value + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + + self.dp_to_attribute: dict[int, DPToAttributeMapping] = { + TUYA_DP_ID_CONTROL: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_MAIN_CONTROL_NAME, + # Converting raw ints to a type allows the attributes UI show meaningful values + CoverMotorStatus, + CoverMotorStatus, + ), + TUYA_DP_ID_PERCENT_STATE: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_LIFTPERCENT_NAME, + self._convert_lift_percent, + self._convert_lift_percent, + ), + TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_DIRECTION_SETTING_NAME, + CoverSettingMotorDirection, + CoverSettingMotorDirection, + ), + TUYA_DP_ID_BATTERY_PERCENT: DPToAttributeMapping( + PowerConfiguration.ep_attribute, + PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, + # Tuya report real percent, zigbee expects value*2, but + # TuyaPowerConfigurationCluster will convert it + ), + } + + data_point_handlers = { + TUYA_DP_ID_CONTROL: "_dp_2_attr_update", + TUYA_DP_ID_PERCENT_STATE: "_dp_2_attr_update", + TUYA_DP_ID_DIRECTION_SETTING: "_dp_2_attr_update", + TUYA_DP_ID_BATTERY_PERCENT: "_dp_2_attr_update", + # Ignore updates from data points that are used as write-only commands to the device, we + # don't need attributes to display their values, but they're they're echoed back in + # get_data and would otherwise log debug messages. + TUYA_DP_ID_PERCENT_CONTROL: "ignore_update", + # I don't know what 7 is, but it's part of set_data_response + 7: "ignore_update", + TUYA_DP_ID_LIMIT_SETTINGS: "ignore_update", + TUYA_DP_ID_SMALL_STEP: "ignore_update", + } + + def _convert_lift_percent(self, input_value: int): + return self.endpoint.window_covering.convert_lift_percent(input_value) + + def ignore_update(self, _datapoint: TuyaDatapointData) -> None: + """Process (and ignore) some data point updates.""" + return None + + class TuyaLevelControl(LevelControl, TuyaLocalCluster): """Tuya MCU Level cluster for dimmable device.""" From 699dd0597baefd7e0008701e7bb0fae855ee834b Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Fri, 17 May 2024 23:01:06 +1000 Subject: [PATCH 08/14] Refactor TuyaNewManufClusterForWindowCover to avoid needing object initialization/state --- zhaquirks/tuya/mcu/__init__.py | 81 +++++++++++++++++----------------- 1 file changed, 41 insertions(+), 40 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 875f9246a0..a40c09d848 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -834,7 +834,7 @@ async def command( elif command_id == WINDOW_COVER_COMMAND_LIFTPERCENT: self.send_tuya_set_datapoints_command( TUYA_DP_ID_PERCENT_CONTROL, - self.convert_lift_percent(args[0]), + self._compute_lift_percent(args[0]), expect_reply=expect_reply, manufacturer=manufacturer, ) @@ -850,8 +850,7 @@ async def command( **kwargs ) - - def convert_lift_percent(self, input_value: int): + def _compute_lift_percent(self, input_value: int): """Convert/invert lift percent when needed. HA shows % open. The zigbee cluster value is called 'lift_percent' but seems to need to @@ -867,6 +866,13 @@ def convert_lift_percent(self, input_value: int): return input_value if invert else 100 - input_value + def update_lift_percent(self, raw_value: int): + """Update lift percent attribute when it's data point data is received.""" + + new_attribute_value = self._compute_lift_percent(raw_value) + self.update_attribute(ATTR_COVER_LIFTPERCENT_NAME, new_attribute_value) + + class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): """Manufacturer Specific Cluster for cover device (based on new TuyaMCUCluster). @@ -874,47 +880,36 @@ class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): Tuya data points. """ - # TODO - work out if I can do this as a class function, yet still access the invert attribute value - def __init__(self, *args, **kwargs): - """Init.""" - super().__init__(*args, **kwargs) - - self.dp_to_attribute: dict[int, DPToAttributeMapping] = { - TUYA_DP_ID_CONTROL: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_MAIN_CONTROL_NAME, - # Converting raw ints to a type allows the attributes UI show meaningful values - CoverMotorStatus, - CoverMotorStatus, - ), - TUYA_DP_ID_PERCENT_STATE: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_LIFTPERCENT_NAME, - self._convert_lift_percent, - self._convert_lift_percent, - ), - TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( - TuyaNewWindowCoverControl.ep_attribute, - ATTR_COVER_DIRECTION_SETTING_NAME, - CoverSettingMotorDirection, - CoverSettingMotorDirection, - ), - TUYA_DP_ID_BATTERY_PERCENT: DPToAttributeMapping( - PowerConfiguration.ep_attribute, - PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, - # Tuya report real percent, zigbee expects value*2, but - # TuyaPowerConfigurationCluster will convert it - ), - } + dp_to_attribute: dict[int, DPToAttributeMapping] = { + TUYA_DP_ID_CONTROL: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_MAIN_CONTROL_NAME, + # Converting raw ints to a type allows the attributes UI show meaningful values + CoverMotorStatus, + CoverMotorStatus, + ), + TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( + TuyaNewWindowCoverControl.ep_attribute, + ATTR_COVER_DIRECTION_SETTING_NAME, + CoverSettingMotorDirection, + CoverSettingMotorDirection, + ), + TUYA_DP_ID_BATTERY_PERCENT: DPToAttributeMapping( + PowerConfiguration.ep_attribute, + PowerConfiguration.AttributeDefs.battery_percentage_remaining.name, + # Tuya report real percent, zigbee expects value*2, but + # TuyaPowerConfigurationCluster will convert it + ), + } data_point_handlers = { TUYA_DP_ID_CONTROL: "_dp_2_attr_update", - TUYA_DP_ID_PERCENT_STATE: "_dp_2_attr_update", + TUYA_DP_ID_PERCENT_STATE: "update_lift_percent", TUYA_DP_ID_DIRECTION_SETTING: "_dp_2_attr_update", TUYA_DP_ID_BATTERY_PERCENT: "_dp_2_attr_update", # Ignore updates from data points that are used as write-only commands to the device, we - # don't need attributes to display their values, but they're they're echoed back in - # get_data and would otherwise log debug messages. + # don't need attributes to display their values, but they're echoed back in get_data and + # would otherwise log debug messages. TUYA_DP_ID_PERCENT_CONTROL: "ignore_update", # I don't know what 7 is, but it's part of set_data_response 7: "ignore_update", @@ -922,8 +917,14 @@ def __init__(self, *args, **kwargs): TUYA_DP_ID_SMALL_STEP: "ignore_update", } - def _convert_lift_percent(self, input_value: int): - return self.endpoint.window_covering.convert_lift_percent(input_value) + def update_lift_percent(self, datapoint: TuyaDatapointData): + """Update lift percent attribute when it's data point data is received. + + This can't be done as a dp_to_attribute entry because it needs access to self which + dp_to_attribute callbacks don't have, but data_point_handlers do. + """ + cluster = self.endpoint.window_covering + cluster.update_lift_percent(datapoint.data.payload) def ignore_update(self, _datapoint: TuyaDatapointData) -> None: """Process (and ignore) some data point updates.""" From 9732e7a6ea71f919432686bd7b7baf5a7aecc44e Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Sat, 18 May 2024 06:29:43 +1000 Subject: [PATCH 09/14] Update command_to_dp map to handle WINDOW_COVER_COMMAND_LIFTPERCENT --- zhaquirks/tuya/mcu/__init__.py | 79 ++++++++++++++++++++++------------ 1 file changed, 52 insertions(+), 27 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index a40c09d848..839d3c6f15 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -4,7 +4,7 @@ import dataclasses import datetime import logging -from typing import Any, Optional, Union +from typing import Any, Optional, Protocol, Union from zigpy.quirks import CustomCluster import zigpy.types as t @@ -62,6 +62,7 @@ _LOGGER = logging.getLogger(__name__) + @dataclasses.dataclass class DPToAttributeMapping: """Container for datapoint to cluster attribute update mapping.""" @@ -87,12 +88,23 @@ class DPToAttributeMapping: endpoint_id: Optional[int] = None +class CommandToDPValueMappingCallback(Protocol): + """Protocol describing CommandToDPValueMapping callbacks.""" + + def __call__( + self, + command_id: Union[foundation.GeneralCommand, int, t.uint8_t], + *args, + **kwargs: Any) -> TuyaData: + """Call back with self, command id, ordered and named variable args.""" + + @dataclasses.dataclass class CommandToDPValueMapping: """Container for command id to datapoint value mapping.""" dp: t.uint8_t - value_source: Union[int, str, Callable[..., TuyaData]] + value_source: Union[int, str, CommandToDPValueMappingCallback] class TuyaClusterData(t.Struct): @@ -114,6 +126,7 @@ class MoesBacklight(t.enum8): light_when_off = 0x02 freeze = 0x03 + class CoverCommandStepDirection(t.enum8): """Window cover step command direction enum.""" @@ -198,8 +211,9 @@ class TuyaCommandCluster(CustomCluster): into data point updates, sent to the tuya mcu cluster to send a set data command to the device. """ - command_to_dp: dict[Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping] = { - } + command_to_dp: dict[ + Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping + ] = {} async def command( self, @@ -228,16 +242,19 @@ async def command( elif isinstance(command_map.value_source, str): value = kwargs[command_map.value_source] else: - value = command_map.value_source(*args, **kwargs) + value = command_map.value_source(self, command_id, *args, **kwargs) - self.send_tuya_set_datapoints_command(command_map.dp, value, expect_reply=expect_reply, - manufacturer=manufacturer) + self.send_tuya_set_datapoints_command( + command_map.dp, + value, + expect_reply=expect_reply, + manufacturer=manufacturer, + ) return self.default_response(command_id) _LOGGER.warning("Unsupported command_id: %s", command_id) return self.unsupported_response(command_id) - def send_tuya_set_datapoints_command( self, dp: t.uint8_t, @@ -254,7 +271,6 @@ def send_tuya_set_datapoints_command( TUYA_MCU_SET_DATAPOINTS, datapoints, manufacturer, expect_reply ) - def default_response( self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] ): @@ -264,7 +280,6 @@ def default_response( foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=foundation.Status.SUCCESS) - def unsupported_response( self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t] ): @@ -551,7 +566,7 @@ async def command( manufacturer: Optional[Union[int, t.uint16_t]] = None, expect_reply: bool = True, tsn: Optional[Union[int, t.uint8_t]] = None, - **_kwargs: Any + **_kwargs: Any, ): """Override the default Cluster command.""" @@ -732,7 +747,9 @@ class MoesSwitchManufCluster(TuyaOnOffManufCluster): data_point_handlers.update({15: "_dp_2_attr_update"}) -class TuyaNewWindowCoverControl(TuyaAttributesCluster, TuyaCommandCluster, WindowCovering): +class TuyaNewWindowCoverControl( + TuyaAttributesCluster, TuyaCommandCluster, WindowCovering +): """Tuya Window Cover Cluster, based on new TuyaNewManufClusterForWindowCover. Derive from TuyaLocalCluster to disable attribute writes, in a way that's compatible with @@ -777,9 +794,27 @@ class TuyaNewWindowCoverControl(TuyaAttributesCluster, TuyaCommandCluster, Windo } ) - command_to_dp: dict[Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping] = { - WINDOW_COVER_COMMAND_SMALL_STEP: CommandToDPValueMapping(TUYA_DP_ID_SMALL_STEP, "direction"), - WINDOW_COVER_COMMAND_UPDATE_LIMITS: CommandToDPValueMapping(TUYA_DP_ID_LIMIT_SETTINGS, "operation"), + def lift_percent_command_dp_value( + self, + _command_id: Union[foundation.GeneralCommand, int, t.uint8_t], + *args, + **_kwargs: Any, + ): + """Compute the tuya data point value to apply when given a set lift percent command.""" + return self._compute_lift_percent(args[0]) + + command_to_dp: dict[ + Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping + ] = { + WINDOW_COVER_COMMAND_LIFTPERCENT: CommandToDPValueMapping( + TUYA_DP_ID_PERCENT_CONTROL, lift_percent_command_dp_value + ), + WINDOW_COVER_COMMAND_SMALL_STEP: CommandToDPValueMapping( + TUYA_DP_ID_SMALL_STEP, "direction" + ), + WINDOW_COVER_COMMAND_UPDATE_LIMITS: CommandToDPValueMapping( + TUYA_DP_ID_LIMIT_SETTINGS, "operation" + ), } # For most tuya devices Up/Open = 0, Stop = 1, Down/Close = 2 @@ -830,15 +865,6 @@ async def command( cluster_data, ) return self.default_response(command_id) - # TODO - Use command to dp mapping for lift percent, but need a way to call convert_lift_percent - elif command_id == WINDOW_COVER_COMMAND_LIFTPERCENT: - self.send_tuya_set_datapoints_command( - TUYA_DP_ID_PERCENT_CONTROL, - self._compute_lift_percent(args[0]), - expect_reply=expect_reply, - manufacturer=manufacturer, - ) - return self.default_response(command_id) # now let TuyaCommandCluster handle any remaining commands mapped to data points return await super().command( @@ -847,8 +873,8 @@ async def command( manufacturer=manufacturer, expect_reply=expect_reply, tsn=tsn, - **kwargs - ) + **kwargs, + ) def _compute_lift_percent(self, input_value: int): """Convert/invert lift percent when needed. @@ -865,7 +891,6 @@ def _compute_lift_percent(self, input_value: int): invert = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 return input_value if invert else 100 - input_value - def update_lift_percent(self, raw_value: int): """Update lift percent attribute when it's data point data is received.""" From 8641f46bab0e78e3550284fa7cc80888e663a411 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Sat, 18 May 2024 07:03:40 +1000 Subject: [PATCH 10/14] Use command_to_dp to implement main window cover move commands --- zhaquirks/tuya/mcu/__init__.py | 135 +++++++++++++++------------------ 1 file changed, 61 insertions(+), 74 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index 839d3c6f15..bf74d11f90 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -134,8 +134,20 @@ class CoverCommandStepDirection(t.enum8): Close = 1 +class CoverMotorCommand(t.enum8): + """Window cover motor command states enum.""" + + Open = 0 + Stop = 1 + Close = 2 + + class CoverMotorStatus(t.enum8): - """Window cover motor states enum.""" + """Window cover motor states enum. + + Uses the same Tuya data point to send a command and receive the status, so needs the same + values as CoverMotorCommand. + """ Opening = 0 Stopped = 1 @@ -758,7 +770,7 @@ class TuyaNewWindowCoverControl( attributes = WindowCovering.attributes.copy() # main control attribute is logically write-only, only used by commands and not very useful - # to HA, but it's return in a set_data and set_data_response packet so I've mapped it to + # to HA, but it's returned in the set_data and set_data_response packets so I've mapped it to # an attribute. attributes.update( { @@ -794,29 +806,6 @@ class TuyaNewWindowCoverControl( } ) - def lift_percent_command_dp_value( - self, - _command_id: Union[foundation.GeneralCommand, int, t.uint8_t], - *args, - **_kwargs: Any, - ): - """Compute the tuya data point value to apply when given a set lift percent command.""" - return self._compute_lift_percent(args[0]) - - command_to_dp: dict[ - Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping - ] = { - WINDOW_COVER_COMMAND_LIFTPERCENT: CommandToDPValueMapping( - TUYA_DP_ID_PERCENT_CONTROL, lift_percent_command_dp_value - ), - WINDOW_COVER_COMMAND_SMALL_STEP: CommandToDPValueMapping( - TUYA_DP_ID_SMALL_STEP, "direction" - ), - WINDOW_COVER_COMMAND_UPDATE_LIMITS: CommandToDPValueMapping( - TUYA_DP_ID_LIMIT_SETTINGS, "operation" - ), - } - # For most tuya devices Up/Open = 0, Stop = 1, Down/Close = 2 tuya_cover_command = { WINDOW_COVER_COMMAND_UPOPEN: 0x0000, @@ -824,57 +813,36 @@ def lift_percent_command_dp_value( WINDOW_COVER_COMMAND_STOP: 0x0001, } - async def command( + def move_command_dp_value( self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t], + *_args, + **_kwargs: Any, + ): + """Translate the tuya data point value to use when commanded to move.""" + return CoverMotorCommand(self.tuya_cover_command[command_id]) + + + def lift_percent_command_dp_value( + self, + _command_id: Union[foundation.GeneralCommand, int, t.uint8_t], *args, - manufacturer: Optional[Union[int, t.uint16_t]] = None, - expect_reply: bool = True, - tsn: Optional[Union[int, t.uint8_t]] = None, - **kwargs: Any, + **_kwargs: Any, ): - """Override the default Cluster command.""" + """Compute the tuya data point value to apply when given a set lift percent command.""" + return self._compute_lift_percent(args[0]) - _LOGGER.debug( - "Sending Tuya Cluster Command... Cluster Command is %x, args=%s, kwargs=%s", - command_id, - args, - kwargs, - ) - # Custom command processing first - # Open Close or Stop commands - # TODO - consider using command to dp mapping for these (knowing that the attribute will be - # updated) when the device echos the dp values. - if command_id in ( - WINDOW_COVER_COMMAND_UPOPEN, - WINDOW_COVER_COMMAND_DOWNCLOSE, - WINDOW_COVER_COMMAND_STOP, - ): - cluster_data = TuyaClusterData( - endpoint_id=self.endpoint.endpoint_id, - cluster_name=self.ep_attribute, - cluster_attr=ATTR_COVER_MAIN_CONTROL_NAME, - # Map from zigbee command to tuya DP value - attr_value=self.tuya_cover_command[command_id], - expect_reply=expect_reply, - manufacturer=manufacturer, - ) - self.endpoint.device.command_bus.listener_event( - TUYA_MCU_SET_CLUSTER_DATA, - cluster_data, - ) - return self.default_response(command_id) + def update_lift_percent(self, raw_value: int): + """Update lift percent attribute when it's data point data is received. + + This can't be done as a dp_to_attribute entry in the mcu cluster because it needs access + to self. + """ + + new_attribute_value = self._compute_lift_percent(raw_value) + self.update_attribute(ATTR_COVER_LIFTPERCENT_NAME, new_attribute_value) - # now let TuyaCommandCluster handle any remaining commands mapped to data points - return await super().command( - command_id, - *args, - manufacturer=manufacturer, - expect_reply=expect_reply, - tsn=tsn, - **kwargs, - ) def _compute_lift_percent(self, input_value: int): """Convert/invert lift percent when needed. @@ -891,11 +859,30 @@ def _compute_lift_percent(self, input_value: int): invert = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 return input_value if invert else 100 - input_value - def update_lift_percent(self, raw_value: int): - """Update lift percent attribute when it's data point data is received.""" - new_attribute_value = self._compute_lift_percent(raw_value) - self.update_attribute(ATTR_COVER_LIFTPERCENT_NAME, new_attribute_value) + command_to_dp: dict[ + Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping + ] = { + + WINDOW_COVER_COMMAND_UPOPEN: CommandToDPValueMapping( + TUYA_DP_ID_CONTROL, move_command_dp_value + ), + WINDOW_COVER_COMMAND_DOWNCLOSE: CommandToDPValueMapping( + TUYA_DP_ID_CONTROL, move_command_dp_value + ), + WINDOW_COVER_COMMAND_STOP: CommandToDPValueMapping( + TUYA_DP_ID_CONTROL, move_command_dp_value + ), + WINDOW_COVER_COMMAND_LIFTPERCENT: CommandToDPValueMapping( + TUYA_DP_ID_PERCENT_CONTROL, lift_percent_command_dp_value + ), + WINDOW_COVER_COMMAND_SMALL_STEP: CommandToDPValueMapping( + TUYA_DP_ID_SMALL_STEP, "direction" + ), + WINDOW_COVER_COMMAND_UPDATE_LIMITS: CommandToDPValueMapping( + TUYA_DP_ID_LIMIT_SETTINGS, "operation" + ), + } class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): @@ -911,7 +898,7 @@ class TuyaNewManufClusterForWindowCover(TuyaMCUCluster): ATTR_COVER_MAIN_CONTROL_NAME, # Converting raw ints to a type allows the attributes UI show meaningful values CoverMotorStatus, - CoverMotorStatus, + CoverMotorCommand, ), TUYA_DP_ID_DIRECTION_SETTING: DPToAttributeMapping( TuyaNewWindowCoverControl.ep_attribute, From e25cb6ad2e1ee4524cb8eacd6fec8eea1923bfb7 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Sat, 18 May 2024 08:23:52 +1000 Subject: [PATCH 11/14] Fix async read_attributes override --- zhaquirks/tuya/mcu/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index bf74d11f90..aea42dc7cc 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -180,13 +180,13 @@ class TuyaPowerConfigurationCluster( class TuyaAttributesCluster(TuyaLocalCluster): """Manufacturer specific cluster for Tuya converting attributes <-> commands.""" - def read_attributes( + async def read_attributes( self, attributes, allow_cache=False, only_cache=False, manufacturer=None ): """Ignore remote reads as the "get_data" command doesn't seem to do anything.""" self.debug("read_attributes --> attrs: %s", attributes) - return super().read_attributes( + return await super().read_attributes( attributes, allow_cache=True, only_cache=True, manufacturer=manufacturer ) From f446e6658d2737091661cb9d749b1f59831661c5 Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Wed, 22 May 2024 21:19:01 +1000 Subject: [PATCH 12/14] Removed unused import --- zhaquirks/tuya/ts0601_cover.py | 1 - 1 file changed, 1 deletion(-) diff --git a/zhaquirks/tuya/ts0601_cover.py b/zhaquirks/tuya/ts0601_cover.py index 239a868c82..fb550f20f7 100644 --- a/zhaquirks/tuya/ts0601_cover.py +++ b/zhaquirks/tuya/ts0601_cover.py @@ -3,7 +3,6 @@ import logging from zigpy.profiles import zha -from zigpy.quirks import _DEVICE_REGISTRY from zigpy.quirks.v2 import add_to_registry_v2 from zigpy.quirks.v2.homeassistant import EntityType from zigpy.zcl.clusters.general import Basic, Groups, Identify, OnOff, Ota, Scenes, Time From 51a30f0516d6fe95892d1cb9bb68f84db2529c4e Mon Sep 17 00:00:00 2001 From: Matt Sullivan <1115325+matt-sullivan@users.noreply.github.com> Date: Wed, 30 Oct 2024 00:04:18 +1000 Subject: [PATCH 13/14] Update tuya cover to latest quirks v2 Update tests to use new zigpy request kwarg format --- tests/test_tuya_cover.py | 43 ++++++++++++++++++++++------------ zhaquirks/tuya/ts0601_cover.py | 22 ++++++++--------- 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/tests/test_tuya_cover.py b/tests/test_tuya_cover.py index d894d9f038..6dc55fa5a8 100644 --- a/tests/test_tuya_cover.py +++ b/tests/test_tuya_cover.py @@ -3,6 +3,7 @@ from unittest import mock import pytest +from zigpy.types import PacketPriority from zigpy.zcl import foundation from tests.common import ClusterListener, wait_for_zigpy_tasks @@ -96,17 +97,21 @@ async def test_cover_move_commands( assert len(tuya_listener.attribute_updates) == 0 with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True ) as m1: rsp = await cover_cluster.command(command, *args, **kwargs) await wait_for_zigpy_tasks() m1.assert_called_with( - 0xEF00, - 1, - expected_frame, - expect_reply=True, + cluster=0xEF00, + sequence=1, + data=expected_frame, command_id=0, + timeout=5, + expect_reply=True, + use_ieee=False, + ask_for_ack=None, + priority=PacketPriority.NORMAL ) assert rsp.status == foundation.Status.SUCCESS @@ -238,17 +243,21 @@ async def test_cover_attributes_set( cover_cluster = device.endpoints[1].window_covering with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True ) as m1: write_results = await cover_cluster.write_attributes({name: value}) await wait_for_zigpy_tasks() m1.assert_called_with( - 0xEF00, - 1, - expected_frame, - expect_reply=False, + cluster=0xEF00, + sequence=1, + data=expected_frame, command_id=0, + timeout=5, + expect_reply=False, + use_ieee=False, + ask_for_ack=None, + priority=PacketPriority.NORMAL ) assert write_results == [ [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)] @@ -303,17 +312,21 @@ async def test_cover_invert( # Now send a command to set that value and assert we send the frame we expect with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS + tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True ) as m1: rsp = await cover_cluster.command(0x05, expected_received_value) await wait_for_zigpy_tasks() m1.assert_called_with( - 0xEF00, - 1, - expected_sent_frame, - expect_reply=True, + cluster=0xEF00, + sequence=1, + data=expected_sent_frame, command_id=0, + timeout=5, + expect_reply=True, + use_ieee=False, + ask_for_ack=None, + priority=PacketPriority.NORMAL ) assert rsp.status == foundation.Status.SUCCESS diff --git a/zhaquirks/tuya/ts0601_cover.py b/zhaquirks/tuya/ts0601_cover.py index fb550f20f7..88643905f8 100644 --- a/zhaquirks/tuya/ts0601_cover.py +++ b/zhaquirks/tuya/ts0601_cover.py @@ -3,7 +3,7 @@ import logging from zigpy.profiles import zha -from zigpy.quirks.v2 import add_to_registry_v2 +from zigpy.quirks.v2 import QuirkBuilder from zigpy.quirks.v2.homeassistant import EntityType from zigpy.zcl.clusters.general import Basic, Groups, Identify, OnOff, Ota, Scenes, Time @@ -637,10 +637,7 @@ class TuyaCloneCover0601(TuyaWindowCover): } } - -def register_v2_quirks() -> None: - """Register v2 Quirks.""" - +( # Tuya window cover device. # # This variant supports: @@ -653,7 +650,7 @@ def register_v2_quirks() -> None: # TuyaNewManufClusterForWindowCover which can handle multiple updates in one zigby frame. # # "_TZE200_eevqq1uv", "TS0601" - Zemismart ZM25R3 roller blind motor - add_to_registry_v2("_TZE200_68nvbio9", "TS0601" + QuirkBuilder("_TZE200_68nvbio9", "TS0601" ).also_applies_to("_TZE200_eevqq1uv", "TS0601" ).replaces(TuyaNewManufClusterForWindowCover ).adds(TuyaNewWindowCoverControl @@ -665,6 +662,7 @@ def register_v2_quirks() -> None: {"direction": CoverCommandStepDirection.Open}, entity_type=EntityType.STANDARD, translation_key="small_step_open", + fallback_name="Small step open", ).command_button( WINDOW_COVER_COMMAND_SMALL_STEP_NAME, TuyaNewWindowCoverControl.cluster_id, @@ -672,38 +670,40 @@ def register_v2_quirks() -> None: {"direction": CoverCommandStepDirection.Close}, entity_type=EntityType.STANDARD, translation_key="small_step_close", + fallback_name="Small step close", ).enum( ATTR_COVER_DIRECTION_SETTING_NAME, CoverSettingMotorDirection, TuyaNewWindowCoverControl.cluster_id, translation_key="motor_direction", + fallback_name="Motor direction", ).command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.SetOpen}, translation_key="set_open_limit", + fallback_name="Set open limit", ).command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.SetClose}, translation_key="set_close_limit", + fallback_name="Set close limit", ).command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.ClearOpen}, translation_key="clear_open_limit", + fallback_name="Clear open limit", ).command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.ClearClose}, translation_key="clear_close_limit", - ) - - -( - register_v2_quirks() + fallback_name="Clear close limit", + ).add_to_registry() ) From ad529f3024869e2f7d0e2579e113febd67e01fba Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 29 Oct 2024 14:31:26 +0000 Subject: [PATCH 14/14] Apply pre-commit auto fixes --- tests/test_tuya_cover.py | 21 +++++++++++++++------ zhaquirks/tuya/__init__.py | 10 +++++++--- zhaquirks/tuya/mcu/__init__.py | 8 ++------ zhaquirks/tuya/ts0601_cover.py | 34 +++++++++++++++++++++------------- 4 files changed, 45 insertions(+), 28 deletions(-) diff --git a/tests/test_tuya_cover.py b/tests/test_tuya_cover.py index 6dc55fa5a8..ab8ddacd8d 100644 --- a/tests/test_tuya_cover.py +++ b/tests/test_tuya_cover.py @@ -97,7 +97,10 @@ async def test_cover_move_commands( assert len(tuya_listener.attribute_updates) == 0 with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True + tuya_cluster.endpoint, + "request", + return_value=foundation.Status.SUCCESS, + autospec=True, ) as m1: rsp = await cover_cluster.command(command, *args, **kwargs) @@ -111,7 +114,7 @@ async def test_cover_move_commands( expect_reply=True, use_ieee=False, ask_for_ack=None, - priority=PacketPriority.NORMAL + priority=PacketPriority.NORMAL, ) assert rsp.status == foundation.Status.SUCCESS @@ -243,7 +246,10 @@ async def test_cover_attributes_set( cover_cluster = device.endpoints[1].window_covering with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True + tuya_cluster.endpoint, + "request", + return_value=foundation.Status.SUCCESS, + autospec=True, ) as m1: write_results = await cover_cluster.write_attributes({name: value}) @@ -257,7 +263,7 @@ async def test_cover_attributes_set( expect_reply=False, use_ieee=False, ask_for_ack=None, - priority=PacketPriority.NORMAL + priority=PacketPriority.NORMAL, ) assert write_results == [ [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)] @@ -312,7 +318,10 @@ async def test_cover_invert( # Now send a command to set that value and assert we send the frame we expect with mock.patch.object( - tuya_cluster.endpoint, "request", return_value=foundation.Status.SUCCESS, autospec=True + tuya_cluster.endpoint, + "request", + return_value=foundation.Status.SUCCESS, + autospec=True, ) as m1: rsp = await cover_cluster.command(0x05, expected_received_value) @@ -326,7 +335,7 @@ async def test_cover_invert( expect_reply=True, use_ieee=False, ask_for_ack=None, - priority=PacketPriority.NORMAL + priority=PacketPriority.NORMAL, ) assert rsp.status == foundation.Status.SUCCESS diff --git a/zhaquirks/tuya/__init__.py b/zhaquirks/tuya/__init__.py index 8719d3b192..d8bce6f6cd 100644 --- a/zhaquirks/tuya/__init__.py +++ b/zhaquirks/tuya/__init__.py @@ -1241,8 +1241,12 @@ class TuyaWindowCoverControl(LocalDataCluster, WindowCovering): # Add additional attributes for direction attributes = WindowCovering.attributes.copy() - attributes.update({ATTR_COVER_DIRECTION_SETTING: (ATTR_COVER_DIRECTION_SETTING_NAME, t.enum8)}) - attributes.update({ATTR_COVER_INVERTED_SETTING: (ATTR_COVER_INVERTED_SETTING_NAME, t.Bool)}) + attributes.update( + {ATTR_COVER_DIRECTION_SETTING: (ATTR_COVER_DIRECTION_SETTING_NAME, t.enum8)} + ) + attributes.update( + {ATTR_COVER_INVERTED_SETTING: (ATTR_COVER_INVERTED_SETTING_NAME, t.Bool)} + ) def __init__(self, *args, **kwargs): """Initialize instance.""" @@ -1359,7 +1363,7 @@ class TuyaWindowCover(CustomDevice): tuya_cover_command = { WINDOW_COVER_COMMAND_UPOPEN: 0x0000, WINDOW_COVER_COMMAND_DOWNCLOSE: 0x0002, - WINDOW_COVER_COMMAND_STOP: 0x0001 + WINDOW_COVER_COMMAND_STOP: 0x0001, } # For all covers which need their position inverted by default diff --git a/zhaquirks/tuya/mcu/__init__.py b/zhaquirks/tuya/mcu/__init__.py index aea42dc7cc..59e296e96d 100644 --- a/zhaquirks/tuya/mcu/__init__.py +++ b/zhaquirks/tuya/mcu/__init__.py @@ -95,7 +95,8 @@ def __call__( self, command_id: Union[foundation.GeneralCommand, int, t.uint8_t], *args, - **kwargs: Any) -> TuyaData: + **kwargs: Any, + ) -> TuyaData: """Call back with self, command id, ordered and named variable args.""" @@ -822,7 +823,6 @@ def move_command_dp_value( """Translate the tuya data point value to use when commanded to move.""" return CoverMotorCommand(self.tuya_cover_command[command_id]) - def lift_percent_command_dp_value( self, _command_id: Union[foundation.GeneralCommand, int, t.uint8_t], @@ -832,7 +832,6 @@ def lift_percent_command_dp_value( """Compute the tuya data point value to apply when given a set lift percent command.""" return self._compute_lift_percent(args[0]) - def update_lift_percent(self, raw_value: int): """Update lift percent attribute when it's data point data is received. @@ -843,7 +842,6 @@ def update_lift_percent(self, raw_value: int): new_attribute_value = self._compute_lift_percent(raw_value) self.update_attribute(ATTR_COVER_LIFTPERCENT_NAME, new_attribute_value) - def _compute_lift_percent(self, input_value: int): """Convert/invert lift percent when needed. @@ -859,11 +857,9 @@ def _compute_lift_percent(self, input_value: int): invert = self._attr_cache.get(ATTR_COVER_INVERTED_SETTING) == 1 return input_value if invert else 100 - input_value - command_to_dp: dict[ Union[foundation.GeneralCommand, int, t.uint8_t], CommandToDPValueMapping ] = { - WINDOW_COVER_COMMAND_UPOPEN: CommandToDPValueMapping( TUYA_DP_ID_CONTROL, move_command_dp_value ), diff --git a/zhaquirks/tuya/ts0601_cover.py b/zhaquirks/tuya/ts0601_cover.py index 88643905f8..5357ab0056 100644 --- a/zhaquirks/tuya/ts0601_cover.py +++ b/zhaquirks/tuya/ts0601_cover.py @@ -637,6 +637,7 @@ class TuyaCloneCover0601(TuyaWindowCover): } } + ( # Tuya window cover device. # @@ -650,12 +651,12 @@ class TuyaCloneCover0601(TuyaWindowCover): # TuyaNewManufClusterForWindowCover which can handle multiple updates in one zigby frame. # # "_TZE200_eevqq1uv", "TS0601" - Zemismart ZM25R3 roller blind motor - QuirkBuilder("_TZE200_68nvbio9", "TS0601" - ).also_applies_to("_TZE200_eevqq1uv", "TS0601" - ).replaces(TuyaNewManufClusterForWindowCover - ).adds(TuyaNewWindowCoverControl - ).adds(TuyaPowerConfigurationCluster - ).command_button( + QuirkBuilder("_TZE200_68nvbio9", "TS0601") + .also_applies_to("_TZE200_eevqq1uv", "TS0601") + .replaces(TuyaNewManufClusterForWindowCover) + .adds(TuyaNewWindowCoverControl) + .adds(TuyaPowerConfigurationCluster) + .command_button( WINDOW_COVER_COMMAND_SMALL_STEP_NAME, TuyaNewWindowCoverControl.cluster_id, None, @@ -663,7 +664,8 @@ class TuyaCloneCover0601(TuyaWindowCover): entity_type=EntityType.STANDARD, translation_key="small_step_open", fallback_name="Small step open", - ).command_button( + ) + .command_button( WINDOW_COVER_COMMAND_SMALL_STEP_NAME, TuyaNewWindowCoverControl.cluster_id, None, @@ -671,39 +673,45 @@ class TuyaCloneCover0601(TuyaWindowCover): entity_type=EntityType.STANDARD, translation_key="small_step_close", fallback_name="Small step close", - ).enum( + ) + .enum( ATTR_COVER_DIRECTION_SETTING_NAME, CoverSettingMotorDirection, TuyaNewWindowCoverControl.cluster_id, translation_key="motor_direction", fallback_name="Motor direction", - ).command_button( + ) + .command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.SetOpen}, translation_key="set_open_limit", fallback_name="Set open limit", - ).command_button( + ) + .command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.SetClose}, translation_key="set_close_limit", fallback_name="Set close limit", - ).command_button( + ) + .command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.ClearOpen}, translation_key="clear_open_limit", fallback_name="Clear open limit", - ).command_button( + ) + .command_button( WINDOW_COVER_COMMAND_UPDATE_LIMITS_NAME, TuyaNewWindowCoverControl.cluster_id, None, {"operation": CoverSettingLimitOperation.ClearClose}, translation_key="clear_close_limit", fallback_name="Clear close limit", - ).add_to_registry() + ) + .add_to_registry() )