Andreas WredeandClaude Opus 4.8
52e948654e
Pushover notification when the router loses mains (now on battery) and when mains is restored. One alert per real transition; first sample is skipped so a restart while already on battery doesn't false-fire. Independent of the low-charge 25/20/15/10% alerts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
230 lines
8.3 KiB
Python
230 lines
8.3 KiB
Python
#!/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)
|
|
|
|
capstr = f"{cap}%" if cap is not None else "unknown charge"
|
|
|
|
# Mains <-> battery transition alerts. Skip the very first sample (prev is None)
|
|
# so a restart while already on battery doesn't fire a spurious "on battery".
|
|
if on_batt is not None and prev_on_batt is not None and on_batt != prev_on_batt:
|
|
if on_batt:
|
|
log(f"mains lost — running on battery at {capstr}", "warn")
|
|
pushover(cfg, f"⚡ {HOST}: on battery",
|
|
f"Mains power lost — now running on battery ({capstr}). "
|
|
f"Low alerts at {cfg['warn_levels']}%, auto-shutdown at {shutdown_level}%.")
|
|
else:
|
|
log(f"mains restored at {capstr} — alerts re-armed")
|
|
pushover(cfg, f"🔌 {HOST}: back on mains",
|
|
f"Mains power restored ({capstr}). Battery alert sequence re-armed.")
|
|
|
|
if on_batt is False:
|
|
# On mains: re-arm the whole sequence for the next discharge episode.
|
|
last_alerted = None
|
|
shutdown_issued = False
|
|
elif on_batt is True and cap is not None:
|
|
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)
|