Clean up Multistate Input, better tests

tests/test_sensor.py

@@ -17,12 +17,7 @@
 from zigpy.zcl.clusters import general, homeautomation, hvac, measurement, smartenergy
 from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
 
-from tests.common import (
-    find_entity,
-    get_entity,
-    send_attributes_report,
-    update_attribute_cache,
-)
+from tests.common import get_entity, send_attributes_report
 from tests.conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
 from zha.application import Platform
 from zha.application.const import ZHA_CLUSTER_HANDLER_READS_PER_REQ
@@ -391,6 +386,65 @@ async def async_test_pi_heating_demand(
     assert_state(entity, 1, "%")
 
 
+async def async_test_general_analog_input(
+    zha_gateway: Gateway, cluster: Cluster, entity: PlatformEntity
+):
+    """Test general analog input."""
+    await entity.async_update()
+
+    assert entity.device_class == SensorDeviceClass.HUMIDITY.value
+
+    if entity._cluster_handler.resolution is not None:
+        assert entity.suggested_display_precision == 1
+    else:
+        assert entity.suggested_display_precision is None
+
+    assert entity._cluster_handler.max_present_value == 100.0
+    assert entity._cluster_handler.min_present_value == 1.0
+    assert entity._cluster_handler.out_of_service == 0
+    assert entity._cluster_handler.reliability == 0
+    assert entity._cluster_handler.status_flags == 0
+    assert entity._cluster_handler.application_type == 0x00010000
+    assert entity._cluster_handler.present_value == 1.0
+
+    await send_attributes_report(
+        zha_gateway, cluster, {general.AnalogInput.AttributeDefs.present_value.id: 1.0}
+    )
+    assert_state(entity, 1.0, "%")
+
+
+async def async_test_general_multistate_input(
+    zha_gateway: Gateway, cluster: Cluster, entity: PlatformEntity
+):
+    """Test general multistate input."""
+    await entity.async_update()
+
+    assert entity._cluster_handler.number_of_states == 2
+    assert entity._cluster_handler.out_of_service == 0
+    assert entity._cluster_handler.present_value == 1
+    assert entity._cluster_handler.reliability == 0
+    assert entity._cluster_handler.status_flags == 0
+    assert entity._cluster_handler.application_type == 0x00000009
+
+    if entity._cluster_handler.state_text is None:
+        assert_state(entity, "state_1", None)
+        await send_attributes_report(
+            zha_gateway,
+            cluster,
+            {general.MultistateInput.AttributeDefs.present_value.id: 2},
+        )
+        assert_state(entity, "state_2", None)
+    else:
+        assert entity._cluster_handler.state_text == ["Night", "Day", "Hold"]
+        assert_state(entity, "Night", None)
+        await send_attributes_report(
+            zha_gateway,
+            cluster,
+            {general.MultistateInput.AttributeDefs.present_value.id: 2},
+        )
+        assert_state(entity, "Day", None)
+
+
 @pytest.mark.parametrize(
     "cluster_id, entity_type, test_func, read_plug, unsupported_attrs",
     (
@@ -544,6 +598,72 @@ async def async_test_pi_heating_demand(
             None,
             None,
         ),
+        (
+            general.AnalogInput.cluster_id,
+            sensor.AnalogInputSensor,
+            async_test_general_analog_input,
+            {
+                "present_value": 1.0,
+                "description": "Analog Input",
+                "max_present_value": 100.0,
+                "min_present_value": 1.0,
+                "out_of_service": 0,
+                "reliability": 0,
+                "resolution": 1.1,
+                "status_flags": 0,
+                "engineering_units": 98,
+                "application_type": 0x00010000,
+            },
+            None,
+        ),
+        (
+            general.AnalogInput.cluster_id,
+            sensor.AnalogInputSensor,
+            async_test_general_analog_input,
+            {
+                "present_value": 1.0,
+                "description": "Analog Input",
+                "max_present_value": 100.0,
+                "min_present_value": 1.0,
+                "out_of_service": 0,
+                "reliability": 0,
+                "status_flags": 0,
+                "engineering_units": 98,
+                "application_type": 0x00010000,
+            },
+            None,
+        ),
+        (
+            general.MultistateInput.cluster_id,
+            sensor.MultiStateInputSensor,
+            async_test_general_multistate_input,
+            {
+                "state_text": t.LVList(["Night", "Day", "Hold"]),
+                "description": "Multistate Input",
+                "number_of_states": 2,
+                "out_of_service": 0,
+                "present_value": 1,
+                "reliability": 0,
+                "status_flags": 0,
+                "application_type": 0x00000009,
+            },
+            None,
+        ),
+        (
+            general.MultistateInput.cluster_id,
+            sensor.MultiStateInputSensor,
+            async_test_general_multistate_input,
+            {
+                "description": "Multistate Input",
+                "number_of_states": 2,
+                "out_of_service": 0,
+                "present_value": 1,
+                "reliability": 0,
+                "status_flags": 0,
+                "application_type": 0x00000009,
+            },
+            None,
+        ),
     ),
 )
 async def test_sensor(
@@ -586,7 +706,12 @@ async def test_sensor(
     )
 
     await zha_gateway.async_block_till_done()
-    assert entity.fallback_name is None
+
+    if read_plug and read_plug.get("description", None):
+        assert entity.fallback_name == read_plug.get("description", None)
+        assert entity.translation_key is None
+    else:
+        assert entity.fallback_name is None
     # test sensor associated logic
     await test_func(zha_gateway, cluster, entity)
 
@@ -1377,32 +1502,3 @@ async def test_danfoss_thermostat_sw_error(
     assert entity.extra_state_attribute_names
     assert "Top_pcb_sensor_error" in entity.extra_state_attribute_names
     assert entity.state["Top_pcb_sensor_error"]
-
-
-async def test_sensor_general(
-    zigpy_device_mock: Callable[..., ZigpyDevice],
-    device_joined: Callable[[ZigpyDevice], Awaitable[Device]],
-    zha_gateway: Gateway,
-) -> None:
-    """Test sensor general - description."""
-    DEVICE_GENERAL = {
-        1: {
-            SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
-            SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.COMBINED_INTERFACE,
-            SIG_EP_INPUT: [general.AnalogInput.cluster_id],
-            SIG_EP_OUTPUT: [],
-        }
-    }
-
-    zigpy_device = zigpy_device_mock(DEVICE_GENERAL)
-
-    cluster = getattr(zigpy_device.endpoints[1], "analog_input")
-    cluster.PLUGGED_ATTR_READS = {"description": "Analog Input", "present_value": 1.0}
-    update_attribute_cache(cluster)
-    zha_device = await device_joined(zigpy_device)
-    entity: PlatformEntity = find_entity(zha_device, Platform.SENSOR)
-
-    await entity.async_update()
-    await zha_gateway.async_block_till_done()
-    assert entity.fallback_name == "Analog Input"
-    assert entity.translation_key is None

zha/application/platforms/sensor/__init__.py

@@ -7,6 +7,7 @@
 import enum
 import functools
 import logging
+import math
 import numbers
 from typing import TYPE_CHECKING, Any, Self
 
@@ -30,7 +31,11 @@
 from zha.application.platforms.climate.const import HVACAction
 from zha.application.platforms.helpers import validate_device_class
 from zha.application.platforms.number.const import UNITS
-from zha.application.platforms.sensor.const import SensorDeviceClass, SensorStateClass
+from zha.application.platforms.sensor.const import (
+    AnalogInputStateClass,
+    SensorDeviceClass,
+    SensorStateClass,
+)
 from zha.application.registries import PLATFORM_ENTITIES
 from zha.decorators import periodic
 from zha.units import (
@@ -538,7 +543,7 @@ def __init__(
             self._enum = enum.Enum(  # type: ignore [misc]
                 "state_text",
                 [
-                    (f"state_{i}", i)
+                    (f"state_{i+1}", i + 1)
                     for i in range(
                         self._cluster_handler.cluster.get("number_of_states")
                     )
@@ -568,6 +573,27 @@ def __init__(
         super().__init__(unique_id, cluster_handlers, endpoint, device, **kwargs)
         engineering_units = self._cluster_handler.engineering_units
         self._attr_native_unit_of_measurement = UNITS.get(engineering_units)
+        self._attr_state_class = SensorStateClass.MEASUREMENT
+
+    @property
+    def device_class(self) -> str | None:
+        """Return the device class."""
+        if self._cluster_handler.application_type is not None:
+            device_type = (self._cluster_handler.application_type >> 16) & 0xFF
+            return AnalogInputStateClass.device_class(device_type)
+        return None
+
+    @property
+    def suggested_display_precision(self) -> int | None:
+        """Return the the display precision."""
+        if self._cluster_handler.resolution is not None:
+            return math.ceil(
+                -math.log10(
+                    abs(self._cluster_handler.resolution)
+                    - abs(math.floor(self._cluster_handler.resolution))
+                )
+            )
+        return None
 
 
 @MULTI_MATCH(cluster_handler_names=CLUSTER_HANDLER_POWER_CONFIGURATION)

zha/application/platforms/sensor/const.py

@@ -390,3 +390,36 @@ class SensorDeviceClass(enum.StrEnum):
     SensorDeviceClass.ENUM,
     SensorDeviceClass.TIMESTAMP,
 }
+
+
+class AnalogInputStateClass(enum.Enum):
+    """State class for AnalogInput Types."""
+
+    TEMPERATURE = (0x00, SensorDeviceClass.TEMPERATURE)
+    RELATIVE_HUMIDITY = (0x01, SensorDeviceClass.HUMIDITY)
+    PRESSURE = (0x02, SensorDeviceClass.PRESSURE)
+    FLOW = (0x03, SensorDeviceClass.VOLUME_FLOW_RATE)
+    PERCENTAGE = (0x04, None)
+    PARTS_PER_MILLION = (0x05, SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS)
+    RPM = (0x06, None)
+    CURRENT = (0x07, SensorDeviceClass.CURRENT)
+    FREQUENCY = (0x08, SensorDeviceClass.FREQUENCY)
+    POWER_WATTS = (0x09, SensorDeviceClass.POWER)
+    POWER_KILOWATTS = (0x0A, SensorDeviceClass.POWER)
+    ENERGY = (0x0B, SensorDeviceClass.ENERGY)
+    COUNT = (0x0C, None)
+    ENTHALPY = (0x0D, None)
+    TIME_SECONDS = (0x0E, None)
+
+    def __init__(self, type, dev_class):
+        """Init this enum."""
+        self.type = type
+        self.dev_class = dev_class
+
+    @classmethod
+    def device_class(cls, type):
+        """Return the device class given a type."""
+        for entry in cls:
+            if entry.type == type:
+                return entry.dev_class.value
+        return None

zha/zigbee/cluster_handlers/general.py

@@ -100,6 +100,7 @@ class AnalogInputClusterHandler(ClusterHandler):
         AnalogInput.AttributeDefs.out_of_service.name: True,
         AnalogInput.AttributeDefs.reliability.name: True,
         AnalogInput.AttributeDefs.resolution.name: True,
+        AnalogInput.AttributeDefs.status_flags.name: True,
         AnalogInput.AttributeDefs.engineering_units.name: True,
         AnalogInput.AttributeDefs.application_type.name: True,
     }
@@ -467,14 +468,63 @@ class MultistateInputClusterHandler(ClusterHandler):
     )
 
     ZCL_INIT_ATTRS = {
+        MultistateInput.AttributeDefs.state_text.name: True,
+        MultistateInput.AttributeDefs.description.name: True,
+        MultistateInput.AttributeDefs.number_of_states.name: True,
+        MultistateInput.AttributeDefs.out_of_service.name: True,
+        MultistateInput.AttributeDefs.present_value.name: True,
+        MultistateInput.AttributeDefs.reliability.name: True,
+        MultistateInput.AttributeDefs.status_flags.name: True,
+        MultistateInput.AttributeDefs.application_type.name: True,
         MultistateInput.AttributeDefs.description.name: True,
     }
 
+    @property
+    def state_text(self) -> t.LVList | None:
+        """Return cached value of state text."""
+        return self.cluster.get(MultistateInput.AttributeDefs.state_text.name)
+
     @property
     def description(self) -> str | None:
         """Return cached value of description."""
         return self.cluster.get(MultistateInput.AttributeDefs.description.name)
 
+    @property
+    def number_of_states(self) -> int:
+        """Return cached value of number of states."""
+        return self.cluster.get(MultistateInput.AttributeDefs.number_of_states.name)
+
+    @property
+    def out_of_service(self) -> bool | None:
+        """Return cached value of out of service."""
+        return self.cluster.get(AnalogInput.AttributeDefs.out_of_service.name)
+
+    @property
+    def present_value(self) -> int:
+        """Return cached value of present value."""
+        return self.cluster.get(AnalogInput.AttributeDefs.present_value.name)
+
+    @property
+    def reliability(self) -> int | None:
+        """Return cached value of reliability."""
+        return self.cluster.get(AnalogInput.AttributeDefs.reliability.name)
+
+    @property
+    def status_flags(self) -> int | None:
+        """Return cached value of status flags."""
+        return self.cluster.get(AnalogInput.AttributeDefs.status_flags.name)
+
+    @property
+    def application_type(self) -> int | None:
+        """Return cached value of application type."""
+        return self.cluster.get(AnalogInput.AttributeDefs.application_type.name)
+
+    async def async_update(self):
+        """Update cluster value attribute."""
+        await self.get_attribute_value(
+            AnalogInput.AttributeDefs.present_value.name, from_cache=False
+        )
+
 
 @registries.CLUSTER_HANDLER_REGISTRY.register(MultistateOutput.cluster_id)
 class MultistateOutputClusterHandler(ClusterHandler):