sam/hub_fixes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
hub_fixes/watchdog/watchdog.py
Sam Rolfe 02d8815e53 Directive 1: voice intake watchdog (layer attribution, daily repeat, Apprise) 
Build the Hub Fixes Directive 1 watchdog as a small standalone service on .13.

- Watch voice/audio_stream (paho); 15 min silence = 3 missed heartbeats.
- Attribute failures in order: broker -> voice_bridge -> ESP32/no-publisher.
  The ESP32 is never blamed when the broker or the bridge broke.
- Alert via Apprise on .35:8210 (tag sam_ntfy; daily repeat escalates to sam).
- Persist state; repeat once a day while broken; recovery notice; alerts stop.
- No auto-heal of any kind. Notification only.
- Require publisher evidence (heartbeat or retained presence) before treating
  silence as a failure, so the unflashed Directive 3 firmware cannot cause the
  cry-wolf false alarms the spec warns about.
- Docker: host network, read-only Docker socket, state volume, unless-stopped.
- 23 unit tests; live evidence captured in watchdog/README.md.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-14 11:10:30 +11:00

693 lines
25 KiB
Python
Raw Permalink Blame History

#!/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())
Reference in New Issue View Git Blame Copy Permalink