diff --git a/tests/test_xiaomi.py b/tests/test_xiaomi.py index 811a66746a..5dd4ee41c4 100644 --- a/tests/test_xiaomi.py +++ b/tests/test_xiaomi.py @@ -8,7 +8,7 @@ import pytest import zigpy.device import zigpy.types as t -from zigpy.zcl import foundation +from zigpy.zcl import Cluster, foundation from zigpy.zcl.clusters.closures import WindowCovering from zigpy.zcl.clusters.general import ( AnalogInput, @@ -1388,15 +1388,12 @@ async def test_xiaomi_power_cluster_not_used(zigpy_device_from_quirk, caplog, qu ) -@pytest.mark.parametrize( - "quirk", (zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ,) -) -async def test_xiaomi_e1_roller_curtain_battery(zigpy_device_from_quirk, quirk): +def test_xiaomi_e1_roller_curtain_battery(zigpy_device_from_v2_quirk): """Test Aqara E1 roller curtain battery reporting.""" # Ideally, get a real Xiaomi "heartbeat" message to test. # For now, fake the heartbeat message and check if battery parsing works. - device = zigpy_device_from_quirk(quirk) + device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002") basic_cluster = device.endpoints[1].basic ClusterListener(basic_cluster) @@ -1588,33 +1585,91 @@ async def test_xiaomi_e1_driver_light_level( @pytest.mark.parametrize( - "command, value", + "command, value, read_current_position", [ - (WindowCovering.ServerCommandDefs.up_open.id, 1), - (WindowCovering.ServerCommandDefs.down_close.id, 0), - (WindowCovering.ServerCommandDefs.stop.id, 2), + (WindowCovering.ServerCommandDefs.up_open.id, 1, False), + (WindowCovering.ServerCommandDefs.down_close.id, 0, False), + (WindowCovering.ServerCommandDefs.stop.id, 2, True), ], ) -async def test_xiaomi_e1_roller_commands_1(zigpy_device_from_quirk, command, value): +async def test_xiaomi_e1_roller_commands_1( + zigpy_device_from_v2_quirk, command, value, read_current_position +): """Test Aqara E1 roller commands for basic movement functions using MultistateOutput Cluster.""" - device = zigpy_device_from_quirk( - zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ - ) + device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002") window_covering_cluster = device.endpoints[1].window_covering + window_covering_listener = ClusterListener(window_covering_cluster) + window_covering_attr_id = ( + WindowCovering.AttributeDefs.current_position_lift_percentage.id + ) + + analog_cluster = device.endpoints[1].analog_output + analog_attr_id = AnalogOutput.AttributeDefs.present_value.id + multistate_cluster = device.endpoints[1].multistate_output - multistate_cluster._write_attributes = mock.AsyncMock( - return_value=( - [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)], - ) + multistate_attr_id = MultistateOutput.AttributeDefs.present_value.id + + # fake read response for attributes: return 1 for all attributes + def mock_read(attributes, manufacturer=None): + records = [ + foundation.ReadAttributeRecord( + attr, foundation.Status.SUCCESS, foundation.TypeValue(None, 1) + ) + for attr in attributes + ] + return (records,) + + # patch read commands + patch_window_covering_read = mock.patch.object( + window_covering_cluster, + "_read_attributes", + mock.AsyncMock(side_effect=mock_read), + ) + patch_analog_read = mock.patch.object( + analog_cluster, "_read_attributes", mock.AsyncMock(side_effect=mock_read) ) - attr_id = MultistateOutput.AttributeDefs.present_value.id - # test command - await window_covering_cluster.command(command) - assert multistate_cluster._write_attributes.call_count == 1 - assert multistate_cluster._write_attributes.call_args[0][0][0].attrid == attr_id - assert multistate_cluster._write_attributes.call_args[0][0][0].value.value == value + # patch write commands + patch_multistate_write = mock.patch.object( + multistate_cluster, + "_write_attributes", + mock.AsyncMock( + return_value=( + [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)], + ) + ), + ) + + with ( + patch_window_covering_read, + patch_analog_read, + patch_multistate_write, + ): + # test command + await window_covering_cluster.command(command) + assert multistate_cluster._write_attributes.call_count == 1 + assert ( + multistate_cluster._write_attributes.call_args[0][0][0].attrid + == multistate_attr_id + ) + assert ( + multistate_cluster._write_attributes.call_args[0][0][0].value.value == value + ) + if read_current_position: + # confirm the window covering cluster read was redirected + assert len(window_covering_cluster._read_attributes.mock_calls) == 0 + # confirm the analog output read occurs + assert len(analog_cluster._read_attributes.mock_calls) == 1 + assert analog_cluster._read_attributes.mock_calls[0][1][0] == [ + analog_attr_id + ] + assert window_covering_listener.attribute_updates[0] == ( + window_covering_attr_id, + 100 - 1, + ) # confirm the position was updated on the ZCL WindowCovering cluster + else: + assert len(analog_cluster._read_attributes.mock_calls) == 0 @pytest.mark.parametrize( @@ -1623,27 +1678,214 @@ async def test_xiaomi_e1_roller_commands_1(zigpy_device_from_quirk, command, val (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id, 60), ], ) -async def test_xiaomi_e1_roller_commands_2(zigpy_device_from_quirk, command, value): +async def test_xiaomi_e1_roller_commands_2(zigpy_device_from_v2_quirk, command, value): """Test Aqara E1 roller commands for go to lift percentage using AnalogOutput cluster.""" - device = zigpy_device_from_quirk( - zhaquirks.xiaomi.aqara.roller_curtain_e1.RollerE1AQ - ) + device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002") window_covering_cluster = device.endpoints[1].window_covering + window_covering_listener = ClusterListener(window_covering_cluster) + analog_cluster = device.endpoints[1].analog_output - analog_cluster._write_attributes = mock.AsyncMock( - return_value=( - [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)], + analog_listener = ClusterListener(analog_cluster) + analog_attr_id = AnalogOutput.AttributeDefs.present_value.id + + # patch write commands + patch_analog_write = mock.patch.object( + analog_cluster, + "_write_attributes", + mock.AsyncMock( + return_value=( + [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)], + ) + ), + ) + + with ( + patch_analog_write, + ): + # test go to lift percentage command + await window_covering_cluster.go_to_lift_percentage(value) + assert analog_cluster._write_attributes.call_count == 1 + assert ( + analog_cluster._write_attributes.call_args[0][0][0].attrid == analog_attr_id ) + assert ( + analog_cluster._write_attributes.call_args[0][0][0].value.value + == 100 - value + ) + assert ( + len(window_covering_listener.attribute_updates) == 0 + ) # confirm the AnalogOutput write did not update the current WindowCovering position + assert analog_listener.attribute_updates[0] == ( + analog_attr_id, + 100 - value, + ) # confirm the AnalogOutput present_value was updated + + +@pytest.mark.parametrize( + "attribute_id, device_value, converted_value", + [ + ( + zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id, + zhaquirks.xiaomi.aqara.roller_curtain_e1.AqaraRollerDriverCharging.true, + t.Bool.true, + ), + ( + zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id, + zhaquirks.xiaomi.aqara.roller_curtain_e1.AqaraRollerDriverCharging.false, + t.Bool.false, + ), + ( + zhaquirks.xiaomi.aqara.roller_curtain_e1.XiaomiAqaraRollerE1.AttributeDefs.charging.id, + 3, + None, + ), + ], +) +async def test_xiaomi_e1_roller_bool_value_conversion( + zigpy_device_from_v2_quirk, attribute_id, device_value, converted_value +): + """Test remap of Aqara values to binary_sensor bool values.""" + device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002") + + opple_cluster = device.endpoints[1].opple_cluster + opple_listener = ClusterListener(opple_cluster) + + # update aqara attribute + opple_cluster.update_attribute(attribute_id, device_value) + if converted_value is None: + assert len(opple_listener.attribute_updates) == 0 + return + assert len(opple_listener.attribute_updates) == 1 + + # confirm the aqara device value has been remapped to a t.Bool state + assert opple_listener.attribute_updates[0][0] == attribute_id + assert opple_listener.attribute_updates[0][1] == converted_value + + +@pytest.mark.parametrize( + "attr, expected_value, target_attr, target_cluster", + [ + ( + WindowCovering.AttributeDefs.current_position_lift_percentage, + 99, + AnalogOutput.AttributeDefs.present_value, + AnalogOutput, + ), # Redirect with read success + ( + WindowCovering.AttributeDefs.current_position_lift_percentage, + None, + AnalogOutput.AttributeDefs.present_value, + AnalogOutput, + ), # Redirect with read failure + ( + WindowCovering.AttributeDefs.config_status, + 1, + None, + None, + ), # Regular read success + ( + WindowCovering.AttributeDefs.config_status, + None, + None, + None, + ), # Regular read failure + ], +) +async def test_xiaomi_e1_roller_window_covering_read_redirection( + zigpy_device_from_v2_quirk, + attr: foundation.ZCLAttributeDef, + expected_value: int | None, + target_attr: foundation.ZCLAttributeDef | None, + target_cluster: Cluster | None, +): + """Test Aqara E1 roller WindowCovering attribute read redirection.""" + device = zigpy_device_from_v2_quirk(LUMI, "lumi.curtain.acn002") + + window_covering_cluster = device.endpoints[1].window_covering + window_covering_listener = ClusterListener(window_covering_cluster) + + redirect = False + if target_attr and target_cluster: + target_cluster = getattr(device.endpoints[1], target_cluster.ep_attribute) + redirect = True + + # fake read response for attributes, a value of 1 is returned if expected_value is not None + def mock_read(attributes, manufacturer=None): + records = [ + foundation.ReadAttributeRecord( + attr, + foundation.Status.SUCCESS + if expected_value + else foundation.Status.FAILURE, + foundation.TypeValue(None, 1), + ) + for attr in attributes + ] + return (records,) + + # patch window covering read command + patch_window_covering_read = mock.patch.object( + window_covering_cluster, + "_read_attributes", + mock.AsyncMock(side_effect=(mock_read)), ) - attr_id = AnalogOutput.AttributeDefs.present_value.id - # test go to lift percentage command - await window_covering_cluster.go_to_lift_percentage(value) - assert analog_cluster._write_attributes.call_count == 1 - assert analog_cluster._write_attributes.call_args[0][0][0].attrid == attr_id - assert ( - analog_cluster._write_attributes.call_args[0][0][0].value.value == 100 - value + if redirect: + # patch target cluster read command + patch_target_read = mock.patch.object( + target_cluster, + "_read_attributes", + mock.AsyncMock(side_effect=(mock_read)), + ) + with ( + patch_window_covering_read, + patch_target_read, + ): + # read attribute from WindowCovering cluster using id and name + await window_covering_cluster.read_attributes([attr.id]) + await window_covering_cluster.read_attributes([attr.name]) + + # confirm the reads were redirected to the target cluster + assert len(window_covering_cluster._read_attributes.mock_calls) == 0 + assert len(target_cluster._read_attributes.mock_calls) == 2 + assert target_cluster._read_attributes.mock_calls[0][1][0] == [ + target_attr.id + ] + assert target_cluster._read_attributes.mock_calls[1][1][0] == [ + target_attr.id + ] + else: + with ( + patch_window_covering_read, + ): + # read attribute from WindowCovering cluster using id and name + await window_covering_cluster.read_attributes([attr.id]) + await window_covering_cluster.read_attributes([attr.name]) + + # confirm the reads occurred normally + assert len(window_covering_cluster._read_attributes.mock_calls) == 2 + assert window_covering_cluster._read_attributes.mock_calls[0][1][0] == [ + attr.id + ] + assert window_covering_cluster._read_attributes.mock_calls[1][1][0] == [ + attr.id + ] + + if not expected_value: + # check read fails do not trigger an attribute update + assert len(window_covering_listener.attribute_updates) == 0 + return + + # check the WindowCovering attribute was updated by the reads + assert len(window_covering_listener.attribute_updates) == 2 + assert window_covering_listener.attribute_updates[0] == ( + attr.id, + expected_value, + ) + assert window_covering_listener.attribute_updates[1] == ( + attr.id, + expected_value, ) @@ -1774,11 +2016,11 @@ async def test_aqara_fp1e_sensor( expected_motion_status, ): """Test Aqara FP1E sensor.""" - quirk = zigpy_device_from_v2_quirk("aqara", "lumi.sensor_occupy.agl1") + device = zigpy_device_from_v2_quirk("aqara", "lumi.sensor_occupy.agl1") - opple_cluster = quirk.endpoints[1].opple_cluster - ias_cluster = quirk.endpoints[1].ias_zone - occupancy_cluster = quirk.endpoints[1].occupancy + opple_cluster = device.endpoints[1].opple_cluster + ias_cluster = device.endpoints[1].ias_zone + occupancy_cluster = device.endpoints[1].occupancy opple_listener = ClusterListener(opple_cluster) ias_listener = ClusterListener(ias_cluster) @@ -1809,15 +2051,15 @@ async def test_aqara_fp1e_sensor( def test_h1_wireless_remotes(zigpy_device_from_v2_quirk): """Test Aqara H1 wireless remote quirk adds missing endpoints.""" # create device with endpoint 1 only and verify we don't get a KeyError - quirk = zigpy_device_from_v2_quirk(LUMI, "lumi.remote.b28ac1") + device = zigpy_device_from_v2_quirk(LUMI, "lumi.remote.b28ac1") # verify the quirk adds endpoints 2 and 3 - assert 2 in quirk.endpoints - assert 3 in quirk.endpoints + assert 2 in device.endpoints + assert 3 in device.endpoints # verify the quirk adds the correct clusters to the new endpoints - assert OnOff.cluster_id in quirk.endpoints[2].out_clusters - assert OnOff.cluster_id in quirk.endpoints[3].out_clusters + assert OnOff.cluster_id in device.endpoints[2].out_clusters + assert OnOff.cluster_id in device.endpoints[3].out_clusters - assert MultistateInput.cluster_id in quirk.endpoints[2].in_clusters - assert MultistateInput.cluster_id in quirk.endpoints[3].in_clusters + assert MultistateInput.cluster_id in device.endpoints[2].in_clusters + assert MultistateInput.cluster_id in device.endpoints[3].in_clusters diff --git a/zhaquirks/xiaomi/aqara/roller_curtain_e1.py b/zhaquirks/xiaomi/aqara/roller_curtain_e1.py index d59f9b1cc1..a6aae3c6c1 100644 --- a/zhaquirks/xiaomi/aqara/roller_curtain_e1.py +++ b/zhaquirks/xiaomi/aqara/roller_curtain_e1.py @@ -2,64 +2,212 @@ from __future__ import annotations -from typing import Any +from collections.abc import Callable +from typing import Any, Final from zigpy import types as t -from zigpy.profiles import zgp, zha -from zigpy.zcl import foundation +from zigpy.quirks.v2 import QuirkBuilder +from zigpy.quirks.v2.homeassistant.binary_sensor import BinarySensorDeviceClass +from zigpy.zcl import Cluster, foundation from zigpy.zcl.clusters.closures import WindowCovering -from zigpy.zcl.clusters.general import ( - Alarms, - AnalogOutput, - Basic, - DeviceTemperature, - GreenPowerProxy, - Groups, - Identify, - MultistateOutput, - OnOff, - Ota, - Scenes, - Time, -) +from zigpy.zcl.clusters.general import AnalogOutput, MultistateOutput, OnOff +from zigpy.zcl.foundation import BaseAttributeDefs, DataTypeId, ZCLAttributeDef from zhaquirks import CustomCluster -from zhaquirks.const import ( - DEVICE_TYPE, - ENDPOINTS, - INPUT_CLUSTERS, - MODELS_INFO, - OUTPUT_CLUSTERS, - PROFILE_ID, -) from zhaquirks.xiaomi import ( LUMI, BasicCluster, XiaomiAqaraE1Cluster, - XiaomiCluster, - XiaomiCustomDevice, XiaomiPowerConfigurationPercent, ) +class AqaraRollerDriverCharging(t.enum8): + """Aqara roller driver charging status attribute values.""" + + true = 0x01 + false = 0x02 + + +class AqaraRollerDriverSpeed(t.enum8): + """Aqara roller driver speed attribute values.""" + + Low = 0x00 + Medium = 0x01 + High = 0x02 + + +class AqaraRollerState(t.enum8): + """Aqara roller state values.""" + + Closing = 0x00 + Opening = 0x01 + Stopped = 0x02 + Blocked = 0x03 + + +class RedirectAttributes: + """Methods for redirecting attribute reads to another cluster.""" + + _REDIRECT_ATTRIBUTES: ( + dict[ZCLAttributeDef, tuple[ZCLAttributeDef, Cluster, Callable | None]] | None + ) = None + + async def read_attributes( + self, + attributes: list[int | str], + allow_cache: bool = False, + only_cache: bool = False, + manufacturer: int | t.uint16_t | None = None, + ): + """Redirect attribute reads to another cluster.""" + + successful, failed = {}, {} + remaining_attributes = attributes.copy() + redirect_attributes = self._REDIRECT_ATTRIBUTES or {} + + for attr in redirect_attributes: + if attr.id not in attributes and attr.name not in attributes: + continue + if attr.id in attributes: + remaining_attributes.remove(attr.id) + if attr.name in attributes: + remaining_attributes.remove(attr.name) + + target_attr, target_cluster, format_func = redirect_attributes[attr] + result_s, result_f = await getattr( + self.endpoint, target_cluster.ep_attribute + ).read_attributes( + [target_attr.id], + allow_cache, + only_cache, + manufacturer, + ) + + if target_attr.id in result_s: + value = result_s[target_attr.id] + successful[attr.id] = format_func(value) if format_func else value + if target_attr.id in result_f: + failed[attr.id] = result_f[target_attr.id] + + if remaining_attributes: + result_s, result_f = await super().read_attributes( + remaining_attributes, allow_cache, only_cache, manufacturer + ) + successful.update(result_s) + failed.update(result_f) + + return successful, failed + + +class WriteAwareUpdateAttribute: + """Methods providing 'is_write' arg to _update_attribute.""" + + async def write_attributes_raw( + self, + attrs: list[foundation.Attribute], + manufacturer: int | None = None, + **kwargs, + ) -> list: + """Provide the is_write=True flag when calling _update_attribute.""" + + result = await self._write_attributes( + attrs, manufacturer=manufacturer, **kwargs + ) + if not isinstance(result[0], list): + return result + + records = result[0] + if len(records) == 1 and records[0].status == foundation.Status.SUCCESS: + for attr_rec in attrs: + self._update_attribute( + attr_rec.attrid, attr_rec.value.value, is_write=True + ) + else: + failed = [rec.attrid for rec in records] + for attr_rec in attrs: + if attr_rec.attrid not in failed: + self._update_attribute( + attr_rec.attrid, attr_rec.value.value, is_write=True + ) + + return result + + def _update_attribute( + self, attrid: int, value: Any, is_write: bool | None = None + ) -> None: + super()._update_attribute(attrid, value) + + class XiaomiAqaraRollerE1(XiaomiAqaraE1Cluster): - """Xiaomi mfg cluster implementation specific for E1 Roller.""" - - attributes = XiaomiCluster.attributes.copy() - attributes.update( - { - 0x0400: ("reverse_direction", t.Bool, True), - 0x0402: ("positions_stored", t.Bool, True), - 0x0407: ("store_position", t.uint8_t, True), - 0x0408: ("speed", t.uint8_t, True), - 0x0409: ("charging", t.uint8_t, True), - 0x00F7: ("aqara_attributes", t.LVBytes, True), - } - ) + """Aqara manufacturer cluster for the Roller Driver E1.""" + + class AttributeDefs(BaseAttributeDefs): + """Manufacturer specific attributes.""" + + reverse_direction = ZCLAttributeDef( + id=0x0400, + type=t.Bool, + access="rwp", + is_manufacturer_specific=True, + ) + positions_stored = ZCLAttributeDef( + id=0x0402, + type=t.Bool, + access="rwp", + is_manufacturer_specific=True, + ) -class AnalogOutputRollerE1(CustomCluster, AnalogOutput): - """Analog output cluster, only used to relay current_value to WindowCovering.""" + store_position = ZCLAttributeDef( + id=0x0407, + type=t.uint8_t, + access="rwp", + is_manufacturer_specific=True, + ) + + speed = ZCLAttributeDef( + id=0x0408, + type=AqaraRollerDriverSpeed, + zcl_type=DataTypeId.uint8, + access="rwp", + is_manufacturer_specific=True, + ) + + charging = ZCLAttributeDef( + id=0x0409, + type=AqaraRollerDriverCharging, + zcl_type=DataTypeId.uint8, + access="rp", + is_manufacturer_specific=True, + ) + + aqara_attributes = ZCLAttributeDef( + id=0x00F7, + type=t.LVBytes, + is_manufacturer_specific=True, + ) + + @staticmethod + def _enum_value_to_bool(enum_type, value: Any) -> t.Bool | None: + """Return the t.Bool equivalent of enum mapped true/false values.""" + + if value == enum_type.true: + return t.Bool.true + if value == enum_type.false: + return t.Bool.false + return None + + def _update_attribute(self, attrid: int, value: Any) -> None: + """Convert Aqara charging status values to binary_sensor charging true/false.""" + + if attrid == self.AttributeDefs.charging.id: + value = self._enum_value_to_bool(AqaraRollerDriverCharging, value) + super()._update_attribute(attrid, value) + + +class AnalogOutputRollerE1(WriteAwareUpdateAttribute, CustomCluster, AnalogOutput): + """Analog output cluster, only used to relay present_value to WindowCovering current_position_lift_percentage.""" _CONSTANT_ATTRIBUTES = { AnalogOutput.AttributeDefs.description.id: "Current position", @@ -70,18 +218,34 @@ class AnalogOutputRollerE1(CustomCluster, AnalogOutput): AnalogOutput.AttributeDefs.status_flags.id: 0x00, } - def _update_attribute(self, attrid: int, value: Any) -> None: - super()._update_attribute(attrid, value) + def _update_attribute( + self, attrid: int, value: Any, is_write: bool | None = None + ) -> None: + """Non-write 'present_value' updates should update the WindowCovering position.""" - if attrid == self.AttributeDefs.present_value.id: + super()._update_attribute(attrid, value) + if attrid == self.AttributeDefs.present_value.id and not is_write: self.endpoint.window_covering.update_attribute( WindowCovering.AttributeDefs.current_position_lift_percentage.id, (100 - value), ) -class WindowCoveringRollerE1(CustomCluster, WindowCovering): - """Window covering cluster to receive commands that are sent to the AnalogOutput's present_value to move the motor.""" +class WindowCoveringRollerE1(RedirectAttributes, CustomCluster, WindowCovering): + """Window covering cluster for handling motor commands.""" + + _CONSTANT_ATTRIBUTES = { + WindowCovering.AttributeDefs.window_covering_type.id: WindowCovering.WindowCoveringType.Rollershade, + } + + # This is used to redirect 'current_position_lift_percentage' reads to AnalogOutput 'present_value' + _REDIRECT_ATTRIBUTES = { + WindowCovering.AttributeDefs.current_position_lift_percentage: ( + AnalogOutput.AttributeDefs.present_value, + AnalogOutput, + lambda x: 100 - x, + ), + } async def command( self, @@ -99,218 +263,95 @@ async def command( """ if command_id == WindowCovering.ServerCommandDefs.up_open.id: (res,) = await self.endpoint.multistate_output.write_attributes( - {"present_value": 1} + { + MultistateOutput.AttributeDefs.present_value.name: AqaraRollerState.Opening + } ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) + if command_id == WindowCovering.ServerCommandDefs.down_close.id: (res,) = await self.endpoint.multistate_output.write_attributes( - {"present_value": 0} + { + MultistateOutput.AttributeDefs.present_value.name: AqaraRollerState.Closing + } ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) + if command_id == WindowCovering.ServerCommandDefs.go_to_lift_percentage.id: (res,) = await self.endpoint.analog_output.write_attributes( - {"present_value": (100 - args[0])} + {AnalogOutput.AttributeDefs.present_value.name: (100 - args[0])} ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) + if command_id == WindowCovering.ServerCommandDefs.stop.id: (res,) = await self.endpoint.multistate_output.write_attributes( - {"present_value": 2} + { + MultistateOutput.AttributeDefs.present_value.name: AqaraRollerState.Stopped + } + ) + # Current position is read because it is not consistently reported + await self.read_attributes( + [self.AttributeDefs.current_position_lift_percentage.id] ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) + return None + class MultistateOutputRollerE1(CustomCluster, MultistateOutput): - """Multistate Output cluster which overwrites present_value. + """Multistate Output cluster which overwrites present_value attribute type. - Otherwise, it gives errors of wrong datatype when using it in the commands. + The device responds with an error when using the standard t.Single type. """ - attributes = MultistateOutput.attributes.copy() - attributes.update( - { - MultistateOutput.AttributeDefs.present_value.id: ( - "present_value", - t.uint16_t, - ), - } - ) - - -class RollerE1AQ(XiaomiCustomDevice): - """Aqara Roller Shade Driver E1 device.""" - - signature = { - MODELS_INFO: [(LUMI, "lumi.curtain.acn002")], - ENDPOINTS: { - # - 1: { - PROFILE_ID: zha.PROFILE_ID, - DEVICE_TYPE: zha.DeviceType.ON_OFF_LIGHT, - INPUT_CLUSTERS: [ - Alarms.cluster_id, - AnalogOutput.cluster_id, - Basic.cluster_id, - DeviceTemperature.cluster_id, - Groups.cluster_id, - Identify.cluster_id, - XiaomiAqaraRollerE1.cluster_id, - MultistateOutput.cluster_id, - OnOff.cluster_id, - Scenes.cluster_id, - WindowCovering.cluster_id, - ], - OUTPUT_CLUSTERS: [ - Ota.cluster_id, - Time.cluster_id, - ], - }, - # - 242: { - PROFILE_ID: zgp.PROFILE_ID, - DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC, - INPUT_CLUSTERS: [], - OUTPUT_CLUSTERS: [ - GreenPowerProxy.cluster_id, - ], - }, - }, - } - replacement = { - ENDPOINTS: { - 1: { - PROFILE_ID: zha.PROFILE_ID, - DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE, - INPUT_CLUSTERS: [ - Alarms.cluster_id, - AnalogOutputRollerE1, - BasicCluster, - DeviceTemperature.cluster_id, - Groups.cluster_id, - Identify.cluster_id, - XiaomiAqaraRollerE1, - MultistateOutputRollerE1, - Scenes.cluster_id, - WindowCoveringRollerE1, - XiaomiPowerConfigurationPercent, - ], - OUTPUT_CLUSTERS: [ - Ota.cluster_id, - Time.cluster_id, - ], - }, - 242: { - PROFILE_ID: zgp.PROFILE_ID, - DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC, - INPUT_CLUSTERS: [], - OUTPUT_CLUSTERS: [ - GreenPowerProxy.cluster_id, - ], - }, - }, - } + class AttributeDefs(MultistateOutput.AttributeDefs): + """Aqara attribute definition overrides.""" + present_value: Final = ZCLAttributeDef( + id=0x0055, type=t.uint16_t, access="r*w", mandatory=True + ) -class RollerE1AQ_2(RollerE1AQ): - """Aqara Roller Shade Driver E1 (version 2) device.""" - - signature = { - MODELS_INFO: [(LUMI, "lumi.curtain.acn002")], - ENDPOINTS: { - # - 1: { - PROFILE_ID: zha.PROFILE_ID, - DEVICE_TYPE: zha.DeviceType.ON_OFF_LIGHT, - INPUT_CLUSTERS: [ - Alarms.cluster_id, - AnalogOutput.cluster_id, - Basic.cluster_id, - DeviceTemperature.cluster_id, - Groups.cluster_id, - Identify.cluster_id, - MultistateOutput.cluster_id, - OnOff.cluster_id, - Scenes.cluster_id, - WindowCovering.cluster_id, - ], - OUTPUT_CLUSTERS: [ - Ota.cluster_id, - Time.cluster_id, - ], - }, - # - 242: { - PROFILE_ID: zgp.PROFILE_ID, - DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC, - INPUT_CLUSTERS: [], - OUTPUT_CLUSTERS: [ - GreenPowerProxy.cluster_id, - ], - }, - }, - } - -class RollerE1AQ_3(RollerE1AQ): - """Aqara Roller Shade Driver E1 (version 3) device.""" - - signature = { - MODELS_INFO: [(LUMI, "lumi.curtain.acn002")], - ENDPOINTS: { - # - 1: { - PROFILE_ID: zha.PROFILE_ID, - DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE, - INPUT_CLUSTERS: [ - Alarms.cluster_id, - AnalogOutput.cluster_id, - Basic.cluster_id, - DeviceTemperature.cluster_id, - Groups.cluster_id, - Identify.cluster_id, - MultistateOutput.cluster_id, - OnOff.cluster_id, - Scenes.cluster_id, - WindowCovering.cluster_id, - ], - OUTPUT_CLUSTERS: [ - Ota.cluster_id, - Time.cluster_id, - ], - }, - # - 242: { - PROFILE_ID: zgp.PROFILE_ID, - DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC, - INPUT_CLUSTERS: [], - OUTPUT_CLUSTERS: [ - GreenPowerProxy.cluster_id, - ], - }, - }, - } +( + QuirkBuilder(LUMI, "lumi.curtain.acn002") + # temporarily commented out due to potentially breaking existing blueprints + # .friendly_name( + # manufacturer="Aqara", model="Roller Shade Driver E1" + # ) + .removes(OnOff.cluster_id) + .replaces(AnalogOutputRollerE1) + .replaces(BasicCluster) + .replaces(MultistateOutputRollerE1) + .replaces(XiaomiPowerConfigurationPercent) + .replaces(WindowCoveringRollerE1) + .replaces(XiaomiAqaraRollerE1) + .enum( + XiaomiAqaraRollerE1.AttributeDefs.speed.name, + AqaraRollerDriverSpeed, + XiaomiAqaraRollerE1.cluster_id, + translation_key="speed", + fallback_name="Speed", + ) + .binary_sensor( + XiaomiAqaraRollerE1.AttributeDefs.charging.name, + XiaomiAqaraRollerE1.cluster_id, + device_class=BinarySensorDeviceClass.BATTERY_CHARGING, + translation_key="charging", + fallback_name="Charging", + ) + .binary_sensor( + XiaomiAqaraRollerE1.AttributeDefs.positions_stored.name, + XiaomiAqaraRollerE1.cluster_id, + translation_key="calibrated", + fallback_name="Calibrated", + ) + .add_to_registry() +)