Improve pressure sensor

Author: Alexwijn

custom_components/sat/binary_sensor.py

@@ -23,6 +23,9 @@
 from .types import BoilerStatus, CycleClassification
 
 PRESSURE_DROP_RATE_SETTLE_SECONDS = 600
+PRESSURE_EMA_ALPHA = 0.05
+PRESSURE_PROBLEM_CONFIRMATION_SECONDS = 120
+PRESSURE_DROP_RATE_MIN_WINDOW_SECONDS = 300
 
 
 async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
@@ -205,6 +208,8 @@ def __init__(self, coordinator, config: SatConfig, heating_control: SatHeatingCo
         self._last_drop_rate: Optional[float] = None
         self._last_seen_pressure: Optional[float] = None
         self._last_pressure_timestamp: Optional[float] = None
+        self._smoothed_pressure: Optional[float] = None
+        self._problem_first_detected: Optional[float] = None
         self._pressure_samples: deque[tuple[float, float]] = deque()
 
     async def async_added_to_hass(self) -> None:
@@ -218,6 +223,7 @@ async def async_added_to_hass(self) -> None:
         self._last_pressure = float_value(attributes.get("last_pressure"))
         self._last_pressure_timestamp = float_value(attributes.get("last_pressure_timestamp"))
         self._last_seen_pressure = float_value(attributes.get("last_seen_pressure_timestamp"))
+        self._smoothed_pressure = float_value(attributes.get("smoothed_pressure"))
 
         if self._last_pressure is not None and self._last_pressure_timestamp is not None:
             self._pressure_samples.append((self._last_pressure_timestamp, self._last_pressure))
@@ -247,15 +253,15 @@ def available(self) -> bool:
     def is_on(self) -> bool:
         """Return the state of the sensor."""
         now = timestamp()
-        pressure = self._coordinator.boiler_pressure
+        raw_pressure = self._coordinator.boiler_pressure
         minimum_pressure = self._pressure_config.minimum_pressure_bar
         maximum_pressure = self._pressure_config.maximum_pressure_bar
         maximum_age_seconds = self._pressure_config.maximum_age_seconds
         maximum_drop_rate = self._pressure_config.maximum_drop_rate_bar_per_hour
 
         self._track_active_state(now)
 
-        if pressure is None:
+        if raw_pressure is None:
             if self._last_seen_pressure is None:
                 return False
 
@@ -265,15 +271,16 @@ def is_on(self) -> bool:
             return (now - self._last_seen_pressure) > maximum_age_seconds
 
         self._last_seen_pressure = now
-        self._record_pressure_sample(now, pressure, maximum_age_seconds)
+        smoothed = self._update_smoothed_pressure(raw_pressure)
+        self._record_pressure_sample(now, raw_pressure, maximum_age_seconds)
 
         drop_rate = self._calculate_drop_rate()
 
-        self._last_pressure = pressure
+        self._last_pressure = raw_pressure
         self._last_pressure_timestamp = now
 
-        pressure_low = pressure < minimum_pressure
-        pressure_high = pressure > maximum_pressure
+        pressure_low = smoothed < minimum_pressure
+        pressure_high = smoothed > maximum_pressure
         drop_rate_allowed = self._drop_rate_allowed(now)
 
         if not drop_rate_allowed:
@@ -283,14 +290,16 @@ def is_on(self) -> bool:
             self._last_drop_rate = round(drop_rate, 3)
 
         drop_rate_high = drop_rate is not None and drop_rate > maximum_drop_rate
+        raw_problem = pressure_low or pressure_high or drop_rate_high
 
-        return pressure_low or pressure_high or drop_rate_high
+        return self._confirm_problem(now, raw_problem)
 
     @property
     def extra_state_attributes(self) -> dict[str, Optional[float]]:
         """Return extra attributes for debugging pressure health decisions."""
         return {
             "pressure": self._coordinator.boiler_pressure,
+            "smoothed_pressure": self._smoothed_pressure,
             "pressure_drop_rate_bar_per_hour": self._last_drop_rate,
 
             "last_pressure": self._last_pressure,
@@ -303,13 +312,33 @@ def unique_id(self) -> str:
         """Return a unique ID to use for this entity."""
         return f"{self._config.entry_id}-pressure-health"
 
+    def _update_smoothed_pressure(self, raw_pressure: float) -> float:
+        if self._smoothed_pressure is None:
+            self._smoothed_pressure = raw_pressure
+        else:
+            self._smoothed_pressure = (
+                PRESSURE_EMA_ALPHA * raw_pressure
+                + (1 - PRESSURE_EMA_ALPHA) * self._smoothed_pressure
+            )
+        return self._smoothed_pressure
+
+    def _confirm_problem(self, now: float, condition: bool) -> bool:
+        if not condition:
+            self._problem_first_detected = None
+            return False
+
+        if self._problem_first_detected is None:
+            self._problem_first_detected = now
+
+        return (now - self._problem_first_detected) >= PRESSURE_PROBLEM_CONFIRMATION_SECONDS
+
     def _track_active_state(self, timestamp_seconds: float) -> None:
         active = self._coordinator.active
         if self._last_active is None:
             self._last_active = active
             return
 
-        if self._last_active and not active:
+        if self._last_active != active:
             self._drop_rate_suspended_until = timestamp_seconds + PRESSURE_DROP_RATE_SETTLE_SECONDS
             self._pressure_samples.clear()
 
@@ -336,17 +365,27 @@ def _record_pressure_sample(self, timestamp_seconds: float, pressure: float, max
             self._pressure_samples.popleft()
 
     def _calculate_drop_rate(self) -> Optional[float]:
-        if len(self._pressure_samples) < 2:
+        if len(self._pressure_samples) < 3:
+            return None
+
+        elapsed = self._pressure_samples[-1][0] - self._pressure_samples[0][0]
+        if elapsed < PRESSURE_DROP_RATE_MIN_WINDOW_SECONDS:
             return None
 
-        oldest_time, oldest_pressure = self._pressure_samples[0]
-        newest_time, newest_pressure = self._pressure_samples[-1]
-        elapsed = newest_time - oldest_time
+        n = len(self._pressure_samples)
+        sum_t = sum_p = sum_tp = sum_t2 = 0.0
+        for t, p in self._pressure_samples:
+            sum_t += t
+            sum_p += p
+            sum_tp += t * p
+            sum_t2 += t * t
 
-        if elapsed <= 0:
+        denom = n * sum_t2 - sum_t * sum_t
+        if denom == 0:
             return None
 
-        return ((oldest_pressure - newest_pressure) / elapsed) * 3600
+        slope = (n * sum_tp - sum_t * sum_p) / denom
+        return -slope * 3600
 
 
 class SatDeviceHealthSensor(SatEntity, BinarySensorEntity):

tests/test_binary_sensor.py

@@ -21,15 +21,33 @@
 )
 
 
-async def test_pressure_health_low_pressure(hass, coordinator, entry, domains, data, options, config):
+def _get_pressure_entity_id(hass, entry):
+    registry = er.async_get(hass)
+    return registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
+
+
+async def test_pressure_health_low_pressure(hass, coordinator, entry, domains, data, options, config, monkeypatch):
+    current_time = 0.0
+
+    def fake_timestamp():
+        return current_time
+
+    monkeypatch.setattr(sat_binary_sensor, "timestamp", fake_timestamp)
+
     await coordinator.async_set_boiler_pressure(0.6)
     coordinator.async_update_listeners()
     await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
+    assert state.state == "off"
+
+    current_time = 130.0
+    await coordinator.async_set_boiler_pressure(0.6)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state.state == "on"
 
 
@@ -38,9 +56,7 @@ async def test_pressure_health_normal_pressure(hass, coordinator, entry, domains
     coordinator.async_update_listeners()
     await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
     assert state.state == "off"
 
@@ -53,23 +69,14 @@ def fake_timestamp():
 
     monkeypatch.setattr(sat_binary_sensor, "timestamp", fake_timestamp)
 
-    await coordinator.async_set_boiler_pressure(1.8)
-    coordinator.async_update_listeners()
-    await hass.async_block_till_done()
-
-    current_time = 1800.0
-    await coordinator.async_set_boiler_pressure(1.6)
-    coordinator.async_update_listeners()
-    await hass.async_block_till_done()
-
-    current_time = 3600.0
-    await coordinator.async_set_boiler_pressure(1.2)
-    coordinator.async_update_listeners()
-    await hass.async_block_till_done()
+    for i in range(60):
+        current_time = float(i * 10)
+        pressure = 1.8 - (i * 0.01)
+        await coordinator.async_set_boiler_pressure(pressure)
+        coordinator.async_update_listeners()
+        await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
     assert state.state == "on"
     assert state.attributes["pressure_drop_rate_bar_per_hour"] is not None
@@ -94,9 +101,7 @@ def fake_timestamp():
     coordinator.async_update_listeners()
     await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
     assert state.state == "off"
     assert state.attributes["pressure_drop_rate_bar_per_hour"] is None
@@ -119,9 +124,7 @@ def fake_timestamp():
     coordinator.async_update_listeners()
     await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
     assert state.state == "on"
 
@@ -141,6 +144,7 @@ def fake_timestamp():
         "last_pressure": 1.4,
         "last_pressure_timestamp": 100.0,
         "last_seen_pressure_timestamp": 100.0,
+        "smoothed_pressure": 1.4,
     }
 
     async def fake_async_get_last_state(self):
@@ -155,8 +159,95 @@ async def fake_async_get_last_state(self):
     coordinator.async_update_listeners()
     await hass.async_block_till_done()
 
-    registry = er.async_get(hass)
-    entity_id = registry.async_get_entity_id("binary_sensor", "sat", f"{entry.entry_id}-pressure-health")
-    state = hass.states.get(entity_id)
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
     assert state is not None
     assert state.attributes["last_pressure"] == 1.4
+    assert state.attributes["smoothed_pressure"] is not None
+
+
+async def test_pressure_health_normal_oscillations_no_false_positive(hass, coordinator, entry, domains, data, options, config, monkeypatch):
+    current_time = 0.0
+
+    def fake_timestamp():
+        return current_time
+
+    monkeypatch.setattr(sat_binary_sensor, "timestamp", fake_timestamp)
+
+    for i in range(60):
+        current_time = float(i * 30)
+        pressure = 2.1 + (0.2 if i % 2 == 0 else -0.2)
+        await coordinator.async_set_boiler_pressure(pressure)
+        coordinator.async_update_listeners()
+        await hass.async_block_till_done()
+
+        state = hass.states.get(_get_pressure_entity_id(hass, entry))
+        assert state.state == "off", f"False positive at iteration {i}, pressure={pressure}"
+
+
+async def test_pressure_health_confirmation_delay_resets(hass, coordinator, entry, domains, data, options, config, monkeypatch):
+    current_time = 0.0
+
+    def fake_timestamp():
+        return current_time
+
+    monkeypatch.setattr(sat_binary_sensor, "timestamp", fake_timestamp)
+
+    await coordinator.async_set_boiler_pressure(0.6)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "off"
+
+    current_time = 100.0
+    await coordinator.async_set_boiler_pressure(0.6)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "off"
+
+    current_time = 110.0
+    await coordinator.async_set_boiler_pressure(1.5)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "off"
+
+    current_time = 120.0
+    await coordinator.async_set_boiler_pressure(0.6)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    current_time = 250.0
+    await coordinator.async_set_boiler_pressure(0.6)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "on"
+
+
+async def test_pressure_health_high_pressure_with_delay(hass, coordinator, entry, domains, data, options, config, monkeypatch):
+    current_time = 0.0
+
+    def fake_timestamp():
+        return current_time
+
+    monkeypatch.setattr(sat_binary_sensor, "timestamp", fake_timestamp)
+
+    await coordinator.async_set_boiler_pressure(3.0)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "off"
+
+    current_time = 130.0
+    await coordinator.async_set_boiler_pressure(3.0)
+    coordinator.async_update_listeners()
+    await hass.async_block_till_done()
+
+    state = hass.states.get(_get_pressure_entity_id(hass, entry))
+    assert state.state == "on"