"""System vitals sampling for the bundled dashboard System tile.

A process-global collector wraps psutil so repeated tile fetches share one
ring buffer of recent samples (sparkline history) and one network-counter
baseline (kbps deltas). All public access goes through ``sample()``.
"""

from __future__ import annotations

import plistlib
import subprocess
import threading
import time
from collections import deque

# ~3 minutes of history at the tile's 5s refresh cadence.
HISTORY_MAXLEN = 36
TOP_PROCESS_COUNT = 5
_PROCESS_NAME_MAX = 40


class SystemVitalsCollector:
    """Thread-safe psutil sampler with sparkline ring buffers."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._history: dict[str, deque[float]] = {
            "cpu": deque(maxlen=HISTORY_MAXLEN),
            "gpu": deque(maxlen=HISTORY_MAXLEN),
            "memory": deque(maxlen=HISTORY_MAXLEN),
            "net_down_kbps": deque(maxlen=HISTORY_MAXLEN),
            "net_up_kbps": deque(maxlen=HISTORY_MAXLEN),
        }
        self._last_net: tuple[float, int, int] | None = None  # (monotonic, bytes_sent, bytes_recv)
        self._primed = False
        self._last_sample: tuple[float, dict] | None = None  # (wall time, sample dict)

    def sample(self) -> dict:
        """Take one sample and return current values + history lists."""
        import psutil

        with self._lock:
            if not self._primed:
                # First cpu_percent call returns 0.0 and sets the baseline.
                psutil.cpu_percent(interval=None)
                self._primed = True

            cpu = float(psutil.cpu_percent(interval=None))
            vm = psutil.virtual_memory()
            disk = psutil.disk_usage("/")

            now = time.monotonic()
            net = psutil.net_io_counters()
            up_kbps = 0.0
            down_kbps = 0.0
            if net is not None:
                if self._last_net is not None:
                    last_t, last_sent, last_recv = self._last_net
                    dt = now - last_t
                    if dt > 0.1:
                        up_kbps = max(0.0, (net.bytes_sent - last_sent) / dt / 1024.0)
                        down_kbps = max(0.0, (net.bytes_recv - last_recv) / dt / 1024.0)
                self._last_net = (now, net.bytes_sent, net.bytes_recv)

            gpu = _read_gpu_percent()
            mem_percent = float(vm.percent)
            self._history["cpu"].append(round(cpu, 1))
            if gpu is not None:
                self._history["gpu"].append(round(gpu, 1))
            self._history["memory"].append(round(mem_percent, 1))
            self._history["net_down_kbps"].append(round(down_kbps, 1))
            self._history["net_up_kbps"].append(round(up_kbps, 1))

            gib = 1024.0**3
            result = {
                "cpu_percent": round(cpu, 1),
                "gpu_percent": round(gpu, 1) if gpu is not None else None,
                "memory_percent": round(mem_percent, 1),
                "memory_used_gb": round(vm.used / gib, 1),
                "memory_total_gb": round(vm.total / gib, 1),
                "disk_percent": round(float(disk.percent), 1),
                "disk_used_gb": round(disk.used / gib, 1),
                "disk_total_gb": round(disk.total / gib, 1),
                "network_up_kbps": round(up_kbps, 1),
                "network_down_kbps": round(down_kbps, 1),
                "history": {k: list(v) for k, v in self._history.items()},
                "processes": _top_processes(psutil),
            }
            self._last_sample = (time.time(), result)
            return result

    def last_sample(self, *, max_age_seconds: float | None = None) -> dict | None:
        """Return the most recent sample without taking a new one (read-only).

        Used by the dashboard chat context injection; never perturbs the
        network-rate baseline or sparkline history. Returns None when no
        sample exists yet or the latest one is older than ``max_age_seconds``.
        """
        with self._lock:
            if self._last_sample is None:
                return None
            ts, data = self._last_sample
            age = max(0.0, time.time() - ts)
            if max_age_seconds is not None and age > max_age_seconds:
                return None
            out = dict(data)
            out["sample_age_seconds"] = round(age, 1)
            return out


def _read_gpu_percent() -> float | None:
    """GPU utilization via IOKit accelerator stats (Apple Silicon, no sudo).

    Reads ``Device Utilization %`` from the IOAccelerator performance
    statistics; returns None when unavailable (Intel iGPU-only configs,
    sandboxed environments, non-macOS).
    """
    try:
        out = subprocess.run(
            ["ioreg", "-r", "-d", "1", "-c", "IOAccelerator", "-a"],
            capture_output=True,
            timeout=3,
        )
        if out.returncode != 0 or not out.stdout:
            return None
        items = plistlib.loads(out.stdout)
        for item in items:
            stats = item.get("PerformanceStatistics") or {}
            value = stats.get("Device Utilization %")
            if isinstance(value, (int, float)):
                return max(0.0, min(100.0, float(value)))
    except Exception:
        return None
    return None


def _top_processes(psutil_mod) -> list[dict]:
    rows: list[dict] = []
    try:
        for proc in psutil_mod.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
            try:
                info = proc.info
            except Exception:
                continue
            name = str(info.get("name") or "").strip()
            if not name:
                continue
            try:
                cpu = float(info.get("cpu_percent") or 0.0)
                mem = float(info.get("memory_percent") or 0.0)
            except (TypeError, ValueError):
                continue
            rows.append(
                {
                    "pid": int(info.get("pid") or 0),
                    "name": name[:_PROCESS_NAME_MAX],
                    "cpu_percent": round(cpu, 1),
                    "memory_percent": round(mem, 1),
                }
            )
    except Exception:
        return rows[:TOP_PROCESS_COUNT]
    rows.sort(key=lambda r: r["cpu_percent"], reverse=True)
    return rows[:TOP_PROCESS_COUNT]


_COLLECTOR: SystemVitalsCollector | None = None
_COLLECTOR_LOCK = threading.Lock()


def get_collector() -> SystemVitalsCollector:
    global _COLLECTOR
    with _COLLECTOR_LOCK:
        if _COLLECTOR is None:
            _COLLECTOR = SystemVitalsCollector()
        return _COLLECTOR


def reset_collector_for_tests() -> None:
    global _COLLECTOR
    with _COLLECTOR_LOCK:
        _COLLECTOR = None
