christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4dac76f703c3df199cedac409c63d8bc7cd3337e
status-indicators/settings/campus.py
T
christianandClaude Opus 5.5 d7dc931bd4 NA sees every campus; SSH from the admin laptop to every machine 
Campuses set up from NA's install line tell NA where they are and learn nothing back (no campus
ever shows another). NA alone reads each campus's key states every two seconds and keeps two pages
in its Companion (added at the end, 98 and 99, since every other page is in use): CAMPUSES, a key
per campus (its ALL key, purple when it stops answering), and CAMPUS, the tapped campus's status
page drawn live in the same places, view only. Tried on NA with a test campus pointing at NA
itself, then removed.

tools/fleet.py: an Apple Remote Desktop command that turns on Remote Login, adds the laptop's key
and reports each Mac to NA, and the laptop SSH shortcuts made from those reports (tried on NA
Translation).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 14:57:18 -04:00

435 lines
21 KiB
Python
Raw Blame History

"""A campus that sets itself up.
On the campus's Mac that runs Companion, one line installs the panel from another campus's settings
page and runs `setup`: Companion gets the panel's custom variables and blink clock, two free pages
are picked, the gear is found (settings/discover.py) and turned into keys (settings/autoconf.py),
and that config is applied like any Apply on the page. Each other computer runs one join line
(Terminal on a Mac, an administrator PowerShell on a PC); from then on the settings page looks again
every ten minutes and adds what is new by itself (Settings > Site: "Add new gear by itself").
python3 -m settings.campus setup --id ab [--name "Apollo Beach"]
python3 -m settings.campus look what it finds (nothing changes)
python3 -m settings.campus add add what is new now
"""
import argparse
import base64
import gzip
import hashlib
import io
import json
import os
import random
import re
import tarfile
import threading
import time
from settings import PARENT # noqa: F401 (puts the project folder on sys.path)
from settings import autoconf, discover, scripts
import push_page
import schema
import status_panel as sp
CAMPUSES = {"na": "North Attleboro", "ab": "Apollo Beach", "sc": "Southcoast", "ws": "Woonsocket"}
VARIABLES = ("sp_state", "sp_hb", "sp_cmd", "flash_tick")
TRIGGER_NAME = "flash_tick_500"
LOOK_EVERY = 600
SEEN, LAST = "discover-seen.json", "discover-last.json"
# What the program kit carries (no settings, keys or history of the campus that serves it).
KIT_FILES = ("status_panel.py", "readiness.py", "coremidi.py", "netgear.py", "schema.py", "make_page.py", "push_page.py",
"nav_keys.py", "atem_watch.js", "atem_state.js", "nav_templates.json")
KIT_DIRS = ("settings", "remote-access", "companion-objects")
# --- Companion: the panel's variables, its blink clock, and two free pages ---------------
def flash_trigger_export(build="5.0.7", name=TRIGGER_NAME, enabled=True):
"""A Companion export holding one trigger: flash_tick 1 for half a second, every second."""
def nid():
return base64.urlsafe_b64encode(os.urandom(15)).decode().rstrip("=")[:21]
def setvar(v):
return {"id": nid(), "definitionId": "custom_variable_set_value", "connectionId": "internal",
"options": {"name": {"isExpression": False, "value": "flash_tick"}, "value": {"isExpression": False, "value": v}},
"type": "action", "children": {}}
tid = nid()
trig = {"type": "trigger", "options": {"name": name, "enabled": enabled, "sortOrder": 0},
"actions": [setvar("1"), {"id": nid(), "definitionId": "wait", "connectionId": "internal",
"options": {"time": {"isExpression": False, "value": "500"}}, "type": "action", "children": {}}, setvar("0")],
"condition": [], "events": [{"id": nid(), "type": "interval", "enabled": True, "options": {"seconds": 1}}], "localVariables": []}
return tid, {"version": 12, "type": "full", "companionBuild": build, "triggers": {tid: trig}, "triggerCollections": [],
"custom_variables": {}, "customVariablesCollections": [], "instances": {}, "connectionCollections": []}
def companion_state(cfg):
raw = discover.companion_get(cfg, "/int/export/custom?buttons=false&connections=false&surfaces=&triggers=true&customVariables=true"
"&expressionVariables=false&includeSecrets=false&imageLibrary=false&format=json", 10)
d = json.loads(raw)
names = {(t.get("options") or {}).get("name") for t in (d.get("triggers") or {}).values()}
return set(d.get("custom_variables") or {}), names
def companion_prepare(cfg):
"""Create what the pages need inside Companion and is otherwise made by hand in its UI. Returns words."""
comp = cfg.get("companion") or {}
host, port = comp.get("host", "127.0.0.1"), int(comp.get("port", 8000))
have_vars, triggers = companion_state(cfg)
done = []
missing = [v for v in VARIABLES if v not in have_vars]
if not missing and TRIGGER_NAME in triggers:
return done
ws = push_page.WebSocket(host, port, "/trpc")
try:
api = push_page.Trpc(ws)
for v in missing:
api.mutation("customVariables.create", {"name": v, "defaultVal": "0" if v == "flash_tick" else ""})
done.append("custom variable %s" % v)
if TRIGGER_NAME not in triggers:
tid, export = flash_trigger_export()
data = gzip.compress(json.dumps(export).encode())
session = api.mutation("importExport.prepareImport.start", {"name": "status-panel-blink.companionconfig", "size": len(data)})
api.mutation("importExport.prepareImport.uploadChunk", {"sessionId": session, "offset": 0, "data": base64.b64encode(data).decode()})
summary = api.mutation("importExport.prepareImport.complete", {"sessionId": session, "expectedChecksum": hashlib.sha1(data).hexdigest(), "userData": None})
if isinstance(summary, list) and summary and summary[0]:
raise RuntimeError("Companion could not read the blink trigger: %s" % summary[0])
api.mutation("importExport.importTriggers", {"selectedTriggerIds": [tid], "connectionIdRemapping": {}, "replaceExisting": False})
done.append("the %s trigger (the keys' blink clock)" % TRIGGER_NAME)
finally:
ws.close()
return done
def pick_pages(cfg, count=2, start=19):
"""Pages that hold nothing a person made (or only this panel's keys): the first `count` from `start`."""
comp = cfg.get("companion") or {}
host, port = comp.get("host", "127.0.0.1"), int(comp.get("port", 8000))
out = []
for page in list(range(start, 100)) + list(range(2, start)):
try:
ok, _ = push_page.safe_target(push_page.fetch_page(host, port, page))
except (OSError, ValueError):
continue
if ok:
out.append(page)
if len(out) == count:
return out
raise RuntimeError("Companion has no %d empty pages for the status panel" % count)
# --- the first setup --------------------------------------------------------------------
def template(data_dir):
for path in (os.path.join(data_dir, "kit", "sites", "_template", "config.json"),
os.path.join(PARENT, "sites", "_template", "config.json")):
if os.path.exists(path):
with open(path) as f:
return json.load(f)
raise RuntimeError("no sites/_template/config.json to start from")
def is_fresh(cfg):
return (cfg.get("site") or {}).get("id") in ("new", "", None) and len(cfg.get("tiles") or []) <= 3
def load_seen(data_dir):
try:
with open(os.path.join(data_dir, SEEN)) as f:
return set(json.load(f))
except (OSError, ValueError):
return set()
def save_seen(data_dir, seen):
sp.write_json(os.path.join(data_dir, SEEN), sorted(seen))
def wait_applied(applier, limit=240):
t0 = time.time()
while time.time() - t0 < limit:
st = applier.status or {}
if st.get("state") not in ("running",):
return st
time.sleep(1)
return applier.status or {}
def setup(data_dir, cid, name=None, say=print, home=None):
"""The first setup of a campus (or, when it is set up already, a look for what is new)."""
from settings import server, campuses
if home: # where it was installed from: the campus list lives there
data = campuses.load(data_dir)
data["home"] = home
campuses.save(data_dir, data)
site = server.Site(data_dir)
cfg = site.load()
if not is_fresh(cfg):
say("%s is set up already: looking for what is new." % (cfg.get("site") or {}).get("name", "This campus"))
return look_and_add(site, say=say, force=True)
name = name or CAMPUSES.get(cid) or cid.upper()
for line in companion_prepare(cfg):
say("Companion: added %s" % line)
pages = pick_pages(cfg)
say("Companion pages %d (status) and %d (restart) are free: using them." % tuple(pages))
inv = discover.look(cfg, data_dir)
if inv.get("companion_error"):
raise RuntimeError("Companion's HTTP API does not answer (%s): turn it on in Companion's settings, then run this again" % inv["companion_error"])
pin = "%04d" % random.SystemRandom().randrange(1000, 10000)
new, seen = autoconf.build(template(data_dir), inv, {"id": cid, "name": name}, pin, pages)
errors, _ = schema.validate(new)
if errors:
raise RuntimeError("the config auto-setup made does not pass the rules: %s" % "; ".join(errors[:4]))
r = site.applier.start(new, "auto-setup: first setup of %s" % name, "auto-setup", "127.0.0.1")
if not r.get("ok"):
raise RuntimeError(r.get("error") or "the Apply did not start")
st = wait_applied(site.applier)
save_seen(data_dir, seen)
keys = [t["label"] for t in new["tiles"] if not t.get("kind")]
record(data_dir, {"at": int(time.time()), "added": ["%d keys: %s" % (len(keys), ", ".join(keys))], "inventory": summary(inv)})
say("%s: %s" % (st.get("state"), st.get("outcome", "")))
say("Keys: %s" % ", ".join(keys))
say("Editing code (PIN) for the settings page: %s" % pin)
return new
def summary(inv):
known = set(autoconf.COMPANION)
return {"connections": len(inv.get("connections") or []),
"not_known": sorted("%s (%s)" % (c["label"], c["module"]) for c in inv.get("connections") or [] if c["module"] not in known and c["module"] != "internal"),
"dante": len((inv.get("dante") or {}).get("devices") or []),
"computers": {a: {"name": c.get("name"), "os": c.get("os"), "ip": c.get("ip"), "reachable": bool(c.get("reachable")),
"error": c.get("error", "")} for a, c in (inv.get("computers") or {}).items()},
"midi_sessions": sum(len(s or {}) for s in (inv.get("midi") or {}).values()),
"wired": ((inv.get("ingest") or {}).get("wired") or {}).get("net")}
def record(data_dir, result):
sp.write_json(os.path.join(data_dir, LAST), result)
# --- every ten minutes: what is new ------------------------------------------------------
def blocked(cfg, applier, now=None):
"""Why auto-setup may not apply right now, or None: never during a service or a recording."""
now = now or time.time()
live, why = applier.live()
if live:
return "live now (%s)" % why
if sp.in_windows((cfg.get("auto_fix") or {}).get("windows") or [], now):
return "inside the auto-fix hours (a service day): waits until they end"
if applier.busy():
return "an Apply is running"
return None
def look_and_add(site, say=lambda s: None, force=False):
"""Look, and add what is new (when the campus allows it, and not during a service)."""
cfg = site.load()
inv = discover.look(cfg, site.data_dir)
seen = load_seen(site.data_dir)
new, added, now_seen = autoconf.extend(cfg, inv, seen)
result = {"at": int(time.time()), "added": [], "waiting": [], "inventory": summary(inv)}
if added:
auto = (cfg.get("discover") or {}).get("auto")
why = blocked(cfg, site.applier)
if not (auto or force):
result["waiting"] = added
result["why"] = "adding by itself is off (Settings > Site)"
elif why:
result["waiting"] = added
result["why"] = why
else:
errors, _ = schema.validate(new)
if errors:
result["waiting"], result["why"] = added, "would not pass the rules: %s" % "; ".join(errors[:3])
else:
r = site.applier.start(new, "auto-setup found: %s" % ", ".join(added)[:180], "auto-setup", "127.0.0.1")
if r.get("ok"):
st = wait_applied(site.applier)
if st.get("state") == "done":
save_seen(site.data_dir, seen | set(now_seen))
result["added"] = added
else:
result["waiting"], result["why"] = added, st.get("outcome") or "the Apply did not finish"
else:
result["waiting"], result["why"] = added, r.get("error", "the Apply did not start")
record(site.data_dir, result)
say("added: %s" % (", ".join(result["added"]) or "nothing"))
if result["waiting"]:
say("waiting: %s (%s)" % (", ".join(result["waiting"]), result.get("why")))
return result
def loop(site, stop, poke):
"""The settings page's own background look: every ten minutes, and soon after a computer joins."""
stop.wait(60)
while not stop.is_set():
try:
if (site.load().get("discover") or {}).get("auto") and not is_fresh(site.cfg):
look_and_add(site)
except Exception as e: # noqa: BLE001 - the page must keep running
record(site.data_dir, dict(load_last(site.data_dir), error=str(e), at=int(time.time())))
poke.wait(LOOK_EVERY)
if poke.is_set():
poke.clear()
stop.wait(20) # a computer just joined: give its SSH a moment
def load_last(data_dir):
try:
with open(os.path.join(data_dir, LAST)) as f:
return json.load(f)
except (OSError, ValueError):
return {}
# --- the lines people run ----------------------------------------------------------------
def program_kit(data_dir):
"""The program as a .tgz: the installed code, the settings page, the templates, install.sh."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for f in KIT_FILES:
p = os.path.join(data_dir, f)
if os.path.exists(p):
tar.add(p, f)
for d in KIT_DIRS:
p = os.path.join(data_dir, d)
if os.path.isdir(p):
tar.add(p, d, filter=lambda ti: None if "__pycache__" in ti.name or ti.name.endswith(".pyc") else ti)
kit = os.path.join(data_dir, "kit")
for name in ("install.sh", "README.md"):
p = os.path.join(kit, name)
if os.path.exists(p):
tar.add(p, name)
p = os.path.join(kit, "sites", "_template")
if os.path.isdir(p):
tar.add(p, "sites/_template")
return buf.getvalue()
def shq(s):
return "'" + str(s).replace("'", "'\"'\"'") + "'"
def install_script(base, cid, support_key=None):
"""What a new campus's Companion Mac runs: install from `base` (this page), then set itself up."""
m = re.match(r"[a-z0-9]+", str(cid).lower())
cid = m.group(0)[:8] if m else "new"
name = CAMPUSES.get(cid, cid.upper())
key = support_key or ""
return """#!/bin/zsh
# Status panel for %(name)s: installs on this Mac (the one that runs Companion) and sets itself up.
# Safe to run again: it then updates the program and looks for new gear.
set -e
SRC=%(base)s
if ! /usr/bin/xcode-select -p >/dev/null 2>&1; then
echo "First the Command Line Tools: a window opens - click Install, wait for it, then run this line again."
/usr/bin/xcode-select --install 2>/dev/null || true
exit 1
fi
DIR="$HOME/status-panel-install"
mkdir -p "$DIR" && cd "$DIR"
echo "Downloading the status panel from $SRC ..."
/usr/bin/curl -fsSL "$SRC/campus/program.tgz" | /usr/bin/tar -xz
[[ -f "$HOME/.ssh/id_status_panel" ]] || { mkdir -p "$HOME/.ssh" && chmod 700 "$HOME/.ssh" && /usr/bin/ssh-keygen -q -t ed25519 -N "" -C "status-panel@$(scutil --get LocalHostName)" -f "$HOME/.ssh/id_status_panel"; }
SUPPORT=%(key)s
if [[ -n "$SUPPORT" ]]; then
touch "$HOME/.ssh/authorized_keys" && chmod 600 "$HOME/.ssh/authorized_keys"
grep -qF "$SUPPORT" "$HOME/.ssh/authorized_keys" || echo "$SUPPORT" >> "$HOME/.ssh/authorized_keys"
fi
zsh install.sh --site %(cid)s
cd "$HOME/Library/Application Support/status-panel"
/usr/bin/python3 -m settings.campus setup --id %(cid)s --name %(qname)s --home "$SRC"
if ! /usr/bin/nc -z -G 2 127.0.0.1 22 >/dev/null 2>&1; then
echo
echo "One more thing: turn on Remote Login (System Settings > General > Sharing), so North Attleboro can help remotely."
fi
echo
echo "Next, on each other computer here (ProPresenter, Tracks, Lighting...):"
echo " Mac: open Terminal and paste curl -fsSL http://$(ipconfig getifaddr en0 2>/dev/null || hostname):8020/join | sh"
echo " PC: open PowerShell as administrator and paste irm http://$(ipconfig getifaddr en0 2>/dev/null || hostname):8020/join.ps1 | iex"
""" % {"name": name, "base": shq(base), "key": shq(key), "cid": cid, "qname": shq(name)}
def join_mac(cfg, base):
key, _ = scripts.public_key(cfg)
if not key:
raise RuntimeError("this Mac has no login key yet (run the install line again)")
ingest = (cfg.get("site") or {}).get("ingest_name") or "the Companion Mac"
return """#!/bin/sh
# Joins this Mac to %(ingest)s's status panel: lets it log in with its key (no password) and says this Mac is here.
KEY=%(key)s
mkdir -p "$HOME/.ssh" && touch "$HOME/.ssh/authorized_keys" && chmod 700 "$HOME/.ssh" && chmod 600 "$HOME/.ssh/authorized_keys"
grep -qF "$KEY" "$HOME/.ssh/authorized_keys" || echo "$KEY" >> "$HOME/.ssh/authorized_keys"
if ! /usr/bin/nc -z -G 2 127.0.0.1 22 >/dev/null 2>&1; then
echo "Remote Login is off. Turning it on (it asks for this Mac's administrator password):"
if ! sudo /usr/sbin/systemsetup -setremotelogin on >/dev/null 2>&1; then
echo "Please turn it on in System Settings > General > Sharing > Remote Login, then paste this line again."
open "x-apple.systempreferences:com.apple.Sharing-Settings.extension" 2>/dev/null
exit 1
fi
fi
esc() { printf '%%s' "$1" | sed 's/[\\\\"]//g'; }
IPS=$(/sbin/ifconfig | awk '/inet /{print $2}' | grep -v '^127\\.' | sed 's/.*/"&"/' | paste -sd, -)
BODY="{\\"user\\":\\"$(esc "$(id -un)")\\",\\"name\\":\\"$(esc "$(scutil --get ComputerName)")\\",\\"hostname\\":\\"$(esc "$(scutil --get LocalHostName).local")\\",\\"os\\":\\"mac\\",\\"ips\\":[$IPS]}"
if /usr/bin/curl -fsS -X POST -H 'Content-Type: application/json' -d "$BODY" %(hello)s >/dev/null; then
echo "Joined. %(ingest)s sets this Mac up in about a minute: watch its status page."
else
echo "Could not reach %(ingest)s at %(base)s - is this Mac on the wired network?"
exit 1
fi
""" % {"ingest": ingest, "key": shq(key), "hello": shq(base + "/api/hello"), "base": base}
def join_windows(cfg, base):
"""The Windows set-up script, with the programs found running on that PC instead of a list."""
text = scripts.fill(scripts.template("windows-setup.ps1.template"), dict(
scripts.site_values(cfg), PC_NAME="this PC", APPS=" # filled in below", PROGRAMS="the ones running now",
SERVICES="'rtpMIDIService'"))
known = "; ".join("'%s' = '%s'" % (a["exe"], a["task"].replace("Start ", "", 1)) for a in autoconf.WIN_APPS)
detect = ("$known = @{ %s }\n"
"$apps = @(Get-Process -ErrorAction SilentlyContinue | Where-Object { $_.Path -and $known.ContainsKey((Split-Path $_.Path -Leaf)) } |\n"
" Sort-Object Path -Unique | ForEach-Object { @{ Name = $known[(Split-Path $_.Path -Leaf)]; Exe = $_.Path } })" % known)
text = re.sub(r"\$apps = @\(\n # filled in below\n\)", lambda m: detect, text)
if "$known = @{" not in text:
raise RuntimeError("the Windows template changed: cannot put the program finder in")
text = text.replace("# Status panel remote access", "# Run in PowerShell as administrator: irm %s/join.ps1 | iex\n# Status panel remote access" % base, 1)
text += """
# 5. Tell %(ingest)s this PC is here: it sets it up within a minute.
$hello = @{ user = $env:USERNAME; name = $env:COMPUTERNAME; os = 'windows'
ips = @(Get-NetIPAddress -AddressFamily IPv4 -ErrorAction SilentlyContinue | Where-Object { $_.IPAddress -notlike '127.*' } | ForEach-Object IPAddress)
macs = @($nics | ForEach-Object MacAddress); tasks = @($apps | ForEach-Object { 'Start ' + $_.Name }) }
try {
Invoke-RestMethod -Method Post -Uri '%(hello)s' -ContentType 'application/json' -Body ($hello | ConvertTo-Json) | Out-Null
Write-Host 'Joined: %(ingest)s sets this PC up in about a minute.'
} catch { Write-Host ('Could not reach %(ingest)s: ' + $_.Exception.Message) }
""" % {"ingest": (cfg.get("site") or {}).get("ingest_name") or "the Companion Mac", "hello": base + "/api/hello"}
return text
# --- command line ------------------------------------------------------------------------
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("what", choices=("setup", "look", "add"))
ap.add_argument("--id", default="new")
ap.add_argument("--name")
ap.add_argument("--data-dir", default=PARENT)
ap.add_argument("--home", help="the settings page this campus was installed from (the campus list lives there)")
a = ap.parse_args(argv)
if a.what == "setup":
setup(a.data_dir, a.id, a.name, home=a.home)
elif a.what == "look":
cfg = sp.load_config(os.path.join(a.data_dir, "config.json"))
inv = discover.look(cfg, a.data_dir)
print(json.dumps(summary(inv), indent=1))
new, added, _ = autoconf.extend(cfg, inv, load_seen(a.data_dir))
print("would add:", ", ".join(added) or "nothing")
else:
from settings import server
look_and_add(server.Site(a.data_dir), say=print, force=True)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink