"""Control daemon main loop.""" from __future__ import annotations import json import logging import os import signal import time from datetime import datetime, timezone from typing import Any from .commands import CommandError, apply_command, parse_command_message from .config import ControlConfig from .filter_policy import desired_filter_pump from .gpio_out import RelayOutputs from .heat_fsm import tick_heat from .redis_io import CHAN_COMMAND, CHAN_TEMPS_READING, ControlRedis from .safety import temps_fresh from .state import ControlState, FsmContext, load_persisted_state, save_persisted_state, state_snapshot from .systemd_notify import notify as sd_notify logger = logging.getLogger("hot_tub.control") _stop = False def _handle_signal(signum: int, _frame: object) -> None: global _stop logger.info("Received signal %s, shutting down", signum) _stop = True def _utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _health_status( state: ControlState, cfg: ControlConfig, reading: dict[str, Any] | None, *, gpio_error: bool = False, publish_failed: bool = False, ) -> tuple[str, str]: if gpio_error: return "error", "gpio failure" if publish_failed: return "error", "redis publish failed" if state.heat_phase.startswith("error_"): return "degraded", state.heat_phase if state.heat_mode == "maintain" and not temps_fresh(reading, cfg): return "degraded", "stale temps while maintain" return "ok", "loop running" def _apply_gpio(relays: RelayOutputs, state: ControlState) -> None: relays.sync( heat=state.heat_pump == "on", filter_pump=state.filter_pump == "on", ) state.heat_pump = "on" if relays.heat_on else "off" state.filter_pump = "on" if relays.filter_on else "off" class ControlEngine: def __init__( self, cfg: ControlConfig, *, dry_run: bool = False, gpio_dry_run: bool = False, ) -> None: self.cfg = cfg self.dry_run = dry_run self.state = load_persisted_state(cfg.state_path, default_target=cfg.target_f_default) self.ctx = FsmContext() self.reading: dict[str, Any] | None = None self.gpio_error = False self.relays = RelayOutputs(cfg.gpio, dry_run=dry_run or gpio_dry_run) self.relays.all_off() self._last_persist_hash: str | None = None self._last_publish_mono = 0.0 self._last_snapshot_json: str | None = None def close(self) -> None: self.relays.all_off() self.relays.close() def _maybe_persist(self) -> None: body = str(self.state.to_persist()) if body == self._last_persist_hash: return if not self.dry_run: save_persisted_state(self.cfg.state_path, self.state) self._last_persist_hash = body def ingest_temps(self, payload: dict[str, Any]) -> None: self.reading = payload def handle_command_raw(self, raw: str) -> None: try: envelope = parse_command_message(raw) changed = apply_command(self.state, self.cfg, envelope) if changed: self._maybe_persist() except CommandError as exc: logger.warning("command rejected: %s", exc) except Exception: logger.exception("command handling failed") def tick(self) -> dict[str, Any]: if self.gpio_error: self.state.heat_pump = "off" self.state.filter_pump = "off" return state_snapshot(self.state, ts=_utc_now()) try: tick_heat(self.state, self.ctx, self.cfg, self.reading) filt = desired_filter_pump(self.state, self.cfg, self.ctx) self.state.filter_pump = filt _apply_gpio(self.relays, self.state) except Exception: logger.exception("tick failed (GPIO or FSM)") self.gpio_error = True self.relays.all_off() self.state.heat_pump = "off" self.state.filter_pump = "off" return state_snapshot(self.state, ts=_utc_now()) def should_publish(self, snapshot: dict[str, Any], now_mono: float) -> bool: body = str(sorted(snapshot.items())) if body != self._last_snapshot_json: return True return (now_mono - self._last_publish_mono) >= self.cfg.heartbeat_s def mark_published(self, snapshot: dict[str, Any], now_mono: float) -> None: self._last_snapshot_json = str(sorted(snapshot.items())) self._last_publish_mono = now_mono def _ingest_redis_messages(engine: ControlEngine, msgs: list[tuple[str, str]]) -> None: for channel, data in msgs: if channel == CHAN_COMMAND: engine.handle_command_raw(data) elif channel == CHAN_TEMPS_READING: try: payload = json.loads(data) if isinstance(payload, dict): engine.ingest_temps(payload) except json.JSONDecodeError: logger.warning("invalid temps.reading JSON") def run_once( cfg: ControlConfig, *, dry_run: bool = False, gpio_dry_run: bool = False, engine: ControlEngine | None = None, redis_io: ControlRedis | None = None, ) -> dict[str, Any]: eng = engine or ControlEngine(cfg, dry_run=dry_run, gpio_dry_run=gpio_dry_run) if eng.reading is None and not dry_run: r = redis_io or ControlRedis(cfg.redis_url) try: latest = r.get_temps_latest() if latest: eng.ingest_temps(latest) finally: if redis_io is None: r.close() snapshot = eng.tick() return snapshot def run_loop( cfg: ControlConfig, *, dry_run: bool = False, gpio_dry_run: bool = False, ) -> None: global _stop _stop = False signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGINT, _handle_signal) logger.info("Control daemon starting (dry_run=%s gpio_dry_run=%s)", dry_run, gpio_dry_run) engine = ControlEngine(cfg, dry_run=dry_run, gpio_dry_run=gpio_dry_run) redis_io: ControlRedis | None = None if not dry_run: redis_io = ControlRedis(cfg.redis_url) latest = redis_io.get_temps_latest() if latest: engine.ingest_temps(latest) sd_notify("READY=1") next_due = time.monotonic() try: while not _stop: if redis_io is not None: timeout = max(0.0, next_due - time.monotonic()) msgs = redis_io.wait_messages(timeout_s=timeout) _ingest_redis_messages(engine, msgs) else: delay = max(0.0, next_due - time.monotonic()) if delay > 0 and not _stop: time.sleep(delay) snapshot = engine.tick() now_mono = time.monotonic() publish_failed = False if engine.should_publish(snapshot, now_mono): status, detail = _health_status( engine.state, cfg, engine.reading, gpio_error=engine.gpio_error, publish_failed=False, ) health = { "service": "control", "ts": snapshot["ts"], "status": status, "detail": detail, "pid": os.getpid(), } if dry_run: logger.debug("state %s", snapshot) elif redis_io is not None: ok_state = redis_io.publish_state(snapshot) ok_health = redis_io.publish_health(health) if ok_state and ok_health: engine.mark_published(snapshot, now_mono) else: status, detail = _health_status( engine.state, cfg, engine.reading, gpio_error=engine.gpio_error, publish_failed=True, ) health["status"] = status health["detail"] = detail health["ts"] = _utc_now() redis_io.publish_health(health) else: engine.mark_published(snapshot, now_mono) if now_mono >= next_due: next_due = now_mono + cfg.loop_interval_s sd_notify("WATCHDOG=1") finally: if redis_io is not None: redis_io.close() engine.close() logger.info("Control daemon stopped")