vanlink/leds/van-leds

#!/usr/bin/env python3
"""van-leds — drive the NanoPi R5C's four front LEDs (power / LAN / WAN / WiFi).

  power  default-on
  LAN    netdev on the LAN port (link + tx + rx)
  WiFi   netdev on the 5GHz AP radio (link + tx + rx)
  WAN    follows van-failover (/run/van-failover/state.json), re-evaluated each poll:
           active WAN healthy            -> netdev on that device
           active WAN unhealthy, or no failover info / no active WAN but the wired
           uplink has carrier            -> heartbeat (link, no healthy internet)
           nothing                       -> trigger none, off
         Missing/invalid state file: fall back to the wired uplink (netdev if carrier).

An LED whose name is empty in the config, or absent under sysfs, is skipped (logged
once at start). Settings are re-asserted every poll but only written when they differ
from what sysfs reads back. SIGTERM/SIGINT exit cleanly and leave the LEDs as they are.
Stdlib only; config is /etc/van-leds/config.json (argv[1] overrides). Test-only config
keys: "sysfs_leds", "sysfs_net", "failover_state".
"""

import json
import os
import signal
import sys
import threading
import time
from datetime import datetime

DEFAULTS = {
    "led_power": "", "led_lan": "", "led_wan": "", "led_wlan": "",
    "lan_iface": "", "wlan_iface": "", "uplink_iface": "",
    "poll_interval_s": 5,
    "sysfs_leds": "/sys/class/leds",
    "sysfs_net": "/sys/class/net",
    "failover_state": "/run/van-failover/state.json",
    "state_max_age_s": 60,      # older failover state is ignored (wired-uplink fallback)
}
LED_KEYS = (("power", "led_power"), ("lan", "led_lan"), ("wan", "led_wan"), ("wlan", "led_wlan"))

stop = threading.Event()


def log(msg):
    print(msg, flush=True)


_last = {}


def log_once(key, msg):
    """Log msg unless it is what this key last logged (a persistent failure is logged once)."""
    if _last.get(key) != msg:
        _last[key] = msg
        log(msg)


def clear_log(key):
    _last.pop(key, None)


def interval(cfg):
    try:
        return max(1.0, float(cfg["poll_interval_s"]))
    except (TypeError, ValueError):
        return 5.0


def load_config(path):
    cfg = dict(DEFAULTS)
    with open(path) as f:
        cfg.update(json.load(f))
    return cfg


def read_attr(path):
    """Stripped file content; for a sysfs trigger list ("none [netdev] ...") the bracketed item."""
    try:
        with open(path) as f:
            s = f.read().strip()
    except OSError:
        return None
    if "[" in s and "]" in s:
        s = s[s.index("[") + 1:s.index("]")]
    return s


def write_attr(path, value):
    with open(path, "w") as f:
        f.write(str(value))


def state_stale(state, now, max_age):
    """True when state["updated"] parses and is older than max_age seconds (unparseable: not stale)."""
    try:
        t = datetime.strptime(state["updated"], "%Y-%m-%dT%H:%M:%S%z").timestamp()
    except (TypeError, KeyError, ValueError):
        return False
    return now - t > max_age


def desired_wan(state, uplink_carrier, uplink=None, now=None, max_age=60):
    """-> ("netdev", device) | ("heartbeat", None) | ("none", None). `now`: epoch seconds (default: clock)."""
    if isinstance(state, dict) and state_stale(state, time.time() if now is None else now, max_age):
        state = None
    valid = isinstance(state, dict) and isinstance(state.get("wans"), list) and "active_device" in state
    if not valid:  # no failover info: judge by the wired uplink alone
        if not uplink_carrier:
            return ("none", None)
        return ("netdev", uplink) if uplink else ("heartbeat", None)
    active = state.get("active_device")
    if active:
        for w in state["wans"]:
            if isinstance(w, dict) and w.get("device") == active and w.get("healthy") is True:
                return ("netdev", active)
        return ("heartbeat", None)
    return ("heartbeat", None) if uplink_carrier else ("none", None)


def apply_led(led_dir, trigger, device=None):
    """Make the LED match; write only what differs. Returns the list of files written."""
    writes = []

    def want(name, value):
        p = os.path.join(led_dir, name)
        if read_attr(p) != str(value):
            write_attr(p, value)
            writes.append(p)

    want("trigger", trigger)
    if trigger == "netdev":
        want("device_name", device)
        for a in ("link", "tx", "rx"):
            want(a, 1)
    elif trigger == "none":
        want("brightness", 0)
    return writes


def load_state(path):
    try:
        with open(path) as f:
            return json.load(f)
    except (OSError, ValueError):
        return None


def carrier(cfg, iface):
    return bool(iface) and read_attr(os.path.join(cfg["sysfs_net"], iface, "carrier")) == "1"


def wanted(cfg, now=None):
    """{role: (trigger, device)} for every role (before skipping absent LEDs)."""
    wan = desired_wan(load_state(cfg["failover_state"]), carrier(cfg, cfg["uplink_iface"]),
                      cfg["uplink_iface"] or None, now, cfg["state_max_age_s"])
    return {
        "power": ("default-on", None),
        "lan": ("netdev", cfg["lan_iface"]),
        "wlan": ("netdev", cfg["wlan_iface"]),
        "wan": wan,
    }


def usable_leds(cfg, announce=False):
    """{role: led_dir} for the configured LEDs present in sysfs; optionally logs the skipped ones."""
    out = {}
    for role, key in LED_KEYS:
        name = cfg[key]
        if not name:
            if announce:
                log(f"LED {role}: no name configured, skipped")
            continue
        iface_key = {"lan": "lan_iface", "wlan": "wlan_iface"}.get(role)
        if iface_key and not cfg[iface_key]:
            if announce:
                log(f"LED {role}: {iface_key} is empty, skipped")
            continue
        d = os.path.join(cfg["sysfs_leds"], name)
        if not os.path.isdir(d):
            if announce:
                log(f"LED {role}: {name} not present under {cfg['sysfs_leds']}, skipped")
            continue
        out[role] = d
    return out


def poll_once(cfg, now=None):
    """One evaluation of every usable LED. Returns the files written."""
    want = wanted(cfg, now)
    writes = []
    for role, d in usable_leds(cfg).items():
        trig, dev = want[role]
        try:
            w = apply_led(d, trig, dev)
        except OSError as e:
            log_once(role, f"LED {role}: cannot set {trig}: {e}")
            continue
        clear_log(role)
        if w:
            log(f"LED {role}: {trig}" + (f" {dev}" if dev else ""))
        writes += w
    return writes


def main(argv):
    path = argv[1] if len(argv) > 1 else "/etc/van-leds/config.json"
    cfg = load_config(path)
    signal.signal(signal.SIGTERM, lambda *_: stop.set())
    signal.signal(signal.SIGINT, lambda *_: stop.set())
    usable_leds(cfg, announce=True)
    log(f"van-leds started ({path})")
    while not stop.is_set():
        try:
            poll_once(cfg)
        except Exception as e:  # never die on a transient sysfs/state oddity
            log_once("poll", f"poll failed: {e}")
        else:
            clear_log("poll")
        stop.wait(interval(cfg))
    log("van-leds stopping, LEDs left as they are")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))