Add attribute_converter/value_converter to number entities

tests/test_number.py

@@ -25,6 +25,7 @@
 from zha.application import Platform
 from zha.application.gateway import Gateway
 from zha.application.platforms import EntityCategory, PlatformEntity
+from zha.application.platforms.number import NumberConfigurationEntity
 from zha.application.platforms.number.const import NumberMode
 
 ZIGPY_ANALOG_OUTPUT_DEVICE = {
@@ -430,3 +431,98 @@ async def test_color_number(
     )
     await zha_gateway.async_block_till_done()
     assert entity.state["state"] == initial_value
+
+
+@pytest.mark.parametrize(
+    (
+        "attribute_converter",
+        "value_converter",
+        "initial_value",
+        "expected_native",
+        "set_native",
+        "expected_written",
+    ),
+    (
+        # invert a 0..5 scale where 0 is the maximum (Schneider outlet brightness)
+        (lambda x: 5 - x, lambda x: 5 - int(x), 1, 4, 1, 4),
+        # non-symmetric converters are allowed: each direction is independent
+        (lambda x: x + 100, lambda x: int(x) - 100, 5, 105, 110, 10),
+    ),
+)
+async def test_number_attribute_converter(
+    zha_gateway: Gateway,
+    attribute_converter,
+    value_converter,
+    initial_value: int,
+    expected_native: int,
+    set_native: int,
+    expected_written: int,
+) -> None:
+    """Test number entity attribute_converter/value_converter round trip."""
+
+    light = await light_mock(zha_gateway)
+    cluster = light.endpoints[1].on_off
+    cluster.PLUGGED_ATTR_READS = {"on_time": initial_value}
+    update_attribute_cache(cluster)
+    zha_device = await join_zigpy_device(zha_gateway, light)
+
+    entity = NumberConfigurationEntity(
+        endpoint=zha_device.endpoints[1],
+        device=zha_device,
+        cluster=cluster,
+        from_quirk=True,
+        fallback_name="Test number",
+        translation_key="test_number",
+        attribute_name="on_time",
+        attribute_converter=attribute_converter,
+        value_converter=value_converter,
+        min_value=0,
+        max_value=255,
+    )
+
+    # the raw attribute value is converted for display
+    assert entity.native_value == expected_native
+    assert entity.state["state"] == expected_native
+
+    # the value from HA is converted back before it is written
+    cluster.write_attributes.reset_mock()
+    await entity.async_set_native_value(set_native)
+    assert cluster.write_attributes.mock_calls == [
+        call({"on_time": expected_written}, manufacturer=UNDEFINED),
+    ]
+
+
+async def test_number_attribute_converter_ignores_multiplier(
+    zha_gateway: Gateway,
+) -> None:
+    """Test that converters take precedence over the multiplier, as for sensors."""
+
+    light = await light_mock(zha_gateway)
+    cluster = light.endpoints[1].on_off
+    cluster.PLUGGED_ATTR_READS = {"on_time": 10}
+    update_attribute_cache(cluster)
+    zha_device = await join_zigpy_device(zha_gateway, light)
+
+    entity = NumberConfigurationEntity(
+        endpoint=zha_device.endpoints[1],
+        device=zha_device,
+        cluster=cluster,
+        from_quirk=True,
+        fallback_name="Test number",
+        translation_key="test_number",
+        attribute_name="on_time",
+        attribute_converter=lambda x: 5 - x,
+        value_converter=lambda x: 5 - int(x),
+        multiplier=0.1,
+        min_value=0,
+        max_value=255,
+    )
+
+    # 10 * 0.1 would be 1.0 if the multiplier applied, but the converter wins
+    assert entity.native_value == -5
+
+    cluster.write_attributes.reset_mock()
+    await entity.async_set_native_value(1)
+    assert cluster.write_attributes.mock_calls == [
+        call({"on_time": 4}, manufacturer=UNDEFINED),
+    ]

zha/application/platforms/number/__init__.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Callable
 from dataclasses import dataclass
 import functools
 import logging
@@ -259,13 +260,17 @@ class NumberConfigurationEntity(BaseNumber):
     _attr_native_step: float = 1.0
     _multiplier: float = 1
     _attribute_name: str
+    _attribute_converter: Callable[[Any], Any] | None = None
+    _value_converter: Callable[[Any], Any] | None = None
 
     def __init__(
         self,
         endpoint: Endpoint,
         device: Device,
         *,
         attribute_name: str | None = None,
+        attribute_converter: Callable[[Any], Any] | None = None,
+        value_converter: Callable[[Any], Any] | None = None,
         min_value: float | None = None,
         max_value: float | None = None,
         step: float | None = None,
@@ -278,6 +283,10 @@ def __init__(
         """Init this number configuration entity."""
         if attribute_name is not None:
             self._attribute_name = attribute_name
+        if attribute_converter is not None:
+            self._attribute_converter = attribute_converter
+        if value_converter is not None:
+            self._value_converter = value_converter
         if min_value is not None:
             self._attr_native_min_value = min_value
         if max_value is not None:
@@ -344,13 +353,17 @@ def native_value(self) -> float | None:
         value = self._cluster.get(self._attribute_name)
         if value is None:
             return None
+        if self._attribute_converter:
+            return self._attribute_converter(value)
         return value * self._multiplier
 
     async def async_set_native_value(self, value: float) -> None:
         """Update the current value from HA."""
-        await write_attributes_safe(
-            self._cluster, {self._attribute_name: int(value / self._multiplier)}
-        )
+        if self._value_converter:
+            attr_value = self._value_converter(value)
+        else:
+            attr_value = int(value / self._multiplier)
+        await write_attributes_safe(self._cluster, {self._attribute_name: attr_value})
         self.maybe_emit_state_changed_event()
 
     async def async_update(self) -> None: