ARIA
e148c83622
Adds an experimental power-cap mode using the undocumented RM ioctl interface (based on panchovix's LACT PR #1205) to set power limits below the VBIOS minimum (down to 30 W). - hal/rm_power.py: RM ioctl power-cap read/write/reset + runtime probe - limits.py: power_cap_mode (nvml/ioctl) with support detection - config.py: persist power_cap_mode per GPU - profiles: record/apply power_cap_mode - server.py: POST /api/limits validates ioctl support (409 on failure) - cli.py: profile save falls back to persisted mode - client.py: power_cap_mode in Limits - frontend: toggle + warning with panchovix attribution (LACT #1205) - tests: test_rm_power.py (unit) + integration coverage - Makefile: add test_rm_power.py to make test Also includes automated linter reformatting (prettier, ruff, shellcheck, isort, markdownlint) that the linter would apply anyway.
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
"""Native profile storage and schema."""
|
|
|
|
import glob
|
|
import json
|
|
import logging
|
|
import os
|
|
from dataclasses import asdict, dataclass
|
|
|
|
log = logging.getLogger("nvcurve.profiles.native")
|
|
|
|
|
|
@dataclass
|
|
class ProfileData:
|
|
name: str
|
|
gpu_name: str
|
|
curve_deltas: dict[str, int] # { "index": delta_khz }
|
|
mem_offset_mhz: int | None = None
|
|
power_limit_w: int | None = None
|
|
# How power_limit_w is applied: "nvml" (default) or "ioctl" (experimental
|
|
# RM power control, permits values below the VBIOS minimum).
|
|
power_cap_mode: str | None = None
|
|
fan_curve: list[dict[str, int]] | None = None
|
|
# Fan indices controlled by fan_curve (0-based); None = all fans.
|
|
fan_targets: list[int] | None = None
|
|
|
|
|
|
def save_profile(profile_dir: str, data: ProfileData) -> str:
|
|
"""Save profile to JSON, sanitising the filename."""
|
|
try:
|
|
os.makedirs(profile_dir, exist_ok=True)
|
|
except OSError as exc:
|
|
raise RuntimeError(f"Cannot create profile dir {profile_dir}: {exc}") from exc
|
|
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")
|
|
try:
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
json.dump(asdict(data), f, indent=2)
|
|
except OSError as exc:
|
|
raise RuntimeError(f"Cannot write profile {filepath}: {exc}") from exc
|
|
return filepath
|
|
|
|
|
|
def load_profile(filepath: str) -> ProfileData:
|
|
"""Load profile from JSON."""
|
|
try:
|
|
with open(filepath, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except FileNotFoundError:
|
|
raise
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise RuntimeError(f"Cannot read profile {filepath}: {exc}") from exc
|
|
# 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)
|
|
# Normalize the experimental power-cap mode; unknown values fall back to NVML.
|
|
if data.get("power_cap_mode") not in (None, "nvml", "ioctl"):
|
|
data["power_cap_mode"] = 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 exc:
|
|
log.debug("Skipping unreadable profile %s: %s", fp, exc)
|
|
# 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
|