Clean up Aqara Roller Curtain E1 quirk

zhaquirks/xiaomi/aqara/roller_curtain_e1.py

@@ -21,7 +21,6 @@
     Scenes,
     Time,
 )
-from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
 
 from zhaquirks import CustomCluster
 from zhaquirks.const import (
@@ -35,24 +34,16 @@
 from zhaquirks.xiaomi import (
     LUMI,
     BasicCluster,
+    XiaomiAqaraE1Cluster,
     XiaomiCluster,
     XiaomiCustomDevice,
     XiaomiPowerConfiguration,
 )
 
-PRESENT_VALUE = 0x0055
-CURRENT_POSITION_LIFT_PERCENTAGE = 0x0008
-GO_TO_LIFT_PERCENTAGE = 0x0005
-DOWN_CLOSE = 0x0001
-UP_OPEN = 0x0000
-STOP = 0x0002
 
-
-class XiaomiAqaraRollerE1(XiaomiCluster, ManufacturerSpecificCluster):
+class XiaomiAqaraRollerE1(XiaomiAqaraE1Cluster):
     """Xiaomi mfg cluster implementation specific for E1 Roller."""
 
-    cluster_id = 0xFCC0
-
     attributes = XiaomiCluster.attributes.copy()
     attributes.update(
         {
@@ -69,36 +60,29 @@ class XiaomiAqaraRollerE1(XiaomiCluster, ManufacturerSpecificCluster):
 class AnalogOutputRollerE1(CustomCluster, AnalogOutput):
     """Analog output cluster, only used to relay current_value to WindowCovering."""
 
-    cluster_id = AnalogOutput.cluster_id
-
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Init."""
         super().__init__(*args, **kwargs)
 
-        self._update_attribute(0x0041, float(0x064))  # max_present_value
-        self._update_attribute(0x0045, 0.0)  # min_present_value
-        self._update_attribute(0x0051, 0)  # out_of_service
-        self._update_attribute(0x006A, 1.0)  # resolution
-        self._update_attribute(0x006F, 0x00)  # status_flags
+        self.update_attribute(self.AttributeDefs.max_present_value.id, float(0x064))
+        self.update_attribute(self.AttributeDefs.min_present_value.id, 0.0)
+        self.update_attribute(self.AttributeDefs.out_of_service.id, 0)
+        self.update_attribute(self.AttributeDefs.resolution.id, 1.0)
+        self.update_attribute(self.AttributeDefs.status_flags.id, 0x00)
 
     def _update_attribute(self, attrid: int, value: Any) -> None:
         super()._update_attribute(attrid, value)
 
-        if attrid == PRESENT_VALUE:
-            self.endpoint.window_covering._update_attribute(
-                CURRENT_POSITION_LIFT_PERCENTAGE, (100 - value)
+        if attrid == self.AttributeDefs.present_value.id:
+            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."""
 
-    cluster_id = WindowCovering.cluster_id
-
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        """Init."""
-        super().__init__(*args, **kwargs)
-
     async def command(
         self,
         command_id: foundation.GeneralCommand | int | t.uint8_t,
@@ -113,28 +97,28 @@ async def command(
         We either overwrite analog_output's current_value or multistate_output's current
         value to make the roller work.
         """
-        if command_id == UP_OPEN:
+        if command_id == WindowCovering.ServerCommandDefs.up_open.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
                 {"present_value": 1}
             )
             return foundation.GENERAL_COMMANDS[
                 foundation.GeneralCommand.Default_Response
             ].schema(command_id=command_id, status=res[0].status)
-        if command_id == DOWN_CLOSE:
+        if command_id == WindowCovering.ServerCommandDefs.down_close.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
                 {"present_value": 0}
             )
             return foundation.GENERAL_COMMANDS[
                 foundation.GeneralCommand.Default_Response
             ].schema(command_id=command_id, status=res[0].status)
-        if command_id == GO_TO_LIFT_PERCENTAGE:
+        if command_id == WindowCovering.ServerCommandDefs.go_to_lift_percentage.id:
             (res,) = await self.endpoint.analog_output.write_attributes(
                 {"present_value": (100 - args[0])}
             )
             return foundation.GENERAL_COMMANDS[
                 foundation.GeneralCommand.Default_Response
             ].schema(command_id=command_id, status=res[0].status)
-        if command_id == STOP:
+        if command_id == WindowCovering.ServerCommandDefs.stop.id:
             (res,) = await self.endpoint.multistate_output.write_attributes(
                 {"present_value": 2}
             )
@@ -152,7 +136,10 @@ class MultistateOutputRollerE1(CustomCluster, MultistateOutput):
     attributes = MultistateOutput.attributes.copy()
     attributes.update(
         {
-            0x0055: ("present_value", t.uint16_t),
+            MultistateOutput.AttributeDefs.present_value.id: (
+                "present_value",
+                t.uint16_t,
+            ),
         }
     )