#!/usr/bin/env python3 """Proof-of-concept control loop for the hot tub. Every LOOP_INTERVAL seconds: 1. Read all DS18B20 probes and average the valid readings (F). 2. Toggle the filter pump relay. 3. Heat pump: ON if avg < HEAT_ON_BELOW_F, OFF if avg > HEAT_OFF_ABOVE_F. Between those thresholds the heat relay is left unchanged (hysteresis). Usage (on the Pi, with 1-Wire overlay enabled): python3 control_poc.py python3 control_poc.py --dry-run # print actions, no GPIO python3 control_poc.py --interval 5 Ctrl+C turns both relays OFF before exit. GPIO note: this script uses FILTER=22 / HEAT=23 as requested for the PoC. hardware-setup.md currently lists the reverse (heat=22, filter=23) — change the constants below if your wiring matches the doc. """ from __future__ import annotations import argparse import sys import time from pathlib import Path from read_temps import DEFAULT_DEVICES_PATH, c_to_f, find_devices, read_temp_c # --- Pin / behavior knobs ----------------------------------------------------- FILTER_GPIO = 22 HEAT_GPIO = 23 HEAT_ON_BELOW_F = 100.0 HEAT_OFF_ABOVE_F = 104.0 LOOP_INTERVAL_S = 5.0 def average_temp_f(devices_path: Path) -> tuple[float | None, list[tuple[str, float | None]]]: """Return (avg_f or None, [(device_id, temp_f or None), ...]).""" readings: list[tuple[str, float | None]] = [] temps_f: list[float] = [] for device_id in find_devices(devices_path): temp_c = read_temp_c(devices_path, device_id) if temp_c is None: readings.append((device_id, None)) continue temp_f = c_to_f(temp_c) readings.append((device_id, temp_f)) temps_f.append(temp_f) if not temps_f: return None, readings return sum(temps_f) / len(temps_f), readings class RelayOutputs: """Thin wrapper so --dry-run can skip real GPIO.""" def __init__(self, dry_run: bool) -> None: self.dry_run = dry_run self.filter_on = False self.heat_on = False self._filter = None self._heat = None if dry_run: return try: from gpiozero import OutputDevice except ImportError as exc: raise SystemExit( "gpiozero is required on the Pi. Install with:\n" " sudo apt install python3-gpiozero\n" "Or run with --dry-run to exercise the loop without GPIO." ) from exc # active_high=True: GPIO high energizes the relay control input. # Invert if your IoT Relay / module is active-low. self._filter = OutputDevice(FILTER_GPIO, active_high=True, initial_value=False) self._heat = OutputDevice(HEAT_GPIO, active_high=True, initial_value=False) def set_filter(self, on: bool) -> None: self.filter_on = on if self._filter is not None: self._filter.on() if on else self._filter.off() def set_heat(self, on: bool) -> None: self.heat_on = on if self._heat is not None: self._heat.on() if on else self._heat.off() def toggle_filter(self) -> bool: self.set_filter(not self.filter_on) return self.filter_on def all_off(self) -> None: self.set_filter(False) self.set_heat(False) def close(self) -> None: self.all_off() if self._filter is not None: self._filter.close() if self._heat is not None: self._heat.close() def decide_heat(avg_f: float | None, heat_currently_on: bool) -> bool | None: """Return desired heat state, or None to leave unchanged. None also covers "no valid average" — heat must not turn on without a reading. If heat is already on and sensors disappear, turn it off for safety. """ if avg_f is None: return False if heat_currently_on else None if avg_f < HEAT_ON_BELOW_F: return True if avg_f > HEAT_OFF_ABOVE_F: return False return None def run_loop(devices_path: Path, interval_s: float, dry_run: bool) -> None: relays = RelayOutputs(dry_run=dry_run) mode = "DRY-RUN" if dry_run else "LIVE" print( f"control_poc starting ({mode}): " f"filter=GPIO{FILTER_GPIO}, heat=GPIO{HEAT_GPIO}, " f"heat on <{HEAT_ON_BELOW_F:g}F / off >{HEAT_OFF_ABOVE_F:g}F, " f"interval={interval_s:g}s" ) print("Ctrl+C to stop (relays forced OFF).\n") try: while True: stamp = time.strftime("%Y-%m-%d %H:%M:%S") avg_f, readings = average_temp_f(devices_path) print(f"--- {stamp} ---") if not readings: print(" no 1-Wire devices found") for device_id, temp_f in readings: if temp_f is None: print(f" {device_id}: --") else: print(f" {device_id}: {temp_f:6.2f} F") if avg_f is None: print(" avg: -- (no valid readings)") else: print(f" avg: {avg_f:6.2f} F ({len([r for r in readings if r[1] is not None])} probes)") filter_on = relays.toggle_filter() print(f" filter: {'ON ' if filter_on else 'OFF'} (toggled)") desired_heat = decide_heat(avg_f, relays.heat_on) if desired_heat is None: print(f" heat: {'ON ' if relays.heat_on else 'OFF'} (unchanged)") else: relays.set_heat(desired_heat) reason = ( "no sensors" if avg_f is None else f"avg {'<' if desired_heat else '>'} " f"{HEAT_ON_BELOW_F if desired_heat else HEAT_OFF_ABOVE_F:g}F" ) print(f" heat: {'ON ' if desired_heat else 'OFF'} ({reason})") print() time.sleep(interval_s) except KeyboardInterrupt: print("\nStopping — turning relays OFF.") finally: relays.close() print("Done.") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--devices-path", type=Path, default=DEFAULT_DEVICES_PATH, help=f"1-Wire devices directory (default: {DEFAULT_DEVICES_PATH})", ) parser.add_argument( "--interval", type=float, default=LOOP_INTERVAL_S, help=f"seconds between loop iterations (default: {LOOP_INTERVAL_S})", ) parser.add_argument( "--dry-run", action="store_true", help="print decisions without driving GPIO", ) args = parser.parse_args() if args.interval <= 0: print("--interval must be > 0", file=sys.stderr) sys.exit(2) run_loop(args.devices_path, args.interval, args.dry_run) if __name__ == "__main__": main()