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

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",
}
LED_KEYS = (("power", "led_power"), ("lan", "led_lan"), ("wan", "led_wan"), ("wlan", "led_wlan"))

stop = threading.Event()


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


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 desired_wan(state, uplink_carrier, uplink=None):
    """-> ("netdev", device) | ("heartbeat", None) | ("none", 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):
    """{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)
    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
        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):
    """One evaluation of every usable LED. Returns the files written."""
    want = wanted(cfg)
    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(f"LED {role}: cannot set {trig}: {e}")
            continue
        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(f"poll failed: {e}")
        stop.wait(float(cfg["poll_interval_s"]))
    log("van-leds stopping, LEDs left as they are")
    return 0


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