vanlink/power/van-battery

#!/usr/bin/env python3
"""van-battery — battery / mains monitor for the campervan router (wayback).

While running **off mains** (AC offline, i.e. on battery) it sends escalating
Pushover alerts as the charge drops past each warn level, and at the shutdown
level it sends a final alert and powers the machine off cleanly.

Alerts are edge-triggered per discharge episode: each severity fires once, and
the whole sequence re-arms when mains power returns. Plug-out already below a
warn level fires a single alert for the current severity, not a burst.

Pushover credentials live in a separate 0600 secrets file (see credentials_path),
never in this repo. Missing/placeholder creds disable sending but NOT the
shutdown — running flat must always power down safely. Stdlib only.
"""

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

CONFIG_PATH = os.environ.get("VAN_BATTERY_CONFIG", "/etc/van-battery/config.json")
STATE_DIR = Path("/run/van-battery")
STATE_PATH = STATE_DIR / "state.json"
PSY = Path("/sys/class/power_supply")
HOST = socket.gethostname()

DEFAULTS = {
    "poll_interval": 30,                    # seconds between reads (battery moves slowly)
    "ac_path": "/sys/class/power_supply/AC0/online",
    "battery_path": "/sys/class/power_supply/BAT0",
    "warn_levels": [25, 20, 15],            # Pushover alert only
    "shutdown_level": 10,                   # Pushover alert + poweroff
    "shutdown_grace": 8,                    # seconds to let the alert flush before poweroff
    "credentials_path": "/etc/van-battery/pushover.json",
}

PLACEHOLDERS = {"", "REPLACE_ME", "your-token-here", "your-user-key-here"}


def log(msg, level="info"):
    pri = {"info": "<6>", "warn": "<4>", "crit": "<2>"}.get(level, "<6>")
    print(pri + msg, flush=True)


def load_config():
    cfg = dict(DEFAULTS)
    try:
        with open(CONFIG_PATH) as f:
            cfg.update(json.load(f))
    except FileNotFoundError:
        log(f"config {CONFIG_PATH} not found, using built-in defaults")
    except Exception as e:
        log(f"config {CONFIG_PATH} unreadable ({e}), using defaults", "warn")
    return cfg


def _by_type(kind):
    """Find a power_supply dir by its `type` (Mains / Battery) — fallback when the
    configured AC0/BAT0 name isn't present on this machine."""
    for d in sorted(PSY.glob("*")):
        try:
            if (d / "type").read_text().strip() == kind:
                return d
        except OSError:
            continue
    return None


def read_on_battery(cfg):
    """True if running on battery (mains absent), False if on mains, None if unknown."""
    p = Path(cfg["ac_path"])
    if not p.exists():
        d = _by_type("Mains")
        p = (d / "online") if d else None
    if not p or not p.exists():
        return None
    try:
        return p.read_text().strip() == "0"
    except OSError:
        return None


def read_capacity(cfg):
    """Battery charge percentage (int), or None."""
    d = Path(cfg["battery_path"])
    if not (d / "capacity").exists():
        d = _by_type("Battery") or d
    try:
        return int((d / "capacity").read_text().strip())
    except (OSError, ValueError):
        return None


def load_creds(cfg):
    try:
        c = json.loads(Path(cfg["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 {cfg['credentials_path']} unreadable ({e})", "warn")
        return None


def pushover(cfg, title, message, priority=0):
    creds = load_creds(cfg)
    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, "priority": priority,
    }).encode()
    req = urllib.request.Request("https://api.pushover.net/1/messages.json", data=data)
    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:
        log(f"pushover send failed: {e}", "warn")
        return False


def severity(cap, levels):
    """Most-severe (lowest) threshold the capacity has reached, or None if above all.
    levels: thresholds sorted ascending. cap=12, levels=[10,15,20,25] -> 15."""
    reached = [t for t in levels if cap <= t]
    return min(reached) if reached else None


def poweroff(cfg):
    log("shutdown level reached — powering off", "crit")
    time.sleep(cfg["shutdown_grace"])   # give the Pushover POST time to land first
    try:
        subprocess.run(["systemctl", "poweroff"], check=False)
    except Exception as e:
        log(f"poweroff failed: {e}", "crit")


def write_state(on_batt, cap, armed):
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    payload = {
        "updated": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "on_battery": on_batt,
        "capacity": cap,
        "alerted_below": armed,   # lowest level alerted this discharge episode, or null
    }
    tmp = STATE_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(payload))
    tmp.replace(STATE_PATH)


def main():
    cfg = load_config()
    levels = sorted(cfg["warn_levels"] + [cfg["shutdown_level"]])
    shutdown_level = cfg["shutdown_level"]
    last_alerted = None       # lowest threshold alerted in the current discharge episode
    shutdown_issued = False
    prev_on_batt = None
    log(f"van-battery up: poll {cfg['poll_interval']}s, warn {cfg['warn_levels']}, "
        f"shutdown {shutdown_level}%")

    while True:
        on_batt = read_on_battery(cfg)
        cap = read_capacity(cfg)

        if on_batt is False:
            # On mains: re-arm the whole sequence for the next discharge episode.
            if prev_on_batt:
                log("mains restored — alerts re-armed")
            last_alerted = None
            shutdown_issued = False
        elif on_batt is True and cap is not None:
            if prev_on_batt is False:
                log(f"mains lost — running on battery at {cap}%", "warn")
            sev = severity(cap, levels)
            if sev is not None and (last_alerted is None or sev < last_alerted):
                last_alerted = sev
                is_shutdown = sev <= shutdown_level
                if is_shutdown:
                    pushover(cfg, f"⚠ {HOST}: battery {cap}% — shutting down",
                             f"On battery at {cap}% (≤{shutdown_level}%). Powering off now to "
                             f"protect the system.", priority=1)
                    if not shutdown_issued:
                        shutdown_issued = True
                        poweroff(cfg)
                else:
                    log(f"battery {cap}% on battery — alerting (level {sev})", "warn")
                    pushover(cfg, f"{HOST}: battery {cap}%",
                             f"Running on battery, charge down to {cap}% (alert at {sev}%). "
                             f"Shutdown at {shutdown_level}%.")

        prev_on_batt = on_batt
        try:
            write_state(on_batt, cap, last_alerted)
        except OSError as e:
            log(f"state write failed: {e}", "warn")
        time.sleep(cfg["poll_interval"])


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