Merge branch 'release/0.0.69'

.gitignore

@@ -10,3 +10,11 @@ build/**/*.*
 .coverage
 htmlcov/
 .mypycache
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/

setup.py

@@ -4,7 +4,7 @@
 
 from setuptools import find_packages, setup
 
-VERSION = "0.0.68"
+VERSION = "0.0.69"
 
 
 setup(

tests/test_quirks.py

@@ -1,17 +1,21 @@
 """General quirk tests."""
+from __future__ import annotations
 
+from pathlib import Path
 from unittest import mock
 
 import pytest
 import zigpy.device
 import zigpy.endpoint
 import zigpy.profiles
 import zigpy.quirks as zq
+from zigpy.quirks import CustomDevice
 import zigpy.types
 import zigpy.zcl as zcl
 import zigpy.zdo.types
 
 import zhaquirks
+import zhaquirks.bosch.motion
 from zhaquirks.const import (
     DEVICE_TYPE,
     ENDPOINTS,
@@ -62,7 +66,7 @@
 
 
 @pytest.mark.parametrize("quirk", ALL_QUIRK_CLASSES)
-def test_quirk_replacements(quirk):
+def test_quirk_replacements(quirk: CustomDevice) -> None:
     """Test all quirks have a replacement."""
 
     assert quirk.signature
@@ -72,15 +76,15 @@ def test_quirk_replacements(quirk):
 
 
 @pytest.fixture
-def raw_device():
+def raw_device() -> zigpy.device.Device:
     """Raw device."""
     app = mock.MagicMock()
     ieee = zigpy.types.EUI64.convert("11:22:33:44:55:66:77:88")
     nwk = 0x1234
     return zigpy.device.Device(app, ieee, nwk)
 
 
-def test_dev_from_signature_incomplete_sig(raw_device):
+def test_dev_from_signature_incomplete_sig(raw_device: zigpy.device.Device) -> None:
     """Test device initialization from quirk's based on incomplete signature."""
 
     class BadSigNoSignature(zhaquirks.QuickInitDevice):
@@ -224,7 +228,9 @@ class BadSigIncompleteEp(zhaquirks.QuickInitDevice):
         },
     ),
 )
-def test_dev_from_signature(raw_device, quirk_signature):
+def test_dev_from_signature(
+    raw_device: zigpy.device.Device, quirk_signature: dict
+) -> None:
     """Test device initialization from quirk's based on signature."""
 
     class QuirkDevice(zhaquirks.QuickInitDevice):
@@ -255,7 +261,7 @@ class QuirkDevice(zhaquirks.QuickInitDevice):
 @pytest.mark.parametrize(
     "quirk", (q for q in ALL_QUIRK_CLASSES if issubclass(q, zhaquirks.QuickInitDevice))
 )
-def test_quirk_quickinit(quirk):
+def test_quirk_quickinit(quirk: zigpy.quirks.CustomDevice) -> None:
     """Make sure signature in QuickInit Devices have all required attributes."""
 
     if not issubclass(quirk, zhaquirks.QuickInitDevice):
@@ -273,10 +279,10 @@ def test_quirk_quickinit(quirk):
 
 
 @pytest.mark.parametrize("quirk", ALL_QUIRK_CLASSES)
-def test_signature(quirk):
+def test_signature(quirk: CustomDevice) -> None:
     """Make sure signature look sane for all custom devices."""
 
-    def _check_range(cluster):
+    def _check_range(cluster: zcl.Cluster) -> bool:
         for range in zcl.Cluster._registry_range.keys():
             if range[0] <= cluster <= range[1]:
                 return True
@@ -360,7 +366,7 @@ def _check_range(cluster):
 
 
 @pytest.mark.parametrize("quirk", ALL_QUIRK_CLASSES)
-def test_quirk_importable(quirk):
+def test_quirk_importable(quirk: CustomDevice) -> None:
     """Ensure all quirks can be imported with a normal Python `import` statement."""
 
     path = f"{quirk.__module__}.{quirk.__name__}"
@@ -369,7 +375,7 @@ def test_quirk_importable(quirk):
     ), f"{path} is not importable"
 
 
-def test_quirk_loading_error(tmp_path):
+def test_quirk_loading_error(tmp_path: Path) -> None:
     """Ensure quirks do not silently fail to load."""
 
     custom_quirks = tmp_path / "custom_zha_quirks"
@@ -394,7 +400,9 @@ def test_quirk_loading_error(tmp_path):
         zhaquirks.setup({zhaquirks.CUSTOM_QUIRKS_PATH: str(custom_quirks)})
 
 
-def test_custom_quirk_loading(zigpy_device_from_quirk, tmp_path):
+def test_custom_quirk_loading(
+    zigpy_device_from_quirk: CustomDevice, tmp_path: Path
+) -> None:
     """Make sure custom quirks take priority over regular quirks."""
 
     device = zigpy_device_from_quirk(
@@ -486,3 +494,17 @@ class TestReplacementISWZPR1WP13(CustomDevice):
 
     assert not isinstance(zq.get_device(device), zhaquirks.bosch.motion.ISWZPR1WP13)
     assert type(zq.get_device(device)).__name__ == "TestReplacementISWZPR1WP13"
+
+
+def test_zigpy_custom_cluster_pollution() -> None:
+    """Ensure all quirks subclass `CustomCluster`."""
+    non_zigpy_clusters = {
+        cluster
+        for cluster in zcl.Cluster._registry.values()
+        if not cluster.__module__.startswith("zigpy.")
+    }
+
+    if non_zigpy_clusters:
+        raise RuntimeError(
+            f"Custom clusters must subclass `CustomCluster`: {non_zigpy_clusters}"
+        )

zhaquirks/tuya/ts0211.py

@@ -10,6 +10,7 @@
 from zigpy.zcl.clusters.homeautomation import Diagnostic
 from zigpy.zcl.clusters.security import IasZone
 
+from zhaquirks import CustomCluster
 from zhaquirks.const import (
     COMMAND_SINGLE,
     DEVICE_TYPE,
@@ -22,7 +23,7 @@
 )
 
 
-class IasZoneDoorbellCluster(IasZone):
+class IasZoneDoorbellCluster(CustomCluster, IasZone):
     """Custom IasZone cluster for the doorbell."""
 
     cluster_id = IasZone.cluster_id

zhaquirks/xiaomi/aqara/roller_curtain_e1.py

@@ -1,7 +1,11 @@
 """Aqara Roller Shade Driver E1 device."""
+from __future__ import annotations
+
+from typing import Any
 
 from zigpy import types as t
 from zigpy.profiles import zha
+from zigpy.zcl import foundation
 from zigpy.zcl.clusters.closures import WindowCovering
 from zigpy.zcl.clusters.general import (
     Alarms,
@@ -20,7 +24,7 @@
 )
 from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
 
-from zhaquirks import Bus, LocalDataCluster
+from zhaquirks import Bus, CustomCluster, LocalDataCluster
 from zhaquirks.const import (
     DEVICE_TYPE,
     ENDPOINTS,
@@ -57,12 +61,12 @@ class XiaomiAqaraRollerE1(XiaomiCluster, ManufacturerSpecificCluster):
     )
 
 
-class AnalogOutputRollerE1(AnalogOutput):
+class AnalogOutputRollerE1(CustomCluster, AnalogOutput):
     """Analog output cluster, only used to relay current_value to WindowCovering."""
 
     cluster_id = AnalogOutput.cluster_id
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Init."""
         super().__init__(*args, **kwargs)
 
@@ -72,7 +76,7 @@ def __init__(self, *args, **kwargs):
         self._update_attribute(0x006A, 1.0)  # resolution
         self._update_attribute(0x006F, 0x00)  # status_flags
 
-    def _update_attribute(self, attrid, value):
+    def _update_attribute(self, attrid: int, value: Any) -> None:
 
         super()._update_attribute(attrid, value)
 
@@ -82,18 +86,25 @@ def _update_attribute(self, attrid, value):
             )
 
 
-class WindowCoveringRollerE1(WindowCovering):
+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, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Init."""
         super().__init__(*args, **kwargs)
 
     async def command(
-        self, command_id, *args, manufacturer=None, expect_reply=True, tsn=None
-    ):
+        self,
+        command_id: foundation.GeneralCommand | int | t.uint8_t,
+        *args: Any,
+        manufacturer: int | t.uint16_t | None = None,
+        expect_reply: bool = True,
+        tries: int = 1,
+        tsn: int | t.uint8_t | None = None,
+        **kwargs: Any,
+    ) -> 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
@@ -104,24 +115,24 @@ async def command(
                 {"present_value": 1}
             )
             return res[0].status
-        elif command_id == DOWN_CLOSE:
+        if command_id == DOWN_CLOSE:
             (res,) = await self.endpoint.multistate_output.write_attributes(
                 {"present_value": 0}
             )
             return res[0].status
-        elif command_id == GO_TO_LIFT_PERCENTAGE:
+        if command_id == GO_TO_LIFT_PERCENTAGE:
             (res,) = await self.endpoint.analog_output.write_attributes(
                 {"present_value": (100 - args[0])}
             )
             return res[0].status
-        elif command_id == STOP:
+        if command_id == STOP:
             (res,) = await self.endpoint.multistate_output.write_attributes(
                 {"present_value": 2}
             )
             return res[0].status
 
 
-class MultistateOutputRollerE1(MultistateOutput):
+class MultistateOutputRollerE1(CustomCluster, MultistateOutput):
     """Multistate Output cluster which overwrites present_value.
 
     Otherwise, it gives errors of wrong datatype when using it in the commands.
@@ -140,12 +151,12 @@ class PowerConfigurationRollerE1(PowerConfiguration, LocalDataCluster):
 
     BATTERY_PERCENTAGE_REMAINING = 0x0021
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Init."""
         super().__init__(*args, **kwargs)
         self.endpoint.device.power_bus_percentage.add_listener(self)
 
-    def update_battery_percentage(self, value):
+    def update_battery_percentage(self, value: int) -> None:
         """Doubles the battery percentage to the Zigbee spec's expected 200% maximum."""
         super()._update_attribute(
             self.BATTERY_PERCENTAGE_REMAINING,
@@ -156,10 +167,10 @@ def update_battery_percentage(self, value):
 class RollerE1AQ(XiaomiCustomDevice):
     """Aqara Roller Shade Driver E1 device."""
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Init."""
-        self.power_bus_percentage = Bus()
-        super().__init__(*args, **kwargs)
+        self.power_bus_percentage: Bus = Bus()  # type: ignore
+        super().__init__(*args, **kwargs)  # type: ignore
 
     signature = {
         MODELS_INFO: [(LUMI, "lumi.curtain.acn002")],