vanlink/power/van-thermal

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

One small daemon that does three jobs off a single sysfs sample loop:
  1. publishes /run/van-thermal/state.json  (the Cockpit "Temps" card reads this,
     exactly like van-failover's state.json — no shelling out to `sensors` per refresh)
  2. warns to the journal on threshold crossings, with hysteresis so a sensor hovering
     on the line doesn't spam (journalctl -u van-thermal)
  3. appends throttled CSV history to /var/log/van-thermal.csv with self-rotation

Sensors are resolved by hwmon *name* + *label* at runtime, never by hwmonN index
(that number is assigned at boot and is not stable). Stdlib only.
"""

import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

CONFIG_PATH = os.environ.get("VAN_THERMAL_CONFIG", "/etc/van-thermal/config.json")
STATE_DIR = Path("/run/van-thermal")
STATE_PATH = STATE_DIR / "state.json"
HWMON = Path("/sys/class/hwmon")

DEFAULTS = {
    "sample_interval": 10,      # seconds between sysfs reads (alerting cadence)
    "log_interval": 60,         # seconds between CSV rows (history cadence)
    "log_path": "/var/log/van-thermal.csv",
    "log_max_bytes": 5 * 1024 * 1024,   # rotate to .1 past this, keep one old file
    "sensors": [
        {"name": "cpu",  "hwmon": "coretemp", "label": "Package id 0",
         "warn": 85, "crit": 95, "clear_margin": 5},
        {"name": "nvme", "hwmon": "nvme",     "label": "Composite",
         "warn": 65, "crit": 70, "clear_margin": 5},
    ],
}

LEVELS = ("ok", "warn", "crit")


def log(msg, level="info"):
    # systemd journal severity prefixes (sd-daemon); shows up via journalctl -p.
    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 built-in defaults", "warn")
    return cfg


def find_hwmon(name):
    """Return the hwmon dir whose name matches, or None. Re-resolved on demand
    because the hwmonN index can shift across boots / module reloads."""
    for d in sorted(HWMON.glob("hwmon*")):
        try:
            if (d / "name").read_text().strip() == name:
                return d
        except OSError:
            continue
    return None


def read_temp(spec, cache):
    """Read one sensor's temperature in °C, or None if unavailable.

    spec: {hwmon, label}. Resolves hwmon dir + the tempN whose *_label matches,
    falling back to temp1 when the chip exposes no labels."""
    d = cache.get(spec["hwmon"])
    if d is None or not d.exists():
        d = find_hwmon(spec["hwmon"])
        cache[spec["hwmon"]] = d
    if d is None:
        return None

    input_file = None
    want = spec.get("label")
    if want:
        for lbl in sorted(d.glob("temp*_label")):
            try:
                if lbl.read_text().strip() == want:
                    input_file = lbl.with_name(lbl.name.replace("_label", "_input"))
                    break
            except OSError:
                continue
    if input_file is None:
        input_file = d / "temp1_input"   # label not found / chip is label-less

    try:
        return int(input_file.read_text().strip()) / 1000.0
    except (OSError, ValueError):
        # hwmon may have re-enumerated; drop the cache so next pass re-resolves.
        cache[spec["hwmon"]] = None
        return None


def classify(temp, spec, prev_level):
    """Level with hysteresis: step up at the threshold, step down only after
    dropping clear_margin below it, so a sensor on the line doesn't oscillate."""
    if temp is None:
        return prev_level
    warn, crit = spec["warn"], spec["crit"]
    margin = spec.get("clear_margin", 5)
    if temp >= crit:
        return "crit"
    if temp >= warn:
        return "crit" if prev_level == "crit" and temp > crit - margin else "warn"
    if temp >= warn - margin:
        return prev_level if prev_level in ("warn", "crit") else "ok"
    return "ok"


def announce(name, temp, old, new):
    if old == new:
        return
    rising = LEVELS.index(new) > LEVELS.index(old)
    sev = {"crit": "crit", "warn": "warn", "ok": "info"}[new]
    arrow = "rose to" if rising else "fell back to"
    log(f"{name} {arrow} {new.upper()} ({temp:.1f}°C)", sev if rising else "info")


def rotate_log(path, max_bytes):
    try:
        if path.exists() and path.stat().st_size > max_bytes:
            path.replace(path.with_suffix(path.suffix + ".1"))
    except OSError as e:
        log(f"log rotate failed: {e}", "warn")


def write_csv(cfg, readings):
    path = Path(cfg["log_path"])
    rotate_log(path, cfg["log_max_bytes"])
    new = not path.exists()
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "a") as f:
            if new:
                f.write(",".join(["time"] + sum(
                    [[f"{s['name']}_c", f"{s['name']}_lvl"] for s in cfg["sensors"]], [])) + "\n")
            row = [datetime.now(timezone.utc).isoformat(timespec="seconds")]
            for s in cfg["sensors"]:
                r = readings[s["name"]]
                row += ["" if r["temp"] is None else f"{r['temp']:.1f}", r["level"]]
            f.write(",".join(row) + "\n")
    except OSError as e:
        log(f"csv write failed: {e}", "warn")


def write_state(readings):
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    payload = {
        "updated": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "sensors": readings,
    }
    tmp = STATE_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(payload))
    tmp.replace(STATE_PATH)   # atomic swap so the Cockpit reader never sees a partial file


def main():
    cfg = load_config()
    cache = {}
    levels = {s["name"]: "ok" for s in cfg["sensors"]}
    last_log = 0.0
    log(f"van-thermal up: sampling every {cfg['sample_interval']}s, "
        f"logging every {cfg['log_interval']}s to {cfg['log_path']}")

    while True:
        readings = {}
        for s in cfg["sensors"]:
            temp = read_temp(s, cache)
            new = classify(temp, s, levels[s["name"]])
            if temp is not None:
                announce(s["name"], temp, levels[s["name"]], new)
            levels[s["name"]] = new
            readings[s["name"]] = {
                "temp": None if temp is None else round(temp, 1),
                "level": new, "warn": s["warn"], "crit": s["crit"],
            }

        try:
            write_state(readings)
        except OSError as e:
            log(f"state write failed: {e}", "warn")

        now = time.monotonic()
        if now - last_log >= cfg["log_interval"]:
            write_csv(cfg, readings)
            last_log = now

        time.sleep(cfg["sample_interval"])


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