ARIA
6cb33187d3
The fan curve previously only controlled fan index 0; secondary fans
stayed on driver control. The curve can now target all fans (new
default) or any individual fan(s).
Backend:
- hal/fans.py: get_num_fans() via nvmlDeviceGetNumFans; get_fan_info()
returns per-fan speeds; set_fan_speed() accepts a fan index list
(None = all fans; all-fans mode is lenient toward driver-locked
fans, explicit lists are strict); reset_fan() restores all fans.
- server.py: per-GPU fan_targets state; the poller applies the curve to
all target fans and logs write failures (once per distinct error);
activation validates targets against the hardware (stale indices fall
back to all fans); POST /api/fans accepts fans, POST /api/fans/speed
accepts a fan index, GET /api/fans returns num_fans/fans/fan_targets.
- Persistence format is now {"curve": ..., "fans": ...}; legacy
bare-curve entries migrate to "all fans" at startup.
- Profiles save/apply fan_targets alongside fan_curve.
- MonitoringSample carries per-fan speeds for live gauges.
Frontend:
- Fans tab: All / Fan 1 / Fan 2 / ... selector with live per-fan %;
the selection is applied together with the curve.
- Live monitor: per-fan gauges with sparklines for multi-fan GPUs.
311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""Hardware Abstraction Layer for Fan Control.
|
|
|
|
Uses NVML (via pynvml) for all operations:
|
|
- nvmlDeviceGetNumFans : number of fans on the device
|
|
- nvmlDeviceGetFanSpeed_v2 : read current fan speed % for a fan index
|
|
- nvmlDeviceSetFanSpeed_v2 : set fan speed % for a fan index
|
|
- nvmlDeviceSetDefaultFanSpeed_v2 : restore automatic control for a fan index
|
|
- nvmlDeviceGetMinMaxFanSpeed : get min/max fan speed constraints
|
|
- nvmlDeviceGetTemperature : read GPU temp for curve interpolation
|
|
|
|
Fans are addressed by 0-based index. Passing ``fans=None`` to the set/reset
|
|
helpers means "all fans on the device".
|
|
"""
|
|
|
|
import ctypes
|
|
import logging
|
|
from typing import Any
|
|
|
|
try:
|
|
import pynvml as _pynvml_import
|
|
|
|
_NVML_AVAILABLE = True
|
|
except ImportError:
|
|
_pynvml_import = None
|
|
_NVML_AVAILABLE = False
|
|
|
|
# Aliased as Any so attribute access is not flagged when the import failed.
|
|
pynvml: Any = _pynvml_import
|
|
|
|
log = logging.getLogger("nvcurve.hal.fans")
|
|
|
|
|
|
def _get_handle(gpu_index: int):
|
|
"""Return an NVML device handle."""
|
|
if not _NVML_AVAILABLE:
|
|
raise RuntimeError("NVML not available (install nvidia-ml-py)")
|
|
return pynvml.nvmlDeviceGetHandleByIndex(gpu_index)
|
|
|
|
|
|
def _num_fans(handle) -> int:
|
|
"""Return the number of fans on the device (>= 1 on query failure)."""
|
|
try:
|
|
return max(0, int(pynvml.nvmlDeviceGetNumFans(handle)))
|
|
except pynvml.NVMLError:
|
|
# GetNumFans unsupported: assume at least the primary fan exists.
|
|
return 1
|
|
|
|
|
|
def get_num_fans(gpu_index: int = 0) -> int:
|
|
"""Return the number of fans on the GPU (0 if NVML is unavailable)."""
|
|
if not _NVML_AVAILABLE:
|
|
return 0
|
|
try:
|
|
return _num_fans(_get_handle(gpu_index))
|
|
except pynvml.NVMLError as exc:
|
|
log.warning("get_num_fans: %s", exc)
|
|
return 0
|
|
|
|
|
|
def get_fan_info(gpu_index: int = 0) -> dict:
|
|
"""Return current fan state for all fans.
|
|
|
|
Returns a dict with:
|
|
fan_pct : current speed % of fan 0 (legacy, None on failure)
|
|
fans : [{"index": i, "fan_pct": pct | None}, ...] per fan
|
|
num_fans : number of fans on the device
|
|
fan_mode : None (the server derives "auto"/"curve")
|
|
min_fan_pct / max_fan_pct : device-wide speed constraints
|
|
|
|
Returns None values on failure.
|
|
"""
|
|
out: dict[str, Any] = {
|
|
"fan_pct": None,
|
|
"fans": [],
|
|
"num_fans": 0,
|
|
"fan_mode": None,
|
|
"min_fan_pct": None,
|
|
"max_fan_pct": None,
|
|
}
|
|
if not _NVML_AVAILABLE:
|
|
return out
|
|
try:
|
|
handle = _get_handle(gpu_index)
|
|
except pynvml.NVMLError as exc:
|
|
log.warning("get_fan_info: %s", exc)
|
|
return out
|
|
|
|
out["num_fans"] = _num_fans(handle)
|
|
|
|
# Per-fan speeds via the v2 API; fan 0 falls back to the legacy v1 API.
|
|
for i in range(out["num_fans"]):
|
|
pct: float | None = None
|
|
try:
|
|
pct = float(pynvml.nvmlDeviceGetFanSpeed_v2(handle, i))
|
|
except pynvml.NVMLError:
|
|
if i == 0:
|
|
try:
|
|
pct = float(pynvml.nvmlDeviceGetFanSpeed(handle))
|
|
except pynvml.NVMLError:
|
|
pct = None
|
|
out["fans"].append({"index": i, "fan_pct": pct})
|
|
|
|
out["fan_pct"] = out["fans"][0]["fan_pct"] if out["fans"] else None
|
|
|
|
# Get min/max fan speed constraints
|
|
try:
|
|
min_s = ctypes.c_uint(0)
|
|
max_s = ctypes.c_uint(0)
|
|
pynvml.nvmlDeviceGetMinMaxFanSpeed(handle, min_s, max_s)
|
|
out["min_fan_pct"] = int(min_s.value)
|
|
out["max_fan_pct"] = int(max_s.value)
|
|
except pynvml.NVMLError:
|
|
out["min_fan_pct"] = 0
|
|
out["max_fan_pct"] = 100
|
|
|
|
return out
|
|
|
|
|
|
def set_fan_speed(
|
|
gpu_index: int,
|
|
pct: int,
|
|
fans: list[int] | None = None,
|
|
) -> tuple[bool, str]:
|
|
"""Set fan speed to a percentage (0-100).
|
|
|
|
fans=None targets every fan on the device; fans=[0, 1] targets the
|
|
listed fan indices. In all-fans mode, fans the driver does not allow
|
|
manual control of (e.g. driver-mirrored secondary fans) are skipped
|
|
with a warning instead of failing the whole operation; explicit fan
|
|
lists are strict and fail if any selected fan cannot be set.
|
|
"""
|
|
try:
|
|
pct = max(0, min(100, int(pct)))
|
|
except (TypeError, ValueError):
|
|
return False, "Invalid fan speed"
|
|
if not _NVML_AVAILABLE:
|
|
return False, "NVML not available"
|
|
try:
|
|
handle = _get_handle(gpu_index)
|
|
num_fans = _num_fans(handle)
|
|
|
|
if fans is None:
|
|
targets = list(range(num_fans))
|
|
strict = False
|
|
else:
|
|
targets: list[int] = []
|
|
for f in fans:
|
|
try:
|
|
f = int(f)
|
|
except (TypeError, ValueError):
|
|
return False, f"Invalid fan index: {f!r}"
|
|
if f < 0 or f >= num_fans:
|
|
return False, f"Fan index {f} out of range (0-{num_fans - 1})"
|
|
targets.append(f)
|
|
strict = True
|
|
if not targets:
|
|
return False, "No fans selected"
|
|
|
|
if not targets:
|
|
return False, "No fans available on this GPU"
|
|
|
|
skipped: list[str] = []
|
|
for i in targets:
|
|
try:
|
|
pynvml.nvmlDeviceSetFanSpeed_v2(handle, i, pct)
|
|
except pynvml.NVMLError as exc:
|
|
if strict:
|
|
log.warning(
|
|
"set_fan_speed(%d, fan %d, %d): %s", gpu_index, i, pct, exc
|
|
)
|
|
return False, str(exc)
|
|
# All-fans mode: secondary fans may be driver-controlled and
|
|
# reject manual writes; skip them and report in the message.
|
|
log.debug("set_fan_speed: fan %d not settable: %s", i, exc)
|
|
skipped.append(f"fan {i + 1}")
|
|
if skipped:
|
|
return (
|
|
True,
|
|
f"OK ({len(skipped)} fan(s) not manually controllable: {', '.join(skipped)})",
|
|
)
|
|
return True, "OK"
|
|
except pynvml.NVMLError as exc:
|
|
log.warning("set_fan_speed(%d, %d): %s", gpu_index, pct, exc)
|
|
return False, str(exc)
|
|
|
|
|
|
def reset_fan(gpu_index: int = 0) -> tuple[bool, str]:
|
|
"""Restore automatic fan control for all fans.
|
|
|
|
Tries nvidia-smi --fan=default first (most reliable, resets all fans on
|
|
the device), then falls back to NVML nvmlDeviceSetDefaultFanSpeed_v2
|
|
per fan index.
|
|
"""
|
|
if not _NVML_AVAILABLE:
|
|
return False, "NVML not available"
|
|
|
|
import subprocess
|
|
|
|
# Try nvidia-smi approach first (most reliable for restoring auto)
|
|
try:
|
|
ret = subprocess.run(
|
|
["nvidia-smi", "-i", str(gpu_index), "-fan", "default"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
if ret.returncode == 0:
|
|
return True, "OK"
|
|
log.debug("nvidia-smi -fan default failed: %s", ret.stderr.strip())
|
|
except FileNotFoundError:
|
|
log.debug("nvidia-smi not found, falling back to NVML")
|
|
except subprocess.TimeoutExpired:
|
|
log.warning("nvidia-smi -fan default timed out")
|
|
except Exception as exc:
|
|
log.debug("nvidia-smi -fan default error: %s", exc)
|
|
|
|
# Fallback: use NVML to reset every fan to default speed
|
|
try:
|
|
handle = _get_handle(gpu_index)
|
|
for i in range(_num_fans(handle)):
|
|
try:
|
|
pynvml.nvmlDeviceSetDefaultFanSpeed_v2(handle, i)
|
|
except pynvml.NVMLError as exc:
|
|
log.debug("reset_fan: fan %d: %s", i, exc)
|
|
return True, "OK"
|
|
except pynvml.NVMLError as exc:
|
|
return False, f"Failed to reset fan: {exc}"
|
|
|
|
|
|
def get_temp(gpu_index: int = 0) -> float | None:
|
|
"""Read current GPU temperature in °C."""
|
|
if not _NVML_AVAILABLE:
|
|
return None
|
|
try:
|
|
handle = _get_handle(gpu_index)
|
|
return float(
|
|
pynvml.nvmlDeviceGetTemperature(handle, pynvml.NVML_TEMPERATURE_GPU)
|
|
)
|
|
except pynvml.NVMLError as exc:
|
|
log.debug("get_temp: %s", exc)
|
|
return None
|
|
|
|
|
|
def interpolate_fan_speed(curve: list[dict], temp_c: float) -> int | None:
|
|
"""Interpolate target fan speed from a curve at a given temperature.
|
|
|
|
curve: list of {temp_c: int, fan_pct: int} sorted by temp_c
|
|
Returns fan_pct clamped to 0-100, or None if curve is empty.
|
|
"""
|
|
if not curve or len(curve) < 2:
|
|
return None
|
|
|
|
try:
|
|
temp = float(temp_c)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
# Find the two surrounding points
|
|
for i in range(len(curve) - 1):
|
|
t0, f0 = curve[i]["temp_c"], curve[i]["fan_pct"]
|
|
t1, f1 = curve[i + 1]["temp_c"], curve[i + 1]["fan_pct"]
|
|
|
|
if t0 == t1:
|
|
continue
|
|
|
|
if t0 <= temp <= t1:
|
|
fraction = (temp - t0) / (t1 - t0)
|
|
result = f0 + fraction * (f1 - f0)
|
|
try:
|
|
return max(0, min(100, int(round(result))))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
# Outside range: clamp to first or last point
|
|
if temp <= curve[0]["temp_c"]:
|
|
return max(0, min(100, curve[0]["fan_pct"]))
|
|
return max(0, min(100, curve[-1]["fan_pct"]))
|
|
|
|
|
|
def validate_curve(curve: list[dict]) -> tuple[bool, str]:
|
|
"""Validate a fan curve.
|
|
|
|
Returns (True, "OK") or (False, error_message).
|
|
"""
|
|
if not curve or len(curve) < 2:
|
|
return False, "Fan curve requires at least 2 points"
|
|
|
|
temps = [p["temp_c"] for p in curve]
|
|
speeds = [p["fan_pct"] for p in curve]
|
|
|
|
# Check for duplicate temperatures
|
|
if len(temps) != len(set(temps)):
|
|
return False, "Fan curve has duplicate temperature values"
|
|
|
|
# Check temperature range
|
|
for t in temps:
|
|
if t < 0 or t > 120:
|
|
return False, f"Temperature {t}°C out of range (0-120)"
|
|
|
|
# Check fan speed range
|
|
for s in speeds:
|
|
if s < 0 or s > 100:
|
|
return False, f"Fan speed {s}% out of range (0-100)"
|
|
|
|
# Check sorted by temperature
|
|
for i in range(len(temps) - 1):
|
|
if temps[i] >= temps[i + 1]:
|
|
return False, "Fan curve points must be sorted by ascending temperature"
|
|
|
|
return True, "OK"
|