feat: add fan curve control via temperature-based fan speed curves
- Backend HAL (hal/fans.py): NVML v2 fan read/set/reset with min/max queries - Server endpoints: GET/POST /api/fans, POST /api/fans/reset, POST /api/fans/speed - Background fan poller: reads GPU temp every 2s, interpolates fan speed from curve - Profile integration: fan_curve field saved/applied, auto-restore on shutdown - Frontend: FanCurveEditor (SVG chart with drag/add/delete points), FanMonitor sidebar - App.tsx: three-tab layout (Curve, Performance, Fans) - GaugeCard: optional history sparkline, Fan Mode card without sparkline - fan_mode field populated as 'curve' or 'auto' in GET /api/fans
This commit is contained in:
ARIA committed
1 parent
7605d21e27
commit
477c4becae
11 files changed
+1288
-10
No files matched your search
@@ -0,0 +1,198 @@
|
||||
"""Hardware Abstraction Layer for Fan Control.
|
||||
|
||||
Uses NVML (via pynvml) for all operations:
|
||||
- nvmlDeviceGetFanSpeed_v2 : read current fan speed % for a fan index
|
||||
- nvmlDeviceSetFanSpeed_v2 : set fan speed % for a fan index
|
||||
- nvmlDeviceGetMinMaxFanSpeed: get min/max fan speed constraints
|
||||
- nvmlDeviceGetTemperature : read GPU temp for curve interpolation
|
||||
"""
|
||||
|
||||
import ctypes
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
try:
|
||||
import pynvml
|
||||
_NVML_AVAILABLE = True
|
||||
except ImportError:
|
||||
_NVML_AVAILABLE = False
|
||||
|
||||
log = logging.getLogger("nvcurve.hal.fans")
|
||||
|
||||
# We use fan index 0 (first/primary fan) for all operations.
|
||||
_FAN_INDEX = 0
|
||||
|
||||
|
||||
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 get_fan_info(gpu_index: int = 0) -> dict:
|
||||
"""Return current fan state: fan_pct, fan_mode, min_fan_pct, max_fan_pct.
|
||||
|
||||
Returns None values on failure.
|
||||
"""
|
||||
out = {
|
||||
"fan_pct": None,
|
||||
"fan_mode": None,
|
||||
"min_fan_pct": None,
|
||||
"max_fan_pct": None,
|
||||
}
|
||||
if not _NVML_AVAILABLE:
|
||||
return out
|
||||
try:
|
||||
handle = _get_handle(gpu_index)
|
||||
|
||||
# Get current fan speed using v2 API (fan index 0)
|
||||
try:
|
||||
out["fan_pct"] = float(pynvml.nvmlDeviceGetFanSpeed_v2(handle, _FAN_INDEX))
|
||||
except pynvml.NVMLError:
|
||||
# Fallback to legacy v1 API
|
||||
try:
|
||||
out["fan_pct"] = float(pynvml.nvmlDeviceGetFanSpeed(handle))
|
||||
except pynvml.NVMLError:
|
||||
pass
|
||||
|
||||
# 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
|
||||
|
||||
except pynvml.NVMLError as exc:
|
||||
log.warning("get_fan_info: %s", exc)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def set_fan_speed(gpu_index: int, pct: int) -> tuple[bool, str]:
|
||||
"""Set fan speed to a percentage (0-100) on the primary fan."""
|
||||
pct = max(0, min(100, int(pct)))
|
||||
if not _NVML_AVAILABLE:
|
||||
return False, "NVML not available"
|
||||
try:
|
||||
handle = _get_handle(gpu_index)
|
||||
pynvml.nvmlDeviceSetFanSpeed_v2(handle, _FAN_INDEX, pct)
|
||||
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.
|
||||
|
||||
Tries nvidia-smi --fan=default first (most reliable), then falls back to
|
||||
NVML nvmlDeviceSetDefaultFanSpeed_v2.
|
||||
"""
|
||||
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 to default fan speed
|
||||
try:
|
||||
handle = _get_handle(gpu_index)
|
||||
pynvml.nvmlDeviceSetDefaultFanSpeed_v2(handle, _FAN_INDEX)
|
||||
return True, "OK"
|
||||
except pynvml.NVMLError as exc:
|
||||
return False, f"Failed to reset fan: {exc}"
|
||||
|
||||
|
||||
def get_temp(gpu_index: int = 0) -> Optional[float]:
|
||||
"""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) -> Optional[int]:
|
||||
"""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
|
||||
|
||||
temp = float(temp_c)
|
||||
|
||||
# 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)
|
||||
return max(0, min(100, int(round(result))))
|
||||
|
||||
# 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"
|
||||
Reference in new issue
Block a user