Add MQTT handover controller and systemd installer

controller.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+"""Zorruno MeshCore Phone App Mode v1.0 — 2026-10-03.
+
+Runs on beaver independently of meshcore-webui. Only ON/OFF commands on the
+fixed MQTT topic are accepted. Container names and executable arguments never
+come from MQTT. No container action occurs at startup or broker reconnect.
+"""
+
+import argparse
+import configparser
+import json
+import logging
+import queue
+import shutil
+import signal
+import subprocess
+import threading
+import time
+from datetime import datetime, timezone
+
+CONTAINER = "meshcore-webui"
+COMMAND_TOPIC = "viewroad-commands/meshcore/phone_app_mode"
+STATE_TOPIC = "viewroad-status/meshcore/phone_app_mode"
+AVAILABILITY_TOPIC = "viewroad-status/meshcore/controller/availability"
+LOG = logging.getLogger("meshcore-phone-mode")
+
+
+def phone_mode(status):
+    if status in {"exited", "created"}:
+        return "ON"
+    if status in {"running", "restarting"}:
+        return "OFF"
+    raise RuntimeError("Unexpected WebUI container state: " + status)
+
+
+class DockerControl:
+    def __init__(self):
+        self.executable = shutil.which("docker")
+        if not self.executable:
+            raise RuntimeError("Docker CLI is not installed")
+
+    def run(self, *args, timeout=15):
+        result = subprocess.run(
+            [self.executable, *args], capture_output=True, text=True,
+            timeout=timeout, check=False,
+        )
+        if result.returncode:
+            raise RuntimeError(result.stderr.strip() or "Docker command failed")
+        return result.stdout.strip()
+
+    def status(self):
+        return self.run("inspect", "--format", "{{.State.Status}}", CONTAINER)
+
+    def set_mode(self, requested):
+        if requested not in {"ON", "OFF"}:
+            raise ValueError("Only ON or OFF is allowed")
+        before = self.status()
+        if phone_mode(before) == requested:
+            return before
+        if requested == "ON":
+            self.run("stop", "--time", "10", CONTAINER, timeout=35)
+        else:
+            self.run("start", CONTAINER, timeout=35)
+        after = self.status()
+        if phone_mode(after) != requested:
+            raise RuntimeError("WebUI did not reach the requested state: " + after)
+        return after
+
+
+class Controller:
+    def __init__(self, client, docker, poll_seconds=15):
+        self.client = client
+        self.docker = docker
+        self.poll_seconds = poll_seconds
+        self.connected = threading.Event()
+        self.stopping = threading.Event()
+        self.refresh = threading.Event()
+        self.commands = queue.Queue(maxsize=10)
+        self.subscription_mid = None
+        client.on_connect = self.on_connect
+        client.on_subscribe = self.on_subscribe
+        client.on_disconnect = self.on_disconnect
+        client.on_message = self.on_message
+
+    def on_connect(self, client, userdata, flags, reason_code, properties):
+        if reason_code != 0:
+            LOG.error("MQTT connection rejected: %s", reason_code)
+            return
+        self.connected.clear()
+        client.publish(AVAILABILITY_TOPIC, "offline", qos=1, retain=True)
+        result, self.subscription_mid = client.subscribe(COMMAND_TOPIC, qos=1)
+        if result != 0:
+            LOG.error("MQTT subscribe failed")
+            client.disconnect()
+            return
+        LOG.info("MQTT connected; waiting for command subscription")
+
+    def on_subscribe(self, client, userdata, mid, reason_codes, properties):
+        if mid != self.subscription_mid:
+            return
+        if not reason_codes or any(code >= 128 for code in reason_codes):
+            LOG.error("MQTT command subscription denied; check broker permissions")
+            self.connected.clear()
+            return
+        # Never advertise online before subscription and Docker are checked.
+        self.refresh.set()
+        self.connected.set()
+        LOG.info("MQTT subscribed; checking current WebUI state")
+
+    def on_disconnect(self, client, userdata, flags, reason_code, properties):
+        self.connected.clear()
+        self.clear_commands()
+        if not self.stopping.is_set():
+            LOG.warning("MQTT disconnected; will reconnect automatically")
+
+    def clear_commands(self):
+        while True:
+            try:
+                self.commands.get_nowait()
+            except queue.Empty:
+                return
+
+    def on_message(self, client, userdata, message):
+        if message.topic != COMMAND_TOPIC or not self.connected.is_set():
+            return
+        if message.retain:
+            LOG.warning("Ignored retained command; publish commands without retain")
+            return
+        if len(message.payload) > 16:
+            LOG.warning("Ignored invalid command")
+            return
+        try:
+            requested = message.payload.decode("utf-8").strip()
+        except UnicodeDecodeError:
+            LOG.warning("Ignored non-text command")
+            return
+        if requested not in {"ON", "OFF"}:
+            LOG.warning("Ignored invalid command; only ON/OFF is accepted")
+            return
+        try:
+            self.commands.put_nowait(requested)
+        except queue.Full:
+            LOG.warning("Command queue full; ignored command")
+
+    def publish_confirmed(self, topic, payload):
+        info = self.client.publish(topic, payload, qos=1, retain=True)
+        if info.rc != 0:
+            raise RuntimeError("MQTT publish failed")
+        info.wait_for_publish(timeout=5)
+        if not info.is_published():
+            raise RuntimeError("MQTT publication was not acknowledged")
+
+    def report(self, status=None):
+        try:
+            status = status if status is not None else self.docker.status()
+            state = {
+                "phone_app_mode": phone_mode(status),
+                "container": CONTAINER,
+                "container_status": status,
+                "observed_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+            }
+            self.publish_confirmed(STATE_TOPIC, json.dumps(state))
+            self.publish_confirmed(AVAILABILITY_TOPIC, "online")
+            return True
+        except Exception as exc:
+            LOG.error("State check failed: %s", exc)
+            try:
+                self.publish_confirmed(AVAILABILITY_TOPIC, "offline")
+            except Exception:
+                pass
+            return False
+
+    def process_command(self, requested):
+        if not self.connected.is_set() or self.stopping.is_set():
+            return
+        try:
+            LOG.info("Phone App Mode requested: %s", requested)
+            status = self.docker.set_mode(requested)
+            self.report(status)
+        except Exception as exc:
+            LOG.error("Container control failed: %s", exc)
+            # Report what actually happened, even if the operation failed.
+            self.report()
+
+    def run(self):
+        next_report = 0
+        self.client.loop_start()
+        try:
+            while not self.stopping.is_set():
+                if not self.connected.is_set():
+                    self.stopping.wait(0.5)
+                    continue
+                if self.refresh.is_set() or time.monotonic() >= next_report:
+                    self.refresh.clear()
+                    self.report()
+                    next_report = time.monotonic() + self.poll_seconds
+                try:
+                    requested = self.commands.get(timeout=0.5)
+                except queue.Empty:
+                    continue
+                self.process_command(requested)
+        finally:
+            if self.connected.is_set():
+                try:
+                    self.publish_confirmed(AVAILABILITY_TOPIC, "offline")
+                except Exception:
+                    pass
+            self.client.disconnect()
+            self.client.loop_stop()
+
+
+def load_config(path):
+    config = configparser.ConfigParser(interpolation=None)
+    if not config.read(path):
+        raise RuntimeError("Cannot read config: " + path)
+    host = config.get("mqtt", "host", fallback="127.0.0.1").strip()
+    port = config.getint("mqtt", "port", fallback=1883)
+    poll = config.getfloat("controller", "poll_seconds", fallback=15)
+    if not host or not 1 <= port <= 65535 or not 5 <= poll <= 300:
+        raise ValueError("Invalid host/port/poll interval in config")
+    return host, port, poll, config
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--config", default="/etc/meshcore-phone-mode.ini")
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    host, port, poll, config = load_config(args.config)
+    import paho.mqtt.client as mqtt
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id="beaver-meshcore-phone-mode",
+        clean_session=True,
+        protocol=mqtt.MQTTv311,
+    )
+    username = config.get("mqtt", "username", fallback="").strip()
+    password = config.get("mqtt", "password", fallback="")
+    if username:
+        client.username_pw_set(username, password)
+    client.will_set(AVAILABILITY_TOPIC, "offline", qos=1, retain=True)
+    client.reconnect_delay_set(min_delay=1, max_delay=30)
+    client.connect_timeout = 10
+    controller = Controller(client, DockerControl(), poll)
+    for sig in (signal.SIGTERM, signal.SIGINT):
+        signal.signal(sig, lambda *_: controller.stopping.set())
+    client.connect_async(host, port, keepalive=30)
+    controller.run()
+
+
+if __name__ == "__main__":
+    main()