"""Apple Silicon hardware detection and telemetry (macOS, graceful degradation elsewhere)."""

from __future__ import annotations

import asyncio
import platform
import re
import shutil
import subprocess
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

import psutil

GB = 1024**3


def _run(cmd: list[str], timeout: float = 5.0) -> str:
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout
    except Exception:
        return ""


def _sysctl(key: str) -> str:
    return _run(["sysctl", "-n", key]).strip()


@dataclass
class Hardware:
    chip: str
    memory_gb: float
    memory_bytes: int
    cpu_cores: int
    performance_cores: int | None
    efficiency_cores: int | None
    gpu_cores: int | None
    os: str
    os_version: str
    hostname: str
    apple_silicon: bool
    disk_total_gb: float
    python: str

    def to_dict(self) -> dict:
        return asdict(self)


_HW_CACHE: Hardware | None = None


def detect_hardware(models_dir: Path | None = None) -> Hardware:
    global _HW_CACHE
    if _HW_CACHE is not None:
        return _HW_CACHE
    is_mac = platform.system() == "Darwin"
    chip = _sysctl("machdep.cpu.brand_string") if is_mac else platform.processor() or "unknown"
    mem_bytes = int(_sysctl("hw.memsize") or psutil.virtual_memory().total) if is_mac else psutil.virtual_memory().total
    ncpu = int(_sysctl("hw.ncpu") or psutil.cpu_count() or 0) if is_mac else (psutil.cpu_count() or 0)
    perf = eff = None
    if is_mac:
        p0 = _sysctl("hw.perflevel0.physicalcpu")
        p1 = _sysctl("hw.perflevel1.physicalcpu")
        perf = int(p0) if p0.isdigit() else None
        eff = int(p1) if p1.isdigit() else None
    gpu_cores = None
    if is_mac:
        out = _run(["ioreg", "-r", "-c", "IOAccelerator", "-d", "1"])
        m = re.search(r'"gpu-core-count"\s*=\s*(\d+)', out)
        if m:
            gpu_cores = int(m.group(1))
    disk_path = models_dir if (models_dir and models_dir.exists()) else Path.home()
    du = shutil.disk_usage(disk_path)
    os_version = ""
    if is_mac:
        os_version = _run(["sw_vers", "-productVersion"]).strip()
    _HW_CACHE = Hardware(
        chip=chip or "unknown",
        memory_gb=round(mem_bytes / GB, 1),
        memory_bytes=mem_bytes,
        cpu_cores=ncpu,
        performance_cores=perf,
        efficiency_cores=eff,
        gpu_cores=gpu_cores,
        os="macOS" if is_mac else platform.system(),
        os_version=os_version or platform.release(),
        hostname=platform.node(),
        apple_silicon=is_mac and platform.machine() == "arm64",
        disk_total_gb=round(du.total / GB, 1),
        python=platform.python_version(),
    )
    return _HW_CACHE


@dataclass
class Telemetry:
    ts: float
    mem_total_gb: float
    mem_used_gb: float
    mem_available_gb: float
    mem_wired_gb: float | None
    mem_compressed_gb: float | None
    mem_pressure_percent: int | None  # 0 = none … 100 = critical (derived from free %)
    mem_pressure_level: str  # normal | warning | critical
    swap_used_gb: float
    swap_total_gb: float
    cpu_percent: float
    cpu_per_core: list[float] = field(default_factory=list)
    load_avg: list[float] = field(default_factory=list)
    gpu_percent: float | None = None
    gpu_renderer_percent: float | None = None
    gpu_memory_gb: float | None = None
    thermal_state: str = "nominal"
    thermal_cpu_speed_limit: int | None = None
    disk_total_gb: float = 0
    disk_used_gb: float = 0
    disk_free_gb: float = 0
    uptime_seconds: float = 0
    process_rss_gb: float = 0

    def to_dict(self) -> dict:
        return asdict(self)


def _read_gpu() -> tuple[float | None, float | None, float | None]:
    out = _run(["ioreg", "-r", "-c", "IOAccelerator", "-d", "1"], timeout=3)
    if not out:
        return None, None, None
    dev = re.search(r'"Device Utilization %"=(\d+)', out)
    rend = re.search(r'"Renderer Utilization %"=(\d+)', out)
    mem = re.search(r'"In use system memory"=(\d+)', out)
    return (
        float(dev.group(1)) if dev else None,
        float(rend.group(1)) if rend else None,
        round(int(mem.group(1)) / GB, 2) if mem else None,
    )


def _read_thermal() -> tuple[str, int | None]:
    out = _run(["pmset", "-g", "therm"], timeout=3)
    if not out:
        return "unknown", None
    m = re.search(r"CPU_Speed_Limit\s*=\s*(\d+)", out)
    limit = int(m.group(1)) if m else None
    if limit is not None and limit < 100:
        state = "throttled" if limit < 80 else "warm"
    elif "No thermal warning" in out or limit == 100:
        state = "nominal"
    else:
        state = "nominal"
    return state, limit


def _read_vm_stat() -> dict[str, int]:
    out = _run(["vm_stat"], timeout=3)
    vals: dict[str, int] = {}
    if not out:
        return vals
    m = re.search(r"page size of (\d+) bytes", out)
    page = int(m.group(1)) if m else 16384
    for line in out.splitlines()[1:]:
        if ":" not in line:
            continue
        k, v = line.split(":", 1)
        v = v.strip().rstrip(".")
        if v.isdigit():
            vals[k.strip()] = int(v) * page
    vals["_page"] = page
    return vals


def sample_telemetry(models_dir: Path | None = None) -> Telemetry:
    vm = psutil.virtual_memory()
    sw = psutil.swap_memory()
    is_mac = platform.system() == "Darwin"
    wired = compressed = None
    if is_mac:
        vs = _read_vm_stat()
        if vs:
            wired = round(vs.get("Pages wired down", 0) / GB, 2)
            compressed = round(vs.get("Pages occupied by compressor", 0) / GB, 2)
    gpu, gpu_r, gpu_mem = _read_gpu() if is_mac else (None, None, None)
    therm, limit = _read_thermal() if is_mac else ("unknown", None)
    disk_path = models_dir if (models_dir and models_dir.exists()) else Path.home()
    du = shutil.disk_usage(disk_path)
    avail_pct = vm.available / vm.total * 100
    if avail_pct > 20 and sw.used < 1 * GB:
        level = "normal"
    elif avail_pct > 8 and sw.used < 4 * GB:
        level = "warning"
    else:
        level = "critical"
    pressure = int(max(0.0, min(100.0, 100 - avail_pct)))
    proc = psutil.Process()
    try:
        load = list(psutil.getloadavg())
    except Exception:
        load = []
    return Telemetry(
        ts=time.time(),
        mem_total_gb=round(vm.total / GB, 2),
        mem_used_gb=round((vm.total - vm.available) / GB, 2),
        mem_available_gb=round(vm.available / GB, 2),
        mem_wired_gb=wired,
        mem_compressed_gb=compressed,
        mem_pressure_percent=pressure,
        mem_pressure_level=level,
        swap_used_gb=round(sw.used / GB, 2),
        swap_total_gb=round(sw.total / GB, 2),
        cpu_percent=psutil.cpu_percent(interval=None),
        cpu_per_core=psutil.cpu_percent(interval=None, percpu=True),
        load_avg=[round(x, 2) for x in load],
        gpu_percent=gpu,
        gpu_renderer_percent=gpu_r,
        gpu_memory_gb=gpu_mem,
        thermal_state=therm,
        thermal_cpu_speed_limit=limit,
        disk_total_gb=round(du.total / GB, 1),
        disk_used_gb=round(du.used / GB, 1),
        disk_free_gb=round(du.free / GB, 1),
        uptime_seconds=round(time.time() - psutil.boot_time(), 0),
        process_rss_gb=round(proc.memory_info().rss / GB, 3),
    )


async def sample_telemetry_async(models_dir: Path | None = None) -> Telemetry:
    return await asyncio.to_thread(sample_telemetry, models_dir)


def process_tree_rss_bytes(pid: int) -> int:
    """RSS of a process and its children (worker memory)."""
    try:
        p = psutil.Process(pid)
        total = p.memory_info().rss
        for c in p.children(recursive=True):
            try:
                total += c.memory_info().rss
            except psutil.Error:
                pass
        return total
    except psutil.Error:
        return 0


def process_footprint_bytes(pid: int) -> int:
    """macOS 'physical footprint' (closest to Activity Monitor 'Memory'). Falls back to RSS."""
    if platform.system() == "Darwin":
        out = _run(["footprint", "-p", str(pid)], timeout=3) if shutil.which("footprint") else ""
        m = re.search(r"phys_footprint:\s*([\d.]+)\s*([KMG]B)", out)
        if m:
            mult = {"KB": 1024, "MB": 1024**2, "GB": 1024**3}[m.group(2)]
            return int(float(m.group(1)) * mult)
    return process_tree_rss_bytes(pid)