christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ready-page
status-indicators/schema.py
T
christianandClaude Opus 5.5 bf128e1bad READY page: drop the MICS and STAGE keys 
BOH MIX is the one tap key left.

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

815 lines
56 KiB
Python
Raw Permalink Blame History

"""The registry of every setting the status panel has.
One record per config key (path, type, default, plain-words help, which screen of the settings
page owns it, whether it sits under Advanced), the check types with their fields, the fix kinds,
the words a key can show, and the rules a config must pass. The settings page draws its forms
from this file; tests/test_schema.py keeps it in step with the code (every key the watchdog reads
must be declared here, every default in the code must equal the one here).
Paths: "companion.host", "tiles[].checks[].conn" (a list element), "fixes.*.run" (a map entry).
"""
import copy
import re
import status_panel as sp
# --- Field records -----------------------------------------------------------
# type: str | int | float | bool | list | dict | enum | path | ip | mac | url | hhmm | days | id
# apply: what changes when the value does: "restart" (the watchdog restarts), "pages" (the Companion
# pages are rebuilt and re-imported when the built page differs), "none" (page-only data), "settings".
F = {}
def field(path, type, default=None, help="", screen="", advanced=False, secret=False, apply="restart",
required=False, options=None, lo=None, hi=None, regex=None, example=None, shows=None):
F[path] = {"path": path, "type": type, "default": default, "help": help, "screen": screen,
"advanced": advanced, "secret": secret, "apply": apply, "required": required,
"options": options, "lo": lo, "hi": hi, "regex": regex, "example": example, "shows": shows}
# -- site --
field("site", "dict", {}, "This campus: its name, this Mac, and when changes must wait.", "Site")
field("site.id", "id", "na", "Short name used for folders and files.", "Site", advanced=True, regex=r"[a-z0-9_-]+")
field("site.name", "str", "", "The campus, as people say it.", "Site", example="North Attleboro", apply="none")
field("site.ingest_name", "str", "NA Ingest", "What this Mac is called in logs, notes and alerts.", "Site", apply="pages")
field("site.ingest_hostname", "str", "", "This Mac's network name, used in the page address and by the page tools.",
"Site", example="na-ingest.local", apply="none")
field("site.ingest_ips", "list", [], "This Mac's addresses, for the PCs' firewall allow-lists.", "Site", apply="none")
field("site.contacts", "str", "", "Printed on the cheat sheet: who to call.", "Site", apply="none")
field("site.alert_prefix", "str", "", "The first words of an alert message; blank = this Mac's name.", "Site", advanced=True)
field("site.live_services", "list", [], "Clock windows during which changes to the panel wait (a recording "
"HyperDeck always counts as live, whatever the clock says).", "Schedules")
field("site.live_services[].days", "days", None, "Days of the week.", "Schedules", required=True)
field("site.live_services[].start", "hhmm", None, "Start time.", "Schedules", required=True)
field("site.live_services[].end", "hhmm", None, "End time; earlier than the start means it runs past midnight.",
"Schedules", required=True)
field("site.editing_code", "str", "", "The code that switches Editing on. Kept here, in the repo, so everyone "
"with the repo has it.", "Site", advanced=True, secret=True, apply="none")
field("site.settings_port", "int", 8020, "Where the settings page listens.", "Site", advanced=True, lo=1024, hi=65535,
apply="settings")
field("site.dvs_helper", "path", "/usr/local/sbin/status-panel-dvs-start", "The DVS helper on each Mac that runs "
"Dante Virtual Soundcard.", "Site", advanced=True)
field("site.solo_dir", "path", "~/Library/Application Support/status-panel", "The Pro Tools helper's folder on "
"the Studio B Mac.", "Site", advanced=True)
# -- companion --
field("companion", "dict", {}, "Where Companion answers.", "Companion", required=True)
field("companion.host", "str", "127.0.0.1", "Companion's address from this Mac.", "Companion", advanced=True)
field("companion.port", "int", 8000, "Companion's HTTP port.", "Companion", advanced=True, lo=1, hi=65535)
field("companion.db", "path", None, "Companion's settings database; blank = the newest one found.", "Companion",
advanced=True)
field("companion.export_version", "int", sp.__dict__.get("EXPORT_VERSION", 12), "Companion's page export format "
"number (changes with a Companion upgrade).", "Companion", advanced=True, apply="pages")
field("companion.build", "str", "5.0.7+9763-stable-cec2f88e6f", "The Companion build the pages are written for.",
"Companion", advanced=True, apply="pages")
field("page", "int", 19, "The Status page's number in Companion.", "Keys", lo=1, apply="pages")
field("page_name", "str", "19\\nStatus", "The Status page's name in Companion.", "Keys", advanced=True, apply="pages")
field("nav_keys", "list", ["pagenum"], "Which of Companion's page keys (page up, page number, page down) are "
"coloured like the ALL key on every page.", "Companion", advanced=True, options=["pageup", "pagenum", "pagedown"],
apply="pages")
field("nav_hold_ms", "int", 1000, "Holding the page-number key this long jumps to the Status page.", "Companion",
advanced=True, lo=200, hi=5000, apply="pages")
field("nav_tap_page", "int", 1, "Tapping the page-number key goes to this page.", "Companion", advanced=True, lo=1,
apply="pages")
field("variables", "dict", {}, "The custom variables the panel uses inside Companion.", "Companion", required=True)
field("variables.state", "str", "sp_state", "The variable the keys are drawn from.", "Companion", advanced=True,
regex=r"[A-Za-z0-9_]+", apply="pages")
field("variables.heartbeat", "str", "sp_hb", "The variable that says the watchdog is alive.", "Companion",
advanced=True, regex=r"[A-Za-z0-9_]+", apply="pages")
field("variables.command", "str", "sp_cmd", "The variable key presses are written to.", "Companion", advanced=True,
regex=r"[A-Za-z0-9_]+", apply="pages")
field("variables.flash", "str", "flash_tick", "The blink clock. Renaming it means renaming it inside the trigger "
"in Companion too.", "Companion", advanced=True, regex=r"[A-Za-z0-9_]+", apply="pages")
field("page_grid", "dict", {"minColumn": 0, "maxColumn": 8, "minRow": 0, "maxRow": 5}, "The export grid "
"(the deck is the 8x4 inside it).", "Companion", advanced=True, apply="pages")
field("page_stale_seconds", "int", 15, "Keys turn purple when the watchdog has been silent this long.",
"Companion", advanced=True, lo=5, hi=120, apply="pages")
field("page_hold_ms", "int", 2000, "How long a key is held to run its fix.", "Schedules", advanced=True, lo=500,
hi=5000, apply="pages", shows="HOLD 2s")
# -- schedules and timing --
field("service_windows", "list", [{"days": list(sp.DAYS), "start": "00:00", "end": "23:59"}],
"When alarms are armed: always (the ALARMS key silences them until midnight).", "Schedules", advanced=True)
field("service_windows[].days", "days", None, "", "Schedules", required=True)
field("service_windows[].start", "hhmm", None, "", "Schedules", required=True)
field("service_windows[].end", "hhmm", None, "", "Schedules", required=True)
field("debounce_seconds", "int", 3, "A fault must last this long before it shows (one bad reading is not an "
"alarm).", "Schedules", advanced=True, lo=0, hi=60)
field("flap", "dict", {}, "Repeated faults show amber 'xN'.", "Schedules", advanced=True)
field("flap.count", "int", 3, "This many faults...", "Schedules", advanced=True, lo=1, hi=20)
field("flap.window_seconds", "int", 600, "...within this many seconds.", "Schedules", advanced=True, lo=60, hi=86400)
field("timing", "dict", {}, "Timing numbers.", "Schedules", advanced=True)
field("timing.fix_cooldown_seconds", "int", 60, "The same fix never re-runs within this.", "Schedules", advanced=True,
lo=10, hi=600, shows="WAIT 1 MIN")
field("timing.alert_after_seconds", "int", 30, "Send the alert message when Companion has not answered for this "
"long.", "Schedules", advanced=True, lo=5, hi=600)
field("alert_webhook", "url", "", "Where alert messages go (a Slack-style web hook); blank = off.", "Site",
secret=True)
field("auto_fix", "dict", {}, "When the panel may fix things by itself.", "Schedules")
field("auto_fix.windows", "list", [], "When auto-fix may act.", "Schedules")
field("auto_fix.windows[].days", "days", None, "", "Schedules", required=True)
field("auto_fix.windows[].start", "hhmm", None, "", "Schedules", required=True)
field("auto_fix.windows[].end", "hhmm", None, "", "Schedules", required=True)
field("auto_fix.while_muted", "list", [], "Auto-fixes that keep running inside the window even while ALARMS is "
"muted (harmless ones, like playing the Spotify playlist).", "Schedules")
field("auto_fix.after_seconds", "int", 20, "A fault must last this long before auto-fix acts.", "Schedules",
advanced=True, lo=5, hi=600)
field("auto_fix.max_tries", "int", 2, "Tries per fault, then it is yours to hold.", "Schedules", advanced=True,
lo=1, hi=10)
field("auto_fix.retry_seconds", "int", 90, "Between tries.", "Schedules", advanced=True, lo=30, hi=3600)
field("auto_fix.pause_after_network_seconds", "int", 60, "Wait this long after a network blip.", "Schedules",
advanced=True, lo=0, hi=3600)
field("auto_fix.forget_seconds", "int", 600, "Forget a fault's tries once it has been gone this long.",
"Schedules", advanced=True, lo=60, hi=86400)
field("auto_fix.restart_connections", "bool", False, "Also reconnect Companion connections by itself.", "Schedules")
field("auto_fix.fixes", "list", [], "The fixes auto-fix may run (gentle ones only).", "Schedules")
field("auto_fix.log_dir", "path", "~/Desktop/Auto-Fix Logs", "One file per day of everything auto-fix does.",
"Schedules", advanced=True)
field("auto_fix.keep_months", "int", 1, "Daily log files are deleted after this many months.", "Schedules",
advanced=True, lo=1, hi=12)
# -- network and Dante --
field("network", "dict", {}, "The NET key.", "Network & Dante")
field("network.interface", "str", "en0", "The network port to watch for link.", "Network & Dante")
field("network.gateway", "ip", "", "The router to ping; blank = off.", "Network & Dante")
field("network.burst_connections", "int", 4, "A blip is this many connections...", "Network & Dante", advanced=True,
lo=1, hi=50)
field("network.burst_groups", "int", 3, "...from this many groups...", "Network & Dante", advanced=True, lo=1, hi=20)
field("network.burst_window_seconds", "int", 10, "...dropping within this many seconds.", "Network & Dante",
advanced=True, lo=1, hi=300)
field("network.remember_seconds", "int", 300, "NET shows BLIP this long after.", "Network & Dante", advanced=True,
lo=10, hi=3600, apply="pages")
field("ssh_key", "path", "~/.ssh/id_status_panel", "The login key this Mac uses for the other computers.", "Site",
advanced=True)
field("dante_switches", "list", [], "The Netgear switches on the Dante network.", "Network & Dante")
field("dante_map_seconds", "int", 600, "Survey the switches this often.", "Network & Dante", advanced=True, lo=60,
hi=86400)
field("dante_max_shared", "int", 2, "Never power-cycle a port carrying more than this many devices (an uplink).",
"Network & Dante", advanced=True, lo=1, hi=10)
# -- Spotify and ATEM --
field("spotify", "dict", {}, "The service playlist.", "Spotify")
field("spotify.playlist", "str", "", "The playlist Spotify must be playing.", "Spotify", regex=r"spotify:playlist:[A-Za-z0-9]+",
required=True)
field("spotify.name", "str", "", "The playlist's name, as the keys and the key guide say it.", "Spotify")
field("spotify.open_wait", "int", 8, "Seconds to wait after opening Spotify.", "Spotify", advanced=True, lo=1, hi=60)
field("spotify.play_wait", "int", 3, "Seconds to wait before checking it plays.", "Spotify", advanced=True, lo=1, hi=60)
field("atem", "dict", {}, "The switcher and its saved file.", "ATEM")
field("atem.host", "ip", "", "The switcher's address.", "ATEM", required=True)
field("atem.node", "path", "/Applications/Companion [0-9]*.app/Contents/Resources/node-runtimes/node22/bin/node",
"Companion's node (pattern; the newest match is used).", "ATEM", advanced=True)
field("atem.library", "path", "~/Library/Application Support/companion/modules/bmd-atem-*/node_modules/atem-connection",
"Companion's ATEM library (pattern).", "ATEM", advanced=True)
field("atem.reference", "path", "atem-reference.xml", "The saved switcher file to keep in line with.", "ATEM")
field("atem.compare", "dict", None, "What to keep in line: audio fields and camera settings; blank = everything.",
"ATEM", advanced=True)
field("atem.source_folder", "path", "", "Where ATEM Software Control saves its files (for 'pick the newest').",
"ATEM", apply="none")
# -- hosts --
field("hosts", "dict", {}, "The computers this Mac logs in to.", "Computers")
field("hosts.*.ssh", "str", None, "How this Mac logs in: user@address.", "Computers", regex=r"[^@\s]+@[^@\s]+")
field("hosts.*.mac", "mac", None, "The PC's hardware address, to wake it when it is off.", "Computers")
field("hosts.*.wake_broadcast", "ip", None, "The broadcast address the wake-up goes to.", "Computers", advanced=True)
field("hosts.*.name", "str", None, "The computer, as people say it.", "Computers", apply="none")
field("hosts.*.os", "enum", "mac", "Mac or Windows (a PC's checks are written for cmd.exe).", "Computers",
options=["mac", "windows", "other"])
field("hosts.*.role", "str", None, "What it is for.", "Computers", apply="none")
field("hosts.*.notes", "str", None, "Notes for operators.", "Computers", apply="none")
field("hosts.*.pingable", "bool", True, "Off for a PC whose firewall blocks ping.", "Computers", apply="none")
field("hosts.*.self", "bool", False, "This Mac itself.", "Computers", apply="none")
field("hosts.*.runs", "list", [], "What runs on it (programs, tasks, services, helpers).", "Computers", apply="none")
field("hosts.*.networks", "list", [], "Its network ports and what each is for (LAN, Dante, switch management).",
"Computers", advanced=True, apply="none")
field("hosts.*.manual_done", "dict", {}, "Set-up steps ticked by hand (who, when).", "Computers", apply="none")
field("gear", "dict", {}, "Devices that are not computers: where each one is.", "Computers", apply="none")
field("gear.*.location", "str", None, "Where it is in the building.", "Computers", apply="none")
field("gear.*.notes", "str", None, "Notes for operators.", "Computers", apply="none")
field("gear.*.kind", "str", None, "What it is (recorder, switcher, wireless receiver...).", "Computers", apply="none")
field("gear.*.address", "str", None, "Its network address, if it has one.", "Computers", apply="none")
# -- fixes --
field("fixes", "dict", {}, "Everything a hold can run.", "Fixes")
field("fixes.*.kind", "enum", "ssh", "How it is built.", "Fixes", options=list(sp.FIX_KINDS))
field("fixes.*.host", "str", None, "The computer it runs on (ssh fixes).", "Fixes")
field("fixes.*.run", "str", None, "The command (zsh on Macs, cmd.exe on PCs).", "Fixes", advanced=True)
field("fixes.*.timeout", "int", 60, "Give up after this many seconds.", "Fixes", advanced=True, lo=5, hi=600)
field("fixes.*.short", "str", None, "The word the key shows while it runs (FIXING ...).", "Fixes", apply="pages")
field("fixes.*.label", "str", None, "Plain words: what it does.", "Fixes", required=True, apply="pages")
field("fixes.*.reconnect", "bool", False, "Reconnect the Companion connection afterwards.", "Fixes")
field("fixes.*.conn", "str", None, "The Companion connection it reconnects.", "Fixes")
field("fixes.*.restart_ok", "bool", False, "May close a running program (and still run by itself).", "Fixes")
field("fixes.*.never_auto", "bool", False, "Never runs by itself.", "Fixes")
field("fixes.*.part", "enum", None, "Which part of the switcher.", "Fixes", options=["audio", "camera"])
field("fixes.*.verify_seconds", "int", 20, "Wait this long before judging a reconnect.", "Fixes", advanced=True,
lo=5, hi=120)
field("fixes.*.template", "str", None, "How it is built (the builder); custom = typed by hand under Advanced.", "Fixes", apply="none")
field("fixes.*.params", "dict", None, "The builder's fields; the command is made from them.", "Fixes", apply="none")
field("tiles[].checks[].template", "str", None, "How the check's command is built.", "Keys", apply="none")
field("tiles[].checks[].params", "dict", None, "The builder's fields.", "Keys", apply="none")
field("fixes.*.off_seconds", "int", 2, "How long a HyperDeck's connection stays off.", "Fixes", advanced=True,
lo=1, hi=10)
field("fixes.*.hung_note", "str", None, "What the key says when the device answers ping but not Companion.",
"Fixes", advanced=True)
field("fixes.*.switch", "ip", None, "The PoE switch.", "Fixes")
field("fixes.*.port", "str", None, "The PoE port (e.g. 0/14).", "Fixes", regex=r"\d+/\d+")
field("fixes.*.mac", "mac", None, "The device expected on that port.", "Fixes")
field("fixes.*.playlist", "str", None, "A playlist other than the site's (rare).", "Fixes", advanced=True)
field("fixes.*.template", "str", None, "The builder this fix was made with.", "Fixes", advanced=True, apply="none")
field("fixes.*.params", "dict", None, "The builder's fields.", "Fixes", advanced=True, apply="none")
field("fixes.*.action", "enum", None, "What a READY page fix does.", "Fixes", options=list(sp.READY_ACTIONS))
# -- tiles (keys) --
field("tiles", "list", [], "The keys on the Status page.", "Keys", required=True)
field("tiles[].id", "id", None, "Used inside Companion.", "Keys", advanced=True, required=True, regex=r"[A-Za-z0-9_]+",
apply="pages")
field("tiles[].label", "str", None, "The first line on the key.", "Keys", required=True, apply="pages")
field("tiles[].pos", "list", None, "[row, column]; absent = no key of its own (shown inside a group).", "Keys",
apply="pages")
field("tiles[].kind", "enum", None, "A special key: master (ALL), service (ALARMS), network (NET), ready (READY: "
"opens the READY page and sums up its checklist).", "Keys", advanced=True,
options=["master", "service", "network", "ready"], apply="pages")
field("tiles[].members", "list", None, "A group: the keys it sums up.", "Keys")
field("tiles[].checks", "list", [], "What the key watches.", "Keys")
field("tiles[].hold_always", "str", None, "A fix a hold runs even when the key is green.", "Keys", apply="pages")
field("tiles[].checks[].type", "enum", None, "What kind of check.", "Keys", required=True)
field("tiles[].checks[].conn", "str", None, "The Companion connection.", "Keys")
field("tiles[].checks[].fix", "str", None, "What holding the key runs for this check (a fix, or a table per "
"sub-check).", "Keys")
field("tiles[].checks[].tag", "str", None, "Names the device when one key covers several.", "Keys")
field("tiles[].checks[].service_only", "bool", False, "Legacy: only during a service (has no effect while alarms "
"are armed all day).", "Keys", advanced=True)
field("tiles[].checks[].debounce_seconds", "int", None, "Wait before this check's fault shows.", "Keys", advanced=True,
lo=0, hi=60)
field("tiles[].checks[].ignore_warning", "bool", False, "Ignore Companion's yellow warning state.", "Keys")
field("tiles[].checks[].optional", "bool", False, "A HyperDeck that may have no disk (grey, no alarm).", "Keys")
field("tiles[].checks[].disk_service_only", "bool", True, "NO DISK only during a service.", "Keys", advanced=True)
field("tiles[].checks[].full_minutes", "int", 5, "FULL below this many minutes left.", "Keys", advanced=True)
field("tiles[].checks[].low_minutes", "int", 30, "LOW below this many minutes left.", "Keys", advanced=True)
field("tiles[].checks[].rec_sync", "bool", True, "Must record whenever another deck records.", "Keys")
field("tiles[].checks[].ip", "ip", None, "The deck's address, instead of what Companion reports.", "Keys", advanced=True)
field("tiles[].checks[].devices", "dict", None, "Dante devices: name -> key text.", "Keys")
field("tiles[].checks[].count_all", "bool", False, "Count every watched Dante device on this key.", "Keys")
field("tiles[].checks[].channels", "int", 1, "Channels on the receiver.", "Keys", lo=1, hi=8)
field("tiles[].checks[].battery_err_bars", "int", 1, "Red at this many bars or fewer.", "Keys", advanced=True, lo=0, hi=5)
field("tiles[].checks[].battery_warn_bars", "int", 2, "Amber at this many bars or fewer.", "Keys", advanced=True, lo=0, hi=5)
field("tiles[].checks[].muted_is_error", "bool", False, "RF mute is red, not amber.", "Keys")
field("tiles[].checks[].host", "str", None, "The computer (an address for ping, a computer name for the rest).", "Keys")
field("tiles[].checks[].name", "str", None, "The name the key shows.", "Keys")
field("tiles[].checks[].msg", "str", None, "The words the key shows when it fails.", "Keys")
field("tiles[].checks[].failures", "int", 2, "Pings missed before NO PING.", "Keys", advanced=True, lo=1, hi=10)
field("tiles[].checks[].label", "str", None, "The launchd helper.", "Keys")
field("tiles[].checks[].match", "str", None, "The program (pgrep -f pattern).", "Keys")
field("tiles[].checks[].url", "url", None, "The web address to fetch.", "Keys")
field("tiles[].checks[].interval_seconds", "int", 30, "Fetch it this often.", "Keys", advanced=True, lo=5, hi=3600)
field("tiles[].checks[].path", "path", None, "The disk to watch.", "Keys")
field("tiles[].checks[].warn_gb", "int", 20, "Amber below this many GB free.", "Keys", advanced=True)
field("tiles[].checks[].err_gb", "int", 5, "Red below this many GB free.", "Keys", advanced=True)
field("tiles[].checks[].session", "str", None, "The network MIDI session.", "Keys")
field("tiles[].checks[].peer", "str", None, "The peer that must be in it.", "Keys")
field("tiles[].checks[].address", "str", None, "Where to reconnect the peer if it was never seen.", "Keys", advanced=True)
field("tiles[].checks[].domain", "str", None, "The macOS preference domain.", "Keys")
field("tiles[].checks[].key", "str", None, "The preference key.", "Keys")
field("tiles[].checks[].want", "str", None, "The value it must have.", "Keys")
field("tiles[].checks[].playlist", "str", None, "A playlist other than the site's (rare).", "Keys", advanced=True)
field("tiles[].checks[].instance", "str", None, "The Bonjour MIDI name.", "Keys")
field("tiles[].checks[].source", "str", None, "The MIDI source, or the Art-Net sender's address.", "Keys")
field("tiles[].checks[].max_silence_seconds", "int", None, "Amber/red after this much silence.", "Keys", lo=1, hi=600)
field("tiles[].checks[].cmd", "str", None, "The command run on that computer.", "Keys", advanced=True)
field("tiles[].checks[].shell", "enum", None, "sh (Macs) or cmd (PCs); blank = from the computer's kind.", "Keys",
advanced=True, options=["sh", "cmd"])
field("tiles[].checks[].part", "enum", None, "Which part of the switcher.", "Keys", options=["audio", "camera"])
field("tiles[].checks[].template", "str", None, "The builder this check was made with.", "Keys", advanced=True, apply="none")
field("tiles[].checks[].item", "str", None, "Which line of the weekly checklist (READY page).", "Ready")
field("tiles[].checks[].params", "dict", None, "The builder's fields.", "Keys", advanced=True, apply="none")
# -- restart page --
field("restart_page", "dict", None, "The Restart page; absent = this campus has none.", "Keys", apply="pages")
field("restart_page.page", "int", 20, "Its number in Companion.", "Keys", lo=1, apply="pages")
field("restart_page.page_name", "str", "20\\nRestart", "Its name in Companion.", "Keys", advanced=True, apply="pages")
field("restart_page.jump_pos", "list", None, "Where the 'go to Restart page' key sits on the Status page.", "Keys",
apply="pages")
field("restart_page.back_pos", "list", None, "Where the 'back to Status' key sits on the Restart page.", "Keys",
apply="pages")
field("restart_page.keys", "list", [], "The restart keys.", "Keys", apply="pages")
field("restart_page.keys[].id", "id", None, "", "Keys", advanced=True, required=True, regex=r"[A-Za-z0-9_]+", apply="pages")
field("restart_page.keys[].label", "str", None, "", "Keys", required=True, apply="pages")
field("restart_page.keys[].pos", "list", None, "[row, column].", "Keys", required=True, apply="pages")
field("restart_page.keys[].fix", "str", None, "What holding it runs.", "Keys", required=True)
field("restart_page.keys[].watch", "list", [], "Keep flashing HOLD TO FIX while these fixes' checks or "
"'conn:<label>' connections are down.", "Keys")
# -- the READY page (the weekly checklist) --
field("ready_page", "dict", None, "The READY page: this weekend's checklist, checked against Planning Center; absent = "
"this campus has none.", "Ready", apply="pages")
field("ready_page.page", "int", 24, "Its number in Companion.", "Ready", lo=1, apply="pages")
field("ready_page.page_name", "str", "24\\nReady", "Its name in Companion.", "Ready", advanced=True, apply="pages")
field("ready_page.back_pos", "list", None, "Where the 'back to Status' key sits on the READY page.", "Ready",
apply="pages")
field("ready_page.keys", "list", [], "The checklist keys (each like a Status page key, placed on the READY page).",
"Ready", apply="pages")
field("ready", "dict", {}, "Where the READY page looks.", "Ready")
field("ready.interval_seconds", "int", 120, "Look this often (never while a HyperDeck is recording; a hold looks at "
"once).", "Ready", advanced=True, lo=30, hi=3600)
field("ready.due", "list", [], "From when gaps are red instead of amber.", "Ready")
field("ready.due[].days", "days", None, "", "Ready", required=True)
field("ready.due[].start", "hhmm", None, "", "Ready", required=True)
field("ready.due[].end", "hhmm", None, "", "Ready", required=True)
field("ready.reset_day", "enum", "mon", "Taps (BOH mix) count from midnight on this day.", "Ready",
options=list(sp.DAYS))
field("ready.pco_credentials", "path", "~/Library/Application Support/ProdDash/modules.json", "ProdDash's settings "
"file, which holds the Planning Center login.", "Ready", advanced=True)
field("ready.pco_module", "str", "pco-plan", "The ProdDash module whose Planning Center login is used.", "Ready",
advanced=True)
field("ready.pco_service_type", "str", None, "The Planning Center service type; blank = ProdDash's first.", "Ready")
field("ready.skip_songs", "list", [], "PCO songs that have no tracks or slides of their own.", "Ready")
field("ready.ltc_note", "str", "LTC", "The PCO note category that holds a song's timecode (checked against the "
"tracks, written from them by a hold); blank = not checked.",
"Ready", advanced=True)
field("ready.prop_url", "url", "", "ProPresenter's network API.", "Ready", example="http://10.1.14.225:1025")
field("ready.prop_lists", "list", [], "PCO items whose description lists things the slides must have: "
"{match, kind (count: one slide each / names: every name on a slide), short}.", "Ready", advanced=True)
field("ready.tracks_host", "str", None, "The computer Ableton runs on.", "Ready")
field("ready.projects", "path", "", "The shared Ableton Projects folder, as that computer sees it.", "Ready")
field("ready.site_folder", "str", "", "This campus's folder inside it.", "Ready", example="NA")
field("ready.folder_format", "str", "{date} {loc} Project", "A weekend's project folder ({date} = YYMMDD).",
"Ready", advanced=True)
field("ready.ltc_track", "str", "LTC", "The Ableton track that plays the timecode (an LTC file named like "
"LTC_00000000_60mins_30fps: it starts at 00:00:00:00, 30 frames a second).", "Ready", advanced=True)
field("ready.lighting_track", "str", "LIGHTING CUES", "The Ableton track that holds the lighting cues.", "Ready")
field("ready.locations", "list", [], "The campuses that get this campus's tracks.", "Ready")
field("ready.remote_timeout", "int", 90, "Seconds allowed for reading the Ableton set.", "Ready", advanced=True,
lo=10, hi=600)
field("ready.lineup_url", "url", "", "ProdDash's Band Lineup module.", "Ready")
field("ready.lineup_teams", "list", ["Band", "Communication"], "PCO teams whose people must be on the lineup TV.",
"Ready")
field("ready.lineup_skip_positions", "list", [], "PCO positions the lineup TV does not show.", "Ready")
field("ready.resolume_url", "url", "", "Resolume Arena's web server.", "Ready")
field("ready.resolume_deck", "str", "", "The deck Resolume must be on.", "Ready")
field("ready.led_clip", "str", "", "The cross background a hold plays; blank = the last one seen playing.", "Ready")
field("ready.prop_bundle_host", "str", None, "The computer that saves the ProPresenter bundle.", "Ready")
field("ready.prop_bundle_folder", "path", "", "Where the bundle for the locations goes.", "Ready")
field("ready.prop_bundle_pattern", "str", "*.probundle", "The bundle's file name pattern.", "Ready", advanced=True)
field("ready.vocal_positions", "list", ["Vocalist", "[Leader]"], "PCO positions that need a vocal mic.", "Ready")
field("ready.mic_positions", "dict", {}, "Other PCO positions and the mic they need (MC -> MC).", "Ready")
# --- Check types ---------------------------------------------------------------
# fields: the keys that type reads; subs: the sub-check names it returns (fix tables use them);
# identity: what it is about; module: the Companion module its connection is expected to use.
CHECKS = {
"conn": {"fields": ["conn", "service_only", "ignore_warning"], "subs": ["conn"], "identity": ["conn"], "module": None,
"question": "Is a device answering?", "words": "A Companion connection is good."},
"hyperdeck": {"fields": ["conn", "optional", "disk_service_only", "full_minutes", "low_minutes", "rec_sync", "ip"],
"subs": ["conn", "disk", "rec"], "identity": ["conn"], "module": "bmd-hyperdeck",
"question": "Is a recorder OK?", "words": "A HyperDeck is connected, has a disk, and records with the others."},
"dante": {"fields": ["conn", "devices", "count_all", "service_only"], "subs": ["conn", "<device name>"], "identity": ["conn"],
"module": "audinate-dantecontroller", "question": "Is Dante complete?", "words": "Every listed Dante device is present."},
"shure": {"fields": ["conn", "channels", "battery_err_bars", "battery_warn_bars"], "subs": ["conn", "ch<N>_batt", "ch<N>_rf"],
"identity": ["conn"], "module": "shure-wireless", "question": "Is a device answering?", "words": "A Shure receiver: batteries and interference."},
"psm": {"fields": ["conn", "channels", "muted_is_error"], "subs": ["conn", "ch<N>_mute"], "identity": ["conn"],
"module": "shure-psm1000", "question": "Is a device answering?", "words": "A PSM transmitter: RF mute."},
"tascam": {"fields": ["conn"], "subs": ["conn", "error", "caution"], "identity": ["conn"], "module": "tascam-da-6400",
"question": "Is a recorder OK?", "words": "The Tascam recorder reports no error or caution."},
"ping": {"fields": ["host", "conn", "name", "msg", "failures", "service_only"], "subs": ["ping"], "identity": ["conn", "host"],
"module": None, "question": "Is a device answering?", "words": "An address answers ping."},
"launchd": {"fields": ["label", "name"], "subs": ["svc"], "identity": ["label"], "module": None,
"question": "Is a program running?", "words": "A helper (LaunchAgent) on this Mac is loaded and running."},
"process": {"fields": ["match", "name", "template", "params"], "subs": ["proc"], "identity": ["match"], "module": None,
"question": "Is a program running?", "words": "A program on this Mac is running."},
"http": {"fields": ["url", "msg", "name", "interval_seconds"], "subs": ["http"], "identity": ["url"], "module": None,
"question": "Is a device answering?", "words": "A web address answers."},
"disk": {"fields": ["path", "warn_gb", "err_gb"], "subs": ["disk"], "identity": ["path"], "module": None,
"question": "Is a setting right?", "words": "Free space on a disk."},
"midi_peer": {"fields": ["session", "peer", "host", "address"], "subs": ["peer"], "identity": ["host", "session", "peer"],
"module": None, "question": "Is a MIDI link up?", "words": "A peer is in a network MIDI session (here or on another Mac)."},
"pref": {"fields": ["domain", "key", "want", "msg", "template", "params"], "subs": ["pref"], "identity": ["domain", "key"],
"module": None, "question": "Is a setting right?", "words": "A macOS preference has the value it should."},
"spotify": {"fields": ["playlist", "fix"], "subs": ["spotify"], "identity": [], "module": None,
"question": "Is Spotify playing?", "words": "Spotify is playing the service playlist."},
"mdns": {"fields": ["instance", "name"], "subs": ["mdns"], "identity": ["instance"], "module": None,
"question": "Is a MIDI link up?", "words": "A network MIDI session is advertised on Bonjour."},
"midi_rx": {"fields": ["source", "name", "max_silence_seconds"], "subs": ["rx"], "identity": ["source"], "module": None,
"question": "Is a MIDI link up?", "words": "MIDI is arriving from a source."},
"remote": {"fields": ["host", "cmd", "msg", "shell", "template", "params"], "subs": ["remote"], "identity": ["host", "cmd"],
"module": None, "question": "Is a program running?", "words": "A command on another computer succeeds."},
"artnet": {"fields": ["source", "max_silence_seconds", "msg"], "subs": ["artnet"], "identity": ["source"], "module": None,
"question": "Is the lighting console sending?", "words": "Art-Net is arriving from the lighting program."},
"atem": {"fields": ["part", "msg"], "subs": ["audio", "camera"], "identity": ["part"], "module": "bmd-atem",
"question": "Is a setting right?", "words": "The switcher matches the saved file."},
"ready": {"fields": ["item"], "subs": ["ready", "<what is missing>"], "identity": ["item"], "module": None,
"question": "Is the weekend ready?", "words": "A line of the weekly checklist, checked against Planning Center."},
"ready_manual": {"fields": ["item", "msg"], "subs": ["done"], "identity": ["item"], "module": None,
"question": "Is the weekend ready?", "words": "A line of the weekly checklist a person taps done."},
}
COMMON_CHECK_FIELDS = ["type", "fix", "tag", "service_only", "debounce_seconds"]
FIX_KINDS = sp.FIX_KINDS
FIX_KIND_FIELDS = {
"ssh": ["host", "run", "timeout", "reconnect", "conn"], "local": ["run", "timeout", "reconnect", "conn"],
"atem": ["part"], "reconnect": ["conn", "verify_seconds", "hung_note"], "midi_connect": [],
"netgear_poe": ["switch", "port", "mac"], "hyperdeck": ["verify_seconds", "off_seconds", "hung_note"],
"spotify": ["playlist"], "ready": ["action"],
}
EXPECTED_MODULE = {t: spec["module"] for t, spec in CHECKS.items() if spec["module"]}
# --- Every word a key can show, grouped by what to do about it -------------------
GLOSSARY = {
"hold the key: the fix runs": [
"NOT PLAYING", "WRONG PLAYLIST", "SPOTIFY CLOSED", "STOPPED", "NOT LOADED", "MIDI LOST", "MIDI", "GONE",
"DISABLED", "OFFLINE", "NO SESSION", "NOT RUNNING", "DOWN", "NOT REC", "SILENT", "WARNING", "STARTING",
"ATEM AUDIO WRONG", "ATEM CAMERA WRONG", "ATEM SW MIDI ON", "PT SOLO NOT AFL", "PROPRES NOT RUNNING",
"ABLETON NOT RUNNING", "TRANSLATE NOT RUNNING", "PROP TV DOWN", "BIBLE APP DOWN", "PRODCOM DOWN", "PRODDASH DOWN",
"ENCODER WEB DOWN", "PT SOLO CHECK OFF", "UNREACH", "DROPPED",
],
"check the device or the cable": [
"NO PING", "NO ANSWER", "UNREACHABLE", "LINK DOWN", "NO GATEWAY", "NO DISK", "FULL", "LOW", "BATT", "INTERF",
"RF MUTE", "HUNG - POWER CYCLE IT", "NO NETWORK CHECK POWER", "STILL DOWN", "REFUSED", "LOGIN", "BLIP",
"NOT REACHABLE", "DISK",
],
"the READY page: what is not ready yet (hold the key to fix it or look again)": [
"TO DO", "ALL DONE", "TAP WHEN DONE", "TAP WHEN SET", "NOT DONE", "DONE", "NOT SET UP", "CHECKING",
"NO TRACKS", "EMPTY", "NOT MIXED", "TO MASTER", "NO LIGHTS", "NO CUE TRACK", "NO SET", "LIVE HAS OLD SET",
"MISSING", "SLIDES OLD?", "SLIDES", "NO TC", "NO LTC", "NO BG", "NO PLAYLIST", "OLD PLAN", "SYNC ERROR",
"NOT ON PLAN", "DECLINED", "WRONG DECK", "NOTHING ON", "NOT SENT", "COPYING TO", "NO PCO LOGIN",
"PCO UNREACHABLE", "NO PLAN IN PCO", "PROP UNREACHABLE", "LINEUP UNREACHABLE", "RESOLUME UNREACHABLE",
"UNREACHABLE", "CHECK FAILED", "NO CLIP KNOWN", "LTC WRONG", "PCO WRITE FAILED", "IN PCO", "NO SET TO COPY", "SYNC SLOW", "NO PLAN YET", "VOX",
],
"ask Chris": ["NOT IN COMPANION", "CHECK BUG", "NO DATA", "WATCHDOG DOWN", "ERR", "PROCESS", "WEB"],
"what a key says while you hold it": [
"NOTHING TO FIX", "NO AUTO FIX", "JUST TRIED", "WAIT 1 MIN", "WAIT", "MIN", "S", "CHECKING", "FIXING",
"AUTO-FIXING", "WAKING PC", "FIX FAILED", "RESTARTING", "RESTARTING SOON", "SOON", "DONE", "FAILED",
"JUST DONE", "STARTING", "RECORDING LEFT ALONE", "NOTHING TO CLEAR", "CLEARED", "STILL CONNECTING",
"STILL OFF IN COMPANION", "DISABLED IN COMPANION", "TURN ON", "RECONNECT", "POWER CYCLE", "FIXES", "HOLD TO FIX",
"HOLD", "REC", "PLAYING", "OK", "OFF", "ON", "ALL OK", "WARN", "TRS", "GB", "NOT", "REACHABLE", "RECORDING",
"LEFT ALONE", "TO FIX", "TO CLEAR", "NO AUTO", "FIX", "AUTO", "IN", "COMPANION", "POWER", "CYCLE",
"STILL", "OFF IN", "DISABLED IN", "CONNECTING", "CHECK", "NETWORK", "NO",
],
}
# --- Rules ---------------------------------------------------------------------
def defaults():
"""{path: default} for every field with one."""
return {p: copy.deepcopy(f["default"]) for p, f in F.items() if f["default"] is not None}
def key_record():
"""{receiver kind: {keys}}: which keys the code may read from each kind of config object."""
rec = {}
for p in F:
parts = p.split(".")
kind = "top"
if p.startswith("tiles[].checks[]."):
kind, parts = "check", parts[2:]
elif p.startswith("tiles[]."):
kind, parts = "tile", parts[1:]
elif p.startswith("fixes.*."):
kind, parts = "fix", parts[2:]
elif p.startswith("hosts.*."):
kind, parts = "host", parts[2:]
elif p.startswith("restart_page.keys[]."):
kind, parts = "restart_key", parts[2:]
elif p.startswith("restart_page."):
kind, parts = "restart_page", parts[1:]
elif p.startswith("ready_page."):
kind, parts = "ready_page", parts[1:]
elif p.startswith("ready."):
kind, parts = "ready", parts[1:]
elif p.startswith("auto_fix."):
kind, parts = "auto", parts[1:]
elif p.startswith("network."):
kind, parts = "network", parts[1:]
elif p.startswith("variables."):
kind, parts = "variables", parts[1:]
elif p.startswith("site."):
kind, parts = "site", parts[1:]
elif p.startswith("atem."):
kind, parts = "atem", parts[1:]
elif p.startswith("spotify."):
kind, parts = "spotify", parts[1:]
elif p.startswith("companion."):
kind, parts = "companion", parts[1:]
elif p.startswith("timing."):
kind, parts = "timing", parts[1:]
elif p.startswith("flap."):
kind, parts = "flap", parts[1:]
elif p.startswith("gear.*."):
kind, parts = "gear", parts[2:]
if "[]" in parts[0] and len(parts) > 1:
kind, parts = "window", parts[1:]
rec.setdefault(kind, set()).add(parts[0].replace("[]", ""))
return rec
def _windows(ws, where, errors):
try:
sp.check_windows(ws, where)
except ValueError as e:
errors.append(str(e))
def validate(cfg, live=None):
"""(errors, warnings). Errors stop an Apply; warnings are shown. `live` (optional) is what the
settings page knows right now: {"connections": {label: module id}, "hosts_up": {alias: bool}}."""
errors, warnings = [], []
live = live or {}
def pos_ok(pos, where):
if not (isinstance(pos, list) and len(pos) == 2 and all(isinstance(n, int) for n in pos)):
errors.append("%s: position must be [row, column]" % where)
return None
grid = cfg.get("page_grid", F["page_grid"]["default"])
if not (grid["minRow"] <= pos[0] <= grid["maxRow"] and grid["minColumn"] <= pos[1] <= grid["maxColumn"]):
errors.append("%s: row %d column %d is off the grid" % (where, pos[0], pos[1]))
if pos[1] == 0 and pos[0] in (0, 1, 2):
errors.append("%s: row %d column 0 is a page key" % (where, pos[0]))
return tuple(pos)
for req in ("companion", "variables", "tiles"):
if req not in cfg:
errors.append("missing %s" % req)
if errors:
return errors, warnings
ids, seen_pos = set(), {}
kinds = {"master": 0, "service": 0, "network": 0}
fixes, hosts = cfg.get("fixes", {}), cfg.get("hosts", {})
for t in cfg["tiles"]:
tid = t.get("id", "")
where = "key %s" % (tid or "?")
if not re.fullmatch(r"[A-Za-z0-9_]+", str(tid)):
errors.append("%s: the id may only hold letters, digits and _" % where)
if tid in ids:
errors.append("%s: two keys with the same id" % where)
ids.add(tid)
if not t.get("label"):
errors.append("%s: no label" % where)
if t.get("kind") in kinds:
kinds[t["kind"]] += 1
if "pos" in t:
pos = pos_ok(t["pos"], where)
if pos in seen_pos:
errors.append("%s and %s share row %d column %d" % (seen_pos[pos], where, pos[0], pos[1]))
seen_pos[pos] = where
if t.get("hold_always") and t["hold_always"] not in fixes:
errors.append("%s: hold runs unknown fix %s" % (where, t["hold_always"]))
for i, c in enumerate(t.get("checks", [])):
cw = "%s check %d" % (where, i + 1)
if c.get("type") not in CHECKS:
errors.append("%s: unknown check type %r" % (cw, c.get("type")))
continue
spec = CHECKS[c["type"]]
for k in c:
if k not in spec["fields"] and k not in COMMON_CHECK_FIELDS:
warnings.append("%s: %s is not a field of a %s check" % (cw, k, c["type"]))
if c["type"] in ("conn", "hyperdeck", "dante", "shure", "psm", "tascam") and not c.get("conn"):
errors.append("%s: a %s check needs a Companion connection" % (cw, c["type"]))
if c["type"] == "ping" and not (c.get("host") or c.get("conn")):
errors.append("%s: a ping needs an address or a connection" % cw)
if c["type"] == "midi_peer" and not (c.get("session") and c.get("peer")):
errors.append("%s: a MIDI link needs a session and a peer" % cw)
if c["type"] == "remote" and not (c.get("host") and c.get("cmd") and c.get("msg")):
errors.append("%s: a remote check needs a computer, a command and the words to show" % cw)
fx = c.get("fix")
for f in (fx.values() if isinstance(fx, dict) else [fx]):
if f and f not in fixes and f not in (sp.RESTART_CONNECTION, sp.DANTE_AUTO):
errors.append("%s: unknown fix %s" % (cw, f))
if c.get("service_only"):
warnings.append("%s: 'only during a service' has no effect while alarms are armed all day" % cw)
conn, mod = c.get("conn"), spec["module"]
if conn and mod and live.get("connections") and conn in live["connections"] \
and live["connections"][conn] and not live["connections"][conn].startswith(mod):
warnings.append("%s: %s uses module %s; this check expects %s" % (cw, conn, live["connections"][conn], mod))
if conn and live.get("connections") and conn not in live["connections"]:
warnings.append("%s: Companion has no connection called %s" % (cw, conn))
for t in cfg["tiles"]:
for m in t.get("members", []):
if m not in ids:
errors.append("key %s: unknown member %s" % (t.get("id"), m))
if m == t.get("id"):
errors.append("key %s: is its own member" % t.get("id"))
in_group = {m for t in cfg["tiles"] for m in t.get("members", [])}
for t in cfg["tiles"]:
if "pos" not in t and "members" not in t and t.get("id") not in in_group and t.get("kind") != "master":
warnings.append("key %s has no place on the deck and is in no group: nobody will see it" % t.get("id"))
if kinds["master"] != 1:
errors.append("there must be exactly one ALL key (kind master); found %d" % kinds["master"])
if kinds["service"] != 1:
errors.append("there must be exactly one ALARMS key (kind service); found %d" % kinds["service"])
if kinds["network"] > 1:
errors.append("at most one NET key (kind network); found %d" % kinds["network"])
for fid, fix in fixes.items():
where = "fix %s" % fid
kind = fix.get("kind", "ssh")
if kind not in FIX_KINDS:
errors.append("%s: unknown kind %s" % (where, kind))
if not fix.get("label"):
errors.append("%s: needs plain words (label)" % where)
if kind == "ssh":
h = hosts.get(fix.get("host"))
if h is None:
errors.append("%s: unknown computer %s" % (where, fix.get("host")))
elif h.get("self"):
errors.append("%s: %s is this Mac - use the local kind" % (where, fix.get("host")))
elif not h.get("ssh"):
errors.append("%s: computer %s has no log-in address" % (where, fix.get("host")))
if kind in ("ssh", "local") and not fix.get("run"):
errors.append("%s: nothing to run" % where)
if kind == "netgear_poe" and not (fix.get("switch") and fix.get("port")):
errors.append("%s: needs a switch and a port" % where)
if kind == "reconnect" and not fix.get("conn"):
errors.append("%s: needs the connection to reconnect" % where)
if kind == "atem" and fix.get("part") not in ("audio", "camera"):
errors.append("%s: part must be audio or camera" % where)
for k in fix:
if k not in FIX_KIND_FIELDS.get(kind, []) and k not in ("kind", "short", "label", "restart_ok", "never_auto",
"template", "params", "timeout"):
warnings.append("%s: %s is not a field of a %s fix" % (where, k, kind))
for alias, h in hosts.items():
if not re.fullmatch(r"[a-z0-9_]+", alias):
errors.append("computer %s: the short name may only hold lowercase letters, digits and _" % alias)
if not h.get("self") and not re.fullmatch(r"[^@\s]+@[^@\s]+", str(h.get("ssh", ""))):
errors.append("computer %s: the log-in address must be user@address" % alias)
if h.get("mac") and not re.fullmatch(r"(?:[0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}", h["mac"]):
errors.append("computer %s: not a hardware (MAC) address: %s" % (alias, h["mac"]))
for t in cfg["tiles"]:
for i, c in enumerate(t.get("checks", [])):
if c.get("type") == "remote" or (c.get("type") == "midi_peer" and c.get("host")):
h = hosts.get(c["host"])
if h is None:
errors.append("key %s check %d: unknown computer %s" % (t.get("id"), i + 1, c["host"]))
elif h.get("self"):
errors.append("key %s check %d: %s is this Mac" % (t.get("id"), i + 1, c["host"]))
a = cfg.get("auto_fix") or {}
for f in a.get("fixes", []):
if f not in fixes:
errors.append("auto-fix: unknown fix %s" % f)
elif fixes[f].get("never_auto"):
errors.append("auto-fix: %s is marked never to run by itself" % f)
elif re.search(r"pkill|taskkill|killall", fixes[f].get("run", "")) and not fixes[f].get("restart_ok"):
errors.append("auto-fix: %s closes a running program; tick 'may close a running program' or take it out" % f)
for f in a.get("while_muted", []):
if f not in a.get("fixes", []):
errors.append("auto-fix: %s runs while ALARMS is muted but is not an auto-fix" % f)
_windows(cfg.get("service_windows"), "alarms window", errors)
_windows(a.get("windows"), "auto-fix window", errors)
_windows((cfg.get("site") or {}).get("live_services"), "live services", errors)
rp = cfg.get("restart_page")
if rp:
rpos = {}
for k in rp.get("keys", []):
kid = k.get("id", "")
where = "restart key %s" % (kid or "?")
if kid in ids:
errors.append("%s: a key on the Status page has that id" % where)
if k.get("fix") not in fixes:
errors.append("%s: unknown fix %s" % (where, k.get("fix")))
for w in k.get("watch", []):
if not w.startswith("conn:") and w not in fixes:
errors.append("%s: unknown fix %s in its watch list" % (where, w))
pos = pos_ok(k.get("pos"), where)
if pos in rpos:
errors.append("%s and %s share row %d column %d" % (rpos[pos], where, pos[0], pos[1]))
rpos[pos] = where
for name in ("jump_pos", "back_pos"):
if name in rp:
pos_ok(rp[name], "the Restart page's %s" % name.replace("_pos", " key"))
if "jump_pos" in rp and tuple(rp["jump_pos"]) in seen_pos:
errors.append("the 'go to Restart page' key sits on %s" % seen_pos[tuple(rp["jump_pos"])])
if "back_pos" in rp and tuple(rp["back_pos"]) in rpos:
errors.append("the 'back to Status' key sits on %s" % rpos[tuple(rp["back_pos"])])
if rp.get("page") == cfg.get("page"):
errors.append("the Status page and the Restart page cannot be the same page")
rd = cfg.get("ready_page")
readies = [t for t in cfg["tiles"] if t.get("kind") == "ready"]
if len(readies) > 1:
errors.append("at most one READY key on the Status page; found %d" % len(readies))
if rd:
dpos = {}
for k in rd.get("keys", []):
kid = k.get("id", "")
where = "READY key %s" % (kid or "?")
if not re.fullmatch(r"[A-Za-z0-9_]+", str(kid)):
errors.append("%s: the id may only hold letters, digits and _" % where)
if kid in ids:
errors.append("%s: another key has that id" % where)
ids.add(kid)
if not k.get("label"):
errors.append("%s: no label" % where)
pos = pos_ok(k.get("pos"), where)
if pos in dpos:
errors.append("%s and %s share row %d column %d" % (dpos[pos], where, pos[0], pos[1]))
dpos[pos] = where
for i, c in enumerate(k.get("checks", [])):
cw = "%s check %d" % (where, i + 1)
if c.get("type") not in CHECKS:
errors.append("%s: unknown check type %r" % (cw, c.get("type")))
continue
if c["type"] in ("ready", "ready_manual") and not c.get("item"):
errors.append("%s: which checklist line (item)?" % cw)
if c["type"] == "ready" and c.get("item") and c["item"] not in sp.READY_ITEMS:
errors.append("%s: unknown checklist line %s" % (cw, c["item"]))
fx = c.get("fix")
for f in (fx.values() if isinstance(fx, dict) else [fx]):
if f and f not in fixes and f not in (sp.RESTART_CONNECTION, sp.DANTE_AUTO):
errors.append("%s: unknown fix %s" % (cw, f))
if "back_pos" in rd:
bp = pos_ok(rd["back_pos"], "the READY page's back key")
if bp in dpos:
errors.append("the 'back to Status' key sits on %s" % dpos[bp])
if rd.get("page") in (cfg.get("page"), (rp or {}).get("page")):
errors.append("the READY page cannot be the Status or Restart page")
if not readies:
warnings.append("the READY page has no READY key on the Status page: nobody can open it from there")
_windows((cfg.get("ready") or {}).get("due"), "READY due window", errors)
for fid, fix in fixes.items():
if fix.get("kind") == "ready" and fix.get("action") not in sp.READY_ACTIONS:
errors.append("fix %s: unknown READY action %s" % (fid, fix.get("action")))
spo = cfg.get("spotify") or {}
if any(c.get("type") == "spotify" for t in cfg["tiles"] for c in t.get("checks", [])):
if not re.fullmatch(r"spotify:playlist:[A-Za-z0-9]+", str(spo.get("playlist", ""))):
errors.append("Spotify: the playlist must look like spotify:playlist:...")
for p, f in F.items():
v = _get(cfg, p)
if v is None:
continue
if f["type"] == "int" and not isinstance(v, int):
errors.append("%s: must be a whole number" % p)
elif f["type"] == "int":
if f["lo"] is not None and v < f["lo"]:
errors.append("%s: %s is below %s" % (p, v, f["lo"]))
if f["hi"] is not None and v > f["hi"]:
errors.append("%s: %s is above %s" % (p, v, f["hi"]))
if f["regex"] and isinstance(v, str) and not re.fullmatch(f["regex"], v):
errors.append("%s: %r is not in the right form" % (p, v))
if f["type"] == "enum" and f["options"] and v not in f["options"]:
errors.append("%s: %r is not one of %s" % (p, v, ", ".join(f["options"])))
try:
from settings import builders
errors += builders.mismatches(cfg)
except ImportError:
pass
return errors, warnings
def _get(cfg, path):
"""The value at a plain path (no [] or *); None when absent."""
if "[]" in path or ".*." in path:
return None
cur = cfg
for part in path.split("."):
if not isinstance(cur, dict) or part not in cur:
return None
cur = cur[part]
return cur
def mask_secrets(obj):
"""A copy with every secret replaced by "set (N characters)" or "" (for API responses and history)."""
secret_paths = {p for p, f in F.items() if f["secret"]}
def walk(node, path):
if isinstance(node, dict):
return {k: ("set (%d characters)" % len(v) if v else "") if "%s.%s" % (path, k) in secret_paths
or k in secret_paths else walk(v, "%s.%s" % (path, k) if path else k) for k, v in node.items()}
if isinstance(node, list):
return [walk(v, path + "[]") for v in node]
return node
return walk(copy.deepcopy(obj), "")
def unmask(draft, live):
"""A draft that came back from the page carries masked secrets ("set (N characters)"): put the
live values back so a masked value never overwrites a real one."""
secret_paths = {p for p, f in F.items() if f["secret"]}
def walk(node, ref, path):
if isinstance(node, dict):
out = {}
for k, v in node.items():
p = "%s.%s" % (path, k) if path else k
if (p in secret_paths or k in secret_paths) and isinstance(v, str) and (v.startswith("set (") or v == "") \
and isinstance(ref, dict) and isinstance(ref.get(k), str) and ref.get(k):
out[k] = ref[k] if v.startswith("set (") else ""
else:
out[k] = walk(v, ref.get(k) if isinstance(ref, dict) else None, p)
return out
if isinstance(node, list):
return [walk(v, ref[i] if isinstance(ref, list) and i < len(ref) else None, path + "[]") for i, v in enumerate(node)]
return node
return walk(copy.deepcopy(draft), live, "")
def phrases(cfg):
"""The sentences the page and the logs build from the config."""
a = cfg.get("auto_fix") or {}
site = cfg.get("site") or {}
return {
"auto_fix": "may run by itself %s" % sp.windows_phrase(a.get("windows", [])) if a.get("windows") else "never runs by itself",
"live": ("changes wait while a HyperDeck is recording" + (" or %s" % sp.windows_phrase(site["live_services"])
if site.get("live_services") else "")),
"alarms": "Alarms are on every day. The ALARMS key silences them until midnight, then they come back by themselves.",
"hold": "HOLD %gs" % (cfg.get("page_hold_ms", 2000) / 1000.0),
"ingest": site.get("ingest_name", "NA Ingest"),
}
Reference in New Issue View Git Blame Copy Permalink