diff --git a/tests/test_sensor.py b/tests/test_sensor.py index ab9c40be..1213be0d 100644 --- a/tests/test_sensor.py +++ b/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 diff --git a/zha/application/platforms/sensor/__init__.py b/zha/application/platforms/sensor/__init__.py index 9838f7de..fc793ba4 100644 --- a/zha/application/platforms/sensor/__init__.py +++ b/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 ( @@ -539,7 +544,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") ) @@ -569,6 +574,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) diff --git a/zha/application/platforms/sensor/const.py b/zha/application/platforms/sensor/const.py index 70644788..b9f55cbe 100644 --- a/zha/application/platforms/sensor/const.py +++ b/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 diff --git a/zha/zigbee/cluster_handlers/general.py b/zha/zigbee/cluster_handlers/general.py index c0dcc8fd..1cc90672 100644 --- a/zha/zigbee/cluster_handlers/general.py +++ b/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):