gistfile1.txt

#!/usr/bin/env python3
"""powermetrics 처럼 GPU 사용률과 CPU 클러스터 잔류율(residency)을 IOReport 로 읽어온다.

Apple Silicon 에서 P-State residency 를 이용해 GPU/CPU 활성 사용률을 계산한다.
관리자 권한 없이도 동작한다.
"""
import ctypes
import time

from ctypes import (
    c_void_p,
    c_char_p,
    c_int,
    c_uint32,
    c_uint64,
    c_int64,
    POINTER,
    byref,
    create_string_buffer,
)

CF = ctypes.CDLL("/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation")
IOR = ctypes.CDLL("/usr/lib/libIOReport.dylib")

kCFStringEncodingUTF8 = 0x08000100

# --- CoreFoundation ---------------------------------------------------------
CF.CFStringCreateWithCString.argtypes = [c_void_p, c_char_p, c_uint32]
CF.CFStringCreateWithCString.restype = c_void_p
CF.CFRelease.argtypes = [c_void_p]
CF.CFRelease.restype = None

CF.CFStringGetCString.argtypes = [c_void_p, c_char_p, c_int64, c_uint32]
CF.CFStringGetCString.restype = ctypes.c_bool

CF.CFDictionaryGetValue.argtypes = [c_void_p, c_void_p]
CF.CFDictionaryGetValue.restype = c_void_p

CF.CFArrayGetCount.argtypes = [c_void_p]
CF.CFArrayGetCount.restype = c_int64
CF.CFArrayGetValueAtIndex.argtypes = [c_void_p, c_int64]
CF.CFArrayGetValueAtIndex.restype = c_void_p

# --- IOReport ---------------------------------------------------------------
IOR.IOReportCopyChannelsInGroup.argtypes = [c_void_p, c_void_p, c_uint64, c_uint64, c_uint64]
IOR.IOReportCopyChannelsInGroup.restype = c_void_p

IOR.IOReportMergeChannels.argtypes = [c_void_p, c_void_p, c_void_p]
IOR.IOReportMergeChannels.restype = None

IOR.IOReportCreateSubscription.argtypes = [c_void_p, c_void_p, POINTER(c_void_p), c_uint64, c_void_p]
IOR.IOReportCreateSubscription.restype = c_void_p

IOR.IOReportCreateSamples.argtypes = [c_void_p, c_void_p, c_void_p]
IOR.IOReportCreateSamples.restype = c_void_p

IOR.IOReportCreateSamplesDelta.argtypes = [c_void_p, c_void_p, c_void_p]
IOR.IOReportCreateSamplesDelta.restype = c_void_p

# 개별 채널(딕셔너리) 접근자
IOR.IOReportChannelGetGroup.argtypes = [c_void_p]
IOR.IOReportChannelGetGroup.restype = c_void_p
IOR.IOReportChannelGetSubGroup.argtypes = [c_void_p]
IOR.IOReportChannelGetSubGroup.restype = c_void_p
IOR.IOReportChannelGetChannelName.argtypes = [c_void_p]
IOR.IOReportChannelGetChannelName.restype = c_void_p

IOR.IOReportStateGetCount.argtypes = [c_void_p]
IOR.IOReportStateGetCount.restype = c_int
IOR.IOReportStateGetNameForIndex.argtypes = [c_void_p, c_int]
IOR.IOReportStateGetNameForIndex.restype = c_void_p
IOR.IOReportStateGetResidency.argtypes = [c_void_p, c_int]
IOR.IOReportStateGetResidency.restype = c_int64

IOR.IOReportSimpleGetIntegerValue.argtypes = [c_void_p, c_int]
IOR.IOReportSimpleGetIntegerValue.restype = c_int64


def cfstr(s):
    return CF.CFStringCreateWithCString(None, s.encode(), kCFStringEncodingUTF8)


def from_cfstr(ref):
    if not ref:
        return None
    buf = create_string_buffer(512)
    ok = CF.CFStringGetCString(ref, buf, len(buf), kCFStringEncodingUTF8)
    if not ok:
        return None
    return buf.value.decode("utf-8", "replace")


def copy_group(group, subgroup=None):
    g = cfstr(group)
    sg = cfstr(subgroup) if subgroup else None

    ch = IOR.IOReportCopyChannelsInGroup(g, sg, 0, 0, 0)

    CF.CFRelease(g)
    if sg:
        CF.CFRelease(sg)

    return ch


def iter_channels(delta):
    """delta 샘플 딕셔너리에서 각 채널 딕셔너리를 순회한다."""
    key = cfstr("IOReportChannels")
    arr = CF.CFDictionaryGetValue(delta, key)
    CF.CFRelease(key)
    if not arr:
        return
    n = CF.CFArrayGetCount(arr)
    for i in range(n):
        yield CF.CFArrayGetValueAtIndex(arr, i)


def read_states(chan):
    """채널의 P-State 잔류율을 {state_name: residency} 로 반환한다."""
    count = IOR.IOReportStateGetCount(chan)
    states = {}
    for i in range(count):
        name = from_cfstr(IOR.IOReportStateGetNameForIndex(chan, i))
        res = IOR.IOReportStateGetResidency(chan, i)
        states[name or f"P{i}"] = res
    return states


def active_ratio(states):
    """idle/off/down 상태를 제외한 활성 잔류 비율(0~1)."""
    total = sum(states.values())
    if total <= 0:
        return 0.0
    idle = 0
    for name, res in states.items():
        n = (name or "").upper()
        if "IDLE" in n or "OFF" in n or n == "DOWN":
            idle += res
    return max(0.0, (total - idle) / total)


def build_subscription():
    chans = copy_group("GPU Stats", "GPU Performance States")
    if not chans:
        raise RuntimeError("no GPU Stats / GPU Performance States channels")

    for group, sub in [
        ("CPU Stats", "CPU Core Performance States"),
        ("CPU Stats", "CPU Complex Performance States"),
    ]:
        extra = copy_group(group, sub)
        if extra:
            IOR.IOReportMergeChannels(chans, extra, None)
            CF.CFRelease(extra)

    sub = c_void_p()
    subscribed = IOR.IOReportCreateSubscription(None, chans, byref(sub), 0, None)
    if not sub.value:
        raise RuntimeError("subscription failed")
    return subscribed, chans, sub


class Sampler:
    """구독을 한 번만 만들어두고 반복적으로 delta 를 읽는다."""

    def __init__(self):
        self.subscribed, self.chans, self.sub = build_subscription()
        self.prev = IOR.IOReportCreateSamples(self.subscribed, self.chans, None)
        # 코어가 파킹되어 이번 샘플에 안 들어오거나 residency 합이 0 일 때
        # 마지막으로 관측한 값을 재사용하기 위한 저장소
        self.last = {}          # key -> entry (마지막 유효 값)
        self.order = {}         # group -> [key, ...] 관측 순서 유지

    def read(self, interval=1.0):
        time.sleep(interval)
        cur = IOR.IOReportCreateSamples(self.subscribed, self.chans, None)
        if not self.prev or not cur:
            raise RuntimeError("samples failed")

        delta = IOR.IOReportCreateSamplesDelta(self.prev, cur, None)
        if not delta:
            raise RuntimeError("delta failed")
        self.prev = cur

        # 이번 주기에 유효 데이터(residency 합 > 0)가 들어온 채널만 갱신
        for chan in iter_channels(delta):
            group = from_cfstr(IOR.IOReportChannelGetGroup(chan))
            if group not in ("GPU Stats", "CPU Stats"):
                continue
            subgroup = from_cfstr(IOR.IOReportChannelGetSubGroup(chan))
            name = from_cfstr(IOR.IOReportChannelGetChannelName(chan))
            states = read_states(chan)
            if not states:
                continue

            key = (group, subgroup, name)
            if key not in self.order.setdefault(group, []):
                self.order[group].append(key)

            total = sum(states.values())
            if total <= 0:
                # 파킹 등으로 이번엔 데이터 없음 → 마지막 값을 stale 로 유지
                if key in self.last:
                    self.last[key] = {**self.last[key], "stale": True}
                continue

            self.last[key] = {
                "name": name,
                "subgroup": subgroup,
                "active": active_ratio(states),
                "total": total,
                "stale": False,
            }

        # 관측 순서대로, 파킹되어 안 나온 채널은 마지막(stale) 값으로 채워 반환
        gpu, cpu = [], []
        for group, keys in self.order.items():
            dst = gpu if group == "GPU Stats" else cpu
            for key in keys:
                if key in self.last:
                    dst.append(self.last[key])
        return {"gpu": gpu, "cpu": cpu}


def sample(interval=1.0):
    """한 번만 샘플링해서 반환하는 편의 함수."""
    return Sampler().read(interval)


# --- 터미널 표시 (asitop 스타일) -------------------------------------------
from collections import deque

ESC = "\x1b["
HIDE_CURSOR = ESC + "?25l"
SHOW_CURSOR = ESC + "?25h"
CLEAR = ESC + "2J" + ESC + "H"
HOME = ESC + "H"
CLEAR_TO_END = ESC + "0J"   # 커서부터 화면 끝까지 지움 (잔상 제거용)

# 아래에서 위로 채워지는 1/8 단위 블록
EIGHTHS = [" ", "▁", "▂", "▃", "▄", "▅", "▆", "▇", "█"]


def color_for(pct):
    if pct >= 80:
        return "\x1b[91m"  # 빨강
    if pct >= 50:
        return "\x1b[93m"  # 노랑
    return "\x1b[92m"      # 초록


def bar(pct, width=40):
    pct = max(0.0, min(100.0, pct))
    filled = int(round(pct / 100 * width))
    reset = "\x1b[0m"
    c = color_for(pct)
    return f"{c}{'█' * filled}{reset}{'─' * (width - filled)}"


def gauge_line(label, pct, width=40, stale=False):
    if stale:
        # 파킹되어 이번 주기에 값이 없어 직전 값을 유지 → 흐리게 + 표시
        dim, reset = "\x1b[2m", "\x1b[0m"
        filled = int(round(max(0.0, min(100.0, pct)) / 100 * width))
        b = f"{'█' * filled}{'─' * (width - filled)}"
        return f"  {dim}{label:<12} {b} {pct:6.2f}% (park){reset}"
    return f"  {label:<12} {bar(pct, width)} {pct:6.2f}%"


def vgraph(history, height=8, width=48):
    """history(%) 를 아래에서 위로 자라는 세로 막대 그래프로 그린다.

    가장 오래된 값이 왼쪽, 최신 값이 오른쪽. 각 열의 높이가 사용률.
    """
    vals = list(history)[-width:]
    vals = [0.0] * (width - len(vals)) + vals  # 왼쪽을 빈 값으로 채움

    # 각 열을 1/8 단위 눈금 개수로 환산
    ticks = [int(round(max(0.0, min(100.0, v)) / 100 * height * 8)) for v in vals]

    rows = []
    for r in range(height):                     # r=0 이 맨 위 행
        level = height - 1 - r                   # 이 행이 담당하는 바닥 기준 칸
        cells = []
        for i, t in enumerate(ticks):
            fill = t - level * 8
            fill = 0 if fill < 0 else (8 if fill > 8 else fill)
            ch = EIGHTHS[fill]
            c = color_for(vals[i]) if fill > 0 else ""
            cells.append(f"{c}{ch}\x1b[0m" if fill > 0 else ch)
        # y축 눈금 (맨위=100, 맨아래=0 근처)
        axis = int(round((level + 1) / height * 100))
        rows.append(f"  {axis:3d}%│{''.join(cells)}")
    rows.append("      └" + "─" * width)
    return rows


def render(data, width=40, gpu_hist=None):
    lines = []
    lines.append("\x1b[1m mac_power — asitop 스타일 모니터\x1b[0m   (sudo 불필요 · Ctrl-C 로 종료)")
    lines.append("")

    gpu_pct = data["gpu"][0]["active"] * 100 if data["gpu"] else 0.0
    lines.append(f"\x1b[1;96m GPU 사용률  {gpu_pct:6.2f}%\x1b[0m")
    if gpu_hist is not None:
        lines.extend(vgraph(gpu_hist, height=8, width=min(48, max(20, width + 8))))
    lines.append("")

    cores = [e for e in data["cpu"] if e["subgroup"] == "CPU Core Performance States"]
    ecores = [e for e in cores if (e["name"] or "").startswith("E")]
    pcores = [e for e in cores if (e["name"] or "").startswith("P")]

    if ecores:
        avg = sum(e["active"] for e in ecores) / len(ecores) * 100
        lines.append(f"\x1b[1;96m CPU E-cluster (avg {avg:5.1f}%)\x1b[0m")
        for e in ecores:
            lines.append(gauge_line(e["name"], e["active"] * 100, width, e.get("stale")))
        lines.append("")

    if pcores:
        avg = sum(e["active"] for e in pcores) / len(pcores) * 100
        lines.append(f"\x1b[1;96m CPU P-cluster (avg {avg:5.1f}%)\x1b[0m")
        for e in pcores:
            lines.append(gauge_line(e["name"], e["active"] * 100, width, e.get("stale")))

    # 각 줄 끝을 지워(ESC[K) 이전 프레임의 더 긴 줄 잔상 제거
    return ("\x1b[K\n").join(lines) + "\x1b[K"


def live(interval=1.0, width=40):
    sampler = Sampler()
    gpu_hist = deque(maxlen=200)
    # 화면 전체 지우기는 최초 1회만. 이후엔 커서만 홈으로 보내 덮어써서 깜빡임 제거
    print(HIDE_CURSOR + CLEAR, end="", flush=True)
    try:
        while True:
            data = sampler.read(interval)
            gpu_hist.append(data["gpu"][0]["active"] * 100 if data["gpu"] else 0.0)
            frame = render(data, width, gpu_hist)
            # 한 번의 write 로 프레임 전체를 덮어쓰고, 남는 아래줄만 지운다
            print(HOME + frame + CLEAR_TO_END, end="", flush=True)
    except KeyboardInterrupt:
        pass
    finally:
        print(SHOW_CURSOR + "\n", end="", flush=True)


def main():
    import argparse

    p = argparse.ArgumentParser(description="Apple Silicon GPU/CPU 사용률 모니터")
    p.add_argument("-i", "--interval", type=float, default=1.0, help="샘플링 주기(초)")
    p.add_argument("-w", "--width", type=int, default=40, help="게이지 막대 너비")
    p.add_argument("--once", action="store_true", help="한 번만 출력하고 종료")
    args = p.parse_args()

    if args.once:
        data = sample(args.interval)
        hist = deque([data["gpu"][0]["active"] * 100 if data["gpu"] else 0.0])
        print(render(data, args.width, hist))
    else:
        live(args.interval, args.width)


if __name__ == "__main__":
    main()