Repository: zigpy/zha-device-handlers
Pull request #3686 (2025-01-08 11:06:18)

Refactor Aqara Roller Driver E1 as v2 quirk for more entities
## Proposed change
This PR refactors the existing Aqara Roller Driver E1 quirk into the new v2 structure to expose the configuration and status attributes, below is a summary of changes:

 - Refactor into v2 quirk format
   - Refactor clusters to use AttributeDefs format
 - Expose attributes as entities:
   - **Motor speed** (select)
   - **Charging status** (binary sensor)
   - **Calibration status** (binary sensor)
 - Represent the device as a **_Rollershade_** WindowCoveringType rather than **_Drapery_**
 - Device cluster attribute handling:
   - Reads/reports of `AnalogOutput: present_value` (current position) are replicated into `WindowCovering: current_position_lift_percentage`
   - Writes to `AnalogOutput: present_value` (go to position) do not update the `WindowCovering: current_position_lift_percentage`
   - Reads of `WindowCovering: current_position_lift_percentage` are redirected to `AnalogOutput: present_value`
 - Window Covering command handling:
   - `up_open`, `down_close`, `stop` are handled by `MultistateOutput: present_value`
   - `go_to_lift_percentage` value is written to `AnalogOutput: present_value`
   - The `current_position_lift_percentage` is read prior to returning command response, this ensures the correct state is shown following a change in direction


**Functional enhancements**

- Users can change the Speed using the select entity
- Users can see if the device is Charging
- Users can see if the device is Calibrated
- The cover uses the Rollershade type
- The redundant number entity for current position is disabled by default
- The redundant switch entity is disabled by default

<img width="202" alt="image" src="https://github.com/user-attachments/assets/04eac041-4319-449c-a43c-4a9014c8e4ac" />

## Additional information
<!--
  Please include any additional information that is important to this PR.
  For example, if this PR is a potentially breaking change, mention that here.
  If this PR requires other PRs to be merged in HA Core or other projects, mention that.
  Lastly, if this PR fixes a specific issue, please include "Fixes #xxxx".
-->

Depends on the following PRs for binary_sensor value conversion:
 - https://github.com/zigpy/zigpy/pull/1540
 - https://github.com/zigpy/zha/pull/360

Depends on the following PR for disabling the redundant entities:
 - https://github.com/zigpy/zigpy/pull/1535
 - https://github.com/zigpy/zha/pull/387

Associated PRs:
 - Fixes cover handling of Rollershade device class (for the correct icon) https://github.com/zigpy/zha/pull/376

Whilst I've validated that is works with my own E1 roller shade devices, I'd like to ensure other users do not encounter any regressions with this refactor.
@dmulcahey / @schwickster / @javicalle / @badrpc / @TheJulianJES I can see in the past you've previously contributed for this device, if you still use it would you mind checking if this updated quirk works as expected without introducing regressions?

## Checklist
<!--
  Put an 'x' in all boxes that apply.
  Note: You do not need to tick all boxes before creating a PR.
-->

- [x] The changes are tested and work correctly
- [x] `pre-commit` checks pass / the code has been formatted using Black
- [x] Tests have been added to verify that the new code works


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/zigpy/zha-device-handlers/pull/3686?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy) Report
Attention: Patch coverage is `97.14286%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 89.86%. Comparing base [(`e3d3849`)](https://app.codecov.io/gh/zigpy/zha-device-handlers/commit/e3d3849ffdd74b809713f9f154172ddad329aad9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy) to head [(`babacac`)](https://app.codecov.io/gh/zigpy/zha-device-handlers/commit/babacac5542ce15e9278e7f72617efe4b5c6f60a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy).

| [Files with missing lines](https://app.codecov.io/gh/zigpy/zha-device-handlers/pull/3686?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy) | Patch % | Lines |
|---|---|---|
| [zhaquirks/xiaomi/aqara/roller\_curtain\_e1.py](https://app.codecov.io/gh/zigpy/zha-device-handlers/pull/3686?src=pr&el=tree&filepath=zhaquirks%2Fxiaomi%2Faqara%2Froller_curtain_e1.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy#diff-emhhcXVpcmtzL3hpYW9taS9hcWFyYS9yb2xsZXJfY3VydGFpbl9lMS5weQ==) | 97.14% | [1 Missing :warning: ](https://app.codecov.io/gh/zigpy/zha-device-handlers/pull/3686?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##              dev    #3686   +/-   ##
=======================================
  Coverage   89.85%   89.86%           
=======================================
  Files         322      322           
  Lines       10380    10400   +20     
=======================================
+ Hits         9327     9346   +19     
- Misses       1053     1054    +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/zigpy/zha-device-handlers/pull/3686?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=zigpy).

jeverley: Have identified an issue with the cover state not reliably changing to open/closed. Investigating this currently.
jeverley: @TheJulianJES, @dmulcahey looking at the code in https://github.com/zigpy/zha/blob/dev/zha/application/platforms/cover/__init__.py
it looks to me like the `_determine_state` function and stored target positions might be the cause of the issue, I'm seeing similar issues reported for other devices which would appear to have the same root cause as below https://github.com/home-assistant/core/issues/93836, https://github.com/home-assistant/core/issues/98933.

I've tested the following scenarios:

**1. _go_to_lift_percentage_ has never been called, _down_close_ completes with reported _position_or_tilt_ of 0**
_target_lift_position == None
position_or_tilt == 0
returns **STATE_CLOSED**

**2. _go_to_lift_percentage_ has never been called, _up_open_ completes with reported _position_or_tilt_ of 100**
_target_lift_position == None
position_or_tilt == 100
returns: **STATE_OPEN**

**3. _go_to_lift_percentage_ has been set to 50, _down_close_ completes with reported _position_or_tilt_ of 0**
_target_lift_position == 50
position_or_tilt == 0
returns **STATE_CLOSED**

**4. _go_to_lift_percentage_ has been set to 50, _up_open_ completes with reported _position_or_tilt_ of 100**
_target_lift_position == 50
position_or_tilt == 100
returns **None, state remains as STATE_OPENING**

**5. stop command issued, _position_or_tilt_ of 100**
_target_lift_position == 100 (the function async_stop_cover sets it to the current cover position)
position_or_tilt == 100
returns **STATE_OPEN**

I'd suggest we can handle this in one of 3 ways:

1. set the stored _target_lift_position to 100 or 0 when calling the _up_open_ or _down_close_ commands respectively
2. clear the stored _target_lift_position when calling the _up_open_ or _down_close_ commands
3. modify the _determine_state function to return STATE_OPEN when `self.is_opening == True and position_or_tilt == 100` is true


Below is the current function for reference:
```
    def _determine_state(self, position_or_tilt, is_lift_update=True) -> None:
        """Determine the state of the cover.

        In HA None is unknown, 0 is closed, 100 is fully open.
        In ZCL 0 is fully open, 100 is fully closed.
        Keep in mind the values have already been flipped to match HA
        in the WindowCovering cluster handler
        """
        if is_lift_update:
            target = self._target_lift_position
            current = self.current_cover_position
        else:
            target = self._target_tilt_position
            current = self.current_cover_tilt_position

        if position_or_tilt == 0:
            self._state = (
                STATE_CLOSED
                if is_lift_update
                else STATE_OPEN
                if self.current_cover_position is not None
                and self.current_cover_position > 0
                else STATE_CLOSED
            )
            return
        if target is not None and target != current:
            # we are mid transition and shouldn't update the state
            return
        self._state = STATE_OPEN
```
jeverley: I've been working on changes to the ZHA cover entity hander to address the underlying status issue outlined above (it also affects the existing v1 quirk code for this and other devices).

I'll raise a bug and PR to address these over the weekend, current WIP code here:
https://github.com/jeverley/zha/commit/655010881864f76e55388bce6f1255cff1e026da
jeverley: @TheJulianJES if I'm reading https://github.com/zigpy/zigpy/pull/1540 I believe I should be able to use the proposed `attribute_converter` functionality to remap the Aqara values for the charging binary_sensor?

If so I'll remove the custom converter I've implemented in this quirk once your PR is merged.
jeverley: @TheJulianJES the `AnalogOutput` cluster `present_value` on this device is used to write the intended target position, whereas reads/reports on the attribute report the device current position.

Right now ZHA exposes this as a separate number entity in addition to the cover entity.
In the UI the number entity is redundant since interactions & current position can all be handled through the cover.

When adding custom entities in a v2 quirk we're able to mark them as disabled by default, do you think it would be beneficial to extend this to regular cluster attributes?
This would allow us to present devices more uniformly, without exposing redundant entities arising from manufacturers' non-standard implementations.

Critically this would be different to entirely removing the cluster/marking an attribute as unsupported in the cluster, since we'd still want to remap attribute updates/writes from the cover commands to the cluster.
TheJulianJES: > When adding custom entities in a v2 quirk we're able to mark them as disabled by default, do you think it would be beneficial to extend this to regular cluster attributes?

With https://github.com/zigpy/zigpy/pull/1535, that should be allowed in the future. Do note that the ZHA part wasn't PR'd yet.
jeverley: @TheJulianJES I think this should be ready for merging, I'll raise a PR for the zha cover status issue outlined in https://github.com/zigpy/zha-device-handlers/pull/3686#issuecomment-2578907234 separately (it's not a pre-req for this change).
jeverley: I've now raised https://github.com/zigpy/zha/pull/376 to address the ZHA logic issues I outlined in https://github.com/zigpy/zha-device-handlers/pull/3686#issuecomment-2578907234.
jeverley: Will update the tests following https://github.com/zigpy/zha-device-handlers/pull/3686/commits/9e5b67ac9a7749c1bfb3f95827462da4cede735c later this evening.

Patch:
diff --git a/zhaquirks/xiaomi/aqara/roller_curtain_e1.py b/zhaquirks/xiaomi/aqara/roller_curtain_e1.py
index d59f9b1..d0d5466 100644
--- a/zhaquirks/xiaomi/aqara/roller_curtain_e1.py
+++ b/zhaquirks/xiaomi/aqara/roller_curtain_e1.py
@@ -2,64 +2,194 @@
 
 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."""
+
+    Charging = 0x01
+    NotCharging = 0x02
+
+
+class AqaraRollerDriverSpeed(t.enum8):
+    """Aqara roller driver speed attribute values."""
+
+    Low = 0x00
+    Medium = 0x01
+    High = 0x02
+
+
+class AqaraRollerControl(t.enum8):
+    """Aqara roller control attribute values."""
+
+    Close = 0x00
+    Open = 0x01
+    Stop = 0x02
+
+
+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,
+        )
 
-class AnalogOutputRollerE1(CustomCluster, AnalogOutput):
-    """Analog output cluster, only used to relay current_value to WindowCovering."""
+        positions_stored = ZCLAttributeDef(
+            id=0x0402,
+            type=t.Bool,
+            access="rwp",
+            is_manufacturer_specific=True,
+        )
+
+        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,
+        )
+
+
+class AnalogOutputRollerE1(WriteAwareUpdateAttribute, CustomCluster, AnalogOutput):
+    """AnalogOutput cluster reporting current position and used for writing target position."""
 
     _CONSTANT_ATTRIBUTES = {
         AnalogOutput.AttributeDefs.description.id: "Current position",
@@ -70,18 +200,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),
+                t.uint8_t(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: t.uint8_t(100 - x),
+        ),
+    }
 
     async def command(
         self,
@@ -94,223 +240,119 @@ class WindowCoveringRollerE1(CustomCluster, WindowCovering):
     ) -> Any:
         """Overwrite the commands to make it work for both firmware 1425 and 1427.
 
-        We either overwrite analog_output's current_value or multistate_output's current
-        value to make the roller work.
+        Write to AnalogOutput current_value for go go_to_lift_percentage.
+        Write to MultistateOutput current_value for up_open/down_close/stop.
+
+        The current_position_lift_percentage is read prior to returning the command response
+        to ensure that ZHA has the correct position during changes in direction/stopping.
         """
         if command_id == WindowCovering.ServerCommandDefs.up_open.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
-                {"present_value": 1}
+                {
+                    MultistateOutput.AttributeDefs.present_value.name: AqaraRollerControl.Open
+                }
+            )
+            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)
+
         if command_id == WindowCovering.ServerCommandDefs.down_close.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
-                {"present_value": 0}
+                {
+                    MultistateOutput.AttributeDefs.present_value.name: AqaraRollerControl.Close
+                }
+            )
+            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)
+
         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])}
+            )
+            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)
+
         if command_id == WindowCovering.ServerCommandDefs.stop.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
-                {"present_value": 2}
+                {
+                    MultistateOutput.AttributeDefs.present_value.name: AqaraRollerControl.Stop
+                }
+            )
+            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 foundation.GENERAL_COMMANDS[
+            foundation.GeneralCommand.Default_Response
+        ].schema(command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND)
+
 
 class MultistateOutputRollerE1(CustomCluster, MultistateOutput):
-    """Multistate Output cluster which overwrites present_value.
+    """MultistateOutput cluster used for writing commands (up_open, down_close, stop).
 
-    Otherwise, it gives errors of wrong datatype when using it in the commands.
+    This requires a change to the present_value attribute type because 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 AttributeDefs(MultistateOutput.AttributeDefs):
+        """Aqara attribute definition overrides."""
 
+        present_value: Final = ZCLAttributeDef(
+            id=0x0055, type=t.uint16_t, access="r*w", mandatory=True
+        )
 
-class RollerE1AQ(XiaomiCustomDevice):
-    """Aqara Roller Shade Driver E1 device."""
-
-    signature = {
-        MODELS_INFO: [(LUMI, "lumi.curtain.acn002")],
-        ENDPOINTS: {
-            # <SizePrefixedSimpleDescriptor endpoint=1 profile=260 device_type=256
-            # device_version=1
-            # input_clusters=[0, 2, 3, 4, 5, 6, 9, 64704, 13, 19, 258]
-            # output_clusters=[10, 25]>
-            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,
-                ],
-            },
-            # <SizePrefixedSimpleDescriptor endpoint=242 profile=41440 device_type=97
-            # device_version=0,
-            # input_clusters=[]
-            # output_clusters=[33]>
-            242: {
-                PROFILE_ID: zgp.PROFILE_ID,
-                DEVICE_TYPE: zgp.DeviceType.PROXY_BASIC,
-                INPUT_CLUSTERS: [],
-                OUTPUT_CLUSTERS: [
-                    GreenPowerProxy.cluster_id,
-                ],
-            },
-        },
-    }
-    replacement = {
-        ENDPOINTS: {
-            1: {
-                PROFILE_ID: zha.PROFILE_ID,
-                DEVICE_TYPE: zha.DeviceType.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 RollerE1AQ_2(RollerE1AQ):
-    """Aqara Roller Shade Driver E1 (version 2) device."""
-
-    signature = {
-        MODELS_INFO: [(LUMI, "lumi.curtain.acn002")],
-        ENDPOINTS: {
-            # <SizePrefixedSimpleDescriptor endpoint=1 profile=260 device_type=256
-            # device_version=1
-            # input_clusters=[0, 2, 3, 4, 5, 6, 9, 13, 19, 258]
-            # output_clusters=[10, 25]>
-            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,
-                ],
-            },
-            # <SizePrefixedSimpleDescriptor endpoint=242 profile=41440 device_type=97
-            # device_version=0,
-            # input_clusters=[]
-            # output_clusters=[33]>
-            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: {
-            # <SizePrefixedSimpleDescriptor endpoint=1 profile=260 device_type=514
-            # device_version=1
-            # input_clusters=[0, 2, 3, 4, 5, 6, 9, 13, 19, 258]
-            # output_clusters=[10, 25]>
-            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,
-                ],
-            },
-            # <SizePrefixedSimpleDescriptor endpoint=242 profile=41440 device_type=97
-            # device_version=0,
-            # input_clusters=[]
-            # output_clusters=[33]>
-            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"
+    #    )
+    .prevent_default_entity_creation(endpoint_id=1, cluster_id=AnalogOutput.cluster_id)
+    .prevent_default_entity_creation(
+        endpoint_id=1, cluster_id=MultistateOutput.cluster_id
+    )
+    .prevent_default_entity_creation(endpoint_id=1, cluster_id=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",
+        attribute_converter=lambda x: x == AqaraRollerDriverCharging.Charging,
+    )
+    .binary_sensor(
+        XiaomiAqaraRollerE1.AttributeDefs.positions_stored.name,
+        XiaomiAqaraRollerE1.cluster_id,
+        translation_key="calibrated",
+        fallback_name="Calibrated",
+    )
+    .add_to_registry()
+)


Test patch:
diff --git a/tests/test_xiaomi.py b/tests/test_xiaomi.py
index fb6a452..546dfd9 100644
--- a/tests/test_xiaomi.py
+++ b/tests/test_xiaomi.py
@@ -8,7 +8,7 @@ from unittest import mock
 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,
@@ -1434,15 +1434,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)
@@ -1640,64 +1637,413 @@ 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, True),
+        (WindowCovering.ServerCommandDefs.down_close.id, 0, True),
+        (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
+            ]
+
+            # confirm the position was updated on the ZCL WindowCovering cluster
+            assert len(window_covering_listener.attribute_updates) == 1
+            assert window_covering_listener.attribute_updates[0] == (
+                window_covering_attr_id,
+                100 - 1,
+            )
+        else:
+            # confirm the command did not read the current position
+            assert len(analog_cluster._read_attributes.mock_calls) == 0
 
 
 @pytest.mark.parametrize(
-    "command, value",
+    "command, value, read_current_position",
     [
-        (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id, 60),
+        (WindowCovering.ServerCommandDefs.go_to_lift_percentage.id, 60, True),
     ],
 )
-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, read_current_position
+):
     """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)
+    window_covering_attr_id = (
+        WindowCovering.AttributeDefs.current_position_lift_percentage.id
+    )
+
     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
+
+    # 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)
+    )
+
+    # patch write commands
+    patch_analog_write = mock.patch.object(
+        analog_cluster,
+        "_write_attributes",
+        mock.AsyncMock(
+            return_value=(
+                [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)],
+            )
+        ),
+    )
+
+    with (
+        patch_window_covering_read,
+        patch_analog_read,
+        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
+        )
+
+        # confirm the AnalogOutput present_value was updated
+        assert analog_listener.attribute_updates[0] == (
+            analog_attr_id,
+            100 - 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
+            ]
+
+            # confirm the position was updated on the ZCL WindowCovering cluster with the read value
+            assert len(window_covering_listener.attribute_updates) == 1
+            assert window_covering_listener.attribute_updates[0] == (
+                window_covering_attr_id,
+                100 - 1,
+            )
+        else:
+            # confirm the command did not read the current position
+            assert len(analog_cluster._read_attributes.mock_calls) == 0
+
+            # confirm the AnalogOutput write did not update the current WindowCovering position
+            assert len(window_covering_listener.attribute_updates) == 0
+
+    # confirm non-mapped commands return status UNSUP_CLUSTER_COMMAND
+    _, status = await window_covering_cluster.go_to_tilt_percentage(value)
+    assert status == foundation.Status.UNSUP_CLUSTER_COMMAND
+
+
+@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,
+    )
+
+
+async def test_xiaomi_e1_roller_write_aware_update_attribute(
+    zigpy_device_from_v2_quirk,
+):
+    """Test Aqara E1 roller AnalogOutput write-aware update_attribute method."""
+    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_listener = ClusterListener(analog_cluster)
+    analog_attr = AnalogOutput.AttributeDefs.present_value
+    analog_attr_max = AnalogOutput.AttributeDefs.max_present_value
+
+    # patch write command for a success response
+    patch_analog_write = mock.patch.object(
+        analog_cluster,
+        "_write_attributes",
+        mock.AsyncMock(
+            return_value=(
+                [foundation.WriteAttributesStatusRecord(foundation.Status.SUCCESS)],
+            )
+        ),
     )
 
+    # patch write command for a fail response
+    patch_analog_write_fail = mock.patch.object(
+        analog_cluster,
+        "_write_attributes",
+        mock.AsyncMock(
+            return_value=(
+                [
+                    foundation.WriteAttributesStatusRecord(
+                        foundation.Status.INVALID_VALUE, analog_attr.id
+                    ),
+                ],
+            )
+        ),
+    )
+
+    with (
+        patch_analog_write,
+    ):
+        # test writing valid and invalid values using name & id
+        await analog_cluster.write_attributes({analog_attr.id: 50})
+        await analog_cluster.write_attributes({analog_attr.name: 60})
+        assert analog_cluster._write_attributes.call_count == 2
+
+        # confirm the two successful writes updated the analog cluster
+        assert len(analog_listener.attribute_updates) == 2
+        assert analog_listener.attribute_updates[0] == (
+            analog_attr.id,
+            50,
+        )
+        assert analog_listener.attribute_updates[1] == (
+            analog_attr.id,
+            60,
+        )
+
+    with (
+        patch_analog_write_fail,
+    ):
+        # test writing valid and invalid values using name & id
+        await analog_cluster.write_attributes(
+            {analog_attr_max.id: 100, analog_attr.id: 150}
+        )
+        await analog_cluster.write_attributes(
+            {analog_attr_max.name: 100, analog_attr.name: 160}
+        )
+        assert analog_cluster._write_attributes.call_count == 2
+
+        # confirm the two failed attr writes did not update the analog cluster
+        assert len(analog_listener.attribute_updates) == 4
+
+        # confirm the two successful writes updated the analog cluster
+        assert analog_listener.attribute_updates[2] == (
+            analog_attr_max.id,
+            100,
+        )
+        assert analog_listener.attribute_updates[3] == (
+            analog_attr_max.id,
+            100,
+        )
+
+    # confirm the write invoked update_attributes did not update the covering cluster
+    assert len(window_covering_listener.attribute_updates) == 0
+
 
 @pytest.mark.parametrize("endpoint", [(1), (2)])
 async def test_aqara_t2_relay(zigpy_device_from_quirk, endpoint):
@@ -1826,11 +2172,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)
@@ -1861,15 +2207,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