107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from ...config import settings
|
|
|
|
|
|
def parse_nvidia_smi_csv(output: str) -> list[dict]:
|
|
gpus: list[dict] = []
|
|
for line in output.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
parts = [part.strip() for part in line.split(",")]
|
|
if len(parts) < 7:
|
|
continue
|
|
index, name, total, used, free, util, temp = parts[:7]
|
|
try:
|
|
gpus.append(
|
|
{
|
|
"index": int(index),
|
|
"name": name,
|
|
"memory_total_mb": int(total),
|
|
"memory_used_mb": int(used),
|
|
"memory_free_mb": int(free),
|
|
"utilization_gpu_percent": int(util),
|
|
"temperature_c": int(temp),
|
|
}
|
|
)
|
|
except ValueError:
|
|
continue
|
|
return gpus
|
|
|
|
|
|
def get_gpus() -> dict:
|
|
cmd = [
|
|
"nvidia-smi",
|
|
"--query-gpu=index,name,memory.total,memory.used,memory.free,utilization.gpu,temperature.gpu",
|
|
"--format=csv,noheader,nounits",
|
|
]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return {"available": True, "gpus": parse_nvidia_smi_csv(result.stdout)}
|
|
except Exception as exc:
|
|
return {"available": False, "gpus": [], "error": str(exc)}
|
|
|
|
|
|
def get_conda_envs() -> dict:
|
|
try:
|
|
result = subprocess.run(["conda", "env", "list"], capture_output=True, text=True, check=True)
|
|
except Exception as exc:
|
|
return {"available": False, "envs": [], "error": str(exc)}
|
|
envs = []
|
|
for line in result.stdout.splitlines():
|
|
raw = line.strip()
|
|
if not raw or raw.startswith("#"):
|
|
continue
|
|
marker = "*" in raw.split()
|
|
parts = raw.replace("*", " ").split()
|
|
if len(parts) >= 2:
|
|
envs.append({"name": parts[0], "path": parts[-1], "active": marker})
|
|
return {"available": True, "envs": envs, "task_default": settings.task_conda_env}
|
|
|
|
|
|
def disk_usage() -> dict:
|
|
usage = shutil.disk_usage(settings.source_root)
|
|
return {
|
|
"path": str(settings.source_root),
|
|
"total": usage.total,
|
|
"used": usage.used,
|
|
"free": usage.free,
|
|
}
|
|
|
|
|
|
def scan_results() -> list[dict]:
|
|
roots = [
|
|
settings.source_root / "DataSet_Public_outputs",
|
|
settings.source_root / "BestMode_Predict_Results_DataSet_Public",
|
|
settings.source_root / "Hardisk",
|
|
settings.source_root / "Seg_All_In_One_Analysis",
|
|
]
|
|
exts = {".csv", ".png", ".jpg", ".jpeg", ".svg", ".log", ".pth", ".pt"}
|
|
results: list[dict] = []
|
|
for root in roots:
|
|
if not root.exists():
|
|
continue
|
|
for path in root.rglob("*"):
|
|
if path.is_file() and path.suffix.lower() in exts:
|
|
try:
|
|
stat = path.stat()
|
|
results.append(
|
|
{
|
|
"name": path.name,
|
|
"path": str(path.resolve()),
|
|
"relative_path": str(path.resolve().relative_to(settings.source_root)),
|
|
"size": stat.st_size,
|
|
"modified": stat.st_mtime,
|
|
"kind": path.suffix.lower().lstrip("."),
|
|
}
|
|
)
|
|
except OSError:
|
|
continue
|
|
results.sort(key=lambda item: item["modified"], reverse=True)
|
|
return results[:1000]
|
|
|