"""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"; 3: ALL AUTO-FIX key, 2026-10-21), so the hub # re-imports them. PAGES_VERSION = 3 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"} REMOTE = make_page.REMOTE_VAR # which other locations have a problem: the blue band on NA's page-number keys ALL_CMD = "sp_all_cmd" # a tap on NA's ALL AUTO-FIX key (LOCATIONS page) ALL_AF = "sp_all_af" # what that key shows: {"t", "x", "paused"} ALL_AF_POS = (3, 1) # where AUTO-FIX sits on every location's own status page 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) make_page.place(controls, ALL_AF_POS, all_autofix_key(local_cfg)) return make_page.page_export("%d\\nLocations" % page, page, controls, local_cfg) def all_autofix_key(local_cfg): """Tap: pause auto-fix at every location at once (NA too); tap again: every location back to its hours (2026-10-21, the user: "add a key on NA to pause auto-fix at all locations"). Blue while none is paused, grey naming the paused ones.""" v = "$(custom:%s)" % ALL_AF stale = "$(internal:time_unix) - jsonpath(%s, '$.t') > 30" % v key = make_page.static_button("all_autofix", "AUTO-FIX\\nALL", 0x0B57D0, 0xFFFFFF, 1, "Tap: pause auto-fix at every location (alarms stay on, holds still fix); tap again: all " "back to their auto-fix hours.", "STATUS tile 'all autofix'") for layer in key["style"]["layers"]: if layer["type"] == "box": layer["color"] = make_page.prop("%s ? %d : (jsonpath(%s, '$.paused') > 0 ? %d : %d)" % (stale, make_page.PURPLE, v, 0x262626, 0x0B57D0), True) elif layer["type"] == "text": layer["text"] = make_page.prop("%s ? `AUTO-FIX\\nALL ?` : `AUTO-FIX\\n${jsonpath(%s, '$.x')}`" % (stale, v), True) key["steps"]["0"]["action_sets"]["up"] = [make_page.command_action("all_autofix", "press", ALL_CMD, "toggle $(internal:time_unix)")] return key def autofix_paused(state_json): """True when that location's AUTO-FIX key says it was paused by hand ("AUTO-FIX\\nOFF\\nTIL ..."), None without one.""" try: af = json.loads(state_json or "{}").get("autofix") except ValueError: return None if not af: return None lines = str(af.get("x") or "").split("\n") return len(lines) > 2 and lines[1] == "OFF" and lines[2].startswith("TIL") def all_autofix_summary(states, now): """states: {id: that location's /api/state "state" JSON}. -> what NA's ALL AUTO-FIX key shows.""" paused = sorted(cid.upper() for cid, s in states.items() if autofix_paused(s)) return {"t": int(now), "paused": len(paused), "x": ("%s\nPAUSED" % " ".join(paused)) if paused else "ALL ON"} 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, REMOTE, ALL_CMD, ALL_AF) 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 all_autofix(self, cfg, mine, others): """NA's ALL AUTO-FIX key: show which locations are paused; a tap pauses them all, or resumes them all.""" states = {cid: (self.remote.get(cid) or {}).get("state") for cid in others} states[mine] = self.site.comp.get_custom((cfg.get("variables") or {}).get("state", "sp_state")) summary = all_autofix_summary(states, time.time()) self.site.comp.set_custom(ALL_AF, json.dumps(summary, separators=(",", ":"))) text = (self.site.comp.get_custom(ALL_CMD) or "").strip(" ;") if not text: return self.site.comp.set_custom(ALL_CMD, "") stamps = [w for w in text.replace(";", " ").split() if w.isdigit()] if not stamps or time.time() - int(stamps[-1]) > 30: return # pressed while this was down: not acted on later word = "auto" if summary["paused"] else "off" hosts = {cid: urllib.parse.urlparse(e["url"]).hostname for cid, e in others.items()} hosts[mine] = "127.0.0.1" for cid, host in sorted(hosts.items()): try: comp = self.site.comp if cid == mine else sp.Companion(host, 8000, timeout=3) before = (comp.get_custom("sp_cmd") or "").strip() comp.set_custom("sp_cmd", ";".join([before, "autofix " + word] if before else ["autofix " + word])) log.info("ALL AUTO-FIX: %s %s", cid, "paused" if word == "off" else "back to its hours") except OSError as e: log.warning("ALL AUTO-FIX: %s not reached: %s", cid, e) 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) if (data.get("pages") or {}).get("overview"): self.site.comp.set_custom(REMOTE, json.dumps(remote_summary(others, self.remote, time.time()), separators=(",", ":"))) self.all_autofix(cfg, mine, others) # 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 remote_summary(others, remote, now, stale=30): """{"t", "n", "x", "f"}: the other locations with something down (their ALL key not "ALL OK", or not answering: "SC?"), named by id; "f" = 1 while any of them is still flashing (not acknowledged there).""" names, flashing = [], False for cid in sorted(others): st = remote.get(cid) tag = cid.upper() if not st or now - float(st.get("hb") or 0) > stale: names.append(tag + "?") flashing = True continue try: all_ = json.loads(st.get("state") or "{}").get("all") or {} except ValueError: all_ = {} if not all_ or str(all_.get("x") or "").startswith("ALL OK"): continue names.append(tag) flashing = flashing or all_.get("b") != all_.get("b2") return {"t": int(now), "n": len(names), "x": " ".join(names), "f": 1 if flashing else 0} def campus_command(keys, text, now, restart=False): """The LOCATION page's taps and holds ("tap 1 3