LuxStats/host/system_monitor.py

"""LuxStats system monitor v3.

Sources (all toggleable in the configurator):

  cpu, gpu, ram         0-1 utilisation
  cpu_temp, gpu_temp    RAW Celsius (not normalised)
  eth                   wired link speed in Mbps; 0 unless the link is
                        actually ACTIVE (carrier up + operstate up)
  wifi                  band GHz (0/2/5/6)
  vpn, cam              bool
  disk_<dev>            0-1 usage per detected physical drive
  disk_<dev>_temp       RAW Celsius where a nvme/drivetemp sensor exists
  disk / disk_temp      aggregates (max usage / max temp)

Camera detection ignores media daemons (PipeWire/WirePlumber/PulseAudio)
that hold /dev/video* open permanently on modern Arch/GNOME setups - only a
"real" consumer (browser, OBS, Zoom, ...) counts as camera-in-use.

Slow probes are cached with per-probe minimum ages so the configurator can
poll at 100 ms without spawning subprocesses at 10 Hz.
"""

import glob
import os
import re
import shutil
import subprocess
import sys
import threading
import time

import psutil

VPN_IF_PATTERNS = re.compile(r"^(tun|tap|wg|ppp|tailscale|zt|nordlynx|proton)", re.I)
VIRTUAL_IF_PATTERNS = re.compile(
    r"^(lo|docker|veth|br-|virbr|vmnet|vbox|kube|flannel|cni)", re.I
)

# Media stacks that keep camera devices open without actually using them.
# /proc/<pid>/comm is truncated to 15 chars, so compare prefixes.
CAMERA_IGNORE_PREFIXES = (
    "pipewire", "wireplumber", "pulseaudio", "v4l2-relayd",
    "pipewire-media", "pipewire-pulse",
)

# Minimum seconds between real reads of each slow probe (values are cached
# in between so 100 ms polling stays cheap).
PROBE_MIN_AGE = {
    "cpu_temp": 1.0, "gpu": 1.0, "wifi": 1.0, "cam": 0.5,
    "eth": 0.3, "drives_usage": 2.0, "drives_temp": 2.0,
}

_cache = {}


def _cached(key, fn):
    now = time.monotonic()
    ent = _cache.get(key)
    if ent and now - ent[0] < PROBE_MIN_AGE.get(key, 0.0):
        return ent[1]
    val = fn()
    _cache[key] = (now, val)
    return val


def _run(cmd, timeout=2):
    try:
        out = subprocess.run(cmd, capture_output=True, text=True,
                             timeout=timeout)
        return out.stdout
    except (OSError, subprocess.SubprocessError):
        return ""


def _wireless_ifaces():
    if sys.platform.startswith("linux"):
        return [os.path.basename(os.path.dirname(p))
                for p in glob.glob("/sys/class/net/*/wireless")]
    return []


def _sensors():
    try:
        return psutil.sensors_temperatures()
    except (AttributeError, OSError):
        return {}


# --------------------------------------------------------------------------
# CPU / RAM / temps (raw Celsius)
# --------------------------------------------------------------------------

def cpu_percent():
    return psutil.cpu_percent(interval=None) / 100.0


def ram_percent():
    return psutil.virtual_memory().percent / 100.0


def cpu_temp():
    """Raw CPU temperature in Celsius, or None."""
    temps = _sensors()
    for key in ("coretemp", "k10temp", "zenpower", "cpu_thermal", "acpitz"):
        if key in temps and temps[key]:
            return float(temps[key][0].current)
    return None


# --------------------------------------------------------------------------
# GPU (NVIDIA via nvidia-smi, AMD via sysfs/hwmon) - temp is raw Celsius
# --------------------------------------------------------------------------

_NVSMI = shutil.which("nvidia-smi")


def _nvidia_query():
    if not _NVSMI:
        return None
    out = _run([_NVSMI, "--query-gpu=utilization.gpu,temperature.gpu",
                "--format=csv,noheader,nounits"], timeout=3)
    m = re.match(r"\s*(\d+)\s*,\s*(\d+)", out or "")
    if m:
        return int(m.group(1)) / 100.0, float(m.group(2))
    return None


def _amd_util():
    for path in glob.glob("/sys/class/drm/card*/device/gpu_busy_percent"):
        try:
            with open(path) as f:
                return int(f.read().strip()) / 100.0
        except (OSError, ValueError):
            continue
    return None


def gpu_stats():
    """Return (util 0-1 or None, temp raw Celsius or None)."""
    nv = _nvidia_query()
    if nv:
        return nv
    util = _amd_util()
    temp = None
    temps = _sensors()
    for key in ("amdgpu", "radeon", "nouveau"):
        if key in temps and temps[key]:
            temp = float(temps[key][0].current)
            break
    return util, temp


# --------------------------------------------------------------------------
# Drives - usage per physical drive + raw Celsius temps
# --------------------------------------------------------------------------

def _device_base(device):
    name = os.path.basename(device.rstrip("\\/"))
    if sys.platform == "win32":
        return name.rstrip(":") or name
    if name.startswith(("nvme", "mmcblk")):
        return re.sub(r"p\d+$", "", name)
    if re.match(r"^(sd[a-z]+|hd[a-z]+|vd[a-z]+|xvd[a-z]+)\d*$", name):
        return re.sub(r"\d+$", "", name)
    return name


def physical_drives():
    drives = {}
    try:
        parts = psutil.disk_partitions(all=False)
    except OSError:
        return drives
    for p in parts:
        if not p.device:
            continue
        if p.fstype in ("squashfs", "iso9660", "overlay", "tmpfs"):
            continue
        drives.setdefault(_device_base(p.device), []).append(p.mountpoint)
    return drives


def _drives_usage():
    out = {}
    for d, mounts in physical_drives().items():
        worst = None
        for m in mounts:
            try:
                worst = max(worst or 0.0,
                            psutil.disk_usage(m).percent / 100.0)
            except (OSError, PermissionError):
                continue
        if worst is not None:
            out[d] = worst
    return out


def drives_usage():
    return _cached("drives_usage", _drives_usage)


def _drives_temp():
    """Best-effort {drive: raw Celsius} from nvme/drivetemp sensors."""
    temps = _sensors()
    result = {}
    drives = sorted(physical_drives())
    nvme = [d for d in drives if d.startswith("nvme")]
    sata = [d for d in drives if re.match(r"^(sd|hd)[a-z]+$", d)]
    for i, entry in enumerate(temps.get("nvme", [])):
        if i < len(nvme):
            result[nvme[i]] = float(entry.current)
    for i, entry in enumerate(temps.get("drivetemp", [])):
        if i < len(sata):
            result[sata[i]] = float(entry.current)
    return result


def drives_temp():
    return _cached("drives_temp", _drives_temp)


# --------------------------------------------------------------------------
# Network - eth requires an ACTIVE link (carrier), not just a reported speed
# --------------------------------------------------------------------------

def _iface_active(name):
    """True only when the interface has a live link."""
    if sys.platform.startswith("linux"):
        try:
            with open(f"/sys/class/net/{name}/carrier") as f:
                if f.read().strip() != "1":
                    return False
        except OSError:
            return False
        try:
            with open(f"/sys/class/net/{name}/operstate") as f:
                return f.read().strip() in ("up", "unknown")
        except OSError:
            return False
    st = psutil.net_if_stats().get(name)
    return bool(st and st.isup)


def _eth_speed():
    """Best speed (Mbps) among ACTIVE wired links; 0 when nothing is live."""
    wireless = set(_wireless_ifaces())
    best = 0
    for name, st in psutil.net_if_stats().items():
        if name in wireless:
            continue
        if VPN_IF_PATTERNS.match(name) or VIRTUAL_IF_PATTERNS.match(name):
            continue
        if not st.isup or not st.speed or st.speed <= 0:
            continue
        if not _iface_active(name):
            continue
        best = max(best, st.speed)
    return best


def eth_speed():
    return _cached("eth", _eth_speed)


def _wifi_band():
    freq_mhz = 0
    if sys.platform.startswith("linux"):
        iw = shutil.which("iw")
        for iface in _wireless_ifaces():
            if not _iface_active(iface):
                continue
            if iw:
                m = re.search(r"freq:\s*(\d+)",
                              _run([iw, "dev", iface, "link"]))
                if m:
                    freq_mhz = max(freq_mhz, int(m.group(1)))
        if not freq_mhz and shutil.which("nmcli"):
            out = _run(["nmcli", "-t", "-f", "ACTIVE,FREQ", "dev", "wifi"])
            for line in out.splitlines():
                if line.startswith(("yes:", "sí:", "oui:")):
                    m = re.search(r"(\d+)\s*MHz", line)
                    if m:
                        freq_mhz = max(freq_mhz, int(m.group(1)))
    elif sys.platform == "win32":
        out = _run(["netsh", "wlan", "show", "interfaces"])
        m = re.search(r"Band\s*:\s*(\d+(?:\.\d+)?)\s*GHz", out)
        if m:
            return {2.4: 2}.get(float(m.group(1)), int(float(m.group(1))))
        m = re.search(r"Channel\s*:\s*(\d+)", out)
        if m:
            return 2 if int(m.group(1)) <= 14 else 5
    if freq_mhz >= 5925:
        return 6
    if freq_mhz >= 4900:
        return 5
    if freq_mhz >= 2400:
        return 2
    return 0


def wifi_band():
    return _cached("wifi", _wifi_band)


def vpn_active():
    for name, st in psutil.net_if_stats().items():
        if st.isup and VPN_IF_PATTERNS.match(name):
            return True
    return False


# --------------------------------------------------------------------------
# Camera - ignore media daemons that hold /dev/video* open permanently
# --------------------------------------------------------------------------

def _pid_comm(pid):
    try:
        with open(f"/proc/{pid}/comm") as f:
            return f.read().strip()
    except OSError:
        return ""


def _camera_holder_pids(devices):
    pids = set()
    fuser = shutil.which("fuser")
    if fuser:
        for tok in _run([fuser] + devices).split():
            tok = tok.strip().rstrip("mrce")   # fuser access-mode suffixes
            if tok.isdigit():
                pids.add(tok)
        return pids
    try:
        for pid in filter(str.isdigit, os.listdir("/proc")):
            fd_dir = f"/proc/{pid}/fd"
            try:
                for fd in os.listdir(fd_dir):
                    if os.readlink(os.path.join(fd_dir, fd)) \
                            .startswith("/dev/video"):
                        pids.add(pid)
                        break
            except OSError:
                continue
    except OSError:
        pass
    return pids


def _camera_active():
    if not sys.platform.startswith("linux"):
        return None
    devices = glob.glob("/dev/video*")
    if not devices:
        return False
    for pid in _camera_holder_pids(devices):
        comm = _pid_comm(pid).lower()
        if comm and not comm.startswith(CAMERA_IGNORE_PREFIXES):
            return True
    return False


def camera_active():
    return _cached("cam", _camera_active)


# --------------------------------------------------------------------------
# Source registry + collection
# --------------------------------------------------------------------------

BASE_SOURCES = ["cpu", "cpu_temp", "gpu", "gpu_temp", "ram",
                "eth", "wifi", "vpn", "cam"]


def discover_sources():
    sources = list(BASE_SOURCES)
    drives = physical_drives()
    temps = drives_temp()
    sources.append("disk")
    if temps:
        sources.append("disk_temp")
    for d in sorted(drives):
        sources.append(f"disk_{d}")
        if d in temps:
            sources.append(f"disk_{d}_temp")
    return sources


def collect(enabled=None):
    def want(key):
        return enabled is None or key in enabled

    values = {}
    if want("cpu"):
        values["cpu"] = round(cpu_percent(), 3)
    if want("cpu_temp"):
        t = _cached("cpu_temp", cpu_temp)
        if t is not None:
            values["cpu_temp"] = round(t, 1)
    if want("gpu") or want("gpu_temp"):
        util, temp = _cached("gpu", gpu_stats)
        if want("gpu") and util is not None:
            values["gpu"] = round(util, 3)
        if want("gpu_temp") and temp is not None:
            values["gpu_temp"] = round(temp, 1)
    if want("ram"):
        values["ram"] = round(ram_percent(), 3)
    if want("eth"):
        values["eth"] = eth_speed()
    if want("wifi"):
        values["wifi"] = wifi_band()
    if want("vpn"):
        values["vpn"] = vpn_active()
    if want("cam"):
        cam = camera_active()
        if cam is not None:
            values["cam"] = cam

    usage = drives_usage() if (want("disk") or any(
        want(f"disk_{d}") for d in physical_drives())) else {}
    temps = drives_temp() if (want("disk_temp") or any(
        want(f"disk_{d}_temp") for d in physical_drives())) else {}
    agg_u, agg_t = None, None
    for d, u in usage.items():
        if want(f"disk_{d}"):
            values[f"disk_{d}"] = round(u, 3)
        agg_u = max(agg_u or 0.0, u)
    for d, t in temps.items():
        if want(f"disk_{d}_temp"):
            values[f"disk_{d}_temp"] = round(t, 1)
        agg_t = max(agg_t or 0.0, t)
    if want("disk") and agg_u is not None:
        values["disk"] = round(agg_u, 3)
    if want("disk_temp") and agg_t is not None:
        values["disk_temp"] = round(agg_t, 1)
    return values


class NotificationWatcher(threading.Thread):
    """Fires `callback()` on any desktop notification (Linux, dbus-monitor)."""

    MATCH = "interface='org.freedesktop.Notifications',member='Notify'"

    def __init__(self, callback):
        super().__init__(daemon=True)
        self.callback = callback
        self.proc = None
        self._stop = threading.Event()

    def run(self):
        if not shutil.which("dbus-monitor"):
            return
        try:
            self.proc = subprocess.Popen(
                ["dbus-monitor", self.MATCH],
                stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
            for line in self.proc.stdout:
                if self._stop.is_set():
                    break
                if "member=Notify" in line:
                    try:
                        self.callback()
                    except Exception:
                        pass
        except OSError:
            pass

    def stop(self):
        self._stop.set()
        if self.proc:
            try:
                self.proc.terminate()
            except OSError:
                pass


if __name__ == "__main__":
    import json

    psutil.cpu_percent(interval=None)
    time.sleep(0.3)
    print("sources:", discover_sources())
    print(json.dumps(collect(), indent=2))