| """Gerçek zamanlı GPU durumu: nvidia-smi (NVIDIA) + Ollama /api/ps (modelin GPU payı). |
| |
| nvidia-smi, uygulama doğrudan bilgisayarda çalışırken PATH'ten; Docker'da ise NVIDIA konteyner çalışma |
| zamanının (docker-compose.gpu.yml) konteynere eklediği ikiliden okunur. Bulunamazsa yalnızca Ollama bilgisi |
| döner. Sonuç kısa süre önbelleğe alınır; böylece birden çok sekme sorgulasa da nvidia-smi saniyede en fazla |
| bir kez çalışır. |
| """ |
| from __future__ import annotations |
|
|
| import shutil |
| import subprocess |
| import threading |
| import time |
| from typing import Any, Optional |
|
|
| FIELDS = [ |
| ("index", int), ("name", str), ("driver_version", str), ("utilization.gpu", float), |
| ("utilization.memory", float), ("memory.used", float), ("memory.total", float), ("temperature.gpu", float), |
| ("power.draw", float), ("power.limit", float), ("clocks.sm", float), ("clocks.max.sm", float), |
| ("fan.speed", float), ("pstate", str), |
| ] |
| KEYS = {"utilization.gpu": "util", "utilization.memory": "mem_util", "memory.used": "mem_used_mib", |
| "memory.total": "mem_total_mib", "temperature.gpu": "temp_c", "power.draw": "power_w", |
| "power.limit": "power_limit_w", "clocks.sm": "clock_mhz", "clocks.max.sm": "clock_max_mhz", |
| "fan.speed": "fan_pct", "driver_version": "driver"} |
|
|
| _lock = threading.Lock() |
| _cache: dict[str, Any] = {"ts": 0.0, "data": None} |
|
|
|
|
| def _num(v: str, cast): |
| v = v.strip() |
| if not v or v.startswith("[") or v.lower() in ("n/a", "not supported"): |
| return None |
| try: |
| return cast(v) |
| except ValueError: |
| return None |
|
|
|
|
| def parse_nvidia_smi(text: str) -> list[dict[str, Any]]: |
| out = [] |
| for line in text.strip().splitlines(): |
| parts = [p.strip() for p in line.split(",")] |
| if len(parts) < len(FIELDS): |
| continue |
| g: dict[str, Any] = {} |
| for (field, cast), raw in zip(FIELDS, parts): |
| g[KEYS.get(field, field)] = _num(raw, cast) if cast is not str else (raw or None) |
| out.append(g) |
| return out |
|
|
|
|
| def query_nvidia(binary: Optional[str] = None, timeout: float = 3.0) -> tuple[list[dict[str, Any]], Optional[str]]: |
| exe = binary or shutil.which("nvidia-smi") |
| if not exe: |
| return [], "nvidia-smi bulunamadı" |
| try: |
| p = subprocess.run([exe, "--query-gpu=" + ",".join(f for f, _ in FIELDS), "--format=csv,noheader,nounits"], |
| capture_output=True, text=True, timeout=timeout) |
| except (OSError, subprocess.TimeoutExpired) as e: |
| return [], str(e) |
| if p.returncode != 0: |
| return [], (p.stderr or p.stdout).strip()[:300] or f"nvidia-smi çıkış kodu {p.returncode}" |
| return parse_nvidia_smi(p.stdout), None |
|
|
|
|
| def status(ollama_running: Optional[list[dict[str, Any]]] = None, max_age: float = 1.0) -> dict[str, Any]: |
| with _lock: |
| now = time.time() |
| if _cache["data"] is not None and now - _cache["ts"] < max_age: |
| gpus, err = _cache["data"] |
| else: |
| gpus, err = query_nvidia() |
| _cache.update(ts=now, data=(gpus, err)) |
| return {"ts": time.time(), "available": bool(gpus), "source": "nvidia-smi" if gpus else None, |
| "gpus": gpus, "error": None if gpus else err, "models": ollama_running or []} |
|
|
|
|
| class EnergyMeter: |
| """İş süresince GPU güç çekişini örnekleyip enerjiyi (Wh) yamuk kuralıyla toplar. |
| |
| Ölçülen değer kartın toplam güç çekişidir (o sırada çalışan başka işler de dahil); deneyler tek tek |
| koşulduğundan iş başına enerji için iyi bir yaklaşımdır. nvidia-smi yoksa ölçüm yapılmaz. |
| """ |
|
|
| def __init__(self, interval: float = 1.0, reader=None) -> None: |
| self.interval = interval |
| self._read = reader or (lambda: status(None, max_age=interval * 0.9)["gpus"]) |
| self._stop = threading.Event() |
| self._thread: Optional[threading.Thread] = None |
| self.wh = 0.0 |
| self.samples = 0 |
| self.util_sum = 0.0 |
| self.peak_mem_mib = 0.0 |
| self._last: Optional[tuple[float, float]] = None |
| self.available = False |
|
|
| def _sample(self) -> None: |
| try: |
| gpus = self._read() |
| except Exception: |
| gpus = [] |
| watts = [g.get("power_w") for g in gpus if g.get("power_w") is not None] |
| if not watts: |
| return |
| self.available = True |
| now, p = time.time(), float(sum(watts)) |
| if self._last: |
| t0, p0 = self._last |
| self.wh += (p0 + p) / 2 * (now - t0) / 3600.0 |
| self._last = (now, p) |
| self.samples += 1 |
| self.util_sum += sum(g.get("util") or 0 for g in gpus) / max(1, len(gpus)) |
| self.peak_mem_mib = max(self.peak_mem_mib, float(sum(g.get("mem_used_mib") or 0 for g in gpus))) |
|
|
| def _loop(self) -> None: |
| while not self._stop.is_set(): |
| self._sample() |
| self._stop.wait(self.interval) |
|
|
| def start(self) -> "EnergyMeter": |
| self._sample() |
| if not self.available: |
| return self |
| self._thread = threading.Thread(target=self._loop, daemon=True) |
| self._thread.start() |
| return self |
|
|
| def stop(self) -> dict[str, Any]: |
| self._stop.set() |
| if self._thread: |
| self._thread.join(timeout=3) |
| self._sample() |
| if not self.available or self.samples < 2: |
| return {} |
| return {"energy_wh": round(self.wh, 3), "gpu_util_avg": round(self.util_sum / self.samples, 1), |
| "gpu_mem_peak_gb": round(self.peak_mem_mib / 1024, 2), "energy_samples": self.samples} |
|
|
