216 lines
6.7 KiB
Python
Executable file
216 lines
6.7 KiB
Python
Executable file
#!/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()
|