Add statistics keep_last_sample option (#88655)

* introduce preserve last value option

* improve comments

* add unit test

* skip scheduling purge on a preserved value

* do not schedule sensor update if preserving last value

* fix unit test to use new mock time pattern

pattern introduced in https://github.com/home-assistant/core/pull/93499

* rename preserve_last_val to keep_last_sample

* add keep_last_sample config validation

homeassistant/components/statistics/sensor.py

@@ -186,6 +186,7 @@ STATS_BINARY_PERCENTAGE = {
 CONF_STATE_CHARACTERISTIC = "state_characteristic"
 CONF_SAMPLES_MAX_BUFFER_SIZE = "sampling_size"
 CONF_MAX_AGE = "max_age"
+CONF_KEEP_LAST_SAMPLE = "keep_last_sample"
 CONF_PRECISION = "precision"
 CONF_PERCENTILE = "percentile"
 
@@ -221,6 +222,16 @@ def valid_boundary_configuration(config: dict[str, Any]) -> dict[str, Any]:
     return config
 
 
+def valid_keep_last_sample(config: dict[str, Any]) -> dict[str, Any]:
+    """Validate that if keep_last_sample is set, max_age must also be set."""
+
+    if config.get(CONF_KEEP_LAST_SAMPLE) is True and config.get(CONF_MAX_AGE) is None:
+        raise vol.RequiredFieldInvalid(
+            "The sensor configuration must provide 'max_age' if 'keep_last_sample' is True"
+        )
+    return config
+
+
 _PLATFORM_SCHEMA_BASE = PLATFORM_SCHEMA.extend(
     {
         vol.Required(CONF_ENTITY_ID): cv.entity_id,
@@ -231,6 +242,7 @@ _PLATFORM_SCHEMA_BASE = PLATFORM_SCHEMA.extend(
             vol.Coerce(int), vol.Range(min=1)
         ),
         vol.Optional(CONF_MAX_AGE): cv.time_period,
+        vol.Optional(CONF_KEEP_LAST_SAMPLE, default=False): cv.boolean,
         vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): vol.Coerce(int),
         vol.Optional(CONF_PERCENTILE, default=50): vol.All(
             vol.Coerce(int), vol.Range(min=1, max=99)
@@ -241,6 +253,7 @@ PLATFORM_SCHEMA = vol.All(
     _PLATFORM_SCHEMA_BASE,
     valid_state_characteristic_configuration,
     valid_boundary_configuration,
+    valid_keep_last_sample,
 )
 
 
@@ -263,6 +276,7 @@ async def async_setup_platform(
                 state_characteristic=config[CONF_STATE_CHARACTERISTIC],
                 samples_max_buffer_size=config.get(CONF_SAMPLES_MAX_BUFFER_SIZE),
                 samples_max_age=config.get(CONF_MAX_AGE),
+                samples_keep_last=config[CONF_KEEP_LAST_SAMPLE],
                 precision=config[CONF_PRECISION],
                 percentile=config[CONF_PERCENTILE],
             )
@@ -282,6 +296,7 @@ class StatisticsSensor(SensorEntity):
         state_characteristic: str,
         samples_max_buffer_size: int | None,
         samples_max_age: timedelta | None,
+        samples_keep_last: bool,
         precision: int,
         percentile: int,
     ) -> None:
@@ -297,6 +312,7 @@ class StatisticsSensor(SensorEntity):
         self._state_characteristic: str = state_characteristic
         self._samples_max_buffer_size: int | None = samples_max_buffer_size
         self._samples_max_age: timedelta | None = samples_max_age
+        self.samples_keep_last: bool = samples_keep_last
         self._precision: int = precision
         self._percentile: int = percentile
         self._value: StateType | datetime = None
@@ -456,13 +472,27 @@ class StatisticsSensor(SensorEntity):
         now = dt_util.utcnow()
 
         _LOGGER.debug(
-            "%s: purging records older then %s(%s)",
+            "%s: purging records older then %s(%s)(keep_last_sample: %s)",
             self.entity_id,
             dt_util.as_local(now - max_age),
             self._samples_max_age,
+            self.samples_keep_last,
         )
 
         while self.ages and (now - self.ages[0]) > max_age:
+            if self.samples_keep_last and len(self.ages) == 1:
+                # Under normal circumstance this will not be executed, as a purge will not
+                # be scheduled for the last value if samples_keep_last is enabled.
+                # If this happens to be called outside normal scheduling logic or a
+                # source sensor update, this ensures the last value is preserved.
+                _LOGGER.debug(
+                    "%s: preserving expired record with datetime %s(%s)",
+                    self.entity_id,
+                    dt_util.as_local(self.ages[0]),
+                    (now - self.ages[0]),
+                )
+                break
+
             _LOGGER.debug(
                 "%s: purging record with datetime %s(%s)",
                 self.entity_id,
@@ -475,6 +505,17 @@ class StatisticsSensor(SensorEntity):
     def _next_to_purge_timestamp(self) -> datetime | None:
         """Find the timestamp when the next purge would occur."""
         if self.ages and self._samples_max_age:
+            if self.samples_keep_last and len(self.ages) == 1:
+                # Preserve the most recent entry if it is the only value.
+                # Do not schedule another purge. When a new source
+                # value is inserted it will restart purge cycle.
+                _LOGGER.debug(
+                    "%s: skipping purge cycle for last record with datetime %s(%s)",
+                    self.entity_id,
+                    dt_util.as_local(self.ages[0]),
+                    (dt_util.utcnow() - self.ages[0]),
+                )
+                return None
             # Take the oldest entry from the ages list and add the configured max_age.
             # If executed after purging old states, the result is the next timestamp
             # in the future when the oldest state will expire.
@@ -491,6 +532,8 @@ class StatisticsSensor(SensorEntity):
         self._update_value()
 
         # If max_age is set, ensure to update again after the defined interval.
+        # By basing updates off the timestamps of sampled data we avoid updating
+        # when none of the observed entities change.
         if timestamp := self._next_to_purge_timestamp():
             _LOGGER.debug("%s: scheduling update at %s", self.entity_id, timestamp)
             if self._update_listener:

tests/components/statistics/test_sensor.py

@@ -313,6 +313,67 @@ async def test_sampling_boundaries_given(hass: HomeAssistant) -> None:
     assert state is not None
 
 
+async def test_keep_last_value_given(hass: HomeAssistant) -> None:
+    """Test if either sampling_size or max_age are given."""
+    assert await async_setup_component(
+        hass,
+        "sensor",
+        {
+            "sensor": [
+                {
+                    "platform": "statistics",
+                    "name": "test_none",
+                    "entity_id": "sensor.test_monitored",
+                    "state_characteristic": "mean",
+                    "keep_last_sample": True,
+                },
+                {
+                    "platform": "statistics",
+                    "name": "test_sampling_size",
+                    "entity_id": "sensor.test_monitored",
+                    "state_characteristic": "mean",
+                    "sampling_size": 20,
+                    "keep_last_sample": True,
+                },
+                {
+                    "platform": "statistics",
+                    "name": "test_max_age",
+                    "entity_id": "sensor.test_monitored",
+                    "state_characteristic": "mean",
+                    "max_age": {"minutes": 4},
+                    "keep_last_sample": True,
+                },
+                {
+                    "platform": "statistics",
+                    "name": "test_both",
+                    "entity_id": "sensor.test_monitored",
+                    "state_characteristic": "mean",
+                    "sampling_size": 20,
+                    "max_age": {"minutes": 4},
+                    "keep_last_sample": True,
+                },
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+
+    hass.states.async_set(
+        "sensor.test_monitored",
+        str(VALUES_NUMERIC[0]),
+        {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.test_none")
+    assert state is None
+    state = hass.states.get("sensor.test_sampling_size")
+    assert state is None
+    state = hass.states.get("sensor.test_max_age")
+    assert state is not None
+    state = hass.states.get("sensor.test_both")
+    assert state is not None
+
+
 async def test_sampling_size_reduced(hass: HomeAssistant) -> None:
     """Test limited buffer size."""
     assert await async_setup_component(
@@ -467,6 +528,124 @@ async def test_age_limit_expiry(hass: HomeAssistant) -> None:
         assert state.attributes.get("age_coverage_ratio") is None
 
 
+async def test_age_limit_expiry_with_keep_last_sample(hass: HomeAssistant) -> None:
+    """Test that values are removed with given max age."""
+    now = dt_util.utcnow()
+    current_time = datetime(now.year + 1, 8, 2, 12, 23, tzinfo=dt_util.UTC)
+
+    with freeze_time(current_time) as freezer:
+        assert await async_setup_component(
+            hass,
+            "sensor",
+            {
+                "sensor": [
+                    {
+                        "platform": "statistics",
+                        "name": "test",
+                        "entity_id": "sensor.test_monitored",
+                        "state_characteristic": "mean",
+                        "sampling_size": 20,
+                        "max_age": {"minutes": 4},
+                        "keep_last_sample": True,
+                    },
+                ]
+            },
+        )
+        await hass.async_block_till_done()
+
+        for value in VALUES_NUMERIC:
+            current_time += timedelta(minutes=1)
+            freezer.move_to(current_time)
+            async_fire_time_changed(hass, current_time)
+            hass.states.async_set(
+                "sensor.test_monitored",
+                str(value),
+                {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
+            )
+        await hass.async_block_till_done()
+
+        # After adding all values, we should only see 5 values in memory
+
+        state = hass.states.get("sensor.test")
+        new_mean = round(sum(VALUES_NUMERIC[-5:]) / len(VALUES_NUMERIC[-5:]), 2)
+        assert state is not None
+        assert state.state == str(new_mean)
+        assert state.attributes.get("buffer_usage_ratio") == round(5 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 1.0
+
+        # Values expire over time. Only two are left
+
+        current_time += timedelta(minutes=3)
+        freezer.move_to(current_time)
+        async_fire_time_changed(hass, current_time)
+        await hass.async_block_till_done()
+
+        state = hass.states.get("sensor.test")
+        new_mean = round(sum(VALUES_NUMERIC[-2:]) / len(VALUES_NUMERIC[-2:]), 2)
+        assert state is not None
+        assert state.state == str(new_mean)
+        assert state.attributes.get("buffer_usage_ratio") == round(2 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 1 / 4
+
+        # Values expire over time. Only one is left
+
+        current_time += timedelta(minutes=1)
+        freezer.move_to(current_time)
+        async_fire_time_changed(hass, current_time)
+        await hass.async_block_till_done()
+
+        state = hass.states.get("sensor.test")
+        new_mean = float(VALUES_NUMERIC[-1])
+        assert state is not None
+        assert state.state == str(new_mean)
+        assert state.attributes.get("buffer_usage_ratio") == round(1 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 0
+
+        # Values expire over time. All values expired, but preserve expired last value
+
+        current_time += timedelta(minutes=1)
+        freezer.move_to(current_time)
+        async_fire_time_changed(hass, current_time)
+        await hass.async_block_till_done()
+
+        state = hass.states.get("sensor.test")
+        assert state is not None
+        assert state.state == str(float(VALUES_NUMERIC[-1]))
+        assert state.attributes.get("buffer_usage_ratio") == round(1 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 0
+
+        # Indefinitely preserve expired last value
+
+        current_time += timedelta(minutes=1)
+        freezer.move_to(current_time)
+        async_fire_time_changed(hass, current_time)
+        await hass.async_block_till_done()
+
+        state = hass.states.get("sensor.test")
+        assert state is not None
+        assert state.state == str(float(VALUES_NUMERIC[-1]))
+        assert state.attributes.get("buffer_usage_ratio") == round(1 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 0
+
+        # New sensor value within max_age, preserved expired value should be dropped
+        last_update_val = 123.0
+        current_time += timedelta(minutes=1)
+        freezer.move_to(current_time)
+        async_fire_time_changed(hass, current_time)
+        hass.states.async_set(
+            "sensor.test_monitored",
+            str(last_update_val),
+            {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS},
+        )
+        await hass.async_block_till_done()
+
+        state = hass.states.get("sensor.test")
+        assert state is not None
+        assert state.state == str(last_update_val)
+        assert state.attributes.get("buffer_usage_ratio") == round(1 / 20, 2)
+        assert state.attributes.get("age_coverage_ratio") == 0
+
+
 async def test_precision(hass: HomeAssistant) -> None:
     """Test correct results with precision set."""
     assert await async_setup_component(