zhuyuan-ice-gl-zy001/STELLAR-GAUGE-20260807/stellar_gauge.py

#!/usr/bin/env python3
"""stellar_gauge.py · 湖面映星 · 悬浮星球用量仪
pywebview 无边框透明窗口，Canvas 手绘星球。
数据 = 本地用量统计 + 用户手抄余量基准 + 本机内存/存储/CPU 实时投影。
"""
import json, os, sys, threading, time
from pathlib import Path
import webview

HERE = os.path.dirname(os.path.abspath(__file__))
HOME_DIR = os.path.expanduser("~/.stellar-gauge")
PLAN_FILE = os.path.join(HOME_DIR, "plan.json")
USAGE_FILE = os.path.join(HOME_DIR, "usage.json")
SYS_FILE = os.path.join(HOME_DIR, "sysmon.json")
DIAG_FILE = os.path.join(HOME_DIR, "diag.txt")
WIN_TITLE = "湖面映星 · 星"
sys.path.insert(0, HERE)
import collector  # noqa: E402


def load(path, default):
    try:
        with open(path) as f:
            return json.load(f)
    except Exception:
        return default


def _our_window():
    """找到本应用的 NSWindow（按标题）。AppKit 调用必须回主线程。"""
    from AppKit import NSApp
    for w in NSApp.windows():
        try:
            if WIN_TITLE in str(w.title() or ""):
                return w
        except Exception:
            continue
    return None


def _on_main(fn):
    """把 AppKit 操作派回主线程（子线程碰 NSWindow 会 trace trap）"""
    from PyObjCTools import AppHelper
    AppHelper.callAfter(fn)


def scan_system():
    """本机内存 / 存储 / CPU 实时快照 → sysmon.json"""
    try:
        import psutil
    except Exception:
        return {}
    vm = psutil.virtual_memory()
    disk = psutil.disk_usage("/")
    snap = {
        "mem": {"used_pct": round(vm.percent, 1),
                "used_gb": round(vm.used / 2**30, 1),
                "total_gb": round(vm.total / 2**30, 1)},
        "disk": {"used_pct": round(disk.percent, 1),
                 "free_gb": round(disk.free / 2**30, 1),
                 "total_gb": round(disk.total / 2**30, 1)},
        "cpu": {"pct": round(psutil.cpu_percent(interval=None), 1)},
        "ts": time.strftime("%H:%M:%S"),
    }
    try:
        tmp = SYS_FILE + ".tmp"
        with open(tmp, "w") as f:
            json.dump(snap, f, ensure_ascii=False)
        os.replace(tmp, SYS_FILE)
    except Exception as e:
        print("sysmon write:", e, flush=True)
    return snap


def diag(msg):
    """诊断通道：open 拉起的进程 stdout 不落文件，直接写盘"""
    try:
        with open(DIAG_FILE, "a") as f:
            f.write(time.strftime("%H:%M:%S") + " " + str(msg)[:400] + "\n")
    except Exception:
        pass


class Api:
    def log(self, msg):
        """JS 侧自检日志回传"""
        diag("[js] " + str(msg))
        return True

    def get_data(self):
        usage = load(USAGE_FILE, {})
        plan = load(PLAN_FILE, {})
        sysm = load(SYS_FILE, {})
        return {"usage": usage, "plan": plan, "sys": sysm,
                "now": time.strftime("%Y-%m-%d %H:%M:%S")}

    def set_plan(self, plan_json):
        try:
            plan = json.loads(plan_json)
            old = load(PLAN_FILE, {})
            old.update(plan)
            tmp = PLAN_FILE + ".tmp"
            with open(tmp, "w") as f:
                json.dump(old, f, ensure_ascii=False, indent=1)
            os.replace(tmp, PLAN_FILE)
            return {"ok": True}
        except Exception as e:
            return {"ok": False, "error": str(e)}

    def drag_by(self, dx, dy):
        """JS 端拖拽 → 移动整颗星球窗口（macOS 屏幕坐标 y 向上）"""
        try:
            def move():
                w = _our_window()
                if w:
                    f = w.frame()
                    w.setFrameOrigin_((f.origin.x + dx, f.origin.y - dy))
            _on_main(move)
            return True
        except Exception as e:
            print("drag:", e, flush=True)
            return False


def collector_loop():
    while True:
        try:
            collector.scan()
        except Exception as e:
            print("collector error:", e, flush=True)
        time.sleep(300)


def sysmon_loop():
    psutil_first = True
    while True:
        try:
            if psutil_first:
                import psutil
                psutil.cpu_percent(interval=None)  # 首次调用丢弃基准
                psutil_first = False
            scan_system()
        except Exception as e:
            print("sysmon error:", e, flush=True)
        time.sleep(5)


def main():
    os.makedirs(HOME_DIR, exist_ok=True)
    diag("main start pid=" + str(os.getpid()))
    try:
        collector.scan()
    except Exception as e:
        print("initial scan error:", e, flush=True)
    scan_system()
    threading.Thread(target=collector_loop, daemon=True).start()
    threading.Thread(target=sysmon_loop, daemon=True).start()

    # 从磁盘文件加载：pywebview 的 html= 字符串通道会静默吞掉内联脚本
    page_url = Path(os.path.join(HERE, "index.html")).as_uri()

    win = webview.create_window(
        WIN_TITLE, url=page_url, js_api=Api(),
        width=500, height=500, x=2310, y=1100,
        frameless=True, resizable=False,
        background_color="#04060f")

    def on_shown():
        def raise_level():
            from AppKit import NSApp, NSFloatingWindowLevel
            w = _our_window()
            if w:
                w.setLevel_(NSFloatingWindowLevel)
            # launchd 拉起的裸 Python 不激活就不渲染：主动激活应用
            try:
                NSApp.activateWithOptions_(3)  # NSApplicationActivateIgnoringOtherApps
            except Exception:
                pass
        try:
            _on_main(raise_level)
        except Exception as e:
            print("on_top:", e, flush=True)

    win.events.shown += on_shown

    def on_loaded():
        try:
            r = win.evaluate_js(
                "JSON.stringify({boot:window.__boot||'NO',"
                "draw:typeof draw,api:typeof window.pywebview,"
                "alive:document.getElementById('alive')?'y':'n',"
                "ua:navigator.userAgent.slice(0,60)})")
            diag("loaded probe=" + str(r))
        except Exception as e:
            diag("loaded probe error: " + str(e))

        def pixel_check():
            # 自验：采样球心像素，非透明才说明 Canvas 真的画出来了
            try:
                px = win.evaluate_js(
                    "(function(){try{var d=cv.getContext('2d')"
                    ".getImageData(250,250,1,1).data;"
                    "return d[0]+','+d[1]+','+d[2]+','+d[3];}"
                    "catch(e){return 'ERR '+e;}})()")
                diag("canvas center pixel rgba=" + str(px))
            except Exception as e:
                diag("pixel check error: " + str(e))

        threading.Timer(4.0, pixel_check).start()

    win.events.loaded += on_loaded
    webview.start(debug=False)


if __name__ == "__main__":
    main()