ledgrab/server/src/ledgrab/utils/metrics/android_provider.py

"""Android metrics provider — reads /proc directly (no psutil needed).

Chaquopy doesn't ship a working psutil on Android, but the kernel
exposes the same data through ``/proc``. This provider tracks the
previous sample of ``/proc/stat`` and ``/proc/self/stat`` so it can
compute CPU% deltas the same way ``psutil.cpu_percent(interval=None)``
does on desktop.

If any of the expected ``/proc`` files become unreadable (some Android
flavors lock down ``/proc/self/stat`` for non-root apps), the provider
silently falls back to zero values for the affected metric instead of
crashing the dashboard. :func:`is_supported` lets the factory decide
whether this provider is even worth instantiating on the host.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Optional

from .types import MemorySnapshot, ProcessSnapshot


def is_supported() -> bool:
    """Return True iff /proc/stat and /proc/meminfo are readable here."""
    try:
        with open("/proc/stat", "r"):
            pass
        with open("/proc/meminfo", "r"):
            pass
    except OSError:
        return False
    return True


@dataclass
class _CpuSample:
    total: int
    busy: int


def _read_proc_stat() -> Optional[_CpuSample]:
    """Aggregate CPU jiffies from the first ``cpu`` line of /proc/stat."""
    try:
        with open("/proc/stat", "r") as f:
            line = f.readline()
    except OSError:
        return None
    parts = line.split()
    if not parts or parts[0] != "cpu":
        return None
    try:
        # user nice system idle iowait irq softirq steal guest guest_nice
        nums = [int(x) for x in parts[1:]]
    except ValueError:
        return None
    if len(nums) < 4:
        return None
    idle = nums[3] + (nums[4] if len(nums) > 4 else 0)  # idle + iowait
    total = sum(nums)
    return _CpuSample(total=total, busy=total - idle)


def _read_proc_self_stat_jiffies() -> Optional[int]:
    """Return user+system jiffies for the current process, or None on failure."""
    try:
        with open("/proc/self/stat", "rb") as f:
            data = f.read()
    except OSError:
        return None
    # The comm field (parens) can contain spaces; parse from the last ')'
    end = data.rfind(b")")
    if end < 0:
        return None
    parts = data[end + 1 :].split()
    # After comm (and state), positions:
    # 0=state 1=ppid 2=pgrp 3=session 4=tty_nr 5=tpgid 6=flags
    # 7=minflt 8=cminflt 9=majflt 10=cmajflt 11=utime 12=stime ...
    if len(parts) < 13:
        return None
    try:
        return int(parts[11]) + int(parts[12])
    except ValueError:
        return None


def _read_meminfo() -> MemorySnapshot:
    """Parse /proc/meminfo into a MemorySnapshot. Zeroed on failure."""
    fields: dict[str, int] = {}
    try:
        with open("/proc/meminfo", "r") as f:
            for line in f:
                key, _, rest = line.partition(":")
                if not rest:
                    continue
                val = rest.strip().split()
                if not val:
                    continue
                try:
                    # Values are in kB
                    fields[key] = int(val[0]) * 1024
                except ValueError:
                    continue
    except OSError:
        return MemorySnapshot(0, 0, 0.0)

    total = fields.get("MemTotal", 0)
    available = fields.get("MemAvailable", fields.get("MemFree", 0))
    if total <= 0:
        return MemorySnapshot(0, 0, 0.0)
    used = max(0, total - available)
    return MemorySnapshot(
        used_bytes=used,
        total_bytes=total,
        percent=round(used * 100.0 / total, 1),
    )


def _read_self_rss_bytes() -> int:
    """Read VmRSS (resident set size) for the current process from /proc/self/status."""
    try:
        with open("/proc/self/status", "r") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    parts = line.split()
                    # "VmRSS:  12345 kB"
                    if len(parts) >= 2:
                        try:
                            return int(parts[1]) * 1024
                        except ValueError:
                            return 0
    except OSError:
        return 0
    return 0


class AndroidMetricsProvider:
    """Reads CPU/RAM from /proc — used on Android via Chaquopy."""

    available: bool = True

    def __init__(self) -> None:
        self._cpu_count = os.cpu_count() or 1
        # Prime the deltas so the first real sample is meaningful.
        self._last_host: Optional[_CpuSample] = _read_proc_stat()
        self._last_proc_jiffies: Optional[int] = _read_proc_self_stat_jiffies()
        self._last_host_total: Optional[int] = self._last_host.total if self._last_host else None

    def cpu_percent(self) -> float:
        sample = _read_proc_stat()
        if sample is None or self._last_host is None:
            self._last_host = sample
            return 0.0
        d_total = sample.total - self._last_host.total
        d_busy = sample.busy - self._last_host.busy
        self._last_host = sample
        if d_total <= 0:
            return 0.0
        return round(d_busy * 100.0 / d_total, 1)

    def cpu_count(self) -> int:
        return self._cpu_count

    def virtual_memory(self) -> MemorySnapshot:
        return _read_meminfo()

    def process_snapshot(self) -> ProcessSnapshot:
        proc_jiffies = _read_proc_self_stat_jiffies()
        host_sample = _read_proc_stat()

        cpu = 0.0
        if (
            proc_jiffies is not None
            and self._last_proc_jiffies is not None
            and host_sample is not None
            and self._last_host_total is not None
        ):
            d_proc = proc_jiffies - self._last_proc_jiffies
            d_host = host_sample.total - self._last_host_total
            if d_host > 0 and d_proc >= 0:
                # d_proc / d_host gives fraction of *one* core; multiply by
                # cpu_count for raw N*100% scale, then normalize to 0–100%.
                cpu = round(d_proc * 100.0 / d_host, 1)

        if proc_jiffies is not None:
            self._last_proc_jiffies = proc_jiffies
        if host_sample is not None:
            self._last_host_total = host_sample.total

        return ProcessSnapshot(cpu_percent=cpu, rss_bytes=_read_self_rss_bytes())