Step 1 in progress: the settings page server, probes, teaching pages and UI (not deployed yet)

- settings/server.py: read-only routes (overview, keys, computers, gear, companion, logs, history,
  help facts, key guide / cheat sheet / where things are), the editing-code check with a lockout,
  LAN-only binding with a per-minute rebind
- settings/probes.py: fixed read-only probes per computer; whole lists fetched and matched locally
- settings/keyguide.py: the generated teaching pages
- settings/ui: the page, every screen read-only
- status_panel.py: snapshot.json every 5 s for the page; the Companion database lookup prefers the
  running build's folder and skips a database that will not open (a stale v5.1 folder was breaking it)
- config.json: every computer's record (what runs there, network ports, done-by-hand ticks), the
  editing code (kept in the repo, the user's choice), an empty gear list
- install.sh: the settings page as a second LaunchAgent; settings/ and schema.py deployed

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
2026-10-13 16:23:59 -04:00
co-authored by Claude Fable 5.1
parent b05ff408d5
commit 2343a27529
11 changed files with 2266 additions and 19 deletions
+82 -3
View File
@@ -1783,14 +1783,39 @@ def remember_peers(snap, host, sessions):
def companion_db_path(cfg):
"""Companion's settings database: "db" in config "companion", else the folder of the build the
pages are written for ("companion.build", e.g. 5.0.7 -> v5.0), else the newest folder whose
database opens (an upgrade that was tried and abandoned leaves a folder with a dead copy)."""
if cfg["companion"].get("db"):
return os.path.expanduser(cfg["companion"]["db"])
root = os.path.expanduser("~/Library/Application Support/companion")
vers = [d for d in os.listdir(root) if re.fullmatch(r"v\d+\.\d+", d)] if os.path.isdir(root) else []
if not vers:
return None
newest = max(vers, key=lambda d: tuple(int(x) for x in d[1:].split(".")))
return os.path.join(root, newest, "db.sqlite")
build = str(cfg["companion"].get("build") or "").split("+")[0]
wanted = "v" + ".".join(build.split(".")[:2]) if build else None
order = sorted(vers, key=lambda d: tuple(int(x) for x in d[1:].split(".")), reverse=True)
if wanted in order:
order.remove(wanted)
order.insert(0, wanted)
for d in order:
path = os.path.join(root, d, "db.sqlite")
if os.path.exists(path) and db_opens(path):
return path
return None
def db_opens(path):
import sqlite3
try:
db = sqlite3.connect("file:%s?mode=ro" % urllib.parse.quote(path), uri=True, timeout=2)
try:
db.execute("SELECT 1 FROM instances LIMIT 1").fetchall()
finally:
db.close()
return True
except sqlite3.Error:
return False
def companion_conn_hosts(path):
@@ -2628,6 +2653,54 @@ def start_pollers(cfg, snap, stop, data_dir=None):
return threads
def snapshot_summary(snap, engine, now):
"""Everything the watchdog knows right now, in plain JSON (the settings page reads it instead of
asking the gear again): what Companion reports, pings, MIDI sessions here and on the other Macs,
helpers and programs, web checks, Spotify, the switcher, the Dante map, and every key's faults."""
out = {
"time": now,
"conns": {label: {"enabled": c.get("enabled"), "category": c.get("category"), "message": c.get("message")}
for label, c in snap.conns.items()},
"conns_fresh": snap.conns_fresh(now),
"conn_hosts": dict(snap.conn_hosts),
"ping": dict(snap.ping),
"link": snap.link,
"midi": snap.midi if snap.midi_time else None,
"remote_midi": {h: v for h, v in snap.remote_midi.items()},
"remote_midi_age": {h: round(now - t, 1) for h, t in snap.remote_midi_time.items()},
"mdns": sorted(snap.mdns) if snap.mdns_time else None,
"launchd": dict(snap.launchd),
"procs": dict(snap.procs),
"http": dict(snap.http),
"disk": dict(snap.disk),
"prefs": {"%s/%s" % k: v for k, v in snap.prefs.items()},
"remote": {k.replace("\0", "|"): v for k, v in snap.remote.items()},
"spotify": snap.spotify,
"spotify_known_songs": len(snap.spotify_tracks),
"atem": ({"connected": snap.atem.get("connected"), "audio": snap.atem.get("audio"), "camera": snap.atem.get("camera"),
"age": round(now - snap.atem.get("time", now), 1)} if snap.atem else None),
"dante_map": snap.dante_map,
"dante_map_age": round(now - snap.dante_map_time, 1) if snap.dante_map_time else None,
"artnet": {src: round(now - t, 1) for src, t in snap.artnet_last.items()},
"midi_rx": {name: {"connected": l.connected, "age": None if l.last_time is None else round(now - l.last_time, 1)}
for name, l in snap.listeners.items()},
"service_on": engine.service.on(now),
"alarms_override": engine.service.override,
"auto_fix": {"on": engine.auto_on, "paused": engine.auto_paused},
"burst": engine.burst and {"start": engine.burst["start"], "conns": sorted(engine.burst["conns"]),
"cleared": engine.burst["cleared"]},
"live": list(engine.live_now(now)),
"faults": {},
}
for key, tr in engine.trackers.items():
if tr.confirmed and key in engine.meta:
tile_id, chk, sub = engine.meta[key]
out["faults"].setdefault(tile_id, []).append(
{"check": check_identity(chk), "type": chk["type"], "sub": sub, "level": tr.result.level,
"msg": tr.result.msg, "since": tr.since, "acked": tr.acked})
return out
def post_webhook(url, text):
if not url:
return
@@ -2712,7 +2785,7 @@ def run(cfg, data_dir, dry_run=False, stop=None, config_sha=None):
except (OSError, http.client.HTTPException) as e:
log.warning("could not read the command variable at start: %s", e)
stop.wait(2.5) # let the first polls land before judging anything
last_json, last_write, last_eval = None, 0.0, 0.0
last_json, last_write, last_eval, last_snapshot = None, 0.0, 0.0, 0.0
down_since, alerted, warned_missing = None, False, set()
log.info("status panel running (code %s, %s)", VERSION, "dry run" if dry_run else "live")
try:
@@ -2740,6 +2813,12 @@ def run(cfg, data_dir, dry_run=False, stop=None, config_sha=None):
for kind, text in engine.events:
log.info("%s %s", kind, text)
engine.events = []
if now - last_snapshot >= 5:
last_snapshot = now
try:
write_json(os.path.join(data_dir, "snapshot.json"), snapshot_summary(snap, engine, now))
except (OSError, TypeError, ValueError) as e:
log.warning("snapshot.json: %s", e)
js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
if dry_run:
if js != last_json: