ProxMenux/AppImage/scripts/temperature_sensor_resolver.py

#!/usr/bin/env python3

import os
import re
import time
from pathlib import Path


_TOPOLOGY_CACHE = {
    "key": None,
    "time": 0.0,
    "entries": [],
}
_TOPOLOGY_CACHE_TTL = 60


def _read_text(path):
    try:
        return Path(path).read_text(encoding="utf-8", errors="replace").strip()
    except (OSError, ValueError):
        return ""


def _read_temperature(path):
    value = _read_text(path)
    if not value:
        return 0.0
    try:
        return float(value) / 1000.0
    except ValueError:
        return 0.0


def _related_path_distance(first, second):
    first_parts = Path(first).parts
    second_parts = Path(second).parts
    common = 0
    for first_part, second_part in zip(first_parts, second_parts):
        if first_part != second_part:
            break
        common += 1
    if common == min(len(first_parts), len(second_parts)):
        return (len(first_parts) - common) + (len(second_parts) - common)
    return None


def _block_devices_for_hwmon(hwmon_device, block_root):
    exact = []
    related = []
    try:
        block_entries = sorted(Path(block_root).iterdir(), key=lambda item: item.name)
    except OSError:
        return []

    hwmon_real = os.path.realpath(hwmon_device)
    for block_entry in block_entries:
        block_device = block_entry / "device"
        if not block_device.exists():
            continue
        block_real = os.path.realpath(block_device)
        if block_real == hwmon_real:
            exact.append(block_entry.name)
            continue
        distance = _related_path_distance(hwmon_real, block_real)
        if distance is not None:
            related.append((distance, block_entry.name))

    if exact:
        return exact
    if not related:
        return []

    minimum_distance = min(distance for distance, _ in related)
    return [name for distance, name in related if distance == minimum_distance]


def _device_metadata(device_name, hwmon_kind, block_root):
    block_path = Path(block_root) / device_name
    rotational = _read_text(block_path / "queue" / "rotational")
    if hwmon_kind == "nvme":
        sensor_type = "nvme"
    elif rotational == "1":
        sensor_type = "hdd"
    elif rotational == "0":
        sensor_type = "ssd"
    else:
        sensor_type = "storage"

    return {
        "type": sensor_type,
        "model": _read_text(block_path / "device" / "model"),
        "serial": _read_text(block_path / "device" / "serial"),
    }


def _build_topology(hwmon_root, block_root):
    entries = []
    try:
        hwmon_entries = sorted(Path(hwmon_root).glob("hwmon*"), key=lambda item: item.name)
    except OSError:
        return entries

    for hwmon_path in hwmon_entries:
        hwmon_kind = _read_text(hwmon_path / "name").lower()
        if hwmon_kind not in {"nvme", "drivetemp"}:
            continue

        devices = _block_devices_for_hwmon(hwmon_path / "device", block_root)
        device = devices[0] if devices else ""
        metadata = _device_metadata(device, hwmon_kind, block_root) if device else {
            "type": "nvme" if hwmon_kind == "nvme" else "storage",
            "model": "",
            "serial": "",
        }
        entries.append({
            "hwmon_path": str(hwmon_path),
            "kind": hwmon_kind,
            "device": device,
            "devices": devices,
            **metadata,
        })
    return entries


def _get_topology(hwmon_root, block_root, cache_ttl):
    cache_key = (os.path.realpath(hwmon_root), os.path.realpath(block_root))
    now = time.monotonic()
    if (
        _TOPOLOGY_CACHE["key"] == cache_key
        and now - _TOPOLOGY_CACHE["time"] < cache_ttl
    ):
        return _TOPOLOGY_CACHE["entries"]

    entries = _build_topology(hwmon_root, block_root)
    _TOPOLOGY_CACHE.update({
        "key": cache_key,
        "time": now,
        "entries": entries,
    })
    return entries


def clear_temperature_topology_cache():
    _TOPOLOGY_CACHE.update({"key": None, "time": 0.0, "entries": []})


def _temperature_inputs(entry):
    hwmon_path = Path(entry["hwmon_path"])
    inputs = sorted(hwmon_path.glob("temp*_input"), key=lambda item: item.name)
    if entry["kind"] == "drivetemp":
        preferred = [item for item in inputs if item.name == "temp1_input"]
        return preferred or inputs[:1]

    selected = []
    for input_path in inputs:
        match = re.fullmatch(r"temp(\d+)_input", input_path.name)
        if not match:
            continue
        index = match.group(1)
        label = _read_text(hwmon_path / f"temp{index}_label")
        if label.lower() == "composite" or (not label and index == "1"):
            selected.append(input_path)
    return selected


def get_storage_temperatures(
    hwmon_root="/sys/class/hwmon",
    block_root="/sys/block",
    cache_ttl=_TOPOLOGY_CACHE_TTL,
):
    temperatures = []
    expected_by_kind = {}
    resolved_by_kind = {}

    for entry in _get_topology(hwmon_root, block_root, cache_ttl):
        inputs = _temperature_inputs(entry)
        if not inputs:
            continue
        expected_by_kind[entry["kind"]] = expected_by_kind.get(entry["kind"], 0) + 1

        input_path = inputs[0]
        match = re.fullmatch(r"temp(\d+)_input", input_path.name)
        if not match:
            continue
        index = match.group(1)
        current = _read_temperature(input_path)
        if current == 0.0 and not _read_text(input_path):
            # Sensor present but no readable temperature (e.g. an NVMe in
            # low power state that leaves temp1_input empty). Mark the
            # kind as covered anyway so the lm-sensors fallback does not
            # add a duplicate entry for the same hwmon device.
            resolved_by_kind[entry["kind"]] = resolved_by_kind.get(entry["kind"], 0) + 1
            continue

        sensor_type = entry["type"]
        if sensor_type == "nvme":
            name = "NVMe SSD"
            adapter = "PCI adapter"
        elif sensor_type == "hdd":
            name = "HDD"
            adapter = "SCSI adapter"
        elif sensor_type == "ssd":
            name = "SSD"
            adapter = "SCSI adapter"
        else:
            name = "Storage device"
            adapter = "SCSI adapter"

        label = _read_text(Path(entry["hwmon_path"]) / f"temp{index}_label")
        temperatures.append({
            "name": name,
            "original_name": label or f"temp{index}",
            "current": current,
            "high": _read_temperature(Path(entry["hwmon_path"]) / f"temp{index}_max"),
            "critical": _read_temperature(Path(entry["hwmon_path"]) / f"temp{index}_crit"),
            "adapter": adapter,
            "type": sensor_type,
            "device": entry["device"],
            "devices": entry["devices"],
            "model": entry["model"],
            "serial": entry["serial"],
        })
        resolved_by_kind[entry["kind"]] = resolved_by_kind.get(entry["kind"], 0) + 1

    covered_kinds = {
        kind
        for kind, expected in expected_by_kind.items()
        if resolved_by_kind.get(kind, 0) == expected
    }
    temperatures.sort(key=lambda item: (item["type"], item["device"], item["name"]))
    return temperatures, covered_kinds