projects/hot-tub/deploy/control/hot-tub-control.logrotate
... ...
@@ -0,0 +1,9 @@
1
+/var/log/hot-tub/control.log {
2
+ weekly
3
+ rotate 8
4
+ compress
5
+ delaycompress
6
+ missingok
7
+ notifempty
8
+ copytruncate
9
+}
projects/hot-tub/deploy/control/hot-tub-control.service
... ...
@@ -0,0 +1,17 @@
1
+[Unit]
2
+Description=Hot Tub Control Daemon
3
+After=network-online.target redis-server.service
4
+Wants=network-online.target
5
+Documentation=file://INSTALL.md
6
+
7
+[Service]
8
+Type=simple
9
+User=hot-tub
10
+Group=hot-tub
11
+Environment=PYTHONPATH=__INSTALL_ROOT__/src
12
+ExecStart=/usr/bin/python3 -m control --config /etc/hot-tub/control.json
13
+Restart=on-failure
14
+RestartSec=5
15
+
16
+[Install]
17
+WantedBy=multi-user.target
projects/hot-tub/deploy/control/install.sh
... ...
@@ -0,0 +1,68 @@
1
+#!/usr/bin/env bash
2
+# Install hot-tub control daemon on Raspberry Pi OS (run with sudo).
3
+set -euo pipefail
4
+
5
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
6
+REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
7
+INSTALL_ROOT="${INSTALL_ROOT:-${REPO_ROOT}}"
8
+
9
+if [[ "$(id -u)" -ne 0 ]]; then
10
+ echo "Run as root: sudo $0" >&2
11
+ exit 1
12
+fi
13
+
14
+echo "==> Installing packages (python3-redis, python3-gpiozero)..."
15
+export DEBIAN_FRONTEND=noninteractive
16
+apt-get update -qq
17
+apt-get install -y -qq python3-redis python3-gpiozero
18
+
19
+if ! systemctl is-active --quiet redis-server 2>/dev/null && ! systemctl is-active --quiet redis 2>/dev/null; then
20
+ echo "==> Redis not running; installing redis-server..."
21
+ apt-get install -y -qq redis-server
22
+fi
23
+
24
+echo "==> Creating hot-tub system user..."
25
+if ! id hot-tub &>/dev/null; then
26
+ useradd --system --home /var/lib/hot-tub --shell /usr/sbin/nologin hot-tub
27
+fi
28
+usermod -aG gpio hot-tub 2>/dev/null || true
29
+
30
+echo "==> Preparing log, config, and state paths..."
31
+mkdir -p /var/log/hot-tub /etc/hot-tub /var/lib/hot-tub
32
+chown hot-tub:hot-tub /var/log/hot-tub /var/lib/hot-tub
33
+
34
+if [[ ! -f /etc/hot-tub/control.json ]]; then
35
+ install -m 0644 "${INSTALL_ROOT}/src/control/config.example.json" /etc/hot-tub/control.json
36
+ echo " Created /etc/hot-tub/control.json from example"
37
+else
38
+ echo " Keeping existing /etc/hot-tub/control.json"
39
+fi
40
+
41
+chown -R hot-tub:hot-tub "${INSTALL_ROOT}/src/control" 2>/dev/null || true
42
+
43
+echo "==> Installing systemd unit..."
44
+sed "s|__INSTALL_ROOT__|${INSTALL_ROOT}|g" "${SCRIPT_DIR}/hot-tub-control.service" \
45
+ > /etc/systemd/system/hot-tub-control.service
46
+
47
+echo "==> Installing logrotate..."
48
+install -m 0644 "${SCRIPT_DIR}/hot-tub-control.logrotate" /etc/logrotate.d/hot-tub-control
49
+
50
+echo "==> Enabling services..."
51
+systemctl enable redis-server 2>/dev/null || systemctl enable redis 2>/dev/null || true
52
+systemctl start redis-server 2>/dev/null || systemctl start redis 2>/dev/null || true
53
+systemctl daemon-reload
54
+systemctl enable hot-tub-control.service
55
+systemctl restart hot-tub-control.service || systemctl start hot-tub-control.service
56
+
57
+echo ""
58
+echo "Installation complete."
59
+echo " Repo / code root: ${INSTALL_ROOT}"
60
+echo " Config: /etc/hot-tub/control.json"
61
+echo " State: /var/lib/hot-tub/control-state.json"
62
+echo " Log: /var/log/hot-tub/control.log"
63
+echo ""
64
+echo "Do not run src/control_poc.py while this service is active (same GPIO pins)."
65
+echo ""
66
+echo "Status: systemctl status hot-tub-control"
67
+echo "Logs: tail -f /var/log/hot-tub/control.log"
68
+echo "Redis: redis-cli GET control:latest"
projects/hot-tub/src/control/INSTALL.md
... ...
@@ -0,0 +1,95 @@
1
+# Control Daemon — Installation (Raspberry Pi)
2
+
3
+Install the control daemon on the Pi 3 controller. Design details: [README.md](README.md).
4
+
5
+## Prerequisites
6
+
7
+- Raspberry Pi OS with Python 3
8
+- **Redis** running locally (`redis-server`)
9
+- **Temperature daemon** recommended (`hot-tub-temps`) so `temps:latest` is fresh
10
+- Heat relay on **GPIO22**, filter relay on **GPIO23** (see [hardware-setup.md](../../hardware-setup.md))
11
+- IoT relay modules wired; confirm `active_high` in config matches your modules
12
+
13
+## 1. Stop conflicting GPIO users
14
+
15
+Do **not** run `src/control_poc.py` or other relay experiments while the control service is enabled — they share the same pins.
16
+
17
+```bash
18
+# If you were testing the PoC:
19
+pkill -f control_poc.py || true
20
+```
21
+
22
+## 2. Install service
23
+
24
+From the hot-tub repository on the Pi:
25
+
26
+```bash
27
+cd /path/to/hot-tub
28
+sudo deploy/control/install.sh
29
+```
30
+
31
+Optional fixed install root:
32
+
33
+```bash
34
+sudo INSTALL_ROOT=/opt/hot-tub deploy/control/install.sh
35
+```
36
+
37
+The script installs `python3-redis`, `python3-gpiozero`, adds user `hot-tub` to group `gpio`, installs `/etc/hot-tub/control.json`, systemd unit `hot-tub-control`, and logrotate for `/var/log/hot-tub/control.log`.
38
+
39
+## 3. Configure
40
+
41
+Edit `/etc/hot-tub/control.json` if needed (GPIO pins, maintain band, filter auto schedule, target limits).
42
+
43
+Persisted operator state lives in `/var/lib/hot-tub/control-state.json` (`heat_mode`, `filter_mode`, `target_f`, latched startup error).
44
+
45
+## 4. Verify
46
+
47
+```bash
48
+systemctl status hot-tub-control
49
+tail -f /var/log/hot-tub/control.log
50
+redis-cli GET control:latest
51
+redis-cli GET health:control
52
+redis-cli SUBSCRIBE control.state
53
+```
54
+
55
+Send test commands:
56
+
57
+```bash
58
+redis-cli PUBLISH control.command '{"op":"set_target_temp","args":{"target_f":102},"id":"1","source":"cli"}'
59
+redis-cli PUBLISH control.command '{"op":"set_heat_mode","args":{"mode":"maintain"},"id":"2","source":"cli"}'
60
+redis-cli GET control:latest
61
+```
62
+
63
+Expect heat to stay off until `temps:latest` is valid and fresh (default stale threshold 15 s).
64
+
65
+## 5. Development (off the Pi or dry-run)
66
+
67
+```bash
68
+cd src
69
+PYTHONPATH=. python3 -m control --dry-run --once
70
+PYTHONPATH=. python3 -m unittest control.tests.test_commands control.tests.test_safety \
71
+ control.tests.test_filter_policy control.tests.test_heat_fsm -v
72
+```
73
+
74
+Dry-run uses `/tmp/hot-tub-control.log` and `/tmp/hot-tub-control-state.json` when system paths are not writable.
75
+
76
+## File locations
77
+
78
+| Path | Purpose |
79
+|---|---|
80
+| `/etc/hot-tub/control.json` | Daemon config |
81
+| `/var/lib/hot-tub/control-state.json` | Persisted modes / target / latched error |
82
+| `/var/log/hot-tub/control.log` | Application log |
83
+| `/etc/systemd/system/hot-tub-control.service` | systemd unit |
84
+| `/etc/logrotate.d/hot-tub-control` | Log rotation |
85
+
86
+## Uninstall
87
+
88
+```bash
89
+sudo systemctl disable --now hot-tub-control
90
+sudo rm /etc/systemd/system/hot-tub-control.service
91
+sudo rm /etc/logrotate.d/hot-tub-control
92
+sudo systemctl daemon-reload
93
+```
94
+
95
+Config and state under `/etc/hot-tub` and `/var/lib/hot-tub` are left in place unless you remove them manually.
projects/hot-tub/src/control/README.md
... ...
@@ -1,5 +1,7 @@
1 1
# Control Daemon
2 2
3
+**Installation:** [INSTALL.md](INSTALL.md)
4
+
3 5
Sole owner of heat and filter pump relays on the Pi. Consumes temperature readings and user commands from Redis, applies mode / safety logic, drives GPIO, and publishes canonical system state (and health) back to Redis.
4 6
5 7
See [software-architecture.md](../../software-architecture.md) for system context. GPIO pin defaults match [hardware-setup.md](../../hardware-setup.md) (heat relay GPIO22, filter relay GPIO23 unless config overrides).
projects/hot-tub/src/control/__init__.py
... ...
@@ -0,0 +1 @@
1
+"""Hot tub control daemon — sole owner of heat and filter pump relays."""
projects/hot-tub/src/control/__main__.py
... ...
@@ -0,0 +1,78 @@
1
+"""CLI entry: python -m control"""
2
+
3
+from __future__ import annotations
4
+
5
+import argparse
6
+import json
7
+import sys
8
+from pathlib import Path
9
+
10
+from .config import load_config
11
+from .daemon import run_loop, run_once
12
+from .logging_setup import setup_logging
13
+
14
+
15
+def _path_writable(dir_path: Path) -> bool:
16
+ try:
17
+ dir_path.mkdir(parents=True, exist_ok=True)
18
+ test = dir_path / ".write_test"
19
+ test.write_text("")
20
+ test.unlink()
21
+ return True
22
+ except OSError:
23
+ return False
24
+
25
+
26
+def main() -> None:
27
+ parser = argparse.ArgumentParser(description="Hot tub control daemon")
28
+ parser.add_argument(
29
+ "--config",
30
+ type=Path,
31
+ default=Path("/etc/hot-tub/control.json"),
32
+ help="Path to JSON config (default: /etc/hot-tub/control.json)",
33
+ )
34
+ parser.add_argument(
35
+ "--once",
36
+ action="store_true",
37
+ help="Run one evaluation cycle and exit",
38
+ )
39
+ parser.add_argument(
40
+ "--dry-run",
41
+ action="store_true",
42
+ help="No Redis; log GPIO decisions only",
43
+ )
44
+ parser.add_argument(
45
+ "-v",
46
+ "--verbose",
47
+ action="store_true",
48
+ help="Debug logging",
49
+ )
50
+ args = parser.parse_args()
51
+
52
+ cfg_path = args.config if args.config.exists() else None
53
+ if cfg_path is None:
54
+ dev_example = Path(__file__).resolve().parent / "config.example.json"
55
+ cfg_path = dev_example if dev_example.exists() else None
56
+
57
+ cfg = load_config(cfg_path)
58
+
59
+ if args.dry_run or not _path_writable(cfg.log_path.parent):
60
+ cfg.log_path = Path("/tmp/hot-tub-control.log")
61
+ if args.dry_run or not _path_writable(cfg.state_path.parent):
62
+ cfg.state_path = Path("/tmp/hot-tub-control-state.json")
63
+
64
+ setup_logging(cfg.log_path, verbose=args.verbose)
65
+
66
+ if args.once:
67
+ snapshot = run_once(cfg, dry_run=args.dry_run)
68
+ print(json.dumps(snapshot, indent=2))
69
+ return
70
+
71
+ run_loop(cfg, dry_run=args.dry_run)
72
+
73
+
74
+if __name__ == "__main__":
75
+ try:
76
+ main()
77
+ except KeyboardInterrupt:
78
+ sys.exit(0)
projects/hot-tub/src/control/commands.py
... ...
@@ -0,0 +1,106 @@
1
+"""Parse and apply control.command payloads."""
2
+
3
+from __future__ import annotations
4
+
5
+import logging
6
+from typing import Any
7
+
8
+from .config import ControlConfig
9
+from .state import FILTER_MODES, HEAT_MODES, ControlState
10
+
11
+logger = logging.getLogger("hot_tub.control")
12
+
13
+
14
+class CommandError(ValueError):
15
+ pass
16
+
17
+
18
+def _clamp_target(target: float, cfg: ControlConfig) -> float:
19
+ return max(cfg.target_f_min, min(cfg.target_f_max, target))
20
+
21
+
22
+def apply_command(state: ControlState, cfg: ControlConfig, envelope: dict[str, Any]) -> bool:
23
+ """Apply one command. Returns True if intents changed (persist recommended)."""
24
+ op = envelope.get("op")
25
+ if not op or not isinstance(op, str):
26
+ raise CommandError("missing or invalid op")
27
+
28
+ cmd_id = envelope.get("id")
29
+ if cmd_id is not None:
30
+ state.last_command_id = str(cmd_id)
31
+
32
+ args = envelope.get("args")
33
+ if args is not None and not isinstance(args, dict):
34
+ raise CommandError("args must be an object")
35
+ args = args or {}
36
+
37
+ changed = False
38
+
39
+ if op == "set_heat_mode":
40
+ mode = args.get("mode")
41
+ if mode not in HEAT_MODES:
42
+ raise CommandError(f"invalid heat mode: {mode!r}")
43
+ if state.heat_mode != mode:
44
+ logger.info("heat_mode %s -> %s", state.heat_mode, mode)
45
+ state.heat_mode = mode # type: ignore[assignment]
46
+ changed = True
47
+ if mode == "sleep" and state.heat_phase == "error_startup":
48
+ state.error = None
49
+ state.startup_attempt = 0
50
+ state.heat_phase = "idle"
51
+ logger.info("error_startup cleared via sleep")
52
+ changed = True
53
+ return changed
54
+
55
+ if op == "set_target_temp":
56
+ raw = args.get("target_f")
57
+ if raw is None:
58
+ raise CommandError("target_f required")
59
+ target = _clamp_target(float(raw), cfg)
60
+ if state.target_f != target:
61
+ logger.info("target_f %.1f -> %.1f", state.target_f, target)
62
+ state.target_f = target
63
+ changed = True
64
+ return changed
65
+
66
+ if op == "target_temp_increment":
67
+ delta = float(args.get("delta_f", 1))
68
+ target = _clamp_target(state.target_f + delta, cfg)
69
+ if state.target_f != target:
70
+ logger.info("target_f %.1f -> %.1f (+%s)", state.target_f, target, delta)
71
+ state.target_f = target
72
+ changed = True
73
+ return changed
74
+
75
+ if op == "target_temp_decrement":
76
+ delta = float(args.get("delta_f", 1))
77
+ target = _clamp_target(state.target_f - delta, cfg)
78
+ if state.target_f != target:
79
+ logger.info("target_f %.1f -> %.1f (-%s)", state.target_f, target, delta)
80
+ state.target_f = target
81
+ changed = True
82
+ return changed
83
+
84
+ if op == "set_filter_mode":
85
+ mode = args.get("mode")
86
+ if mode not in FILTER_MODES:
87
+ raise CommandError(f"invalid filter mode: {mode!r}")
88
+ if state.filter_mode != mode:
89
+ logger.info("filter_mode %s -> %s", state.filter_mode, mode)
90
+ state.filter_mode = mode # type: ignore[assignment]
91
+ changed = True
92
+ return changed
93
+
94
+ raise CommandError(f"unknown op: {op!r}")
95
+
96
+
97
+def parse_command_message(raw: str) -> dict[str, Any]:
98
+ import json
99
+
100
+ try:
101
+ data = json.loads(raw)
102
+ except json.JSONDecodeError as exc:
103
+ raise CommandError("invalid JSON") from exc
104
+ if not isinstance(data, dict):
105
+ raise CommandError("command must be a JSON object")
106
+ return data
projects/hot-tub/src/control/config.example.json
... ...
@@ -0,0 +1,32 @@
1
+{
2
+ "redis_url": "redis://127.0.0.1:6379/0",
3
+ "log_path": "/var/log/hot-tub/control.log",
4
+ "state_path": "/var/lib/hot-tub/control-state.json",
5
+ "gpio": {
6
+ "heat_pump": 22,
7
+ "filter_pump": 23,
8
+ "active_high": true
9
+ },
10
+ "maintain": {
11
+ "band_f": 5.0,
12
+ "startup_delay_s": 20.0,
13
+ "hot_in_rise_f": 10.0,
14
+ "startup_attempts": 3,
15
+ "retry_off_s": 5.0
16
+ },
17
+ "backflush": {
18
+ "duration_s": 10.0,
19
+ "revert_to_heat_mode": "sleep"
20
+ },
21
+ "filter_auto": {
22
+ "start_time": "00:00",
23
+ "run_time_hours": 2.0,
24
+ "timezone": "local"
25
+ },
26
+ "target_f_default": 102,
27
+ "target_f_min": 50,
28
+ "target_f_max": 110,
29
+ "temps_stale_s": 15,
30
+ "loop_interval_s": 1.0,
31
+ "heartbeat_s": 5.0
32
+}
projects/hot-tub/src/control/config.py
... ...
@@ -0,0 +1,132 @@
1
+"""Load control daemon configuration."""
2
+
3
+from __future__ import annotations
4
+
5
+import json
6
+from dataclasses import dataclass, field
7
+from pathlib import Path
8
+
9
+
10
+@dataclass
11
+class GpioConfig:
12
+ heat_pump: int = 22
13
+ filter_pump: int = 23
14
+ active_high: bool = True
15
+
16
+
17
+@dataclass
18
+class MaintainConfig:
19
+ band_f: float = 5.0
20
+ startup_delay_s: float = 20.0
21
+ hot_in_rise_f: float = 10.0
22
+ startup_attempts: int = 3
23
+ retry_off_s: float = 5.0
24
+
25
+
26
+@dataclass
27
+class BackflushConfig:
28
+ duration_s: float = 10.0
29
+ revert_to_heat_mode: str = "sleep"
30
+
31
+
32
+@dataclass
33
+class FilterAutoConfig:
34
+ start_time: str = "00:00"
35
+ run_time_hours: float = 2.0
36
+ timezone: str = "local"
37
+
38
+
39
+@dataclass
40
+class ControlConfig:
41
+ redis_url: str = "redis://127.0.0.1:6379/0"
42
+ log_path: Path = Path("/var/log/hot-tub/control.log")
43
+ state_path: Path = Path("/var/lib/hot-tub/control-state.json")
44
+ gpio: GpioConfig = field(default_factory=GpioConfig)
45
+ maintain: MaintainConfig = field(default_factory=MaintainConfig)
46
+ backflush: BackflushConfig = field(default_factory=BackflushConfig)
47
+ filter_auto: FilterAutoConfig = field(default_factory=FilterAutoConfig)
48
+ target_f_default: float = 102.0
49
+ target_f_min: float = 50.0
50
+ target_f_max: float = 110.0
51
+ temps_stale_s: float = 15.0
52
+ loop_interval_s: float = 1.0
53
+ heartbeat_s: float = 5.0
54
+
55
+
56
+def _merge_gpio(data: dict) -> GpioConfig:
57
+ g = GpioConfig()
58
+ block = data.get("gpio") or {}
59
+ if "heat_pump" in block:
60
+ g.heat_pump = int(block["heat_pump"])
61
+ if "filter_pump" in block:
62
+ g.filter_pump = int(block["filter_pump"])
63
+ if "active_high" in block:
64
+ g.active_high = bool(block["active_high"])
65
+ return g
66
+
67
+
68
+def _merge_maintain(data: dict) -> MaintainConfig:
69
+ m = MaintainConfig()
70
+ block = data.get("maintain") or {}
71
+ for key, typ in (
72
+ ("band_f", float),
73
+ ("startup_delay_s", float),
74
+ ("hot_in_rise_f", float),
75
+ ("startup_attempts", int),
76
+ ("retry_off_s", float),
77
+ ):
78
+ if key in block:
79
+ setattr(m, key, typ(block[key]))
80
+ return m
81
+
82
+
83
+def _merge_backflush(data: dict) -> BackflushConfig:
84
+ b = BackflushConfig()
85
+ block = data.get("backflush") or {}
86
+ if "duration_s" in block:
87
+ b.duration_s = float(block["duration_s"])
88
+ if "revert_to_heat_mode" in block:
89
+ b.revert_to_heat_mode = str(block["revert_to_heat_mode"])
90
+ return b
91
+
92
+
93
+def _merge_filter_auto(data: dict) -> FilterAutoConfig:
94
+ f = FilterAutoConfig()
95
+ block = data.get("filter_auto") or {}
96
+ if "start_time" in block:
97
+ f.start_time = str(block["start_time"])
98
+ if "run_time_hours" in block:
99
+ f.run_time_hours = float(block["run_time_hours"])
100
+ if "timezone" in block:
101
+ f.timezone = str(block["timezone"])
102
+ return f
103
+
104
+
105
+def load_config(path: Path | None) -> ControlConfig:
106
+ cfg = ControlConfig()
107
+ if path is None or not path.exists():
108
+ return cfg
109
+ data = json.loads(path.read_text())
110
+ if "redis_url" in data:
111
+ cfg.redis_url = str(data["redis_url"])
112
+ if "log_path" in data:
113
+ cfg.log_path = Path(data["log_path"])
114
+ if "state_path" in data:
115
+ cfg.state_path = Path(data["state_path"])
116
+ cfg.gpio = _merge_gpio(data)
117
+ cfg.maintain = _merge_maintain(data)
118
+ cfg.backflush = _merge_backflush(data)
119
+ cfg.filter_auto = _merge_filter_auto(data)
120
+ if "target_f_default" in data:
121
+ cfg.target_f_default = float(data["target_f_default"])
122
+ if "target_f_min" in data:
123
+ cfg.target_f_min = float(data["target_f_min"])
124
+ if "target_f_max" in data:
125
+ cfg.target_f_max = float(data["target_f_max"])
126
+ if "temps_stale_s" in data:
127
+ cfg.temps_stale_s = float(data["temps_stale_s"])
128
+ if "loop_interval_s" in data:
129
+ cfg.loop_interval_s = float(data["loop_interval_s"])
130
+ if "heartbeat_s" in data:
131
+ cfg.heartbeat_s = float(data["heartbeat_s"])
132
+ return cfg
projects/hot-tub/src/control/daemon.py
... ...
@@ -0,0 +1,190 @@
1
+"""Control daemon main loop."""
2
+
3
+from __future__ import annotations
4
+
5
+import logging
6
+import os
7
+import signal
8
+import time
9
+from datetime import datetime, timezone
10
+from typing import Any
11
+
12
+from .commands import CommandError, apply_command, parse_command_message
13
+from .config import ControlConfig
14
+from .filter_policy import desired_filter_pump
15
+from .gpio_out import RelayOutputs
16
+from .heat_fsm import tick_heat
17
+from .redis_io import CHAN_COMMAND, CHAN_TEMPS_READING, ControlRedis
18
+from .safety import temps_fresh
19
+from .state import ControlState, FsmContext, load_persisted_state, save_persisted_state, state_snapshot
20
+
21
+logger = logging.getLogger("hot_tub.control")
22
+
23
+_stop = False
24
+
25
+
26
+def _handle_signal(signum: int, _frame: object) -> None:
27
+ global _stop
28
+ logger.info("Received signal %s, shutting down", signum)
29
+ _stop = True
30
+
31
+
32
+def _utc_now() -> str:
33
+ return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
34
+
35
+
36
+def _health_status(state: ControlState, cfg: ControlConfig, reading: dict[str, Any] | None) -> tuple[str, str]:
37
+ if state.heat_phase.startswith("error_"):
38
+ return "degraded", state.heat_phase
39
+ if state.heat_mode == "maintain" and not temps_fresh(reading, cfg):
40
+ return "degraded", "stale temps while maintain"
41
+ return "ok", "loop running"
42
+
43
+
44
+def _apply_gpio(relays: RelayOutputs, state: ControlState) -> None:
45
+ relays.sync(
46
+ heat=state.heat_pump == "on",
47
+ filter_pump=state.filter_pump == "on",
48
+ )
49
+ state.heat_pump = "on" if relays.heat_on else "off"
50
+ state.filter_pump = "on" if relays.filter_on else "off"
51
+
52
+
53
+class ControlEngine:
54
+ def __init__(self, cfg: ControlConfig, *, dry_run: bool = False) -> None:
55
+ self.cfg = cfg
56
+ self.dry_run = dry_run
57
+ self.state = load_persisted_state(cfg.state_path, default_target=cfg.target_f_default)
58
+ self.ctx = FsmContext()
59
+ self.reading: dict[str, Any] | None = None
60
+ self.relays = RelayOutputs(cfg.gpio, dry_run=dry_run)
61
+ self.relays.all_off()
62
+ self._last_persist_hash: str | None = None
63
+ self._last_publish_mono = 0.0
64
+ self._last_snapshot_json: str | None = None
65
+
66
+ def close(self) -> None:
67
+ self.relays.all_off()
68
+ self.relays.close()
69
+
70
+ def _maybe_persist(self) -> None:
71
+ body = str(self.state.to_persist())
72
+ if body == self._last_persist_hash:
73
+ return
74
+ if not self.dry_run:
75
+ save_persisted_state(self.cfg.state_path, self.state)
76
+ self._last_persist_hash = body
77
+
78
+ def ingest_temps(self, payload: dict[str, Any]) -> None:
79
+ self.reading = payload
80
+
81
+ def handle_command_raw(self, raw: str) -> None:
82
+ try:
83
+ envelope = parse_command_message(raw)
84
+ changed = apply_command(self.state, self.cfg, envelope)
85
+ if changed:
86
+ self._maybe_persist()
87
+ except CommandError as exc:
88
+ logger.warning("command rejected: %s", exc)
89
+ except Exception:
90
+ logger.exception("command handling failed")
91
+
92
+ def tick(self) -> dict[str, Any]:
93
+ tick_heat(self.state, self.ctx, self.cfg, self.reading)
94
+ filt = desired_filter_pump(self.state, self.cfg, self.ctx)
95
+ self.state.filter_pump = filt
96
+ _apply_gpio(self.relays, self.state)
97
+ return state_snapshot(self.state, ts=_utc_now())
98
+
99
+ def should_publish(self, snapshot: dict[str, Any], now_mono: float) -> bool:
100
+ body = str(sorted(snapshot.items()))
101
+ if body != self._last_snapshot_json:
102
+ return True
103
+ return (now_mono - self._last_publish_mono) >= self.cfg.heartbeat_s
104
+
105
+ def mark_published(self, snapshot: dict[str, Any], now_mono: float) -> None:
106
+ self._last_snapshot_json = str(sorted(snapshot.items()))
107
+ self._last_publish_mono = now_mono
108
+
109
+
110
+def run_once(
111
+ cfg: ControlConfig,
112
+ *,
113
+ dry_run: bool = False,
114
+ engine: ControlEngine | None = None,
115
+ redis_io: ControlRedis | None = None,
116
+) -> dict[str, Any]:
117
+ eng = engine or ControlEngine(cfg, dry_run=dry_run)
118
+ if eng.reading is None and not dry_run:
119
+ r = redis_io or ControlRedis(cfg.redis_url)
120
+ try:
121
+ latest = r.get_temps_latest()
122
+ if latest:
123
+ eng.ingest_temps(latest)
124
+ finally:
125
+ if redis_io is None:
126
+ r.close()
127
+ snapshot = eng.tick()
128
+ return snapshot
129
+
130
+
131
+def run_loop(cfg: ControlConfig, *, dry_run: bool = False) -> None:
132
+ global _stop
133
+ _stop = False
134
+ signal.signal(signal.SIGTERM, _handle_signal)
135
+ signal.signal(signal.SIGINT, _handle_signal)
136
+
137
+ logger.info("Control daemon starting (dry_run=%s)", dry_run)
138
+ engine = ControlEngine(cfg, dry_run=dry_run)
139
+ redis_io: ControlRedis | None = None
140
+ if not dry_run:
141
+ redis_io = ControlRedis(cfg.redis_url)
142
+ latest = redis_io.get_temps_latest()
143
+ if latest:
144
+ engine.ingest_temps(latest)
145
+
146
+ try:
147
+ while not _stop:
148
+ loop_start = time.monotonic()
149
+ if redis_io is not None:
150
+ for channel, data in redis_io.poll_messages(timeout_s=min(0.5, cfg.loop_interval_s)):
151
+ if channel == CHAN_COMMAND:
152
+ engine.handle_command_raw(data)
153
+ elif channel == CHAN_TEMPS_READING:
154
+ try:
155
+ import json
156
+
157
+ payload = json.loads(data)
158
+ if isinstance(payload, dict):
159
+ engine.ingest_temps(payload)
160
+ except json.JSONDecodeError:
161
+ logger.warning("invalid temps.reading JSON")
162
+
163
+ snapshot = engine.tick()
164
+ now_mono = time.monotonic()
165
+
166
+ if engine.should_publish(snapshot, now_mono):
167
+ status, detail = _health_status(engine.state, cfg, engine.reading)
168
+ health = {
169
+ "service": "control",
170
+ "ts": snapshot["ts"],
171
+ "status": status,
172
+ "detail": detail,
173
+ "pid": os.getpid(),
174
+ }
175
+ if dry_run:
176
+ logger.debug("state %s", snapshot)
177
+ elif redis_io is not None:
178
+ redis_io.publish_state(snapshot)
179
+ redis_io.publish_health(health)
180
+ engine.mark_published(snapshot, now_mono)
181
+
182
+ elapsed = time.monotonic() - loop_start
183
+ sleep_s = max(0.0, cfg.loop_interval_s - elapsed)
184
+ if sleep_s > 0 and not _stop:
185
+ time.sleep(sleep_s)
186
+ finally:
187
+ if redis_io is not None:
188
+ redis_io.close()
189
+ engine.close()
190
+ logger.info("Control daemon stopped")
projects/hot-tub/src/control/filter_policy.py
... ...
@@ -0,0 +1,94 @@
1
+"""Filter pump mode policy (on / off / auto daily window)."""
2
+
3
+from __future__ import annotations
4
+
5
+from dataclasses import dataclass
6
+from datetime import date, datetime, timedelta, timezone
7
+from zoneinfo import ZoneInfo
8
+
9
+from .config import ControlConfig
10
+from .state import ControlState, FsmContext, PumpState
11
+
12
+
13
+@dataclass
14
+class FilterWindow:
15
+ active: bool
16
+ next_start_iso: str | None
17
+
18
+
19
+def _local_tz(cfg: ControlConfig) -> ZoneInfo:
20
+ if cfg.filter_auto.timezone == "local":
21
+ return datetime.now().astimezone().tzinfo or ZoneInfo("UTC") # type: ignore[return-value]
22
+ return ZoneInfo(cfg.filter_auto.timezone)
23
+
24
+
25
+def _parse_start_time(s: str) -> tuple[int, int]:
26
+ parts = s.strip().split(":")
27
+ if len(parts) != 2:
28
+ raise ValueError(f"invalid start_time: {s!r}")
29
+ return int(parts[0]), int(parts[1])
30
+
31
+
32
+def window_for_day(
33
+ cfg: ControlConfig,
34
+ day: date,
35
+ *,
36
+ tz: ZoneInfo | None = None,
37
+) -> tuple[datetime, datetime]:
38
+ tz = tz or _local_tz(cfg)
39
+ hour, minute = _parse_start_time(cfg.filter_auto.start_time)
40
+ start = datetime(day.year, day.month, day.day, hour, minute, tzinfo=tz)
41
+ end = start + timedelta(hours=cfg.filter_auto.run_time_hours)
42
+ return start, end
43
+
44
+
45
+def evaluate_filter_window(cfg: ControlConfig, now: datetime | None = None) -> FilterWindow:
46
+ tz = _local_tz(cfg)
47
+ now = now or datetime.now(tz)
48
+ if now.tzinfo is None:
49
+ now = now.replace(tzinfo=tz)
50
+ else:
51
+ now = now.astimezone(tz)
52
+
53
+ start, end = window_for_day(cfg, now.date(), tz=tz)
54
+ if start <= now < end:
55
+ return FilterWindow(active=True, next_start_iso=None)
56
+
57
+ if now < start:
58
+ next_start = start
59
+ else:
60
+ next_day = now.date() + timedelta(days=1)
61
+ next_start, _ = window_for_day(cfg, next_day, tz=tz)
62
+ return FilterWindow(
63
+ active=False,
64
+ next_start_iso=next_start.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
65
+ )
66
+
67
+
68
+def desired_filter_pump(
69
+ state: ControlState,
70
+ cfg: ControlConfig,
71
+ ctx: FsmContext,
72
+ now: datetime | None = None,
73
+) -> PumpState:
74
+ mode = state.filter_mode
75
+ if mode == "off":
76
+ state.filter_window_active = False
77
+ state.filter_next_start = None
78
+ ctx.auto_run_started_mono = None
79
+ return "off"
80
+ if mode == "on":
81
+ state.filter_window_active = False
82
+ state.filter_next_start = None
83
+ ctx.auto_run_started_mono = None
84
+ return "on"
85
+
86
+ win = evaluate_filter_window(cfg, now=now)
87
+ state.filter_next_start = win.next_start_iso
88
+ if not win.active:
89
+ state.filter_window_active = False
90
+ ctx.auto_run_started_mono = None
91
+ return "off"
92
+
93
+ state.filter_window_active = True
94
+ return "on"
projects/hot-tub/src/control/gpio_out.py
... ...
@@ -0,0 +1,80 @@
1
+"""Heat and filter relay GPIO outputs."""
2
+
3
+from __future__ import annotations
4
+
5
+import logging
6
+
7
+from .config import GpioConfig
8
+
9
+logger = logging.getLogger("hot_tub.control")
10
+
11
+
12
+class RelayOutputs:
13
+ """Thin wrapper so --dry-run can skip real GPIO."""
14
+
15
+ def __init__(self, gpio: GpioConfig, *, dry_run: bool) -> None:
16
+ self.dry_run = dry_run
17
+ self.heat_on = False
18
+ self.filter_on = False
19
+ self._heat = None
20
+ self._filter = None
21
+
22
+ if dry_run:
23
+ return
24
+
25
+ try:
26
+ from gpiozero import OutputDevice
27
+ except ImportError as exc:
28
+ raise SystemExit(
29
+ "gpiozero is required on the Pi. Install with:\n"
30
+ " sudo apt install python3-gpiozero\n"
31
+ "Or run with --dry-run to exercise the loop without GPIO."
32
+ ) from exc
33
+
34
+ self._heat = OutputDevice(
35
+ gpio.heat_pump,
36
+ active_high=gpio.active_high,
37
+ initial_value=False,
38
+ )
39
+ self._filter = OutputDevice(
40
+ gpio.filter_pump,
41
+ active_high=gpio.active_high,
42
+ initial_value=False,
43
+ )
44
+ logger.info(
45
+ "GPIO initialized heat=%s filter=%s active_high=%s",
46
+ gpio.heat_pump,
47
+ gpio.filter_pump,
48
+ gpio.active_high,
49
+ )
50
+
51
+ def set_heat(self, on: bool) -> None:
52
+ if self.heat_on == on:
53
+ return
54
+ self.heat_on = on
55
+ if self._heat is not None:
56
+ self._heat.on() if on else self._heat.off()
57
+ logger.info("heat relay %s", "ON" if on else "OFF")
58
+
59
+ def set_filter(self, on: bool) -> None:
60
+ if self.filter_on == on:
61
+ return
62
+ self.filter_on = on
63
+ if self._filter is not None:
64
+ self._filter.on() if on else self._filter.off()
65
+ logger.info("filter relay %s", "ON" if on else "OFF")
66
+
67
+ def sync(self, *, heat: bool, filter_pump: bool) -> None:
68
+ self.set_heat(heat)
69
+ self.set_filter(filter_pump)
70
+
71
+ def all_off(self) -> None:
72
+ self.set_heat(False)
73
+ self.set_filter(False)
74
+
75
+ def close(self) -> None:
76
+ self.all_off()
77
+ if self._heat is not None:
78
+ self._heat.close()
79
+ if self._filter is not None:
80
+ self._filter.close()
projects/hot-tub/src/control/heat_fsm.py
... ...
@@ -0,0 +1,208 @@
1
+"""Heat mode FSM: sleep, maintain (startup + heating), backflush."""
2
+
3
+from __future__ import annotations
4
+
5
+import logging
6
+import time
7
+from typing import Any
8
+
9
+from .config import ControlConfig
10
+from .safety import apply_safety_trip, check_heat_allowed
11
+from .state import ControlState, FsmContext, HeatPhase, PumpState
12
+
13
+logger = logging.getLogger("hot_tub.control")
14
+
15
+STARTUP_PHASES: tuple[HeatPhase, ...] = ("startup_1", "startup_2", "startup_3")
16
+
17
+
18
+def _attempt_to_phase(attempt: int) -> HeatPhase:
19
+ if attempt <= 1:
20
+ return "startup_1"
21
+ if attempt == 2:
22
+ return "startup_2"
23
+ return "startup_3"
24
+
25
+
26
+def _mono() -> float:
27
+ return time.monotonic()
28
+
29
+
30
+def _enter_phase(state: ControlState, ctx: FsmContext, phase: HeatPhase) -> None:
31
+ if state.heat_phase != phase:
32
+ logger.info("heat_phase %s -> %s", state.heat_phase, phase)
33
+ state.heat_phase = phase
34
+ ctx.phase_entered_mono = _mono()
35
+ ctx.wait_until_mono = 0.0
36
+
37
+
38
+def _maintain_needs_heat(state: ControlState, cfg: ControlConfig) -> bool:
39
+ if state.tub_avg_f is None:
40
+ return False
41
+ low = state.target_f - cfg.maintain.band_f
42
+ if state.tub_avg_f <= low:
43
+ return True
44
+ if state.tub_avg_f >= state.target_f:
45
+ return False
46
+ # Between band edges: keep heating if already in heating or startup; else idle
47
+ if state.heat_phase in ("heating", *STARTUP_PHASES):
48
+ return True
49
+ return False
50
+
51
+
52
+def _startup_check_passed(state: ControlState, cfg: ControlConfig) -> bool:
53
+ if state.tub_avg_f is None or state.hot_in_f is None:
54
+ return False
55
+ return state.hot_in_f >= state.tub_avg_f + cfg.maintain.hot_in_rise_f
56
+
57
+
58
+def _begin_startup(state: ControlState, ctx: FsmContext, cfg: ControlConfig) -> None:
59
+ if state.startup_attempt <= 0:
60
+ state.startup_attempt = 1
61
+ state.heat_pump = "on"
62
+ _enter_phase(state, ctx, _attempt_to_phase(state.startup_attempt))
63
+ ctx.wait_until_mono = ctx.phase_entered_mono + cfg.maintain.startup_delay_s
64
+
65
+
66
+def _fail_startup_attempt(
67
+ state: ControlState,
68
+ ctx: FsmContext,
69
+ cfg: ControlConfig,
70
+) -> None:
71
+ state.heat_pump = "off"
72
+ attempt = state.startup_attempt
73
+ if attempt >= cfg.maintain.startup_attempts:
74
+ _enter_phase(state, ctx, "error_startup")
75
+ state.error = {
76
+ "code": "startup_failed",
77
+ "detail": "Three startup attempts failed — check propane or run backflush",
78
+ }
79
+ logger.warning("startup failed after %s attempts", attempt)
80
+ return
81
+ state.startup_attempt = attempt + 1
82
+ _enter_phase(state, ctx, "idle")
83
+ ctx.wait_until_mono = _mono() + cfg.maintain.retry_off_s
84
+ logger.info("startup attempt %s failed; retry after %.0fs", attempt, cfg.maintain.retry_off_s)
85
+
86
+
87
+def tick_heat(
88
+ state: ControlState,
89
+ ctx: FsmContext,
90
+ cfg: ControlConfig,
91
+ reading: dict[str, Any] | None,
92
+) -> None:
93
+ """Advance heat FSM one tick; may set heat_pump and heat_phase."""
94
+ now_mono = _mono()
95
+
96
+ if state.heat_mode == "sleep":
97
+ if state.heat_pump != "off":
98
+ state.heat_pump = "off"
99
+ if state.heat_phase not in ("idle", "error_startup"):
100
+ _enter_phase(state, ctx, "idle")
101
+ if state.heat_phase != "error_startup":
102
+ state.startup_attempt = 0
103
+ if state.error and state.error.get("code") == "startup_failed":
104
+ state.error = None
105
+ return
106
+
107
+ if state.heat_mode == "backflush":
108
+ safety = check_heat_allowed(
109
+ state,
110
+ cfg,
111
+ reading,
112
+ require_hot_in=False,
113
+ )
114
+ if not safety.allow_heat and state.heat_phase == "backflush":
115
+ if safety.trip:
116
+ apply_safety_trip(state, safety.reason or "safety")
117
+ state.heat_pump = "off"
118
+ return
119
+
120
+ if state.heat_phase != "backflush":
121
+ _enter_phase(state, ctx, "backflush")
122
+ state.heat_pump = "on"
123
+ ctx.wait_until_mono = ctx.phase_entered_mono + cfg.backflush.duration_s
124
+ return
125
+
126
+ if now_mono >= ctx.wait_until_mono:
127
+ state.heat_pump = "off"
128
+ _enter_phase(state, ctx, "idle")
129
+ revert = cfg.backflush.revert_to_heat_mode
130
+ if revert in ("sleep", "maintain", "backflush"):
131
+ logger.info("backflush complete; heat_mode -> %s", revert)
132
+ state.heat_mode = revert # type: ignore[assignment]
133
+ else:
134
+ state.heat_mode = "sleep"
135
+ return
136
+
137
+ # maintain
138
+ if state.heat_phase == "error_startup":
139
+ state.heat_pump = "off"
140
+ return
141
+
142
+ if state.heat_phase == "error_safety":
143
+ safety = check_heat_allowed(state, cfg, reading, require_hot_in=False)
144
+ if safety.allow_heat:
145
+ state.error = None
146
+ _enter_phase(state, ctx, "idle")
147
+ else:
148
+ state.heat_pump = "off"
149
+ return
150
+
151
+ safety = check_heat_allowed(
152
+ state,
153
+ cfg,
154
+ reading,
155
+ require_hot_in=state.heat_phase in STARTUP_PHASES and now_mono >= ctx.wait_until_mono,
156
+ )
157
+ if not safety.allow_heat:
158
+ if state.heat_pump == "on" and safety.trip:
159
+ apply_safety_trip(state, safety.reason or "safety")
160
+ elif state.heat_pump == "on":
161
+ state.heat_pump = "off"
162
+ _enter_phase(state, ctx, "idle")
163
+ return
164
+
165
+ needs = _maintain_needs_heat(state, cfg)
166
+
167
+ if not needs:
168
+ if state.heat_pump != "off":
169
+ state.heat_pump = "off"
170
+ if state.heat_phase not in ("idle",):
171
+ _enter_phase(state, ctx, "idle")
172
+ state.startup_attempt = 0
173
+ return
174
+
175
+ # Need heat
176
+ if state.heat_pump == "off" and state.heat_phase == "idle":
177
+ if ctx.wait_until_mono > now_mono:
178
+ return
179
+ _begin_startup(state, ctx, cfg)
180
+ return
181
+
182
+ if state.heat_phase in STARTUP_PHASES:
183
+ if now_mono < ctx.wait_until_mono:
184
+ return
185
+ if _startup_check_passed(state, cfg):
186
+ logger.info(
187
+ "startup success (attempt %s): hot_in=%.1f tub_avg=%.1f",
188
+ state.startup_attempt,
189
+ state.hot_in_f or 0,
190
+ state.tub_avg_f or 0,
191
+ )
192
+ _enter_phase(state, ctx, "heating")
193
+ state.heat_pump = "on"
194
+ state.startup_attempt = 0
195
+ else:
196
+ _fail_startup_attempt(state, ctx, cfg)
197
+ return
198
+
199
+ if state.heat_phase == "heating":
200
+ state.heat_pump = "on"
201
+ if state.tub_avg_f is not None and state.tub_avg_f >= state.target_f:
202
+ state.heat_pump = "off"
203
+ _enter_phase(state, ctx, "idle")
204
+ return
205
+
206
+ # Pump on but phase idle — start startup
207
+ if state.heat_pump == "off":
208
+ _begin_startup(state, ctx, cfg)
projects/hot-tub/src/control/logging_setup.py
... ...
@@ -0,0 +1,20 @@
1
+"""File logging for the control daemon."""
2
+
3
+from __future__ import annotations
4
+
5
+import logging
6
+from pathlib import Path
7
+
8
+
9
+def setup_logging(log_path: Path, *, verbose: bool = False) -> logging.Logger:
10
+ log_path.parent.mkdir(parents=True, exist_ok=True)
11
+ logger = logging.getLogger("hot_tub.control")
12
+ logger.setLevel(logging.DEBUG if verbose else logging.INFO)
13
+ logger.handlers.clear()
14
+
15
+ handler = logging.FileHandler(log_path, encoding="utf-8")
16
+ handler.setFormatter(
17
+ logging.Formatter("%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
18
+ )
19
+ logger.addHandler(handler)
20
+ return logger
projects/hot-tub/src/control/redis_io.py
... ...
@@ -0,0 +1,134 @@
1
+"""Redis subscribe/publish for control daemon."""
2
+
3
+from __future__ import annotations
4
+
5
+import json
6
+import logging
7
+import time
8
+from typing import Any
9
+
10
+try:
11
+ import redis
12
+except ImportError:
13
+ redis = None # type: ignore[assignment]
14
+
15
+logger = logging.getLogger("hot_tub.control")
16
+
17
+KEY_CONTROL_LATEST = "control:latest"
18
+CHAN_CONTROL_STATE = "control.state"
19
+KEY_HEALTH = "health:control"
20
+CHAN_HEALTH = "health.control"
21
+KEY_TEMPS_LATEST = "temps:latest"
22
+CHAN_COMMAND = "control.command"
23
+CHAN_TEMPS_READING = "temps.reading"
24
+
25
+
26
+class ControlRedis:
27
+ def __init__(self, redis_url: str) -> None:
28
+ if redis is None:
29
+ raise RuntimeError("redis package not installed (apt: python3-redis)")
30
+ self._url = redis_url
31
+ self._client: redis.Redis | None = None
32
+ self._pubsub: redis.client.PubSub | None = None
33
+ self._backoff_s = 1.0
34
+
35
+ def _connect(self) -> redis.Redis:
36
+ if self._client is None:
37
+ self._client = redis.from_url(self._url, decode_responses=True)
38
+ return self._client
39
+
40
+ def _ensure(self) -> redis.Redis | None:
41
+ try:
42
+ client = self._connect()
43
+ client.ping()
44
+ self._backoff_s = 1.0
45
+ return client
46
+ except Exception as exc:
47
+ logger.warning("Redis unavailable: %s (retry in %.0fs)", exc, self._backoff_s)
48
+ self._client = None
49
+ self._pubsub = None
50
+ time.sleep(min(self._backoff_s, 30.0))
51
+ self._backoff_s = min(self._backoff_s * 2, 30.0)
52
+ return None
53
+
54
+ def get_temps_latest(self) -> dict[str, Any] | None:
55
+ client = self._ensure()
56
+ if client is None:
57
+ return None
58
+ try:
59
+ raw = client.get(KEY_TEMPS_LATEST)
60
+ if not raw:
61
+ return None
62
+ data = json.loads(raw)
63
+ return data if isinstance(data, dict) else None
64
+ except Exception as exc:
65
+ logger.warning("GET temps:latest failed: %s", exc)
66
+ return None
67
+
68
+ def subscribe_commands_and_temps(self) -> redis.client.PubSub | None:
69
+ client = self._ensure()
70
+ if client is None:
71
+ return None
72
+ try:
73
+ if self._pubsub is None:
74
+ self._pubsub = client.pubsub(ignore_subscribe_messages=True)
75
+ self._pubsub.subscribe(CHAN_COMMAND, CHAN_TEMPS_READING)
76
+ return self._pubsub
77
+ except Exception as exc:
78
+ logger.warning("Redis subscribe failed: %s", exc)
79
+ self._pubsub = None
80
+ self._client = None
81
+ return None
82
+
83
+ def poll_messages(self, timeout_s: float = 0.5) -> list[tuple[str, str]]:
84
+ pubsub = self.subscribe_commands_and_temps()
85
+ if pubsub is None:
86
+ return []
87
+ out: list[tuple[str, str]] = []
88
+ deadline = time.monotonic() + timeout_s
89
+ while time.monotonic() < deadline:
90
+ msg = pubsub.get_message(timeout=0.1)
91
+ if msg is None:
92
+ continue
93
+ if msg.get("type") != "message":
94
+ continue
95
+ channel = msg.get("channel")
96
+ data = msg.get("data")
97
+ if isinstance(channel, str) and isinstance(data, str):
98
+ out.append((channel, data))
99
+ if len(out) >= 32:
100
+ break
101
+ return out
102
+
103
+ def publish_state(self, payload: dict[str, Any]) -> bool:
104
+ client = self._ensure()
105
+ if client is None:
106
+ return False
107
+ body = json.dumps(payload)
108
+ client.set(KEY_CONTROL_LATEST, body)
109
+ client.publish(CHAN_CONTROL_STATE, body)
110
+ return True
111
+
112
+ def publish_health(self, payload: dict[str, Any]) -> bool:
113
+ client = self._ensure()
114
+ if client is None:
115
+ return False
116
+ body = json.dumps(payload)
117
+ client.set(KEY_HEALTH, body)
118
+ client.publish(CHAN_HEALTH, body)
119
+ return True
120
+
121
+ def close(self) -> None:
122
+ if self._pubsub is not None:
123
+ try:
124
+ self._pubsub.unsubscribe()
125
+ self._pubsub.close()
126
+ except Exception:
127
+ pass
128
+ self._pubsub = None
129
+ if self._client is not None:
130
+ try:
131
+ self._client.close()
132
+ except Exception:
133
+ pass
134
+ self._client = None
projects/hot-tub/src/control/safety.py
... ...
@@ -0,0 +1,85 @@
1
+"""Safety interlocks before energizing heat."""
2
+
3
+from __future__ import annotations
4
+
5
+from dataclasses import dataclass
6
+from datetime import datetime, timezone
7
+from typing import Any
8
+
9
+from .config import ControlConfig
10
+from .state import ControlState
11
+
12
+
13
+@dataclass
14
+class SafetyResult:
15
+ allow_heat: bool
16
+ reason: str | None = None
17
+ trip: bool = False
18
+
19
+
20
+def _parse_ts(ts: str | None) -> datetime | None:
21
+ if not ts:
22
+ return None
23
+ try:
24
+ if ts.endswith("Z"):
25
+ return datetime.fromisoformat(ts.replace("Z", "+00:00"))
26
+ return datetime.fromisoformat(ts)
27
+ except ValueError:
28
+ return None
29
+
30
+
31
+def temps_fresh(reading: dict[str, Any] | None, cfg: ControlConfig, now: datetime | None = None) -> bool:
32
+ if reading is None:
33
+ return False
34
+ ts = _parse_ts(reading.get("ts"))
35
+ if ts is None:
36
+ return False
37
+ if ts.tzinfo is None:
38
+ ts = ts.replace(tzinfo=timezone.utc)
39
+ now = now or datetime.now(timezone.utc)
40
+ age = (now - ts.astimezone(timezone.utc)).total_seconds()
41
+ return age <= cfg.temps_stale_s
42
+
43
+
44
+def check_heat_allowed(
45
+ state: ControlState,
46
+ cfg: ControlConfig,
47
+ reading: dict[str, Any] | None,
48
+ *,
49
+ require_hot_in: bool = False,
50
+ now: datetime | None = None,
51
+) -> SafetyResult:
52
+ """Evaluate whether heat pump may be turned on."""
53
+ # TODO: GPIO17 low-water float when hardware is added.
54
+
55
+ if state.heat_phase == "error_startup":
56
+ return SafetyResult(False, "latched error_startup")
57
+
58
+ if reading is None or not temps_fresh(reading, cfg, now=now):
59
+ return SafetyResult(False, "stale or missing temps", trip=state.heat_pump == "on")
60
+
61
+ setup = reading.get("setup_state")
62
+ if setup == "need_reinstall":
63
+ return SafetyResult(False, "need_reinstall", trip=state.heat_pump == "on")
64
+
65
+ tub = reading.get("tub_avg_f")
66
+ if tub is None:
67
+ return SafetyResult(False, "no tub_avg_f", trip=state.heat_pump == "on")
68
+
69
+ state.tub_avg_f = float(tub)
70
+ hot = reading.get("hot_in_f")
71
+ state.hot_in_f = float(hot) if hot is not None else None
72
+
73
+ if state.tub_avg_f >= state.target_f:
74
+ return SafetyResult(False, "at or above target")
75
+
76
+ if require_hot_in and state.hot_in_f is None:
77
+ return SafetyResult(False, "hot_in required for startup check")
78
+
79
+ return SafetyResult(True)
80
+
81
+
82
+def apply_safety_trip(state: ControlState, reason: str) -> None:
83
+ state.heat_pump = "off"
84
+ state.heat_phase = "error_safety"
85
+ state.error = {"code": "safety", "detail": reason}
projects/hot-tub/src/control/state.py
... ...
@@ -0,0 +1,130 @@
1
+"""In-memory control state and persistence."""
2
+
3
+from __future__ import annotations
4
+
5
+import json
6
+import logging
7
+import os
8
+import tempfile
9
+from dataclasses import dataclass
10
+from pathlib import Path
11
+from typing import Any, Literal
12
+
13
+logger = logging.getLogger("hot_tub.control")
14
+
15
+HeatMode = Literal["sleep", "maintain", "backflush"]
16
+FilterMode = Literal["on", "auto", "off"]
17
+HeatPhase = Literal[
18
+ "idle",
19
+ "startup_1",
20
+ "startup_2",
21
+ "startup_3",
22
+ "heating",
23
+ "backflush",
24
+ "error_startup",
25
+ "error_safety",
26
+]
27
+PumpState = Literal["on", "off"]
28
+
29
+HEAT_MODES: frozenset[str] = frozenset({"sleep", "maintain", "backflush"})
30
+FILTER_MODES: frozenset[str] = frozenset({"on", "auto", "off"})
31
+
32
+
33
+@dataclass
34
+class ControlState:
35
+ heat_mode: HeatMode = "sleep"
36
+ filter_mode: FilterMode = "off"
37
+ target_f: float = 102.0
38
+ heat_phase: HeatPhase = "idle"
39
+ startup_attempt: int = 0
40
+ heat_pump: PumpState = "off"
41
+ filter_pump: PumpState = "off"
42
+ error: dict[str, Any] | None = None
43
+ last_command_id: str | None = None
44
+ tub_avg_f: float | None = None
45
+ hot_in_f: float | None = None
46
+ filter_window_active: bool = False
47
+ filter_next_start: str | None = None
48
+
49
+ def to_persist(self) -> dict[str, Any]:
50
+ return {
51
+ "heat_mode": self.heat_mode,
52
+ "filter_mode": self.filter_mode,
53
+ "target_f": self.target_f,
54
+ "error": self.error,
55
+ "heat_phase": self.heat_phase if self.heat_phase == "error_startup" else "idle",
56
+ }
57
+
58
+ @classmethod
59
+ def from_persist(cls, data: dict[str, Any], *, default_target: float) -> ControlState:
60
+ st = cls(target_f=default_target)
61
+ if "heat_mode" in data and data["heat_mode"] in HEAT_MODES:
62
+ st.heat_mode = data["heat_mode"] # type: ignore[assignment]
63
+ if "filter_mode" in data and data["filter_mode"] in FILTER_MODES:
64
+ st.filter_mode = data["filter_mode"] # type: ignore[assignment]
65
+ if "target_f" in data:
66
+ st.target_f = float(data["target_f"])
67
+ if data.get("error"):
68
+ st.error = dict(data["error"])
69
+ phase = data.get("heat_phase", "idle")
70
+ if phase == "error_startup" and st.error:
71
+ st.heat_phase = "error_startup"
72
+ code = st.error.get("code")
73
+ if code == "startup_failed":
74
+ st.startup_attempt = 3
75
+ return st
76
+
77
+
78
+@dataclass
79
+class FsmContext:
80
+ """Transient FSM timing (not persisted)."""
81
+
82
+ phase_entered_mono: float = 0.0
83
+ wait_until_mono: float = 0.0
84
+ auto_run_started_mono: float | None = None
85
+
86
+
87
+def load_persisted_state(path: Path, *, default_target: float) -> ControlState:
88
+ if not path.exists():
89
+ return ControlState(target_f=default_target)
90
+ try:
91
+ data = json.loads(path.read_text())
92
+ return ControlState.from_persist(data, default_target=default_target)
93
+ except (json.JSONDecodeError, OSError, TypeError, ValueError) as exc:
94
+ logger.warning("Could not load state from %s: %s", path, exc)
95
+ return ControlState(target_f=default_target)
96
+
97
+
98
+def save_persisted_state(path: Path, state: ControlState) -> None:
99
+ path.parent.mkdir(parents=True, exist_ok=True)
100
+ body = json.dumps(state.to_persist(), indent=2) + "\n"
101
+ fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".control-state-", suffix=".tmp")
102
+ try:
103
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
104
+ f.write(body)
105
+ os.replace(tmp, path)
106
+ except OSError:
107
+ try:
108
+ os.unlink(tmp)
109
+ except OSError:
110
+ pass
111
+ raise
112
+
113
+
114
+def state_snapshot(state: ControlState, *, ts: str) -> dict[str, Any]:
115
+ return {
116
+ "ts": ts,
117
+ "heat_mode": state.heat_mode,
118
+ "filter_mode": state.filter_mode,
119
+ "target_f": state.target_f,
120
+ "tub_avg_f": state.tub_avg_f,
121
+ "hot_in_f": state.hot_in_f,
122
+ "heat_pump": state.heat_pump,
123
+ "filter_pump": state.filter_pump,
124
+ "heat_phase": state.heat_phase,
125
+ "startup_attempt": state.startup_attempt,
126
+ "error": state.error,
127
+ "last_command_id": state.last_command_id,
128
+ "filter_window_active": state.filter_window_active,
129
+ "filter_next_start": state.filter_next_start,
130
+ }
projects/hot-tub/src/control/tests/__init__.py
projects/hot-tub/src/control/tests/test_commands.py
... ...
@@ -0,0 +1,50 @@
1
+"""Tests for command parsing and application."""
2
+
3
+from __future__ import annotations
4
+
5
+import unittest
6
+
7
+from control.commands import CommandError, apply_command
8
+from control.config import ControlConfig
9
+from control.state import ControlState
10
+
11
+
12
+class TestCommands(unittest.TestCase):
13
+ def setUp(self) -> None:
14
+ self.cfg = ControlConfig()
15
+ self.state = ControlState(target_f=100.0)
16
+
17
+ def test_set_heat_mode_sleep_clears_startup_error(self) -> None:
18
+ self.state.heat_phase = "error_startup"
19
+ self.state.error = {"code": "startup_failed", "detail": "x"}
20
+ apply_command(
21
+ self.state,
22
+ self.cfg,
23
+ {"op": "set_heat_mode", "args": {"mode": "sleep"}, "id": "1"},
24
+ )
25
+ self.assertEqual(self.state.heat_mode, "sleep")
26
+ self.assertIsNone(self.state.error)
27
+ self.assertEqual(self.state.heat_phase, "idle")
28
+
29
+ def test_target_clamped(self) -> None:
30
+ self.cfg.target_f_max = 110
31
+ apply_command(
32
+ self.state,
33
+ self.cfg,
34
+ {"op": "set_target_temp", "args": {"target_f": 200}},
35
+ )
36
+ self.assertEqual(self.state.target_f, 110)
37
+
38
+ def test_increment_decrement(self) -> None:
39
+ apply_command(self.state, self.cfg, {"op": "target_temp_increment", "args": {}})
40
+ self.assertEqual(self.state.target_f, 101)
41
+ apply_command(self.state, self.cfg, {"op": "target_temp_decrement", "args": {"delta_f": 2}})
42
+ self.assertEqual(self.state.target_f, 99)
43
+
44
+ def test_invalid_op(self) -> None:
45
+ with self.assertRaises(CommandError):
46
+ apply_command(self.state, self.cfg, {"op": "nope"})
47
+
48
+
49
+if __name__ == "__main__":
50
+ unittest.main()
projects/hot-tub/src/control/tests/test_filter_policy.py
... ...
@@ -0,0 +1,39 @@
1
+"""Tests for filter auto schedule."""
2
+
3
+from __future__ import annotations
4
+
5
+import unittest
6
+from datetime import datetime
7
+from zoneinfo import ZoneInfo
8
+
9
+from control.config import ControlConfig, FilterAutoConfig
10
+from control.filter_policy import evaluate_filter_window, desired_filter_pump
11
+from control.state import ControlState, FsmContext
12
+
13
+
14
+class TestFilterPolicy(unittest.TestCase):
15
+ def test_midnight_two_hour_window(self) -> None:
16
+ cfg = ControlConfig(
17
+ filter_auto=FilterAutoConfig(start_time="00:00", run_time_hours=2.0, timezone="UTC")
18
+ )
19
+ tz = ZoneInfo("UTC")
20
+ inside = datetime(2026, 10, 19, 1, 0, tzinfo=tz)
21
+ outside = datetime(2026, 10, 19, 3, 0, tzinfo=tz)
22
+ self.assertTrue(evaluate_filter_window(cfg, inside).active)
23
+ self.assertFalse(evaluate_filter_window(cfg, outside).active)
24
+
25
+ def test_auto_mode_pump_on_in_window(self) -> None:
26
+ cfg = ControlConfig(
27
+ filter_auto=FilterAutoConfig(start_time="00:00", run_time_hours=2.0, timezone="UTC")
28
+ )
29
+ state = ControlState(filter_mode="auto")
30
+ ctx = FsmContext()
31
+ tz = ZoneInfo("UTC")
32
+ now = datetime(2026, 10, 19, 1, 0, tzinfo=tz)
33
+ pump = desired_filter_pump(state, cfg, ctx, now=now)
34
+ self.assertEqual(pump, "on")
35
+ self.assertTrue(state.filter_window_active)
36
+
37
+
38
+if __name__ == "__main__":
39
+ unittest.main()
projects/hot-tub/src/control/tests/test_heat_fsm.py
... ...
@@ -0,0 +1,68 @@
1
+"""Tests for heat FSM transitions."""
2
+
3
+from __future__ import annotations
4
+
5
+import unittest
6
+from datetime import datetime, timezone
7
+
8
+from control.config import ControlConfig, MaintainConfig, BackflushConfig
9
+from control.heat_fsm import tick_heat
10
+from control.state import ControlState, FsmContext
11
+
12
+
13
+def _reading(tub: float, hot: float | None = 120.0) -> dict:
14
+ return {
15
+ "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
16
+ "setup_state": "ready",
17
+ "tub_avg_f": tub,
18
+ "hot_in_f": hot,
19
+ }
20
+
21
+
22
+class TestHeatFsm(unittest.TestCase):
23
+ def setUp(self) -> None:
24
+ self.cfg = ControlConfig(
25
+ maintain=MaintainConfig(
26
+ band_f=5.0,
27
+ startup_delay_s=0.0,
28
+ hot_in_rise_f=10.0,
29
+ startup_attempts=3,
30
+ retry_off_s=0.0,
31
+ ),
32
+ backflush=BackflushConfig(duration_s=0.0, revert_to_heat_mode="sleep"),
33
+ )
34
+
35
+ def test_maintain_startup_success_then_heating(self) -> None:
36
+ state = ControlState(heat_mode="maintain", target_f=102.0)
37
+ ctx = FsmContext()
38
+ reading = _reading(90.0, hot=105.0)
39
+ tick_heat(state, ctx, self.cfg, reading)
40
+ self.assertEqual(state.heat_phase, "startup_1")
41
+ self.assertEqual(state.heat_pump, "on")
42
+ tick_heat(state, ctx, self.cfg, reading)
43
+ self.assertEqual(state.heat_phase, "heating")
44
+
45
+ def test_three_failures_error_startup(self) -> None:
46
+ state = ControlState(heat_mode="maintain", target_f=102.0)
47
+ ctx = FsmContext()
48
+ reading = _reading(90.0, hot=90.0)
49
+ for _ in range(24):
50
+ tick_heat(state, ctx, self.cfg, reading)
51
+ if state.heat_phase == "error_startup":
52
+ break
53
+ self.assertEqual(state.heat_phase, "error_startup")
54
+ self.assertEqual(state.heat_pump, "off")
55
+ self.assertIsNotNone(state.error)
56
+
57
+ def test_backflush_reverts_to_sleep(self) -> None:
58
+ state = ControlState(heat_mode="backflush")
59
+ ctx = FsmContext()
60
+ tick_heat(state, ctx, self.cfg, _reading(90.0))
61
+ self.assertEqual(state.heat_phase, "backflush")
62
+ tick_heat(state, ctx, self.cfg, _reading(90.0))
63
+ self.assertEqual(state.heat_mode, "sleep")
64
+ self.assertEqual(state.heat_pump, "off")
65
+
66
+
67
+if __name__ == "__main__":
68
+ unittest.main()
projects/hot-tub/src/control/tests/test_safety.py
... ...
@@ -0,0 +1,49 @@
1
+"""Tests for safety interlocks."""
2
+
3
+from __future__ import annotations
4
+
5
+import unittest
6
+from datetime import datetime, timezone
7
+
8
+from control.config import ControlConfig
9
+from control.safety import check_heat_allowed, temps_fresh
10
+from control.state import ControlState
11
+
12
+
13
+def _reading(**kwargs: object) -> dict:
14
+ base = {
15
+ "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
16
+ "setup_state": "ready",
17
+ "tub_avg_f": 95.0,
18
+ "hot_in_f": 110.0,
19
+ }
20
+ base.update(kwargs)
21
+ return base
22
+
23
+
24
+class TestSafety(unittest.TestCase):
25
+ def setUp(self) -> None:
26
+ self.cfg = ControlConfig()
27
+ self.state = ControlState(target_f=102.0)
28
+
29
+ def test_stale_temps_block(self) -> None:
30
+ old = {"ts": "2020-01-01T00:00:00Z", "tub_avg_f": 90.0, "setup_state": "ready"}
31
+ self.assertFalse(temps_fresh(old, self.cfg))
32
+ r = check_heat_allowed(self.state, self.cfg, old)
33
+ self.assertFalse(r.allow_heat)
34
+
35
+ def test_need_reinstall_blocks(self) -> None:
36
+ r = check_heat_allowed(self.state, self.cfg, _reading(setup_state="need_reinstall"))
37
+ self.assertFalse(r.allow_heat)
38
+
39
+ def test_at_target_blocks(self) -> None:
40
+ r = check_heat_allowed(self.state, self.cfg, _reading(tub_avg_f=102.0))
41
+ self.assertFalse(r.allow_heat)
42
+
43
+ def test_ok_below_target(self) -> None:
44
+ r = check_heat_allowed(self.state, self.cfg, _reading(tub_avg_f=96.0))
45
+ self.assertTrue(r.allow_heat)
46
+
47
+
48
+if __name__ == "__main__":
49
+ unittest.main()