Files
nvcurve/nvcurve/server.py
T
ARIA 930e56bd07 Clean up LSP diagnostics across backend and frontend
Backend (nvcurve/):
- hal/fans.py, hal/limits.py, hal/gpu.py: replace conditional pynvml
  imports with the established 'pynvml: Any = _pynvml_import' pattern
  (fixes ~50 'possibly unbound' errors); type the result dicts; guard
  query_interface() results; explicit uuid/pci-bus parsing (int, hex
  convention documented); modernize Optional[T] -> T | None
- cli.py: fix 'curve_state' possibly-unbound and snap_path None handling
  in cmd_setup; wrap unchecked int()/open()/makedirs() calls in
  try/except with clean CLI errors; add module logger for silent
  except-pass blocks; raise ... from exc; fix unused loop vars and
  set-comprehension
- hal/snapshot.py: filepath: str | None; wrap all file ops; sorted
  imports; remove unused CT_POINTS import
- daemon.py: extract 0o666 to _SOCKET_MODE constant (intentional for
  /run sockets) with nosemgrep
- server.py: nosemgrep for Python 3.7-compat false positive (project
  requires >= 3.12); log previously-swallowed exception
- profiles/native.py, profiles/apply.py: wrap file ops and int(k)
  profile-key parsing; sorted imports; modernize typing

Frontend (frontend/src):
- Add .js extensions to all relative imports (standard TS-ESM; Vite
  resolves .js -> .ts)
- React.FormEvent (deprecated in React 19 types) -> React.SubmitEvent
- catch (e: any) -> catch (e: unknown) + instanceof Error narrowing
- React-hooks: move ref writes from render into effects; convert
  viewport reset to render-phase state adjustment; split
  selectPoint(index, multi) into selectPoint + togglePoint (no flag
  argument); remove non-null assertion
- Static inline styles -> Tailwind classes (dynamic positioning/cursor
  styles kept)
- Remove non-standard 'container' option from scrollIntoView (browsers
  ignore unknown options) which had orphaned a @ts-expect-error
- Object.fromEntries for Map -> Record conversion

Tooling:
- .gitignore: ignore .codegraph/ local tool data

Verified: tsc --noEmit, vite production build, python imports, and
full LSP scan (0 errors/warnings in both projects).
2026-09-08 23:57:30 +02:00

1799 lines
61 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""FastAPI API server — REST + WebSocket.
Run via: nvcurve serve [--host 127.0.0.1 --port 8042]
Or: uvicorn nvcurve.server:app
Requires root (NvAPI needs it).
"""
import asyncio
import logging
import os
from contextlib import asynccontextmanager, suppress
from pathlib import Path
from typing import Any
from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from . import auth
from .config import Config, default_config
from .hal.dashboard import get_dashboard_info
from .hal.fans import (
get_fan_info,
get_temp,
interpolate_fan_speed,
reset_fan,
set_fan_speed,
validate_curve,
)
from .hal.gpu import discover_gpus, get_gpu
from .hal.limits import (
get_clock_offsets,
get_mem_offset_range,
get_power_limit,
set_clock_offsets,
set_power_limit,
)
from .hal.monitoring import (
get_driver_version,
get_vram_total,
init_nvml,
poll,
shutdown_nvml,
throttle_reasons_label,
)
from .hal.ranges import get_clock_ranges
from .hal.snapshot import (
list_snapshots,
)
from .hal.snapshot import (
restore as snapshot_restore,
)
from .hal.snapshot import (
save as snapshot_save,
)
from .hal.vfcurve import (
read_clock_offsets,
read_curve,
reset_offsets,
write_global_offset,
write_offsets,
)
from .profiles.native import (
ProfileData,
delete_profile,
list_profiles,
load_profile,
rename_profile,
save_profile,
)
from .safety import check_negative_freq_warnings, validate_write
log = logging.getLogger("nvcurve.server")
def _open_browser_as_user(url: str) -> None:
"""Open URL as the original (non-root) user when running under sudo."""
import os
import subprocess
sudo_user = os.environ.get("SUDO_USER")
if sudo_user and os.geteuid() == 0:
try:
subprocess.Popen(
["runuser", "-u", sudo_user, "--", "xdg-open", url],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return
except Exception as exc:
log.debug("runuser xdg-open failed, falling back to webbrowser: %s", exc)
import webbrowser
webbrowser.open(url)
# ── Shared app state ──────────────────────────────────────────────────────────
_state: dict[str, Any] = {
"gpus": {}, # dict[int, dict] mapping gpu_index -> gpu state
"config": default_config,
}
def _get_gpu_state(gpu_index: int) -> dict:
if gpu_index not in _state["gpus"]:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail=f"GPU {gpu_index} not found")
return _state["gpus"][gpu_index]
# ── Serialization helpers ─────────────────────────────────────────────────────
def _vfpoint_dict(p) -> dict:
return {
"index": p.index,
"freq_khz": p.freq_khz,
"freq_mhz": p.freq_mhz,
"volt_uv": p.volt_uv,
"volt_mv": p.volt_mv,
"delta_khz": p.delta_khz,
"delta_mhz": p.delta_mhz,
"effective_freq_khz": p.effective_freq_khz,
"effective_freq_mhz": p.effective_freq_mhz,
"domain": p.domain,
}
def _curve_state_dict(state) -> dict:
return {
"gpu_name": state.gpu_name,
"timestamp": state.timestamp,
"points": [_vfpoint_dict(p) for p in state.points],
}
def _sample_dict(s) -> dict:
return {
"timestamp": s.timestamp,
"voltage_uv": s.voltage_uv,
"voltage_mv": s.voltage_uv / 1000.0 if s.voltage_uv is not None else None,
"clock_mhz": s.clock_mhz,
"mem_clock_mhz": s.mem_clock_mhz,
"temp_c": s.temp_c,
"power_w": s.power_w,
"fan_pct": s.fan_pct,
"pstate": s.pstate,
"pstate_label": f"P{s.pstate}" if s.pstate is not None else None,
"mem_used_bytes": s.mem_used_bytes,
"mem_total_bytes": s.mem_total_bytes,
"mem_used_mib": round(s.mem_used_bytes / (1024**2), 1)
if s.mem_used_bytes is not None
else None,
"mem_total_mib": round(s.mem_total_bytes / (1024**2), 1)
if s.mem_total_bytes is not None
else None,
"gpu_util_pct": s.gpu_util_pct,
"mem_util_pct": s.mem_util_pct,
"throttle_reasons": s.throttle_reasons,
"throttle_reasons_label": throttle_reasons_label(s.throttle_reasons),
"pcie_link_width": s.pcie_link_width,
"pcie_link_generation": s.pcie_link_generation,
"mem_temp_c": s.mem_temp_c,
}
def _int_key_deltas(deltas: dict) -> dict[int, int]:
"""Convert string-keyed deltas (from JSON) to int-keyed.
Raises ValueError if any key is not a valid integer (corrupted profile),
so a bad profile fails closed rather than partially applying to hardware.
"""
out: dict[int, int] = {}
for k, v in deltas.items():
try:
out[int(k)] = v
except (TypeError, ValueError) as exc:
raise ValueError(f"Invalid curve point index in profile: {k!r}") from exc
return out
# ── WebSocket broadcast ───────────────────────────────────────────────────────
async def _broadcast(clients: set, payload: dict) -> None:
"""Send JSON payload to all connected WebSocket clients, evict dead ones."""
dead = set()
for ws in list(clients):
try:
await ws.send_json(payload)
except Exception:
dead.add(ws)
clients -= dead
# ── Background monitoring poller ─────────────────────────────────────────────
async def _monitor_poller(gpu_index: int) -> None:
"""Continuously poll GPU state and push to connected monitor WebSocket clients."""
cfg: Config = _state["config"]
while True:
try:
g_state = _state["gpus"].get(gpu_index)
if g_state and g_state["gpu"] is not None and g_state["monitor_clients"]:
loop = asyncio.get_running_loop()
sample = await loop.run_in_executor(
None, poll, g_state["gpu"], gpu_index
)
await _broadcast(g_state["monitor_clients"], _sample_dict(sample))
except Exception as exc:
log.warning("Monitor poller error for GPU %d: %s", gpu_index, exc)
await asyncio.sleep(cfg.poll_interval_s)
async def _fan_poller(gpu_index: int) -> None:
"""Continuously read GPU temp, interpolate fan speed from active curve, and apply."""
while True:
try:
g_state = _state["gpus"].get(gpu_index)
if g_state and g_state.get("fan_curve_active") and g_state.get("fan_curve"):
temp = await _run(get_temp, gpu_index)
if temp is not None:
curve = g_state["fan_curve"]
target = interpolate_fan_speed(curve, temp)
if target is not None:
await _run(set_fan_speed, gpu_index, target)
except asyncio.CancelledError:
return
except Exception as exc:
log.warning("Fan poller error for GPU %d: %s", gpu_index, exc)
await asyncio.sleep(2.0)
async def _activate_fan_curve(gpu_index: int, curve: list) -> None:
"""Set the active fan curve, (re)start the poller, and persist it.
Persistence (config.json) is what makes the curve survive server restarts:
fan control is volatile, so the driver reverts to automatic mode on reboot
and the saved curve is re-applied at the next server start.
"""
g_state = _get_gpu_state(gpu_index)
# Stop existing poller if running
if g_state.get("fan_poller_task"):
g_state["fan_poller_task"].cancel()
with suppress(asyncio.CancelledError):
await g_state["fan_poller_task"]
g_state["fan_curve"] = curve
g_state["fan_curve_active"] = True
g_state["fan_poller_task"] = asyncio.create_task(_fan_poller(gpu_index))
cfg: Config = _state["config"]
cfg.fan_curves[_gpu_stable_key(gpu_index)] = curve
_persist_fan_curves(cfg.fan_curves)
async def _deactivate_fan_curve(gpu_index: int, reset_hardware: bool = True) -> None:
"""Clear the active fan curve, stop the poller, and clear its persistence.
When reset_hardware is True the GPU is returned to automatic fan control.
"""
g_state = _get_gpu_state(gpu_index)
if g_state.get("fan_poller_task"):
g_state["fan_poller_task"].cancel()
with suppress(asyncio.CancelledError):
await g_state["fan_poller_task"]
g_state["fan_poller_task"] = None
g_state["fan_curve"] = None
g_state["fan_curve_active"] = False
if reset_hardware:
ok, msg = await _run(reset_fan, gpu_index)
if not ok:
log.warning("Fan reset warning: %s", msg)
cfg: Config = _state["config"]
key = _gpu_stable_key(gpu_index)
if key in cfg.fan_curves:
del cfg.fan_curves[key]
_persist_fan_curves(cfg.fan_curves)
# ── Lifespan ──────────────────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
loop = asyncio.get_running_loop()
# Initialize NVML (best-effort)
await loop.run_in_executor(None, init_nvml)
gpu_infos = await loop.run_in_executor(None, discover_gpus)
if not gpu_infos:
log.warning("No GPUs discovered.")
poller_tasks = []
for info in gpu_infos:
idx = info.index
g_state = {
"gpu": None,
"gpu_name": info.name,
"uuid": info.uuid,
"pci_bus_id": info.pci_bus_id,
"write_lock": asyncio.Lock(),
"last_offsets": None,
"active_profile": None,
"monitor_clients": set(),
"curve_clients": set(),
"fan_curve": None,
"fan_curve_active": False,
"fan_poller_task": None,
}
_state["gpus"][idx] = g_state
try:
gpu, name = await loop.run_in_executor(None, get_gpu, idx)
g_state["gpu"] = gpu
g_state["gpu_name"] = name
log.info("GPU %d: %s", idx, name)
# Read initial offsets for reconciliation baseline
offsets, err = await loop.run_in_executor(None, read_clock_offsets, gpu)
if offsets:
g_state["last_offsets"] = offsets
poller_tasks.append(asyncio.create_task(_monitor_poller(idx)))
except Exception as exc:
log.error("Failed to initialize GPU %d: %s", idx, exc)
# ── Backward Compatibility Bridge ──────────────────────────────────────────
# NOTE: This auto-load path is for users running the server directly (e.g.
# via an old systemd unit file that lacks the new daemon mode).
# In the future, this will be removed and auto-loading will be the
# responsibility of daemon.py only.
cfg: Config = _state["config"]
if cfg.auto_load_profiles:
# Build a reverse map: stable_key → current gpu_index
key_to_idx = {_gpu_stable_key(idx): idx for idx in _state["gpus"]}
for gpu_key, profile_name in cfg.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) [compat path]",
profile_name,
gpu_idx,
gpu_key,
)
try:
await _auto_apply_profile_with_retry(profile_name, gpu_idx)
except FileNotFoundError:
log.warning(
"Auto-load profile %r not found in %s — skipping GPU %d",
profile_name,
cfg.profile_dir,
gpu_idx,
)
except Exception as exc:
log.warning(
"Auto-load profile %r failed on GPU %d: %s — skipping",
profile_name,
gpu_idx,
exc,
)
# ──────────────────────────────────────────────────────────────────────────
# ── Restore persisted fan curves ──────────────────────────────────────────
# Fan control is volatile: the driver reverts to automatic mode on reboot,
# so a curve applied via the UI is persisted in config.json and re-applied
# here at startup. Runs after the auto-load profile path so the user's
# explicit fan curve setting takes precedence.
for gpu_index, g_state in _state["gpus"].items():
if g_state["gpu"] is None:
continue
if g_state.get("fan_curve_active"):
continue # already activated by the auto-load profile path
key = _gpu_stable_key(gpu_index)
curve = cfg.fan_curves.get(key)
if not curve:
continue
ok, msg = validate_curve(curve)
if not ok:
log.warning("Skipping persisted fan curve for GPU %d: %s", gpu_index, msg)
continue
log.info("Restoring persisted fan curve on GPU %d (%s)", gpu_index, key)
try:
await _activate_fan_curve(gpu_index, curve)
except Exception as exc:
log.warning(
"Failed to restore persisted fan curve on GPU %d: %s",
gpu_index,
exc,
)
# ──────────────────────────────────────────────────────────────────────────
yield # server is running
for task in poller_tasks:
task.cancel()
for task in poller_tasks:
with suppress(asyncio.CancelledError):
await task
for gpu_index, g_state in _state["gpus"].items():
if g_state.get("fan_poller_task"):
g_state["fan_poller_task"].cancel()
with suppress(asyncio.CancelledError):
await g_state["fan_poller_task"]
if g_state.get("fan_curve_active"):
g_state["fan_curve_active"] = False
g_state["fan_curve"] = None
try:
await loop.run_in_executor(None, reset_fan, gpu_index)
log.info(
"GPU %d: restored automatic fan control on shutdown", gpu_index
)
except Exception as exc:
log.warning(
"GPU %d: failed to restore automatic fan control on shutdown: %s",
gpu_index,
exc,
)
await loop.run_in_executor(None, shutdown_nvml)
# ── App ───────────────────────────────────────────────────────────────────────
app = FastAPI(title="nvcurve", version="0.5.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
# The SPA is served same-origin by this server, so CORS only matters for
# local development (e.g. the Vite dev server). Restrict to localhost
# origins rather than a wildcard.
allow_origin_regex=r"https?://(localhost|127\.0\.0\.1)(:\d+)?$",
allow_methods=["*"],
allow_headers=["*"],
)
# ── Authentication middleware (dual mode) ─────────────────────────────────────
# When the user store contains at least one user, every /api/* endpoint requires
# a valid session (cookie or Bearer token). With no users configured, the API is
# open — exactly like before. Static files (the SPA, including the login page)
# and the public auth/ping endpoints are always reachable.
PUBLIC_API_PATHS = {
"/api/ping",
"/api/auth/login",
"/api/auth/status",
"/api/auth/logout",
}
class AuthMiddleware:
"""Require a valid session for /api/* when authentication is enabled."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
path = scope["path"]
if (
path.startswith("/api/")
and path not in PUBLIC_API_PATHS
and scope.get("method") != "OPTIONS"
):
cfg: Config = _state["config"]
if auth.auth_enabled(cfg.users_file):
token = auth.extract_token(Request(scope))
if auth.get_session(token) is None:
response = JSONResponse(
{"detail": "Authentication required"},
status_code=401,
)
await response(scope, receive, send)
return
await self.app(scope, receive, send)
app.add_middleware(AuthMiddleware)
# ── Request models ────────────────────────────────────────────────────────────
class WriteRequest(BaseModel):
deltas: dict[int, int] # {point_index: delta_kHz}
max_delta_khz: int | None = None # per-request safety limit override
class GlobalOffsetRequest(BaseModel):
delta_khz: int
max_delta_khz: int | None = None # per-request safety limit override
class VerifyRequest(BaseModel):
deltas: dict[int, int] # {point_index: delta_kHz} — pre-expanded by CLI
class SnapshotRestoreRequest(BaseModel):
filepath: str | None = None
class LimitsRequest(BaseModel):
power_limit_w: int | None = None
mem_offset_mhz: int | None = None
class ProfileSaveRequest(BaseModel):
name: str
class ProfileRenameRequest(BaseModel):
new_name: str
class ConfigUpdateRequest(BaseModel):
auto_load_profile: str | None = None
gpu_index: int = 0
class FanCurvePoint(BaseModel):
temp_c: int
fan_pct: int
class FanCurveRequest(BaseModel):
curve: list[FanCurvePoint]
class FanSpeedRequest(BaseModel):
fan_pct: int
class LoginRequest(BaseModel):
username: str
password: str
# ── Helper: run blocking HAL call in thread pool ──────────────────────────────
async def _run(fn, *args):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, fn, *args)
# ── Auth endpoints ────────────────────────────────────────────────────────────
@app.get("/api/ping")
async def api_ping():
"""Public liveness probe (no auth). Used by the CLI to detect a running server."""
return {"ok": True}
@app.get("/api/auth/status")
async def api_auth_status(request: Request):
"""Report whether auth is required and whether this request is authenticated."""
cfg: Config = _state["config"]
required = auth.auth_enabled(cfg.users_file)
token = auth.extract_token(request)
username = auth.get_session(token) if required else None
expires_at = auth.get_session_expires_at(token) if username else None
return {
"auth_required": required,
"authenticated": username is not None,
"username": username,
"expires_at": expires_at,
}
@app.post("/api/auth/login")
async def api_auth_login(req: LoginRequest, request: Request):
"""Authenticate with username+password. Sets a 24-hour session cookie.
The plaintext password is checked against the stored bcrypt hash and never
persisted. On success a random session token is returned (for Bearer use)
and set as an HttpOnly cookie (for browser use).
"""
cfg: Config = _state["config"]
users = auth.load_users(cfg.users_file)
if not users:
raise HTTPException(status_code=404, detail="Authentication is not enabled")
client_ip = request.client.host if request.client else "unknown"
if auth.is_locked_out(client_ip):
raise HTTPException(
status_code=429, detail="Too many failed attempts. Try again later."
)
if auth.check_credentials(users, req.username, req.password):
auth.clear_failures(client_ip)
token, expires_at = auth.create_session(req.username)
response = JSONResponse(
{
"ok": True,
"username": req.username,
"expires_at": expires_at,
"token": token,
}
)
response.set_cookie(
auth.COOKIE_NAME,
token,
max_age=auth.SESSION_TTL_S,
httponly=True,
samesite="lax",
path="/",
)
return response
auth.record_failure(client_ip)
raise HTTPException(status_code=401, detail="Invalid username or password")
@app.post("/api/auth/logout")
async def api_auth_logout(request: Request):
"""End the current session (idempotent)."""
token = auth.extract_token(request)
auth.destroy_session(token)
response = JSONResponse({"ok": True})
response.delete_cookie(auth.COOKIE_NAME, path="/")
return response
@app.get("/api/auth/users")
async def api_auth_users():
"""List configured usernames (requires auth)."""
cfg: Config = _state["config"]
return {"users": auth.list_users(cfg.users_file)}
def _require_gpu(gpu_index: int = 0):
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
if gpu is None:
raise HTTPException(status_code=503, detail=f"GPU {gpu_index} not initialized")
return gpu, g_state
# ── REST endpoints ────────────────────────────────────────────────────────────
@app.get("/api/gpus")
async def api_gpus():
"""List all discovered GPUs."""
from .hal.gpu import discover_gpus
gpu_infos = await _run(discover_gpus)
return [
{
"index": info.index,
"name": info.name,
"uuid": info.uuid,
"pci_bus_id": info.pci_bus_id,
}
for info in gpu_infos
]
@app.get("/api/gpu")
async def api_gpu(gpu_index: int = 0):
"""GPU info: name, driver version, VRAM."""
gpu, g_state = _require_gpu(gpu_index)
driver = get_driver_version()
vram = get_vram_total(gpu_index)
return {
"name": g_state["gpu_name"],
"index": gpu_index,
"driver_version": driver,
"vram_bytes": vram,
"vram_gib": round(vram / (1024**3), 2) if vram else None,
}
@app.get("/api/dashboard")
async def api_dashboard(gpu_index: int = 0):
"""Static GPU info for the Dashboard tab (VBIOS, CUDA cores, PCIe, BAR1, etc.).
Live values (clocks, temps, power, throttle) come from the monitor WebSocket.
"""
_, g_state = _require_gpu(gpu_index)
info = await _run(get_dashboard_info, gpu_index, g_state["gpu_name"])
return info
@app.get("/api/curve")
async def api_curve(gpu_index: int = 0):
"""Full CurveState: all V/F points with base freq, voltage, delta, effective freq."""
gpu, g_state = _require_gpu(gpu_index)
state, err = await _run(read_curve, gpu, g_state["gpu_name"])
if state is None:
raise HTTPException(status_code=500, detail=f"Failed to read curve: {err}")
# Update reconciliation baseline
g_state["last_offsets"] = [p.delta_khz for p in state.points]
return _curve_state_dict(state)
@app.get("/api/curve/{point}")
async def api_curve_point(point: int, gpu_index: int = 0):
"""Single V/F point detail."""
gpu, g_state = _require_gpu(gpu_index)
state, err = await _run(read_curve, gpu, g_state["gpu_name"])
if state is None:
raise HTTPException(status_code=500, detail=f"Failed to read curve: {err}")
if point < 0 or point >= len(state.points):
raise HTTPException(
status_code=400, detail=f"Point index must be 0–{len(state.points) - 1}"
)
return _vfpoint_dict(state.points[point])
@app.get("/api/ranges")
async def api_ranges(gpu_index: int = 0):
"""Clock boost domain ranges (min/max offset per domain)."""
gpu, g_state = _require_gpu(gpu_index)
ranges, err = await _run(get_clock_ranges, gpu)
if ranges is None:
raise HTTPException(status_code=500, detail=f"Failed to read ranges: {err}")
return ranges
@app.get("/api/voltage")
async def api_voltage(gpu_index: int = 0):
"""Current GPU core voltage."""
from .hal.monitoring import read_voltage
gpu, g_state = _require_gpu(gpu_index)
voltage_uv, err = await _run(read_voltage, gpu)
if voltage_uv is None:
raise HTTPException(status_code=500, detail=f"Failed to read voltage: {err}")
return {"voltage_uv": voltage_uv, "voltage_mv": voltage_uv / 1000.0}
@app.get("/api/monitor")
async def api_monitor(gpu_index: int = 0):
"""One-shot monitoring snapshot: voltage, clock, temp, power, fan, p-state, VRAM, utilization."""
gpu, g_state = _require_gpu(gpu_index)
sample = await _run(poll, gpu, gpu_index)
return _sample_dict(sample)
@app.get("/api/snapshots")
async def api_snapshots():
"""List saved ClockBoostTable snapshots."""
cfg: Config = _state["config"]
snapshots = await _run(list_snapshots, cfg.snapshot_dir)
return [
{
"filepath": s.filepath,
"timestamp": s.timestamp,
"gpu": s.gpu,
"nonzero_offsets": s.nonzero_offsets,
"size": s.size,
}
for s in snapshots
]
def _persist_config_field(key: str, value) -> None:
"""Write a single key into /etc/nvcurve/config.json if the file exists.
The file is created by `service install`. If it doesn't exist (e.g. the
service was never installed), config changes are in-memory only for the
current server session. Silently ignores errors.
"""
import json as _json
import os as _os
config_path = "/etc/nvcurve/config.json"
if not _os.path.exists(config_path):
return
try:
with open(config_path) as f:
data = _json.load(f)
except Exception:
data = {}
if value is not None:
data[key] = value
else:
data.pop(key, None)
try:
with open(config_path, "w") as f:
_json.dump(data, f)
except Exception as exc:
log.warning("Failed to persist config field %r: %s", key, exc)
def _gpu_stable_key(gpu_index: int) -> str:
"""Return a stable identifier for a GPU suitable for use as a config key.
Preference order: NVML UUID → PCI bus ID → 'idx:{n}' fallback.
UUID is the most stable across reboots and GPU slot changes.
"""
g_state = _state["gpus"].get(gpu_index, {})
uuid = g_state.get("uuid")
if uuid:
return uuid
pci = g_state.get("pci_bus_id")
if pci is not None:
return f"pci:{pci:04x}"
return f"idx:{gpu_index}"
def _persist_auto_load_profiles(profiles: dict[str, str]) -> None:
"""Persist auto_load_profiles dict to config.json."""
_persist_config_field("auto_load_profiles", profiles if profiles else None)
def _persist_fan_curves(fan_curves: dict) -> None:
"""Persist the per-GPU active fan curves dict to config.json."""
_persist_config_field("fan_curves", fan_curves if fan_curves else None)
@app.get("/api/profiles")
async def api_profiles(gpu_index: int = 0):
"""List saved native profiles, the active profile name, and the auto-load profile name."""
cfg: Config = _state["config"]
profiles = await _run(list_profiles, cfg.profile_dir)
g_state = _state["gpus"].get(gpu_index)
active = g_state["active_profile"] if g_state else None
return {
"profiles": profiles,
"active": active,
"auto_load": cfg.auto_load_profiles.get(_gpu_stable_key(gpu_index)),
}
@app.post("/api/profiles")
async def api_profile_save(req: ProfileSaveRequest, gpu_index: int = 0):
"""Save current GPU state (curve deltas + limits) as a named profile."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
state, err = await _run(read_curve, gpu, g_state["gpu_name"])
if state is None:
raise HTTPException(status_code=500, detail=f"Failed to read curve: {err}")
curve_deltas = {str(p.index): p.delta_khz for p in state.points if p.delta_khz != 0}
try:
power_info = await _run(get_power_limit, gpu_index)
offsets = await _run(get_clock_offsets, gpu_index)
power_limit_w = power_info.get("power_limit_w")
mem_offset_mhz = offsets.get("mem_offset_mhz")
except Exception:
power_limit_w = None
mem_offset_mhz = None
data = ProfileData(
name=req.name,
gpu_name=g_state["gpu_name"],
curve_deltas=curve_deltas,
mem_offset_mhz=mem_offset_mhz,
power_limit_w=power_limit_w,
fan_curve=g_state.get("fan_curve") if g_state.get("fan_curve_active") else None,
)
filepath = await _run(save_profile, cfg.profile_dir, data)
g_state["active_profile"] = req.name
return {"ok": True, "filepath": filepath}
async def _auto_apply_profile_with_retry(
name: str, gpu_index: int = 0, max_retries: int = 3
) -> None:
"""Apply profile on startup with read-back verification and exponential backoff retry.
Raises FileNotFoundError if the profile does not exist (no point retrying).
Logs a warning and gives up after max_retries failed attempts.
"""
import os as _os
cfg: Config = _state["config"]
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
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 = await _run(load_profile, filepath) # raises FileNotFoundError if missing
expected: dict[int, int] = (
_int_key_deltas(profile.curve_deltas) if profile.curve_deltas else {}
)
for attempt in range(max_retries):
errs = await _apply_profile(name, gpu_index)
if errs:
log.warning(
"Auto-load attempt %d/%d had errors: %s",
attempt + 1,
max_retries,
"; ".join(errs),
)
elif expected:
offsets, err = await _run(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
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
if attempt < max_retries - 1:
delay = 2**attempt # 1 s, 2 s, 4 s
log.info("Retrying auto-load in %ds…", delay)
await asyncio.sleep(delay)
log.warning(
"Auto-load profile %r failed after %d attempts — giving up", name, max_retries
)
async def _apply_profile(name: str, gpu_index: int = 0) -> list[str]:
"""Load and apply a saved profile to hardware.
Returns a list of error strings. An empty list means success.
Raises FileNotFoundError if the profile file does not exist.
Sets g_state["active_profile"] on full success.
"""
import os as _os
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
cfg: Config = _state["config"]
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")
# Let FileNotFoundError propagate so callers can map it to 404 or a warning.
profile = await _run(load_profile, filepath)
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 = await _run(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 = await _run(set_power_limit, profile.power_limit_w, gpu_index)
if not ok:
errs.append(f"Power limit: {msg}")
# Apply curve deltas (after mem offset which may have wiped them).
async with g_state["write_lock"]:
if profile.curve_deltas:
deltas = _int_key_deltas(profile.curve_deltas)
errors = validate_write(deltas, cfg.max_delta_khz)
if errors:
errs.append("Curve: " + "; ".join(errors))
else:
if cfg.auto_snapshot:
await _run(
snapshot_save,
gpu,
g_state["gpu_name"],
cfg.snapshot_dir,
cfg.max_snapshots,
)
ret, desc = await _run(write_offsets, gpu, deltas)
if ret != 0:
errs.append(f"Curve write failed ({ret}): {desc}")
else:
await _run(reset_offsets, gpu)
await _update_offsets_and_broadcast(gpu_index)
# Apply fan curve if present in profile (persists it so it survives restarts);
# otherwise deactivate any active fan curve (and clear its persistence).
if profile.fan_curve:
ok, msg = validate_curve(profile.fan_curve)
if not ok:
errs.append(f"Fan curve: {msg}")
else:
await _activate_fan_curve(gpu_index, profile.fan_curve)
elif g_state.get("fan_curve_active"):
await _deactivate_fan_curve(gpu_index, reset_hardware=True)
if not errs:
g_state["active_profile"] = name
return errs
@app.post("/api/profiles/{name}/apply")
async def api_profile_apply(name: str, gpu_index: int = 0):
"""Apply a saved profile to hardware (curve deltas + limits)."""
_require_gpu(gpu_index)
try:
errs = await _apply_profile(name, gpu_index)
except FileNotFoundError as err:
raise HTTPException(
status_code=404, detail=f"Profile '{name}' not found"
) from err
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to load profile: {e}"
) from e
if errs:
raise HTTPException(status_code=500, detail="; ".join(errs))
return {"ok": True}
@app.delete("/api/profiles/{name}")
async def api_profile_delete(name: str):
"""Delete a saved profile by name."""
cfg: Config = _state["config"]
ok = await _run(delete_profile, cfg.profile_dir, name)
if not ok:
raise HTTPException(status_code=404, detail=f"Profile '{name}' not found")
for g_state in _state["gpus"].values():
if g_state["active_profile"] == name:
g_state["active_profile"] = None
changed = any(v == name for v in cfg.auto_load_profiles.values())
if changed:
cfg.auto_load_profiles = {
k: v for k, v in cfg.auto_load_profiles.items() if v != name
}
_persist_auto_load_profiles(cfg.auto_load_profiles)
return {"ok": True}
@app.post("/api/profiles/{name}/rename")
async def api_profile_rename(name: str, req: ProfileRenameRequest):
"""Rename a profile."""
cfg: Config = _state["config"]
if not req.new_name.strip():
raise HTTPException(status_code=400, detail="New name cannot be empty")
ok = await _run(rename_profile, cfg.profile_dir, name, req.new_name.strip())
if not ok:
raise HTTPException(status_code=404, detail=f"Profile '{name}' not found")
for g_state in _state["gpus"].values():
if g_state["active_profile"] == name:
g_state["active_profile"] = req.new_name.strip()
changed = any(v == name for v in cfg.auto_load_profiles.values())
if changed:
cfg.auto_load_profiles = {
k: (req.new_name.strip() if v == name else v)
for k, v in cfg.auto_load_profiles.items()
}
_persist_auto_load_profiles(cfg.auto_load_profiles)
return {"ok": True}
@app.get("/api/config")
async def api_config_get(gpu_index: int = 0):
"""Get mutable server configuration for a specific GPU."""
cfg: Config = _state["config"]
return {"auto_load_profile": cfg.auto_load_profiles.get(_gpu_stable_key(gpu_index))}
@app.post("/api/config")
async def api_config_update(req: ConfigUpdateRequest):
"""Update mutable server configuration. Changes persist to /etc/nvcurve/config.json if present."""
if req.gpu_index not in _state["gpus"]:
raise HTTPException(status_code=404, detail=f"GPU {req.gpu_index} not found")
cfg: Config = _state["config"]
key = _gpu_stable_key(req.gpu_index)
if req.auto_load_profile:
cfg.auto_load_profiles[key] = req.auto_load_profile
else:
cfg.auto_load_profiles.pop(key, None)
_persist_auto_load_profiles(cfg.auto_load_profiles)
return {"ok": True, "auto_load_profile": cfg.auto_load_profiles.get(key)}
@app.get("/api/limits")
async def api_limits(gpu_index: int = 0):
"""Current performance limits: power and clock offsets."""
power = await _run(get_power_limit, gpu_index)
offsets = await _run(get_clock_offsets, gpu_index)
mem_off_range = await _run(get_mem_offset_range, gpu_index)
return {
**power,
**offsets, # gpc_offset_mhz, mem_offset_mhz
**mem_off_range, # min_mem_offset_mhz, max_mem_offset_mhz
}
@app.post("/api/limits")
async def api_limits_update(req: LimitsRequest, gpu_index: int = 0):
"""Update performance limits."""
g_state = _get_gpu_state(gpu_index)
errs = []
if req.power_limit_w is not None:
ok, msg = await _run(set_power_limit, req.power_limit_w, gpu_index)
if not ok:
errs.append(f"Power Limit: {msg}")
if req.mem_offset_mhz is not None:
ok, msg = await _run(set_clock_offsets, None, req.mem_offset_mhz, gpu_index)
if not ok:
errs.append(f"Mem Offset: {msg}")
else:
# Setting mem offset may reset the GPC/curve table as a driver side-effect.
# Re-apply the last known curve offsets to restore them.
await _reapply_curve(gpu_index)
if errs:
raise HTTPException(status_code=500, detail="; ".join(errs))
g_state["active_profile"] = None
return {"ok": True}
async def _reapply_curve(gpu_index: int) -> None:
"""Re-write the last known V/F curve offsets to hardware and notify WS clients."""
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
last = g_state["last_offsets"]
if gpu is None or not last:
return
deltas = {i: off for i, off in enumerate(last) if off != 0}
if not deltas:
return
try:
await _run(write_offsets, gpu, deltas)
await _update_offsets_and_broadcast(gpu_index)
except Exception as exc:
log.warning("_reapply_curve: %s", exc)
async def _update_offsets_and_broadcast(gpu_index: int) -> None:
"""Re-read curve offsets, update the reconciliation baseline, and push to WS clients.
When curve WS clients are connected, a single read_curve call covers both
updating the baseline and the broadcast payload — avoiding a redundant
read_clock_offsets (ClockBoostTable) call that would otherwise happen first.
"""
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
if gpu is None:
return
if g_state["curve_clients"]:
state, _ = await _run(read_curve, gpu, g_state["gpu_name"])
if state:
g_state["last_offsets"] = [p.delta_khz for p in state.points]
await _broadcast(g_state["curve_clients"], _curve_state_dict(state))
else:
offsets, _ = await _run(read_clock_offsets, gpu)
g_state["last_offsets"] = offsets
@app.post("/api/limits/reset")
async def api_limits_reset(gpu_index: int = 0):
"""Reset power limit to hardware default and memory clock offset to 0."""
g_state = _get_gpu_state(gpu_index)
errs = []
power = await _run(get_power_limit, gpu_index)
default_w = power.get("default_power_limit_w")
if default_w is not None:
ok, msg = await _run(set_power_limit, default_w, gpu_index)
if not ok:
errs.append(f"Power Limit: {msg}")
ok, msg = await _run(set_clock_offsets, None, 0, gpu_index)
if not ok:
errs.append(f"Mem Offset: {msg}")
else:
await _reapply_curve(gpu_index)
if errs:
raise HTTPException(status_code=500, detail="; ".join(errs))
g_state["active_profile"] = None
return {"ok": True}
# ── Fan endpoints ──────────────────────────────────────────────────────────────
@app.get("/api/fans")
async def api_fans(gpu_index: int = 0):
"""Current fan state: fan %, curve, and whether curve control is active."""
_get_gpu_state(gpu_index)
g_state = _state["gpus"][gpu_index]
info = await _run(get_fan_info, gpu_index)
curve_active = g_state.get("fan_curve_active", False)
return {
**info,
"fan_mode": "curve" if curve_active else "auto",
"curve": g_state.get("fan_curve"),
"curve_active": curve_active,
}
@app.post("/api/fans")
async def api_fans_update(req: FanCurveRequest, gpu_index: int = 0):
"""Set or update the fan curve. Starts the fan control poller."""
g_state = _get_gpu_state(gpu_index)
curve_data = [{"temp_c": p.temp_c, "fan_pct": p.fan_pct} for p in req.curve]
ok, msg = validate_curve(curve_data)
if not ok:
raise HTTPException(status_code=400, detail=msg)
# Test that fan control is available on this GPU, probing with the
# target for the *current* temperature so the fan is never briefly
# set to an inappropriate speed.
if curve_data:
sample = await _run(poll, g_state["gpu"], gpu_index)
test_temp = (
sample.temp_c
if sample and sample.temp_c is not None
else curve_data[0]["temp_c"]
)
target = interpolate_fan_speed(curve_data, test_temp)
if target is not None:
fan_ok, fan_msg = await _run(set_fan_speed, gpu_index, target)
if not fan_ok:
raise HTTPException(
status_code=500, detail=f"Fan control not available: {fan_msg}"
)
# Activate the curve (starts the poller) and persist it so it survives restarts.
await _activate_fan_curve(gpu_index, curve_data)
return {"ok": True}
@app.post("/api/fans/reset")
async def api_fans_reset(gpu_index: int = 0):
"""Deactivate fan curve control and restore automatic fan mode."""
_get_gpu_state(gpu_index)
# Stop the poller, clear state, restore automatic fan control, and clear
# the persisted curve so it is not re-applied on the next server start.
await _deactivate_fan_curve(gpu_index, reset_hardware=True)
return {"ok": True}
@app.post("/api/fans/speed")
async def api_fans_speed(req: FanSpeedRequest, gpu_index: int = 0):
"""One-shot set fan to an exact percentage (bypasses curve)."""
_get_gpu_state(gpu_index)
pct = max(0, min(100, req.fan_pct))
ok, msg = await _run(set_fan_speed, gpu_index, pct)
if not ok:
raise HTTPException(status_code=500, detail=msg)
return {"ok": True}
# ── Write endpoints ────────────────────────────────────────────────────────────
async def _reconcile_check(gpu_index: int) -> dict | None:
"""Re-read current offsets and return a warning dict if they differ from our last known state.
Returns None if no external change detected (or no baseline).
"""
g_state = _get_gpu_state(gpu_index)
gpu = g_state["gpu"]
last = g_state["last_offsets"]
if last is None:
return None
current, err = await _run(read_clock_offsets, gpu)
if current is None:
return None # Can't read — let the write attempt proceed
changed = [i for i, (a, b) in enumerate(zip(last, current, strict=False)) if a != b]
if not changed:
return None
# External tool changed the curve — active profile is no longer current.
g_state["active_profile"] = None
return {
"warning": "external_change_detected",
"message": (
f"{len(changed)} point(s) changed since last read "
f"(e.g. by LACT, nvidia-smi, or another tool). "
"The write will proceed using the current hardware state."
),
"changed_points": changed[:20], # cap list for readability
}
@app.post("/api/curve/write")
async def api_curve_write(req: WriteRequest, gpu_index: int = 0):
"""Write per-point frequency offsets. {deltas: {point_index: delta_kHz}}"""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
vfp_state, _ = await _run(read_curve, gpu, g_state["gpu_name"])
effective_limit = (
req.max_delta_khz if req.max_delta_khz is not None else cfg.max_delta_khz
)
errors = validate_write(req.deltas, effective_limit)
if errors:
raise HTTPException(status_code=400, detail={"errors": errors})
# Check for negative-freq warnings before writing (best-effort, non-blocking)
freq_warnings: list[str] = []
if vfp_state:
vfp_freqs = [p.freq_khz for p in vfp_state.points]
freq_warnings = check_negative_freq_warnings(
req.deltas, vfp_freqs, g_state["last_offsets"] or []
)
async with g_state["write_lock"]:
warning = await _reconcile_check(gpu_index)
if cfg.auto_snapshot:
await _run(
snapshot_save,
gpu,
g_state["gpu_name"],
cfg.snapshot_dir,
cfg.max_snapshots,
)
ret, desc = await _run(write_offsets, gpu, req.deltas)
if ret != 0:
raise HTTPException(status_code=500, detail=f"Write failed ({ret}): {desc}")
# Update baseline and push curve update to WS clients
await _update_offsets_and_broadcast(gpu_index)
g_state["active_profile"] = None
result = {"ok": True, "return_code": ret, "description": desc}
if warning:
result["warning"] = warning
if freq_warnings:
result["freq_warnings"] = freq_warnings
return result
@app.post("/api/curve/write/global")
async def api_curve_write_global(req: GlobalOffsetRequest, gpu_index: int = 0):
"""Apply a uniform frequency offset to all curve points."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
vfp_state, _ = await _run(read_curve, gpu, g_state["gpu_name"])
if not vfp_state:
raise HTTPException(status_code=500, detail="Failed to read curve")
all_deltas = {p.index: req.delta_khz for p in vfp_state.points if p.domain == "gpu"}
effective_limit = (
req.max_delta_khz if req.max_delta_khz is not None else cfg.max_delta_khz
)
errors = validate_write(all_deltas, effective_limit)
if errors:
raise HTTPException(status_code=400, detail={"errors": errors})
freq_warnings: list[str] = []
if vfp_state:
vfp_freqs = [p.freq_khz for p in vfp_state.points]
freq_warnings = check_negative_freq_warnings(
all_deltas, vfp_freqs, g_state["last_offsets"] or []
)
async with g_state["write_lock"]:
warning = await _reconcile_check(gpu_index)
if cfg.auto_snapshot:
await _run(
snapshot_save,
gpu,
g_state["gpu_name"],
cfg.snapshot_dir,
cfg.max_snapshots,
)
ret, desc = await _run(write_global_offset, gpu, req.delta_khz)
if ret != 0:
raise HTTPException(status_code=500, detail=f"Write failed ({ret}): {desc}")
await _update_offsets_and_broadcast(gpu_index)
g_state["active_profile"] = None
result = {"ok": True, "return_code": ret, "description": desc}
if warning:
result["warning"] = warning
if freq_warnings:
result["freq_warnings"] = freq_warnings
return result
@app.post("/api/curve/reset")
async def api_curve_reset(gpu_index: int = 0):
"""Reset all frequency offsets to zero."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
async with g_state["write_lock"]:
warning = await _reconcile_check(gpu_index)
if cfg.auto_snapshot:
await _run(
snapshot_save,
gpu,
g_state["gpu_name"],
cfg.snapshot_dir,
cfg.max_snapshots,
)
ret, desc = await _run(reset_offsets, gpu)
if ret != 0:
raise HTTPException(status_code=500, detail=f"Reset failed ({ret}): {desc}")
await _update_offsets_and_broadcast(gpu_index)
g_state["active_profile"] = None
result = {"ok": True, "return_code": ret, "description": desc}
if warning:
result["warning"] = warning
return result
@app.post("/api/curve/verify")
async def api_curve_verify(req: VerifyRequest, gpu_index: int = 0):
"""Write-verify-read cycle. Returns per-point match results and collateral changes."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
errors = validate_write(req.deltas, cfg.max_delta_khz)
if errors:
raise HTTPException(status_code=400, detail={"errors": errors})
before_offsets, err = await _run(read_clock_offsets, gpu)
if before_offsets is None:
raise HTTPException(
status_code=500, detail=f"Failed to read current state: {err}"
)
# Always snapshot before verify — it's a testing operation
await _run(
snapshot_save, gpu, g_state["gpu_name"], cfg.snapshot_dir, cfg.max_snapshots
)
async with g_state["write_lock"]:
ret, desc = await _run(write_offsets, gpu, req.deltas)
if ret != 0:
raise HTTPException(status_code=500, detail=f"Write failed ({ret}): {desc}")
await asyncio.sleep(0.2)
after_offsets, err = await _run(read_clock_offsets, gpu)
if after_offsets is None:
raise HTTPException(
status_code=500, detail=f"Verification read failed: {err}"
)
g_state["active_profile"] = None
await _update_offsets_and_broadcast(gpu_index)
points_result = []
all_matched = True
for point, expected in sorted(req.deltas.items()):
actual = after_offsets[point]
match = actual == expected
if not match:
all_matched = False
points_result.append(
{
"point": point,
"expected_khz": expected,
"actual_khz": actual,
"match": match,
}
)
collateral = [
{"point": i, "before_khz": before_offsets[i], "after_khz": after_offsets[i]}
for i in range(len(before_offsets))
if i not in req.deltas and before_offsets[i] != after_offsets[i]
]
return {
"ok": all_matched and not collateral,
"all_matched": all_matched,
"no_side_effects": not collateral,
"return_code": ret,
"description": desc,
"points": points_result,
"collateral_changes": collateral,
}
@app.post("/api/shutdown")
async def api_shutdown():
"""Gracefully shut down the server process."""
import os
import signal
loop = asyncio.get_running_loop()
loop.call_later(0.1, lambda: os.kill(os.getpid(), signal.SIGTERM))
return {"ok": True}
@app.post("/api/snapshot/save")
async def api_snapshot_save(gpu_index: int = 0):
"""Save a ClockBoostTable snapshot."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
path = await _run(
snapshot_save, gpu, g_state["gpu_name"], cfg.snapshot_dir, cfg.max_snapshots
)
if path is None:
raise HTTPException(status_code=500, detail="Failed to save snapshot")
return {"ok": True, "filepath": path}
@app.post("/api/snapshot/restore")
async def api_snapshot_restore(req: SnapshotRestoreRequest, gpu_index: int = 0):
"""Restore a ClockBoostTable snapshot. Uses most recent if filepath not specified."""
gpu, g_state = _require_gpu(gpu_index)
cfg: Config = _state["config"]
async with g_state["write_lock"]:
ok = await _run(snapshot_restore, gpu, cfg.snapshot_dir, req.filepath)
if not ok:
raise HTTPException(status_code=500, detail="Failed to restore snapshot")
await _update_offsets_and_broadcast(gpu_index)
g_state["active_profile"] = None
return {"ok": True}
# ── WebSocket endpoints ───────────────────────────────────────────────────────
def _ws_authenticated(ws: WebSocket) -> bool:
"""True if the WebSocket connection is allowed (auth disabled or valid session).
The session token is read from the Authorization header or the cookie
(browsers send the cookie automatically on the WS handshake). The token is
deliberately NOT accepted via a query string, since uvicorn's access log
records the full path including the query string.
"""
cfg: Config = _state["config"]
if not auth.auth_enabled(cfg.users_file):
return True
return auth.get_session(auth.extract_token(ws)) is not None
@app.websocket("/ws/monitor")
async def ws_monitor(ws: WebSocket):
"""Stream MonitoringSample at poll_interval_s. Clients receive JSON objects."""
if not _ws_authenticated(ws):
await ws.close(code=1008)
return
await ws.accept()
try:
data = await ws.receive_json()
if data.get("action") != "subscribe":
await ws.close()
return
gpu_index = data.get("gpu_index", 0)
except WebSocketDisconnect:
return
except Exception:
await ws.close()
return
g_state = _state["gpus"].get(gpu_index)
if not g_state:
await ws.close()
return
g_state["monitor_clients"].add(ws)
try:
gpu = g_state["gpu"]
if gpu is not None:
sample = await _run(poll, gpu, gpu_index)
await ws.send_json(_sample_dict(sample))
while True:
await ws.receive_text()
except WebSocketDisconnect:
pass
except Exception:
log.debug("monitor ws client error", exc_info=True)
finally:
g_state["monitor_clients"].discard(ws)
@app.websocket("/ws/curve")
async def ws_curve(ws: WebSocket):
"""Push CurveState whenever the curve changes (after writes)."""
if not _ws_authenticated(ws):
await ws.close(code=1008)
return
await ws.accept()
try:
data = await ws.receive_json()
if data.get("action") != "subscribe":
await ws.close()
return
gpu_index = data.get("gpu_index", 0)
except WebSocketDisconnect:
return
except Exception:
await ws.close()
return
g_state = _state["gpus"].get(gpu_index)
if not g_state:
await ws.close()
return
g_state["curve_clients"].add(ws)
try:
gpu = g_state["gpu"]
if gpu is not None:
state, _ = await _run(read_curve, gpu, g_state["gpu_name"])
if state:
await ws.send_json(_curve_state_dict(state))
while True:
await ws.receive_text()
except WebSocketDisconnect:
pass
except Exception:
log.debug("curve ws client error", exc_info=True)
finally:
g_state["curve_clients"].discard(ws)
# ── Frontend SPA ──────────────────────────────────────────────────────────────
# When set, suppresses the auto-open browser behaviour so the dev can open
# the Vite dev server (pnpm dev) manually instead.
_DEV_PORT = os.environ.get("NVCURVE_DEV_PORT")
def _resolve_dist_dir() -> str:
"""Resolve the frontend dist directory to a string path.
Prefers the packaged location (importlib.resources), then falls back to
the project-root layout used during local development.
"""
try:
# Project requires Python >= 3.12, so the 3.7-compat finding is a false positive.
from importlib.resources import ( # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2
files as _resource_files,
)
candidate = _resource_files("nvcurve") / "frontend" / "dist"
if candidate.is_dir():
return str(candidate)
except (ImportError, TypeError):
pass
here = Path(__file__).parent
for base in (here, here.parent):
dist = base / "frontend" / "dist"
if dist.is_dir():
return str(dist)
return str(here / "frontend" / "dist")
_dist_dir = _resolve_dist_dir()
if os.path.isdir(os.path.join(_dist_dir, "assets")):
app.mount(
"/assets",
StaticFiles(directory=os.path.join(_dist_dir, "assets")),
name="assets",
)
@app.get("/{catchall:path}")
async def serve_spa(catchall: str):
if catchall.startswith(("api/", "ws/")):
raise HTTPException(status_code=404, detail="Not Found")
if not os.path.isdir(_dist_dir):
return {"error": "Frontend not built. Run pnpm build in frontend/."}
path = os.path.join(_dist_dir, catchall)
if os.path.isfile(path) and catchall:
return FileResponse(path)
index = os.path.join(_dist_dir, "index.html")
if os.path.isfile(index):
return FileResponse(index)
raise HTTPException(status_code=404, detail="Not Found")
# ── Factory for configured app ────────────────────────────────────────────────
def create_app(config: Config = default_config) -> FastAPI:
"""Create a server app with a custom config (e.g. different gpu_index)."""
_state["config"] = config
return app
def run(
host: str = "127.0.0.1",
port: int = 8042,
gpu_index: int = 0,
config: Config = default_config,
open_browser: bool = False,
) -> None:
"""Start the uvicorn server. Blocking."""
import socket
import threading
import uvicorn
_state["config"] = config
# Suppress noisy websockets keepalive ping-timeout tracebacks — these are
# normal disconnection events (browser tab closed, network hiccup) and
# logging them at ERROR level creates false alarm noise.
logging.getLogger("websockets").setLevel(logging.CRITICAL)
# Fail fast if the port is already in use — silently shifting ports breaks
# client discovery. Users should configure a different port explicitly.
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
s.bind((host, port))
except OSError:
print(f"Error: port {port} is already in use.")
print(
f"Use --port N to specify a different port, or free port {port} first."
)
return
url = f"http://{host}:{port}"
# Print banner *before* uvicorn starts so it appears above uvicorn's own output.
# GPU name is populated by the lifespan; we omit it here since the server
# hasn't started yet, and the lifespan logs it via log.info.
print("\033[1;36m" + "─" * 60 + "\033[0m")
print("\033[1;32m" + " NVCurve".center(60) + "\033[0m")
print(f" {url}".center(60))
print("\033[1;36m" + "─" * 60 + "\033[0m")
print(" Press Ctrl+C to stop.")
print()
if open_browser and not _DEV_PORT:
threading.Timer(1.2, lambda: _open_browser_as_user(url)).start()
uvicorn.run(app, host=host, port=port, log_level="warning", access_log=False)