"""Every location's status, on North Attleboro's Companion only. Each other location's settings page tells North Attleboro's (the hub: a location with no "home") where it is, and learns nothing back: no other location ever sees another location (2026-10-20, the user: "NA should be the only one that sees every single location's status"). Only the hub, every two seconds, copies each location's key states (what that location's watchdog draws, /api/state) into its own Companion as custom variables sp_state_ / sp_hb_, and keeps two kinds of page there: LOCATIONS page one key per location: that location's ALL key, with its name; tap = its status page LOCATION page whichever location was tapped, as its own status page looks (same places, same colours, live), view only: holds and presses are for that location's own deck. The ALARMS corner goes back to the LOCATIONS page. Two pages, whatever the number of locations (NA's Companion has 97 pages of its own; Companion stops at 99): the LOCATION page's keys are cells ("c_1_3"), and the hub fills them from the tapped location's layout (sp_view says which). A location that stops answering goes purple. Pages are only written where nothing a person made sits (push_page.safe_target): an empty page, else one added at the end. State: campuses.json in the data folder. A standalone location (Apollo Beach, 2026-10-21: "they want to take care of it on their own ... i dont need to monitor AB from NA") has "standalone" set instead of a home: it tells nobody, is no hub, and lists only itself. """ import gzip import logging import hashlib import json import os import tempfile import threading import time import urllib.parse import urllib.request from settings import PARENT # noqa: F401 (puts the project folder on sys.path) from settings import discover import make_page import push_page import status_panel as sp FILE = "campuses.json" log = logging.getLogger("status-settings") # the settings page's own log (status-settings.log) POLL, REFRESH = 2.0, 60.0 SHORT = {"na": "NA", "ab": "AB", "sc": "SC", "ws": "WS"} MIRROR_MARK = "STATUS tile" # copied keys keep the tile marker, so a later rebuild may overwrite them def load(data_dir): try: with open(os.path.join(data_dir, FILE)) as f: return json.load(f) except (OSError, ValueError): return {} def save(data_dir, data): sp.write_json(os.path.join(data_dir, FILE), data) def short(cid, name=""): return SHORT.get(cid) or (name or cid)[:4].upper() def my_url(cfg): w = (discover.this_mac().get("wired") or {}).get("ip") or "127.0.0.1" return "http://%s:%d" % (w, int((cfg.get("site") or {}).get("settings_port", 8020))) def me(cfg, url=None): import pwd site = cfg.get("site") or {} # The account this panel runs as: the hub logs in with it for the nightly backup. return {"id": site.get("id") or "new", "name": site.get("name") or "This location", "url": url or my_url(cfg), "user": pwd.getpwuid(os.getuid()).pw_name} # Bumped when the pages are built differently (2: "campus" became "location", 2026-10-21), so the hub # re-imports them. PAGES_VERSION = 2 def layout_sha(cfg): keys = [(t.get("id"), t.get("label"), t.get("pos"), t.get("kind"), t.get("members")) for t in cfg.get("tiles", [])] return hashlib.sha256(json.dumps([keys, cfg.get("page_grid"), cfg.get("variables"), PAGES_VERSION], sort_keys=True, default=str).encode()).hexdigest()[:16] def state_now(site): """What /api/state answers: this location's key states as its watchdog draws them, right now.""" cfg = site.load() names = cfg.get("variables") or {} comp = site.comp return {"id": (cfg.get("site") or {}).get("id"), "name": (cfg.get("site") or {}).get("name"), "state": comp.get_custom(names.get("state", "sp_state")) or "", "hb": comp.get_custom(names.get("heartbeat", "sp_hb")) or "", "layout": layout_sha(cfg), "time": int(time.time())} def http_json(url, body=None, timeout=3): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"} if data else {}, method="POST" if data else "GET") with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8", "replace")) # --- the pages (the hub only) ----------------------------------------------------------- VIEW = {"state": "sp_state_view", "heartbeat": "sp_hb_view"} SELECT = "sp_view" VIEW_CMD = "sp_view_cmd" # a tap or hold on the LOCATION page, passed on to that location's own watchdog MODE = "sp_view_mode" # "" = that location's status page, "RESTART" = its Restart page CELLS = [(r, c) for r in range(4) for c in range(1, 8)] def cell_id(r, c): return "c_%d_%d" % (r, c) def view_state(state, keys): """A location's key states, moved onto the cells its keys sit in (empty cells blank).""" out = {cell_id(r, c): {"x": "", "b": 0, "b2": 0, "f": 0xFFFFFF, "f2": 0xFFFFFF, "s": 14} for r, c in CELLS} for k in keys: r, c = k["pos"] if (r, c) in CELLS and k["id"] in state: out[cell_id(r, c)] = state[k["id"]] return out def build_view(local_cfg, page, overview_page): """The LOCATION page: a key per cell, drawn from sp_state_view; tap = acknowledge, hold = fix, at that location (2026-10-20: the user wants to test and fix Woonsocket from NA). The ALARMS corner goes back.""" opts = make_page.settings(local_cfg) controls = make_page.nav_controls(local_cfg) hold_ms = opts["hold_ms"] for r, c in CELLS: layers = make_page.layers(cell_id(r, c), {}, VIEW, "''", opts) tap = make_page.command_action("view_%d_%d" % (r, c), "press", VIEW_CMD, "tap %d %d $(internal:time_unix)" % (r, c)) hold = make_page.command_action("view_%d_%d" % (r, c), "hold", VIEW_CMD, "hold %d %d $(internal:time_unix)" % (r, c)) make_page.place(controls, (r, c), { "type": "button-layered", "style": {"layers": layers}, "options": {"stepProgression": "auto", "stepExpression": "", "rotaryActions": False, "canModifyStyleInApis": True, "notes": "STATUS tile 'location view %d/%d' - the tapped location's key here. Press: acknowledge it there. " "Hold %g s: fix it there, like a hold on that location's own deck." % (r, c, hold_ms / 1000.0)}, "feedbacks": [], "steps": {"0": {"action_sets": {"down": [], "up": [tap], str(hold_ms): [hold]}, "options": {"runWhileHeld": [hold_ms]}}}, "localVariables": []}) back = make_page.static_button("campus_back", "ALL\\nSITES", 0x37474F, 0xFFFFFF, overview_page, "Tap: back to the LOCATIONS page (page %d)." % overview_page, "STATUS jump key") for layer in back["style"]["layers"]: if layer["type"] == "text": # which location this page shows, then the way back layer["text"] = make_page.prop("`${$(custom:%s)} ${$(custom:%s)}\\nALL SITES`" % (SELECT, MODE), True) make_page.place(controls, (3, 0), back) return make_page.page_export("%d\\nLocation" % page, page, controls, local_cfg) def select_action(cid, mode=""): return [{"type": "action", "id": make_page.stable_id("campus_" + cid + mode, "select"), "definitionId": "custom_variable_set_value", "connectionId": "internal", "options": {"name": make_page.prop(SELECT), "value": make_page.prop(cid.upper())}, "children": {}}, {"type": "action", "id": make_page.stable_id("campus_" + cid + mode, "mode"), "definitionId": "custom_variable_set_value", "connectionId": "internal", "options": {"name": make_page.prop(MODE), "value": make_page.prop(mode)}, "children": {}}] def campus_key(cid, name, names, actions, note, local_cfg): """One LOCATIONS key: that location's ALL key with its name on top.""" opts = make_page.settings(local_cfg) tag = short(cid, name) stale_text = "`%s\\n?`" % tag layers = make_page.layers("all", {}, names, stale_text, opts) text = next(layer for layer in layers if layer["type"] == "text") stale = "$(internal:time_unix) - $(custom:%s) > %d" % (names["heartbeat"], opts["stale"]) text["text"] = make_page.prop("%s ? %s : `%s\\n${jsonpath($(custom:%s), '$.all.x')}`" % (stale, stale_text, tag, names["state"]), True) text["fontsize"] = make_page.prop("auto") return {"type": "button-layered", "style": {"layers": layers}, "options": {"stepProgression": "auto", "stepExpression": "", "rotaryActions": False, "canModifyStyleInApis": True, "notes": "STATUS tile 'location %s' - %s's ALL key. %s" % (cid, name, note)}, "feedbacks": [], "steps": {"0": {"action_sets": {"down": [], "up": actions}, "options": {"runWhileHeld": []}}}, "localVariables": []} def build_overview(local_cfg, campuses, page, view_page): """Row 0: each location's ALL key (tap: its status page). Row 1: its RESTART key (tap: its Restart page).""" controls = make_page.nav_controls(local_cfg) mine = (local_cfg.get("site") or {}).get("id") rp = (local_cfg.get("restart_page") or {}).get("page") for col, cid in enumerate(sorted(campuses, key=lambda c: (c != mine, c))[:7], start=1): name = campuses[cid].get("name", cid) tag = short(cid, name) if cid == mine: key = campus_key(cid, name, local_cfg["variables"], [make_page.set_page_action("campus_" + cid, "go", local_cfg["page"])], "Tap: its status page (page %d)." % local_cfg["page"], local_cfg) restart = make_page.static_button("campus_rs_" + cid, "%s\\nRESTART" % tag, 0x37474F, 0xFFFFFF, rp, "Tap: this location's Restart page (page %s)." % rp, "STATUS jump key") if rp else None else: key = campus_key(cid, name, {"state": "sp_state_" + cid, "heartbeat": "sp_hb_" + cid}, select_action(cid) + [make_page.set_page_action("campus_" + cid, "go", view_page)], "Tap: its status page, live (page %d)." % view_page, local_cfg) restart = make_page.static_button("campus_rs_" + cid, "%s\\nRESTART" % tag, 0x37474F, 0xFFFFFF, view_page, "Tap: %s's Restart page, live (page %d); hold a key there to restart it at %s." % (name, view_page, name), "STATUS jump key") restart["steps"]["0"]["action_sets"]["up"] = select_action(cid, "RESTART") + restart["steps"]["0"]["action_sets"]["up"] make_page.place(controls, (0, col), key) if restart: make_page.place(controls, (1, col), restart) return make_page.page_export("%d\\nLocations" % page, page, controls, local_cfg) def ensure_variables(cfg, ids): from settings import campus have, _ = campus.companion_state(cfg) want = [n for cid in ids for n in ("sp_state_" + cid, "sp_hb_" + cid) if n not in have] want += [n for n in (VIEW["state"], VIEW["heartbeat"], SELECT, VIEW_CMD, MODE) if n not in have] if not want: return comp = cfg.get("companion") or {} ws = push_page.WebSocket(comp.get("host", "127.0.0.1"), int(comp.get("port", 8000)), "/trpc") try: api = push_page.Trpc(ws) for n in want: api.mutation("customVariables.create", {"name": n, "defaultVal": "0" if n.startswith("sp_hb_") else ""}) finally: ws.close() def free_page(cfg, taken, start=21): """The first page from `start` that holds nothing but Companion's own page keys.""" comp = cfg.get("companion") or {} host, port = comp.get("host", "127.0.0.1"), int(comp.get("port", 8000)) for page in range(start, 100): if page in taken: continue if not exists(host, port, page): break try: export = push_page.fetch_page(host, port, page) except (OSError, ValueError): continue ctrls = (export.get("page") or {}).get("controls") or {} if not any(ctl and ctl.get("type") not in push_page.NAV_TYPES for row in ctrls.values() for ctl in (row or {}).values()): return page # None empty: a new page at the end (Companion's own page keys on it), never in between. last = max(p for p in range(1, 100) if exists(host, port, p)) if last >= 99: raise RuntimeError("Companion has 99 pages and none is empty: free one for the locations") ws = push_page.WebSocket(host, port, "/trpc") try: push_page.Trpc(ws).mutation("pages.insert", {"asPageNumber": last + 1, "pageNames": ["LOCATIONS"]}) finally: ws.close() return last + 1 def exists(host, port, page): try: push_page.fetch_page(host, port, page) return True except (OSError, ValueError): return False def push(cfg, page, export): comp = cfg.get("companion") or {} host, port = comp.get("host", "127.0.0.1"), int(comp.get("port", 8000)) ok, why = push_page.safe_target(push_page.fetch_page(host, port, page)) if not ok: raise RuntimeError("page %d: %s" % (page, why)) with tempfile.NamedTemporaryFile(suffix=".companionconfig", delete=False) as f: f.write(gzip.compress(json.dumps(export).encode())) path = f.name try: push_page.push(host, port, page, path, export["page"]["name"]) finally: os.remove(path) # --- the loops -------------------------------------------------------------------------- class Mirror: def __init__(self, site): self.site = site self.remote = {} # id -> last /api/state answer self.keys = {} # id -> [{"id", "pos"}] of its status page (from its /api/config) self.rkeys = {} # id -> the same for its Restart page @staticmethod def is_hub(data): return not data.get("home") and not data.get("standalone") def sync_registry(self): """A location: tell the hub where this settings page is (it learns nothing back). The hub: know itself.""" cfg = self.site.load() data = load(self.site.data_dir) mine = me(cfg) home = data.get("home") if home and home.rstrip("/") != mine["url"]: try: http_json(home.rstrip("/") + "/api/campus/hello", mine) data.pop("error", None) except (OSError, ValueError) as e: data["error"] = "hub %s: %s" % (home, e) data["list"] = {mine["id"]: mine} else: data.setdefault("list", {})[mine["id"]] = mine save(self.site.data_dir, data) return data def taken_pages(self, cfg): t = {cfg.get("page"), (cfg.get("restart_page") or {}).get("page"), (cfg.get("ready_page") or {}).get("page")} return {p for p in t if p} def refresh_pages(self): """The hub: each location's layout (when it changed), the LOCATION page once, the LOCATIONS page when the roster changes.""" cfg = self.site.load() data = self.sync_registry() if not self.is_hub(data): return data mine = (cfg.get("site") or {}).get("id") others = {cid: e for cid, e in data.get("list", {}).items() if cid != mine} if not others and not (data.get("pages") or {}).get("overview"): return data # no other location yet: no pages needed pages = data.setdefault("pages", {}) layouts = data.setdefault("layouts", {}) for cid, e in sorted(others.items()): st = self.remote.get(cid) if st and (layouts.get(cid) != st.get("layout") or cid not in self.keys): remote_cfg = http_json(e["url"].rstrip("/") + "/api/config", timeout=8)["config"] self.keys[cid] = [{"id": t["id"], "pos": t["pos"]} for t in remote_cfg.get("tiles", []) if t.get("pos")] self.rkeys[cid] = [{"id": k["id"], "pos": k["pos"]} for k in (remote_cfg.get("restart_page") or {}).get("keys", []) if k.get("pos")] layouts[cid] = st.get("layout") ensure_variables(cfg, list(others)) taken = self.taken_pages(cfg) | {p for p in pages.values() if p} for which in ("overview", "view"): if not pages.get(which): pages[which] = free_page(cfg, taken) taken.add(pages[which]) built = layout_sha(cfg) if data.get("view_built") != built: push(cfg, pages["view"], build_view(cfg, pages["view"], pages["overview"])) data["view_built"] = built roster = sorted(data["list"]) if data.get("overview_roster") != roster or data.get("overview_built") != built: push(cfg, pages["overview"], build_overview(cfg, data["list"], pages["overview"], pages["view"])) data["overview_roster"], data["overview_built"] = roster, built data.pop("error", None) save(self.site.data_dir, data) return data def pass_on(self, pick, others, restart=False): """A tap or hold on the LOCATION page goes to the shown location's watchdog, through its Companion.""" text = (self.site.comp.get_custom(VIEW_CMD) or "").strip() if not text: return self.site.comp.set_custom(VIEW_CMD, "") keys = (self.rkeys if restart else self.keys).get(pick) cmds = campus_command(keys, text, time.time(), restart) if pick else [] if not cmds: return host = urllib.parse.urlparse(others[pick]["url"]).hostname try: remote = sp.Companion(host, 8000, timeout=3) before = (remote.get_custom("sp_cmd") or "").strip() remote.set_custom("sp_cmd", ";".join([before] + cmds if before else cmds)) log.info("LOCATION %s: %s (from %s's LOCATION page)", pick, "; ".join(cmds), (self.site.cfg.get("site") or {}).get("name", "here")) except OSError as e: log.warning("LOCATION %s: could not pass on %s: %s", pick, cmds, e) def poll(self): """The hub: copy each other location's key states into this Companion.""" cfg = self.site.load() data = load(self.site.data_dir) if not self.is_hub(data): return mine = (cfg.get("site") or {}).get("id") others = {cid: e for cid, e in (data.get("list") or {}).items() if cid != mine and e.get("url")} threads = [] def one(cid, url): try: st = http_json(url.rstrip("/") + "/api/state", timeout=1.5) except (OSError, ValueError): return # not answering: its heartbeat goes stale, its keys purple self.remote[cid] = st if (data.get("pages") or {}).get("overview"): self.site.comp.set_custom("sp_state_" + cid, st.get("state") or "{}") self.site.comp.set_custom("sp_hb_" + cid, str(st.get("hb") or "0")) for cid, e in others.items(): t = threading.Thread(target=one, args=(cid, e["url"]), daemon=True) t.start() threads.append(t) for t in threads: t.join(2) # The LOCATION page shows the tapped location: its keys moved onto their cells. pick = (self.site.comp.get_custom(SELECT) or "").strip().lower() # held as "WS": the back key shows it if pick not in others and others: # Nothing tapped yet (or a location that left): show the first one rather than a purple page. pick = sorted(others)[0] self.site.comp.set_custom(SELECT, pick.upper()) restart = (self.site.comp.get_custom(MODE) or "").strip().upper() == "RESTART" self.pass_on(pick if pick in others else None, others, restart) st = self.remote.get(pick) keys = (self.rkeys if restart else self.keys).get(pick) if st and keys is not None and (data.get("pages") or {}).get("view"): try: state = json.loads(st.get("state") or "{}") except ValueError: state = {} self.site.comp.set_custom(VIEW["state"], json.dumps(view_state(state, keys), separators=(",", ":"))) self.site.comp.set_custom(VIEW["heartbeat"], str(st.get("hb") or "0")) def campus_command(keys, text, now, restart=False): """The LOCATION page's taps and holds ("tap 1 3