Merge branch 'release/0.0.31'

README.md

@@ -116,6 +116,7 @@ If you are looking to make your first code contribution to this project then we
 - Some functionality requires a coordinator device to be XBee as well
 - GPIO pins are exposed to Home Assistant as switches
 - Analog inputs are exposed as sensors
+- PWM output on XBee3 can be controlled by writing 0x0055 (present_value) cluster attribute with `zha.set_zigbee_cluster_attribute` service
 - Outgoing UART data can be sent with `zha.issue_zigbee_cluster_command` service
 - Incoming UART data will generate `zha_event` event.
 

setup.py

@@ -2,7 +2,7 @@
 
 from setuptools import find_packages, setup
 
-VERSION = "0.0.30"
+VERSION = "0.0.31"
 
 
 def readme():

zhaquirks/__init__.py

@@ -4,8 +4,8 @@
 import pkgutil
 
 from zigpy.quirks import CustomCluster
-import zigpy.types as types
 from zigpy.util import ListenableMixin
+from zigpy.zcl import foundation
 from zigpy.zcl.clusters.general import PowerConfiguration
 from zigpy.zdo import types as zdotypes
 
@@ -35,9 +35,28 @@ class LocalDataCluster(CustomCluster):
 
     async def read_attributes_raw(self, attributes, manufacturer=None):
         """Prevent remote reads."""
-        attributes = [types.uint16_t(a) for a in attributes]
-        values = [self._attr_cache.get(attr) for attr in attributes]
-        return values
+        records = [
+            foundation.ReadAttributeRecord(
+                attr, foundation.Status.UNSUPPORTED_ATTRIBUTE, foundation.TypeValue()
+            )
+            for attr in attributes
+        ]
+        for record in records:
+            record.value.value = self._attr_cache.get(record.attrid)
+            if record.value.value is not None:
+                record.status = foundation.Status.SUCCESS
+        return (records,)
+
+    async def write_attributes(self, attributes, manufacturer=None):
+        """Prevent remote writes."""
+        for attrid, value in attributes.items():
+            if isinstance(attrid, str):
+                attrid = self._attridx[attrid]
+            if attrid not in self.attributes:
+                self.error("%d is not a valid attribute id", attrid)
+                continue
+            self._update_attribute(attrid, value)
+        return (foundation.Status.SUCCESS,)
 
 
 class EventableCluster(CustomCluster):

zhaquirks/ikea/motionzha.py

@@ -1,13 +1,15 @@
 """Device handler for IKEA of Sweden TRADFRI remote control."""
 from zigpy.profiles import zha
-from zigpy.quirks import CustomDevice
+from zigpy.quirks import CustomCluster, CustomDevice
 from zigpy.zcl.clusters.general import (
     Alarms,
     Basic,
     Groups,
     Identify,
+    LevelControl,
     OnOff,
     Ota,
+    PollControl,
     PowerConfiguration,
 )
 from zigpy.zcl.clusters.lightlink import LightLink
@@ -23,9 +25,24 @@
 )
 from . import IKEA, LightLinkCluster
 
+IKEA_CLUSTER_ID = 0xFC7C  # decimal = 64636
 DIAGNOSTICS_CLUSTER_ID = 0x0B05  # decimal = 2821
 
 
+class LightLinkClusterNew(CustomCluster, LightLink):
+    """Ikea LightLink cluster."""
+
+    async def bind(self):
+        """Bind LightLink cluster to coordinator."""
+        application = self._endpoint.device.application
+        try:
+            coordinator = application.get_device(application.ieee)
+        except KeyError:
+            return
+        status = await coordinator.add_to_group(0x0000)
+        return [status]
+
+
 class IkeaTradfriMotion(CustomDevice):
     """Custom device representing IKEA of Sweden TRADFRI remote control."""
 
@@ -81,3 +98,64 @@ class IkeaTradfriMotion(CustomDevice):
             }
         }
     }
+
+
+class IkeaTradfriMotionE1745(CustomDevice):
+    """Custom device representing IKEA of Sweden TRADFRI motion sensor E1745."""
+
+    signature = {
+        # <SimpleDescriptor endpoint=1 profile=260 device_type=2128
+        # device_version=1
+        # input_clusters=[0, 1, 3, 9, 32, 4096, 64636]
+        # output_clusters=[3, 4, 6, 8, 25, 4096]>
+        MODELS_INFO: [(IKEA, "TRADFRI motion sensor")],
+        ENDPOINTS: {
+            1: {
+                PROFILE_ID: zha.PROFILE_ID,
+                DEVICE_TYPE: zha.DeviceType.ON_OFF_SENSOR,
+                INPUT_CLUSTERS: [
+                    Basic.cluster_id,
+                    PowerConfiguration.cluster_id,
+                    Identify.cluster_id,
+                    Alarms.cluster_id,
+                    PollControl.cluster_id,
+                    LightLink.cluster_id,
+                    IKEA_CLUSTER_ID,
+                ],
+                OUTPUT_CLUSTERS: [
+                    Identify.cluster_id,
+                    Groups.cluster_id,
+                    OnOff.cluster_id,
+                    LevelControl.cluster_id,
+                    Ota.cluster_id,
+                    LightLink.cluster_id,
+                ],
+            }
+        },
+    }
+
+    replacement = {
+        ENDPOINTS: {
+            1: {
+                PROFILE_ID: zha.PROFILE_ID,
+                DEVICE_TYPE: zha.DeviceType.ON_OFF_SENSOR,
+                INPUT_CLUSTERS: [
+                    Basic.cluster_id,
+                    DoublingPowerConfigurationCluster,
+                    Identify.cluster_id,
+                    Alarms.cluster_id,
+                    PollControl.cluster_id,
+                    LightLinkClusterNew,
+                    IKEA_CLUSTER_ID,
+                ],
+                OUTPUT_CLUSTERS: [
+                    Identify.cluster_id,
+                    Groups.cluster_id,
+                    OnOff.cluster_id,
+                    LevelControl.cluster_id,
+                    Ota.cluster_id,
+                    LightLink.cluster_id,
+                ],
+            }
+        }
+    }

zhaquirks/ikea/twobtnremote.py

@@ -1,5 +1,5 @@
 """Device handler for IKEA of Sweden TRADFRI remote control."""
-from zigpy.profiles import zha
+from zigpy.profiles import zha, zll
 from zigpy.quirks import CustomCluster, CustomDevice
 from zigpy.zcl.clusters.closures import WindowCovering
 from zigpy.zcl.clusters.general import (
@@ -135,3 +135,51 @@ class IkeaTradfriRemote2Btn(CustomDevice):
             ARGS: [1, 83],
         },
     }
+
+
+class IkeaTradfriRemote2BtnZLL(CustomDevice):
+    """Custom device representing IKEA of Sweden TRADFRI remote control."""
+
+    signature = {
+        # <SimpleDescriptor endpoint=1 profile=49246 device_type=2080
+        # device_version=248
+        # input_clusters=[0, 1, 3, 9, 258, 4096, 64636]
+        # output_clusters=[3, 4, 6, 8, 25, 258, 4096]>
+        MODELS_INFO: IkeaTradfriRemote2Btn.signature[MODELS_INFO].copy(),
+        ENDPOINTS: {
+            1: {
+                PROFILE_ID: zll.PROFILE_ID,
+                DEVICE_TYPE: zll.DeviceType.CONTROLLER,
+                INPUT_CLUSTERS: [
+                    Basic.cluster_id,
+                    PowerConfiguration.cluster_id,
+                    Identify.cluster_id,
+                    Alarms.cluster_id,
+                    WindowCovering.cluster_id,
+                    LightLink.cluster_id,
+                    IKEA_CLUSTER_ID,
+                ],
+                OUTPUT_CLUSTERS: IkeaTradfriRemote2Btn.signature[ENDPOINTS][1][
+                    OUTPUT_CLUSTERS
+                ].copy(),
+            }
+        },
+    }
+    signature[ENDPOINTS][1][INPUT_CLUSTERS].append(WindowCovering.cluster_id)
+
+    replacement = {
+        ENDPOINTS: {
+            1: {
+                PROFILE_ID: zll.PROFILE_ID,
+                DEVICE_TYPE: zll.DeviceType.CONTROLLER,
+                INPUT_CLUSTERS: IkeaTradfriRemote2Btn.replacement[ENDPOINTS][1][
+                    INPUT_CLUSTERS
+                ].copy(),
+                OUTPUT_CLUSTERS: IkeaTradfriRemote2Btn.replacement[ENDPOINTS][1][
+                    OUTPUT_CLUSTERS
+                ].copy(),
+            }
+        }
+    }
+
+    device_automation_triggers = IkeaTradfriRemote2Btn.device_automation_triggers.copy()

zhaquirks/xbee/__init__.py

@@ -22,7 +22,13 @@
 from zigpy.quirks import CustomCluster, CustomDevice
 import zigpy.types as t
 from zigpy.zcl import foundation
-from zigpy.zcl.clusters.general import AnalogInput, BinaryInput, LevelControl, OnOff
+from zigpy.zcl.clusters.general import (
+    AnalogInput,
+    AnalogOutput,
+    BinaryInput,
+    LevelControl,
+    OnOff,
+)
 
 from .. import EventableCluster, LocalDataCluster
 from ..const import ENDPOINTS, INPUT_CLUSTERS, OUTPUT_CLUSTERS
@@ -40,6 +46,8 @@
 XBEE_PROFILE_ID = 0xC105
 XBEE_REMOTE_AT = 0x17
 XBEE_SRC_ENDPOINT = 0xE8
+ATTR_PRESENT_VALUE = 0x0055
+PIN_ANALOG_OUTPUT = 2
 
 
 class IOSample(bytes):
@@ -125,26 +133,27 @@ def deserialize(cls, data):
 }
 
 
+ENDPOINT_TO_AT = {
+    0xD0: "D0",
+    0xD1: "D1",
+    0xD2: "D2",
+    0xD3: "D3",
+    0xD4: "D4",
+    0xD5: "D5",
+    0xD8: "D8",
+    0xD9: "D9",
+    0xDA: "P0",
+    0xDB: "P1",
+    0xDC: "P2",
+}
+
+
 class XBeeOnOff(CustomCluster, OnOff):
     """XBee on/off cluster."""
 
-    ep_id_2_pin = {
-        0xD0: "D0",
-        0xD1: "D1",
-        0xD2: "D2",
-        0xD3: "D3",
-        0xD4: "D4",
-        0xD5: "D5",
-        0xD8: "D8",
-        0xD9: "D9",
-        0xDA: "P0",
-        0xDB: "P1",
-        0xDC: "P2",
-    }
-
     async def command(self, command, *args, manufacturer=None, expect_reply=True):
         """Xbee change pin state command, requires zigpy_xbee."""
-        pin_name = self.ep_id_2_pin.get(self._endpoint.endpoint_id)
+        pin_name = ENDPOINT_TO_AT.get(self._endpoint.endpoint_id)
         if command not in [0, 1] or pin_name is None:
             return super().command(command, *args)
         if command == 0:
@@ -155,6 +164,47 @@ async def command(self, command, *args, manufacturer=None, expect_reply=True):
         return 0, foundation.Status.SUCCESS
 
 
+class XBeePWM(LocalDataCluster, AnalogOutput):
+    """XBee PWM Cluster."""
+
+    ep_id_2_pwm = {0xDA: "M0", 0xDB: "M1"}
+
+    def __init__(self, endpoint, is_server=True):
+        """Set known attributes and store them in cache."""
+        super().__init__(endpoint, is_server)
+        self._update_attribute(0x0041, float(0x03FF))  # 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
+
+    async def write_attributes(self, attributes, manufacturer=None):
+        """Intercept present_value attribute write."""
+        if ATTR_PRESENT_VALUE in attributes:
+            duty_cycle = int(round(float(attributes.pop(ATTR_PRESENT_VALUE))))
+            at_command = self.ep_id_2_pwm.get(self._endpoint.endpoint_id)
+            result = await self._endpoint.device.remote_at(at_command, duty_cycle)
+            if result != foundation.Status.SUCCESS:
+                return result
+
+            at_command = ENDPOINT_TO_AT.get(self._endpoint.endpoint_id)
+            result = await self._endpoint.device.remote_at(
+                at_command, PIN_ANALOG_OUTPUT
+            )
+            if result != foundation.Status.SUCCESS or not attributes:
+                return result
+
+        return await super().write_attributes(attributes, manufacturer)
+
+    async def read_attributes_raw(self, attributes, manufacturer=None):
+        """Intercept present_value attribute read."""
+        if ATTR_PRESENT_VALUE in attributes:
+            at_command = self.ep_id_2_pwm.get(self._endpoint.endpoint_id)
+            result = await self._endpoint.device.remote_at(at_command)
+            self._update_attribute(ATTR_PRESENT_VALUE, float(result))
+        return await super().read_attributes_raw(attributes, manufacturer)
+
+
 class XBeeCommon(CustomDevice):
     """XBee common class."""
 
@@ -200,7 +250,7 @@ def handle_cluster_request(self, tsn, command_id, args):
                         self._endpoint.device.__getitem__(
                             ENDPOINT_MAP[pin]
                         ).__getattr__(AnalogInput.ep_attribute)._update_attribute(
-                            0x0055,  # "present_value"
+                            ATTR_PRESENT_VALUE,
                             values["analog_samples"][pin]
                             / (10.23 if pin != 7 else 1000),  # supply voltage is in mV
                         )

zhaquirks/xbee/xbee3_io.py

@@ -3,7 +3,7 @@
 from zigpy.profiles import zha
 from zigpy.zcl.clusters.general import AnalogInput
 
-from . import XBEE_PROFILE_ID, XBeeCommon, XBeeOnOff
+from . import XBEE_PROFILE_ID, XBeeCommon, XBeeOnOff, XBeePWM
 from ..const import DEVICE_TYPE, ENDPOINTS, INPUT_CLUSTERS, OUTPUT_CLUSTERS, PROFILE_ID
 
 
@@ -91,15 +91,15 @@ def __init__(self, application, ieee, nwk, replaces):
                     "model": "DIO10/PWM0",
                     DEVICE_TYPE: zha.DeviceType.LEVEL_CONTROL_SWITCH,
                     PROFILE_ID: XBEE_PROFILE_ID,
-                    INPUT_CLUSTERS: [XBeeOnOff],
+                    INPUT_CLUSTERS: [XBeeOnOff, XBeePWM],
                     OUTPUT_CLUSTERS: [],
                 },
                 0xDB: {
                     "manufacturer": "XBEE",
                     "model": "DIO11/PWM1",
                     DEVICE_TYPE: zha.DeviceType.LEVEL_CONTROL_SWITCH,
                     PROFILE_ID: XBEE_PROFILE_ID,
-                    INPUT_CLUSTERS: [XBeeOnOff],
+                    INPUT_CLUSTERS: [XBeeOnOff, XBeePWM],
                     OUTPUT_CLUSTERS: [],
                 },
                 0xDC: {

zhaquirks/xiaomi/__init__.py

@@ -264,9 +264,8 @@ def _parse_mija_attributes(self, value):
     @staticmethod
     def _calculate_remaining_battery_percentage(voltage):
         """Calculate percentage."""
-        # Min/Max values from https://github.com/louisZL/lumi-gateway-local-api
         min_voltage = 2800
-        max_voltage = 3300
+        max_voltage = 3000
         percent = (voltage - min_voltage) / (max_voltage - min_voltage) * 200
         return min(200, percent)