106 lines
3.6 KiB
Python
106 lines
3.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Mesh control — the for-dummies GUI (v0, read-only + safe verbs).
|
||
|
|
|
||
|
|
A pywebview window over the same data planes everything else uses: the declared
|
||
|
|
truth (mesh-hosts.json), the discovered overlay (lan-state.json), and the
|
||
|
|
agent's snapshot (agent-status.json). Every action in the right-click menu is a
|
||
|
|
`net` verb — the GUI invents no behavior.
|
||
|
|
|
||
|
|
Run via `net gui` (uses the tray venv, which carries pywebview).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import shutil
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
|
||
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
AGENT_PY = os.path.join(ROOT, "smart-lan-router", "smart-lan-router.py")
|
||
|
|
_spec = importlib.util.spec_from_file_location("slr", AGENT_PY)
|
||
|
|
slr = importlib.util.module_from_spec(_spec)
|
||
|
|
sys.modules["slr"] = slr
|
||
|
|
_spec.loader.exec_module(slr)
|
||
|
|
|
||
|
|
FRIENDLY_ROLE = {
|
||
|
|
"gpu": "the AI box", "cpu": "storage + forge", "cloud": "cloud hub, Iceland",
|
||
|
|
"laptop": "this laptop", "phone": "phone",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _load(path: str) -> dict:
|
||
|
|
try:
|
||
|
|
with open(path, encoding="utf-8") as fh:
|
||
|
|
return json.load(fh)
|
||
|
|
except (OSError, json.JSONDecodeError):
|
||
|
|
return {}
|
||
|
|
|
||
|
|
|
||
|
|
class Api:
|
||
|
|
"""Exposed to the page as window.pywebview.api.*"""
|
||
|
|
|
||
|
|
def fleet(self) -> dict:
|
||
|
|
d = _load(os.path.join(ROOT, "data", "mesh-hosts.json"))
|
||
|
|
ov = _load(os.path.join(ROOT, "data", "lan-state.json"))
|
||
|
|
status = _load(os.path.join(ROOT, "data", "agent-status.json"))
|
||
|
|
me = slr.identify_self(d)
|
||
|
|
my = me["name"] if me else None
|
||
|
|
hosts = []
|
||
|
|
for h in d.get("hosts", []):
|
||
|
|
if h["name"] == my:
|
||
|
|
continue
|
||
|
|
hosts.append({
|
||
|
|
"name": h["name"],
|
||
|
|
"aliases": h.get("aliases") or [],
|
||
|
|
"klass": h.get("class"),
|
||
|
|
"friendly": FRIENDLY_ROLE.get(h.get("class", ""), h.get("class", "")),
|
||
|
|
"ip": ov.get(h["name"]) or h.get("lan") or h.get("wg"),
|
||
|
|
"wg": h.get("wg"),
|
||
|
|
"phone": h.get("class") == "phone",
|
||
|
|
})
|
||
|
|
return {"self": my, "location": status.get("location"),
|
||
|
|
"route": status.get("lan_route_via"), "agent_ts": status.get("ts"),
|
||
|
|
"hosts": hosts}
|
||
|
|
|
||
|
|
def probe(self, ip: str) -> dict:
|
||
|
|
ping = shutil.which("ping") or "/sbin/ping"
|
||
|
|
flag = "-t" if slr.PLATFORM == "darwin" else "-W"
|
||
|
|
rc, out, _ = slr._run([ping, "-c", "1", flag, "2", ip], 5)
|
||
|
|
if rc != 0:
|
||
|
|
return {"ok": False}
|
||
|
|
m = re.search(r"time=([\d.]+)", out)
|
||
|
|
return {"ok": True, "ms": round(float(m.group(1)), 1) if m else 0}
|
||
|
|
|
||
|
|
def copy(self, text: str) -> bool:
|
||
|
|
p = subprocess.run(["pbcopy"], input=text.encode())
|
||
|
|
return p.returncode == 0
|
||
|
|
|
||
|
|
def ssh_terminal(self, host: str) -> bool:
|
||
|
|
script = f'tell application "Terminal" to do script "ssh {host}"'
|
||
|
|
subprocess.Popen(["osascript", "-e", 'tell application "Terminal" to activate',
|
||
|
|
"-e", script])
|
||
|
|
return True
|
||
|
|
|
||
|
|
def doctor(self, host: str) -> str:
|
||
|
|
p = subprocess.run([os.path.join(ROOT, "bin", "net"), "doctor", host],
|
||
|
|
capture_output=True, text=True, timeout=60)
|
||
|
|
return p.stdout or p.stderr or "(no output)"
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
import webview
|
||
|
|
api = Api()
|
||
|
|
html = os.path.join(ROOT, "gui", "index.html")
|
||
|
|
webview.create_window("Mesh control", html, js_api=api,
|
||
|
|
width=620, height=560, resizable=True)
|
||
|
|
webview.start()
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|