Initial commit

STELLAR-GAUGE-20260807/stellar_gauge.py

@@ -0,0 +1,216 @@
+#!/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()