nvcurve with some fixes and better limits

nvcurve/__init__.py

@@ -0,0 +1,3 @@
+"""nvcurve — Linux GPU V/F curve editor for NVIDIA GPUs."""
+
+__version__ = "0.5.0"

nvcurve/__main__.py

@@ -0,0 +1,10 @@
+"""Allow running as: python -m nvcurve"""
+
+import sys
+
+if len(sys.argv) > 1 and sys.argv[1] == "daemon":
+    from .daemon import run
+    run()
+else:
+    from .cli import main
+    main()

nvcurve/client.py

@@ -0,0 +1,148 @@
+"""HTTP client for communicating with a running nvcurve server."""
+
+import httpx
+from typing import Any
+
+DEFAULT_BASE = "http://127.0.0.1:8042"
+_TIMEOUT = 5.0
+
+
+class ServerNotRunning(Exception):
+    """Raised when the nvcurve server cannot be reached."""
+
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: Any):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(f"HTTP {status_code}: {detail}")
+
+
+class NvCurveClient:
+    def __init__(self, base: str = DEFAULT_BASE, gpu_index: int = 0):
+        self._base = base.rstrip("/")
+        self.gpu_index = gpu_index
+
+    def _url(self, path: str) -> str:
+        sep = "&" if "?" in path else "?"
+        return f"{self._base}{path}{sep}gpu_index={self.gpu_index}"
+
+    def _raise(self, r: httpx.Response) -> None:
+        if r.is_error:
+            try:
+                detail = r.json().get("detail", r.text)
+            except Exception:
+                detail = r.text
+            raise ApiError(r.status_code, detail)
+
+    def _get(self, path: str) -> Any:
+        try:
+            r = httpx.get(self._url(path), timeout=_TIMEOUT)
+        except httpx.ConnectError:
+            raise ServerNotRunning()
+        self._raise(r)
+        return r.json()
+
+    def _post(self, path: str, body: Any = None) -> Any:
+        try:
+            r = httpx.post(self._url(path), json=body, timeout=_TIMEOUT)
+        except httpx.ConnectError:
+            raise ServerNotRunning()
+        self._raise(r)
+        return r.json()
+
+    def _delete(self, path: str) -> Any:
+        try:
+            r = httpx.delete(self._url(path), timeout=_TIMEOUT)
+        except httpx.ConnectError:
+            raise ServerNotRunning()
+        self._raise(r)
+        return r.json()
+
+    def ping(self) -> bool:
+        try:
+            httpx.get(self._url("/api/gpu"), timeout=1.0)
+            return True
+        except Exception:
+            return False
+
+    def gpus(self) -> list:
+        return self._get("/api/gpus")
+
+    # ── GPU ──────────────────────────────────────────────────────────────────
+
+    def gpu(self) -> dict:
+        return self._get("/api/gpu")
+
+    # ── Curve ────────────────────────────────────────────────────────────────
+
+    def curve(self) -> dict:
+        """Returns {gpu_name, timestamp, points: [{index, freq_khz, volt_uv, delta_khz, ...}]}"""
+        return self._get("/api/curve")
+
+    def voltage(self) -> int | None:
+        """Returns current GPU voltage in µV, or None if unavailable."""
+        try:
+            return self._get("/api/voltage")["voltage_uv"]
+        except Exception:
+            return None
+
+    def write_curve(
+        self,
+        deltas: dict[int, int],
+        max_delta_khz: int | None = None,
+    ) -> dict:
+        body: dict = {"deltas": deltas}
+        if max_delta_khz is not None:
+            body["max_delta_khz"] = max_delta_khz
+        return self._post("/api/curve/write", body)
+
+    def write_global(self, delta_khz: int, max_delta_khz: int | None = None) -> dict:
+        body: dict = {"delta_khz": delta_khz}
+        if max_delta_khz is not None:
+            body["max_delta_khz"] = max_delta_khz
+        return self._post("/api/curve/write/global", body)
+
+    def reset_curve(self) -> dict:
+        return self._post("/api/curve/reset")
+
+    def verify_write(self, deltas: dict[int, int]) -> dict:
+        return self._post("/api/curve/verify", {"deltas": deltas})
+
+    # ── Snapshots ────────────────────────────────────────────────────────────
+
+    def snapshot_save(self) -> dict:
+        return self._post("/api/snapshot/save")
+
+    def snapshot_restore(self, filepath: str | None = None) -> dict:
+        return self._post("/api/snapshot/restore", {"filepath": filepath})
+
+    def snapshots(self) -> list:
+        return self._get("/api/snapshots")
+
+    # ── Profiles ─────────────────────────────────────────────────────────────
+
+    def profiles(self) -> dict:
+        return self._get("/api/profiles")
+
+    def profile_save(self, name: str) -> dict:
+        return self._post("/api/profiles", {"name": name})
+
+    def profile_apply(self, name: str) -> dict:
+        return self._post(f"/api/profiles/{name}/apply")
+
+    def profile_delete(self, name: str) -> dict:
+        return self._delete(f"/api/profiles/{name}")
+
+    # ── Config ───────────────────────────────────────────────────────────────
+
+    def config_get(self) -> dict:
+        return self._get("/api/config")
+
+    def config_update(self, auto_load_profile: str | None, gpu_index: int = 0) -> dict:
+        return self._post("/api/config", {"auto_load_profile": auto_load_profile, "gpu_index": gpu_index})
+
+    # ── Server control ───────────────────────────────────────────────────────
+
+    def shutdown(self) -> dict:
+        return self._post("/api/shutdown")

nvcurve/config.py

@@ -0,0 +1,31 @@
+"""User-configurable settings with sensible defaults."""
+
+import os
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Config:
+    # Safety limits
+    max_delta_khz: int = 3000_000        # ±3000 MHz hard cap
+    auto_snapshot: bool = True          # Save snapshot before every write
+    max_snapshots: int = 20             # Maximum snapshots to keep (0 = unlimited)
+
+    # Monitoring
+    poll_interval_s: float = 1.0        # WebSocket monitor poll rate
+
+    # API server
+    host: str = "127.0.0.1"
+    port: int = 8042
+
+    snapshot_dir: str = "/var/cache/nvcurve/snapshots"
+    profile_dir: str = "/etc/nvcurve/profiles"
+
+    # Per-GPU default profiles: applied automatically on server startup.
+    # Key = stable GPU identifier (UUID string, "pci:{bus_id}", or "idx:{n}" fallback).
+    # Value = profile name (str).
+    auto_load_profiles: dict[str, str] = field(default_factory=dict)
+
+
+# Module-level default config instance.
+default_config = Config()

nvcurve/daemon.py

@@ -0,0 +1,209 @@
+"""Minimal daemon: apply auto-load profiles on boot, manage web server via Unix socket.
+
+Run via:  nvcurve daemon
+Or via systemd:  nvcurve service install
+
+Protocol: newline-delimited JSON, one request → one response, connection closed.
+
+Commands:
+  {"cmd": "ping"}
+  {"cmd": "serve_start", "host": "127.0.0.1", "port": 8042}
+  {"cmd": "serve_stop"}
+  {"cmd": "serve_status"}
+
+Requires root.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import signal
+import subprocess
+import sys
+
+log = logging.getLogger("nvcurve.daemon")
+
+SOCKET_PATH = "/run/nvcurve-daemon.sock"
+_PERSISTENT_CONFIG_FILE = "/etc/nvcurve/config.json"
+
+# Global server subprocess — only touched from the asyncio event loop.
+_server_proc: subprocess.Popen | None = None
+_cfg = None  # Config instance, set in run()
+
+
+# ── Socket command handlers ────────────────────────────────────────────────────
+
+async def _handle_serve_start(host: str, port: int) -> dict:
+    global _server_proc
+    if _server_proc is not None and _server_proc.poll() is None:
+        return {"ok": False, "error": "web server already running", "pid": _server_proc.pid}
+
+    cmd = [sys.executable, "-m", "nvcurve", "serve", "start",
+           "--host", host, "--port", str(port), "--direct"]
+    log_path = "/var/log/nvcurve-server.log"
+    log.info("Starting web server on %s:%d (log: %s)", host, port, log_path)
+    with open(log_path, "a") as lf:
+        _server_proc = subprocess.Popen(
+            cmd,
+            stdout=lf,
+            stderr=lf,
+            env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
+        )
+    log.info("Web server started (PID %d)", _server_proc.pid)
+    return {"ok": True, "pid": _server_proc.pid}
+
+
+async def _handle_serve_stop() -> dict:
+    global _server_proc
+    if _server_proc is None or _server_proc.poll() is not None:
+        _server_proc = None
+        return {"ok": False, "error": "web server is not running"}
+    log.info("Stopping web server (PID %d)…", _server_proc.pid)
+    _server_proc.terminate()
+    try:
+        await asyncio.get_running_loop().run_in_executor(
+            None, _server_proc.wait, 5
+        )
+    except Exception:
+        _server_proc.kill()
+    _server_proc = None
+    return {"ok": True}
+
+
+async def _handle_serve_status() -> dict:
+    global _server_proc
+    if _server_proc is not None and _server_proc.poll() is None:
+        return {"ok": True, "running": True, "pid": _server_proc.pid}
+    _server_proc = None
+    return {"ok": True, "running": False}
+
+
+async def _dispatch(req: dict) -> dict:
+    cmd = req.get("cmd")
+    if cmd == "ping":
+        return {"ok": True}
+    elif cmd == "serve_start":
+        host = req.get("host", _cfg.host)
+        port = req.get("port", _cfg.port)
+        return await _handle_serve_start(host, port)
+    elif cmd == "serve_stop":
+        return await _handle_serve_stop()
+    elif cmd == "serve_status":
+        return await _handle_serve_status()
+    else:
+        return {"ok": False, "error": f"unknown command: {cmd!r}"}
+
+
+async def _handle_client(
+    reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+) -> None:
+    resp: dict = {"ok": False, "error": "internal error"}
+    try:
+        data = await asyncio.wait_for(reader.readline(), timeout=5.0)
+        req = json.loads(data)
+        resp = await _dispatch(req)
+    except asyncio.TimeoutError:
+        resp = {"ok": False, "error": "timeout reading request"}
+    except json.JSONDecodeError as exc:
+        resp = {"ok": False, "error": f"invalid JSON: {exc}"}
+    except Exception as exc:
+        resp = {"ok": False, "error": str(exc)}
+    finally:
+        try:
+            writer.write(json.dumps(resp).encode() + b"\n")
+            await writer.drain()
+        except Exception:
+            pass
+        writer.close()
+        try:
+            await writer.wait_closed()
+        except Exception:
+            pass
+
+
+# ── Entrypoint ─────────────────────────────────────────────────────────────────
+
+def _load_persistent_config() -> dict:
+    try:
+        with open(_PERSISTENT_CONFIG_FILE) as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def run() -> None:
+    """Run the nvcurve daemon: apply auto-load profiles, then serve the Unix socket."""
+    global _cfg
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    if os.geteuid() != 0:
+        print("nvcurve daemon: must run as root", file=sys.stderr)
+        sys.exit(1)
+
+    cfg_data = _load_persistent_config()
+
+    from .config import Config
+    _cfg = Config()
+    for key in ("max_delta_khz", "auto_snapshot", "max_snapshots",
+                "snapshot_dir", "profile_dir", "host", "port"):
+        if key in cfg_data:
+            setattr(_cfg, key, cfg_data[key])
+
+    # Apply auto-load profiles in a subprocess so the daemon process itself
+    # never loads NvAPI/NVML/HAL modules — keeps steady-state RSS low.
+    auto_load_profiles: dict = cfg_data.get("auto_load_profiles", {})
+    if auto_load_profiles:
+        log.info("Applying auto-load profiles…")
+        subprocess.run(
+            [sys.executable, "-m", "nvcurve", "autoload"],
+            check=False,
+        )
+
+    auto_serve: bool = cfg_data.get("auto_serve", False)
+    asyncio.run(_serve_socket(auto_serve=auto_serve))
+
+    log.info("Daemon stopped.")
+
+
+async def _serve_socket(auto_serve: bool = False) -> None:
+    global _server_proc
+
+    # Clean up stale socket from a previous (unclean) run.
+    if os.path.exists(SOCKET_PATH):
+        os.unlink(SOCKET_PATH)
+
+    server = await asyncio.start_unix_server(_handle_client, path=SOCKET_PATH)
+    os.chmod(SOCKET_PATH, 0o666)  # allow non-root CLI to connect
+    log.info("Daemon listening on %s", SOCKET_PATH)
+
+    if auto_serve:
+        log.info("auto_serve enabled — starting web server on boot")
+        await _handle_serve_start(_cfg.host, _cfg.port)
+
+    stop_event = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    loop.add_signal_handler(signal.SIGTERM, stop_event.set)
+    loop.add_signal_handler(signal.SIGINT, stop_event.set)
+
+    async with server:
+        await stop_event.wait()
+
+    # Gracefully stop the web server if it's running.
+    if _server_proc is not None and _server_proc.poll() is None:
+        log.info("Stopping web server (PID %d)…", _server_proc.pid)
+        _server_proc.terminate()
+        try:
+            await asyncio.wait_for(
+                asyncio.get_running_loop().run_in_executor(None, _server_proc.wait),
+                timeout=5,
+            )
+        except asyncio.TimeoutError:
+            _server_proc.kill()
+
+    if os.path.exists(SOCKET_PATH):
+        os.unlink(SOCKET_PATH)

nvcurve/hal/__init__.py

@@ -0,0 +1,13 @@
+from .gpu import get_gpu, discover_gpus
+from .vfcurve import read_curve, read_clock_offsets, write_offsets, reset_offsets
+from .monitoring import poll, read_voltage
+from .ranges import get_clock_ranges
+from .snapshot import save as snapshot_save, restore as snapshot_restore, list_snapshots
+
+__all__ = [
+    "get_gpu", "discover_gpus",
+    "read_curve", "read_clock_offsets", "write_offsets", "reset_offsets",
+    "poll", "read_voltage",
+    "get_clock_ranges",
+    "snapshot_save", "snapshot_restore", "list_snapshots",
+]

nvcurve/hal/gpu.py

@@ -0,0 +1,93 @@
+"""GPU discovery and initialization."""
+
+import ctypes
+import sys
+
+from ..nvapi.bootstrap import query_interface
+from ..nvapi.constants import FUNC
+from ..nvapi.types import GpuInfo
+
+
+def init_nvapi() -> None:
+    """Initialize NvAPI. Must be called before any GPU operations."""
+    init_fn = query_interface(FUNC["Initialize"], nargs=0)
+    if not init_fn or init_fn() != 0:
+        print("NvAPI_Initialize failed")
+        sys.exit(1)
+
+
+def enumerate_gpus() -> tuple[ctypes.Array, int]:
+    """Return (gpu_handles_array, count). Exits if no GPUs found."""
+    gpus = (ctypes.c_void_p * 64)()
+    ngpu = ctypes.c_int32()
+    query_interface(FUNC["EnumPhysicalGPUs"])(ctypes.byref(gpus), ctypes.byref(ngpu))
+    if ngpu.value == 0:
+        print("No NVIDIA GPUs found")
+        sys.exit(1)
+    return gpus, ngpu.value
+
+
+def get_gpu_name(gpu) -> str:
+    """Return the full name string for a GPU handle."""
+    name_buf = ctypes.create_string_buffer(256)
+    query_interface(FUNC["GetFullName"])(gpu, name_buf)
+    return name_buf.value.decode(errors="replace")
+
+
+def discover_gpus() -> list[GpuInfo]:
+    """Initialize NvAPI and NVML, and return a list of GpuInfo for all physical GPUs."""
+    init_nvapi()
+    gpus, count = enumerate_gpus()
+    infos = []
+
+    try:
+        import pynvml
+        pynvml.nvmlInit()
+        has_nvml = True
+    except Exception:
+        has_nvml = False
+
+    for i in range(count):
+        name = get_gpu_name(gpus[i])
+        uuid = None
+        pci_bus_id = None
+        if has_nvml:
+            try:
+                handle = pynvml.nvmlDeviceGetHandleByIndex(i)
+                uuid = pynvml.nvmlDeviceGetUUID(handle)
+                # NVML might return bytes
+                if isinstance(uuid, bytes):
+                    uuid = uuid.decode('utf-8', errors='ignore')
+                pci_info = pynvml.nvmlDeviceGetPciInfo(handle)
+                # Parse something like "00000000:01:00.0" -> bus is 1
+                if isinstance(pci_info.bus, bytes):
+                    pci_bus_id = int(pci_info.bus.decode('utf-8', errors='ignore'), 16)
+                else:
+                    pci_bus_id = pci_info.bus
+            except Exception:
+                pass
+        infos.append(GpuInfo(name=name, index=i, uuid=uuid, pci_bus_id=pci_bus_id))
+
+    if has_nvml:
+        try:
+            pynvml.nvmlShutdown()
+        except Exception:
+            pass
+
+    return infos
+
+
+def get_gpu(index: int = 0):
+    """Initialize NvAPI, enumerate GPUs, and return the handle for `index`.
+
+    Also returns the GPU name as a convenience.
+    Returns (handle, name).
+    """
+    init_nvapi()
+    gpus, count = enumerate_gpus()
+    if index >= count:
+        print(f"GPU index {index} out of range (found {count} GPU(s))")
+        sys.exit(1)
+    gpu = gpus[index]
+    name = get_gpu_name(gpu)
+    return gpu, name

nvcurve/hal/limits.py

@@ -0,0 +1,321 @@
+"""Hardware Abstraction Layer for Global Limits (Power, Clock Offsets).
+
+Uses NVML (via pynvml) for all operations.
+
+Clock offsets use nvmlDeviceSetClockOffsets / nvmlDeviceGetClockOffsets
+(introduced in driver 555.85).  The older per-domain functions
+(nvmlDeviceSet/GetGpcClkVfOffset, nvmlDeviceSet/GetMemClkVfOffset) are used
+as a fallback when the new API is unavailable or returns an error.
+set_clock_offsets accepts Optional values and only touches the domains
+that are explicitly specified, leaving others unchanged on hardware.
+"""
+
+import ctypes
+import subprocess
+import logging
+from typing import Optional
+
+try:
+    import pynvml
+    _NVML_AVAILABLE = True
+except ImportError:
+    _NVML_AVAILABLE = False
+
+log = logging.getLogger("nvcurve.hal.limits")
+
+# ── NVML library / handle helpers ─────────────────────────────────────────────
+
+_nvml_lib: Optional[ctypes.CDLL] = None
+
+
+def _nvml_cdll() -> ctypes.CDLL:
+    """Return a ctypes handle to libnvidia-ml, reusing pynvml's load if possible."""
+    global _nvml_lib
+    if _nvml_lib is not None:
+        return _nvml_lib
+    # Prefer to reuse the library already loaded by pynvml to avoid dlopen races.
+    for attr in ("nvml", "_nvml"):          # attribute name varies by pynvml version
+        mod = getattr(pynvml, attr, None)
+        lib = getattr(mod, "_lib", None) or getattr(mod, "_nvmlLib", None)
+        if lib is not None:
+            _nvml_lib = lib
+            return _nvml_lib
+    _nvml_lib = ctypes.CDLL("libnvidia-ml.so.1")
+    return _nvml_lib
+
+
+def _get_handle(gpu_index: int):
+    """Return an NVML device handle, initialising pynvml if needed."""
+    if not _NVML_AVAILABLE:
+        raise RuntimeError("NVML not available (install nvidia-ml-py)")
+    return pynvml.nvmlDeviceGetHandleByIndex(gpu_index)
+
+
+# ── Power limit ───────────────────────────────────────────────────────────────
+
+def get_power_limit(gpu_index: int = 0) -> dict:
+    """Return dict with power_limit_w, default_power_limit_w, min_power_limit_w, max_power_limit_w."""
+    out = {
+        "power_limit_w": None,
+        "default_power_limit_w": None,
+        "min_power_limit_w": None,
+        "max_power_limit_w": None,
+    }
+    try:
+        handle = _get_handle(gpu_index)
+        limit = pynvml.nvmlDeviceGetPowerManagementLimit(handle)
+        constrs = pynvml.nvmlDeviceGetPowerManagementLimitConstraints(handle)
+        out["power_limit_w"] = limit // 1000
+        out["min_power_limit_w"] = constrs[0] // 1000
+        out["max_power_limit_w"] = constrs[1] // 1000
+        try:
+            default = pynvml.nvmlDeviceGetPowerManagementDefaultLimit(handle)
+            out["default_power_limit_w"] = default // 1000
+        except Exception:
+            pass
+    except Exception as exc:
+        log.warning("get_power_limit: %s", exc)
+    return out
+
+
+def set_power_limit(limit_w: int, gpu_index: int = 0) -> tuple[bool, str]:
+    """Set the board power limit (Watts)."""
+    try:
+        handle = _get_handle(gpu_index)
+        pynvml.nvmlDeviceSetPowerManagementLimit(handle, limit_w * 1000)
+        return True, "OK"
+    except Exception as exc:
+        log.debug("NVML set_power_limit failed: %s — falling back to nvidia-smi", exc)
+
+    ret = subprocess.run(
+        ["nvidia-smi", "-i", str(gpu_index), "-pl", str(limit_w)],
+        capture_output=True, text=True,
+    )
+    if ret.returncode == 0:
+        return True, "OK"
+    return False, ret.stderr.strip() or ret.stdout.strip()
+
+
+# ── Clock offsets (GPC + memory) ──────────────────────────────────────────────
+
+# The correct struct layout (per NVML docs and driver 590.x headers):
+#
+#   typedef struct {
+#       unsigned int  version;          // nvmlClockOffset_v1
+#       nvmlClockType_t type;           // NVML_CLOCK_GRAPHICS (0) or NVML_CLOCK_MEM (2)
+#       nvmlPstates_t   pstate;         // NVML_PSTATE_0 (0)
+#       int             clockOffsetMHz;
+#   } nvmlClockOffset_t;
+#
+# nvmlDeviceSet/GetClockOffsets are called ONCE PER CLOCK DOMAIN.
+# pynvml (nvidia-ml-py ≥ 12) exposes c_nvmlClockOffset_t and nvmlClockOffset_v1
+# as ctypes objects; we use them when available and fall back to our own definition.
+
+class _ClockOffset(ctypes.Structure):
+    _fields_ = [
+        ("version",        ctypes.c_uint),
+        ("type",           ctypes.c_uint),   # nvmlClockType_t
+        ("pstate",         ctypes.c_uint),   # nvmlPstates_t
+        ("clockOffsetMHz", ctypes.c_int),
+    ]
+
+_CLOCK_OFFSET_VER = (1 << 24) | ctypes.sizeof(_ClockOffset)  # = 0x01000010 (16 bytes)
+
+# NVML clock-type constants (same values as pynvml).
+_NVML_CLOCK_GRAPHICS = 0
+_NVML_CLOCK_MEM      = 2
+
+
+def _make_clock_offset(clock_type: int, pstate: int = 0, offset_mhz: int = 0) -> ctypes.Structure:
+    """Return a populated nvmlClockOffset_t struct, using pynvml's type when available."""
+    if hasattr(pynvml, "c_nvmlClockOffset_t") and hasattr(pynvml, "nvmlClockOffset_v1"):
+        info = pynvml.c_nvmlClockOffset_t()
+        info.version        = pynvml.nvmlClockOffset_v1
+        info.type           = clock_type
+        info.pstate         = pstate
+        info.clockOffsetMHz = offset_mhz
+        return info
+    info = _ClockOffset()
+    info.version        = _CLOCK_OFFSET_VER
+    info.type           = clock_type
+    info.pstate         = pstate
+    info.clockOffsetMHz = offset_mhz
+    return info
+
+
+def _try_nvml_fn(name: str):
+    """Return a ctypes-callable for an NVML function, or None if not found."""
+    lib = _nvml_cdll()
+    try:
+        return getattr(lib, name)
+    except AttributeError:
+        return None
+
+
+def get_clock_offsets(gpu_index: int = 0) -> dict:
+    """Return GPC and memory clock offsets (MHz).
+
+    Keys: gpc_offset_mhz, mem_offset_mhz (both int or None on failure).
+    Calls nvmlDeviceGetClockOffsets once per clock domain (GRAPHICS, MEM).
+    """
+    out = {"gpc_offset_mhz": None, "mem_offset_mhz": None}
+    if not _NVML_AVAILABLE:
+        return out
+    try:
+        handle = _get_handle(gpu_index)
+
+        # Try pynvml wrapper first (nvidia-ml-py ≥ 12 exposes it correctly).
+        # Fall back to ctypes-direct if pynvml doesn't have it.
+        _pynvml_get = getattr(pynvml, "nvmlDeviceGetClockOffsets", None)
+        fn_get = _try_nvml_fn("nvmlDeviceGetClockOffsets") if _pynvml_get is None else None
+
+        used_new_api = False
+        for clock_type, key in ((_NVML_CLOCK_GRAPHICS, "gpc_offset_mhz"),
+                                 (_NVML_CLOCK_MEM,      "mem_offset_mhz")):
+            info = _make_clock_offset(clock_type, pstate=0)
+            try:
+                if _pynvml_get is not None:
+                    rc = _pynvml_get(handle, ctypes.byref(info))
+                elif fn_get is not None:
+                    rc = fn_get(handle, ctypes.byref(info))
+                else:
+                    break
+                if rc == 0:
+                    out[key] = int(info.clockOffsetMHz)
+                    used_new_api = True
+                else:
+                    log.debug("nvmlDeviceGetClockOffsets(type=%d) returned %d", clock_type, rc)
+            except Exception as exc:
+                log.debug("nvmlDeviceGetClockOffsets(type=%d): %s", clock_type, exc)
+
+        if used_new_api:
+            return out
+
+        # Deprecated per-domain fallback.
+        if hasattr(pynvml, "nvmlDeviceGetGpcClkVfOffset"):
+            try:
+                out["gpc_offset_mhz"] = int(pynvml.nvmlDeviceGetGpcClkVfOffset(handle))
+            except Exception as exc:
+                log.debug("nvmlDeviceGetGpcClkVfOffset: %s", exc)
+        if hasattr(pynvml, "nvmlDeviceGetMemClkVfOffset"):
+            try:
+                res = pynvml.nvmlDeviceGetMemClkVfOffset(handle)
+                out["mem_offset_mhz"] = int(res[0] if isinstance(res, (list, tuple)) else res)
+            except Exception as exc:
+                log.debug("nvmlDeviceGetMemClkVfOffset: %s", exc)
+
+    except Exception as exc:
+        log.warning("get_clock_offsets: %s", exc)
+    return out
+
+
+def set_clock_offsets(
+    gpc_offset_mhz: Optional[int] = None,
+    mem_offset_mhz: Optional[int] = None,
+    gpu_index: int = 0,
+) -> tuple[bool, str]:
+    """Set clock offsets (MHz) for the specified domains only.
+
+    Pass None for a domain to leave it untouched on hardware.
+    Calls nvmlDeviceSetClockOffsets once per requested domain (GRAPHICS, MEM).
+    Falls back to deprecated per-domain functions when the new API returns an
+    error (e.g. NVML_ERROR_DEPRECATED=25 on Blackwell with driver 590.x).
+    """
+    if gpc_offset_mhz is None and mem_offset_mhz is None:
+        return True, "OK"
+    if not _NVML_AVAILABLE:
+        return False, "NVML not available (install nvidia-ml-py)"
+    try:
+        handle = _get_handle(gpu_index)
+
+        domains = []
+        if gpc_offset_mhz is not None:
+            domains.append((_NVML_CLOCK_GRAPHICS, gpc_offset_mhz))
+        if mem_offset_mhz is not None:
+            domains.append((_NVML_CLOCK_MEM, mem_offset_mhz))
+
+        _pynvml_set = getattr(pynvml, "nvmlDeviceSetClockOffsets", None)
+        fn_set = _try_nvml_fn("nvmlDeviceSetClockOffsets") if _pynvml_set is None else None
+
+        if _pynvml_set is not None or fn_set is not None:
+            all_ok = True
+            for clock_type, offset in domains:
+                info = _make_clock_offset(clock_type, pstate=0, offset_mhz=offset)
+                try:
+                    rc = _pynvml_set(handle, ctypes.byref(info)) if _pynvml_set else fn_set(handle, ctypes.byref(info))
+                    if rc != 0:
+                        log.debug("nvmlDeviceSetClockOffsets(type=%d) returned %d — trying fallback", clock_type, rc)
+                        all_ok = False
+                        break
+                except Exception as exc:
+                    log.debug("nvmlDeviceSetClockOffsets(type=%d): %s — trying fallback", clock_type, exc)
+                    all_ok = False
+                    break
+            if all_ok:
+                return True, "OK"
+            # Non-zero rc (e.g. 25=DEPRECATED on Blackwell) — fall through to deprecated path.
+
+        # Deprecated per-domain fallback (works on Blackwell/driver 590.x).
+        errs = []
+        if gpc_offset_mhz is not None and hasattr(pynvml, "nvmlDeviceSetGpcClkVfOffset"):
+            try:
+                pynvml.nvmlDeviceSetGpcClkVfOffset(handle, gpc_offset_mhz)
+            except Exception as exc:
+                errs.append(f"GPC: {exc}")
+        if mem_offset_mhz is not None and hasattr(pynvml, "nvmlDeviceSetMemClkVfOffset"):
+            try:
+                pynvml.nvmlDeviceSetMemClkVfOffset(handle, mem_offset_mhz)
+            except Exception as exc:
+                errs.append(f"MEM: {exc}")
+        if errs:
+            return False, "; ".join(errs)
+        return True, "OK"
+
+    except Exception as exc:
+        log.warning("set_clock_offsets: %s", exc)
+        return False, str(exc)
+
+
+# ── Range queries ─────────────────────────────────────────────────────────────
+
+def get_mem_offset_range(gpu_index: int = 0) -> dict:
+    """Return the min/max allowed memory clock offset (MHz).
+
+    Keys: min_mem_offset_mhz, max_mem_offset_mhz.
+    Uses nvmlDeviceGetMemClkMinMaxVfOffset; falls back to observed RTX values.
+    """
+    # Observed RTX 5090 defaults (NvAPI GetClockBoostRanges says -1000/+3000).
+    out = {"min_mem_offset_mhz": -2000, "max_mem_offset_mhz": 3000}
+    if not _NVML_AVAILABLE:
+        return out
+    try:
+        handle = _get_handle(gpu_index)
+
+        if hasattr(pynvml, "nvmlDeviceGetMemClkMinMaxVfOffset"):
+            result = pynvml.nvmlDeviceGetMemClkMinMaxVfOffset(handle)
+            if isinstance(result, (list, tuple)) and len(result) >= 2:
+                out["min_mem_offset_mhz"] = int(result[0])
+                out["max_mem_offset_mhz"] = int(result[1])
+            else:
+                min_v = getattr(result, "minOffset", None)
+                max_v = getattr(result, "maxOffset", None)
+                if min_v is not None:
+                    out["min_mem_offset_mhz"] = int(min_v)
+                if max_v is not None:
+                    out["max_mem_offset_mhz"] = int(max_v)
+            return out
+
+        fn = _try_nvml_fn("nvmlDeviceGetMemClkMinMaxVfOffset")
+        if fn is not None:
+            min_v = ctypes.c_int(0)
+            max_v = ctypes.c_int(0)
+            rc = fn(handle, ctypes.byref(min_v), ctypes.byref(max_v))
+            if rc == 0:
+                out["min_mem_offset_mhz"] = int(min_v.value)
+                out["max_mem_offset_mhz"] = int(max_v.value)
+
+    except Exception as exc:
+        log.debug("get_mem_offset_range: %s", exc)
+    return out
+
+

nvcurve/hal/monitoring.py

@@ -0,0 +1,124 @@
+"""Live GPU monitoring.
+
+Voltage is read via NvAPI GetCurrentVoltage.
+Clock, temperature, power draw, and fan speed are read via NVML (nvidia-ml-py).
+"""
+
+import struct
+import time
+from typing import Optional
+
+from ..nvapi.bootstrap import nvcall
+from ..nvapi.constants import FUNC, VOLT_SIZE
+from ..nvapi.types import MonitoringSample
+
+try:
+    import pynvml as _pynvml
+    _NVML_AVAILABLE = True
+except ImportError:
+    _pynvml = None
+    _NVML_AVAILABLE = False
+
+_nvml_initialized = False
+
+
+def init_nvml() -> bool:
+    """Initialize NVML. Call once at startup. Returns True on success."""
+    global _nvml_initialized
+    if not _NVML_AVAILABLE:
+        return False
+    try:
+        _pynvml.nvmlInit()
+        _nvml_initialized = True
+        return True
+    except _pynvml.NVMLError:
+        return False
+
+
+def shutdown_nvml() -> None:
+    """Shut down NVML. Call at process exit."""
+    global _nvml_initialized
+    if _NVML_AVAILABLE and _nvml_initialized:
+        try:
+            _pynvml.nvmlShutdown()
+        except _pynvml.NVMLError:
+            pass
+        _nvml_initialized = False
+
+
+def get_driver_version() -> Optional[str]:
+    """Return the NVIDIA driver version string, or None if unavailable."""
+    if not (_NVML_AVAILABLE and _nvml_initialized):
+        return None
+    try:
+        return _pynvml.nvmlSystemGetDriverVersion()
+    except _pynvml.NVMLError:
+        return None
+
+
+def get_vram_total(gpu_index: int = 0) -> Optional[int]:
+    """Return total VRAM in bytes, or None if unavailable."""
+    if not (_NVML_AVAILABLE and _nvml_initialized):
+        return None
+    try:
+        handle = _pynvml.nvmlDeviceGetHandleByIndex(gpu_index)
+        return _pynvml.nvmlDeviceGetMemoryInfo(handle).total
+    except _pynvml.NVMLError:
+        return None
+
+
+def read_voltage(gpu) -> tuple[Optional[int], str]:
+    """Read current GPU core voltage in µV via NvAPI GetCurrentVoltage.
+
+    Returns (voltage_uV, "OK") or (None, error).
+    """
+    d, err = nvcall(FUNC["GetCurrentVoltage"], gpu, VOLT_SIZE, ver=1)
+    if not d:
+        return None, err
+    return struct.unpack_from("<I", d, 0x28)[0], "OK"
+
+
+def _nvml_read(gpu_index: int) -> dict:
+    """Read all NVML fields. Returns a dict with keys matching MonitoringSample fields."""
+    out = {
+        "clock_mhz": None, "temp_c": None, "power_w": None, "fan_pct": None,
+        "pstate": None, "mem_used_bytes": None, "mem_total_bytes": None,
+        "gpu_util_pct": None, "mem_util_pct": None, "mem_clock_mhz": None,
+    }
+    if not (_NVML_AVAILABLE and _nvml_initialized):
+        return out
+    try:
+        handle = _pynvml.nvmlDeviceGetHandleByIndex(gpu_index)
+
+        out["clock_mhz"] = float(_pynvml.nvmlDeviceGetClockInfo(handle, _pynvml.NVML_CLOCK_GRAPHICS))
+        out["mem_clock_mhz"] = float(_pynvml.nvmlDeviceGetClockInfo(handle, _pynvml.NVML_CLOCK_MEM))
+        out["temp_c"] = float(_pynvml.nvmlDeviceGetTemperature(handle, _pynvml.NVML_TEMPERATURE_GPU))
+        out["power_w"] = _pynvml.nvmlDeviceGetPowerUsage(handle) / 1000.0  # mW → W
+        out["pstate"] = int(_pynvml.nvmlDeviceGetPerformanceState(handle))
+
+        mem = _pynvml.nvmlDeviceGetMemoryInfo(handle)
+        out["mem_used_bytes"] = mem.used
+        out["mem_total_bytes"] = mem.total
+
+        util = _pynvml.nvmlDeviceGetUtilizationRates(handle)
+        out["gpu_util_pct"] = float(util.gpu)
+        out["mem_util_pct"] = float(util.memory)
+
+        try:
+            out["fan_pct"] = float(_pynvml.nvmlDeviceGetFanSpeed(handle))
+        except _pynvml.NVMLError:
+            pass
+    except _pynvml.NVMLError:
+        pass
+    return out
+
+
+def poll(gpu, gpu_index: int = 0) -> MonitoringSample:
+    """Read all available monitoring data and return a MonitoringSample."""
+    voltage_uv, _ = read_voltage(gpu)
+    nvml = _nvml_read(gpu_index)
+    return MonitoringSample(
+        timestamp=time.time(),
+        voltage_uv=voltage_uv,
+        **nvml,
+    )

nvcurve/hal/ranges.py

@@ -0,0 +1,31 @@
+"""Clock boost range queries."""
+
+import struct
+from typing import Optional
+
+from ..nvapi.bootstrap import nvcall
+from ..nvapi.constants import FUNC, RANGES_SIZE
+
+
+def get_clock_ranges(gpu) -> tuple[Optional[dict], str]:
+    """Read clock domain min/max offset ranges via GetClockBoostRanges.
+
+    Returns ({"num_domains": int, "domains": [[int, ...], ...]}, "OK")
+    or (None, error).
+
+    On RTX 5090: GPU core ±3000 MHz, memory -3000/+3000 MHz.
+    """
+    d, err = nvcall(FUNC["GetClockBoostRanges"], gpu, RANGES_SIZE, ver=1)
+    if not d:
+        return None, err
+
+    num = struct.unpack_from("<I", d, 4)[0]
+    domains = []
+    for i in range(min(num, 32)):
+        base = 0x08 + i * 0x48
+        if base + 0x48 > len(d):
+            break
+        words = [struct.unpack_from("<i", d, base + j)[0] for j in range(0, 0x48, 4)]
+        domains.append(words)
+
+    return {"num_domains": num, "domains": domains}, "OK"

nvcurve/hal/snapshot.py

@@ -0,0 +1,156 @@
+"""Save and restore ClockBoostTable snapshots to/from disk."""
+
+import ctypes
+import json
+import os
+import struct
+from datetime import datetime
+from typing import Optional
+
+from ..nvapi.bootstrap import nvcall_raw
+from ..nvapi.constants import FUNC, CT_SIZE, CT_BASE, CT_STRIDE, CT_DELTA_OFF, CT_POINTS
+from ..nvapi.types import SnapshotInfo
+from .vfcurve import read_clock_table_raw, get_boost_mask
+
+
+def save(gpu, gpu_name: str, snapshot_dir: str, max_snapshots: int = 0) -> Optional[str]:
+    """Save the current ClockBoostTable to disk.
+
+    Writes both a binary .bin file and a human-readable .json metadata file.
+    If max_snapshots > 0, deletes the oldest snapshots to stay within the limit.
+    Returns the binary filepath on success, or None on failure.
+    """
+    raw, err = read_clock_table_raw(gpu)
+    if not raw:
+        print(f"Failed to read ClockBoostTable: {err}")
+        return None
+
+    os.makedirs(snapshot_dir, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    bin_path = os.path.join(snapshot_dir, f"clock_boost_table_{ts}.bin")
+    meta_path = os.path.join(snapshot_dir, f"clock_boost_table_{ts}.json")
+
+    with open(bin_path, "wb") as f:
+        f.write(raw)
+
+    offsets = []
+    max_entries = (len(raw) - CT_BASE) // CT_STRIDE
+    for i in range(max_entries):
+        off = CT_BASE + i * CT_STRIDE + CT_DELTA_OFF
+        delta = struct.unpack_from("<i", raw, off)[0]
+        offsets.append(delta)
+
+    meta = {
+        "gpu": gpu_name,
+        "timestamp": datetime.now().isoformat(),
+        "file": bin_path,
+        "size": len(raw),
+        "offsets_kHz": offsets,
+        "nonzero_offsets": sum(1 for o in offsets if o != 0),
+    }
+    with open(meta_path, "w") as f:
+        json.dump(meta, f, indent=2)
+
+    print(f"Snapshot saved:")
+    print(f"  Binary:   {bin_path}")
+    print(f"  Metadata: {meta_path}")
+    print(f"  Size:     {len(raw)} bytes")
+    print(f"  Non-zero offsets: {meta['nonzero_offsets']}")
+
+    if max_snapshots > 0:
+        _prune_snapshots(snapshot_dir, max_snapshots)
+
+    return bin_path
+
+
+def _prune_snapshots(snapshot_dir: str, max_snapshots: int) -> None:
+    """Delete oldest snapshots (both .bin and .json) to stay within max_snapshots."""
+    bins = sorted(
+        f for f in os.listdir(snapshot_dir) if f.endswith(".bin")
+    )  # oldest first (lexicographic = chronological for our timestamp format)
+    excess = len(bins) - max_snapshots
+    for fname in bins[:excess]:
+        stem = fname[:-4]  # strip .bin
+        for ext in (".bin", ".json"):
+            try:
+                os.remove(os.path.join(snapshot_dir, stem + ext))
+            except OSError:
+                pass
+
+
+def restore(gpu, snapshot_dir: str, filepath: str = None) -> bool:
+    """Restore a ClockBoostTable snapshot from disk.
+
+    If no filepath is given, uses the most recent snapshot in snapshot_dir.
+    Returns True on success.
+    """
+    if filepath is None:
+        if not os.path.isdir(snapshot_dir):
+            print(f"No snapshots found in {snapshot_dir}")
+            return False
+        bins = sorted(
+            [f for f in os.listdir(snapshot_dir) if f.endswith(".bin")],
+            reverse=True,
+        )
+        if not bins:
+            print(f"No snapshot .bin files in {snapshot_dir}")
+            return False
+        filepath = os.path.join(snapshot_dir, bins[0])
+
+    if not os.path.isfile(filepath):
+        print(f"Snapshot file not found: {filepath}")
+        return False
+
+    with open(filepath, "rb") as f:
+        raw = f.read()
+
+    if len(raw) != CT_SIZE:
+        print(f"Snapshot size mismatch: expected {CT_SIZE}, got {len(raw)}")
+        return False
+
+    vw = struct.unpack_from("<I", raw, 0)[0]
+    expected_vw = (1 << 16) | CT_SIZE
+    if vw != expected_vw:
+        print(f"Version word mismatch: 0x{vw:08X} (expected 0x{expected_vw:08X})")
+        return False
+
+    buf = ctypes.create_string_buffer(CT_SIZE)
+    ctypes.memmove(buf, raw, CT_SIZE)
+
+    # Full mask for restore — write all points
+    mask, _ = get_boost_mask(gpu)
+    if mask:
+        for i in range(32):
+            buf[4 + i] = mask[i]
+
+    print(f"Restoring from: {filepath}")
+    ret, desc = nvcall_raw(FUNC["SetClockBoostTable"], gpu, buf)
+    print(f"SetClockBoostTable returned: {ret} ({desc})")
+    return ret == 0
+
+
+def list_snapshots(snapshot_dir: str) -> list[SnapshotInfo]:
+    """Return metadata for all snapshots in snapshot_dir, newest first."""
+    if not os.path.isdir(snapshot_dir):
+        return []
+
+    results = []
+    for fname in sorted(os.listdir(snapshot_dir), reverse=True):
+        if not fname.endswith(".json"):
+            continue
+        meta_path = os.path.join(snapshot_dir, fname)
+        try:
+            with open(meta_path) as f:
+                meta = json.load(f)
+            bin_path = meta.get("file", meta_path.replace(".json", ".bin"))
+            results.append(SnapshotInfo(
+                filepath=bin_path,
+                timestamp=meta.get("timestamp", ""),
+                gpu=meta.get("gpu", ""),
+                nonzero_offsets=meta.get("nonzero_offsets", 0),
+                size=meta.get("size", 0),
+            ))
+        except (json.JSONDecodeError, KeyError):
+            continue
+
+    return results

nvcurve/hal/vfcurve.py

@@ -0,0 +1,286 @@
+"""Read and write the GPU V/F curve via NvAPI."""
+
+import ctypes
+import struct
+from typing import Optional
+
+from ..nvapi.bootstrap import nvcall, nvcall_raw
+from ..nvapi.constants import (
+    FUNC,
+    VFP_SIZE, VFP_BASE, VFP_STRIDE,
+    CT_SIZE, CT_BASE, CT_STRIDE, CT_DELTA_OFF, CT_POINTS,
+)
+from ..nvapi.types import VFPoint, CurveState
+
+
+# ── Mask helpers ─────────────────────────────────────────────────────────────
+
+# The boost mask is a static property of the GPU/driver — it does not change
+# at runtime. Cache it per GPU handle to avoid redundant GetClockBoostMask
+# calls on every HAL operation.
+_boost_mask_cache: dict[int, bytes] = {}
+
+
+def get_boost_mask(gpu) -> tuple[Optional[bytes], str]:
+    """Read the canonical 32-byte clock boost mask from the driver.
+
+    The result is cached per GPU handle: subsequent calls return the cached
+    value without hitting the driver again.
+
+    Returns (mask_bytes, "OK") or (None, error).
+    """
+    if gpu in _boost_mask_cache:
+        return _boost_mask_cache[gpu], "OK"
+
+    from ..nvapi.constants import MASK_SIZE
+    def fill(b):
+        for i in range(4, 4 + 32):
+            b[i] = 0xFF
+    d, err = nvcall(FUNC["GetClockBoostMask"], gpu, MASK_SIZE, ver=1, pre_fill=fill)
+    if d and len(d) >= 36:
+        mask = d[4:36]
+        _boost_mask_cache[gpu] = mask
+        return mask, "OK"
+    return None, err
+
+
+def set_mask_bit(buf, point: int, offset: int = 4) -> None:
+    """Set a single bit in the 256-bit mask for one point."""
+    byte_idx = offset + (point // 8)
+    bit_idx = point % 8
+    buf[byte_idx] = int.from_bytes(buf[byte_idx:byte_idx + 1], "little") | (1 << bit_idx)
+
+
+def set_mask_bits(buf, points: set[int], offset: int = 4) -> None:
+    """Set mask bits for a set of points."""
+    for p in points:
+        set_mask_bit(buf, p, offset)
+
+
+# ── Readers ───────────────────────────────────────────────────────────────────
+
+def read_vfp_curve(gpu) -> tuple[Optional[list[tuple[int, int]]], str]:
+    """Read the base V/F curve (frequency + voltage pairs).
+
+    Returns ([(freq_kHz, volt_uV), ...], "OK") or (None, error).
+    """
+    mask, mask_err = get_boost_mask(gpu)
+    if not mask:
+        return None, f"GetClockBoostMask failed: {mask_err}"
+
+    def fill(buf):
+        for i in range(32):
+            buf[4 + i] = mask[i]
+
+    d, err = nvcall(FUNC["GetVFPCurve"], gpu, VFP_SIZE, ver=1, pre_fill=fill)
+    if not d:
+        return None, err
+
+    points = []
+    max_entries = (len(d) - VFP_BASE) // VFP_STRIDE
+    for i in range(max_entries):
+        off = VFP_BASE + i * VFP_STRIDE
+        freq = struct.unpack_from("<I", d, off)[0]
+        volt = struct.unpack_from("<I", d, off + 4)[0]
+        points.append((freq, volt))
+    return points, "OK"
+
+
+def read_clock_table_raw(gpu) -> tuple[Optional[bytes], str]:
+    """Read the raw ClockBoostTable buffer.
+
+    Used for snapshots, inspection, and as the baseline for writes.
+    Returns (bytes, "OK") or (None, error).
+    """
+    mask, mask_err = get_boost_mask(gpu)
+    if not mask:
+        return None, f"GetClockBoostMask failed: {mask_err}"
+
+    def fill(buf):
+        for i in range(32):
+            buf[4 + i] = mask[i]
+
+    return nvcall(FUNC["GetClockBoostTable"], gpu, CT_SIZE, ver=1, pre_fill=fill)
+
+
+def read_clock_table_parsed(gpu) -> tuple[Optional[list[tuple[int, int]]], str]:
+    """Read per-point offsets and flags from the ClockBoostTable.
+
+    Returns a list of (delta_kHz, flags) tuples, or (None, error).
+    """
+    d, err = read_clock_table_raw(gpu)
+    if not d:
+        return None, err
+
+    entries = []
+    max_entries = (len(d) - CT_BASE) // CT_STRIDE
+    for i in range(max_entries):
+        base_off = CT_BASE + i * CT_STRIDE
+        flags = struct.unpack_from("<I", d, base_off)[0]
+        delta = struct.unpack_from("<i", d, base_off + CT_DELTA_OFF)[0]
+        entries.append((delta, flags))
+
+    return entries, "OK"
+
+
+def read_clock_offsets(gpu) -> tuple[Optional[list[int]], str]:
+    """Read per-point frequency offsets (kHz, signed) from the ClockBoostTable.
+
+    Returns a list of integers, or (None, error).
+    """
+    parsed, err = read_clock_table_parsed(gpu)
+    if not parsed:
+        return None, err
+
+    offsets = [delta for delta, flags in parsed]
+    return offsets, "OK"
+
+
+def read_clock_entry_full(data: bytes, point: int) -> dict:
+    """Extract all 9 raw fields from a single ClockBoostTable entry.
+
+    Useful for diagnostics and verifying unknown fields.
+    """
+    base = CT_BASE + point * CT_STRIDE
+    fields = {}
+    for j in range(9):
+        off = base + j * 4
+        if j == 5:  # freqDelta is signed
+            fields[f"field_{j:02d}_0x{j * 4:02X}"] = struct.unpack_from("<i", data, off)[0]
+        else:
+            fields[f"field_{j:02d}_0x{j * 4:02X}"] = struct.unpack_from("<I", data, off)[0]
+    fields["freqDelta_kHz"] = fields["field_05_0x14"]
+    return fields
+
+
+def read_curve(gpu, gpu_name: str = "") -> tuple[Optional[CurveState], str]:
+    """Read both the VFP curve and ClockBoostTable and merge into CurveState.
+
+    Returns (CurveState, "OK") or (None, error).
+    """
+    import time
+    vfp_points, vfp_err = read_vfp_curve(gpu)
+    if not vfp_points:
+        return None, vfp_err
+
+    ct_entries, ct_err = read_clock_table_parsed(gpu)
+    if not ct_entries:
+        return None, ct_err
+
+    points = []
+    in_memory = False
+    for i, (freq_khz, volt_uv) in enumerate(vfp_points):
+        if freq_khz == 0 and volt_uv == 0:
+            break  # end of populated entries
+
+        delta_khz = ct_entries[i][0] if i < len(ct_entries) else 0
+        flags = ct_entries[i][1] if i < len(ct_entries) else 0
+
+        if flags == 1:
+            in_memory = True
+
+        points.append(VFPoint(
+            index=i,
+            freq_khz=freq_khz,
+            volt_uv=volt_uv,
+            delta_khz=delta_khz,
+            domain="memory" if in_memory else "gpu",
+        ))
+
+    return CurveState(points=points, timestamp=time.time(), gpu_name=gpu_name), "OK"
+
+
+# ── Writers ───────────────────────────────────────────────────────────────────
+
+def build_write_buffer(
+    gpu,
+    point_deltas: dict[int, int],
+    full_mask: bool = False,
+) -> tuple[Optional[ctypes.Array], str]:
+    """Build a SetClockBoostTable buffer with specified per-point deltas.
+
+    Strategy: read the current ClockBoostTable, modify only the targeted
+    entries' freqDelta fields, set only the targeted mask bits (single-bit per
+    write to avoid touching neighbouring points).
+
+    Args:
+        full_mask: if True, copy the complete GetClockBoostMask into the write
+                   buffer instead of the default sparse (per-point) mask.
+                   Older GPUs (e.g. Pascal) may require this.
+
+    Returns (mutable_buffer, "OK") or (None, error).
+    """
+    current_raw, err = read_clock_table_raw(gpu)
+    if not current_raw:
+        return None, f"Cannot read current ClockBoostTable: {err}"
+
+    buf = ctypes.create_string_buffer(CT_SIZE)
+    ctypes.memmove(buf, current_raw, CT_SIZE)
+
+    # Rewrite version word explicitly
+    struct.pack_into("<I", buf, 0, (1 << 16) | CT_SIZE)
+
+    if full_mask:
+        # Copy the complete boost mask — required by some older drivers/GPUs
+        mask, mask_err = get_boost_mask(gpu)
+        if not mask:
+            return None, f"Cannot read boost mask: {mask_err}"
+        for i, b in enumerate(mask):
+            buf[4 + i] = b
+    else:
+        # Sparse mask — set only bits for points we're writing
+        for i in range(4, 4 + 32):
+            buf[i] = 0x00
+        set_mask_bits(buf, set(point_deltas.keys()))
+
+    for point, delta_khz in point_deltas.items():
+        off = CT_BASE + point * CT_STRIDE + CT_DELTA_OFF
+        struct.pack_into("<i", buf, off, delta_khz)
+
+    return buf, "OK"
+
+
+def write_offsets(
+    gpu,
+    point_deltas: dict[int, int],
+    dry_run: bool = False,
+    full_mask: bool = False,
+) -> tuple[int, str]:
+    """Write per-point frequency offsets via SetClockBoostTable.
+
+    Args:
+        gpu: NvAPI GPU handle
+        point_deltas: {point_index: delta_kHz} — only these points are written
+        dry_run: if True, build the buffer but don't call the driver
+        full_mask: if True, use the full GetClockBoostMask (for older GPUs)
+
+    Returns (return_code, description).
+    """
+    buf, err = build_write_buffer(gpu, point_deltas, full_mask=full_mask)
+    if buf is None:
+        return -999, err
+
+    if dry_run:
+        return 0, "DRY RUN — buffer built but not sent to driver"
+
+    return nvcall_raw(FUNC["SetClockBoostTable"], gpu, buf)
+
+
+def write_global_offset(gpu, delta_khz: int, dry_run: bool = False) -> tuple[int, str]:
+    """Apply a uniform frequency offset to all GPU core points."""
+    curve, err = read_curve(gpu)
+    if not curve:
+        return -999, f"Failed to read curve: {err}"
+
+    point_deltas = {p.index: delta_khz for p in curve.points if p.domain == "gpu"}
+    return write_offsets(gpu, point_deltas, dry_run=dry_run)
+
+
+def reset_offsets(gpu, dry_run: bool = False) -> tuple[int, str]:
+    """Zero all GPU core frequency offsets."""
+    curve, err = read_curve(gpu)
+    if not curve:
+        return -999, f"Failed to read curve: {err}"
+
+    point_deltas = {p.index: 0 for p in curve.points if p.domain == "gpu"}
+    return write_offsets(gpu, point_deltas, dry_run=dry_run)

nvcurve/nvapi/__init__.py

@@ -0,0 +1,11 @@
+from .bootstrap import nvcall, nvcall_raw, query_interface
+from .constants import FUNC
+from .errors import NvAPIError, NVAPI_ERRORS
+from .types import VFPoint, CurveState, MonitoringSample, GpuInfo, SnapshotInfo
+
+__all__ = [
+    "nvcall", "nvcall_raw", "query_interface",
+    "FUNC",
+    "NvAPIError", "NVAPI_ERRORS",
+    "VFPoint", "CurveState", "MonitoringSample", "GpuInfo", "SnapshotInfo",
+]

nvcurve/nvapi/bootstrap.py

@@ -0,0 +1,81 @@
+"""NvAPI bootstrap — library loading, function resolution, versioned struct calls."""
+
+import ctypes
+import struct
+import sys
+
+from .errors import NVAPI_ERRORS
+
+
+def load_nvapi() -> ctypes.CDLL:
+    """Load libnvidia-api.so from the NVIDIA driver."""
+    for name in ("libnvidia-api.so", "libnvidia-api.so.1"):
+        try:
+            return ctypes.CDLL(name)
+        except OSError:
+            continue
+    print("Error: Cannot load libnvidia-api.so")
+    print("Ensure the NVIDIA proprietary driver is installed.")
+    sys.exit(1)
+
+
+# Module-level library handle and QueryInterface function pointer.
+_nvapi = load_nvapi()
+_QI = _nvapi.nvapi_QueryInterface
+_QI.restype = ctypes.c_void_p
+_QI.argtypes = [ctypes.c_uint32]
+
+
+def query_interface(fid: int, nargs: int = 2):
+    """Resolve an NvAPI function pointer by its 32-bit ID.
+
+    Returns a callable ctypes function, or None if the driver doesn't expose it.
+    """
+    ptr = _QI(fid)
+    if not ptr:
+        return None
+    return ctypes.CFUNCTYPE(ctypes.c_int32, *[ctypes.c_void_p] * nargs)(ptr)
+
+
+def nvcall(
+    fid: int,
+    gpu,
+    size: int,
+    ver: int = 1,
+    pre_fill=None,
+) -> tuple[bytes | None, str]:
+    """Call an NvAPI function with a versioned struct buffer.
+
+    Allocates a buffer of `size` bytes, writes the version word
+    ``(ver << 16) | size`` at offset 0, optionally calls ``pre_fill(buf)``
+    to populate request fields, then invokes the function.
+
+    Returns ``(bytes, "OK")`` on success or ``(None, error_description)`` on
+    failure.
+    """
+    func = query_interface(fid)
+    if not func:
+        return None, "function pointer not found (driver too old?)"
+    buf = ctypes.create_string_buffer(size)
+    struct.pack_into("<I", buf, 0, (ver << 16) | size)
+    if pre_fill:
+        pre_fill(buf)
+    ret = func(gpu, buf)
+    if ret != 0:
+        return None, f"error {ret} ({NVAPI_ERRORS.get(ret, 'unknown')})"
+    return bytes(buf), "OK"
+
+
+def nvcall_raw(fid: int, gpu, buf: ctypes.Array) -> tuple[int, str]:
+    """Call an NvAPI function with a pre-built mutable buffer.
+
+    Used for write operations where the caller needs full control over the
+    buffer contents (e.g. SetClockBoostTable).
+
+    Returns ``(return_code, description)``.
+    """
+    func = query_interface(fid)
+    if not func:
+        return -999, "function pointer not found (driver too old?)"
+    ret = func(gpu, buf)
+    return ret, NVAPI_ERRORS.get(ret, f"unknown ({ret})")

nvcurve/nvapi/constants.py

@@ -0,0 +1,55 @@
+# NvAPI function IDs, struct sizes, offsets, and safety constants.
+# All values verified by hex-dump analysis on RTX 5090 (GB202), driver 590.48.01.
+
+# ── Function IDs ────────────────────────────────────────────────────────────
+
+FUNC = {
+    # Bootstrap
+    "Initialize":          0x0150E828,
+    "EnumPhysicalGPUs":    0xE5AC921F,
+    "GetFullName":         0xCEEE8E9F,
+
+    # V/F curve (read)
+    "GetVFPCurve":         0x21537AD4,  # ClkVfPointsGetStatus
+    "GetClockBoostMask":   0x507B4B59,  # ClkVfPointsGetInfo
+    "GetClockBoostTable":  0x23F1B133,  # ClkVfPointsGetControl
+    "GetCurrentVoltage":   0x465F9BCF,  # ClientVoltRailsGetStatus
+    "GetClockBoostRanges": 0x64B43A6A,  # ClkDomainsGetInfo
+
+    # Additional read
+    "GetPerfLimits":       0xE440B867,  # PerfClientLimitsGetStatus
+    "GetVoltBoostPercent": 0x9DF23CA1,  # ClientVoltRailsGetControl
+
+    # Write
+    "SetClockBoostTable":  0x0733E009,  # ClkVfPointsSetControl
+}
+
+# ── GetVFPCurve (0x21537AD4) ─────────────────────────────────────────────────
+# 128 × 28-byte entries at offset 0x48; freq_kHz at +0x00, volt_uV at +0x04.
+
+VFP_SIZE   = 0x1C28   # 7208 bytes
+VFP_BASE   = 0x48
+VFP_STRIDE = 0x1C     # 28 bytes
+VFP_POINTS = (VFP_SIZE - VFP_BASE) // VFP_STRIDE
+
+# ── Get/SetClockBoostTable (0x23F1B133 / 0x0733E009) ────────────────────────
+# 128 × 36-byte entries at offset 0x44; freqDelta (int32, kHz) at entry+0x14.
+# Correct layout verified against LACT issue #936 (which had wrong stride 0x48).
+
+CT_SIZE      = 0x2420  # 9248 bytes
+CT_BASE      = 0x44
+CT_STRIDE    = 0x24    # 36 bytes
+CT_DELTA_OFF = 0x14    # freqDelta offset within entry
+CT_POINTS    = (CT_SIZE - CT_BASE) // CT_STRIDE
+
+# ── Other struct sizes ───────────────────────────────────────────────────────
+
+MASK_SIZE   = 0x182C   # GetClockBoostMask
+VOLT_SIZE   = 0x004C   # GetCurrentVoltage  (voltage µV at offset 0x28)
+RANGES_SIZE = 0x0928   # GetClockBoostRanges
+PERF_SIZE   = 0x030C   # GetPerfLimits (version 2, not 1)
+VBOOST_SIZE = 0x0028   # GetVoltBoostPercent
+
+# ── Safety ───────────────────────────────────────────────────────────────────
+
+MAX_DELTA_KHZ = 3_000_000  # ±3000 MHz hard cap (Blackwell driver limit)

nvcurve/nvapi/errors.py

@@ -0,0 +1,27 @@
+# NvAPI error codes and exception class.
+# Linux NvAPI uses negative integers, not the 0x80000000+ range from Windows.
+
+NVAPI_ERRORS: dict[int, str] = {
+    0:   "OK",
+    -1:  "GENERIC_ERROR",
+    -5:  "INVALID_ARGUMENT",
+    -6:  "NVIDIA_DEVICE_NOT_FOUND",
+    -7:  "END_ENUMERATION",
+    -8:  "INVALID_HANDLE",
+    -9:  "INCOMPATIBLE_STRUCT_VERSION",
+    -10: "HANDLE_INVALIDATED",
+    -14: "INVALID_POINTER",
+}
+
+
+class NvAPIError(Exception):
+    """Raised when an NvAPI call returns a non-zero error code."""
+
+    def __init__(self, code: int, context: str = ""):
+        self.code = code
+        self.name = NVAPI_ERRORS.get(code, f"unknown ({code})")
+        self.context = context
+        msg = f"NvAPI error {code} ({self.name})"
+        if context:
+            msg = f"{context}: {msg}"
+        super().__init__(msg)

nvcurve/nvapi/types.py

@@ -0,0 +1,76 @@
+"""Clean Python dataclasses for V/F curve data.
+
+Raw struct manipulation stays in the call layer (bootstrap.py + hal/).
+Everything above HAL works with these types.
+"""
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class VFPoint:
+    index: int
+    freq_khz: int       # Base frequency from VFP curve
+    volt_uv: int        # Voltage from VFP curve
+    delta_khz: int      # Offset from ClockBoostTable (signed)
+    domain: str = "gpu"  # "gpu" or "memory"
+
+    @property
+    def effective_freq_khz(self) -> int:
+        return self.freq_khz + self.delta_khz
+
+    @property
+    def freq_mhz(self) -> float:
+        return self.freq_khz / 1000.0
+
+    @property
+    def effective_freq_mhz(self) -> float:
+        return self.effective_freq_khz / 1000.0
+
+    @property
+    def volt_mv(self) -> float:
+        return self.volt_uv / 1000.0
+
+    @property
+    def delta_mhz(self) -> float:
+        return self.delta_khz / 1000.0
+
+
+@dataclass
+class CurveState:
+    points: list[VFPoint]
+    timestamp: float
+    gpu_name: str
+
+
+@dataclass
+class MonitoringSample:
+    timestamp: float
+    voltage_uv: int | None
+    clock_mhz: float | None
+    temp_c: float | None
+    power_w: float | None
+    fan_pct: float | None
+    pstate: int | None          # Performance state: 0 (P0, max) – 15 (P15, min)
+    mem_used_bytes: int | None  # VRAM used (bytes)
+    mem_total_bytes: int | None # VRAM total (bytes)
+    gpu_util_pct: float | None  # GPU core utilization (0–100)
+    mem_util_pct: float | None  # Memory bus utilization (0–100)
+    mem_clock_mhz: float | None = None  # Current memory clock (NVML_CLOCK_MEM)
+
+
+@dataclass
+class GpuInfo:
+    name: str
+    index: int
+    uuid: str | None = None
+    pci_bus_id: int | None = None
+
+
+@dataclass
+class SnapshotInfo:
+    filepath: str
+    timestamp: str
+    gpu: str
+    nonzero_offsets: int
+    size: int

nvcurve/profiles/__init__.py

@@ -0,0 +1 @@
+# Make profiles a package

nvcurve/profiles/apply.py

@@ -0,0 +1,187 @@
+"""Apply saved profiles to hardware, with optional read-back verification."""
+
+import json
+import logging
+import os
+import sys
+import time
+
+log = logging.getLogger("nvcurve.profiles.apply")
+
+_PERSISTENT_CONFIG_FILE = "/etc/nvcurve/config.json"
+
+
+def _gpu_stable_key(info) -> str:
+    if info.uuid:
+        return info.uuid
+    if info.pci_bus_id is not None:
+        return f"pci:{info.pci_bus_id:04x}"
+    return f"idx:{info.index}"
+
+
+def apply_profile(gpu_index: int, name: str, cfg) -> list[str]:
+    """Apply a named profile to the given GPU. Returns a list of error strings."""
+    from .native import load_profile
+    from ..hal.gpu import get_gpu
+    from ..hal.limits import set_clock_offsets, set_power_limit
+    from ..hal.vfcurve import write_offsets, reset_offsets
+    from ..hal.snapshot import save as snapshot_save
+    from ..safety import validate_write
+
+    safe_name = "".join(c for c in name if c.isalnum() or c in " _-()").strip()
+    filepath = os.path.join(cfg.profile_dir, f"{safe_name}.json")
+
+    profile = load_profile(filepath)  # raises FileNotFoundError if missing
+    gpu, gpu_name = get_gpu(index=gpu_index)
+
+    errs: list[str] = []
+
+    # Apply mem offset first — driver may reset curve table as a side-effect.
+    if profile.mem_offset_mhz is not None:
+        ok, msg = set_clock_offsets(None, profile.mem_offset_mhz, gpu_index)
+        if not ok:
+            errs.append(f"Mem offset: {msg}")
+
+    if profile.power_limit_w is not None:
+        ok, msg = set_power_limit(profile.power_limit_w, gpu_index)
+        if not ok:
+            errs.append(f"Power limit: {msg}")
+
+    if profile.curve_deltas:
+        deltas = {int(k): v for k, v in profile.curve_deltas.items()}
+        errors = validate_write(deltas, cfg.max_delta_khz)
+        if errors:
+            errs.append("Curve: " + "; ".join(errors))
+        else:
+            if cfg.auto_snapshot:
+                try:
+                    snapshot_save(gpu, gpu_name, cfg.snapshot_dir, cfg.max_snapshots)
+                except Exception as exc:
+                    log.warning("Auto-snapshot failed: %s", exc)
+            ret, desc = write_offsets(gpu, deltas)
+            if ret != 0:
+                errs.append(f"Curve write failed ({ret}): {desc}")
+    else:
+        reset_offsets(gpu)
+
+    return errs
+
+
+def apply_with_retry(gpu_index: int, name: str, cfg, max_retries: int = 3) -> bool:
+    """Apply a named profile with read-back verification, retrying on mismatch."""
+    from .native import load_profile
+    from ..hal.gpu import get_gpu
+    from ..hal.vfcurve import read_clock_offsets
+
+    safe_name = "".join(c for c in name if c.isalnum() or c in " _-()").strip()
+    filepath = os.path.join(cfg.profile_dir, f"{safe_name}.json")
+
+    try:
+        profile = load_profile(filepath)
+    except FileNotFoundError:
+        log.warning("Auto-load profile %r not found — skipping GPU %d", name, gpu_index)
+        return False
+
+    expected: dict[int, int] = (
+        {int(k): v for k, v in profile.curve_deltas.items()}
+        if profile.curve_deltas else {}
+    )
+
+    for attempt in range(max_retries):
+        try:
+            errs = apply_profile(gpu_index, name, cfg)
+        except Exception as exc:
+            log.warning("Auto-load attempt %d/%d exception: %s", attempt + 1, max_retries, exc)
+            errs = [str(exc)]
+
+        if errs:
+            log.warning("Auto-load attempt %d/%d errors: %s",
+                        attempt + 1, max_retries, "; ".join(errs))
+        elif expected:
+            gpu, _ = get_gpu(index=gpu_index)
+            offsets, err = read_clock_offsets(gpu)
+            if offsets is None:
+                log.warning("Auto-load attempt %d/%d: read-back failed: %s",
+                            attempt + 1, max_retries, err)
+            else:
+                mismatches = [
+                    f"pt{idx}: expected {val/1000:+.0f}MHz got {offsets[idx]/1000:+.0f}MHz"
+                    for idx, val in expected.items()
+                    if idx < len(offsets) and offsets[idx] != val
+                ]
+                if not mismatches:
+                    log.info("Auto-load profile %r verified on GPU %d (attempt %d/%d)",
+                             name, gpu_index, attempt + 1, max_retries)
+                    return True
+                log.warning("Auto-load attempt %d/%d: read-back mismatch — %s",
+                            attempt + 1, max_retries, "; ".join(mismatches))
+        else:
+            log.info("Auto-load profile %r applied on GPU %d (attempt %d/%d)",
+                     name, gpu_index, attempt + 1, max_retries)
+            return True
+
+        if attempt < max_retries - 1:
+            delay = 2 ** attempt  # 1s, 2s, 4s
+            log.info("Retrying auto-load in %ds…", delay)
+            time.sleep(delay)
+
+    log.warning("Auto-load profile %r failed after %d attempts on GPU %d",
+                name, max_retries, gpu_index)
+    return False
+
+
+def run_autoload() -> None:
+    """Read config and apply all configured auto-load profiles. Requires root."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    if os.geteuid() != 0:
+        print("nvcurve autoload: must run as root", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        with open(_PERSISTENT_CONFIG_FILE) as f:
+            cfg_data = json.load(f)
+    except Exception:
+        cfg_data = {}
+
+    auto_load_profiles: dict = cfg_data.get("auto_load_profiles", {})
+    if not auto_load_profiles:
+        log.info("No auto-load profiles configured.")
+        return
+
+    from ..config import Config
+    cfg = Config()
+    for key in ("max_delta_khz", "auto_snapshot", "max_snapshots",
+                "snapshot_dir", "profile_dir"):
+        if key in cfg_data:
+            setattr(cfg, key, cfg_data[key])
+
+    from ..hal.gpu import init_nvapi, discover_gpus
+    from ..hal.monitoring import init_nvml, shutdown_nvml
+
+    try:
+        init_nvapi()
+    except Exception as exc:
+        log.error("Failed to initialize NvAPI: %s", exc)
+        sys.exit(1)
+
+    init_nvml()  # best-effort
+    gpus = discover_gpus()
+    if not gpus:
+        log.warning("No GPUs discovered.")
+
+    key_to_idx = {_gpu_stable_key(info): info.index for info in gpus}
+    for gpu_key, profile_name in auto_load_profiles.items():
+        if not profile_name:
+            continue
+        gpu_idx = key_to_idx.get(gpu_key)
+        if gpu_idx is None:
+            log.warning("Auto-load: no GPU found with key %r — skipping", gpu_key)
+            continue
+        log.info("Auto-loading profile %r on GPU %d (%s)", profile_name, gpu_idx, gpu_key)
+        apply_with_retry(gpu_idx, profile_name, cfg)
+
+    shutdown_nvml()

nvcurve/profiles/native.py

@@ -0,0 +1,93 @@
+"""Native profile storage and schema."""
+
+import json
+import os
+import glob
+from dataclasses import dataclass, asdict
+from typing import Dict, Optional, List
+
+
+@dataclass
+class ProfileData:
+    name: str
+    gpu_name: str
+    curve_deltas: Dict[str, int]  # { "index": delta_khz }
+    mem_offset_mhz: Optional[int] = None
+    power_limit_w: Optional[int] = None
+
+
+def save_profile(profile_dir: str, data: ProfileData) -> str:
+    """Save profile to JSON, sanitising the filename."""
+    os.makedirs(profile_dir, exist_ok=True)
+    safe_name = "".join(c for c in data.name if c.isalnum() or c in " _-()").strip()
+    if not safe_name:
+        safe_name = "Unnamed"
+        
+    filepath = os.path.join(profile_dir, f"{safe_name}.json")
+    with open(filepath, "w", encoding="utf-8") as f:
+        json.dump(asdict(data), f, indent=2)
+    return filepath
+
+
+def load_profile(filepath: str) -> ProfileData:
+    """Load profile from JSON."""
+    with open(filepath, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    # Migrate old field names.
+    if "vram_p0_offset_mhz" in data and "mem_offset_mhz" not in data:
+        data["mem_offset_mhz"] = data.pop("vram_p0_offset_mhz")
+    # Drop removed fields so old profiles don't cause TypeError.
+    for obsolete in ("gpu_locked_min_mhz", "gpu_locked_max_mhz", "vram_p0_offset_mhz"):
+        data.pop(obsolete, None)
+    return ProfileData(**data)
+
+
+def list_profiles(profile_dir: str) -> List[ProfileData]:
+    """Return a list of all safely readable profiles."""
+    if not os.path.exists(profile_dir):
+        return []
+    profiles = []
+    for fp in glob.glob(os.path.join(profile_dir, "*.json")):
+        try:
+            profiles.append(load_profile(fp))
+        except Exception as e:
+            # log warning ideally, but swallowing for robustness
+            pass
+    # Sort alphabetically by name
+    profiles.sort(key=lambda p: p.name.lower())
+    return profiles
+
+
+def rename_profile(profile_dir: str, old_name: str, new_name: str) -> bool:
+    """Rename a profile: update the name field and move the file."""
+    old_safe = "".join(c for c in old_name if c.isalnum() or c in " _-()").strip()
+    new_safe = "".join(c for c in new_name if c.isalnum() or c in " _-()").strip()
+    if not new_safe:
+        return False
+    old_path = os.path.join(profile_dir, f"{old_safe}.json")
+    new_path = os.path.join(profile_dir, f"{new_safe}.json")
+    if not os.path.exists(old_path):
+        return False
+    try:
+        profile = load_profile(old_path)
+        profile.name = new_name
+        with open(new_path, "w", encoding="utf-8") as f:
+            json.dump(asdict(profile), f, indent=2)
+        if old_path != new_path:
+            os.remove(old_path)
+        return True
+    except Exception:
+        return False
+
+
+def delete_profile(profile_dir: str, name: str) -> bool:
+    """Delete a profile by name."""
+    safe_name = "".join(c for c in name if c.isalnum() or c in " _-()").strip()
+    filepath = os.path.join(profile_dir, f"{safe_name}.json")
+    if os.path.exists(filepath):
+        try:
+            os.remove(filepath)
+            return True
+        except Exception:
+            return False
+    return False

nvcurve/safety.py

@@ -0,0 +1,72 @@
+"""Centralized safety validation for all write operations.
+
+Every write path calls validate_write() before touching hardware.
+"""
+
+from .nvapi.constants import CT_POINTS
+
+
+def check_negative_freq_warnings(
+    point_deltas: dict[int, int],
+    vfp_freqs_khz: list[int],
+    current_offsets_khz: list[int] | None = None,
+) -> list[str]:
+    """Return warnings for deltas that would produce negative effective frequency.
+
+    GetVFPCurve returns the *current effective* frequencies (already reflecting
+    any applied boost delta). The true hardware base for each point is therefore:
+        true_base = vfp_freq - current_offset
+
+    The new effective with the proposed delta:
+        new_effective = true_base + new_delta = vfp_freq + (new_delta - current_offset)
+
+    The driver clamps results, but users should be informed when their delta
+    would push a point's effective frequency below zero.
+    """
+    warnings = []
+    for point, new_delta_khz in point_deltas.items():
+        if point < 0 or point >= len(vfp_freqs_khz):
+            continue
+        vfp_freq = vfp_freqs_khz[point]
+        if vfp_freq == 0:
+            continue  # unused / placeholder point
+        current_delta = (
+            current_offsets_khz[point]
+            if current_offsets_khz and point < len(current_offsets_khz)
+            else 0
+        )
+        new_effective = vfp_freq + (new_delta_khz - current_delta)
+        if new_effective < 0:
+            warnings.append(
+                f"Point {point}: delta {new_delta_khz / 1000:+.0f} MHz would produce "
+                f"effective frequency {new_effective / 1000:.0f} MHz — driver will clamp to 0."
+            )
+    return warnings
+
+
+def validate_write(
+    point_deltas: dict[int, int],
+    max_delta_khz: int,
+) -> list[str]:
+    """Validate a proposed write request.
+
+    Args:
+        point_deltas: {point_index: delta_kHz}
+        max_delta_khz: absolute delta limit (e.g. 300_000 for ±300 MHz)
+
+    Returns a list of error message strings. Empty list means the request is safe.
+    """
+    errors = []
+    for point, delta_khz in point_deltas.items():
+        if point < 0 or point >= CT_POINTS:
+            errors.append(f"Point {point} out of range (0–{CT_POINTS - 1})")
+            continue
+
+        if abs(delta_khz) > max_delta_khz:
+            errors.append(
+                f"Delta {delta_khz / 1000:+.0f} MHz for point {point} exceeds "
+                f"safety limit of ±{max_delta_khz / 1000:.0f} MHz. "
+                "Use --max-delta to raise the limit if needed."
+            )
+
+    return errors