Fix IKEA double battery percentage for old/new firmware (#2948)

We always double the battery pct first if the firmware is unknown (for devices which have the doubling clusters, very new IKEA devices do not at all).
But if the firmware is unknown, we cause a read to be sent (as the device is hopefully awake).
If that read now returns a new firmware, we "undouble" the battery pct again, so it's correct.

We also always read `sw_build_id` on `bind()`

Author: TheJulianJES

tests/test_ikea.py

@@ -2,12 +2,16 @@
 
 from unittest import mock
 
+import pytest
 from zigpy.zcl import foundation
+from zigpy.zcl.clusters.general import Basic, PowerConfiguration
 from zigpy.zcl.clusters.measurement import PM25
 
 import zhaquirks
 import zhaquirks.ikea.starkvind
 
+from tests.common import ClusterListener
+
 zhaquirks.setup()
 
 
@@ -122,3 +126,91 @@ def mock_read(attributes, manufacturer=None):
         assert success
         assert 6 in success.values()
         assert not fail
+
+
+@mock.patch("zigpy.zcl.Cluster.bind", mock.AsyncMock())
+@pytest.mark.parametrize(
+    "firmware, pct_device, pct_correct, expected_pct_updates, expect_log_warning",
+    (
+        ("2.3.075", 50, 100, 1, False),
+        ("24.4.5", 50, 50, 2, False),
+        ("invalid_fw_string", 50, 50, 2, True),
+    ),
+)
+async def test_double_power_config_firmware(
+    caplog,
+    zigpy_device_from_quirk,
+    firmware,
+    pct_device,
+    pct_correct,
+    expected_pct_updates,
+    expect_log_warning,
+):
+    """Test battery percentage remaining is doubled for old firmware."""
+
+    device = zigpy_device_from_quirk(zhaquirks.ikea.fivebtnremote.IkeaTradfriRemote1)
+
+    basic_cluster = device.endpoints[1].basic
+    ClusterListener(basic_cluster)
+    sw_build_id = Basic.AttributeDefs.sw_build_id.id
+
+    power_cluster = device.endpoints[1].power
+    power_listener = ClusterListener(power_cluster)
+    battery_pct_id = PowerConfiguration.AttributeDefs.battery_percentage_remaining.id
+
+    # fake read response for attributes: return plug_read argument for all attributes
+    def mock_read(attributes, manufacturer=None):
+        records = [
+            foundation.ReadAttributeRecord(
+                attr, foundation.Status.SUCCESS, foundation.TypeValue(None, firmware)
+            )
+            for attr in attributes
+        ]
+        return (records,)
+
+    p1 = mock.patch.object(power_cluster, "create_catching_task")
+    p2 = mock.patch.object(
+        basic_cluster, "_read_attributes", mock.AsyncMock(side_effect=mock_read)
+    )
+
+    with p1 as mock_task, p2 as request_mock:
+        # update battery percentage with no firmware in attr cache, check pct doubled for now
+        power_cluster.update_attribute(battery_pct_id, pct_device)
+        assert len(power_listener.attribute_updates) == 1
+        assert power_listener.attribute_updates[0] == (battery_pct_id, pct_device * 2)
+
+        # but also check that sw_build_id read is requested in the background for next update
+        assert mock_task.call_count == 1
+        await mock_task.call_args[0][0]  # await coroutine to read attribute
+        assert request_mock.call_count == 1  # verify request to read sw_build_id
+        assert request_mock.mock_calls[0][1][0][0] == sw_build_id
+
+        # battery pct might be updated again when the attribute read returned new firmware, check pct not doubled then
+        # if firmware turned out to be old or still unknown, do not update battery pct again, as we doubled it already
+        assert len(power_listener.attribute_updates) == expected_pct_updates
+        if expected_pct_updates > 2:
+            assert power_listener.attribute_updates[1] == (battery_pct_id, pct_correct)
+
+        # reset mocks for testing when sw_build_id is known next
+        mock_task.reset_mock()
+        request_mock.reset_mock()
+        power_listener = ClusterListener(power_cluster)
+
+        # update battery percentage with firmware in attr cache, check pct doubled if needed
+        basic_cluster.update_attribute(sw_build_id, firmware)
+        power_cluster.update_attribute(battery_pct_id, pct_device)
+        assert len(power_listener.attribute_updates) == 1
+        assert power_listener.attribute_updates[0] == (battery_pct_id, pct_correct)
+
+        # check no attribute reads were requested when sw_build_id is known
+        assert mock_task.call_count == 0
+        assert request_mock.call_count == 0
+
+        # make sure a call to bind() always reads sw_build_id (e.g. on join or to refresh when repaired/reconfigured)
+        await power_cluster.bind()
+        assert request_mock.call_count == 1
+        assert request_mock.mock_calls[0][1][0][0] == sw_build_id
+
+        # check log output if we expect a warning
+        if expect_log_warning:
+            assert f"sw_build_id is not a number: {firmware} for device" in caplog.text

zhaquirks/ikea/__init__.py

@@ -1,11 +1,14 @@
 """Ikea module."""
+import logging
 
 from zigpy.quirks import CustomCluster
 import zigpy.types as t
 from zigpy.zcl import foundation
-from zigpy.zcl.clusters.general import PowerConfiguration, Scenes
+from zigpy.zcl.clusters.general import Basic, PowerConfiguration, Scenes
 
-from zhaquirks import DoublingPowerConfigurationCluster, EventableCluster
+from zhaquirks import EventableCluster
+
+_LOGGER = logging.getLogger(__name__)
 
 IKEA = "IKEA of Sweden"
 IKEA_CLUSTER_ID = 0xFC7C  # decimal = 64636
@@ -192,25 +195,96 @@ class PowerConfig1CRXCluster(CustomCluster, PowerConfiguration):
 
 
 # doubling IKEA power configuration clusters:
+
+
+class DoublingPowerConfigClusterIKEA(CustomCluster, PowerConfiguration):
+    """PowerConfiguration cluster implementation for IKEA devices.
+
+    This implementation doubles battery pct remaining for IKEA devices with old firmware.
+    """
+
+    async def bind(self):
+        """Bind cluster and read the sw_build_id for later use."""
+        result = await super().bind()
+        await self.endpoint.basic.read_attributes([Basic.AttributeDefs.sw_build_id.id])
+        return result
+
+    def _is_firmware_old(self):
+        """Checks if firmware is old or unknown."""
+        # get sw_build_id from attribute cache if available
+        sw_build_id = self.endpoint.basic.get(Basic.AttributeDefs.sw_build_id.id, None)
+
+        # guard against possible future version formatting which includes more than just numbers
+        try:
+            # if first part of sw_build_id is 24 or higher, then firmware is new
+            if sw_build_id and int(sw_build_id.split(".")[0]) >= 24:
+                return False
+        except ValueError:
+            _LOGGER.warning(
+                "sw_build_id is not a number: %s for device %s",
+                sw_build_id,
+                self.endpoint.device.ieee,
+            )
+            # sw_build_id is not a number, so it must be new firmware
+            return False
+
+        # unknown or old firmware
+        return True
+
+    async def _read_fw_and_update_battery_pct(self, reported_battery_pct):
+        """Read firmware version and update battery percentage remaining if necessary."""
+        # read sw_build_id from device
+        await self.endpoint.basic.read_attributes([Basic.AttributeDefs.sw_build_id.id])
+
+        # check if sw_build_id was read successfully and new firmware is installed
+        # if so, update cache with reported battery percentage (non-doubled)
+        if not self._is_firmware_old():
+            self._update_attribute(
+                PowerConfiguration.AttributeDefs.battery_percentage_remaining.id,
+                reported_battery_pct,
+            )
+
+    def _update_attribute(self, attrid, value):
+        """Update attribute to double battery percentage if firmware is old/unknown.
+
+        If the firmware version is unknown, a background task to read the firmware version is also started,
+        but the percentage is also doubled for now then, as that task happens asynchronously.
+        """
+        if attrid == PowerConfiguration.AttributeDefs.battery_percentage_remaining.id:
+            # if sw_build_id is not cached, create task to read from device, since it should be awake now
+            if (
+                self.endpoint.basic.get(Basic.AttributeDefs.sw_build_id.id, None)
+                is None
+            ):
+                self.create_catching_task(self._read_fw_and_update_battery_pct(value))
+
+            # double percentage if the firmware is old or unknown
+            # the coroutine above will not have executed yet if the firmware is unknown,
+            # so we double for now in that case too, and it updates again later if our doubling was wrong
+            if self._is_firmware_old():
+                value = value * 2
+        super()._update_attribute(attrid, value)
+
+
 class DoublingPowerConfig2AAACluster(
-    DoublingPowerConfigurationCluster, PowerConfig2AAACluster
+    DoublingPowerConfigClusterIKEA, PowerConfig2AAACluster
 ):
     """Doubling power configuration cluster. Updating power attributes: 2 AAA."""
 
 
 class DoublingPowerConfig2CRCluster(
-    DoublingPowerConfigurationCluster, PowerConfig2CRCluster
+    DoublingPowerConfigClusterIKEA, PowerConfig2CRCluster
 ):
     """Doubling power configuration cluster. Updating power attributes: 2 CR2032."""
 
 
 class DoublingPowerConfig1CRCluster(
-    DoublingPowerConfigurationCluster, PowerConfig1CRCluster
+    DoublingPowerConfigClusterIKEA, PowerConfig1CRCluster
 ):
     """Doubling power configuration cluster. Updating power attributes: 1 CR2032."""
 
 
 class DoublingPowerConfig1CRXCluster(
-    DoublingPowerConfigurationCluster, PowerConfig1CRXCluster
+    DoublingPowerConfigClusterIKEA, PowerConfig1CRXCluster
 ):
     """Doubling power configuration cluster. Updating power attributes: 1 CR2032 and zero voltage."""

zhaquirks/ikea/blinds.py

@@ -13,7 +13,6 @@
 )
 from zigpy.zcl.clusters.lightlink import LightLink
 
-from zhaquirks import DoublingPowerConfigurationCluster
 from zhaquirks.const import (
     DEVICE_TYPE,
     ENDPOINTS,
@@ -22,7 +21,7 @@
     OUTPUT_CLUSTERS,
     PROFILE_ID,
 )
-from zhaquirks.ikea import IKEA, IKEA_CLUSTER_ID
+from zhaquirks.ikea import IKEA, IKEA_CLUSTER_ID, DoublingPowerConfigClusterIKEA
 
 
 class IkeaTradfriRollerBlinds(CustomDevice):
@@ -66,7 +65,7 @@ class IkeaTradfriRollerBlinds(CustomDevice):
                 DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE,
                 INPUT_CLUSTERS: [
                     Basic.cluster_id,
-                    DoublingPowerConfigurationCluster,
+                    DoublingPowerConfigClusterIKEA,
                     Identify.cluster_id,
                     Groups.cluster_id,
                     Scenes.cluster_id,

zhaquirks/ikea/fivebtnremote.py

@@ -16,7 +16,6 @@
 from zigpy.zcl.clusters.homeautomation import Diagnostic
 from zigpy.zcl.clusters.lightlink import LightLink
 
-from zhaquirks import DoublingPowerConfigurationCluster
 from zhaquirks.const import (
     CLUSTER_ID,
     COMMAND,
@@ -98,7 +97,7 @@ class IkeaTradfriRemote1(CustomDevice):
                 DEVICE_TYPE: zll.DeviceType.SCENE_CONTROLLER,
                 INPUT_CLUSTERS: [
                     Basic.cluster_id,
-                    DoublingPowerConfigurationCluster,
+                    DoublingPowerConfig1CRCluster,
                     Identify.cluster_id,
                     Alarms.cluster_id,
                     Diagnostic.cluster_id,