1243 lines
69 KiB
Python
1243 lines
69 KiB
Python
#!/usr/bin/env python3
|
|
"""The settings page's server: a small stdlib HTTP server on the Ingest Mac, beside the watchdog.
|
|
|
|
Everything is readable by anyone on the church network (the alert web hook and the editing code
|
|
are masked); nothing changes without the editing code. The server never runs a command before an
|
|
Apply, never writes a file the watchdog writes, and reads the watchdog's own files for what it knows.
|
|
|
|
python3 -m settings.server --data-dir DIR [--port 8020] [--bind 127.0.0.1 ...]
|
|
"""
|
|
import argparse
|
|
import gzip
|
|
import glob
|
|
import hashlib
|
|
import tempfile
|
|
import http.server
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import re
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
import io
|
|
import plistlib
|
|
|
|
from settings import PARENT, HERE # noqa: F401
|
|
import schema
|
|
import status_panel as sp
|
|
from settings import apply as applying
|
|
from settings import builders
|
|
from settings import scripts
|
|
from settings import keyguide, probes
|
|
|
|
log = logging.getLogger("status-settings")
|
|
UI = os.path.join(HERE, "ui")
|
|
PRIVATE = [ipaddress.ip_network(n) for n in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "127.0.0.0/8")]
|
|
LOGS = {"watchdog": "~/Library/Logs/status-panel.log", "errors": "~/Library/Logs/status-panel.err.log",
|
|
"settings": "~/Library/Logs/status-settings.log"}
|
|
LOG_FILTERS = {"holds": r"\b(HOLD|FORCE|TEST|FIX|ENABLE|RESTART|ACK)\b", "fixes": r"\bFIX\b|-> ",
|
|
"auto": r"\bAUTO\b|auto-fix", "errors": r"WARNING|ERROR|Traceback", "faults": r"\b(FAULT|CLEAR)\b"}
|
|
|
|
# What the Companion experiments found (the Help page shows this; filled in as they are run).
|
|
COMPANION_FACTS = [
|
|
("Which module a connection uses", "The HTTP API reports it (moduleId) - no need to read Companion's database for it."),
|
|
("Does OBS say when it is streaming?", "Yes: the OBS-Youtube connection's variable 'streaming' reads 'Off-Air' when it is not; "
|
|
"the on-air value is recorded the first time it streams."),
|
|
("Can the page create the custom variables?", "Not yet known; they are made by hand in Companion for now."),
|
|
("Which Companion runs here", "The one listening on port 8000 (5.0.7); an abandoned 5.1 upgrade left a dead settings folder, which the panel now skips."),
|
|
("Can an old page's keys be blanked over the API?", "Not yet tried; after a page-number change the old keys are deleted by hand."),
|
|
]
|
|
|
|
|
|
def allowed_client(ip):
|
|
try:
|
|
addr = ipaddress.ip_address(ip.split("%")[0])
|
|
except ValueError:
|
|
return False
|
|
return any(addr in n for n in PRIVATE)
|
|
|
|
|
|
def bind_addresses(cfg, ifaces=None):
|
|
"""127.0.0.1 plus the IPv4 address of every LAN-use port named for this Mac (every private
|
|
address when none is named)."""
|
|
ifaces = probes.interfaces() if ifaces is None else ifaces
|
|
me = next((h for h in cfg.get("hosts", {}).values() if h.get("self")), {})
|
|
lan = [n["iface"] for n in me.get("networks", []) if n.get("use") == "LAN"]
|
|
out = ["127.0.0.1"]
|
|
for name, ip, active in ifaces:
|
|
if ip in out:
|
|
continue
|
|
if (name in lan) if lan else allowed_client(ip):
|
|
out.append(ip)
|
|
return out
|
|
|
|
|
|
def tail(path, lines=400, pattern=None):
|
|
path = os.path.expanduser(path)
|
|
try:
|
|
with open(path, "rb") as f:
|
|
f.seek(0, 2)
|
|
size = f.tell()
|
|
f.seek(max(0, size - 512 * 1024))
|
|
text = f.read().decode("utf-8", "replace")
|
|
except OSError:
|
|
return []
|
|
rows = text.split("\n")[1:] if len(text) >= 512 * 1024 else text.split("\n")
|
|
if pattern:
|
|
rx = re.compile(pattern)
|
|
rows = [r for r in rows if rx.search(r)]
|
|
return [r for r in rows if r][-lines:]
|
|
|
|
|
|
class Site:
|
|
"""Everything the handlers share: the config, the data dir, Companion, caches."""
|
|
|
|
def __init__(self, data_dir, config_path=None, companion=None):
|
|
self.data_dir = data_dir
|
|
self.config_path = config_path or os.path.join(data_dir, "config.json")
|
|
self.cfg, self.cfg_sha, self.cfg_mtime = None, None, 0
|
|
self.load()
|
|
self.comp = companion or sp.make_companion(self.cfg)
|
|
self.lock = threading.Lock()
|
|
self.probing = None
|
|
self.attempts = {} # client ip -> [count, first attempt time]
|
|
self.page_cache = {} # name -> (time, value)
|
|
self.draft = applying.Draft(data_dir)
|
|
self.applier = applying.Applier(self)
|
|
|
|
def load(self):
|
|
st = os.stat(self.config_path)
|
|
if st.st_mtime != self.cfg_mtime:
|
|
with open(self.config_path, "rb") as f:
|
|
raw = f.read()
|
|
self.cfg = sp.load_config(self.config_path)
|
|
self.cfg_sha, self.cfg_mtime = hashlib.sha256(raw).hexdigest(), st.st_mtime
|
|
return self.cfg
|
|
|
|
def read_json(self, name, default=None):
|
|
try:
|
|
with open(os.path.join(self.data_dir, name)) as f:
|
|
return json.load(f)
|
|
except (OSError, ValueError):
|
|
return default
|
|
|
|
def companion_get(self, path, timeout=3):
|
|
comp = self.cfg["companion"]
|
|
url = "http://%s:%d%s" % (comp.get("host", "127.0.0.1"), comp.get("port", 8000), path)
|
|
with urllib.request.urlopen(url, timeout=timeout) as r:
|
|
return r.status, r.read()
|
|
|
|
def cached(self, name, seconds, fn):
|
|
now = time.time()
|
|
hit = self.page_cache.get(name)
|
|
if hit and now - hit[0] < seconds:
|
|
return hit[1]
|
|
value = fn()
|
|
self.page_cache[name] = (now, value)
|
|
return value
|
|
|
|
# --- what the screens show ---------------------------------------------
|
|
|
|
def overview(self):
|
|
cfg = self.load()
|
|
names = cfg["variables"]
|
|
deck, hb, companion_ok = None, None, True
|
|
try:
|
|
raw = self.comp.get_custom(names["state"])
|
|
deck = json.loads(raw) if raw else None
|
|
hb = self.comp.get_custom(names["heartbeat"])
|
|
except Exception as e: # Companion down: the page still works
|
|
companion_ok, deck = False, None
|
|
log.warning("Companion: %s", e)
|
|
now = time.time()
|
|
hb = float(hb) if hb and str(hb).strip().replace(".", "").isdigit() else None
|
|
snap = self.read_json("snapshot.json", {}) or {}
|
|
started = self.read_json("started.json", {}) or {}
|
|
live = self.read_json("live-now.json", {}) or {}
|
|
running = self.read_json("fixes-running.json", {}) or {}
|
|
tiles = {t["id"]: t for t in cfg["tiles"]}
|
|
problems, needs = [], []
|
|
for tid, faults in (snap.get("faults") or {}).items():
|
|
tile = tiles.get(tid)
|
|
if not tile:
|
|
continue
|
|
for f in faults:
|
|
if f["level"] not in ("err", "warn"):
|
|
continue
|
|
text = f["msg"] if f["msg"] and f["msg"] != tile["label"] else tile["label"]
|
|
hold = keyguide.plain(keyguide.hold_words(tile, cfg))
|
|
problems.append({"key": tid, "label": tile["label"], "text": "%s: %s" % (tile["label"], f["msg"]),
|
|
"level": f["level"], "acked": f.get("acked"), "since": f.get("since"),
|
|
"action": "hold the %s key (%s)" % (tile["label"], hold) if hold and not hold.startswith("nothing")
|
|
else "no automatic fix: check the device"})
|
|
fresh = snap.get("time") and now - snap["time"] < 30
|
|
if not problems and deck and not fresh:
|
|
# The watchdog has not written a snapshot (older code, or just started): read the deck itself.
|
|
for tid, s in deck.items():
|
|
tile = tiles.get(tid)
|
|
if not tile or tile.get("members") or tile.get("kind") == "master":
|
|
continue # groups and ALL repeat their members
|
|
if s.get("b") not in (0xFF0000, 0x8B0000, 0xFFA000, 0x3A0000):
|
|
continue
|
|
lines = (s.get("x") or "").split("\n")
|
|
text = " ".join(lines[1:]) if len(lines) > 1 else lines[0]
|
|
hold = keyguide.plain(keyguide.hold_words(tile, cfg))
|
|
problems.append({"key": tid, "label": tile["label"], "text": "%s: %s" % (tile["label"], text),
|
|
"level": "warn" if s.get("b") == 0xFFA000 else "err", "acked": s.get("b") == 0x8B0000, "since": None,
|
|
"action": "hold the %s key (%s)" % (tile["label"], hold) if hold and not hold.startswith("nothing")
|
|
else "no automatic fix: check the device"})
|
|
page_only = self.read_json("page-only.json", {}) or {} # the last Apply changed nothing the watchdog reads
|
|
if started and started.get("config_sha") and started["config_sha"] != self.cfg_sha \
|
|
and not (page_only.get("sha") == self.cfg_sha and page_only.get("watchdog_sha") == started["config_sha"]):
|
|
needs.append("The watchdog is running an older settings file than the one on disk (restart it to pick up the change).")
|
|
pages = self.pages_state()
|
|
if pages.get("status") and not pages["status"]["ok"]:
|
|
needs.append("Companion's page %d is not called %s." % (cfg["page"], pages["status"]["expected"]))
|
|
return {
|
|
"time": now, "site": cfg.get("site", {}), "deck": deck, "heartbeat_age": (now - hb) if hb else None,
|
|
"companion_ok": companion_ok, "watchdog": self.watchdog_state(started, hb, now),
|
|
"live": live, "running": running, "service_on": snap.get("service_on") if fresh else None,
|
|
"alarms_override": snap.get("alarms_override") if fresh else None,
|
|
"auto_fix": {"phrase": schema.phrases(cfg)["auto_fix"], "state": snap.get("auto_fix")},
|
|
"problems": sorted(problems, key=lambda p: (p["level"] != "err", p["since"] or 0)), "needs": needs,
|
|
"computers": self.computer_lines(snap), "pages": pages, "snapshot_age": (now - snap["time"]) if snap.get("time") else None,
|
|
"last_change": self.last_change(), "phrases": schema.phrases(cfg), "version": started.get("version"),
|
|
}
|
|
|
|
def watchdog_state(self, started, hb, now):
|
|
state = {"running": False, "since": started.get("started_at"), "version": started.get("version"), "pid": started.get("pid")}
|
|
try:
|
|
out = sp.subprocess.run(["/bin/launchctl", "print", "gui/%d/org.waterschurch.status-panel" % os.getuid()],
|
|
capture_output=True, text=True, timeout=5).stdout
|
|
m = re.search(r"state = (\w+)", out)
|
|
state["launchd"] = m.group(1) if m else "not loaded"
|
|
state["running"] = bool(m and m.group(1) == "running")
|
|
except (OSError, sp.subprocess.TimeoutExpired):
|
|
state["launchd"] = "unknown"
|
|
state["heartbeat_fresh"] = bool(hb and now - hb < cfg_stale(self.cfg))
|
|
return state
|
|
|
|
def check_hosts(self, cfg):
|
|
"""{(key id, check identity): {computer aliases}}: which computers a check is about - the Mac a
|
|
MIDI session or remote command runs on, the peer's computer when a computer is named like it,
|
|
the computer a check's fix runs on, this Mac for local things."""
|
|
hosts = cfg.get("hosts", {})
|
|
me = next((a for a, h in hosts.items() if h.get("self")), None)
|
|
by_session = {}
|
|
for alias, h in hosts.items():
|
|
for run in h.get("runs", []):
|
|
if run.get("kind") == "midi_session" and run.get("name"):
|
|
by_session[run["name"]] = alias
|
|
if h.get("name"):
|
|
by_session.setdefault(h["name"], alias)
|
|
out = {}
|
|
for t in cfg["tiles"]:
|
|
for c in t.get("checks", []):
|
|
who = set()
|
|
if c["type"] in ("remote", "midi_peer") and c.get("host"):
|
|
who.add(c["host"])
|
|
elif c["type"] == "midi_peer" and me:
|
|
who.add(me)
|
|
if c["type"] == "midi_peer" and c.get("peer") in by_session:
|
|
who.add(by_session[c["peer"]])
|
|
if c["type"] in ("launchd", "process", "disk", "pref", "spotify", "atem", "midi_rx") and me:
|
|
who.add(me)
|
|
fx = c.get("fix")
|
|
for f in (fx.values() if isinstance(fx, dict) else [fx]):
|
|
fix = cfg.get("fixes", {}).get(f) if f else None
|
|
if fix and fix.get("host"):
|
|
who.add(fix["host"])
|
|
elif fix and fix.get("kind") in ("local", "spotify") and me:
|
|
who.add(me)
|
|
if t.get("hold_always"):
|
|
fix = cfg["fixes"].get(t["hold_always"], {})
|
|
if fix.get("host"):
|
|
who.add(fix["host"])
|
|
out[(t["id"], sp.check_identity(c))] = who
|
|
return out
|
|
|
|
def computer_lines(self, snap):
|
|
"""One line per computer: a status word and the reason, from what the watchdog sees."""
|
|
cfg = self.cfg
|
|
live = self.read_json("machines-live.json", {}) or {}
|
|
owners = self.check_hosts(cfg)
|
|
tiles_by_host, labels = {}, {t["id"]: t["label"] for t in cfg["tiles"]}
|
|
for (tid, _), who in owners.items():
|
|
for alias in who:
|
|
tiles_by_host.setdefault(alias, set()).add(tid)
|
|
lines = []
|
|
for alias, h in cfg.get("hosts", {}).items():
|
|
probe = live.get(alias) or {}
|
|
word, reason = "OK", ""
|
|
faults = []
|
|
for tid, fs in (snap.get("faults") or {}).items():
|
|
for f in fs:
|
|
if f["level"] == "err" and alias in owners.get((tid, f.get("check")), set()):
|
|
faults.append("%s %s" % (labels.get(tid, tid), f["msg"]))
|
|
if probe and not probe.get("reachable") and not h.get("self"):
|
|
word, reason = "Problem", "cannot be reached (%s)" % probe.get("error", "no answer")
|
|
elif faults:
|
|
word, reason = "Problem", "; ".join(faults[:2])
|
|
elif probe and any(i.get("light") == "problem" and i.get("watched", True) for i in probe.get("items", [])):
|
|
bad = [i for i in probe.get("items", []) if i.get("light") == "problem" and i.get("watched", True)]
|
|
word, reason = "Problem", "; ".join("%s %s" % (i["name"], i.get("reason", "")) for i in bad[:2])
|
|
elif not probe and not h.get("self"):
|
|
word, reason = "Not looked yet", ""
|
|
lines.append({"alias": alias, "name": h.get("name", alias), "role": h.get("role", ""), "word": word, "reason": reason,
|
|
"keys": sorted(tiles_by_host.get(alias, [])), "self": bool(h.get("self")),
|
|
"looked": probe.get("time")})
|
|
return lines
|
|
|
|
def pages_state(self):
|
|
cfg = self.cfg
|
|
|
|
def look():
|
|
out = {}
|
|
for which, page, name in (("status", cfg["page"], cfg["page_name"]),) + (
|
|
(("restart", cfg["restart_page"]["page"], cfg["restart_page"]["page_name"]),) if cfg.get("restart_page") else ()):
|
|
try:
|
|
status, raw = self.companion_get("/api/variable/internal/page_number_%d_name/value" % page)
|
|
live = raw.decode().replace("\\n", "\n") if status == 200 else None
|
|
except (OSError, urllib.error.URLError):
|
|
live = None
|
|
out[which] = {"page": page, "expected": name.replace("\\n", " "), "live": live.replace("\n", " ") if live else None,
|
|
"ok": live is not None and live.replace("\n", " ") == name.replace("\\n", " ")}
|
|
return out
|
|
return self.cached("pages", 60, look)
|
|
|
|
def last_change(self):
|
|
entries = applying.history_entries(self.data_dir)
|
|
if not entries:
|
|
return None
|
|
e = entries[0]
|
|
return {"when": time.strftime("%Y-%m-%d %H:%M", time.localtime(e["when"])) if e.get("when") else None, "by": e.get("by"), "note": e.get("note"),
|
|
"outcome": e.get("outcome")}
|
|
|
|
def companion_view(self):
|
|
cfg = self.load()
|
|
watched = {}
|
|
for t in cfg["tiles"]:
|
|
for c in t.get("checks", []):
|
|
if c.get("conn"):
|
|
watched.setdefault(c["conn"], set()).add(t["id"])
|
|
for k in (cfg.get("restart_page") or {}).get("keys", []):
|
|
for w in k.get("watch", []):
|
|
if w.startswith("conn:"):
|
|
watched.setdefault(w[5:], set()).add(k["id"])
|
|
try:
|
|
conns = self.comp.connections()
|
|
ok = True
|
|
except Exception as e:
|
|
conns, ok = [], False
|
|
rows = []
|
|
for c in sorted(conns, key=lambda c: c["label"].lower()):
|
|
st = c.get("status") or {}
|
|
rows.append({"label": c["label"], "module": c.get("moduleId"), "enabled": c.get("enabled"), "category": st.get("category"),
|
|
"message": st.get("message"), "watched_by": sorted(watched.get(c["label"], []))})
|
|
labels = {c["label"] for c in conns}
|
|
missing = sorted(l for l in watched if ok and l not in labels)
|
|
names = cfg["variables"]
|
|
lights = {}
|
|
for key in ("state", "heartbeat", "command"):
|
|
try:
|
|
lights["variable " + names[key]] = self.comp.get_custom(names[key]) is not None
|
|
except Exception:
|
|
lights["variable " + names[key]] = None
|
|
try:
|
|
lights["blink clock " + names.get("flash", "flash_tick")] = self.comp.get_custom(names.get("flash", "flash_tick")) is not None
|
|
except Exception:
|
|
lights["blink clock " + names.get("flash", "flash_tick")] = None
|
|
modules = {c.get("moduleId") for c in conns}
|
|
lights["ATEM module"] = ("bmd-atem" in modules) if ok else None
|
|
lights["Tascam module"] = ("tascam-da-6400" in modules) if ok else None
|
|
pages = self.pages_state()
|
|
for which, p in pages.items():
|
|
lights["page %d is the %s page" % (p["page"], which.capitalize())] = p["ok"]
|
|
nav = self.read_json("nav_locations.json", {}) or {}
|
|
try:
|
|
here = json.load(open(os.path.join(PARENT, "nav_locations.json")))
|
|
except (OSError, ValueError):
|
|
here = {}
|
|
coloured = len({k.split("/")[0] for k in (nav or here)})
|
|
return {"ok": ok, "build": self.cached("build", 600, self.companion_build), "connections": rows, "missing": missing,
|
|
"lights": lights, "page_number_buttons": {"pages_coloured": coloured, "recorded_in": "data dir" if nav else "repo copy"},
|
|
"address": "%s:%d" % (cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000)),
|
|
"variables": names}
|
|
|
|
def companion_build(self):
|
|
try:
|
|
status, raw = self.companion_get("/int/export/page/%d?includeSecrets=false" % self.cfg["page"], timeout=8)
|
|
try:
|
|
raw = gzip.decompress(raw)
|
|
except OSError:
|
|
pass
|
|
return json.loads(raw).get("companionBuild")
|
|
except (OSError, ValueError, urllib.error.URLError):
|
|
return None
|
|
|
|
def machines_view(self):
|
|
cfg = self.load()
|
|
live = self.read_json("machines-live.json", {}) or {}
|
|
snap = self.read_json("snapshot.json", {}) or {}
|
|
out = []
|
|
for line in self.computer_lines(snap):
|
|
alias = line["alias"]
|
|
h = cfg["hosts"][alias]
|
|
probe = dict(live.get(alias) or {})
|
|
items = probe.get("items") or [{"name": r.get("name") or r.get("label") or r.get("kind"), "kind": r.get("kind"),
|
|
"light": "unknown", "reason": "not looked yet", "watched": r.get("watched", True)}
|
|
for r in h.get("runs", [])]
|
|
# what the watchdog already knows beats a stale probe
|
|
for item, run in zip(items, h.get("runs", [])):
|
|
k = run.get("kind")
|
|
if k == "midi_session":
|
|
sessions = snap.get("midi") if h.get("self") else (snap.get("remote_midi") or {}).get(alias)
|
|
if sessions is not None and run.get("name") in sessions:
|
|
peers = sessions[run["name"]].get("peers", [])
|
|
item.update({"light": "ok", "reason": "", "detail": "%d peer%s: %s" % (len(peers), "" if len(peers) == 1 else "s",
|
|
", ".join(p["name"] for p in peers))})
|
|
elif run.get("peer_of"):
|
|
for host_sessions in list((snap.get("remote_midi") or {}).values()) + [snap.get("midi") or {}]:
|
|
sess = (host_sessions or {}).get(run["peer_of"])
|
|
if sess and any(p["name"] == run["name"] for p in sess.get("peers", [])):
|
|
item.update({"light": "ok", "reason": "", "detail": "in %s" % run["peer_of"]})
|
|
break
|
|
else:
|
|
item.update({"light": "problem", "reason": "not in the %s session" % run.get("peer_of")})
|
|
elif sessions is not None:
|
|
item.update({"light": "problem", "reason": "no such MIDI session"})
|
|
elif k == "web" or (k == "app" and run.get("http")):
|
|
ok = (snap.get("http") or {}).get(run.get("url") or run.get("http"))
|
|
if ok is not None:
|
|
item.update({"light": "ok" if ok else "problem", "reason": "" if ok else "its web page does not answer"})
|
|
elif k == "artnet":
|
|
age = (snap.get("artnet") or {}).get(run.get("source"))
|
|
if age is not None:
|
|
item.update({"light": "ok" if age < 10 else "problem", "reason": "" if age < 10 else "silent for %d s" % age})
|
|
probe["items"] = items
|
|
probe.update({"alias": alias, "name": h.get("name", alias), "os": h.get("os", "mac"), "role": h.get("role", ""),
|
|
"notes": h.get("notes", ""), "ssh": h.get("ssh"), "mac": h.get("mac"), "pingable": h.get("pingable", True),
|
|
"self": bool(h.get("self")), "networks": h.get("networks", []), "manual_done": h.get("manual_done", {}),
|
|
"word": line["word"], "reason": line["reason"], "keys": line["keys"]})
|
|
if not h.get("self") and h.get("pingable", True):
|
|
ip = h.get("ssh", "").split("@")[-1]
|
|
fails = (snap.get("ping") or {}).get(ip)
|
|
probe["ping"] = None if fails is None else (fails == 0)
|
|
out.append(probe)
|
|
return {"computers": out, "probing": bool(self.probing and self.probing.is_alive()),
|
|
"setup_steps": self.setup_steps(cfg)}
|
|
|
|
def setup_steps(self, cfg):
|
|
"""The one-time checklist per computer kind, with the light each line gets from the probe."""
|
|
return {
|
|
"mac": ["Remote Login is on", "the panel's login key is accepted", "python3 is present (for MIDI reads)",
|
|
"the DVS helper and its sudo rule (Macs that run Dante Virtual Soundcard)", "Audio MIDI Setup sessions named as the keys expect",
|
|
"Studio B only: the Pro Tools helper runs and Accessibility is allowed"],
|
|
"windows": ["the SSH service runs and starts with Windows", "the firewall lets this Mac in", "the panel's login key is accepted",
|
|
"a \\StatusPanel\\ task per program", "rtpMIDI's service restarts by itself (Lighting PC)", "wake-on-LAN is on in Windows and in the BIOS",
|
|
"the network profile is Public where rtpMIDI runs", "Wi-Fi power saving is off (Bible TV)"],
|
|
"self": ["Command Line Tools / python 3.9-3.12", "the watchdog and the settings page are loaded", "Companion's custom variables and blink clock exist",
|
|
"the login key pair exists", "the Netgear login is in the keychain", "the Spotify permission was allowed once",
|
|
"the DVS helper and its sudo rule", "the Dante and switch-management ports are up", "Google Drive is signed in (saved switcher files)"],
|
|
}
|
|
|
|
def gear_view(self):
|
|
cfg = self.load()
|
|
snap = self.read_json("snapshot.json", {}) or {}
|
|
conns, pings, http = snap.get("conns") or {}, snap.get("ping") or {}, snap.get("http") or {}
|
|
hosts_ips = {h.get("ssh", "").split("@")[-1] for h in cfg.get("hosts", {}).values()}
|
|
gear = cfg.get("gear", {})
|
|
kinds = {"hyperdeck": "recorder", "tascam": "recorder", "shure": "wireless receiver", "psm": "in-ear transmitter",
|
|
"dante": "Dante", "atem": "switcher"}
|
|
rows = {}
|
|
for t in cfg["tiles"]:
|
|
for c in t.get("checks", []):
|
|
name, kind, address = None, None, None
|
|
if c.get("conn") and c["type"] != "dante":
|
|
name, address = c["conn"], (snap.get("conn_hosts") or {}).get(c["conn"])
|
|
mod = next((r.get("module") for r in [] if False), None)
|
|
kind = kinds.get(c["type"], "device")
|
|
elif c["type"] == "ping" and c.get("host") and c["host"] not in hosts_ips:
|
|
name, address, kind = c.get("name") or c["host"], c["host"], "device"
|
|
elif c["type"] == "http" and not any(ip in c["url"] for ip in hosts_ips) and "127.0.0.1" not in c["url"]:
|
|
name, address, kind = c.get("name") or c.get("msg") or c["url"], c["url"], "web address"
|
|
elif c["type"] == "artnet":
|
|
continue
|
|
if not name:
|
|
continue
|
|
row = rows.setdefault(name, {"name": name, "kind": kind, "address": address, "watched_by": set(), "light": "unknown",
|
|
"reason": "", "location": gear.get(name, {}).get("location", ""), "notes": gear.get(name, {}).get("notes", "")})
|
|
row["watched_by"].add(t["id"])
|
|
if address is None and row["address"] is None and c.get("host"):
|
|
row["address"] = c["host"]
|
|
if c.get("conn") and c["conn"] in conns:
|
|
st = conns[c["conn"]]
|
|
row["light"] = "ok" if st.get("category") == "good" else ("problem" if st.get("enabled") else "not_needed")
|
|
row["reason"] = "" if row["light"] == "ok" else (st.get("message") or st.get("category") or "")
|
|
if c["type"] == "ping" and c.get("host") in pings and row["light"] != "problem":
|
|
row["light"] = "ok" if pings[c["host"]] == 0 else "problem"
|
|
row["reason"] = "" if pings[c["host"]] == 0 else "does not answer ping"
|
|
if c["type"] == "http" and c["url"] in http and row["light"] == "unknown":
|
|
row["light"] = "ok" if http[c["url"]] else "problem"
|
|
for name in (snap.get("dante_map") or {}):
|
|
e = snap["dante_map"][name]
|
|
row = rows.setdefault(name, {"name": name, "kind": "Dante device", "address": e.get("ip"), "watched_by": {"dante"}, "light": "unknown",
|
|
"reason": "", "location": gear.get(name, {}).get("location", ""), "notes": gear.get(name, {}).get("notes", "")})
|
|
row["detail"] = "switch %s port %s%s" % (e.get("switch", "?"), e.get("port", "?"), ", PoE" if e.get("poe") else "")
|
|
for s in cfg.get("dante_switches", []):
|
|
rows.setdefault("Netgear " + s, {"name": "Netgear " + s, "kind": "Dante switch", "address": s, "watched_by": {"dante"}, "light": "unknown",
|
|
"reason": "", "location": gear.get("Netgear " + s, {}).get("location", ""), "notes": ""})
|
|
for name, g in gear.items(): # gear added by hand, watched by no key
|
|
if not isinstance(g, dict):
|
|
continue
|
|
row = rows.setdefault(name, {"name": name, "kind": g.get("kind") or "device", "address": g.get("address"), "watched_by": set(), "light": "not_needed",
|
|
"reason": "not watched", "location": g.get("location", ""), "notes": g.get("notes", "")})
|
|
if g.get("kind"):
|
|
row["kind"] = g["kind"]
|
|
if g.get("address") and not row.get("address"):
|
|
row["address"] = g["address"]
|
|
row["own"] = True
|
|
out = []
|
|
for r in rows.values():
|
|
r["watched_by"] = sorted(r["watched_by"])
|
|
out.append(r)
|
|
return sorted(out, key=lambda r: (r["light"] != "problem", r["kind"], r["name"].lower()))
|
|
|
|
def keys_view(self):
|
|
cfg = self.load()
|
|
tiles = []
|
|
for t in cfg["tiles"]:
|
|
words = []
|
|
for c in t.get("checks", []):
|
|
ws = [c["msg"]] if c.get("msg") else list(keyguide.FAULT_WORDS.get(c["type"], []))
|
|
if c.get("tag"):
|
|
ws = ["%s %s" % (c["tag"], w) for w in ws]
|
|
words += ws
|
|
tiles.append(dict(t, says=sorted(set(words)), holding=keyguide.plain(keyguide.hold_words(t, cfg)),
|
|
checks=[dict(c, sentence=keyguide.plain(keyguide.check_sentence(c, cfg)),
|
|
identity=sp.check_identity(c)) for c in t.get("checks", [])]))
|
|
rp = cfg.get("restart_page")
|
|
keys = []
|
|
if rp:
|
|
for k in rp["keys"]:
|
|
keys.append(dict(k, holding=keyguide.plain(keyguide.fix_words(k["fix"], cfg))))
|
|
return {"page": cfg["page"], "page_name": cfg["page_name"], "tiles": tiles, "restart_page": dict(rp, keys=keys) if rp else None,
|
|
"hold": schema.phrases(cfg)["hold"], "grid": cfg.get("page_grid", schema.F["page_grid"]["default"])}
|
|
|
|
def history_view(self):
|
|
rows = []
|
|
for path in sorted(glob.glob(os.path.join(self.data_dir, "config.json.bak-*")), reverse=True):
|
|
m = re.search(r"bak-(\d{8})-(\d{6})$", path)
|
|
rows.append({"file": os.path.basename(path), "when": "%s-%s-%s %s:%s:%s" % (m.group(1)[:4], m.group(1)[4:6], m.group(1)[6:],
|
|
m.group(2)[:2], m.group(2)[2:4], m.group(2)[4:]) if m else "?",
|
|
"size": os.path.getsize(path)})
|
|
return {"entries": rows, "note": "Full history (who, what, Go back) arrives with step 2; these are the installer's copies."}
|
|
|
|
# --- editing code (the gate; nothing to gate yet in step 1) --------------
|
|
|
|
def check_code(self, ip, code):
|
|
count, first = self.attempts.get(ip, (0, 0))
|
|
now = time.time()
|
|
if count >= 5 and now - first < 600:
|
|
return {"ok": False, "locked_for": int(600 - (now - first))}
|
|
wanted = (self.cfg.get("site") or {}).get("editing_code") or ""
|
|
if wanted and code and code.strip().lower() == wanted.strip().lower():
|
|
self.attempts.pop(ip, None)
|
|
return {"ok": True}
|
|
if now - first >= 600:
|
|
count, first = 0, now
|
|
self.attempts[ip] = (count + 1, first)
|
|
return {"ok": False, "tries_left": max(0, 5 - count - 1), "no_code": not wanted}
|
|
|
|
|
|
def spotify_info(url):
|
|
"""Name, song count and URI of a public playlist, from open.spotify.com only."""
|
|
m = re.search(r"(?:playlist[/:])([A-Za-z0-9]{22})", url or "")
|
|
if not m:
|
|
return {"ok": False, "error": "paste a Spotify playlist link (open.spotify.com/playlist/...)"}
|
|
pid = m.group(1)
|
|
try:
|
|
req = urllib.request.Request("https://open.spotify.com/embed/playlist/" + pid, headers={"User-Agent": "Mozilla/5.0"})
|
|
html = urllib.request.urlopen(req, timeout=10, context=sp._UNVERIFIED).read().decode("utf-8", "replace")
|
|
mm = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', html)
|
|
entity = json.loads(mm.group(1))["props"]["pageProps"]["state"]["data"]["entity"]
|
|
return {"ok": True, "uri": "spotify:playlist:" + pid, "name": entity.get("name") or entity.get("title"),
|
|
"songs": len(entity.get("trackList", [])), "public": True}
|
|
except Exception as e:
|
|
return {"ok": True, "uri": "spotify:playlist:" + pid, "name": None, "songs": None, "public": False,
|
|
"warning": "the page cannot read this playlist (private?): the key will show WRONG PLAYLIST less reliably (%s)" % e}
|
|
|
|
|
|
def cfg_stale(cfg):
|
|
return cfg.get("page_stale_seconds", 15)
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
site = None # set by serve()
|
|
server_version = "status-settings"
|
|
|
|
def log_message(self, fmt, *args):
|
|
log.info("%s %s", self.client_address[0], fmt % args)
|
|
|
|
def send(self, status, body, ctype="application/json; charset=utf-8"):
|
|
data = body if isinstance(body, bytes) else (json.dumps(body, default=str) if ctype.startswith("application/json") else body).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
if not allowed_client(self.client_address[0]):
|
|
return self.send(403, {"error": "only from the church network"})
|
|
url = urllib.parse.urlparse(self.path)
|
|
q = urllib.parse.parse_qs(url.query)
|
|
path = url.path
|
|
site = self.site
|
|
try:
|
|
if path in ("/", "/index.html"):
|
|
return self.static("index.html", "text/html; charset=utf-8")
|
|
if path in ("/app.js", "/style.css"):
|
|
return self.static(path[1:], "text/javascript; charset=utf-8" if path.endswith(".js") else "text/css; charset=utf-8")
|
|
if path == "/key-guide.html":
|
|
return self.send(200, keyguide.key_guide(site.load()), "text/html; charset=utf-8")
|
|
if path == "/cheat.html":
|
|
return self.send(200, keyguide.cheat_sheet(site.load(), self.my_address()), "text/html; charset=utf-8")
|
|
if path == "/where.html":
|
|
return self.send(200, keyguide.where_things_are(site.load(), site.data_dir, self.my_address()), "text/html; charset=utf-8")
|
|
if path == "/api/config":
|
|
cfg = site.load()
|
|
return self.send(200, {"config": schema.mask_secrets(cfg), "sha": site.cfg_sha, "mtime": site.cfg_mtime})
|
|
if path == "/api/schema":
|
|
return self.send(200, {"fields": schema.F, "checks": schema.CHECKS, "fix_kinds": list(schema.FIX_KINDS),
|
|
"fix_kind_fields": schema.FIX_KIND_FIELDS, "glossary": schema.GLOSSARY})
|
|
if path == "/api/overview":
|
|
return self.send(200, site.overview())
|
|
if path == "/api/companion":
|
|
return self.send(200, site.companion_view())
|
|
if path == "/api/machines":
|
|
return self.send(200, site.machines_view())
|
|
if path == "/api/gear":
|
|
return self.send(200, site.gear_view())
|
|
if path == "/api/keys":
|
|
return self.send(200, site.keys_view())
|
|
if path == "/api/history":
|
|
return self.send(200, {"entries": applying.history_entries(site.data_dir)})
|
|
if path.startswith("/api/history/"):
|
|
entry = path[len("/api/history/"):]
|
|
cfg = applying.history_config(site.data_dir, entry)
|
|
if cfg is None:
|
|
return self.send(404, {"error": "no such version"})
|
|
live_cfg = site.load()
|
|
changes = [applying.words_for(p, a, b, cfg, live_cfg) for p, a, b in applying.changed_paths(cfg, live_cfg)]
|
|
return self.send(200, {"config": schema.mask_secrets(cfg), "changes_to_now": changes})
|
|
if path == "/api/draft":
|
|
d = site.draft.read()
|
|
return self.send(200, {"draft": self.masked_draft(d) if d else None, "live_sha": site.cfg_sha,
|
|
"stale": bool(d and d.get("base_sha") != site.cfg_sha),
|
|
"atem_pending": os.path.exists(os.path.join(site.data_dir, "draft-atem.xml"))})
|
|
if path == "/api/apply/status":
|
|
st = dict(site.applier.status)
|
|
st["live"] = dict(zip(("live", "reason"), site.applier.live()))
|
|
st["failed_last"] = site.read_json("last-apply-failed.json", None)
|
|
return self.send(200, st)
|
|
if path == "/api/live/spotify":
|
|
url = q.get("url", [""])[0]
|
|
return self.send(200, spotify_info(url))
|
|
if path == "/api/atem-reference/folder":
|
|
folder = os.path.expanduser((site.cfg.get("atem") or {}).get("source_folder") or "")
|
|
mounted = bool(folder and os.path.isdir(folder))
|
|
files, via, blocked = cloud_files(site.cfg, folder) if mounted else ([], "", False)
|
|
return self.send(200, {"folder": folder, "mounted": mounted, "blocked": blocked, "via": via,
|
|
"files": [{"name": n, "when": t} for n, t in files[:10]]})
|
|
if path == "/api/live/midi":
|
|
snap = site.read_json("snapshot.json", {}) or {}
|
|
return self.send(200, {"local": snap.get("midi"), "remote": snap.get("remote_midi")})
|
|
if path == "/api/site/names":
|
|
return self.send(200, self.names_check((site.draft.read() or {}).get("config") or site.load()))
|
|
if path == "/api/setup":
|
|
return self.send(200, {"computers": scripts.setup_all(site.load())})
|
|
if path == "/api/program/staged":
|
|
return self.send(200, self.staged())
|
|
if path == "/api/site/bundle":
|
|
cfg = site.load()
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
|
|
z.writestr("config.json", applying.config_text(cfg))
|
|
for name in ("atem-reference.xml", "nav_locations.json"):
|
|
fp = os.path.join(site.data_dir, name)
|
|
if os.path.exists(fp):
|
|
z.write(fp, name)
|
|
z.writestr("README.txt", "Site bundle of %s, exported %s from the settings page.\nDrop it in the repo as sites/%s/ or feed it to "
|
|
"'New campus from a bundle' on another campus's page.\n" % ((cfg.get("site") or {}).get("name", "?"), time.strftime("%Y-%m-%d %H:%M"), (cfg.get("site") or {}).get("id", "site")))
|
|
data = buf.getvalue()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/zip")
|
|
self.send_header("Content-Disposition", "attachment; filename=\"site-%s-%s.zip\"" % ((cfg.get("site") or {}).get("id", "site"), time.strftime("%Y%m%d")))
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
return self.wfile.write(data)
|
|
if path == "/api/builders":
|
|
return self.send(200, {"fixes": builders.template_list(site.cfg), "checks": builders.check_template_list()})
|
|
if path == "/api/facts":
|
|
return self.send(200, {"facts": COMPANION_FACTS})
|
|
if path == "/api/snapshot":
|
|
return self.send(200, site.read_json("snapshot.json", {}) or {})
|
|
if path.startswith("/api/logs/"):
|
|
name = path[len("/api/logs/"):]
|
|
if name == "autofix":
|
|
folder = os.path.expanduser((site.cfg.get("auto_fix") or {}).get("log_dir", "~/Desktop/Auto-Fix Logs"))
|
|
files = sorted(f for f in (os.listdir(folder) if os.path.isdir(folder) else []) if sp.AutoLog.NAME.match(f))
|
|
want = q.get("file", [files[-1] if files else ""])[0]
|
|
if want and want in files:
|
|
return self.send(200, {"files": files, "file": want, "lines": tail(os.path.join(folder, want), 1000)})
|
|
return self.send(200, {"files": files, "file": None, "lines": []})
|
|
if name not in LOGS:
|
|
return self.send(404, {"error": "no such log"})
|
|
flt = q.get("filter", [""])[0]
|
|
return self.send(200, {"lines": tail(LOGS[name], int(q.get("lines", ["400"])[0]), LOG_FILTERS.get(flt))})
|
|
return self.send(404, {"error": "not found"})
|
|
except Exception as e:
|
|
log.exception("GET %s", path)
|
|
return self.send(500, {"error": str(e)})
|
|
|
|
def gate(self):
|
|
"""Every change needs the editing code and a name. (ok, who) or (False, response sent)."""
|
|
code, who = self.headers.get("X-Edit-Key", ""), (self.headers.get("X-Editor") or "").strip()[:40]
|
|
r = self.site.check_code(self.client_address[0], code)
|
|
if not r.get("ok"):
|
|
self.send(401, dict(r, error="the editing code is needed" if not r.get("locked_for") else "locked out after too many wrong codes"))
|
|
return False, None
|
|
if not who:
|
|
self.send(400, {"error": "your first name is needed"})
|
|
return False, None
|
|
return True, who
|
|
|
|
def read_body(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
return self.rfile.read(length) if length else b""
|
|
|
|
def do_PUT(self):
|
|
self.do_POST()
|
|
|
|
def do_DELETE(self):
|
|
self.do_POST()
|
|
|
|
def do_POST(self):
|
|
if not allowed_client(self.client_address[0]):
|
|
return self.send(403, {"error": "only from the church network"})
|
|
path = urllib.parse.urlparse(self.path).path
|
|
raw = self.read_body()
|
|
ctype = self.headers.get("Content-Type") or ""
|
|
body = {}
|
|
if raw and ctype.startswith("application/json"):
|
|
try:
|
|
body = json.loads(raw or b"{}")
|
|
except ValueError:
|
|
return self.send(400, {"error": "bad JSON"})
|
|
host = (self.headers.get("Host") or "").split(":")[0]
|
|
origin = self.headers.get("Origin")
|
|
if origin and urllib.parse.urlparse(origin).hostname not in (host, "127.0.0.1", "localhost"):
|
|
return self.send(403, {"error": "wrong origin"})
|
|
site = self.site
|
|
try:
|
|
if path == "/api/editing/check":
|
|
return self.send(200, site.check_code(self.client_address[0], str(body.get("code", ""))))
|
|
if path == "/api/machines/probe":
|
|
if site.read_json("fixes-running.json", {}):
|
|
return self.send(409, {"error": "a fix is running; try again in a minute"})
|
|
with site.lock:
|
|
if site.probing and site.probing.is_alive():
|
|
return self.send(200, {"started": False, "already": True})
|
|
only = body.get("only")
|
|
site.probing = threading.Thread(target=probes.probe_all, args=(site.load(), site.data_dir, only), daemon=True)
|
|
site.probing.start()
|
|
return self.send(200, {"started": True})
|
|
# ---- everything below changes something: the editing code and a name are needed ----
|
|
ok, who = self.gate()
|
|
if not ok:
|
|
return
|
|
if path == "/api/draft" and self.command == "PUT":
|
|
cfg = schema.unmask(body.get("config") or {}, site.load())
|
|
draft, conflict = site.draft.update(cfg, who, site.cfg_sha, body.get("expect"))
|
|
if conflict:
|
|
changed = applying.changed_paths(cfg, conflict["config"])
|
|
return self.send(409, {"error": "%s changed this a moment ago" % conflict.get("by", "someone"), "draft": self.masked_draft(conflict),
|
|
"changed": [applying.words_for(p, a, b, cfg, conflict["config"])["where"] for p, a, b in changed[:5]]})
|
|
return self.send(200, {"draft": self.masked_draft(draft)})
|
|
if path == "/api/draft" and self.command == "DELETE":
|
|
site.draft.clear()
|
|
self.discard_atem()
|
|
return self.send(200, {"ok": True})
|
|
if path in ("/api/build", "/api/build-check"):
|
|
cfg = (site.draft.read() or {}).get("config") or site.load()
|
|
try:
|
|
if path == "/api/build":
|
|
fix = builders.build(cfg, body.get("fix") or {})
|
|
return self.send(200, {"ok": True, "fix": fix, "label": builders.suggest_label(cfg, fix)})
|
|
return self.send(200, {"ok": True, "check": builders.build_check(cfg, body.get("check") or {})})
|
|
except ValueError as e:
|
|
return self.send(200, {"ok": False, "error": str(e)})
|
|
if path == "/api/validate":
|
|
cfg = schema.unmask(body.get("config") or (site.draft.read() or {}).get("config") or site.load(), site.load())
|
|
return self.send(200, self.check(cfg))
|
|
if path == "/api/apply":
|
|
draft = site.draft.read()
|
|
if not draft:
|
|
return self.send(400, {"error": "there is no draft to apply"})
|
|
if draft.get("base_sha") != site.cfg_sha:
|
|
return self.send(409, {"error": "the live settings changed since this draft was started - review it again",
|
|
"changed_since": True})
|
|
atem = os.path.join(site.data_dir, "draft-atem.xml")
|
|
r = site.applier.start(draft["config"], str(body.get("note", ""))[:200], who, self.client_address[0], body.get("emergency"),
|
|
atem if os.path.exists(atem) else None)
|
|
if r.get("ok"):
|
|
site.draft.clear()
|
|
return self.send(200 if r.get("ok") else 409, r)
|
|
if path.startswith("/api/restore/"):
|
|
entry = path[len("/api/restore/"):]
|
|
cfg = applying.history_config(site.data_dir, entry)
|
|
if cfg is None:
|
|
return self.send(404, {"error": "no such version"})
|
|
draft, _ = site.draft.update(cfg, who, site.cfg_sha)
|
|
return self.send(200, {"draft": self.masked_draft(draft), "from": entry})
|
|
if path == "/api/undo":
|
|
entries = [e for e in applying.history_entries(site.data_dir) if e["kind"] == "apply"]
|
|
if not entries:
|
|
return self.send(404, {"error": "nothing to undo"})
|
|
cfg = applying.history_config(site.data_dir, entries[0]["id"])
|
|
draft, _ = site.draft.update(cfg, who, site.cfg_sha)
|
|
return self.send(200, {"draft": self.masked_draft(draft), "from": entries[0]["id"], "note": entries[0]["note"]})
|
|
if path in ("/api/watchdog/restart", "/api/watchdog/stop", "/api/watchdog/start"):
|
|
live, reason = site.applier.live()
|
|
if live and path != "/api/watchdog/start":
|
|
return self.send(409, {"error": "live now (%s)" % reason})
|
|
if path.endswith("/stop") and body.get("word") != "STOP":
|
|
return self.send(400, {"error": "type STOP to confirm"})
|
|
verb = path.rsplit("/", 1)[1]
|
|
return self.send(200, self.watchdog_control(verb, who))
|
|
if path == "/api/command":
|
|
text = str(body.get("text", "")).strip()
|
|
verb = text.split(" ")[0].lower() if text else ""
|
|
if verb not in ("test", "resurvey", "service"):
|
|
return self.send(400, {"error": "not a command the page may send"})
|
|
if verb == "test":
|
|
live, reason = site.applier.live()
|
|
if live:
|
|
return self.send(409, {"error": "live now (%s): no test runs" % reason})
|
|
text = "%s %d" % (text, int(time.time()))
|
|
names = site.cfg["variables"]
|
|
cur = site.comp.get_custom(names["command"]) or ""
|
|
ok, status = site.comp.set_custom(names["command"], (cur + ";" + text) if cur.strip() else text)
|
|
log.info("%s sent command %r -> %s", who, text, ok)
|
|
return self.send(200 if ok else 502, {"ok": ok, "sent": text})
|
|
if path == "/api/atem-reference":
|
|
return self.send(200, self.take_atem(raw))
|
|
if path == "/api/program/install":
|
|
live, reason = site.applier.live()
|
|
if live:
|
|
return self.send(409, {"error": "live now (%s): the program is not updated during a service" % reason})
|
|
return self.send(200, self.install_staged(who, keep_config=body.get("keep_config", True)))
|
|
if path == "/api/site/import":
|
|
try:
|
|
new_cfg = self.bundle_config(raw)
|
|
except (ValueError, KeyError, zipfile.BadZipFile) as e:
|
|
return self.send(400, {"error": "not a site bundle or settings file: %s" % e})
|
|
cfg = site.load()
|
|
mine = cfg.get("site") or {}
|
|
src = (new_cfg.get("site") or {}).get("name") or "the other campus"
|
|
new_cfg["site"] = dict(new_cfg.get("site") or {}, **{k: mine[k] for k in ("id", "name", "ingest_name", "ingest_hostname", "ingest_ips", "editing_code", "settings_port") if k in mine})
|
|
flag = "copied from %s - check it" % src
|
|
for alias, h in (new_cfg.get("hosts") or {}).items():
|
|
if h.get("self"):
|
|
for k in ("ssh", "mac", "wake_broadcast"):
|
|
h.pop(k, None)
|
|
h["notes"] = (flag + (". " + h["notes"] if h.get("notes") else ""))
|
|
for f in (new_cfg.get("fixes") or {}).values():
|
|
f["label"] = f.get("label", "")
|
|
errors, warnings = schema.validate(new_cfg)
|
|
draft, _ = site.draft.update(new_cfg, who, site.cfg_sha)
|
|
log.info("%s started a new campus from a bundle of %s", who, src)
|
|
return self.send(200, {"draft": self.masked_draft(draft), "from": src, "errors": errors, "warnings": warnings})
|
|
if path == "/api/atem-reference/pick-newest":
|
|
folder = os.path.expanduser((site.cfg.get("atem") or {}).get("source_folder") or "")
|
|
files, via, blocked = cloud_files(site.cfg, folder) if folder and os.path.isdir(folder) else ([], "", False)
|
|
if not files:
|
|
return self.send(404, {"error": "no .xml files in the folder (is Google Drive signed in?)" if not blocked else
|
|
"this page is not allowed to read the cloud folder; use Upload a saved file"})
|
|
newest = os.path.join(folder, files[0][0])
|
|
raw = cloud_read(site.cfg, newest, via)
|
|
if raw is None:
|
|
return self.send(502, {"error": "could not read %s" % files[0][0]})
|
|
return self.send(200, dict(self.take_atem(raw), file=files[0][0]))
|
|
if path == "/api/atem-reference/discard":
|
|
self.discard_atem()
|
|
return self.send(200, {"ok": True})
|
|
if path == "/api/gear":
|
|
cfg = site.load()
|
|
gear = dict(cfg.get("gear") or {})
|
|
name = str(body.get("name", "")).strip()[:80]
|
|
if not name:
|
|
return self.send(400, {"error": "a name is needed"})
|
|
if body.get("remove"):
|
|
gear.pop(name, None)
|
|
else:
|
|
entry = {k: str(body.get(k, "")).strip()[:300] for k in ("location", "notes", "kind", "address")}
|
|
gear[name] = {k: v for k, v in entry.items() if v}
|
|
draft = site.draft.read() or {"config": cfg}
|
|
draft_cfg = dict(draft["config"], gear=gear)
|
|
d, _ = site.draft.update(draft_cfg, who, site.cfg_sha)
|
|
return self.send(200, {"draft": self.masked_draft(d)})
|
|
return self.send(404, {"error": "not found"})
|
|
except Exception as e:
|
|
log.exception("%s %s", self.command, path)
|
|
return self.send(500, {"error": str(e)})
|
|
|
|
def names_check(self, cfg):
|
|
"""Every device name the keys use, against what this Companion, the Dante network and the Macs report now:
|
|
the mapping step of a copied campus."""
|
|
site = self.site
|
|
try:
|
|
live = {c["label"]: c.get("moduleId") for c in site.comp.connections()}
|
|
companion_ok = True
|
|
except Exception:
|
|
live, companion_ok = {}, False
|
|
used = {}
|
|
for t in cfg.get("tiles", []):
|
|
for c in t.get("checks", []):
|
|
if c.get("conn") and c["type"] != "dante" or (c["type"] == "dante" and c.get("conn")):
|
|
used.setdefault(c["conn"], {"keys": set(), "module": schema.EXPECTED_MODULE.get(c["type"])})["keys"].add(t.get("label", t["id"]))
|
|
for k in (cfg.get("restart_page") or {}).get("keys", []):
|
|
for w in k.get("watch", []):
|
|
if w.startswith("conn:"):
|
|
used.setdefault(w[5:], {"keys": set(), "module": None})["keys"].add("restart key " + k.get("label", k["id"]))
|
|
conns = []
|
|
for name, u in sorted(used.items()):
|
|
cands = [l for l, m in live.items() if l not in used and (not u["module"] or not m or m == u["module"])]
|
|
conns.append({"name": name, "keys": sorted(u["keys"]), "live": name in live, "module": u["module"],
|
|
"live_module": live.get(name), "candidates": sorted(cands)})
|
|
snap = site.read_json("snapshot.json", {}) or {}
|
|
dmap = snap.get("dante_map") or {}
|
|
dante = []
|
|
for t in cfg.get("tiles", []):
|
|
for c in t.get("checks", []):
|
|
if c["type"] == "dante":
|
|
for n in c.get("devices", {}):
|
|
dante.append({"name": n, "seen": n in dmap, "candidates": sorted(x for x in dmap if x not in c["devices"])})
|
|
local, remote = snap.get("midi") or {}, snap.get("remote_midi") or {}
|
|
midi = []
|
|
for t in cfg.get("tiles", []):
|
|
for c in t.get("checks", []):
|
|
if c["type"] == "midi_peer":
|
|
src = (remote.get(c["host"]) or {}) if c.get("host") else local
|
|
sess = src.get(c.get("session")) if isinstance(src, dict) else None
|
|
peers = [p.get("name") for p in (sess or {}).get("peers", [])] if sess else []
|
|
midi.append({"session": c.get("session"), "peer": c.get("peer"), "host": c.get("host"), "key": t.get("label", t["id"]),
|
|
"session_seen": sess is not None, "peer_seen": c.get("peer") in peers,
|
|
"sessions_seen": sorted(src) if isinstance(src, dict) else [], "peers_seen": peers})
|
|
return {"companion_ok": companion_ok, "connections": conns, "dante": dante, "midi": midi,
|
|
"missing": sum(1 for c in conns if not c["live"]) + sum(1 for d in dante if not d["seen"]) + sum(1 for m in midi if not m["session_seen"] or not m["peer_seen"])}
|
|
|
|
def staged(self):
|
|
"""The staged folder (~/status-panel-install) against the running code."""
|
|
src = os.path.expanduser("~/status-panel-install")
|
|
out = {"folder": src, "present": os.path.isfile(os.path.join(src, "install.sh")), "staged_version": None, "staged_when": None,
|
|
"running_version": (self.site.read_json("started.json", {}) or {}).get("version"), "job": None, "log": []}
|
|
sp_path = os.path.join(src, "status_panel.py")
|
|
if os.path.isfile(sp_path):
|
|
m = re.search(r'^VERSION\s*=\s*"([^"]+)"', open(sp_path, encoding="utf-8").read(), re.M)
|
|
out["staged_version"] = m.group(1) if m else "?"
|
|
out["staged_when"] = os.path.getmtime(sp_path)
|
|
logp = os.path.expanduser("~/Library/Logs/status-install.log")
|
|
if os.path.exists(logp):
|
|
try:
|
|
out["log"] = open(logp, encoding="utf-8", errors="replace").read().splitlines()[-25:]
|
|
out["log_when"] = os.path.getmtime(logp)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
r = subprocess.run(["/bin/launchctl", "print", "gui/%d/org.waterschurch.status-install" % os.getuid()], capture_output=True, text=True, timeout=5)
|
|
out["job"] = "running" if "state = running" in r.stdout else ("loaded" if r.returncode == 0 else None)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
return out
|
|
|
|
def install_staged(self, who, keep_config=True):
|
|
"""Run install.sh from the staged folder as a one-shot launchd job (it survives this agent's own restart)."""
|
|
src = os.path.expanduser("~/status-panel-install")
|
|
if not os.path.isfile(os.path.join(src, "install.sh")):
|
|
return {"ok": False, "error": "nothing is staged in %s (copy the folder there with scp first)" % src}
|
|
label = "org.waterschurch.status-install"
|
|
plist = os.path.expanduser("~/Library/LaunchAgents/%s.plist" % label)
|
|
args = ["/bin/zsh", os.path.join(src, "install.sh")] + (["--keep-config"] if keep_config else [])
|
|
with open(plist, "wb") as f:
|
|
plistlib.dump({"Label": label, "ProgramArguments": args, "RunAtLoad": True, "AbandonProcessGroup": True,
|
|
"StandardOutPath": os.path.expanduser("~/Library/Logs/status-install.log"),
|
|
"StandardErrorPath": os.path.expanduser("~/Library/Logs/status-install.log"),
|
|
"EnvironmentVariables": {"HOME": os.path.expanduser("~"), "PATH": "/usr/bin:/bin:/usr/sbin:/sbin"}}, f)
|
|
domain = "gui/%d" % os.getuid()
|
|
subprocess.run(["/bin/launchctl", "bootout", "%s/%s" % (domain, label)], capture_output=True, text=True, timeout=10)
|
|
try:
|
|
open(os.path.expanduser("~/Library/Logs/status-install.log"), "a").write("\n==== %s install started by %s from the settings page ====\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), who))
|
|
except OSError:
|
|
pass
|
|
r = subprocess.run(["/bin/launchctl", "bootstrap", domain, plist], capture_output=True, text=True, timeout=10)
|
|
log.info("%s started the install job (%s): rc=%s %s", who, " ".join(args), r.returncode, (r.stderr or r.stdout).strip())
|
|
if r.returncode != 0:
|
|
return {"ok": False, "error": "launchctl: %s" % (r.stderr or r.stdout).strip()[-200:]}
|
|
return {"ok": True, "detail": "install.sh is running; this page comes back in about 20 seconds"}
|
|
|
|
def bundle_config(self, raw):
|
|
"""config.json out of a site bundle (zip) or a bare settings file."""
|
|
if raw[:2] == b"PK":
|
|
with zipfile.ZipFile(io.BytesIO(raw)) as z:
|
|
names = [n for n in z.namelist() if n.endswith("config.json")]
|
|
if not names:
|
|
raise ValueError("no config.json in the bundle")
|
|
cfg = json.loads(z.read(names[0]).decode("utf-8"))
|
|
xml = [n for n in z.namelist() if n.endswith("atem-reference.xml")]
|
|
if xml:
|
|
with open(os.path.join(self.site.data_dir, "draft-atem.xml"), "wb") as f:
|
|
f.write(z.read(xml[0]))
|
|
else:
|
|
cfg = json.loads(raw.decode("utf-8"))
|
|
if not isinstance(cfg, dict) or "tiles" not in cfg:
|
|
raise ValueError("no keys in it")
|
|
return cfg
|
|
|
|
def masked_draft(self, draft):
|
|
return dict(draft, config=schema.mask_secrets(draft["config"]))
|
|
|
|
def check(self, cfg):
|
|
"""Everything Review shows: errors, warnings, the plain-words changes, what Apply will do."""
|
|
site = self.site
|
|
live_cfg = site.load()
|
|
try:
|
|
conns = {c["label"]: c.get("moduleId") for c in site.comp.connections()}
|
|
except Exception:
|
|
conns = None
|
|
errors, warnings = schema.validate(cfg, {"connections": conns} if conns else None)
|
|
tmp = tempfile.NamedTemporaryFile("wb", suffix=".json", delete=False)
|
|
tmp.write(applying.config_text(cfg)); tmp.close()
|
|
try:
|
|
sp.load_config(tmp.name)
|
|
except (ValueError, KeyError, TypeError) as e:
|
|
errors.append("the watchdog would not accept it: %s" % e)
|
|
finally:
|
|
os.remove(tmp.name)
|
|
plan = applying.what_apply_does(live_cfg, cfg, site.applier.hashes()) if not errors else {"restart": False, "pages": [], "nav": False, "changes": applying.changed_paths(live_cfg, cfg)}
|
|
changes = [applying.words_for(p, a, b, live_cfg, cfg) for p, a, b in plan["changes"]]
|
|
live, reason = site.applier.live()
|
|
atem_pending = os.path.exists(os.path.join(site.data_dir, "draft-atem.xml"))
|
|
return {"errors": errors, "warnings": warnings, "changes": changes, "plan": {k: plan[k] for k in ("restart", "pages", "nav")},
|
|
"live": {"live": live, "reason": reason}, "atem_pending": atem_pending, "feel": self.feel(plan, atem_pending)}
|
|
|
|
def feel(self, plan, atem_pending):
|
|
if plan.get("error"):
|
|
return "The pages could not be built: " + plan["error"]
|
|
parts = []
|
|
if plan["restart"] or atem_pending:
|
|
parts.append("the deck goes purple for about 5 seconds and comes back with your changes")
|
|
if plan["pages"]:
|
|
parts.append("the %s re-imported" % " and ".join("%s page is" % p.capitalize() for p in plan["pages"] if p in ("status", "restart")))
|
|
if plan["nav"]:
|
|
parts.append("the page-number keys are recoloured")
|
|
return (", ".join(parts) + ".").capitalize() if parts else "Nothing on the deck changes."
|
|
|
|
def take_atem(self, raw):
|
|
"""An uploaded saved switcher file: capped, no DOCTYPE or entities, parsed with the watchdog's own
|
|
parser; kept as draft-atem.xml until Apply. Returns what differs from the current file."""
|
|
site = self.site
|
|
if len(raw) > 20 * 1024 * 1024:
|
|
return {"ok": False, "error": "the file is bigger than 20 MB"}
|
|
head = raw[:4096].decode("utf-8", "replace")
|
|
if "<!DOCTYPE" in head or "<!ENTITY" in head:
|
|
return {"ok": False, "error": "the file holds a DOCTYPE or entities; a saved switcher file never does"}
|
|
pending = os.path.join(site.data_dir, "draft-atem.xml")
|
|
with open(pending + ".tmp", "wb") as f:
|
|
f.write(raw)
|
|
try:
|
|
new = sp.atem_reference(pending + ".tmp", (site.cfg.get("atem") or {}).get("compare"))
|
|
except Exception as e:
|
|
os.remove(pending + ".tmp")
|
|
return {"ok": False, "error": "not a saved switcher file: %s" % e}
|
|
os.replace(pending + ".tmp", pending)
|
|
os.chmod(pending, 0o600)
|
|
current = os.path.join(site.data_dir, "atem-reference.xml")
|
|
try:
|
|
old = sp.atem_reference(current, (site.cfg.get("atem") or {}).get("compare"))
|
|
except Exception:
|
|
old = {"audio": {}, "camera": {}}
|
|
product = re.search(r'product="([^"]+)"', head)
|
|
diffs = []
|
|
for part in ("audio", "camera"):
|
|
for k in sorted(set(old[part]) | set(new[part])):
|
|
if old[part].get(k) != new[part].get(k):
|
|
diffs.append({"part": part, "key": k, "before": old[part].get(k), "after": new[part].get(k)})
|
|
return {"ok": True, "product": product.group(1) if product else None, "audio": len(new["audio"]), "camera": len(new["camera"]),
|
|
"diffs": diffs[:50], "diff_count": len(diffs)}
|
|
|
|
def discard_atem(self):
|
|
try:
|
|
os.remove(os.path.join(self.site.data_dir, "draft-atem.xml"))
|
|
except OSError:
|
|
pass
|
|
|
|
def watchdog_control(self, verb, who):
|
|
domain = "gui/%d" % os.getuid()
|
|
label = applying.WATCHDOG
|
|
plist = os.path.expanduser("~/Library/LaunchAgents/%s.plist" % label)
|
|
try:
|
|
if verb == "stop":
|
|
r = sp.subprocess.run(["/bin/launchctl", "bootout", "%s/%s" % (domain, label)], capture_output=True, text=True, timeout=30)
|
|
log.info("%s stopped the watchdog -> %s", who, r.returncode)
|
|
return {"ok": r.returncode == 0, "detail": (r.stderr or "stopped").strip()[-200:]}
|
|
if verb == "start":
|
|
r = sp.subprocess.run(["/bin/launchctl", "bootstrap", domain, plist], capture_output=True, text=True, timeout=30)
|
|
log.info("%s started the watchdog -> %s", who, r.returncode)
|
|
return {"ok": r.returncode == 0, "detail": (r.stderr or "started").strip()[-200:]}
|
|
ok, detail = self.site.applier.restart_watchdog(None)
|
|
log.info("%s restarted the watchdog -> %s %s", who, ok, detail)
|
|
return {"ok": ok, "detail": detail}
|
|
except (OSError, sp.subprocess.TimeoutExpired) as e:
|
|
return {"ok": False, "detail": str(e)}
|
|
|
|
def static(self, name, ctype):
|
|
try:
|
|
with open(os.path.join(UI, name), "rb") as f:
|
|
return self.send(200, f.read(), ctype)
|
|
except OSError:
|
|
return self.send(404, {"error": "missing %s" % name})
|
|
|
|
def my_address(self):
|
|
cfg = self.site.cfg
|
|
host = cfg.get("site", {}).get("ingest_hostname") or (self.headers.get("Host") or "").split(":")[0] or "127.0.0.1"
|
|
return "http://%s:%d/" % (host, self.server.server_address[1])
|
|
|
|
|
|
def self_ssh(cfg):
|
|
"""ssh to this very Mac with the panel's own key: sshd may read the cloud folders a background
|
|
program is kept out of (macOS privacy), and the key is already on this Mac."""
|
|
key = os.path.expanduser(cfg.get("ssh_key", "~/.ssh/id_status_panel"))
|
|
return ["/usr/bin/ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=4", "-o", "StrictHostKeyChecking=accept-new", "-i", key, "localhost"]
|
|
|
|
|
|
def cloud_files(cfg, folder):
|
|
"""(files [(name, mtime)] newest first, how, blocked). Reads the folder here; when macOS keeps this process
|
|
out of a cloud folder, asks this Mac over ssh instead."""
|
|
try:
|
|
names = [n for n in os.listdir(folder) if n.lower().endswith(".xml")]
|
|
files = sorted(((n, os.path.getmtime(os.path.join(folder, n))) for n in names), key=lambda x: x[1], reverse=True)
|
|
if files or "/Library/CloudStorage/" not in folder:
|
|
return files, "local", False
|
|
except OSError:
|
|
pass
|
|
try:
|
|
r = subprocess.run(self_ssh(cfg) + ["cd %s && stat -f '%%m %%N' *.xml" % shlex.quote(folder)], capture_output=True, text=True, timeout=15)
|
|
if r.returncode == 0:
|
|
files = []
|
|
for line in r.stdout.splitlines():
|
|
m, _, n = line.partition(" ")
|
|
if n.lower().endswith(".xml"):
|
|
files.append((n, float(m)))
|
|
return sorted(files, key=lambda x: x[1], reverse=True), "ssh", False
|
|
log.warning("cloud folder over ssh: %s", (r.stderr or r.stdout).strip()[-200:])
|
|
except (OSError, subprocess.TimeoutExpired, ValueError) as e:
|
|
log.warning("cloud folder over ssh: %s", e)
|
|
return [], "", True
|
|
|
|
|
|
def cloud_read(cfg, path, via):
|
|
if via != "ssh":
|
|
try:
|
|
with open(path, "rb") as f:
|
|
return f.read()
|
|
except OSError:
|
|
pass
|
|
try:
|
|
r = subprocess.run(self_ssh(cfg) + ["cat %s" % shlex.quote(path)], capture_output=True, timeout=30)
|
|
return r.stdout if r.returncode == 0 else None
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
|
|
|
|
class Server(http.server.ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
allow_reuse_address = True
|
|
|
|
|
|
def serve(data_dir, port, binds=None, config_path=None):
|
|
site = Site(data_dir, config_path)
|
|
Handler.site = site
|
|
binds = binds or bind_addresses(site.cfg)
|
|
servers = []
|
|
for addr in binds:
|
|
try:
|
|
srv = Server((addr, port), Handler)
|
|
except OSError as e:
|
|
log.warning("cannot listen on %s:%d: %s", addr, port, e)
|
|
continue
|
|
threading.Thread(target=srv.serve_forever, name="http-" + addr, daemon=True).start()
|
|
servers.append(srv)
|
|
log.info("listening on http://%s:%d/", addr, port)
|
|
if not servers:
|
|
raise SystemExit("could not listen anywhere")
|
|
return site, servers
|
|
|
|
|
|
def rebind_loop(site, servers, port, stop):
|
|
"""Every minute: if a LAN port's address changed, listen there too (DHCP, a cable moved)."""
|
|
have = {s.server_address[0] for s in servers}
|
|
while not stop.wait(60):
|
|
try:
|
|
for addr in bind_addresses(site.load()):
|
|
if addr not in have:
|
|
srv = Server((addr, port), Handler)
|
|
threading.Thread(target=srv.serve_forever, name="http-" + addr, daemon=True).start()
|
|
servers.append(srv)
|
|
have.add(addr)
|
|
log.info("now also listening on http://%s:%d/", addr, port)
|
|
except Exception as e:
|
|
log.warning("rebind: %s", e)
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--data-dir", default=PARENT)
|
|
ap.add_argument("--config")
|
|
ap.add_argument("--port", type=int)
|
|
ap.add_argument("--bind", nargs="*")
|
|
ap.add_argument("--log", default=os.path.expanduser("~/Library/Logs/status-settings.log"))
|
|
args = ap.parse_args(argv)
|
|
handler = logging.handlers.RotatingFileHandler(args.log, maxBytes=1_000_000, backupCount=2)
|
|
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
log.addHandler(handler)
|
|
if sys.stderr.isatty():
|
|
log.addHandler(logging.StreamHandler())
|
|
log.setLevel(logging.INFO)
|
|
cfg = sp.load_config(args.config or os.path.join(args.data_dir, "config.json"))
|
|
port = args.port or cfg.get("site", {}).get("settings_port", 8020)
|
|
site, servers = serve(args.data_dir, port, args.bind, args.config)
|
|
stop = threading.Event()
|
|
threading.Thread(target=rebind_loop, args=(site, servers, port, stop), daemon=True).start()
|
|
try:
|
|
while True:
|
|
time.sleep(3600)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|