#!/usr/bin/env python3 """voice intake watchdog — Hub Fixes Directive 1. Watches the house voice path and alerts Sam through Apprise when intake dies. It NEVER heals anything: no smart plug, no power-cycle, no restart. Notification only. Sam fixes the layer by hand. What it watches --------------- "Voice intake alive" means all three layers are healthy, checked in this order: 1. BROKER the .13 Mosquitto broker (:1883) is reachable. 2. BRIDGE the `voice_bridge` container is running AND subscribed to `voice/audio_stream` (checked through the Docker socket). 3. PUBLISHER the ESP32 is publishing. Since the Directive 3 firmware, the device sends a heartbeat on `voice/audio_stream` every 5 minutes, so 15 minutes of topic silence = three missed heartbeats. A retained `voice/status == "offline"` Last Will is a second, immediate signal. The layer attribution rule from the spec: if something else broke, the ESP32 is NOT blamed. The alert names the layer that actually failed and what to do about it. Repeat policy ------------- When a layer fails, an alert goes out immediately. While the same layer stays down, the alert repeats once per `REPEAT_INTERVAL_SECONDS` (default: one day). When a different layer fails, that is a new alert. When intake recovers, the state is cleared and a single recovery notice is sent. State is persisted so a watchdog restart does not re-spam or silently forget an ongoing outage. Environment (all optional, sane defaults for the lab) ------------------------------------------------------ MQTT_HOST=192.168.20.13 MQTT_PORT=1883 MQTT_USER / MQTT_PASS (optional; the 1883 plane is anonymous) AUDIO_TOPIC=voice/audio_stream STATUS_TOPIC=voice/status SILENCE_TIMEOUT_SECONDS=900 (15 minutes) CHECK_INTERVAL_SECONDS=30 REPEAT_INTERVAL_SECONDS=86400 (one day) APPRISE_NOTIFY_URL=http://192.168.20.35:8210/notify/apprise/ APPRISE_TAG=sam_ntfy (first alert) APPRISE_TAG_REPEAT=sam (daily repeat escalates to all Sam channels) APPRISE_TIMEOUT_SECONDS=20 BRIDGE_CONTAINER=voice_bridge DOCKER_SOCKET=/var/run/docker.sock DOCKER_TIMEOUT_SECONDS=8 STATE_FILE=/var/lib/voice_intake_watchdog/state.json DRY_RUN=0 (1 => print alerts instead of sending them) REQUIRE_PUBLISHER_EVIDENCE=1 (1 => do not treat silence as a publisher failure until a heartbeat or presence has been seen; prevents false alarms before the Directive 3 heartbeat firmware is flashed) LOG_LEVEL=INFO """ from __future__ import annotations import argparse import dataclasses import datetime as dt import http.client import json import logging import os import socket import sys import threading import time import urllib.request from enum import Enum from typing import Callable, Optional LOG = logging.getLogger("voice_intake_watchdog") try: import paho.mqtt.client as mqtt except ImportError: # pragma: no cover - only hit when paho is missing mqtt = None # --------------------------------------------------------------------------- # Layers # --------------------------------------------------------------------------- class Layer(str, Enum): BROKER = "broker" BRIDGE = "bridge" PUBLISHER = "publisher" HEALTHY = "healthy" # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- def _env_int(name: str, default: int) -> int: raw = os.environ.get(name) if raw is None or raw == "": return default return int(raw) def _env_float(name: str, default: float) -> float: raw = os.environ.get(name) if raw is None or raw == "": return default return float(raw) def _env_bool(name: str, default: bool) -> bool: raw = os.environ.get(name) if raw is None or raw == "": return default return raw.strip().lower() in ("1", "true", "yes", "on") @dataclasses.dataclass class Config: mqtt_host: str = "192.168.20.13" mqtt_port: int = 1883 mqtt_user: Optional[str] = None mqtt_pass: Optional[str] = None audio_topic: str = "voice/audio_stream" status_topic: str = "voice/status" silence_timeout: float = 900.0 check_interval: float = 30.0 repeat_interval: float = 86400.0 apprise_notify_url: str = "http://192.168.20.35:8210/notify/apprise/" apprise_tag: str = "sam_ntfy" apprise_tag_repeat: str = "sam" apprise_timeout: float = 20.0 bridge_container: str = "voice_bridge" docker_socket: str = "/var/run/docker.sock" docker_timeout: float = 8.0 state_file: str = "/var/lib/voice_intake_watchdog/state.json" dry_run: bool = False require_publisher_evidence: bool = True @classmethod def from_env(cls) -> "Config": return cls( mqtt_host=os.environ.get("MQTT_HOST", cls.mqtt_host), mqtt_port=_env_int("MQTT_PORT", cls.mqtt_port), mqtt_user=os.environ.get("MQTT_USER") or None, mqtt_pass=os.environ.get("MQTT_PASS") or None, audio_topic=os.environ.get("AUDIO_TOPIC", cls.audio_topic), status_topic=os.environ.get("STATUS_TOPIC", cls.status_topic), silence_timeout=_env_float("SILENCE_TIMEOUT_SECONDS", cls.silence_timeout), check_interval=_env_float("CHECK_INTERVAL_SECONDS", cls.check_interval), repeat_interval=_env_float("REPEAT_INTERVAL_SECONDS", cls.repeat_interval), apprise_notify_url=os.environ.get("APPRISE_NOTIFY_URL", cls.apprise_notify_url), apprise_tag=os.environ.get("APPRISE_TAG", cls.apprise_tag), apprise_tag_repeat=os.environ.get("APPRISE_TAG_REPEAT", cls.apprise_tag_repeat), apprise_timeout=_env_float("APPRISE_TIMEOUT_SECONDS", cls.apprise_timeout), bridge_container=os.environ.get("BRIDGE_CONTAINER", cls.bridge_container), docker_socket=os.environ.get("DOCKER_SOCKET", cls.docker_socket), docker_timeout=_env_float("DOCKER_TIMEOUT_SECONDS", cls.docker_timeout), state_file=os.environ.get("STATE_FILE", cls.state_file), dry_run=_env_bool("DRY_RUN", cls.dry_run), require_publisher_evidence=_env_bool( "REQUIRE_PUBLISHER_EVIDENCE", cls.require_publisher_evidence ), ) # --------------------------------------------------------------------------- # Classification and repeat state machine (pure, unit-testable) # --------------------------------------------------------------------------- @dataclasses.dataclass class Snapshot: now: float broker_reachable: bool bridge_running: bool bridge_subscribed: bool last_audio_at: Optional[float] started_at: float presence: Optional[str] = None # ESP32 retained voice/status, or None def publisher_armed(snap: Snapshot) -> bool: """True once we have evidence that a publisher exists for this topic. The Directive 3 firmware publishes a retained `voice/status` presence (online/offline) and a heartbeat on the audio topic. Until one of those has been seen, silence on `voice/audio_stream` only means "nobody spoke" or "the heartbeat firmware is not flashed yet" — not "the device is dead". Alerting in that state would cry wolf every quiet night, which is the exact failure this project exists to prevent. """ return snap.presence is not None or snap.last_audio_at is not None def classify(snap: Snapshot, silence_timeout: float, require_evidence: bool = True) -> Layer: """Name the layer that failed, in order: broker, bridge, publisher.""" if not snap.broker_reachable: return Layer.BROKER if not (snap.bridge_running and snap.bridge_subscribed): return Layer.BRIDGE if snap.presence == "offline": return Layer.PUBLISHER if require_evidence and not publisher_armed(snap): return Layer.HEALTHY reference = snap.last_audio_at if snap.last_audio_at is not None else snap.started_at if snap.now - reference > silence_timeout: return Layer.PUBLISHER return Layer.HEALTHY @dataclasses.dataclass class AlertState: layer: Optional[str] = None # currently failing layer, or None first_seen_at: Optional[float] = None last_alert_at: Optional[float] = None last_audio_at: Optional[float] = None started_at: Optional[float] = None @dataclasses.dataclass class Alert: kind: str # "alert" | "repeat" | "recovery" layer: str since: Optional[float] = None def advance(state: AlertState, snap: Snapshot, cfg: Config) -> list[Alert]: """Pure state transition. Returns the alerts that must be delivered now.""" layer = classify(snap, cfg.silence_timeout, cfg.require_publisher_evidence) if layer is Layer.HEALTHY: if state.layer is not None: recovered_from = state.layer since = state.first_seen_at state.layer = None state.first_seen_at = None state.last_alert_at = None return [Alert(kind="recovery", layer=recovered_from, since=since)] return [] if state.layer != layer.value: state.layer = layer.value state.first_seen_at = snap.now state.last_alert_at = snap.now return [Alert(kind="alert", layer=layer.value, since=snap.now)] if state.last_alert_at is None or (snap.now - state.last_alert_at) >= cfg.repeat_interval: state.last_alert_at = snap.now return [Alert(kind="repeat", layer=layer.value, since=state.first_seen_at)] return [] # --------------------------------------------------------------------------- # Message text # --------------------------------------------------------------------------- def _hhmm(ts: Optional[float]) -> str: if ts is None: return "??:??" return dt.datetime.fromtimestamp(ts).strftime("%H:%M") def build_message(alert: Alert, cfg: Config) -> tuple[str, str, str]: """Return (title, body, tag) for an alert. Names what failed and what to do.""" silent_since = _hhmm(alert.since) if alert.kind == "recovery": title = "Voice intake recovered" body = ( f"Voice intake recovered at {_hhmm(time.time())}.\n\n" "The watchdog cleared its alert state. It will alert again if intake dies." ) return title, body, cfg.apprise_tag tag = cfg.apprise_tag if alert.kind == "alert" else cfg.apprise_tag_repeat repeat_note = { "alert": "This alert repeats once a day while the fault stays.", "repeat": "Repeat of an unresolved fault. It repeats again in a day.", }[alert.kind] if alert.layer == Layer.BROKER.value: title = "Voice intake DOWN — .13 MQTT broker unreachable" body = ( f"Voice intake is dead: the .13 MQTT broker ({cfg.mqtt_host}:{cfg.mqtt_port}) " f"has been unreachable since {silent_since}.\n\n" "Nothing can reach the voice path. n8n and the ESP32 also use this broker.\n\n" "What to do: check the Mosquitto container on .13 " "(docker ps; docker logs mosquitto).\n" "Do NOT power-cycle the ESP32 — the broker layer failed, not the device.\n\n" f"{repeat_note}" ) elif alert.layer == Layer.BRIDGE.value: title = "Voice intake DOWN — voice_bridge is not running" body = ( f"Voice intake is dead: the broker is up, but `{cfg.bridge_container}` has not " f"been running and subscribed since {silent_since}.\n\n" "The ESP32 may still be publishing, but nothing is consuming the audio.\n\n" "What to do: start the bridge on .13\n" " cd ~/voice_bridge && docker compose up -d\n" "Do NOT power-cycle the ESP32 — the bridge layer failed, not the device.\n\n" f"{repeat_note}" ) else: # publisher title = "Voice intake silent — no publisher (likely the ESP32)" body = ( f"Voice intake silent since {silent_since}: no message on " f"`{cfg.audio_topic}` for {int(cfg.silence_timeout // 60)} minutes. " "The broker and voice_bridge are healthy, so the publisher stopped.\n\n" "Likely the ESP32 — try a power-cycle.\n\n" f"{repeat_note}" ) return title, body, tag # --------------------------------------------------------------------------- # Notifier (Apprise) # --------------------------------------------------------------------------- class AppriseNotifier: def __init__(self, cfg: Config): self.cfg = cfg self.sent: list[dict] = [] def send(self, title: str, body: str, tag: str) -> bool: record = { "title": title, "tag": tag, "at": time.time(), "dry_run": self.cfg.dry_run, } if self.cfg.dry_run: LOG.info("[dry-run] notify tag=%s title=%s", tag, title) record["body"] = body self.sent.append(record) return True payload = json.dumps({"title": title, "body": body, "tag": tag}).encode() request = urllib.request.Request( self.cfg.apprise_notify_url, data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=self.cfg.apprise_timeout) as resp: ok = 200 <= resp.status < 300 record["status"] = resp.status self.sent.append(record) if ok: LOG.info("apprise notify OK HTTP %s tag=%s: %s", resp.status, tag, title) else: LOG.error("apprise returned HTTP %s for %r", resp.status, title) return ok except Exception as exc: # network/auth/HTTP error record["error"] = str(exc) self.sent.append(record) LOG.error("apprise notify failed for %r: %s", title, exc) return False # --------------------------------------------------------------------------- # Broker watcher (paho) # --------------------------------------------------------------------------- class MqttWatcher: """Keeps a subscription on the audio topic and records the last message time.""" def __init__(self, cfg: Config): if mqtt is None: raise RuntimeError("paho-mqtt is not installed") self.cfg = cfg self.lock = threading.Lock() self.last_audio_at: Optional[float] = None self.presence: Optional[str] = None self.connected = False self.client = mqtt.Client( mqtt.CallbackAPIVersion.VERSION2, client_id="voice_intake_watchdog", ) if cfg.mqtt_user: self.client.username_pw_set(cfg.mqtt_user, cfg.mqtt_pass) self.client.on_connect = self._on_connect self.client.on_disconnect = self._on_disconnect self.client.on_message = self._on_message self.client.reconnect_delay_set(min_delay=1, max_delay=15) # callbacks ------------------------------------------------------------- def _on_connect(self, client, userdata, flags, reason_code, properties=None): if getattr(reason_code, "is_failure", False): LOG.warning("broker refused connection: %s", reason_code) with self.lock: self.connected = False return with self.lock: self.connected = True LOG.info( "connected to %s:%s; subscribing %s + %s", self.cfg.mqtt_host, self.cfg.mqtt_port, self.cfg.audio_topic, self.cfg.status_topic, ) client.subscribe([(self.cfg.audio_topic, 0), (self.cfg.status_topic, 0)]) def _on_disconnect(self, client, userdata, flags, reason_code, properties=None): with self.lock: self.connected = False LOG.warning("broker disconnected: %s", reason_code) def _on_message(self, client, userdata, msg): now = time.time() with self.lock: if msg.topic == self.cfg.audio_topic: self.last_audio_at = now elif msg.topic == self.cfg.status_topic: value = msg.payload.decode("utf-8", "replace").strip().lower() self.presence = value # A retained "offline" proves the device's Last Will fired. LOG.info("voice/status presence: %s", value) # lifecycle ------------------------------------------------------------- def start(self): # connect_async + loop_start so a broker outage cannot block startup. try: self.client.connect_async(self.cfg.mqtt_host, self.cfg.mqtt_port, keepalive=30) except Exception as exc: # pragma: no cover - defensive LOG.error("mqtt connect_async failed: %s", exc) self.client.loop_start() def stop(self): self.client.loop_stop() try: self.client.disconnect() except Exception: pass def tcp_probe(host: str, port: int, timeout: float = 3.0) -> bool: try: with socket.create_connection((host, port), timeout=timeout): return True except OSError: return False # --------------------------------------------------------------------------- # Docker bridge check (unix socket, stdlib only) # --------------------------------------------------------------------------- class DockerClient: def __init__(self, socket_path: str, timeout: float): self.socket_path = socket_path self.timeout = timeout def _get(self, path: str) -> tuple[int, bytes]: conn = http.client.HTTPConnection("localhost", timeout=self.timeout) sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) conn.sock = sock try: conn.request("GET", path) resp = conn.getresponse() return resp.status, resp.read() finally: conn.close() def container_json(self, name: str) -> Optional[dict]: try: status, body = self._get(f"/containers/{name}/json") except OSError: return None if status == 404: return {} if status != 200: return None try: return json.loads(body) except json.JSONDecodeError: return None def logs_since(self, name: str, since_epoch: float) -> bytes: try: status, body = self._get( f"/containers/{name}/logs?stdout=1&stderr=1&since={int(since_epoch)}" ) except OSError: return b"" if status != 200: return b"" return body def _parse_docker_time(value: str) -> float: # e.g. 2026-10-14T09:03:11.123456789Z value = value.replace("Z", "+00:00") if "." in value: head, _, tail = value.partition(".") digits = "" for ch in tail: if ch.isdigit(): digits += ch else: break rest = tail[len(digits):] # keep at most 6 fractional digits for fromisoformat value = f"{head}.{digits[:6]}{rest}" try: return dt.datetime.fromisoformat(value).timestamp() except ValueError: return time.time() @dataclasses.dataclass class BridgeState: checked: bool # False when the Docker socket is unavailable running: bool subscribed: bool detail: str def check_bridge(docker: DockerClient, container: str) -> BridgeState: obj = docker.container_json(container) if obj is None: # Docker socket unavailable: do not invent a bridge failure. return BridgeState(False, True, True, "docker socket unavailable; bridge check skipped") if obj == {}: return BridgeState(True, False, False, "container does not exist") state = obj.get("State", {}) running = bool(state.get("Running")) if not running: return BridgeState(True, False, False, f"container not running (status={state.get('Status')})") started = state.get("StartedAt") or "" since = _parse_docker_time(started) if started else time.time() logs = docker.logs_since(container, since) subscribed = b"subscribed voice/audio_stream" in logs detail = "running and subscribed" if subscribed else "running but no subscribe line in logs" return BridgeState(True, running, subscribed, detail) # --------------------------------------------------------------------------- # State persistence # --------------------------------------------------------------------------- def load_state(path: str, started_at: float) -> AlertState: state = AlertState(started_at=started_at) try: with open(path) as handle: raw = json.load(handle) state.layer = raw.get("layer") state.first_seen_at = raw.get("first_seen_at") state.last_alert_at = raw.get("last_alert_at") state.last_audio_at = raw.get("last_audio_at") if raw.get("started_at"): state.started_at = raw["started_at"] except (OSError, json.JSONDecodeError): pass return state def save_state(path: str, state: AlertState): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) tmp = f"{path}.tmp" with open(tmp, "w") as handle: json.dump(dataclasses.asdict(state), handle, indent=2) os.replace(tmp, path) # --------------------------------------------------------------------------- # Main loop # --------------------------------------------------------------------------- def build_snapshot( cfg: Config, watcher: MqttWatcher, docker: DockerClient, started_at: float, now: Optional[float] = None, broker_probe: Callable[[str, int, float], bool] = tcp_probe, ) -> Snapshot: now = now if now is not None else time.time() with watcher.lock: connected = watcher.connected last_audio_at = watcher.last_audio_at presence = watcher.presence broker_reachable = connected or broker_probe(cfg.mqtt_host, cfg.mqtt_port, 3.0) bridge = check_bridge(docker, cfg.bridge_container) return Snapshot( now=now, broker_reachable=broker_reachable, bridge_running=bridge.running, bridge_subscribed=bridge.subscribed, last_audio_at=last_audio_at, started_at=started_at, presence=presence, ) def run(cfg: Config, cycles: Optional[int] = None, notifier: Optional[AppriseNotifier] = None) -> int: started_at = time.time() state = load_state(cfg.state_file, started_at) watcher = MqttWatcher(cfg) if state.last_audio_at is not None: watcher.last_audio_at = state.last_audio_at # survive watchdog restarts watcher.start() docker = DockerClient(cfg.docker_socket, cfg.docker_timeout) notifier = notifier or AppriseNotifier(cfg) LOG.info( "watchdog starting: broker=%s:%s topic=%s silence=%ss repeat=%ss check=%ss dry_run=%s", cfg.mqtt_host, cfg.mqtt_port, cfg.audio_topic, cfg.silence_timeout, cfg.repeat_interval, cfg.check_interval, cfg.dry_run, ) delivered = 0 warned_unarmed = False try: count = 0 while True: snap = build_snapshot(cfg, watcher, docker, started_at) if snap.last_audio_at is not None: state.last_audio_at = snap.last_audio_at if ( cfg.require_publisher_evidence and not warned_unarmed and not publisher_armed(snap) and (snap.now - snap.started_at) > cfg.silence_timeout ): warned_unarmed = True LOG.warning( "no heartbeat or presence on %s/%s yet: silence is NOT " "treated as a publisher failure. Flash the Directive 3 " "firmware (heartbeat + presence) to arm this check.", cfg.audio_topic, cfg.status_topic, ) alerts = advance(state, snap, cfg) for alert in alerts: title, body, tag = build_message(alert, cfg) LOG.warning( "ALERT kind=%s layer=%s -> tag=%s", alert.kind, alert.layer, tag ) notifier.send(title, body, tag) delivered += 1 save_state(cfg.state_file, state) count += 1 if cycles is not None and count >= cycles: break time.sleep(cfg.check_interval) finally: watcher.stop() save_state(cfg.state_file, state) return delivered def main(argv: Optional[list[str]] = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--cycles", type=int, default=None, help="run N checks then exit") parser.add_argument("--once", action="store_true", help="run one check then exit") parser.add_argument("--dry-run", action="store_true", help="print alerts instead of sending") parser.add_argument("--log-level", default=os.environ.get("LOG_LEVEL", "INFO")) args = parser.parse_args(argv) cfg = Config.from_env() if args.dry_run: cfg.dry_run = True logging.basicConfig( level=getattr(logging, args.log_level.upper(), logging.INFO), format="%(asctime)s %(levelname)s %(message)s", ) cycles = 1 if args.once else args.cycles run(cfg, cycles=cycles) return 0 if __name__ == "__main__": sys.exit(main())