ResourceMonitor

def __init__(self, period: float = 1.0, gpu_id: int = 0, dri_node: str = "",
             metrics: Optional[Any] = None) -> None:
    self.period = max(1.0, float(period))
    self.gpu_id = gpu_id
    self.dri_node = dri_node
    self.metrics = metrics
    self.system: Optional[Dict[str, Any]] = None
    self.gpu: Optional[Dict[str, Any]] = None
    self.on_tick: Optional[Callable[[float], Awaitable[None]]] = None
    self.watched: Optional[Callable[[], bool]] = None
    self._usage = SystemUsage()
    self._probe_gpu = True
    self._gpu_seen = False
    self._stop: Optional[asyncio.Event] = None
    self._task: Optional[asyncio.Task] = None

def _gpu_sample(self) -> Optional[Dict[str, Any]]:
    """One GPU reading, or None where there is nothing to read.

    A card is listed for what it is even when it exposes no counters, so
    the pipeline can match it by vendor or PCI address, but a card nothing
    counts and that reports no memory is not a reading: published every
    tick it would leave a page showing a utilization that can never move
    and a memory total of nothing. None instead stops the probe and leaves
    those off the page. A card that is merely idle has a utilization, so it
    keeps reporting.
    """
    gpus = get_gpus(self.dri_node)
    idx = 0 if (self.dri_node and len(gpus) == 1) else self.gpu_id
    if not gpus or not 0 <= idx < len(gpus):
        return None
    gpu = gpus[idx]
    if gpu.load is None and gpu.memoryTotal <= 0:
        return None
    return {
        "gpu_percent": (gpu.load or 0.0) * 100,
        "memory_total": gpu.memoryTotal * 1024 * 1024,
        "memory_used": gpu.memoryUsed * 1024 * 1024,
    }

def _sample(self) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]:
    """One blocking sample of both, for a worker thread."""
    cpu, total, used = self._usage.sample()
    system = {"cpu_percent": cpu, "mem_total": total, "mem_used": used}
    gpu = None
    if self._probe_gpu:
        try:
            gpu = self._gpu_sample()
        except Exception as exc:
            logger.warning(f"GPU stats unavailable this tick: {exc}")
        if gpu is not None:
            self._gpu_seen = True
        elif not self._gpu_seen:
            self._probe_gpu = False
            logger.info(
                f"No GPU with ID {self.gpu_id} reports utilization or memory; "
                "GPU stats disabled.")
    return system, gpu

async def _loop(self) -> None:
    try:
        while self._stop is not None and not self._stop.is_set():
            if self.metrics is not None or self.watched is None or self.watched():
                self.system, self.gpu = await asyncio.to_thread(self._sample)
            else:
                self.system = self.gpu = None
            if self.metrics is not None and self.gpu is not None:
                self.metrics.set_gpu_utilization(self.gpu["gpu_percent"])
            if self.on_tick is not None:
                await self.on_tick(time.time())
            try:
                await asyncio.wait_for(self._stop.wait(), timeout=self.period)
            except asyncio.TimeoutError:
                pass
    except asyncio.CancelledError:
        pass
    except Exception as exc:
        logger.error(f"Resource monitor error: {exc}", exc_info=True)

def start(self) -> None:
    """Starts sampling on the running loop; a second start is a no-op."""
    if self._task is not None and not self._task.done():
        return
    self._stop = asyncio.Event()
    self._task = asyncio.create_task(self._loop())

async def stop(self) -> None:
    """Ends the loop at once and waits for it."""
    if self._stop is not None:
        self._stop.set()
    if self._task is not None:
        await self._task
        self._task = None