diff --git a/config.ini b/config.ini index 838957c..4f8eb95 100644 --- a/config.ini +++ b/config.ini @@ -58,6 +58,8 @@ security_duration = 7200 ph_pwm_period = 120 # PWM period in seconds for Cl cl_pwm_period = 120 +# Number of filtration cycles (~2 days) used to detect an empty pH- or Cl canister +canister_cycles = 3 [heating] # Delay in seconds that the flow pump will run after the heat pump stopped when going from heating diff --git a/controller/disinfection.py b/controller/disinfection.py index be7e59c..1d1225f 100644 --- a/controller/disinfection.py +++ b/controller/disinfection.py @@ -17,8 +17,9 @@ import logging import time +from collections import deque from datetime import datetime, timedelta -from typing import Final +from typing import Final, NamedTuple from .actor import PoupoolActor, PoupoolModel from .config import config @@ -102,12 +103,170 @@ def compute(self): return constrain((self.pterm * self.__scale) * error, 0, 1) +class CycleStats(NamedTuple): + peak: float + mean: float + + +class CanisterMonitor: + MIN_CYCLE_SAMPLES: Final = 15 + MAX_CYCLE_SAMPLES: Final = 120 + + def __init__( + self, + name: str, + cycles: int = 3, + min_samples: int = MIN_CYCLE_SAMPLES, + max_samples: int = MAX_CYCLE_SAMPLES, + ) -> None: + self.__name: str = name + self.__cycles_count: int = cycles + self.__min_samples: int = min_samples + self.__max_samples: int = max_samples + self.__history: deque[CycleStats] = deque(maxlen=cycles) + self.__current_samples: list[float] = [] + self.__is_empty: bool = False + self.__alert_stats: CycleStats | None = None + + @property + def is_empty(self) -> bool: + return self.__is_empty + + @property + def history(self) -> list[CycleStats]: + return list(self.__history) + + @property + def current_samples(self) -> list[float]: + return list(self.__current_samples) + + @property + def alert_stats(self) -> CycleStats | None: + return self.__alert_stats + + def restore(self, is_empty: bool) -> None: + self.__is_empty = is_empty + if not is_empty: + self.__alert_stats = None + self.__history.clear() + self.__current_samples.clear() + + def add_sample(self, value: float) -> bool: + """Record a sample. Returns True if is_empty state changed.""" + self.__current_samples.append(value) + if self.__is_empty and len(self.__current_samples) >= self.__min_samples: + current = self.__compute_current_stats() + if self.__is_low_demand(current): + logger.info(f"{self.__name} canister refilled: low duty cycle restored ({current.peak * 100:.1f}%)") + self.__clear_empty() + return True + if len(self.__current_samples) >= self.__max_samples: + return self.finish_cycle() + return False + + def finish_cycle(self) -> bool: + """Finalize the current cycle. Returns True if is_empty state changed.""" + if len(self.__current_samples) < self.__min_samples: + self.__current_samples.clear() + return False + stats = self.__compute_current_stats() + self.__current_samples.clear() + if self.__is_empty: + if self.__should_clear(stats): + logger.info( + f"{self.__name} canister refilled: duty cycle dropped " + f"(peak: {stats.peak * 100:.1f}%, mean: {stats.mean * 100:.1f}%)" + ) + self.__clear_empty() + return True + if self.__alert_stats is None or stats.peak > self.__alert_stats.peak: + self.__alert_stats = stats + return False + self.__history.append(stats) + if self.__should_trigger(): + self.__is_empty = True + self.__alert_stats = stats + logger.warning( + f"{self.__name} canister empty detected! Peak: {stats.peak * 100:.1f}%, mean: {stats.mean * 100:.1f}%" + ) + return True + return False + + def __compute_current_stats(self) -> CycleStats: + return CycleStats( + peak=max(self.__current_samples), + mean=sum(self.__current_samples) / len(self.__current_samples), + ) + + def __should_trigger(self) -> bool: + if len(self.__history) < self.__cycles_count: + return False + peaks = [c.peak for c in self.__history] + means = [c.mean for c in self.__history] + + # All cycles in the window must show persistent demand (never recovering to low duty cycle) + if any(p < 0.08 or m < 0.06 for p, m in zip(peaks, means, strict=True)): + return False + + # 1. High-duty saturation fallback (all cycles sustained high) + if all(p >= 0.35 and m >= 0.25 for p, m in zip(peaks, means, strict=True)): + return True + + # 2. Consistent multi-cycle upward trend in peak or mean + if peaks[-1] < 0.20 and means[-1] < 0.15: + return False + # Trend deltas scale adaptively based on the cycle window size: + # - Shorter windows (<= 3 cycles) detect rapid empty canister climb earlier (~12-18h) + # with smaller total delta (min_delta_p = 0.07, min_delta_m = 0.05). + # - Longer windows (4-6 cycles) require a larger cumulative delta (0.10 - 0.12) + # to avoid drift over multiple days. + min_delta_p = 0.07 if self.__cycles_count <= 3 else (0.10 if self.__cycles_count <= 4 else 0.12) + min_delta_m = 0.05 if self.__cycles_count <= 3 else (0.08 if self.__cycles_count <= 4 else 0.10) + return self.__has_upward_trend(peaks, min_delta=min_delta_p) or self.__has_upward_trend( + means, min_delta=min_delta_m + ) + + def __has_upward_trend(self, values: list[float], min_delta: float, tolerance: float = 0.01) -> bool: + """Check if values follow a consistent upward trend. + + For short windows (n <= 4), every step must be strictly increasing (n - 1) + to prevent false positives from brief fluctuations. For longer windows (n > 4), + at most one step is allowed to plateau (n - 2). + """ + n = len(values) + if values[-1] - values[0] < min_delta: + return False + min_incr = n - 1 if n <= 4 else n - 2 + increasing = sum(1 for i in range(1, n) if values[i] > values[i - 1]) + no_drops = all(values[i] >= values[i - 1] - tolerance for i in range(1, n)) + return increasing >= min_incr and no_drops + + @staticmethod + def __is_low_demand(stats: CycleStats) -> bool: + return stats.peak < 0.10 and stats.mean < 0.08 + + def __should_clear(self, stats: CycleStats) -> bool: + if self.__is_low_demand(stats): + return True + return ( + self.__alert_stats is not None + and stats.peak <= self.__alert_stats.peak - 0.15 + and stats.mean <= self.__alert_stats.mean * 0.5 + ) + + def __clear_empty(self) -> None: + self.__is_empty = False + self.__alert_stats = None + self.__history.clear() + + class Disinfection(PoupoolActor): STATE_REFRESH_DELAY = 10 START_DELAY = int(config["disinfection", "start_delay"]) WAITING_DELAY = int(config["disinfection", "waiting_delay"]) PH_PWM_PERIOD = int(config["disinfection", "ph_pwm_period"]) CL_PWM_PERIOD = int(config["disinfection", "cl_pwm_period"]) + CANISTER_CYCLES = int(config["disinfection", "canister_cycles"]) states: Final = [ "halt", @@ -135,6 +294,9 @@ def __init__(self, encoder, devices, sensors_reader, sensors_writer, disable=Fal # Chlorine self.__cl = PWM.start("cl", self.__devices.get_pump("cl")).proxy() self.__cl.period = Disinfection.CL_PWM_PERIOD + # Canister empty monitors + self.__ph_canister = CanisterMonitor("pH", cycles=Disinfection.CANISTER_CYCLES) + self.__cl_canister = CanisterMonitor("Cl", cycles=Disinfection.CANISTER_CYCLES) # Initialize the state machine self.__machine = PoupoolModel(model=self, states=Disinfection.states, initial="halt") @@ -169,6 +331,16 @@ def orp_pterm(self, value): self.__orp_controller.pterm = value logger.info(f"ORP pterm set to: {self.__orp_controller.pterm:.2f}") + def ph_canister_empty(self, value: bool) -> None: + if self.__ph_canister.is_empty != value: + self.__ph_canister.restore(value) + logger.info(f"pH canister empty flag set to {value}") + + def cl_canister_empty(self, value: bool) -> None: + if self.__cl_canister.is_empty != value: + self.__cl_canister.restore(value) + logger.info(f"Cl canister empty flag set to {value}") + def is_disabled(self): return self.__is_disabled @@ -182,6 +354,10 @@ def on_enter_halt(self): self.__encoder.disinfection_cl_feedback(0) self.__encoder.disinfection_ph_feedback(0) self.__sensors_writer.do_cancel.defer() + if self.__ph_enable and self.__ph_canister.finish_cycle(): + self.__encoder.disinfection_ph_canister__empty(1 if self.__ph_canister.is_empty else 0, retain=True) + if self.__orp_enable and self.__cl_canister.finish_cycle(): + self.__encoder.disinfection_cl_canister__empty(1 if self.__cl_canister.is_empty else 0, retain=True) def on_enter_waiting(self): logger.info("Entering waiting state") @@ -205,6 +381,8 @@ def on_enter_running_adjusting(self): self.__encoder.disinfection_ph_feedback(round(ph_feedback * 100)) logger.debug(f"pH: {ph:.2f} feedback: {ph_feedback:.2f}") self.__ph.value = ph_feedback + if self.__ph_enable and self.__ph_canister.add_sample(ph_feedback): + self.__encoder.disinfection_ph_canister__empty(1 if self.__ph_canister.is_empty else 0, retain=True) # ORP/Chlorine orp = self.__sensors_reader.get_orp().get() self.__orp_controller.current = orp @@ -212,6 +390,8 @@ def on_enter_running_adjusting(self): self.__encoder.disinfection_cl_feedback(round(cl_feedback * 100)) logger.debug(f"ORP: {orp} feedback: {cl_feedback:.2f}") self.__cl.value = cl_feedback + if self.__orp_enable and self.__cl_canister.add_sample(cl_feedback): + self.__encoder.disinfection_cl_canister__empty(1 if self.__cl_canister.is_empty else 0, retain=True) self._proxy.treat.defer() def on_enter_running_treating(self): diff --git a/controller/dispatcher.py b/controller/dispatcher.py index af93b5e..513d7b0 100644 --- a/controller/dispatcher.py +++ b/controller/dispatcher.py @@ -295,6 +295,20 @@ def register(self, filtration, tank, swim, light, heater, heating, disinfection, to_float, False, ), + "/status/disinfection/ph/canister_empty": ( + disinfection, + lambda x: x.lower() in ("0", "1", "off", "on"), + lambda _: "ph_canister_empty", + to_bool, + False, + ), + "/status/disinfection/cl/canister_empty": ( + disinfection, + lambda x: x.lower() in ("0", "1", "off", "on"), + lambda _: "cl_canister_empty", + to_bool, + False, + ), "/status/water/counter": ( arduino, greater_equal(0), diff --git a/default_settings.py b/default_settings.py index 3b2b584..a3f6f2f 100644 --- a/default_settings.py +++ b/default_settings.py @@ -75,6 +75,8 @@ def publish(topic, value): publish("/settings/disinfection/orp/enable", "1") publish("/settings/disinfection/orp/setpoint", "600") publish("/settings/disinfection/orp/pterm", "1.0") + publish("/status/disinfection/ph/canister_empty", "0") + publish("/status/disinfection/cl/canister_empty", "0") print("\n**** Missing default parameters ***") [print(t) for t in remaining] diff --git a/home-assistant/mqtt.yaml b/home-assistant/mqtt.yaml index f4359db..5bde3f8 100644 --- a/home-assistant/mqtt.yaml +++ b/home-assistant/mqtt.yaml @@ -1,4 +1,28 @@ mqtt: + binary_sensor: + # DESINFECTION + - name: "Desinfection pH Canister Empty" + unique_id: "desinfection_ph_canister_empty" + icon: mdi:bottle-tonic-alert + state_topic: "/status/disinfection/ph/canister_empty" + payload_on: "1" + payload_off: "0" + device_class: "problem" + device: + identifiers: "Poupool01" + name: "Poupool" + + - name: "Desinfection Cl Canister Empty" + unique_id: "desinfection_cl_canister_empty" + icon: mdi:bottle-tonic-alert + state_topic: "/status/disinfection/cl/canister_empty" + payload_on: "1" + payload_off: "0" + device_class: "problem" + device: + identifiers: "Poupool01" + name: "Poupool" + climate: - name: "Heating" unique_id: "heating" diff --git a/test/test_disinfection.py b/test/test_disinfection.py index ebc6435..000a4db 100644 --- a/test/test_disinfection.py +++ b/test/test_disinfection.py @@ -93,3 +93,226 @@ def test_pwm_many_iterations_0_0(self, pwm, pump): self.run_many_iterations(pwm, iterations) assert pump.on.call_count == 0 assert pump.off.call_count == 0 + + +class TestCanisterMonitor: + def test_insufficient_samples_discarded(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", min_samples=15) + for _ in range(10): + assert not monitor.add_sample(0.50) + assert not monitor.finish_cycle() + assert len(monitor.history) == 0 + assert not monitor.is_empty + + def test_normal_cycles_no_alert(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # Normal fluctuating duty cycles between 1% and 4% + normal_values = [0.01, 0.02, 0.01, 0.03, 0.02, 0.01, 0.04, 0.02, 0.01] + for val in normal_values: + for _ in range(20): + monitor.add_sample(val) + assert not monitor.finish_cycle() + assert not monitor.is_empty + + def test_upward_trend_detection(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # 6 cycles with increasing feedback matching empty canister progression + trend_values = [0.11, 0.15, 0.18, 0.21, 0.23, 0.26] + for idx, val in enumerate(trend_values): + for _ in range(20): + monitor.add_sample(val) + is_last = idx == len(trend_values) - 1 + changed = monitor.finish_cycle() + if is_last: + assert changed + assert monitor.is_empty + assert monitor.alert_stats is not None + assert monitor.alert_stats.peak == pytest.approx(0.26) + else: + assert not changed + assert not monitor.is_empty + + def test_saturation_fallback(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # 6 cycles with high duty cycle saturation + high_values = [0.40, 0.50, 0.45, 0.60, 0.55, 0.50] + for idx, val in enumerate(high_values): + for _ in range(20): + monitor.add_sample(val) + is_last = idx == len(high_values) - 1 + changed = monitor.finish_cycle() + if is_last: + assert changed + assert monitor.is_empty + else: + assert not changed + assert not monitor.is_empty + + def test_summer_spike_recovering_at_night(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # Alternating afternoon spike and overnight low recovery + day_night_values = [0.45, 0.02, 0.50, 0.02, 0.40, 0.02, 0.55, 0.02] + for val in day_night_values: + for _ in range(20): + monitor.add_sample(val) + assert not monitor.finish_cycle() + assert not monitor.is_empty + + def test_auto_clear_on_low_demand(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # Trigger empty state + for val in [0.11, 0.15, 0.18, 0.21, 0.23, 0.26]: + for _ in range(20): + monitor.add_sample(val) + monitor.finish_cycle() + assert monitor.is_empty + + # Canister refilled: low demand for 15 samples clears flag mid-cycle + for _ in range(14): + assert not monitor.add_sample(0.02) + assert monitor.is_empty + assert monitor.add_sample(0.02) + assert not monitor.is_empty + assert len(monitor.history) == 0 + + def test_auto_clear_on_relative_drop(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", cycles=6, min_samples=15) + # Trigger empty state at high saturation + for val in [0.70, 0.75, 0.80, 0.85, 0.90, 0.95]: + for _ in range(20): + monitor.add_sample(val) + monitor.finish_cycle() + assert monitor.is_empty + + # Next cycle drops significantly (peak 0.40, mean 0.20) + # First 14 samples: moderate values + for _ in range(14): + assert not monitor.add_sample(0.20) + # Finish cycle with 20 samples total + for _ in range(6): + assert not monitor.add_sample(0.20) + assert monitor.finish_cycle() + assert not monitor.is_empty + + def test_max_samples_auto_split(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", min_samples=15, max_samples=30) + for _ in range(29): + assert not monitor.add_sample(0.20) + assert len(monitor.history) == 0 + # 30th sample triggers cycle split + monitor.add_sample(0.20) + assert len(monitor.history) == 1 + assert len(monitor.current_samples) == 0 + + def test_restore_and_clear(self): + from controller.disinfection import CanisterMonitor + + monitor = CanisterMonitor("Cl", min_samples=15) + monitor.restore(True) + assert monitor.is_empty + + # Auto-clear after 15 low samples + for _ in range(15): + monitor.add_sample(0.01) + assert not monitor.is_empty + + # Restore False resets state + monitor.restore(True) + assert monitor.is_empty + monitor.restore(False) + assert not monitor.is_empty + + +class TestDisinfectionCanisterIntegration: + @pytest.fixture + def encoder(self, mocker): + return mocker.Mock() + + @pytest.fixture + def devices(self, mocker): + from controller.device import DeviceRegistry, SwitchDevice + + registry = mocker.Mock(DeviceRegistry) + pump_mock = mocker.Mock(SwitchDevice) + registry.get_pump.return_value = pump_mock + return registry + + @pytest.fixture + def sensors_reader(self, mocker): + reader = mocker.Mock() + ph_mock = mocker.Mock() + ph_mock.get.return_value = 7.2 + orp_mock = mocker.Mock() + orp_mock.get.return_value = 650 + reader.get_ph.return_value = ph_mock + reader.get_orp.return_value = orp_mock + return reader + + @pytest.fixture + def sensors_writer(self, mocker): + return mocker.Mock() + + @pytest.fixture + def disinfection(self, encoder, devices, sensors_reader, sensors_writer): + import pykka + + from controller.disinfection import Disinfection + + yield Disinfection(encoder, devices, sensors_reader, sensors_writer, disable=True) + pykka.ActorRegistry.stop_all() + + def test_dispatcher_restore_callbacks(self, disinfection, encoder, sensors_reader): + # Restoring True sets empty state; subsequent low-demand samples will trigger clear publish + encoder.reset_mock() + disinfection.cl_canister_empty(True) + encoder.disinfection_cl_canister__empty.assert_not_called() + + # Calling cl_canister_empty(True) again is idempotent + disinfection.cl_canister_empty(True) + encoder.disinfection_cl_canister__empty.assert_not_called() + + # 15 low-demand samples trigger auto-clear publish + for _ in range(15): + disinfection.on_enter_running_adjusting() + encoder.disinfection_cl_canister__empty.assert_called_once_with(0, retain=True) + + # Restoring False + encoder.reset_mock() + disinfection.cl_canister_empty(False) + encoder.disinfection_cl_canister__empty.assert_not_called() + # When already False, 15 low-demand samples do not publish clear + for _ in range(15): + disinfection.on_enter_running_adjusting() + encoder.disinfection_cl_canister__empty.assert_not_called() + + def test_canister_empty_published_on_transition(self, disinfection, encoder, sensors_reader): + encoder.reset_mock() + # Setpoint is 600, scale is 0.005, pterm is 1.0. + # feedback = (setpoint - current) * pterm * scale = (600 - orp) * 0.005 + # ORPs: 578 (0.11), 570 (0.15), 564 (0.18) + orps = [578, 570, 564] + for idx, orp in enumerate(orps): + sensors_reader.get_orp().get.return_value = orp + for _ in range(15): + disinfection.on_enter_running_adjusting() + disinfection.on_enter_halt() + if idx < len(orps) - 1: + encoder.disinfection_cl_canister__empty.assert_not_called() + else: + encoder.disinfection_cl_canister__empty.assert_called_once_with(1, retain=True)