vanlink/failover/van-modem-watch

#!/usr/bin/env python3
"""van-modem-watch — pages if the EC25 modem never enumerates at boot.

Detection only, deliberately no recovery attempt. On 2026-08-04 the modem
(then on a shared powered USB hub) sometimes failed to enumerate on a cold
boot. Investigation ruled out a boot-timing race and a bad cable/port: a
genuine hub-commanded VBUS power-cycle (confirmed via kernel disconnect/
reconnect events, held off for a full 10s) did not recover it, even with the
rest of the system already up and stable — only a real physical unplug/
replug of the connector ever did. That means no software action from this
host can fix it once it happens; the modem was moved off the hub onto the
Pi's native USB port as the actual fix (a direct port doesn't reproduce the
failure). This just watches for a recurrence and pages, since if it comes
back the only real fix is someone physically reseating the connector.

Pushover credentials shared with van-battery/van-thermal/van-nvme-watch
(/etc/van-battery/pushover.json, 0600). Publishes /run/van-modem-watch/
state.json (same convention as the other watchdogs). Stdlib only.

Usage: van-modem-watch [modem-usb-vendor]
Defaults to 2c7c (Quectel) — see deploy.conf.
"""

import json
import socket
import sys
import time
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

MODEM_VENDOR = sys.argv[1] if len(sys.argv) > 1 else "2c7c"
INTERVAL = 20            # seconds between checks
BUDGET_S = 6 * INTERVAL  # ~2 minutes of retries after boot, then give up quietly
USB_DEVICES = Path("/sys/bus/usb/devices")
STATE_DIR = Path("/run/van-modem-watch")
STATE_PATH = STATE_DIR / "state.json"
CREDENTIALS_PATH = Path("/etc/van-battery/pushover.json")
HOST = socket.gethostname()
PLACEHOLDERS = {"", "REPLACE_ME", "your-token-here", "your-user-key-here"}


def log(msg, level="info"):
    # systemd journal severity prefixes (sd-daemon), same convention as van-ap-watchdog.
    pri = {"info": "<6>", "warn": "<4>", "crit": "<2>"}.get(level, "<6>")
    print(pri + msg, flush=True)


def modem_present(vendor):
    for f in USB_DEVICES.glob("*/idVendor"):
        try:
            if f.read_text().strip() == vendor:
                return True
        except OSError:
            continue
    return False


def load_creds():
    try:
        c = json.loads(CREDENTIALS_PATH.read_text())
        token, user = str(c.get("token", "")).strip(), str(c.get("user", "")).strip()
        if token in PLACEHOLDERS or user in PLACEHOLDERS:
            return None
        return token, user
    except FileNotFoundError:
        return None
    except Exception as e:
        log(f"credentials {CREDENTIALS_PATH} unreadable ({e})", "warn")
        return None


def pushover(title, message, attempts=3, retry_delay=15):
    creds = load_creds()
    if not creds:
        log(f"pushover skipped (no credentials): {title} — {message}", "warn")
        return False
    token, user = creds
    data = urllib.parse.urlencode({
        "token": token, "user": user, "title": title, "message": message,
    }).encode()
    req = urllib.request.Request("https://api.pushover.net/1/messages.json", data=data)
    for attempt in range(1, attempts + 1):
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                ok = resp.status == 200
            if not ok:
                log(f"pushover HTTP {resp.status}", "warn")
            return ok
        except Exception as e:
            last = attempt == attempts
            log(f"pushover send failed ({attempt}/{attempts}): {e}", "warn")
            if not last:
                time.sleep(retry_delay)
    return False


def write_state(payload):
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    tmp = STATE_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(payload))
    tmp.replace(STATE_PATH)


def main():
    log(f"van-modem-watch up: watching for USB vendor {MODEM_VENDOR} for up to "
        f"{BUDGET_S}s after boot")
    elapsed = 0
    while elapsed < BUDGET_S:
        if modem_present(MODEM_VENDOR):
            log("modem present — exiting")
            write_state({
                "updated": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                "present": True,
            })
            return
        time.sleep(INTERVAL)
        elapsed += INTERVAL

    message = (f"modem (USB vendor {MODEM_VENDOR}) not seen {BUDGET_S}s after boot — "
               f"needs a physical unplug/replug, no software fix works for this")
    log(message, "crit")
    write_state({
        "updated": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "present": False,
    })
    pushover(f"🚨 {HOST}: cellular modem absent after boot", message)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)