status-indicators/settings/keyguide.py

"""The teaching pages, built from the config: "What each key does", the cheat sheet, "Where things are"."""
import html
import os
import re

from settings import PARENT  # noqa: F401
import schema
import status_panel as sp

CSS = """
body { font-family: -apple-system, 'Helvetica Neue', Arial, sans-serif; margin: 0; padding: 24px 16px; color: #1d2420; background: #fff; }
h1 { font-size: 1.7rem; margin: 0 0 6px; } h2 { font-size: 1.2rem; margin: 28px 0 8px; border-top: 2px solid #d6dcd6; padding-top: 10px; }
h3 { font-size: 1.05rem; margin: 18px 0 4px; }
p, li { line-height: 1.45; } .muted { color: #5a665f; } code { background: #eef2ee; padding: 1px 5px; border-radius: 4px; font-size: .9em; }
.grid { display: grid; grid-template-columns: repeat(8, 1fr); gap: 4px; max-width: 560px; margin: 10px 0; }
.cell { background: #1b1f1d; color: #e8ebe6; font: 600 .72rem/1.1 -apple-system, sans-serif; border-radius: 5px; aspect-ratio: 1; display: flex; align-items: center; justify-content: center; text-align: center; padding: 2px; }
.cell.nav { background: #37474f; } .cell.empty { background: #2a2e2c; opacity: .4; }
table { border-collapse: collapse; width: 100%; font-size: .95rem; } th, td { text-align: left; vertical-align: top; padding: 6px 10px; border-bottom: 1px solid #e3e7e3; }
.key { margin: 14px 0 18px; padding: 10px 14px; border: 1px solid #d6dcd6; border-radius: 8px; }
.key h3 { margin-top: 0; } .words { font-family: ui-monospace, Menlo, monospace; font-size: .85rem; }
.legend span { display: inline-block; margin-right: 14px; } .legend i { display: inline-block; width: 11px; height: 11px; border-radius: 3px; margin-right: 5px; vertical-align: -1px; }
@media print { body { padding: 0; } .key { break-inside: avoid; } }
"""

FAULT_WORDS = {
    "conn": ["NO ANSWER", "UNREACHABLE", "REFUSED", "LOGIN", "OFFLINE", "WARNING", "DISABLED", "NOT IN COMPANION", "NO DATA"],
    "hyperdeck": ["NO DISK", "FULL (minutes left)", "LOW (minutes left)", "NOT REC", "OFFLINE", "NO ANSWER"],
    "dante": ["<device> GONE"], "shure": ["<channel> BATT", "<channel> INTERF"], "psm": ["<channel> RF MUTE"],
    "tascam": ["<the recorder's own error or caution>"], "ping": ["<name> NO PING"], "launchd": ["<name> NOT LOADED", "<name> STOPPED"],
    "process": ["<name> STOPPED"], "http": ["<name> UNREACH"], "disk": ["DISK <free>GB"],
    "midi_peer": ["MIDI LOST", "NO SESSION", "MIDI <state>"], "pref": [], "spotify": ["SPOTIFY CLOSED", "NOT PLAYING", "WRONG PLAYLIST"],
    "mdns": ["<name> GONE"], "midi_rx": ["<name> SILENT"], "remote": [], "artnet": ["SILENT"], "atem": [],
}


def e(text):
    return html.escape(str(text))


def check_sentence(chk, cfg):
    t = chk["type"]
    hosts = cfg.get("hosts", {})
    who = lambda alias: hosts.get(alias, {}).get("name", alias)
    if t == "conn":
        return "Companion's connection %s is good" % e(chk["conn"])
    if t == "hyperdeck":
        s = "HyperDeck %s is connected" % e(chk["conn"])
        s += ", may have no disk" if chk.get("optional") else ", has a disk with time left"
        return s + (", and records whenever another deck records" if chk.get("rec_sync", True) else "")
    if t == "dante":
        devices = chk.get("devices", {})
        return "all %d Dante devices are present (%s)" % (len(devices), ", ".join(sorted(devices.values() if isinstance(devices, dict) else devices)))
    if t == "shure":
        return "Shure receiver %s: %d channel%s, batteries and interference" % (e(chk["conn"]), chk.get("channels", 1), "" if chk.get("channels", 1) == 1 else "s")
    if t == "psm":
        return "PSM transmitter %s: %d channel%s not RF-muted" % (e(chk["conn"]), chk.get("channels", 1), "" if chk.get("channels", 1) == 1 else "s")
    if t == "tascam":
        return "the Tascam recorder (%s) reports no error or caution" % e(chk["conn"])
    if t == "ping":
        return "%s answers ping (%s)" % (e(chk.get("name") or chk.get("conn") or chk.get("host")), e(chk.get("conn") and "at Companion's address for %s" % chk["conn"] or chk.get("host")))
    if t == "launchd":
        return "%s (a helper on this Mac) is loaded and running" % e(chk.get("name") or chk["label"])
    if t == "process":
        return "%s is running on this Mac" % e(chk.get("name") or chk["match"])
    if t == "http":
        return "%s answers (%s)" % (e(chk.get("name") or chk.get("msg") or "a web address"), e(chk["url"]))
    if t == "disk":
        return "the disk %s has more than %d GB free" % (e(chk["path"]), chk.get("warn_gb", 20))
    if t == "midi_peer":
        where = ("on %s" % who(chk["host"])) if chk.get("host") else "on this Mac"
        return "the network MIDI session %s %s has %s in it" % (e(chk["session"]), where, e(chk["peer"]))
    if t == "midi_wired":
        return "every peer in the network MIDI sessions (here and on the other Macs) comes over the wired network %s, not Wi-Fi or Dante" % e(", ".join(chk.get("nets") or []))
    if t == "pref":
        return "the setting %s in %s is %s" % (e(chk["key"]), e(chk["domain"]), e(chk["want"]))
    if t == "spotify":
        return "Spotify on this Mac is playing %s" % e((cfg.get("spotify") or {}).get("name") or "the service playlist")
    if t == "mdns":
        return "%s is advertised on the network (Bonjour)" % e(chk.get("name") or chk["instance"])
    if t == "midi_rx":
        return "MIDI is arriving from %s" % e(chk.get("name") or chk["source"])
    if t == "remote":
        return "on %s: %s" % (who(chk["host"]), e(chk.get("msg", "a command succeeds").lower().replace(" not ", " is ")))
    if t == "artnet":
        return "the lighting program at %s keeps sending Art-Net" % e(chk["source"])
    if t == "atem":
        return "the switcher's %s settings match the saved file" % e(chk["part"])
    return e(t)


def fix_words(fix_id, cfg):
    if fix_id == sp.RESTART_CONNECTION:
        return "reconnects its Companion connection"
    if fix_id == sp.DANTE_AUTO:
        return "power-cycles that device's PoE port if it has one, otherwise reconnects Companion's Dante module"
    fix = cfg.get("fixes", {}).get(fix_id)
    return e(fix.get("label", fix_id)) if fix else e(fix_id)


def plain(markup):
    """The same words without tags or entities, for the page's JSON."""
    return html.unescape(re.sub(r"<[^>]+>", "", markup))


def hold_words(tile, cfg):
    always = tile.get("hold_always")
    fixes, per_sub = [], []
    for chk in tile.get("checks", []):
        fx = chk.get("fix")
        if isinstance(fx, dict) and len(set(fx.values())) == 1:
            fx = next(iter(fx.values()))                 # one fix for every sub-check: a plain fix
        if isinstance(fx, dict):
            for sub, f in fx.items():
                if f == sp.DANTE_AUTO or sub == "*":
                    per_sub.append(fix_words(f, cfg) if sub == "*" else "%s: %s" % (e(sub), fix_words(f, cfg)))
                else:
                    per_sub.append("%s: %s" % (e(sub), fix_words(f, cfg)))
        elif fx:
            fixes.append(fx)
        elif chk["type"] in sp.RESTARTABLE and chk.get("conn"):
            fixes.append(sp.RESTART_CONNECTION + ":" + chk["conn"])
    words, seen = [], set()
    for f in fixes:
        if f in seen:
            continue
        seen.add(f)
        if f.startswith(sp.RESTART_CONNECTION + ":"):
            words.append("reconnects Companion's %s connection" % e(f.split(":", 1)[1]))
        elif f == always:
            words.append(fix_words(f, cfg) + " <b>(also when the key is green)</b>")
        else:
            words.append(fix_words(f, cfg))
    if always and always not in seen:
        words.insert(0, fix_words(always, cfg) + " <b>(also when the key is green)</b>")
    out = []
    if words:
        out.append("; ".join(words))
    if per_sub:
        out.append("for a named device, " + "; ".join(dict.fromkeys(per_sub)))
    if tile.get("kind") == "network":
        out.append("clears a remembered network blip")
    if tile.get("members"):
        out.append("fixes whatever is failing on %s" % ", ".join(e(m.upper()) for m in tile["members"]))
    return "; ".join(out) if out else "nothing (there is no automatic fix for this key)"


def key_guide(cfg):
    """What each key does: one entry per key, in deck order, plus the restart keys."""
    ph = schema.phrases(cfg)
    auto = set((cfg.get("auto_fix") or {}).get("fixes", []))
    tiles = {t["id"]: t for t in cfg["tiles"]}
    parts = ["<!doctype html><meta charset='utf-8'><title>What each key does</title><style>%s</style>" % CSS,
             "<h1>What each key does</h1><p class='muted'>Made from the panel's settings; it changes whenever they do. "
             "Press a key = acknowledge its alarm. Hold %s = run its fix.</p>" % e(ph["hold"].replace("HOLD ", "")),
             grid_html(cfg)]
    order = sorted((t for t in cfg["tiles"] if "pos" in t), key=lambda t: (t["pos"][0], t["pos"][1]))
    order += [t for t in cfg["tiles"] if "pos" not in t]
    for t in order:
        parts.append("<div class='key' id='key-%s'><h3>%s</h3>" % (e(t["id"]), e(t["label"])))
        pos = "row %d, column %d" % tuple(t["pos"]) if "pos" in t else "no key of its own (shown inside its group)"
        parts.append("<p class='muted'>%s%s</p>" % (pos, "" if t.get("kind") != "master" else " - sums up every key"))
        if t.get("members"):
            parts.append("<p>Shows the worst of: %s. Hold: %s.</p>" % (", ".join(e(tiles[m]["label"]) for m in t["members"]), hold_words(t, cfg)))
        if t.get("kind") == "service":
            parts.append("<p>%s</p>" % e(ph["alarms"]))
        if t.get("kind") == "network":
            n = cfg.get("network", {})
            parts.append("<p>Watches this Mac's network port%s, and calls it a blip when %d connections from %d groups drop within %d s "
                         "(shown for %d minutes). Hold: clears the blip.</p>" % (
                             (" and pings %s" % e(n["gateway"])) if n.get("gateway") else "", n.get("burst_connections", 4),
                             n.get("burst_groups", 3), n.get("burst_window_seconds", 10), n.get("remember_seconds", 300) // 60))
        if t.get("checks"):
            parts.append("<p><b>Checks:</b></p><ul>")
            for chk in t["checks"]:
                words = FAULT_WORDS.get(chk["type"], [])
                if chk.get("msg"):
                    words = [chk["msg"]]
                parts.append("<li>%s%s</li>" % (check_sentence(chk, cfg), (" <span class='words muted'>&rarr; %s</span>" % e(", ".join(words))) if words else ""))
            parts.append("</ul>")
            parts.append("<p><b>Holding:</b> %s.</p>" % hold_words(t, cfg))
            own = set()
            for chk in t["checks"]:
                fx = chk.get("fix")
                own |= set(fx.values()) if isinstance(fx, dict) else ({fx} if fx else set())
            if t.get("hold_always"):
                own.add(t["hold_always"])
            byself = sorted(f for f in own if f in auto)
            if byself:
                parts.append("<p class='muted'>Auto-fix %s: %s.</p>" % (e(ph["auto_fix"]), ", ".join(e(cfg["fixes"][f].get("short", f)) for f in byself)))
            failover = sorted(f for f in own if f in cfg.get("fixes", {}) and cfg["fixes"][f].get("failover"))
            if failover:
                parts.append("<p class='muted'>Failover %s: %s.</p>" % (e(ph["failover"]), ", ".join(e(cfg["fixes"][f].get("short", f)) for f in failover)))
        parts.append("</div>")
    rp = cfg.get("restart_page")
    if rp:
        parts.append("<h2>Restart page (page %d)</h2><p>Each key restarts one thing even when the Status page says it is fine. "
                     "It flashes HOLD TO FIX while that thing is down.</p>" % rp["page"])
        for k in sorted(rp.get("keys", []), key=lambda k: (k["pos"][0], k["pos"][1])):
            parts.append("<div class='key'><h3>%s</h3><p class='muted'>row %d, column %d</p><p>Hold: %s.</p></div>"
                         % (e(k["label"]), k["pos"][0], k["pos"][1], fix_words(k["fix"], cfg)))
    parts.append("<h2>Colours</h2><p class='legend'><span><i style='background:#007A2E'></i>green: fine</span>"
                 "<span><i style='background:#FF0000'></i>flashing red: the problem, named</span>"
                 "<span><i style='background:#8B0000'></i>dark red: acknowledged</span>"
                 "<span><i style='background:#FFA000'></i>amber: a warning, or a fault that keeps coming back (xN)</span>"
                 "<span><i style='background:#6A1B9A'></i>purple: the watchdog is down</span>"
                 "<span><i style='background:#2962FF'></i>blue: a fix is running</span></p>")
    return "\n".join(parts)


def grid_html(cfg, restart=False):
    rp = cfg.get("restart_page")
    cells = {}
    if restart and rp:
        for k in rp.get("keys", []):
            cells[tuple(k["pos"])] = k["label"]
        cells[tuple(rp["back_pos"])] = "STATUS PAGE"
    else:
        for t in cfg["tiles"]:
            if "pos" in t:
                cells[tuple(t["pos"])] = t["label"]
        if rp:
            cells[tuple(rp["jump_pos"])] = "RESTART PAGE"
    nav = {(0, 0): "UP", (1, 0): str(rp["page"] if restart and rp else cfg["page"]), (2, 0): "DOWN"}
    out = ["<div class='grid'>"]
    for r in range(4):
        for c in range(8):
            if (r, c) in nav:
                out.append("<div class='cell nav'>%s</div>" % e(nav[(r, c)]))
            elif (r, c) in cells:
                out.append("<div class='cell'>%s</div>" % e(cells[(r, c)]))
            else:
                out.append("<div class='cell empty'></div>")
    out.append("</div>")
    return "".join(out)


def cheat_sheet(cfg, address):
    ph = schema.phrases(cfg)
    site = cfg.get("site", {})
    hosts = cfg.get("hosts", {})
    rows = "".join("<tr><td>%s</td><td>%s</td></tr>" % (e(h.get("name", a)), e(h.get("role", ""))) for a, h in hosts.items())
    return "\n".join([
        "<!doctype html><meta charset='utf-8'><title>Status panel cheat sheet</title><style>%s</style>" % CSS,
        "<h1>Status panel - %s</h1>" % e(site.get("name", "")),
        "<p><b>Green</b> = fine. <b>Flashing red</b> = the problem, named on the key. <b>Press</b> = acknowledge (it stops flashing). "
        "<b>%s</b> = run the fix. <b>Purple</b> = the watchdog is down: open the settings page and press Restart the watchdog.</p>" % e(ph["hold"].replace("HOLD ", "Hold ")),
        "<p>%s Auto-fix %s. %s.</p>" % (e(ph["alarms"]), e(ph["auto_fix"]), e(ph["live"][0].upper() + ph["live"][1:])),
        "<h2>Status page (page %d)</h2>" % cfg["page"], grid_html(cfg),
        ("<h2>Restart page (page %d)</h2>" % cfg["restart_page"]["page"] + grid_html(cfg, restart=True)) if cfg.get("restart_page") else "",
        "<h2>The computers</h2><table>%s</table>" % rows,
        "<h2>Where to look</h2><p>The settings page: <b>%s</b> (every key, every computer, the logs, the history, help). "
        "Contacts: %s</p>" % (e(address), e(site.get("contacts") or "-")),
    ])


def where_things_are(cfg, data_dir, address):
    """Every file, folder, log, helper and port, per computer. The one page allowed to show file names."""
    site = cfg.get("site", {})
    hosts = cfg.get("hosts", {})
    parts = ["<!doctype html><meta charset='utf-8'><title>Where things are</title><style>%s</style>" % CSS,
             "<h1>Where things are</h1><p class='muted'>For the person who maintains it.</p>",
             "<h2>%s (this Mac)</h2><ul>" % e(site.get("ingest_name", "the Ingest Mac")),
             "<li>The watchdog and the settings page: LaunchAgents <code>org.waterschurch.status-panel</code> and "
             "<code>org.waterschurch.status-settings</code>; files in <code>%s</code></li>" % e(data_dir or "~/Library/Application Support/status-panel"),
             "<li>Logs: <code>~/Library/Logs/status-panel.log</code>, <code>status-panel.err.log</code>, <code>status-settings.log</code>; "
             "auto-fix daily logs in <code>%s</code></li>" % e((cfg.get("auto_fix") or {}).get("log_dir", "~/Desktop/Auto-Fix Logs")),
             "<li>Working files: <code>config.json</code> (the settings), <code>state.json</code> (what survives a restart), "
             "<code>started.json</code>, <code>live-now.json</code>, <code>fixes-running.json</code>, <code>snapshot.json</code> (what the watchdog sees), "
             "<code>machines-live.json</code> (the last look at the computers), <code>dante-map.json</code>, <code>atem-reference.xml</code> (the saved switcher file), "
             "<code>prev/</code> (the previous code, for <code>install.sh --rollback</code>)</li>",
             "<li>Companion: <code>http://127.0.0.1:%d</code>, custom variables <code>%s</code>, <code>%s</code>, <code>%s</code>, blink clock <code>%s</code>; "
             "pages %d (Status)%s</li>" % (cfg["companion"].get("port", 8000), e(cfg["variables"]["state"]), e(cfg["variables"]["heartbeat"]),
                                         e(cfg["variables"]["command"]), e(cfg["variables"].get("flash", "flash_tick")), cfg["page"],
                                         (" and %d (Restart)" % cfg["restart_page"]["page"]) if cfg.get("restart_page") else ""),
             "<li>The login key for the other computers: <code>%s</code>; the Netgear login: keychain item <code>netgear-status-panel</code></li>" % e(cfg.get("ssh_key", "~/.ssh/id_status_panel")),
             "<li>This page: <code>%s</code></li></ul>" % e(address)]
    for alias, h in hosts.items():
        if h.get("self"):
            continue
        parts.append("<h2>%s</h2><ul><li>Log in: <code>%s</code>%s</li>" % (e(h.get("name", alias)), e(h.get("ssh", "")),
                     (" (wake-on-LAN %s)" % e(h["mac"])) if h.get("mac") else ""))
        for run in h.get("runs", []):
            k = run.get("kind")
            if k == "task":
                parts.append("<li>Scheduled task <code>\\StatusPanel\\%s</code> starts <code>%s</code></li>" % (e(run["name"]), e(run.get("exe", ""))))
            elif k == "service":
                parts.append("<li>Windows service <code>%s</code></li>" % e(run["name"]))
            elif k == "launchagent":
                parts.append("<li>LaunchAgent <code>%s</code> (%s)</li>" % (e(run.get("label", "")), e(run.get("name", ""))))
            elif k == "helper_dvs":
                parts.append("<li>DVS helper <code>%s</code> with a sudoers rule in <code>/etc/sudoers.d/status-panel</code></li>" % e(site.get("dvs_helper", "/usr/local/sbin/status-panel-dvs-start")))
            elif k == "helper_solo":
                parts.append("<li>Pro Tools helper in <code>%s</code> (LaunchAgent <code>%s</code>; needs Accessibility)</li>" % (e(site.get("solo_dir", "")), e(run.get("label", ""))))
            elif k == "app":
                parts.append("<li>%s (process <code>%s</code>)</li>" % (e(run.get("name", "")), e(run.get("process", ""))))
        parts.append("</ul>")
    return "\n".join(parts)