Add SensorDeviceClass to statistics component, improve structures

homeassistant/components/statistics/sensor.py

@@ -16,6 +16,7 @@
 from homeassistant.components.recorder.util import execute, session_scope
 from homeassistant.components.sensor import (
     PLATFORM_SCHEMA,
+    SensorDeviceClass,
     SensorEntity,
     SensorStateClass,
 )
@@ -50,10 +51,12 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+# Stats for attributes only
 STAT_AGE_COVERAGE_RATIO = "age_coverage_ratio"
 STAT_BUFFER_USAGE_RATIO = "buffer_usage_ratio"
 STAT_SOURCE_VALUE_VALID = "source_value_valid"
 
+# All sensor statistics
 STAT_AVERAGE_LINEAR = "average_linear"
 STAT_AVERAGE_STEP = "average_step"
 STAT_AVERAGE_TIMELESS = "average_timeless"
@@ -76,28 +79,57 @@
 STAT_VALUE_MIN = "value_min"
 STAT_VARIANCE = "variance"
 
-STAT_DEFAULT = "default"
-DEPRECATION_WARNING = (
+DEPRECATION_WARNING_CHARACTERISTIC = (
     "The configuration parameter 'state_characteristic' will become "
     "mandatory in a future release of the statistics integration. "
     "Please add 'state_characteristic: %s' to the configuration of "
-    'sensor "%s" to keep the current behavior. Read the documentation '
+    "sensor '%s' to keep the current behavior. Read the documentation "
     "for further details: "
     "https://www.home-assistant.io/integrations/statistics/"
 )
 
-STATS_NOT_A_NUMBER = (
-    STAT_DATETIME_OLDEST,
+# Statistics supported by a sensor source (numeric)
+STATS_NUMERIC_SUPPORT = (
+    STAT_AVERAGE_LINEAR,
+    STAT_AVERAGE_STEP,
+    STAT_AVERAGE_TIMELESS,
+    STAT_CHANGE_SAMPLE,
+    STAT_CHANGE_SECOND,
+    STAT_CHANGE,
+    STAT_COUNT,
     STAT_DATETIME_NEWEST,
+    STAT_DATETIME_OLDEST,
+    STAT_DISTANCE_95P,
+    STAT_DISTANCE_99P,
+    STAT_DISTANCE_ABSOLUTE,
+    STAT_MEAN,
+    STAT_MEDIAN,
+    STAT_NOISINESS,
     STAT_QUANTILES,
+    STAT_STANDARD_DEVIATION,
+    STAT_TOTAL,
+    STAT_VALUE_MAX,
+    STAT_VALUE_MIN,
+    STAT_VARIANCE,
 )
 
+# Statistics supported by a binary_sensor source
 STATS_BINARY_SUPPORT = (
     STAT_AVERAGE_STEP,
     STAT_AVERAGE_TIMELESS,
     STAT_COUNT,
     STAT_MEAN,
-    STAT_DEFAULT,
+)
+
+STATS_NOT_A_NUMBER = (
+    STAT_DATETIME_NEWEST,
+    STAT_DATETIME_OLDEST,
+    STAT_QUANTILES,
+)
+
+STATS_DATETIME = (
+    STAT_DATETIME_NEWEST,
+    STAT_DATETIME_OLDEST,
 )
 
 CONF_STATE_CHARACTERISTIC = "state_characteristic"
@@ -115,15 +147,27 @@
 ICON = "mdi:calculator"
 
 
-def valid_binary_characteristic_configuration(config: dict[str, Any]) -> dict[str, Any]:
+def valid_state_characteristic_configuration(config: dict[str, Any]) -> dict[str, Any]:
     """Validate that the characteristic selected is valid for the source sensor type, throw if it isn't."""
-    if split_entity_id(str(config.get(CONF_ENTITY_ID)))[0] == BINARY_SENSOR_DOMAIN:
-        if config.get(CONF_STATE_CHARACTERISTIC) not in STATS_BINARY_SUPPORT:
-            raise ValueError(
-                "The configured characteristic '"
-                + str(config.get(CONF_STATE_CHARACTERISTIC))
-                + "' is not supported for a binary source sensor."
+    is_binary = split_entity_id(config[CONF_ENTITY_ID])[0] == BINARY_SENSOR_DOMAIN
+
+    if config.get(CONF_STATE_CHARACTERISTIC) is None:
+        config[CONF_STATE_CHARACTERISTIC] = STAT_COUNT if is_binary else STAT_MEAN
+        _LOGGER.warning(
+            DEPRECATION_WARNING_CHARACTERISTIC,
+            config[CONF_STATE_CHARACTERISTIC],
+            config[CONF_NAME],
+        )
+
+    characteristic = cast(str, config[CONF_STATE_CHARACTERISTIC])
+    if (is_binary and characteristic not in STATS_BINARY_SUPPORT) or (
+        not is_binary and characteristic not in STATS_NUMERIC_SUPPORT
+    ):
+        raise vol.ValueInvalid(
+            "The configured characteristic '{}' is not supported for the configured source sensor".format(
+                characteristic
             )
+        )
     return config
 
 
@@ -132,32 +176,7 @@ def valid_binary_characteristic_configuration(config: dict[str, Any]) -> dict[st
         vol.Required(CONF_ENTITY_ID): cv.entity_id,
         vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
         vol.Optional(CONF_UNIQUE_ID): cv.string,
-        vol.Optional(CONF_STATE_CHARACTERISTIC, default=STAT_DEFAULT): vol.In(
-            [
-                STAT_AVERAGE_LINEAR,
-                STAT_AVERAGE_STEP,
-                STAT_AVERAGE_TIMELESS,
-                STAT_CHANGE_SAMPLE,
-                STAT_CHANGE_SECOND,
-                STAT_CHANGE,
-                STAT_COUNT,
-                STAT_DATETIME_NEWEST,
-                STAT_DATETIME_OLDEST,
-                STAT_DISTANCE_95P,
-                STAT_DISTANCE_99P,
-                STAT_DISTANCE_ABSOLUTE,
-                STAT_MEAN,
-                STAT_MEDIAN,
-                STAT_NOISINESS,
-                STAT_QUANTILES,
-                STAT_STANDARD_DEVIATION,
-                STAT_TOTAL,
-                STAT_VALUE_MAX,
-                STAT_VALUE_MIN,
-                STAT_VARIANCE,
-                STAT_DEFAULT,
-            ]
-        ),
+        vol.Optional(CONF_STATE_CHARACTERISTIC): cv.string,
         vol.Optional(
             CONF_SAMPLES_MAX_BUFFER_SIZE, default=DEFAULT_BUFFER_SIZE
         ): vol.All(vol.Coerce(int), vol.Range(min=1)),
@@ -173,7 +192,7 @@ def valid_binary_characteristic_configuration(config: dict[str, Any]) -> dict[st
 )
 PLATFORM_SCHEMA = vol.All(
     _PLATFORM_SCHEMA_BASE,
-    valid_binary_characteristic_configuration,
+    valid_state_characteristic_configuration,
 )
 
 
@@ -230,9 +249,6 @@ def __init__(
             split_entity_id(self._source_entity_id)[0] == BINARY_SENSOR_DOMAIN
         )
         self._state_characteristic: str = state_characteristic
-        if self._state_characteristic == STAT_DEFAULT:
-            self._state_characteristic = STAT_COUNT if self.is_binary else STAT_MEAN
-            _LOGGER.warning(DEPRECATION_WARNING, self._state_characteristic, name)
         self._samples_max_buffer_size: int = samples_max_buffer_size
         self._samples_max_age: timedelta | None = samples_max_age
         self._precision: int = precision
@@ -346,12 +362,9 @@ def _derive_unit_of_measurement(self, new_state: State) -> str | None:
             STAT_VALUE_MIN,
         ):
             unit = base_unit
-        elif self._state_characteristic in (
-            STAT_COUNT,
-            STAT_DATETIME_NEWEST,
-            STAT_DATETIME_OLDEST,
-            STAT_QUANTILES,
-        ):
+        elif self._state_characteristic in STATS_NOT_A_NUMBER:
+            unit = None
+        elif self._state_characteristic == STAT_COUNT:
             unit = None
         elif self._state_characteristic == STAT_VARIANCE:
             unit = base_unit + "²"
@@ -361,6 +374,13 @@ def _derive_unit_of_measurement(self, new_state: State) -> str | None:
             unit = base_unit + "/s"
         return unit
 
+    @property
+    def device_class(self) -> Literal[SensorDeviceClass.TIMESTAMP] | None:
+        """Return the class of this device."""
+        if self._state_characteristic in STATS_DATETIME:
+            return SensorDeviceClass.TIMESTAMP
+        return None
+
     @property
     def state_class(self) -> Literal[SensorStateClass.MEASUREMENT] | None:
         """Return the state class of this entity."""

tests/components/statistics/test_sensor.py

@@ -1,6 +1,9 @@
 """The test for the statistics sensor platform."""
+from __future__ import annotations
+
 from datetime import datetime, timedelta
 import statistics
+from typing import Any, Sequence
 from unittest.mock import patch
 
 from homeassistant import config as hass_config
@@ -542,7 +545,7 @@ async def test_state_characteristics(hass: HomeAssistant):
     def mock_now():
         return mock_data["return_time"]
 
-    characteristics = (
+    characteristics: Sequence[dict[str, Any]] = (
         {
             "source_sensor_domain": "sensor",
             "name": "average_linear",
@@ -615,16 +618,16 @@ def mock_now():
             "source_sensor_domain": "sensor",
             "name": "datetime_newest",
             "value_0": STATE_UNKNOWN,
-            "value_1": start_datetime + timedelta(minutes=9),
-            "value_9": start_datetime + timedelta(minutes=9),
+            "value_1": (start_datetime + timedelta(minutes=9)).isoformat(),
+            "value_9": (start_datetime + timedelta(minutes=9)).isoformat(),
             "unit": None,
         },
         {
             "source_sensor_domain": "sensor",
             "name": "datetime_oldest",
             "value_0": STATE_UNKNOWN,
-            "value_1": start_datetime + timedelta(minutes=9),
-            "value_9": start_datetime + timedelta(minutes=1),
+            "value_1": (start_datetime + timedelta(minutes=9)).isoformat(),
+            "value_9": (start_datetime + timedelta(minutes=1)).isoformat(),
             "unit": None,
         },
         {
@@ -805,7 +808,11 @@ def mock_now():
             state = hass.states.get(
                 f"sensor.test_{characteristic['source_sensor_domain']}_{characteristic['name']}"
             )
-            assert state is not None
+            assert state is not None, (
+                f"no state object for characteristic "
+                f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
+                f"(buffer filled)"
+            )
             assert state.state == str(characteristic["value_9"]), (
                 f"value mismatch for characteristic "
                 f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
@@ -826,7 +833,11 @@ def mock_now():
             state = hass.states.get(
                 f"sensor.test_{characteristic['source_sensor_domain']}_{characteristic['name']}"
             )
-            assert state is not None
+            assert state is not None, (
+                f"no state object for characteristic "
+                f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
+                f"(one stored value)"
+            )
             assert state.state == str(characteristic["value_1"]), (
                 f"value mismatch for characteristic "
                 f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
@@ -844,7 +855,11 @@ def mock_now():
             state = hass.states.get(
                 f"sensor.test_{characteristic['source_sensor_domain']}_{characteristic['name']}"
             )
-            assert state is not None
+            assert state is not None, (
+                f"no state object for characteristic "
+                f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
+                f"(buffer empty)"
+            )
             assert state.state == str(characteristic["value_0"]), (
                 f"value mismatch for characteristic "
                 f"'{characteristic['source_sensor_domain']}/{characteristic['name']}' "
@@ -987,7 +1002,7 @@ def mock_purge(self, *args):
     # The max_age timestamp should be 1 hour before what we have right
     # now in mock_data['return_time'].
     assert mock_data["return_time"] == datetime.strptime(
-        state.state, "%Y-%m-%d %H:%M:%S%z"
+        state.state, "%Y-%m-%dT%H:%M:%S%z"
     ) + timedelta(hours=1)