Add Timestamp Sensor (zigpy#224)

Use Subclass Sensor - Add Timestamp Sensor

tests/test_sensor.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
 import math
 from typing import Any, Optional
 from unittest.mock import AsyncMock, MagicMock
@@ -394,6 +395,18 @@ async def async_test_pi_heating_demand(
     assert_state(entity, 1, "%")
 
 
+async def async_test_change_source_timestamp(
+    zha_gateway: Gateway, cluster: Cluster, entity: PlatformEntity
+):
+    """Test change source timestamp is correctly returned."""
+    await send_attributes_report(
+        zha_gateway,
+        cluster,
+        {hvac.Thermostat.AttributeDefs.setpoint_change_source_timestamp.id: 2674725315},
+    )
+    assert entity.state["state"] == datetime(2024, 10, 4, 11, 15, 15, tzinfo=UTC)
+
+
 @pytest.mark.parametrize(
     "cluster_id, entity_type, test_func, read_plug, unsupported_attrs",
     (
@@ -547,6 +560,13 @@ async def async_test_pi_heating_demand(
             None,
             None,
         ),
+        (
+            hvac.Thermostat.cluster_id,
+            sensor.SetpointChangeSourceTimestamp,
+            async_test_change_source_timestamp,
+            None,
+            None,
+        ),
     ),
 )
 async def test_sensor(
@@ -1126,6 +1146,78 @@ async def test_elec_measurement_skip_unsupported_attribute(
     assert read_attrs == supported_attributes
 
 
+class TimestampCluster(CustomCluster, ManufacturerSpecificCluster):
+    """Timestamp Quirk V2 Cluster."""
+
+    cluster_id = 0xEF00
+    ep_attribute = "time_test_cluster"
+    attributes = {
+        0xEF65: ("start_time", t.uint32_t, True),
+    }
+
+    def __init__(self, *args, **kwargs) -> None:
+        """Initialize."""
+        super().__init__(*args, **kwargs)
+        # populate cache to create config entity
+        self._attr_cache.update({0xEF65: 10})
+
+
+(
+    QuirkBuilder("Fake_Timestamp_sensor", "Fake_Model_sensor")
+    .replaces(TimestampCluster)
+    .sensor(
+        "start_time",
+        TimestampCluster.cluster_id,
+        device_class=SensorDeviceClass.TIMESTAMP,
+        translation_key="start_time",
+        fallback_name="Start Time",
+    )
+    .add_to_registry()
+)
+
+
+@pytest.fixture
+async def zigpy_device_timestamp_sensor_v2(
+    zha_gateway: Gateway,  # pylint: disable=unused-argument
+):
+    """Timestamp Test device."""
+
+    zigpy_device = create_mock_zigpy_device(
+        zha_gateway,
+        {
+            1: {
+                SIG_EP_INPUT: [
+                    general.Basic.cluster_id,
+                    TimestampCluster.cluster_id,
+                ],
+                SIG_EP_OUTPUT: [],
+                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
+            }
+        },
+        manufacturer="Fake_Timestamp_sensor",
+        model="Fake_Model_sensor",
+    )
+
+    zigpy_device = get_device(zigpy_device)
+
+    zha_device = await join_zigpy_device(zha_gateway, zigpy_device)
+    return zha_device, zigpy_device.endpoints[1].time_test_cluster
+
+
+async def test_timestamp_sensor_v2(
+    zha_gateway: Gateway,
+    zigpy_device_timestamp_sensor_v2,  # pylint: disable=redefined-outer-name
+) -> None:
+    """Test quirks defined sensor."""
+
+    zha_device, cluster = zigpy_device_timestamp_sensor_v2
+    assert isinstance(zha_device.device, CustomDeviceV2)
+    entity = get_entity(zha_device, platform=Platform.SENSOR, qualifier="start_time")
+
+    await send_attributes_report(zha_gateway, cluster, {0xEF65: 2674725315})
+    assert entity.state["state"] == datetime(2024, 10, 4, 11, 15, 15, tzinfo=UTC)
+
+
 class OppleCluster(CustomCluster, ManufacturerSpecificCluster):
     """Aqara manufacturer specific cluster."""
 

zha/application/discovery.py

@@ -40,6 +40,7 @@
     switch,
     update,
 )
+from zha.application.platforms.sensor.const import SensorDeviceClass
 from zha.application.registries import (
     DEVICE_CLASS,
     PLATFORM_ENTITIES,
@@ -163,6 +164,10 @@
     ): switch.ConfigurableAttributeSwitch,
 }
 
+QUIRKS_SENSOR_DEV_CLASS_TO_ENTITY_CLASS = {
+    SensorDeviceClass.TIMESTAMP: sensor.TimestampSensor
+}
+
 
 class DeviceProbe:
     """Probe to discover entities for a device."""
@@ -280,6 +285,11 @@ def discover_quirks_v2_entities(self, device: Device) -> None:
                     )
                     continue
 
+                if entity_class is sensor.Sensor:
+                    entity_class = QUIRKS_SENSOR_DEV_CLASS_TO_ENTITY_CLASS.get(
+                        entity_metadata.device_class, entity_class
+                    )
+
                 # automatically add the attribute to ZCL_INIT_ATTRS for the cluster
                 # handler if it is not already in the list
                 if (

zha/application/platforms/sensor/__init__.py

@@ -4,6 +4,7 @@
 
 from asyncio import Task
 from dataclasses import dataclass
+from datetime import UTC, date, datetime
 import enum
 import functools
 import logging
@@ -29,7 +30,11 @@
 )
 from zha.application.platforms.climate.const import HVACAction
 from zha.application.platforms.helpers import validate_device_class
-from zha.application.platforms.sensor.const import SensorDeviceClass, SensorStateClass
+from zha.application.platforms.sensor.const import (
+    UNIX_EPOCH_TO_ZCL_EPOCH,
+    SensorDeviceClass,
+    SensorStateClass,
+)
 from zha.application.registries import PLATFORM_ENTITIES
 from zha.decorators import periodic
 from zha.units import (
@@ -240,7 +245,7 @@ def state(self) -> dict:
         return response
 
     @property
-    def native_value(self) -> str | int | float | None:
+    def native_value(self) -> date | datetime | str | int | float | None:
         """Return the state of the entity."""
         assert self._attribute_name is not None
         raw_state = self._cluster_handler.cluster.get(self._attribute_name)
@@ -264,7 +269,9 @@ def handle_cluster_handler_attribute_updated(
         ):
             self.maybe_emit_state_changed_event()
 
-    def formatter(self, value: int | enum.IntEnum) -> int | float | str | None:
+    def formatter(
+        self, value: int | enum.IntEnum
+    ) -> datetime | int | float | str | None:
         """Numeric pass-through formatter."""
         if self._decimals > 0:
             return round(
@@ -273,6 +280,14 @@ def formatter(self, value: int | enum.IntEnum) -> int | float | str | None:
         return round(float(value * self._multiplier) / self._divisor)
 
 
+class TimestampSensor(Sensor):
+    """Timestamp ZHA sensor."""
+
+    def formatter(self, value: int | enum.IntEnum) -> datetime | None:
+        """Pass-through formatter."""
+        return datetime.fromtimestamp(value - UNIX_EPOCH_TO_ZCL_EPOCH, tz=UTC)
+
+
 class PollableSensor(Sensor):
     """Base ZHA sensor that polls for state."""
 
@@ -1631,6 +1646,20 @@ class SetpointChangeSource(EnumSensor):
     _enum = SetpointChangeSourceEnum
 
 
+@CONFIG_DIAGNOSTIC_MATCH(cluster_handler_names=CLUSTER_HANDLER_THERMOSTAT)
+class SetpointChangeSourceTimestamp(TimestampSensor):
+    """Sensor that displays the timestamp the setpoint change.
+
+    Optional thermostat attribute.
+    """
+
+    _unique_id_suffix = "setpoint_change_source_timestamp"
+    _attribute_name = "setpoint_change_source_timestamp"
+    _attr_translation_key: str = "setpoint_change_source_timestamp"
+    _attr_entity_category = EntityCategory.DIAGNOSTIC
+    _attr_device_class = SensorDeviceClass.TIMESTAMP
+
+
 @CONFIG_DIAGNOSTIC_MATCH(cluster_handler_names=CLUSTER_HANDLER_COVER)
 class WindowCoveringTypeSensor(EnumSensor):
     """Sensor that displays the type of a cover device."""

zha/application/platforms/sensor/const.py

@@ -390,3 +390,5 @@ class SensorDeviceClass(enum.StrEnum):
     SensorDeviceClass.ENUM,
     SensorDeviceClass.TIMESTAMP,
 }
+
+UNIX_EPOCH_TO_ZCL_EPOCH = 946684800