monitor.py

"""性能日志 + 资源监测 — 永久埋点基础设施。

单例 monitor（ResourceMonitor）：内存 / CPU / GPU 采样，默认开启，可通过
--no-monitor / RVTOL_MONITOR=0 / GUI 复选框关闭。关闭时零线程、零子进程。

设计约束：
- psutil 缺失时 RSS/CPU 字段为 None（一次性警告），GPU 采样不受影响；
- 无 nvidia-smi 时自动降级到 cuda.bindings 的 cuMemGetInfo（仅显存），再失败则跳过 GPU；
- start()/stop() 幂等；stop() 返回冻结统计（GUI 二次 finalize 读取用）。
"""
from __future__ import annotations

import logging
import subprocess
import threading
import time
from datetime import datetime

import config

logger = logging.getLogger("RaceVideoToLog.monitor")

# psutil 可选：缺失时 RSS/CPU 降级为 None（不阻塞其他功能）。
try:
    import psutil
except ImportError:  # pragma: no cover
    psutil = None

_CREATE_NO_WINDOW = 0x08000000  # nvidia-smi 子进程不闪控制台窗口


def _to_float(s: str) -> float | None:
    try:
        return float(s)
    except (TypeError, ValueError):
        return None


class ResourceMonitor:
    """内存 / CPU / GPU 后台采样。

    采样线程 daemon，间隔默认 1s。维护每指标的 last/peak/avg。
    start()/stop() 幂等；stop() 返回冻结统计。
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._thread: threading.Thread | None = None
        self._stop_ev = threading.Event()
        self._interval = 1.0
        self._with_gpu = True
        self._proc = None  # psutil.Process（复用实例以计算 cpu_percent 差值）
        self._samples: list[dict] = []
        self._stats: dict | None = None
        self._gpu_name = ""
        self._gpu_backend = "none"  # none | smi | cuda
        self._gpu_missing = False
        self._psutil_warned = False

    @property
    def active(self) -> bool:
        return self._thread is not None and self._thread.is_alive()

    def start(self, interval_s: float = 1.0, with_gpu: bool = True) -> None:
        if self.active:
            return
        with self._lock:
            self._interval = max(0.2, float(interval_s))
            self._with_gpu = bool(with_gpu)
            self._samples = []
            self._stats = None
            self._stop_ev.clear()
            if psutil is not None:
                try:
                    self._proc = psutil.Process()
                    self._proc.cpu_percent(None)  # 预热差值基线，首个样本即有真实 CPU%
                except Exception:
                    self._proc = None
        self._gpu_name = self._probe_gpu_name()
        self._thread = threading.Thread(target=self._run, name="rvtol-monitor", daemon=True)
        self._thread.start()

    def _run(self) -> None:
        try:
            self._sample_once()
        except Exception:
            pass
        while not self._stop_ev.wait(self._interval):
            try:
                self._sample_once()
            except Exception:
                pass

    def stop(self) -> dict | None:
        th = self._thread
        if th is None:
            return self._stats
        self._stop_ev.set()
        th.join(timeout=max(2.0, self._interval * 2 + 1.0))
        self._thread = None
        self._finalize_stats()
        return self._stats

    # ── 采样 ──────────────────────────────────────────────────────

    def _sample_once(self) -> None:
        s: dict = {"t": time.perf_counter()}
        if psutil is not None and self._proc is not None:
            try:
                s["rss_mb"] = self._proc.memory_info().rss / 1048576.0
                cpu = self._proc.cpu_percent(None)
                s["cpu_pct"] = 0.0 if cpu is None else float(cpu)
            except Exception:
                s["rss_mb"] = None
                s["cpu_pct"] = None
        else:
            if not self._psutil_warned:
                self._psutil_warned = True
                logger.warning("psutil 未安装，RSS/CPU 监测不可用（GPU 采样不受影响）")
            s["rss_mb"] = None
            s["cpu_pct"] = None
        if self._with_gpu:
            s.update(self._sample_gpu())
        else:
            s.update({"util_pct": None, "vram_mb": None, "gpu_temp_c": None})
        with self._lock:
            self._samples.append(s)

    def _probe_gpu_name(self) -> str:
        if not self._with_gpu:
            return ""
        try:
            out = subprocess.run(
                ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"],
                capture_output=True, text=True, timeout=5, creationflags=_CREATE_NO_WINDOW)
            name = out.stdout.strip().splitlines()[0].strip() if out.stdout.strip() else ""
            if name:
                self._gpu_backend = "smi"
            return name
        except Exception:
            return ""

    def _sample_gpu(self) -> dict:
        none_all = {"util_pct": None, "vram_mb": None, "gpu_temp_c": None}
        if self._gpu_missing:
            return none_all
        if self._gpu_backend == "none":  # 无 nvidia-smi → 尝试 cudaMemGetInfo
            self._gpu_backend = "cuda"
        if self._gpu_backend == "smi":
            try:
                out = subprocess.run(
                    ["nvidia-smi",
                     "--query-gpu=utilization.gpu,memory.used,memory.total,temperature.gpu",
                     "--format=csv,noheader,nounits"],
                    capture_output=True, text=True, timeout=5, creationflags=_CREATE_NO_WINDOW)
                line = out.stdout.strip().splitlines()[0] if out.stdout.strip() else ""
                if not line:
                    raise ValueError("empty nvidia-smi output")
                util, used, _total, temp = (p.strip() for p in line.split(","))
                return {"util_pct": _to_float(util), "vram_mb": _to_float(used),
                        "gpu_temp_c": _to_float(temp)}
            except Exception:
                self._gpu_backend = "cuda"  # 永久降级到 cudaMemGetInfo
        try:
            from cuda.bindings import runtime as cuda_runtime
            free, total = cuda_runtime.cudaMemGetInfo()
            return {"util_pct": None, "vram_mb": (total - free) / 1048576.0, "gpu_temp_c": None}
        except Exception:
            self._gpu_missing = True
            return none_all

    # ── 统计读取 ──────────────────────────────────────────────────

    def _compute_stats(self, samples: list[dict]) -> dict:
        if not samples:
            return {"samples": 0, "elapsed_s": 0.0, "gpu_name": self._gpu_name}

        def agg(key: str) -> dict:
            vals = [s[key] for s in samples if s.get(key) is not None]
            if not vals:
                return {"last": None, "peak": None, "avg": None}
            return {"last": vals[-1], "peak": max(vals), "avg": sum(vals) / len(vals)}

        return {
            "samples": len(samples),
            "elapsed_s": samples[-1]["t"] - samples[0]["t"],
            "gpu_name": self._gpu_name,
            "rss_mb": agg("rss_mb"),
            "cpu_pct": agg("cpu_pct"),
            "gpu_util_pct": agg("util_pct"),
            "vram_used_mb": agg("vram_mb"),
            "gpu_temp_c": agg("gpu_temp_c"),
        }

    def _finalize_stats(self) -> None:
        with self._lock:
            samples = list(self._samples)
        self._stats = self._compute_stats(samples)


monitor = ResourceMonitor()


# ── 模块级委托函数（对外 API） ──────────────────────────────────────

def start(interval_s: float | None = None, with_gpu: bool | None = None) -> None:
    monitor.start(interval_s if interval_s is not None else 1.0,
                  with_gpu if with_gpu is not None else True)


def stop() -> dict | None:
    return monitor.stop()


def _fmt_stats(stats: dict) -> str:
    def p(d: dict, key: str) -> str:
        v = d.get(key)
        return "-" if v is None else f"{v:.1f}"
    parts = [
        f"RSS peak {p(stats.get('rss_mb') or {}, 'peak')}MB",
        f"CPU peak {p(stats.get('cpu_pct') or {}, 'peak')}%",
    ]
    gpu = stats.get("gpu_util_pct") or {}
    if gpu.get("peak") is not None:
        parts.append(f"GPU util peak {p(gpu, 'peak')}%")
        vram = stats.get("vram_used_mb") or {}
        parts.append(f"VRAM peak {p(vram, 'peak')}MB")
        temp = stats.get("gpu_temp_c") or {}
        if temp.get("peak") is not None:
            parts.append(f"GPU temp peak {p(temp, 'peak')}°C")
    if stats.get("gpu_name"):
        parts.append(stats["gpu_name"])
    if stats.get("samples"):
        parts.append(f"{stats['samples']} samples")
    return " | ".join(parts)


def format_stats(stats: dict) -> str:
    """人类可读的资源汇总（无 GPU 时自动省略 GPU 字段）。"""
    return _fmt_stats(stats)


def log_run(label: str, stats: dict | None, timing: dict | None = None) -> None:
    """把一次运行写入 <程序目录>/logs/monitor.log（追加，失败静默）。

    程序目录 = 免安装软件所在文件夹（config.app_logs_dir），不写
    %LOCALAPPDATA% —— 卸载/移动时删除程序目录即清理。
    """
    try:
        log_dir = config.app_logs_dir()
        log_dir.mkdir(parents=True, exist_ok=True)
        path = log_dir / "monitor.log"
        lines = [f"=== {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} === {label}"]
        if stats:
            lines.append("  " + _fmt_stats(stats))
        if timing:
            flat = {k: v for k, v in timing.items() if isinstance(v, (int, float))}
            if flat:
                lines.append("  " + " ".join(f"{k}={v:.1f}s" for k, v in sorted(flat.items())))
        with open(path, "a", encoding="utf-8") as f:
            f.write("\n".join(lines) + "\n")
    except Exception:
        pass


def gui_mark(mark: str) -> None:
    """GUI 时间标记（原 gui.py _t() 的实体）：追加到 <程序目录>/logs/gui_timing.log。"""
    try:
        log_dir = config.app_logs_dir()
        log_dir.mkdir(parents=True, exist_ok=True)
        path = log_dir / "gui_timing.log"
        with open(path, "a", encoding="utf-8") as f:
            f.write(f"{time.perf_counter():.3f} {mark}\n")
    except Exception:
        pass