#!/usr/bin/env python3
"""Status panel watchdog for the Companion STATUS page on NA Ingest.
Every second it reads every connection's status and a handful of module
variables from Companion's HTTP API, adds what Companion cannot see (network
MIDI peers, Bonjour sessions, pings, launchd services, disk, Resi cloud),
decides what is wrong, and writes one JSON custom variable (sp_state) that the
STATUS page buttons render. It also writes a heartbeat (sp_hb) so the page can
tell when this script has stopped, and reads button presses from sp_cmd.
Stdlib only; runs on the macOS /usr/bin/python3 (3.9).
status_panel.py run (LaunchAgent)
status_panel.py --once poll everything once and print the tiles
status_panel.py --dry-run run, but never write to Companion
"""
import argparse
import calendar
import hashlib
import http.client
import json
import logging
import math
import logging.handlers
import os
import re
import shutil
import signal
import socket
import ssl
import subprocess
import sys
import threading
import time
import urllib.parse
import urllib.request
from collections import deque
from datetime import datetime
HERE = os.path.dirname(os.path.abspath(__file__))
log = logging.getLogger("status-panel")
VERSION = "2026.10.15" # logged at start ("status panel running (code ...)") and written to started.json
# Every kind of fix the watchdog can run (validated by load_config).
FIX_KINDS = ("ssh", "local", "atem", "reconnect", "midi_connect", "netgear_poe", "hyperdeck", "spotify", "ready",
"companion_press")
# Failover: a fix marked "failover" (pressing the FOH Backup button when Studio B dies) runs by itself
# inside these windows when the "failover" block names none, "after_seconds" after its fault is confirmed.
FAILOVER_WINDOWS = [{"days": ["sun"], "start": "06:30", "end": "13:30"}]
REARM_SECONDS = 10 # a failover is armed again once its fault has been gone this long (not on one good reading)
PING_INTERVAL = 3 # seconds between pings; a ping check may ask for less ("interval_seconds")
REMOTE_INTERVAL = 10 # seconds between a remote check's ssh runs; a check may ask for less
OK, WARN, ERR, OFF = "ok", "warn", "err", "off"
# Display levels, least to most urgent. "idle" is a service-only fault while
# service mode is off (OBS PC off midweek); "quiet" is any other fault then.
ORDER = ["off", "ok", "idle", "warn", "notice", "quiet", "ack", "err"]
RANK = {name: i for i, name in enumerate(ORDER)}
# (background, text) for the flash-on phase and the flash-off phase.
COLORS = {
"ok": ((0x007A2E, 0xFFFFFF), (0x007A2E, 0xFFFFFF)),
"warn": ((0xFFA000, 0x000000), (0xFFA000, 0x000000)),
"notice": ((0xFFA000, 0x000000), (0x000000, 0xFFA000)), # flashing amber: something happened (NET BLIP)
"quiet": ((0x3A0000, 0xFF8080), (0x3A0000, 0xFF8080)),
"ack": ((0x8B0000, 0xFFFFFF), (0x8B0000, 0xFFFFFF)),
"err": ((0xFF0000, 0xFFFFFF), (0x000000, 0xFF3030)),
"off": ((0x262626, 0x8C8C8C), (0x262626, 0x8C8C8C)),
"idle": ((0x262626, 0xD0D0D0), (0x262626, 0xD0D0D0)),
"service": ((0x0B57D0, 0xFFFFFF), (0x0B57D0, 0xFFFFFF)),
"fixing": ((0x2962FF, 0xFFFFFF), (0x2962FF, 0xFFFFFF)), # a hold is being acted on
"failed": ((0xFF6D00, 0x000000), (0xFF6D00, 0x000000)), # the fix command itself failed
}
class Result:
"""One check's verdict. `named` means msg already says which device it is."""
__slots__ = ("level", "msg", "info", "service_only", "named", "flash")
def __init__(self, level, msg="", info="", service_only=False, named=False, flash=False):
self.level = level
self.msg = msg
self.info = info
self.service_only = service_only
self.named = named
self.flash = flash # a warning that flashes until acknowledged (NET BLIP)
def __repr__(self):
return "Result(%s, %r, info=%r)" % (self.level, self.msg, self.info)
# --- Companion HTTP API ----------------------------------------------------
class Companion:
def __init__(self, host, port, timeout=3.0):
self.host, self.port, self.timeout = host, port, timeout
self._conn = None
def _request(self, method, path, body=None, headers=None):
for attempt in (1, 2):
try:
if self._conn is None:
self._conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout)
self._conn.request(method, path, body=body, headers=headers or {})
resp = self._conn.getresponse()
return resp.status, resp.read().decode("utf-8", "replace")
except (OSError, http.client.HTTPException):
if self._conn is not None:
self._conn.close()
self._conn = None
if attempt == 2:
raise
def connections(self):
status, text = self._request("GET", "/api/connections")
if status != 200:
raise OSError("GET /api/connections -> %d" % status)
return json.loads(text)
def variable(self, label, name):
path = "/api/variable/%s/%s/value" % (urllib.parse.quote(label, safe=""), urllib.parse.quote(name, safe=""))
status, text = self._request("GET", path)
return text if status == 200 else None
def get_custom(self, name):
status, text = self._request("GET", "/api/custom-variable/%s/value" % urllib.parse.quote(name, safe=""))
return text if status == 200 else None
def set_custom(self, name, value):
path = "/api/custom-variable/%s/value" % urllib.parse.quote(name, safe="")
if value == "":
# Companion answers 400 to an empty body; an empty ?value= clears it.
status, text = self._request("POST", path + "?value=")
else:
# In the body, so sp_state (several KB) never runs into a URL length limit.
status, text = self._request("POST", path, body=value.encode("utf-8"),
headers={"Content-Type": "text/plain; charset=utf-8"})
return status == 200, status
def restart_connection(self, conn_id):
status, _ = self._request("POST", "/api/connections/%s/restart" % urllib.parse.quote(conn_id, safe=""))
return status == 200
def disable_connection(self, conn_id):
status, _ = self._request("POST", "/api/connections/%s/disable" % urllib.parse.quote(conn_id, safe=""))
return status == 200
def enable_connection(self, conn_id):
status, _ = self._request("POST", "/api/connections/%s/enable" % urllib.parse.quote(conn_id, safe=""))
return status == 200
def press(self, page, row, col):
"""Press a button (down, then up 20 ms later) and return the HTTP status: 200 when a button is
there, 204 ("No control") when that cell is empty."""
status, _ = self._request("POST", "/api/location/%d/%d/%d/press" % (page, row, col))
return status
# --- Shared snapshot written by the pollers --------------------------------
class Snapshot:
def __init__(self):
self.conns = {} # label -> {"id", "enabled", "category", "message"}
self.conns_time = 0.0
self.vars = {} # (label, name) -> str or None
self.midi = {} # session name -> {"peers": [...]}
self.remote_midi = {} # host -> the same, read on that Mac over SSH (None: unreachable)
self.remote_midi_time = {} # host -> when it last reported
self.peer_addr = {} # (host or "", session, peer) -> address it was last seen at
self.conn_hosts = {} # Companion connection label -> its "host" setting
self.prefs = {} # (preferences domain, key) -> value as "defaults read" prints it
self.atem = None # atem_watch.js's latest report
self.spotify = None # {"running", "state", "track", "name", "artist"} on NA Ingest
self.spotify_recent = deque(maxlen=3) # last distinct tracks played
self.spotify_tracks = set() # songs known to be in the service playlist
self.spotify_complete = False # ...and that is all of them
self.spotify_tracks_time = 0.0
self.midi_time = 0.0
self.mdns = set() # Bonjour _apple-midi._udp instance names
self.mdns_time = 0.0
self.ping = {} # host -> consecutive failures (0 = answering)
self.link = None # interface active?
self.launchd = {} # label -> pid or None (loaded, not running)
self.launchd_time = 0.0
self.procs = {} # match -> running
self.http = {} # url -> reachable
self.disk = {} # path -> free GB
self.listeners = {} # MIDI source name -> coremidi.Listener
self.remote = {} # (host, command) -> True / False / None (could not ask)
self.dante_map = {} # Dante device name -> {"ip", "mac", "switch", "port", "shared", "poe", "seen"}
self.dante_map_time = 0.0 # last successful survey of the Dante switches
self.artnet_last = {} # source IP -> time its last Art-Net packet arrived
self.artnet_since = None # when the Art-Net listener started (None: not listening)
self.ready = None # readiness.py's latest look at this weekend's checklist
def var(self, label, name):
return self.vars.get((label, name))
def conns_fresh(self, now):
return now - self.conns_time < 10
# --- Checks ------------------------------------------------------------------
def short_reason(message, default):
m = (message or "").upper()
for pattern, text in (("ETIMEDOUT", "NO ANSWER"), ("TIMED OUT", "NO ANSWER"), ("TIMEOUT", "NO ANSWER"),
("ECONNREFUSED", "REFUSED"), ("EHOSTUNREACH", "UNREACHABLE"),
("ENETUNREACH", "UNREACHABLE"), ("EHOSTDOWN", "UNREACHABLE"),
("AUTH", "LOGIN"), ("LOGIN", "LOGIN"), ("PASSWORD", "LOGIN"),
("401", "LOGIN"), ("403", "LOGIN")):
if pattern in m:
return text
return default
def conn_result(snap, label, now, service_only=False):
if not snap.conns_fresh(now):
return Result(WARN, "NO DATA")
c = snap.conns.get(label)
if c is None:
return Result(ERR, "NOT IN COMPANION", service_only=service_only)
if not c.get("enabled"):
# Everything on the page is expected on; a disabled connection is a warning, not silence.
return Result(WARN, "DISABLED")
cat = c.get("category")
if cat == "good":
return Result(OK)
if cat == "warning":
return Result(WARN, short_reason(c.get("message"), "WARNING"), service_only=service_only)
if cat == "error":
return Result(ERR, short_reason(c.get("message"), "OFFLINE"), service_only=service_only)
return Result(WARN, "STARTING", service_only=service_only)
def parse_hms_minutes(text):
m = re.match(r"^\s*(\d+):(\d{1,2}):(\d{1,2})\s*$", text or "")
if not m:
return None
return int(m.group(1)) * 60 + int(m.group(2))
def fmt_minutes(mins):
return "%dh" % (mins // 60) if mins >= 60 else "%dm" % mins
def fmt_age(seconds):
s = int(seconds)
if s < 60:
return "%ds" % s
if s < 3600:
return "%dm" % (s // 60)
return "%dh" % (s // 3600)
def to_int(text):
try:
return int(str(text).strip())
except (TypeError, ValueError):
return None
def check_conn(chk, snap, ctx):
res = conn_result(snap, chk["conn"], ctx["now"], chk.get("service_only", False))
# "ignore_warning": the module's yellow warning state is harmless for this one (ProPresenter's
# API flickers to warning every 15-20 min); offline, error and disabled still alarm.
c = snap.conns.get(chk["conn"], {})
if chk.get("ignore_warning") and res.level == WARN and c.get("enabled") and c.get("category") == "warning":
res = Result(OK)
return [("conn", res)]
def hyperdeck_recording(snap, label):
return (snap.var(label, "status") or "").strip().lower() == "record"
def check_hyperdeck(chk, snap, ctx):
label = chk["conn"]
base = conn_result(snap, label, ctx["now"])
if base.level != OK:
return [("conn", base)]
recording = hyperdeck_recording(snap, label)
slot = (snap.var(label, "slotId") or "").strip()
mins = parse_hms_minutes(snap.var(label, "recordingTime"))
out = []
if (slot in ("", "-") or mins is None) and chk.get("optional"):
# An optional deck with no disk is simply not in use today: grey, no alarm.
# Insert a disk and it is watched like the others.
return [("conn", Result(OFF, info="NO DISK"))]
if slot in ("", "-") or mins is None:
out.append(("disk", Result(ERR, "NO DISK", info="NO DISK", service_only=chk.get("disk_service_only", True))))
info = "NO DISK"
else:
left = fmt_minutes(mins)
if mins < chk.get("full_minutes", 5):
out.append(("disk", Result(ERR, "FULL " + left)))
elif mins < chk.get("low_minutes", 30):
out.append(("disk", Result(WARN, "LOW " + left)))
else:
out.append(("disk", Result(OK)))
info = ("REC " + left) if recording else left
if chk.get("rec_sync", True):
if ctx["any_recording"] and not recording:
out.append(("rec", Result(ERR, "NOT REC")))
else:
out.append(("rec", Result(OK)))
out.insert(0, ("conn", Result(OK, info=info)))
return out
def check_dante(chk, snap, ctx):
label = chk["conn"]
base = conn_result(snap, label, ctx["now"])
if base.level != OK:
return [("conn", base)]
raw = snap.var(label, "devices")
try:
present = set(json.loads(raw)) if raw else None
except ValueError:
present = None
if present is None:
return [("conn", Result(WARN, "NO DATA"))]
devices = chk["devices"]
if isinstance(devices, list):
devices = {d: d for d in devices}
out, seen = [], 0
for name, alias in devices.items():
if name in present:
seen += 1
out.append((name, Result(OK)))
else:
out.append((name, Result(ERR, alias + " GONE", service_only=chk.get("service_only", False))))
if chk.get("count_all"):
# Count every Dante device watched anywhere on the page, so the number
# matches Dante Controller; alarms still belong to each device's own button.
everything = ctx["dante_all"].get(label, set())
out.insert(0, ("conn", Result(OK, info="%d/%d" % (len(everything & present), len(everything)))))
elif len(devices) > 1:
out.insert(0, ("conn", Result(OK, info="%d/%d" % (seen, len(devices)))))
return out
def channel_name(snap, label, ch):
name = (snap.var(label, "ch_%d_name" % ch) or "").strip().upper()
return name[:9] if name and name != "UNKNOWN" else "CH%d" % ch
def check_shure(chk, snap, ctx):
label = chk["conn"]
base = conn_result(snap, label, ctx["now"])
if base.level != OK:
return [("conn", base)]
out = []
for ch in range(1, chk.get("channels", 1) + 1):
name = channel_name(snap, label, ch)
bars = to_int(snap.var(label, "ch_%d_battery_bars" % ch))
res = Result(OK)
if bars is not None and 0 <= bars <= 5:
if bars <= chk.get("battery_err_bars", 1):
res = Result(ERR, name + " BATT", named=True)
elif bars <= chk.get("battery_warn_bars", 2):
res = Result(WARN, name + " BATT", named=True)
out.append(("ch%d_batt" % ch, res))
# Only while that channel's transmitter is on (battery reads): an idle
# receiver on a noisy frequency flags interference on and off all week.
intf = (snap.var(label, "ch_%d_interference_status" % ch) or "").strip().upper()
if bars is not None and 0 <= bars <= 5 and intf and intf not in ("NONE", "UNKNOWN"):
out.append(("ch%d_rf" % ch, Result(WARN, name + " INTERF", named=True)))
else:
out.append(("ch%d_rf" % ch, Result(OK)))
return out
def check_psm(chk, snap, ctx):
label = chk["conn"]
base = conn_result(snap, label, ctx["now"])
if base.level != OK:
return [("conn", base)]
out = []
level = ERR if chk.get("muted_is_error") else WARN
for ch in range(1, chk.get("channels", 1) + 1):
muted = (snap.var(label, "ch_%d_rf_mute" % ch) or "").strip().upper() == "MUTED"
res = Result(level, channel_name(snap, label, ch) + " RF MUTE", named=True) if muted else Result(OK)
out.append(("ch%d_mute" % ch, res))
return out
def check_tascam(chk, snap, ctx):
label = chk["conn"]
base = conn_result(snap, label, ctx["now"])
if base.level != OK:
return [("conn", base)]
out = [("conn", Result(OK))]
error = (snap.var(label, "error") or "").strip()
caution = (snap.var(label, "caution") or "").strip()
if error and error.lower() != "no error":
out.append(("error", Result(ERR, error.upper()[:14])))
else:
out.append(("error", Result(OK)))
if caution and caution.lower() != "no caution":
out.append(("caution", Result(WARN, caution.upper()[:14])))
else:
out.append(("caution", Result(OK)))
return out
def ping_target(chk, snap):
"""A ping check names a "host", or a Companion "conn" whose own address is used (so a new IP set
in Companion needs no config change); "host" is then only the fallback."""
return (snap.conn_hosts.get(chk["conn"]) if chk.get("conn") else None) or chk.get("host")
def check_ping(chk, snap, ctx):
fails = snap.ping.get(ping_target(chk, snap))
if fails is not None and fails >= chk.get("failures", 2):
msg = chk.get("msg") or chk.get("name", chk["host"]) + " NO PING"
return [("ping", Result(ERR, msg, service_only=chk.get("service_only", False), named=True))]
return [("ping", Result(OK))]
def check_launchd(chk, snap, ctx):
if not snap.launchd_time:
return [("svc", Result(OK))]
name = chk.get("name", chk["label"])
if chk["label"] not in snap.launchd:
return [("svc", Result(ERR, name + " NOT LOADED"))]
if snap.launchd[chk["label"]] is None:
return [("svc", Result(ERR, name + " STOPPED"))]
return [("svc", Result(OK))]
def check_process(chk, snap, ctx):
running = snap.procs.get(chk["match"])
if running is False:
return [("proc", Result(ERR, chk.get("name", "PROCESS") + " STOPPED"))]
return [("proc", Result(OK))]
def check_http(chk, snap, ctx):
if snap.http.get(chk["url"]) is False:
return [("http", Result(ERR, chk.get("msg") or chk.get("name", "WEB") + " UNREACH", named=True))]
return [("http", Result(OK))]
def check_spotify(chk, snap, ctx):
"""Spotify on NA Ingest must be playing the service playlist."""
sp_ = snap.spotify
if sp_ is None:
return [("spotify", Result(OK))] # not read yet
state = "CLOSED" if not sp_.get("running") else (sp_.get("state") or "?").upper()
if not sp_.get("running"):
return [("spotify", Result(ERR, "SPOTIFY CLOSED", named=True))]
if state != "PLAYING":
return [("spotify", Result(ERR, "NOT PLAYING"))]
known = snap.spotify_tracks
if known:
recent = list(snap.spotify_recent)
if snap.spotify_complete:
wrong = bool(recent) and recent[-1] not in known
else: # only part of the playlist is known: 3 strangers in a row
wrong = len(recent) >= 3 and all(t not in known for t in recent[-3:])
if wrong:
return [("spotify", Result(ERR, "WRONG PLAYLIST"))]
return [("spotify", Result(OK, info="PLAYING"))]
def check_pref(chk, snap, ctx):
"""An app setting on NA Ingest, e.g. ATEM Software Control's "Enable MIDI control" must stay off."""
value = snap.prefs.get((chk["domain"], chk["key"]))
if value is None or value == str(chk["want"]):
return [("pref", Result(OK))] # not read yet, or as it should be
return [("pref", Result(ERR, chk["msg"], named=True))]
def check_disk(chk, snap, ctx):
free = snap.disk.get(chk["path"])
if free is None:
return [("disk", Result(OK))]
text = "DISK %dGB" % free
if free < chk.get("err_gb", 5):
return [("disk", Result(ERR, text))]
if free < chk.get("warn_gb", 20):
return [("disk", Result(WARN, text))]
return [("disk", Result(OK, info="%dGB free" % free))]
def check_midi_peer(chk, snap, ctx):
# "host": a session on another Mac (e.g. NA Tracks -> Lighting PC, the FX lighting cues),
# read over SSH. Unreachable or not read yet = no verdict (that Mac's own ping says if it is down).
sessions = snap.remote_midi.get(chk["host"]) if chk.get("host") else (snap.midi if snap.midi_time else None)
if chk.get("host") and ctx["now"] - snap.remote_midi_time.get(chk["host"], ctx["now"]) > 12:
sessions = None # that Mac stopped reporting: no verdict
if sessions is None:
return [("peer", Result(OK))]
sess = sessions.get(chk["session"])
if sess is None:
return [("peer", Result(ERR, "NO SESSION"))]
for p in sess["peers"]:
if p["name"] == chk["peer"]:
# Measured 2026-10-07: a joining peer reads "connected" for ~6 s until
# clock sync, then "running". A peer that dies is removed from the list:
# at once if it says goodbye, after ~60 s if it crashes or its cable is pulled.
if p["state"] in ("running", "connected"):
return [("peer", Result(OK))]
return [("peer", Result(ERR, "MIDI " + (p["state"] or "?").upper()[:8]))]
return [("peer", Result(ERR, "MIDI LOST"))]
def check_mdns(chk, snap, ctx):
if snap.mdns_time and chk["instance"] not in snap.mdns:
return [("mdns", Result(ERR, chk.get("name", chk["instance"]) + " GONE", named=True))]
return [("mdns", Result(OK))]
def check_midi_rx(chk, snap, ctx):
listener = snap.listeners.get(chk["source"])
name = chk.get("name", "MIDI")
if listener is None or not listener.connected:
return [("rx", Result(OFF))]
if listener.last_time is None:
return [("rx", Result(OK, info=name + " --"))]
age = ctx["now"] - listener.last_time
limit = chk.get("max_silence_seconds")
if limit and ctx["service"] and age > limit:
return [("rx", Result(WARN, name + " SILENT"))]
return [("rx", Result(OK, info="%s %s" % (name, fmt_age(age))))]
def check_artnet(chk, snap, ctx):
"""A lighting program's Art-Net traffic (Vista polls for nodes every 0.5 s): its heartbeat."""
limit = chk.get("max_silence_seconds", 5)
since = snap.artnet_since
if since is None or ctx["now"] - since < limit: # not listening, or only just started
return [("artnet", Result(OK))]
last = snap.artnet_last.get(chk["source"])
if last is None or ctx["now"] - last > limit:
# A custom "msg" names the program itself ("VISTA SILENT"); on its own button plain "SILENT" will do.
return [("artnet", Result(ERR, chk.get("msg", "SILENT"), named="msg" in chk))]
return [("artnet", Result(OK))]
def remote_key(chk):
return "%s\0%s" % (chk["host"], chk["cmd"])
def check_remote(chk, snap, ctx):
"""A command run over SSH on another machine, e.g. "is ProPresenter running?"."""
if snap.remote.get(remote_key(chk)) is False:
return [("remote", Result(ERR, chk["msg"], named=True))]
return [("remote", Result(OK))]
def no_verdict(chk, snap, now):
"""True when a check has nothing to say yet: its ping has not run, ssh could not ask, or Companion's
Dante module has no device list. The failover treats that as "not clear yet"."""
t = chk["type"]
if t == "ping":
return snap.ping.get(ping_target(chk, snap)) is None
if t == "remote":
return snap.remote.get(remote_key(chk)) is None
if t == "dante":
raw = snap.var(chk["conn"], "devices")
return conn_result(snap, chk["conn"], now).level != OK or not raw
return False
def check_ready(chk, snap, ctx):
"""One line of the weekly checklist on the READY page; readiness.py does the looking. Until the
checklist is due (Saturday noon on) every gap is amber; from then on a real gap is red."""
if snap.ready is None:
return [("ready", Result(OFF, info="CHECKING"))]
item = snap.ready.get("items", {}).get(chk["item"])
if item is None:
return [("ready", Result(OFF, info="NOT SET UP"))]
if not item.get("faults"):
return [("ready", Result(OK, info=item.get("info", "")))]
return [(msg, Result(ERR if level == ERR and ctx.get("ready_due") else WARN, msg))
for level, msg in item["faults"]]
def check_ready_manual(chk, snap, ctx):
"""A checklist line only a person can judge (BOH mix): a tap marks it done for the week."""
done = ctx.get("ready_done", {}).get(chk["item"])
if done:
return [("done", Result(OK, info="DONE " + datetime.fromtimestamp(done).strftime("%a %H:%M").upper()))]
info = ((snap.ready or {}).get("items", {}).get(chk["item"]) or {}).get("info") # e.g. the mics PCO needs
msg = chk.get("msg", "TAP WHEN DONE")
return [("done", Result(ERR if ctx.get("ready_due") else WARN, (info + "\n" + msg) if info else msg))]
# What readiness.py looks at (check type "ready") and what a "ready" fix can do.
READY_ITEMS = ("tracks", "mix_foh", "lights", "prop", "timecode", "backgrounds", "led_cross", "lineup", "mics",
"track_bundle", "prop_bundle")
READY_ACTIONS = ("recheck", "lineup_sync", "led_cross", "copy_tracks", "ltc_to_pco")
CHECKS = {
"conn": check_conn, "hyperdeck": check_hyperdeck, "dante": check_dante, "shure": check_shure,
"psm": check_psm, "tascam": check_tascam, "ping": check_ping, "launchd": check_launchd,
"process": check_process, "http": check_http, "disk": check_disk, "midi_peer": check_midi_peer, "pref": check_pref, "spotify": check_spotify,
"mdns": check_mdns, "midi_rx": check_midi_rx, "remote": check_remote, "artnet": check_artnet,
"ready": check_ready, "ready_manual": check_ready_manual,
}
def ready_keys(cfg):
"""The READY page's keys (tile-shaped: label, pos, checks)."""
return (cfg.get("ready_page") or {}).get("keys", [])
def effective_config(cfg):
"""The config as the watchdog runs it: a key that is switched off ("disabled": true) keeps its
place on the page but checks nothing, and a check that is switched off is left out. Like a
connection switched off in Companion, everything stays in the settings, ready to switch on."""
out = json.loads(json.dumps(cfg))
# A computer switched off: nothing logs in to it, so its checks and fixes go too.
off_hosts = {a for a, h in cfg.get("hosts", {}).items() if h.get("disabled")}
off_ips = {str(h.get("ssh", "")).split("@")[-1] for a, h in cfg.get("hosts", {}).items() if a in off_hosts}
for fid, fix in (out.get("fixes") or {}).items():
if fix.get("host") in off_hosts:
fix["disabled"] = True
ready = out.get("ready") or {}
if (cfg.get("ready") or {}).get("disabled"): # the READY page switched off as a whole
for k in ready_keys(out):
k["disabled"] = True
for name in ("tracks_host", "prop_bundle_host"):
if ready.get(name) in off_hosts:
ready.pop(name)
for t in all_tiles(out):
if "checks" in t:
t["checks"] = [] if t.get("disabled") else [
c for c in t["checks"] if not c.get("disabled") and c.get("host") not in off_hosts
and not (c.get("type") == "ping" and c.get("host") in off_ips)
and not (c.get("type") == "http" and urllib.parse.urlparse(c.get("url", "")).hostname in off_ips)]
for c in t["checks"]:
if c.get("devices_off") and isinstance(c.get("devices"), dict): # Dante devices switched off
c["devices"] = {n: v for n, v in c["devices"].items() if n not in c["devices_off"]}
# Auto-fix or the auto switch switched off as a whole: their hours are emptied, so they never act.
if (cfg.get("auto_fix") or {}).get("disabled"):
out["auto_fix"]["windows"] = []
if (cfg.get("failover") or {}).get("disabled"):
out["failover"]["windows"] = []
return out
def all_tiles(cfg):
"""Every key the watchdog judges: the Status page's tiles and the READY page's keys. The READY
page's keys are not in ALL (nor in auto-fix): a weekday gap in the checklist is not an alarm."""
return cfg["tiles"] + ready_keys(cfg)
# Checks whose connection a long press may restart (Dante is shared, so not that one).
RESTARTABLE = ("conn", "hyperdeck", "shure", "psm", "tascam")
# A check's "fix" may name this instead of an entry in "fixes": restart its own Companion
# connection. The Dante button uses it for hardware, which cannot be restarted from here.
RESTART_CONNECTION = "restart-connection"
# The Dante button's catch-all: power-cycle the device's PoE port if the self-updating
# Dante map says it is PoE-powered (found by MAC, so re-patching needs no config change),
# otherwise restart Companion's Dante module.
DANTE_AUTO = "dante-auto"
def needed_variables(cfg):
"""(fast, slow) lists of (connection label, variable name) to poll."""
fast, slow = [], []
for tile in all_tiles(cfg):
for chk in tile.get("checks", []):
t, label = chk["type"], chk.get("conn")
if t == "hyperdeck":
fast += [(label, "status"), (label, "slotId"), (label, "recordingTime")]
slow.append((label, "ip")) # the deck's address, from Companion
elif t == "dante":
fast.append((label, "devices"))
elif t == "tascam":
fast += [(label, "error"), (label, "caution")]
elif t == "shure":
for ch in range(1, chk.get("channels", 1) + 1):
slow += [(label, "ch_%d_name" % ch), (label, "ch_%d_battery_bars" % ch),
(label, "ch_%d_interference_status" % ch)]
elif t == "psm":
for ch in range(1, chk.get("channels", 1) + 1):
slow += [(label, "ch_%d_name" % ch), (label, "ch_%d_rf_mute" % ch)]
dedupe = lambda xs: list(dict.fromkeys(xs))
return dedupe(fast), dedupe(slow)
# --- Debounce, flapping and acknowledgement --------------------------------
class Tracker:
def __init__(self):
self.bad_since = None
self.confirmed = False
self.acked = False
self.result = Result(OK)
self.since = None # when the current confirmed fault started
self.starts = deque() # confirmed fault start times (for flapping)
self.last_fault_msg = ""
def update(self, res, now, debounce, flap_window):
"""Returns "fault", "clear" or None."""
event = None
bad = res.level in (WARN, ERR)
if bad:
if self.bad_since is None:
self.bad_since = now
if not self.confirmed and now - self.bad_since >= debounce:
self.confirmed, self.since = True, now
self.starts.append(now)
event = "fault"
self.result = res
self.last_fault_msg = res.msg
else:
if self.confirmed:
event = "clear"
self.bad_since, self.confirmed, self.acked, self.since = None, False, False, None
self.result = res
while self.starts and now - self.starts[0] > flap_window:
self.starts.popleft()
return event
def effective(self, flap_count):
if self.confirmed:
return self.result
if self.bad_since is not None: # bad but still inside the debounce
return Result(OK)
if len(self.starts) >= flap_count:
return Result(WARN, "%s x%d" % (self.last_fault_msg, len(self.starts)), named=True)
return self.result
# --- Service mode ------------------------------------------------------------
DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
class AutoLog:
"""One plain-text file per day of everything auto-fix does, written line by line as it happens,
e.g. "~/Desktop/Auto-Fix Logs/2026-10-11 Sunday Auto-Fix.log" (a .log opens in Console, which
shows new lines live)."""
NAME = re.compile(r"^(\d{4})-(\d{2})-(\d{2}) [A-Za-z]+ Auto-Fix\.log$")
def __init__(self, directory, who="NA Ingest", when="on Sundays 06:30-13:30", keep_months=1):
self.dir = os.path.expanduser(directory)
self.lock = threading.Lock()
self.who, self.when, self.keep_months = who, when, max(1, int(keep_months))
def cleanup(self, now):
"""Delete day files once they are keep_months calendar months old (the 11 Oct file goes on
11 Nov). Only files named exactly like these logs are ever touched."""
today = datetime.fromtimestamp(now).date()
removed = []
try:
names = os.listdir(self.dir)
except OSError:
return removed
for name in names:
m = self.NAME.match(name)
if not m:
continue
y, mo, d = (int(x) for x in m.groups())
total = mo - 1 + self.keep_months
y2, mo2 = y + total // 12, total % 12 + 1
try:
due = datetime(y2, mo2, min(d, calendar.monthrange(y2, mo2)[1])).date()
except ValueError:
continue
if today >= due:
try:
os.remove(os.path.join(self.dir, name))
removed.append(name)
except OSError as e:
log.warning("auto-fix log cleanup %s: %s", name, e)
if removed:
log.info("auto-fix logs older than a month deleted: %s", ", ".join(sorted(removed)))
return removed
def path(self, now):
return os.path.join(self.dir, datetime.fromtimestamp(now).strftime("%Y-%m-%d %A") + " Auto-Fix.log")
def write(self, now, button, text):
dt = datetime.fromtimestamp(now)
line = "%s %-8s %s\n" % (dt.strftime("%H:%M:%S"), button, text.replace("\n", " "))
try:
with self.lock:
os.makedirs(self.dir, exist_ok=True)
path = self.path(now)
new = not os.path.exists(path)
if new:
self.cleanup(now) # a new day: drop files a month old
with open(path, "a", encoding="utf-8") as f:
if new:
f.write("Status panel auto-fixes - %s\n"
"(%s fixes these by itself %s; anything that closes a\n"
" running program stays a hold. Newest lines at the bottom.)\n\n"
% (dt.strftime("%A %-d %B %Y"), self.who, self.when))
f.write(line)
except OSError as e:
log.warning("auto-fix log %s: %s", self.dir, e)
def window_holds(w, day, hm, prev_day):
"""One window at one moment. "end" <= "start" means the window runs past midnight: it starts on
one of its days and ends early the next morning (a Saturday night service: sat 18:00-00:30)."""
if w["start"] < w["end"]:
return day in w["days"] and w["start"] <= hm < w["end"]
return (day in w["days"] and hm >= w["start"]) or (prev_day in w["days"] and hm < w["end"])
def in_windows(windows, now):
"""Is `now` inside one of [{"days": [...], "start": "HH:MM", "end": "HH:MM"}]?"""
dt = datetime.fromtimestamp(now)
hm, day = dt.strftime("%H:%M"), DAYS[dt.weekday()]
return any(window_holds(w, day, hm, DAYS[dt.weekday() - 1]) for w in windows)
def windows_phrase(windows):
"""Windows in words, for log headers and notes: "on Sundays 06:30-13:30"."""
parts = []
for w in windows:
days = [d for d in DAYS if d in w["days"]]
names = [datetime(2024, 1, 1 + DAYS.index(d)).strftime("%As") for d in days] # 2024-01-01 is a Monday
when = "every day" if len(days) == 7 else " and ".join(names)
parts.append("%s %s-%s" % (when, w["start"], w["end"]))
return ("on " + " and ".join(parts)) if parts else "never"
def check_identity(chk):
"""What a check is about, independent of its position in the list: the key that remembered
state (acknowledgements, auto-fix tries) is filed under, so editing the list keeps it."""
t = chk["type"]
if t == "midi_peer":
return "%s/%s/%s" % (chk.get("host", ""), chk["session"], chk["peer"])
if t == "remote":
return "%s/%s" % (chk["host"], hashlib.sha1(chk["cmd"].encode()).hexdigest()[:8])
if t == "ping":
return chk.get("conn") or chk.get("host", "")
if t == "pref":
return "%s/%s" % (chk["domain"], chk["key"])
single = {"http": "url", "mdns": "instance", "launchd": "label", "process": "match", "artnet": "source",
"atem": "part", "disk": "path", "midi_rx": "source", "ready": "item", "ready_manual": "item"}
if t in single:
return str(chk.get(single[t], ""))
return chk.get("conn") or t
def write_json(path, obj):
"""Write a small JSON file in one go (a reader never sees half of it)."""
tmp = "%s.%d-%x.tmp" % (path, os.getpid(), threading.get_ident() & 0xFFFF) # one per writer: two threads never trip over one file
with open(tmp, "w") as f:
json.dump(obj, f, indent=1, sort_keys=True)
os.replace(tmp, path)
class State:
"""One file (state.json) for everything the watchdog remembers across restarts: the ALARMS
override, acknowledged faults, auto-fix tries and when each fix last ran. One writer; saved whole."""
def __init__(self, path):
self.path, self.data, self.lock = path, {}, threading.Lock()
if not path:
return
try:
with open(path) as f:
self.data = json.load(f)
except (OSError, ValueError):
legacy = os.path.join(os.path.dirname(path), "service-mode.json") # the file before state.json
try:
with open(legacy) as f:
old = json.load(f)
self.data = {"service": {"override": old.get("override"), "last_sched": old.get("last_sched")}}
except (OSError, ValueError):
pass
def get(self, key, default=None):
return self.data.get(key, default)
def set(self, key, value):
with self.lock:
self.data[key] = value
if self.path:
try:
write_json(self.path, self.data)
except OSError as e:
log.warning("could not save %s: %s", self.path, e)
class ServiceMode:
"""On inside a scheduled window; a button press overrides until the next window edge."""
def __init__(self, windows, state):
self.windows = windows
self.state = state if isinstance(state, State) else State(state)
saved = self.state.get("service") or {}
self.override, self.last_sched = saved.get("override"), saved.get("last_sched")
def scheduled(self, now):
return in_windows(self.windows, now)
def on(self, now):
sched = self.scheduled(now)
if sched != self.last_sched:
self.last_sched, self.override = sched, None
self._save()
return self.override if self.override is not None else sched
def set(self, value, now):
sched = self.scheduled(now)
self.last_sched = sched
self.override = None if value == sched else value
self._save()
def _save(self):
self.state.set("service", {"override": self.override, "last_sched": self.last_sched})
# --- Engine ------------------------------------------------------------------
class Engine:
def __init__(self, cfg, state_path=None, data_dir=None):
cfg = effective_config(cfg)
self.cfg = cfg
self.tiles = {t["id"]: t for t in all_tiles(cfg)}
# The READY page's checklist keys (a "ready" kind key sums them up, on either page).
self.ready_ids = [r["id"] for r in ready_keys(cfg) if r.get("kind") != "ready"]
self.ready_cfg = cfg.get("ready") or {}
self.debounce = cfg.get("debounce_seconds", 3)
self.flap_count = cfg.get("flap", {}).get("count", 3)
self.flap_window = cfg.get("flap", {}).get("window_seconds", 600)
self.net = cfg.get("network", {})
self.site = cfg.get("site", {})
timing = cfg.get("timing", {})
self.cooldown = timing.get("fix_cooldown_seconds", 60) # the same fix never re-runs within this
self.hold_ms = cfg.get("page_hold_ms", 2000) # how long a key is held to run its fix
self.state = State(state_path)
self.service = ServiceMode(cfg.get("service_windows", []), self.state)
self.data_dir = data_dir
self.stopping = False # SIGTERM arrived: finish what runs, start nothing new
self.running = {} # fix thread name -> {"key", "what", "since"} (published as fixes-running.json)
self.fix_threads = [] # the threads behind self.running, joined at exit
self.any_recording = False
self.live = None # (live?, reason) last written to live-now.json
self.live_written = 0.0
self.restore_acks = set(self.state.get("acks") or []) # acknowledged before the last restart
self.trackers = {}
self.group_of = {}
for t in cfg["tiles"]:
for m in t.get("members", []):
self.group_of[m] = t["id"]
self.detail_ids = [t["id"] for t in cfg["tiles"] if "checks" in t or t.get("kind") == "network"]
self.prev_conn_bad = {}
self.watched_conns = {c["conn"] for t in cfg["tiles"] for c in t.get("checks", []) if c.get("conn")}
self.conn_good_at = {} # connection -> last time it was "good"
self.conn_drops = deque() # (time, conn label, group)
self.burst = None # {"start", "conns", "cleared"}
self.events = []
self.meta = {} # key -> (tile id, check, sub key)
self.fix_last = {tuple(k.split("\0")): v for k, v in (self.state.get("fix_last") or {}).items()}
self.executor = execute_fix
self.poe_executor = execute_poe
self.notes = {} # tile id -> hold feedback {"text", "level", "until"}
# The Restart page: one key per thing, hold 2 s = restart it even when it looks fine.
self.restart_keys = {k["id"]: k for k in cfg.get("restart_page", {}).get("keys", [])}
self.auto = cfg.get("auto_fix") or {}
saved = self.state.get("auto") or {}
self.auto_tries = {tuple(k.split("\0")): v for k, v in saved.get("tries", {}).items()}
self.auto_seen = {tuple(k.split("\0")): v for k, v in saved.get("seen", {}).items()}
# Failover: fixes marked "failover" run by themselves when their check fails (the stream's
# audio goes to FOH Backup when Studio B dies). One record per fix that fired, kept until the
# "undo" fix runs or the fault has been gone note_seconds; saved, so a restart never fires twice.
fo = cfg.get("failover") or {}
self.failover_cfg = fo
self.failover_windows = fo.get("windows", FAILOVER_WINDOWS)
self.failover_after = fo.get("after_seconds", 2)
self.failover_note = fo.get("note_seconds", 900)
self.failed = dict(self.state.get("failover") or {}) # fix id -> {"tile", "label", "note", "at", "faults", "cleared"}
self.failover_held = {} # fix id -> when the Dante-wide hold-back was last logged
self.notify = post_webhook # the alert message (replaced in tests)
self.autolog = AutoLog(self.auto.get("log_dir", "~/Desktop/Auto-Fix Logs"),
self.site.get("ingest_name", "NA Ingest"), windows_phrase(self.auto.get("windows", [])),
self.auto.get("keep_months", 1)) if (self.auto or fo) else None
if self.autolog:
self.autolog.cleanup(time.time())
self.auto_watch = {} # "BUTTON FAULT" auto-fix went after -> when (to log when it clears)
self.auto_paused = False
self.auto_on = False # inside the window (for the "auto-fix on" line)
self.auto_muted = False # ...with ALARMS muted, so only the "while_muted" fixes run
self.auto_last = 0.0
self.deck_ips = {} # HyperDeck connection -> last address Companion reported
self.now = 0.0
@property
def hold_text(self):
return "HOLD %ss" % ("%g" % (self.hold_ms / 1000.0))
@property
def wait_text(self):
c = int(self.cooldown)
return "WAIT %d MIN" % (c // 60) if c % 60 == 0 else "WAIT %d S" % c
def identity_key(self, key):
"""A tracker key ("tile:index:sub") as the position-free name its state is remembered by."""
tile_id, chk, sub = self.meta[key]
return "%s|%s|%s" % (tile_id, check_identity(chk), sub)
def remember(self):
"""Save what must survive a restart: acknowledgements, auto-fix tries, when fixes last ran."""
acks = sorted({self.identity_key(k) for k, tr in self.trackers.items()
if tr.confirmed and tr.acked and k in self.meta} | self.restore_acks)
self.state.set("acks", acks)
self.state.set("fix_last", {"\0".join(k): v for k, v in self.fix_last.items()})
self.state.set("auto", {"tries": {"\0".join(k): v for k, v in self.auto_tries.items()},
"seen": {"\0".join(k): v for k, v in self.auto_seen.items()}})
self.state.set("failover", self.failed)
def live_now(self, now):
"""(live?, why): changes to the panel wait while this is true. A recording HyperDeck is
always live; a campus may add clock windows ("live_services" in "site")."""
if self.any_recording:
return True, "a HyperDeck is recording"
if in_windows(self.site.get("live_services", []), now):
return True, "a live service is scheduled"
return False, ""
def publish(self, now):
"""live-now.json and fixes-running.json in the data dir (the settings page reads them)."""
if not self.data_dir:
return
live = self.live_now(now)
if live != self.live or now - self.live_written > 5:
try:
write_json(os.path.join(self.data_dir, "live-now.json"),
{"live": live[0], "reason": live[1], "time": now})
except OSError as e:
log.warning("live-now.json: %s", e)
self.live, self.live_written = live, now
def fix_started(self, name, key_id, what):
self.running[name] = {"key": key_id, "what": what, "since": time.time()}
self.publish_running()
def fix_done(self, name):
self.running.pop(name, None)
self.publish_running()
def publish_running(self):
if self.data_dir:
try:
write_json(os.path.join(self.data_dir, "fixes-running.json"), self.running)
except OSError as e:
log.warning("fixes-running.json: %s", e)
def start_fix_thread(self, target, key_id, what):
"""Every fix runs in a thread the watchdog knows about: a restart waits for them."""
name = "fix-%s-%d" % (key_id, int(time.time() * 1000) % 100000)
def wrapped():
try:
target()
finally:
self.fix_done(name)
self.fix_started(name, key_id, what)
t = threading.Thread(target=wrapped, name=name, daemon=True)
self.fix_threads = [x for x in self.fix_threads if x.is_alive()] + [t]
t.start()
return t
# raw results for every check key
def evaluate(self, snap, now, service_on):
ctx = {"now": now, "service": service_on, "any_recording": False, "dante_all": {},
"ready_due": in_windows(self.ready_cfg.get("due", []), now), "ready_done": self.ready_done(now)}
for t in all_tiles(self.cfg):
for chk in t.get("checks", []):
if chk["type"] == "dante":
ctx["dante_all"].setdefault(chk["conn"], set()).update(chk["devices"])
for t in all_tiles(self.cfg):
for chk in t.get("checks", []):
if chk["type"] == "hyperdeck" and chk.get("rec_sync", True):
if conn_result(snap, chk["conn"], now).level == OK and hyperdeck_recording(snap, chk["conn"]):
ctx["any_recording"] = True
self.any_recording = ctx["any_recording"]
raw = {}
for t in all_tiles(self.cfg):
for i, chk in enumerate(t.get("checks", [])):
try:
results = CHECKS[chk["type"]](chk, snap, ctx)
except Exception as e: # a bad check must not take the panel down
log.exception("check %s/%s failed", t["id"], chk["type"])
results = [("error", Result(WARN, "CHECK BUG"))]
for sub, res in results:
# "tag" names the device when one button covers several (APPS, MIDI).
if chk.get("tag") and res.level in (WARN, ERR) and not res.named:
res.msg, res.named = "%s %s" % (chk["tag"], res.msg), True
raw["%s:%d:%s" % (t["id"], i, sub)] = (t["id"], res, chk)
self.meta["%s:%d:%s" % (t["id"], i, sub)] = (t["id"], chk, sub)
for sub, res in self.network_results(snap, now):
raw["net:%s" % sub] = ("net", res, {"type": "network"})
return raw
def network_results(self, snap, now):
out = []
out.append(("link", Result(ERR, "LINK DOWN") if snap.link is False else Result(OK)))
gw = self.net.get("gateway")
fails = snap.ping.get(gw) if gw else None
out.append(("gateway", Result(ERR, "NO GATEWAY") if fails is not None and fails >= 2 else Result(OK)))
# Burst: several unrelated connections dropping together means the network, not the gear.
bad_now = set()
if snap.conns_fresh(now):
for label, c in snap.conns.items():
if label not in self.watched_conns:
# Helper connections no button watches (the six tcp-HyperDeck ones behind the
# "reboot all" button flickered together on 2026-10-11) are not the network.
continue
bad = bool(c.get("enabled")) and c.get("category") == "error"
if bad:
bad_now.add(label)
if c.get("category") == "good":
self.conn_good_at[label] = now
# Only a connection that was fine moments ago counts as dropping: one that a fix just
# restarted goes error -> starting -> error, and is not the network (2026-10-11 review).
if bad and not self.prev_conn_bad.get(label) and now - self.conn_good_at.get(label, -1e9) <= 30:
self.conn_drops.append((now, label, self.group_for_conn(label)))
self.prev_conn_bad[label] = bad
window = self.net.get("burst_window_seconds", 10)
while self.conn_drops and now - self.conn_drops[0][0] > window:
self.conn_drops.popleft()
conns = {c for _, c, _ in self.conn_drops}
groups = {g for _, _, g in self.conn_drops}
if (len(conns) >= self.net.get("burst_connections", 4) and len(groups) >= self.net.get("burst_groups", 3)
and (self.burst is None or self.burst["cleared"] is not None)):
self.burst = {"start": now, "conns": set(conns), "cleared": None}
self.events.append(("NETWORK", "burst: %s" % ", ".join(sorted(conns))))
if self.burst:
if self.burst["cleared"] is None:
self.burst["conns"] |= conns
still = self.burst["conns"] & bad_now
if still:
out.append(("burst", Result(ERR, "%d DROPPED" % len(self.burst["conns"]))))
else:
self.burst["cleared"] = now
if self.burst["cleared"] is not None:
if now - self.burst["cleared"] < self.net.get("remember_seconds", 300):
stamp = datetime.fromtimestamp(self.burst["start"]).strftime("%H:%M")
# Flashes (to be noticed) until acknowledged, then goes by itself; a hold clears it.
out.append(("burst", Result(WARN, "BLIP " + stamp, flash=True)))
else:
self.burst = None
if not any(sub == "burst" for sub, _ in out):
out.append(("burst", Result(OK)))
return out
def group_for_conn(self, label):
for t in self.cfg["tiles"]:
for chk in t.get("checks", []):
if chk.get("conn") == label:
return self.group_of.get(t["id"], t["id"])
return label
def step(self, snap, now, debounce=None):
self.now = now
service_on = self.service.on(now)
raw = self.evaluate(snap, now, service_on)
deb = self.debounce if debounce is None else debounce
for key, (tile_id, res, chk) in raw.items():
tr = self.trackers.setdefault(key, Tracker())
was_acked = tr.acked
# A MIDI peer leaving a session is a clean event: show it after 1 s, not the general 3 s.
event = tr.update(res, now, 0 if chk["type"] == "network" else
chk.get("debounce_seconds", 1 if chk["type"] == "midi_peer" else deb),
self.flap_window)
if event == "fault":
self.events.append(("FAULT", "%s %s" % (self.tiles[tile_id]["label"], res.msg)))
ident = self.identity_key(key) if key in self.meta else None
if ident in self.restore_acks: # acknowledged before the restart: still acknowledged
self.restore_acks.discard(ident)
tr.acked = True
self.remember()
elif event == "clear":
ident = self.identity_key(key) if key in self.meta else None
if was_acked or ident in self.restore_acks:
self.restore_acks.discard(ident)
self.remember() # an acknowledgement is for one fault only
text = "%s %s" % (self.tiles[tile_id]["label"], tr.last_fault_msg)
self.events.append(("CLEAR", text))
fixed_at = self.auto_watch.pop(text, None)
if fixed_at is not None and self.autolog:
self.autolog.write(now, self.tiles[tile_id]["label"],
"OK again: %s (%d s after the auto-fix)" % (tr.last_fault_msg, now - fixed_at))
# A check may stop reporting a fault instead of reporting it OK: a Shure or PSM check says
# "conn" while its connection is down and per channel once it is back (2026-10-08). Such a
# fault is over - clear it, or it lingers unseen and a hold would still try to fix it.
for key, tr in self.trackers.items():
if key not in raw and (tr.confirmed or tr.bad_since is not None):
was_acked = tr.acked
if tr.update(Result(OK), now, 0, self.flap_window) == "clear" and key in self.meta:
tile_id = self.meta[key][0]
self.events.append(("CLEAR", "%s %s" % (self.tiles[tile_id]["label"], tr.last_fault_msg)))
if was_acked:
self.remember()
self.items = {}
for key, (tile_id, res, chk) in raw.items():
tr = self.trackers[key]
eff = tr.effective(self.flap_count)
self.items.setdefault(tile_id, []).append(
(display_level(eff, tr.acked, service_on), eff, tr.since or now))
self.publish(now)
return render(self, service_on)
def keys_for(self, tile_id):
tile = self.tiles.get(tile_id, {})
if tile.get("kind") == "master":
ids = self.detail_ids
elif tile.get("kind") == "ready":
ids = self.ready_ids
elif "members" in tile:
ids = tile["members"]
else:
ids = [tile_id]
prefixes = tuple(i + ":" for i in ids)
return [k for k in self.trackers if k.startswith(prefixes)]
def ack(self, tile_id):
n = 0
for key in self.keys_for(tile_id):
tr = self.trackers[key]
if tr.confirmed and not tr.acked:
tr.acked = True
n += 1
if n:
self.remember()
return n
def restart_targets(self, tile_id):
tile = self.tiles.get(tile_id, {})
return [c["conn"] for c in tile.get("checks", []) if c["type"] in RESTARTABLE and c.get("conn")]
def fix_plan(self, tile_id, snap, now=None, min_age=0, errors_only=False):
"""What a hold on this button would do: fix only what is failing right now.
[(identity, "fix", fix id, check) or (identity, "restart", connection label, check)]
A check's own "fix" wins; otherwise a failing Companion connection is restarted.
Nothing is done for a healthy button, and a recording HyperDeck is never touched.
"""
tile = self.tiles.get(tile_id, {})
ids = self.ready_ids if tile.get("kind") == "ready" else (tile.get("members") or [tile_id])
plan, seen = [], set()
for key in sorted(self.meta):
tid, chk, sub = self.meta[key]
tr = self.trackers.get(key)
if tid not in ids or not tr or not tr.confirmed or tr.result.level not in (WARN, ERR):
continue
if errors_only and (tr.result.level != ERR or now - (tr.since or now) < min_age):
continue # auto-fix: only real faults that have lasted
# A connection switched off in Companion is fixed by switching it on - never by
# acting on the device (2026-10-08: a disabled HyperDeck connection got a deck reboot).
label = chk.get("conn")
if label and label in snap.conns and not snap.conns[label].get("enabled"):
ident = ("enable", label)
if ident not in seen:
seen.add(ident)
plan.append((ident, "enable", label, chk))
continue
fix_id = chk.get("fix")
if isinstance(fix_id, dict):
fix_id = fix_id.get(sub, fix_id.get("*")) # "*": every other sub-check
if fix_id == DANTE_AUTO:
entry = snap.dante_map.get(sub) if chk["type"] == "dante" else None
if entry and entry.get("poe") and entry.get("mac"):
ident = ("poe", entry["mac"])
if ident not in seen:
seen.add(ident)
plan.append((ident, "poe", sub, chk))
continue
fix_id = RESTART_CONNECTION # not PoE-powered, or not mapped yet
if fix_id == RESTART_CONNECTION and chk.get("conn"):
ident = ("restart", chk["conn"]) # e.g. Companion's Dante module
if ident not in seen:
seen.add(ident)
plan.append((ident, "restart", chk["conn"], chk))
continue
if fix_id:
if chk["type"] == "hyperdeck" and hyperdeck_recording(snap, chk["conn"]):
continue
# One run per fix, except a HyperDeck fix, which is per deck. The deck's
# address comes from its Companion connection (remembered if it goes away).
fix_kind = self.cfg["fixes"][fix_id].get("kind")
per_deck = fix_kind == "hyperdeck"
target = ""
if per_deck:
ip = chk.get("ip") or snap.var(chk["conn"], "ip") or self.deck_ips.get(chk["conn"])
if not ip:
continue
self.deck_ips[chk["conn"]] = ip
chk = dict(chk, ip=ip)
target = chk["conn"]
if fix_kind == "midi_connect":
# Reconnect the peer where it was last seen (or the config's address).
key = (chk.get("host", ""), chk["session"], chk["peer"])
address = snap.peer_addr.get(key) or chk.get("address")
if not address:
continue
chk = dict(chk, address=address)
target = "|".join(key)
ident = ("fix", fix_id, target)
if ident not in seen:
seen.add(ident)
plan.append((ident, "fix", fix_id, chk))
elif chk["type"] in RESTARTABLE and sub == "conn" and chk.get("conn"):
ident = ("restart", chk["conn"])
if ident not in seen:
seen.add(ident)
plan.append((ident, "restart", chk["conn"], chk))
return plan
def failing_targets(self):
"""(fix ids, Companion connections) of every check that is down right now: a Restart page
key whose fix (or a fix/connection in its "watch") is among them flashes."""
fixes, conns = set(), set()
for key, tr in self.trackers.items():
if not tr.confirmed or tr.result.level != ERR or key not in self.meta:
continue
_, chk, sub = self.meta[key]
fix_id = chk.get("fix")
if isinstance(fix_id, dict):
fix_id = fix_id.get(sub, fix_id.get("*"))
if fix_id:
fixes.add(fix_id)
if chk.get("conn"):
conns.add(chk["conn"])
return fixes, conns
def failing(self, tile_id):
"""Is anything on this button (or its members) in a confirmed fault right now?"""
for key in self.keys_for(tile_id):
tr = self.trackers[key]
if tr.confirmed and tr.result.level in (WARN, ERR):
return True
return False
def ready_done(self, now):
"""{checklist item: when it was tapped done} for this week (the week starts on "reset_day")."""
import readiness
since = readiness.week_start(now, self.ready_cfg.get("reset_day", "mon"))
return {k: v for k, v in (self.state.get("ready_done") or {}).items() if v >= since}
def toggle_done(self, key_id, now):
"""A tap on a READY page key that only a person can check: done for the week, or (tapped
again) not done after all."""
chk = next((c for c in self.tiles.get(key_id, {}).get("checks", []) if c["type"] == "ready_manual"), None)
if chk is None:
log.warning("DONE %s: not a tap-to-confirm key", key_id)
return None
done = self.ready_done(now)
if chk["item"] in done:
del done[chk["item"]]
log.info("DONE %s: taken back", key_id)
self.note(key_id, "NOT DONE", "idle", now, 2)
else:
done[chk["item"]] = now
log.info("DONE %s: %s", key_id, self.tiles[key_id]["label"])
self.note(key_id, "DONE", "ok", now, 2)
self.state.set("ready_done", done)
return chk["item"] in done
def note(self, tile_id, text, level, now, seconds):
"""Show `text` on the button for a while (hold feedback), instead of its usual text."""
self.notes[tile_id] = {"text": text, "level": level, "until": now + seconds}
def run_fixes(self, tile_id, snap, comp, now, cooldown=None):
cooldown = self.cooldown if cooldown is None else cooldown
if self.stopping:
self.note(tile_id, "RESTARTING\nSOON", "idle", now, 4)
return []
if self.tiles.get(tile_id, {}).get("kind") == "network":
# NET: a hold clears a remembered network blip (link and gateway clear themselves).
if self.burst is None:
self.note(tile_id, "NOTHING\nTO CLEAR", "idle", now, 4)
else:
log.info("HOLD net: network blip cleared by hand")
self.burst = None
self.conn_drops.clear()
self.note(tile_id, "CLEARED", "ok", now, 3)
return []
plan = self.fix_plan(tile_id, snap)
always = self.tiles.get(tile_id, {}).get("hold_always")
if not plan and always:
# e.g. TASCAM, HD1-6: the hold reconnects even when the button looks fine.
chk = next((c for c in self.tiles[tile_id].get("checks", []) if c.get("conn")), {})
if self.cfg["fixes"][always].get("kind") == "hyperdeck":
if hyperdeck_recording(snap, chk["conn"]):
log.info("HOLD %s: %s is recording, left alone", tile_id, chk["conn"])
self.note(tile_id, "RECORDING\nLEFT ALONE", "idle", now, 4)
return []
chk = dict(chk, ip=chk.get("ip") or snap.var(chk["conn"], "ip") or self.deck_ips.get(chk["conn"]))
return self.run_force(tile_id, snap, comp, now, cooldown, fix_id=always, chk=chk)
if not plan:
if self.failing(tile_id):
log.info("HOLD %s: failing, but there is no automatic fix for it", tile_id)
self.note(tile_id, "NO AUTO\nFIX", "idle", now, 4)
else:
log.info("HOLD %s: nothing failing, nothing to fix", tile_id)
self.note(tile_id, "NOTHING\nTO FIX", "idle", now, 4)
return plan
return self.run_plan(tile_id, plan, snap, comp, now, cooldown)
def run_plan(self, tile_id, plan, snap, comp, now, cooldown=None, auto=False, on_done=None):
"""Carry out fix_plan() items for one button (a hold, or auto-fix during service)."""
cooldown = self.cooldown if cooldown is None else cooldown
started, names, waited, midi = [], [], False, []
for ident, kind, what, chk in plan:
if now - self.fix_last.get(ident, 0) < cooldown:
log.info("HOLD %s: %s %s ran less than %ds ago, skipped", tile_id, kind, what, cooldown)
waited = True
continue
self.fix_last[ident] = now
if kind == "fix":
self.failover_undone(what) # switching back by hand ends the failover note
self.remember()
if kind == "enable":
c = snap.conns.get(what)
if c:
ok = comp.enable_connection(c["id"])
log.info("ENABLE %s -> %s", what, "ok" if ok else "failed")
names.append("TURN ON")
elif kind == "restart":
c = snap.conns.get(what)
if c and c.get("enabled"):
ok = comp.restart_connection(c["id"])
log.info("RESTART %s -> %s", what, "ok" if ok else "failed")
names.append("RECONNECT")
elif kind == "poe":
log.info("FIX %s: power-cycle %s's PoE port (found by MAC)", tile_id, what)
names.append("POWER CYCLE")
started.append((self.poe_executor, (self.cfg, what, dict(snap.dante_map.get(what, {}))), what))
else:
fix = self.cfg["fixes"][what]
log.info("FIX %s: %s", tile_id, fix.get("label", what))
names.append(fix.get("short", what.upper()))
if fix.get("kind") == "midi_connect":
midi.append((what, chk)) # one worker for all of them (see below)
else:
started.append((self.executor, (self.cfg, what, chk), fix.get("short", what.upper())))
if midi:
# One after another: two reconnects writing the same session at once overwrite each other.
def midi_worker(batch):
rs = [self.executor(self.cfg, f, c) for f, c in batch]
return next((r for r in rs if isinstance(r, str)), False if False in rs else True)
started.append((midi_worker, (midi,), self.cfg["fixes"][midi[0][0]].get("short", "MIDI")))
if not names:
if not auto:
self.note(tile_id, ("JUST TRIED\n" + self.wait_text) if waited else "NOTHING\nTO FIX", "idle", now, 4)
return plan
summary = names[0] if len(set(names)) == 1 else "%d FIXES" % len(names)
if not started: # only Companion reconnects: quick
self.note(tile_id, "CHECKING\n" + summary, "fixing", now, 15)
if on_done:
on_done("asked Companion to reconnect")
return plan
self.note(tile_id, ("AUTO-FIXING\n" if auto else "FIXING\n") + summary, "fixing", now, 180)
results = []
group = "members" in self.tiles.get(tile_id, {}) or len(started) > 1
def work(fn, args, short):
ok = fn(*args)
results.append(ok)
if isinstance(ok, str) and ok != WAKING:
said.append(((short + " ") if group else "") + ok) # on a group button, say which
if len(results) == len(started): # the last one to finish reports for all
done = time.time()
if WAKING in results:
self.note(tile_id, "WAKING PC\n" + summary, "fixing", done, 120)
outcome = "the PC was off: sent Wake-on-LAN"
elif said: # the fix says what to do next
self.note(tile_id, said[0], "failed", done, 60)
outcome = "did not work: " + said[0]
elif all(r is not False for r in results):
self.note(tile_id, "CHECKING\n" + summary, "fixing", done, 20)
outcome = "done, checking it comes back"
else:
self.note(tile_id, "FIX FAILED\n" + summary, "failed", done, 20)
outcome = "the fix command failed"
if on_done:
on_done(outcome)
said = []
for fn, args, short in started:
self.start_fix_thread(lambda fn=fn, args=args, short=short: work(fn, args, short), tile_id, short)
return plan
def auto_fix(self, snap, comp, now):
"""During the auto-fix windows (Sunday service), fix by itself what a hold would fix - but
only gentle fixes (reconnects, starting a closed app, DVS for a Mac that dropped), only for
real faults that have lasted "after_seconds", at most "max_tries" times each, never while
the network is having a blip, and never while ALARMS has muted the panel (maintenance).
Restarting a running program stays a hold."""
a = self.auto
if not a or now - self.auto_last < 5 or self.stopping:
return []
self.auto_last = now
# ALARMS muted (maintenance) stops auto-fix, except the harmless fixes in "while_muted"
# (playing the Spotify playlist): those keep running inside the window.
muted = not self.service.on(now)
keep = set(a.get("while_muted", [])) if muted else None
if not in_windows(a.get("windows", []), now) or (muted and not keep):
if self.auto_on and self.autolog:
self.autolog.write(now, "PANEL", "auto-fix off (%s)" % ("ALARMS muted" if in_windows(
a.get("windows", []), now) else "end of the window"))
self.auto_on = False
self.auto_tries.clear()
self.auto_seen.clear()
return []
if (not self.auto_on or self.auto_muted != muted) and self.autolog: # opens today's file, so it can be watched from the start
ends = [w["end"] for w in a.get("windows", []) if in_windows([w], now)]
self.autolog.write(now, "PANEL", "auto-fix on%s%s" % (" until " + ends[0] if ends else "",
" (ALARMS muted: only %s)" % ", ".join(sorted(keep)) if muted else ""))
self.auto_on, self.auto_muted = True, muted
net_down = any(self.trackers.get("net:" + k) and self.trackers["net:" + k].confirmed
and self.trackers["net:" + k].result.level == ERR for k in ("link", "gateway"))
if net_down or (self.burst and (self.burst["cleared"] is None
or now - self.burst["cleared"] < a.get("pause_after_network_seconds", 60))):
if not self.auto_paused and self.autolog:
self.autolog.write(now, "NET", "network problem: no auto-fixes until it settles")
self.auto_paused = True
return [] # everything looks broken at once: wait it out
if self.auto_paused and self.autolog:
self.autolog.write(now, "NET", "network settled: auto-fix is back on")
self.auto_paused = False
gentle = {f for f in a.get("fixes", []) if not self.cfg["fixes"].get(f, {}).get("never_auto")
and not self.cfg["fixes"].get(f, {}).get("disabled")}
done = []
for t in self.cfg["tiles"]:
if "checks" not in t:
continue # groups repeat their members
gone = [k for k in self.keys_for(t["id"]) if self.meta.get(k) and self.meta[k][1]["type"] == "dante"
and self.meta[k][2] != "conn" and self.trackers[k].confirmed and self.trackers[k].result.level == ERR]
if len(gone) > 1:
# Several Dante devices "gone" at once is NA Ingest's view of Dante or the Dante network,
# not those Macs: restarting their DVS would cut audio that is working. Leave it to a hold.
if self.auto_seen.get(("dante-wide", t["id"]), 0) < now - 600:
log.info("AUTO %s: %d Dante devices gone at once - not touching them, hold DANTE if needed",
t["id"], len(gone))
if self.autolog:
self.autolog.write(now, t["label"], "left alone: %d Dante devices gone at once (NA Ingest's "
"view of Dante or the Dante network) - hold DANTE if needed" % len(gone))
self.auto_seen[("dante-wide", t["id"])] = now
continue
plan = []
for ident, kind, what, chk in self.fix_plan(t["id"], snap, now, a.get("after_seconds", 20), True):
if not ((kind == "fix" and what in gentle) or (kind == "restart" and a.get("restart_connections"))):
continue # not gentle: stays a hold
if keep is not None and not (kind == "fix" and what in keep):
continue # ALARMS muted: only the "while_muted" fixes
self.auto_seen[ident] = now
tries, last = self.auto_tries.get(ident, (0, 0.0))
if tries >= a.get("max_tries", 2):
if tries == a.get("max_tries", 2):
log.info("AUTO %s: gave up on %s %s after %d tries - hold it to try again", t["id"], kind, what, tries)
if self.autolog:
self.autolog.write(now, t["label"], "gave up after %d tries: %s - hold %s to try again"
% (tries, self.describe(kind, what), t["label"]))
self.auto_tries[ident] = (tries + 1, last)
continue
if now - last < a.get("retry_seconds", 90) or now - self.fix_last.get(ident, 0) < 60:
continue
self.auto_tries[ident] = (tries + 1, now)
plan.append((ident, kind, what, chk))
if plan:
try_no = max(self.auto_tries[i][0] for i, _, _, _ in plan)
log.info("AUTO %s: fixing %s (try %d)", t["id"], ", ".join("%s %s" % (k, w) for _, k, w, _ in plan), try_no)
faults = [r.msg for k in self.keys_for(t["id"]) for r in [self.trackers[k].result]
if self.trackers[k].confirmed and r.level == ERR]
for f in faults:
self.auto_watch["%s %s" % (t["label"], f)] = now
label = t["label"]
if self.autolog:
self.autolog.write(now, label, "problem: %s - fixing: %s (try %d of %d)" % (
", ".join(faults) or "?", "; ".join(self.describe(k, w) for _, k, w, _ in plan),
try_no, a.get("max_tries", 2)))
started = time.time()
self.run_plan(t["id"], plan, snap, comp, now, auto=True,
on_done=(lambda outcome, label=label, started=started: self.autolog.write(
now + (time.time() - started), label, "-> " + outcome)) if self.autolog else None)
done.append(t["id"])
forget = a.get("forget_seconds", 600)
for ident in list(self.auto_tries):
# Forget the tries only once the fault has been gone a long while: a fix makes its own
# fault blink out for a moment (restarting, re-confirming), which must not reset the count.
if now - self.auto_seen.get(ident, 0) >= forget:
del self.auto_tries[ident]
if done or self.auto_tries:
self.remember()
return done
def failover_fix(self, key):
"""The failover fix a tracker key's check names (per sub-check, "*" for the rest), or None."""
tile_id, chk, sub = self.meta[key]
fix_id = chk.get("fix")
if isinstance(fix_id, dict):
fix_id = fix_id.get(sub, fix_id.get("*"))
fix = self.cfg["fixes"].get(fix_id) if fix_id else None
return fix_id if fix and fix.get("failover") and not fix.get("disabled") else None
def dante_gone(self, conn):
"""How many devices on this Dante connection are gone right now, counted over every dante
check on every key (the same device may be watched by two keys)."""
return len({sub for k, (tid, chk, sub) in self.meta.items()
if chk["type"] == "dante" and chk["conn"] == conn and sub != "conn"
and self.trackers.get(k) and self.trackers[k].confirmed and self.trackers[k].result.level == ERR})
def failover_note_for(self, fix_id):
"""The note ("ON FOH BACKUP") of a failover that fired for this fix and is not switched back yet."""
rec = self.failed.get(fix_id)
return rec.get("note") if rec else None
def failover_undone(self, fix_id):
"""A fix that switches back by hand (the "undo" of a failover fix) was started: drop the
failover's note. If the fault is still there the failover fires again after_seconds later,
which is right: the stream must never stay silent."""
for fid in [f for f, r in self.failed.items() if self.cfg["fixes"].get(f, {}).get("undo") == fix_id]:
log.info("FAILOVER %s: switched back by hand (%s)", fid, fix_id)
del self.failed[fid]
def failover(self, snap, comp, now):
"""Run a "failover" fix by itself the moment its check fails for after_seconds (Studio B dies:
press FOH Backup, so the stream is not silent until someone notices). Once per fault episode,
only inside the failover windows, even while ALARMS is muted and even during a network blip
(a dead network is no reason to leave the stream silent). Never switches back by itself.
Timing, from a crash to the press: the Studio B Mac dies -> two missed 1-second pings (2-3 s)
+ 1 s debounce + 2 s after_seconds = about 5-6 s. Pro Tools crashes -> the 2-second ssh check
(2-3 s) + 1 s + 2 s = about 5-6 s. Its Dante device vanishing is seen as fast as Companion's
Dante module notices, then 3 s debounce + 2 s. Returns the fix ids it fired."""
if self.stopping:
return []
faults = {} # failover fix id -> [(tracker key, tracker)] in a confirmed fault
bad = set() # ...and those with a check that is bad at all (still debouncing)
for key in self.meta:
fix_id = self.failover_fix(key)
tr = self.trackers.get(key)
if not fix_id or not tr:
continue
if tr.confirmed and tr.result.level == ERR:
faults.setdefault(fix_id, []).append((key, tr))
# A check with no verdict (ssh could not ask, no ping result yet) is not a clear: a blip in
# the asking must not re-arm the failover and press a second time.
if tr.confirmed or tr.bad_since is not None or no_verdict(self.meta[key][1], snap, now):
bad.add(fix_id)
changed = False
for fix_id, rec in list(self.failed.items()):
if fix_id in bad:
# Still there, or back inside its debounce (the first seconds after a restart): not clear.
continue
if rec.get("cleared") is None: # the fault went: armed again, the note stays
rec["cleared"] = now
log.info("FAILOVER %s: fault clear - armed again (the stream stays %s until switched back by hand)",
fix_id, rec.get("note", "").lower())
changed = True
elif now - rec["cleared"] > self.failover_note:
del self.failed[fix_id]
changed = True
fired = []
for fix_id, items in faults.items():
rec = self.failed.get(fix_id)
if rec and (rec.get("cleared") is None or now - rec["cleared"] < REARM_SECONDS):
continue # fired for this fault already (or it only just cleared)
if not in_windows(self.failover_windows, now):
continue
lasting = [(k, tr) for k, tr in items if now - (tr.since or now) >= self.failover_after]
triggers = []
for k, tr in lasting:
chk = self.meta[k][1]
if chk["type"] == "dante" and self.dante_gone(chk["conn"]) > 1:
# Several Dante devices gone at once is NA Ingest's view of Dante, not that device.
if self.failover_held.get(fix_id, 0) < now - 600:
log.info("FAILOVER %s: %d Dante devices gone at once - that is NA Ingest's view of Dante, "
"not %s; not switching", fix_id, self.dante_gone(chk["conn"]), self.meta[k][2])
if self.autolog:
self.autolog.write(now, self.tiles[self.meta[k][0]]["label"],
"failover held back: %d Dante devices gone at once (NA Ingest's view of "
"Dante, not %s)" % (self.dante_gone(chk["conn"]), self.meta[k][2]))
self.failover_held[fix_id] = now
continue
triggers.append((k, tr))
if not triggers:
continue
key, tr = triggers[0]
tile_id, chk, sub = self.meta[key]
fix, label = self.cfg["fixes"][fix_id], self.tiles[tile_id]["label"]
msgs = list(dict.fromkeys(t.result.msg for _, t in items))
what = fix.get("label", fix_id)
self.failed[fix_id] = {"tile": tile_id, "label": label, "note": "ON " + fix.get("short", fix_id.upper()),
"at": now, "faults": msgs, "cleared": None}
self.fix_last[("fix", fix_id, "")] = now # a hold or Restart key right after says JUST TRIED / JUST DONE
log.info("FAILOVER %s: %s - %s", tile_id, ", ".join(msgs), what)
if self.autolog:
self.autolog.write(now, label, "FAILOVER: %s - %s" % (", ".join(msgs), what))
# How to switch back: a Restart key that runs the undo, else the undo fix's own words.
back = next((k["label"] for k in self.restart_keys.values() if k["fix"] == fix.get("undo")), None)
undo = self.cfg["fixes"].get(fix.get("undo") or "", {}).get("label")
text = "%s: %s %s - switched over by itself: %s. Switch back by hand%s once it is fixed." % (
self.site.get("alert_prefix") or self.site.get("ingest_name") or "Status panel", label, ", ".join(msgs),
what, (" (hold %s on the Restart page)" % back) if back else (" (%s)" % undo) if undo else "")
threading.Thread(target=self.notify, args=(self.cfg.get("alert_webhook"), text), daemon=True).start()
started = time.time()
self.run_plan(tile_id, [(("failover", fix_id), "fix", fix_id, chk)], snap, comp, now, cooldown=0, auto=True,
on_done=(lambda outcome, label=label, started=started: self.autolog.write(
now + (time.time() - started), label, "-> " + outcome)) if self.autolog else None)
fired.append(fix_id)
changed = True
if changed:
self.remember()
return fired
def describe(self, kind, what):
"""A plan item in words, for the auto-fix log."""
if kind == "restart":
return "reconnect Companion's %s connection" % what
if kind == "fix":
return self.cfg["fixes"][what].get("label", what)
return "%s %s" % (kind, what)
def run_force(self, key_id, snap, comp, now, cooldown=None, fix_id=None, chk=None, verb="FORCE"):
"""Run one fix whatever the status says: a Restart page key, or a button's "hold_always".
Shares the cooldown guard with holds, so a hold and a restart key never double up."""
cooldown = self.cooldown if cooldown is None else cooldown
if self.stopping:
self.note(key_id, "RESTARTING\nSOON", "idle", now, 4)
return None
if fix_id is None:
key = self.restart_keys.get(key_id)
if not key:
log.warning("FORCE %s: no such restart key", key_id)
return None
fix_id = key["fix"]
if key.get("disabled"):
self.note(key_id, "TURNED\nOFF", "idle", now, 4)
return None
busy = [w for w in key.get("watch", []) if not w.startswith("conn:")
and now - self.fix_last.get(("fix", w, ""), 0) < cooldown]
if busy: # e.g. auto-fix is starting Ableton right now
log.info("FORCE %s: %s ran less than %ds ago, skipped", key_id, busy[0], cooldown)
self.note(key_id, "STARTING\n" + self.wait_text, "idle", now, 4)
return None
per_deck = self.cfg["fixes"][fix_id].get("kind") == "hyperdeck" and chk and chk.get("conn")
ident = ("fix", fix_id, chk["conn"] if per_deck else "") # the same guard a failing hold uses
if now - self.fix_last.get(ident, 0) < cooldown:
log.info("%s %s: %s ran less than %ds ago, skipped", verb, key_id, fix_id, cooldown)
cur = self.notes.get(key_id)
if not (cur and cur["level"] == "fixing" and now < cur["until"]): # still running: keep saying so
self.note(key_id, "JUST DONE\n" + self.wait_text, "idle", now, 4)
return None
self.fix_last[ident] = now
self.failover_undone(fix_id)
self.remember()
fix = self.cfg["fixes"][fix_id]
log.info("%s %s: %s", verb, key_id, fix.get("label", fix_id))
self.note(key_id, "PRESSING" if fix.get("kind") == "companion_press" else "RESTARTING", "fixing", now, 180)
def work():
r = self.executor(self.cfg, fix_id, chk or {})
done = time.time()
if r == WAKING:
self.note(key_id, "WAKING PC", "fixing", done, 120)
elif isinstance(r, str):
self.note(key_id, r, "failed", done, 60)
elif r is False:
self.note(key_id, "FAILED", "failed", done, 30)
else:
self.note(key_id, "DONE", "ok", done, 20)
self.start_fix_thread(work, key_id, fix.get("short", fix_id.upper()))
return fix_id
def run_test(self, key_id, fix_id, which, snap, comp, now):
"""The settings page's "Run this fix now": one check's fix, run exactly as a hold on that key
would run it (deck address, MIDI peer address, recording guard, cooldown), logged as TEST.
`which` picks the check: its number on the key, or its identity. Refused for anything a hold
would not do from that key, and for fixes that act on the switcher, a PoE port or Dante."""
tile = self.tiles.get(key_id)
if not tile:
log.warning("TEST %s: no such key", key_id)
return None
fix = self.cfg["fixes"].get(fix_id)
if not fix or fix.get("kind") in ("atem", "netgear_poe") or fix_id in (RESTART_CONNECTION, DANTE_AUTO):
log.warning("TEST %s: %s is not a fix the page may run", key_id, fix_id)
return None
checks = tile.get("checks", [])
chk = None
for i, c in enumerate(checks):
if which in (str(i), check_identity(c), urllib.parse.quote(check_identity(c), safe="")):
chk = c
if chk is None and tile.get("hold_always") == fix_id:
chk = next((c for c in checks if c.get("conn")), {})
if chk is None:
log.warning("TEST %s: no check %r on it", key_id, which)
return None
own = chk.get("fix")
own = set(own.values()) if isinstance(own, dict) else {own}
if fix_id not in own and tile.get("hold_always") != fix_id:
log.warning("TEST %s: %s is not attached to that check", key_id, fix_id)
return None
if chk.get("type") == "hyperdeck":
if hyperdeck_recording(snap, chk["conn"]):
log.info("TEST %s: %s is recording, left alone", key_id, chk["conn"])
self.note(key_id, "RECORDING\nLEFT ALONE", "idle", now, 4)
return None
chk = dict(chk, ip=chk.get("ip") or snap.var(chk["conn"], "ip") or self.deck_ips.get(chk["conn"]))
if fix.get("kind") == "midi_connect":
address = snap.peer_addr.get((chk.get("host", ""), chk["session"], chk["peer"])) or chk.get("address")
if not address:
log.info("TEST %s: no address known for %s", key_id, chk["peer"])
return None
chk = dict(chk, address=address)
return self.run_force(key_id, snap, comp, now, fix_id=fix_id, chk=chk, verb="TEST")
def display_level(res, acked, service_on):
if res.level == OFF:
return "off"
if res.level == OK:
return "ok"
if res.level == WARN:
return "notice" if res.flash and not acked else "warn"
if res.service_only and not service_on:
return "idle"
if not service_on:
return "quiet"
return "ack" if acked else "err"
def worst(items):
"""items: [(level, Result, since)] -> (level, ordered faults)."""
if not items:
return "off", []
level = max((lv for lv, _, _ in items), key=lambda lv: RANK[lv])
if all(lv == "off" for lv, _, _ in items):
level = "off"
faults = [it for it in items if it[0] in ("idle", "warn", "notice", "quiet", "ack", "err")]
faults.sort(key=lambda it: (-RANK[it[0]], it[2]))
return level, faults
def fault_text(tile, res):
return res.msg if res.named else "%s %s" % (tile["label"], res.msg)
# Characters x text size (in % of key height) that fit across one key, measured
# on Companion's web buttons: "VIDEO" fits at 29.4%, "TASCAM" does not.
LINE_CAPACITY = 145
LINE_HEIGHT = 1.15
def fit_size(text, largest=30, smallest=12):
"""Largest text size at which no word breaks in the middle and all lines fit."""
for size in range(largest, smallest - 1, -1):
cap = int(LINE_CAPACITY / size)
lines, fits = 0, True
for para in text.split("\n"):
width = 0
lines += 1
for word in para.split():
if len(word) > cap:
fits = False
break
if width and width + 1 + len(word) > cap:
lines += 1
width = len(word)
else:
width += (1 if width else 0) + len(word)
if not fits:
break
if fits and lines * size * LINE_HEIGHT <= 95:
return size
return smallest
def colored(text, level):
(b, f), (b2, f2) = COLORS[level]
return {"x": text, "b": b, "f": f, "b2": b2, "f2": f2, "s": fit_size(text)}
def failover_line(engine, ids):
"""An extra last line ("ON FOH BACKUP") for a key one of whose tiles had a failover fire, until
it is switched back by hand (or the fault has been gone a while)."""
notes = list(dict.fromkeys(r["note"] for r in engine.failed.values() if r.get("tile") in ids and r.get("note")))
return ("\n" + ", ".join(notes)) if notes else ""
def render(engine, service_on):
out = {}
down_fixes, down_conns = engine.failing_targets()
for kid, key in engine.restart_keys.items():
note = engine.notes.get(kid)
if key.get("disabled") or engine.cfg["fixes"].get(key["fix"], {}).get("disabled"):
out[kid] = colored(key["label"] + "\nOFF", "off") # switched off in the settings
continue
watch = [key["fix"]] + key.get("watch", [])
down = any(w[5:] in down_conns if w.startswith("conn:") else w in down_fixes for w in watch)
pressed = engine.failover_note_for(key["fix"])
if note and engine.now < note["until"]:
out[kid] = colored(key["label"] + "\n" + note["text"], note["level"])
elif pressed: # the failover pressed it already: say so, no flashing
out[kid] = colored(key["label"] + "\n" + pressed, "ok")
elif down: # flashes: this is the one to hold
out[kid] = colored(key["label"] + "\nHOLD TO FIX", "err")
else:
out[kid] = colored(key["label"] + "\n" + engine.hold_text, "idle")
for tile in all_tiles(engine.cfg):
tid, kind = tile["id"], tile.get("kind")
if tile.get("disabled"): # switched off in the settings: grey, checks nothing
out[tid] = colored(tile["label"] + "\nOFF", "off")
continue
note = engine.notes.get(tid)
if note and engine.now < note["until"]: # a hold is being acted on: say so
out[tid] = colored(tile["label"] + "\n" + note["text"], note["level"])
continue
if kind == "service":
manual = engine.service.override is not None
text = tile["label"] + "\n" + ("ON" if service_on else "OFF") + ("\nmanual" if manual else "")
out[tid] = colored(text, "service" if service_on else "off")
continue
if kind == "master":
# Idle faults are expected outside service; they do not stop "ALL OK".
items = [("ok" if lv == "idle" else lv, res, since, engine.tiles[t]) for t in engine.detail_ids
for lv, res, since in engine.items.get(t, [])]
level, faults = worst([(lv, res, since) for lv, res, since, _ in items])
if not faults:
text, level = "ALL OK", "ok"
else:
first = next(fault_text(t, r) for lv, r, s, t in items if (lv, r, s) == faults[0])
n_bad = sum(1 for f in faults if f[0] not in ("warn", "notice"))
text = ("%d DOWN" % n_bad if n_bad else "%d WARN" % len(faults)) + "\n" + first
out[tid] = colored(text, level)
continue
if kind == "ready":
# The checklist: amber while things are still to do, red once it is due and they are not.
items = [(lv, res, since, engine.tiles[m]) for m in engine.ready_ids
for lv, res, since in engine.items.get(m, [])]
level, faults = worst([(lv, res, since) for lv, res, since, _ in items])
if faults:
first = next(t for lv, r, s, t in items if (lv, r, s) == faults[0])
todo = len({t["id"] for lv, r, s, t in items if lv not in ("ok", "off")}) # keys, not faults
text = "%s\n%d TO DO\n%s" % (tile["label"], todo, first["label"]) # the key to look at first
elif level == "ok":
text = tile["label"] + "\nALL DONE"
else:
text = tile["label"] + "\nCHECKING"
out[tid] = colored(text, level)
continue
if "members" in tile:
items = [(lv, res, since, engine.tiles[m]) for m in tile["members"]
for lv, res, since in engine.items.get(m, [])]
level, faults = worst([(lv, res, since) for lv, res, since, _ in items])
if faults:
first = next(fault_text(t, r) for lv, r, s, t in items if (lv, r, s) == faults[0])
text = tile["label"] + "\n" + first + ("\n+%d" % (len(faults) - 1) if len(faults) > 1 else "")
else:
text = tile["label"] + "\n" + ("OK" if level == "ok" else "OFF")
out[tid] = colored(text + failover_line(engine, tile["members"]), level)
continue
items = engine.items.get(tid, [])
level, faults = worst(items)
if faults:
text = tile["label"] + "\n" + faults[0][1].msg + ("\n+%d" % (len(faults) - 1) if len(faults) > 1 else "")
else:
infos = [res.info for lv, res, _ in items if res.info]
if level == "off":
text = tile["label"] + "\n" + (infos[0] if infos else "OFF")
else:
text = tile["label"] + "\n" + (infos[0] if infos else "OK")
out[tid] = colored(text + failover_line(engine, [tid]), level)
return out
# --- Pollers -----------------------------------------------------------------
def poll_connections(comp, snap):
data = comp.connections()
conns = {}
for c in data:
st = c.get("status") or {}
conns[c["label"]] = {"id": c["id"], "enabled": c.get("enabled", False),
"category": st.get("category"), "message": st.get("message")}
snap.conns, snap.conns_time = conns, time.time()
def poll_variables(comp, snap, pairs):
values = dict(snap.vars)
for label, name in pairs:
values[(label, name)] = comp.variable(label, name)
snap.vars = values
def ping_once(host):
try:
return subprocess.run(["/sbin/ping", "-c", "1", "-t", "1", "-q", host], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=3).returncode == 0
except subprocess.TimeoutExpired:
return False
_ping_lock = threading.Lock() # the 3-second and the 1-second ping pollers both update snap.ping
def poll_pings(snap, hosts, interface):
results = {}
def one(h):
results[h] = ping_once(h)
threads = [threading.Thread(target=one, args=(h,)) for h in hosts]
for t in threads:
t.start()
for t in threads:
t.join()
with _ping_lock:
pings = dict(snap.ping)
for h, ok in results.items():
pings[h] = 0 if ok else pings.get(h, 0) + 1
snap.ping = pings
if interface:
try:
out = subprocess.run(["/sbin/ifconfig", interface], capture_output=True, text=True, timeout=3).stdout
snap.link = "status: active" in out
except (OSError, subprocess.TimeoutExpired):
snap.link = None
def poll_system(snap, cfg):
checks = [c for t in all_tiles(cfg) for c in t.get("checks", [])]
if any(c["type"] == "launchd" for c in checks):
out = subprocess.run(["/bin/launchctl", "list"], capture_output=True, text=True, timeout=5).stdout
agents = {}
for line in out.splitlines()[1:]:
parts = line.split("\t")
if len(parts) == 3:
agents[parts[2]] = int(parts[0]) if parts[0].isdigit() else None
snap.launchd, snap.launchd_time = agents, time.time()
procs = {}
for c in checks:
if c["type"] == "process":
r = subprocess.run(["/usr/bin/pgrep", "-f", c["match"]], stdout=subprocess.DEVNULL, timeout=5)
procs[c["match"]] = r.returncode == 0
elif c["type"] == "disk":
usage = shutil.disk_usage(c["path"])
snap.disk = dict(snap.disk, **{c["path"]: int(usage.free / 1e9)})
elif c["type"] == "pref":
r = subprocess.run(["/usr/bin/defaults", "read", c["domain"], c["key"]], capture_output=True,
text=True, timeout=5)
snap.prefs[(c["domain"], c["key"])] = r.stdout.strip() if r.returncode == 0 else None
snap.procs = procs
def parse_dns_sd(out):
"""Instance names still present in `dns-sd -B` output."""
names = set()
for line in out.splitlines():
m = re.match(r"^\s*\S+\s+(Add|Rmv)\s+\d+\s+\d+\s+\S+\s+_apple-midi\._udp\.\s+(.+?)\s*$", line)
if m:
(names.add if m.group(1) == "Add" else names.discard)(m.group(2))
return names
def poll_mdns(snap, seconds=3):
proc = subprocess.Popen(["/usr/bin/dns-sd", "-B", "_apple-midi._udp", "local."],
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
try:
out, _ = proc.communicate(timeout=seconds)
except subprocess.TimeoutExpired:
proc.kill()
out, _ = proc.communicate()
snap.mdns, snap.mdns_time = parse_dns_sd(out), time.time()
_UNVERIFIED = ssl._create_unverified_context() # reachability only; nothing is sent
_http_last = {}
def poll_http(snap, checks):
"""checks: [(url, interval seconds)]; each URL is fetched when its interval is due."""
result = dict(snap.http)
now = time.time()
due = [u for u, every in checks if now - _http_last.get(u, 0) >= every]
for url in due:
_http_last[url] = now
try:
urllib.request.urlopen(urllib.request.Request(url, method="HEAD"), timeout=6, context=_UNVERIFIED)
result[url] = True
except urllib.error.HTTPError:
result[url] = True # it answered; that is all we ask
except (OSError, ValueError):
result[url] = False
snap.http = result
def remember_peers(snap, host, sessions):
"""Keep each peer's last address, so a hold can reconnect it after it has gone."""
for name, sess in (sessions or {}).items():
for p in sess.get("peers", []):
if p.get("address"):
snap.peer_addr[(host, name, p["name"])] = p["address"]
def companion_db_path(cfg):
"""Companion's settings database: "db" in config "companion", else the folder of the build the
pages are written for ("companion.build", e.g. 5.0.7 -> v5.0), else the newest folder whose
database opens (an upgrade that was tried and abandoned leaves a folder with a dead copy)."""
if cfg["companion"].get("db"):
return os.path.expanduser(cfg["companion"]["db"])
root = os.path.expanduser("~/Library/Application Support/companion")
vers = [d for d in os.listdir(root) if re.fullmatch(r"v\d+\.\d+", d)] if os.path.isdir(root) else []
if not vers:
return None
build = str(cfg["companion"].get("build") or "").split("+")[0]
wanted = "v" + ".".join(build.split(".")[:2]) if build else None
order = sorted(vers, key=lambda d: tuple(int(x) for x in d[1:].split(".")), reverse=True)
if wanted in order:
order.remove(wanted)
order.insert(0, wanted)
for d in order:
path = os.path.join(root, d, "db.sqlite")
if os.path.exists(path) and db_opens(path):
return path
return None
def db_opens(path):
import sqlite3
try:
db = sqlite3.connect("file:%s?mode=ro" % urllib.parse.quote(path), uri=True, timeout=2)
try:
db.execute("SELECT 1 FROM instances LIMIT 1").fetchall()
finally:
db.close()
return True
except sqlite3.Error:
return False
def companion_conn_hosts(path):
"""{connection label: host} from Companion's own settings database, opened read-only."""
import sqlite3
if not path or not os.path.exists(path):
return {}
db = sqlite3.connect("file:%s?mode=ro" % urllib.parse.quote(path), uri=True, timeout=2)
try:
rows = db.execute("SELECT value FROM instances").fetchall()
finally:
db.close()
hosts = {}
for (value,) in rows:
try:
inst = json.loads(value)
except (TypeError, ValueError):
continue
host = (inst.get("config") or {}).get("host") if isinstance(inst, dict) else None
if inst.get("label") and isinstance(host, str) and host.strip():
hosts[inst["label"]] = host.strip()
return hosts
def poll_conn_hosts(snap, cfg):
if any(c["type"] == "ping" and c.get("conn") for t in all_tiles(cfg) for c in t.get("checks", [])):
try:
snap.conn_hosts = companion_conn_hosts(companion_db_path(cfg)) or snap.conn_hosts
except Exception as e: # Companion's schema is its own business: fall back to "host"
log.warning("could not read connection addresses from Companion: %s", e)
SPOTIFY_STATE = ('tell application "Spotify" to return (player state as text) & "|" & (id of current track) '
'& "|" & (name of current track) & "|" & (artist of current track)')
def playlist_tracks(playlist_uri):
"""(set of track uris, complete?) for a public playlist. The embed preview page lists the first
100 songs and carries an anonymous token, which reads the whole list from Spotify's playlist service."""
pid = playlist_uri.split(":")[-1]
req = lambda url, **h: urllib.request.urlopen(urllib.request.Request(url, headers=dict({"User-Agent": "Mozilla/5.0"}, **h)),
timeout=10, context=_UNVERIFIED).read().decode("utf-8", "replace")
m = re.search(r'',
req("https://open.spotify.com/embed/playlist/" + pid))
state = json.loads(m.group(1))["props"]["pageProps"]["state"]
tracks = {t["uri"] for t in state["data"]["entity"].get("trackList", []) if t.get("uri")}
try:
token = state["settings"]["session"]["accessToken"]
full = json.loads(req("https://spclient.wg.spotify.com/playlist/v2/playlist/%s?from=0&length=1000" % pid,
Authorization="Bearer " + token, Accept="application/json"))
items = {i["uri"] for i in full.get("contents", {}).get("items", []) if i.get("uri")}
if items and len(items) >= int(full.get("length") or 0):
return items, True
except (OSError, ValueError, KeyError, TypeError) as e:
log.warning("Spotify full playlist: %s (using the first %d songs)", e, len(tracks))
return tracks, False
def spotify_playlist(cfg, chk=None):
"""The service playlist: "spotify" in config is the one place it is set; a check may override it."""
return (chk or {}).get("playlist") or (cfg.get("spotify") or {}).get("playlist")
def poll_spotify(snap, cfg):
checks = [c for t in all_tiles(cfg) for c in t.get("checks", []) if c["type"] == "spotify"]
if not checks:
return
def refresh():
try:
snap.spotify_tracks, snap.spotify_complete = playlist_tracks(spotify_playlist(cfg, checks[0]))
except (OSError, ValueError, KeyError, AttributeError, TypeError) as e:
log.warning("Spotify playlist list: %s", e)
snap.spotify_tracks_time = time.time()
if time.time() - snap.spotify_tracks_time > 3600:
refresh()
# Never "tell" Spotify unless it is running: that would open it.
running = subprocess.run(["/usr/bin/pgrep", "-xq", "Spotify"], timeout=5).returncode == 0
if not running:
snap.spotify = {"running": False}
return
r = subprocess.run(["/usr/bin/osascript", "-e", SPOTIFY_STATE], capture_output=True, text=True, timeout=10)
if r.returncode != 0:
snap.spotify = {"running": True, "state": None, "error": r.stderr.strip()[-120:]}
return
parts = (r.stdout.strip().split("|") + ["", "", "", ""])[:4]
snap.spotify = {"running": True, "state": parts[0], "track": parts[1], "name": parts[2], "artist": parts[3]}
if parts[1] and (not snap.spotify_recent or snap.spotify_recent[-1] != parts[1]):
if parts[1] not in snap.spotify_tracks and time.time() - snap.spotify_tracks_time > 60:
refresh() # a song just added to the playlist?
snap.spotify_recent.append(parts[1])
def poll_midi(snap, cfg):
import coremidi
snap.midi, snap.midi_time = coremidi.sessions(), time.time()
remember_peers(snap, "", snap.midi)
for t in all_tiles(cfg):
for c in t.get("checks", []):
if c["type"] == "midi_rx":
lis = snap.listeners.get(c["source"])
if lis is None:
lis = snap.listeners[c["source"]] = coremidi.Listener(c["source"])
lis.poll()
# --- Other machines: SSH checks and fixes ------------------------------------
def remote_midi_hosts(cfg):
return sorted({c["host"] for t in all_tiles(cfg) for c in t.get("checks", [])
if c["type"] == "midi_peer" and c.get("host")})
def remote_midi_script(call="ns['sessions']()"):
"""Python for /usr/bin/python3 on another Mac: this folder's coremidi.py, then print `call`'s result.
Nothing is installed there; the code goes over stdin each time."""
src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "coremidi.py"), encoding="utf-8").read()
return ("import json\nns = {'__name__': 'spcm'}\nexec(compile(%r, 'coremidi.py', 'exec'), ns)\n"
"print('SPMIDI' + json.dumps(%s))\n" % (src, call))
def parse_remote_midi(out):
for line in (out or "").splitlines():
if line.startswith("SPMIDI"):
try:
return json.loads(line[6:])
except ValueError:
return None
return None
def remote_midi_stream_script(interval=0.5):
"""Like remote_midi_script, but keeps running: prints the sessions whenever they change (and every
5 s anyway), checking twice a second. It ends by itself when the ssh connection goes."""
src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "coremidi.py"), encoding="utf-8").read()
return ("import json, time\nns = {'__name__': 'spcm'}\nexec(compile(%r, 'coremidi.py', 'exec'), ns)\n"
"last, sent = None, 0.0\n"
"while True:\n"
" s = json.dumps(ns['sessions']())\n"
" if s != last or time.time() - sent > 5:\n"
" print('SPMIDI' + s, flush=True)\n"
" last, sent = s, time.time()\n"
" time.sleep(%s)\n" % (src, interval))
def stream_remote_midi(snap, cfg, host, stop):
"""One long-lived ssh per Mac (NA Tracks, ProPresenter) reporting its MIDI sessions as they
change, so a dropped link shows in about a second instead of up to 5 s + an ssh login."""
script = remote_midi_stream_script()
while not stop.is_set():
proc = None
try:
proc = subprocess.Popen(ssh_cmd(cfg, host)[:1] + ["-o", "ServerAliveInterval=2", "-o", "ServerAliveCountMax=3"]
+ ssh_cmd(cfg, host)[1:] + ["/usr/bin/python3", "-u", "-"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
proc.stdin.write(script)
proc.stdin.close()
empty_since = None
for line in proc.stdout:
data = parse_remote_midi(line)
if data is not None:
snap.remote_midi[host], snap.remote_midi_time[host] = data, time.time()
remember_peers(snap, host, data)
# A reader started while that Mac was still at the login window never sees the
# user's MIDI sessions: it reports none, for ever. Start again after a minute of nothing.
if data == {}:
empty_since = empty_since or time.time()
if time.time() - empty_since > 60:
log.info("MIDI stream from %s reports no sessions for a minute: reconnecting", host)
break
else:
empty_since = None
if stop.is_set():
break
except (OSError, ValueError) as e:
log.warning("MIDI stream from %s: %s", host, e)
finally:
if proc and proc.poll() is None:
proc.terminate()
snap.remote_midi[host] = None # no verdict until it reports again
stop.wait(3)
def start_remote_midi_streams(snap, cfg, stop):
threads = []
for host in remote_midi_hosts(cfg):
t = threading.Thread(target=stream_remote_midi, args=(snap, cfg, host, stop), name="midi-" + host, daemon=True)
t.start()
threads.append(t)
return threads
def poll_remote_midi(snap, cfg):
script = remote_midi_script()
for host in remote_midi_hosts(cfg):
try:
r = subprocess.run(ssh_cmd(cfg, host) + ["/usr/bin/python3", "-"], input=script,
capture_output=True, text=True, timeout=20)
snap.remote_midi[host] = parse_remote_midi(r.stdout) if r.returncode == 0 else None
remember_peers(snap, host, snap.remote_midi[host])
except (OSError, subprocess.SubprocessError):
snap.remote_midi[host] = None
def ssh_cmd(cfg, host):
"""ssh to a machine in config "hosts", with NA Ingest's status-panel key; one shared connection per host."""
return ["/usr/bin/ssh", "-i", os.path.expanduser(cfg.get("ssh_key", "~/.ssh/id_status_panel")),
"-o", "BatchMode=yes", "-o", "ConnectTimeout=5", "-o", "IdentitiesOnly=yes", "-o", "LogLevel=ERROR",
"-o", "StrictHostKeyChecking=accept-new", "-o", "ControlMaster=auto",
"-o", "ControlPath=/tmp/status-panel-ssh-%C", "-o", "ControlPersist=120", cfg["hosts"][host]["ssh"]]
WAKING = "waking" # execute_fix result: the PC was off, a Wake-on-LAN packet went out
def wake_on_lan(mac, broadcast=None, port=9):
"""Send a Wake-on-LAN magic packet (to the PC's subnet broadcast as well, when given)."""
raw = bytes.fromhex(re.sub(r"[^0-9A-Fa-f]", "", mac))
if len(raw) != 6:
raise ValueError("not a MAC address: %r" % mac)
packet = b"\xff" * 6 + raw * 16
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
for addr in [a for a in (broadcast, "255.255.255.255") if a]:
s.sendto(packet, (addr, port))
return packet
def remote_script(cmds, shell="sh"):
"""One line that runs every check and prints SP=<0 or 1>. On a Mac the shell evaluates $?
after each command; cmd.exe expands %ERRORLEVEL% once before the line runs, so Windows gets a
conditional echo per command instead (never a bare %ERRORLEVEL%)."""
if shell == "cmd":
return " & ".join("((%s) >nul 2>&1 && echo SP%d=0 || echo SP%d=1)" % (cmd, i, i) for i, cmd in enumerate(cmds))
return "; ".join("(%s) >/dev/null 2>&1; echo SP%d=$?" % (cmd, i) for i, cmd in enumerate(cmds))
def host_shell(cfg, host, chk=None):
"""Which shell a remote check's command line is written for: "cmd" on a Windows PC, else "sh"."""
if chk and chk.get("shell"):
return chk["shell"]
return "cmd" if cfg.get("hosts", {}).get(host, {}).get("os") == "windows" else "sh"
def parse_remote(out, n):
codes = dict(re.findall(r"SP(\d+)=(\d+)", out or ""))
return [None if str(i) not in codes else codes[str(i)] == "0" for i in range(n)]
def remote_groups(cfg):
"""{(host, interval seconds): [commands]}: each machine's remote checks, grouped by how often they
ask to run ("interval_seconds"; Pro Tools on Studio B every 2 s, the rest every REMOTE_INTERVAL)."""
groups = {}
for t in all_tiles(cfg):
for c in t.get("checks", []):
if c["type"] == "remote":
key = (c["host"], c.get("interval_seconds", REMOTE_INTERVAL))
groups.setdefault(key, [])
if c["cmd"] not in groups[key]:
groups[key].append(c["cmd"])
return groups
_remote_lock = threading.Lock()
_remote_last = {} # (host, interval) -> when that group last ran
_remote_running = {} # (host, interval) -> the thread running it now
def run_remote_group(snap, cfg, host, cmds):
"""One ssh for one group of commands; its verdicts go into snap.remote."""
try:
out = subprocess.run(ssh_cmd(cfg, host) + [remote_script(cmds, host_shell(cfg, host))], capture_output=True,
text=True, timeout=20).stdout
except (OSError, subprocess.SubprocessError):
out = ""
with _remote_lock:
result = dict(snap.remote)
for cmd, ok in zip(cmds, parse_remote(out, len(cmds))):
result["%s\0%s" % (host, cmd)] = ok # None = could not reach it: no verdict
snap.remote = result
def poll_remote(snap, cfg, now=None, last=None):
"""Run the remote check groups that are due, each in its own thread (one ssh process per group,
sharing the host's ControlMaster), so a dead machine (ssh gives up after 5 s) never delays the
2-second group on another. A group still running is not started again. With no `last` every
group runs and the call waits for them (run_once, tests). Returns the threads it started."""
now = time.time() if now is None else now
threads = []
for key, cmds in remote_groups(cfg).items():
if last is not None:
running = _remote_running.get(key)
if now - last.get(key, 0) < key[1] or (running and running.is_alive()):
continue
last[key] = now
t = threading.Thread(target=run_remote_group, args=(snap, cfg, key[0], cmds), name="remote-" + key[0], daemon=True)
t.start()
threads.append(t)
if last is not None:
_remote_running[key] = t
if last is None:
for t in threads:
t.join()
return threads
# --- The self-updating Dante map: name -> IP -> MAC -> switch port -> PoE --------
def dante_map_path(data_dir):
return os.path.join(data_dir, "dante-map.json") if data_dir else None
def load_dante_map(path):
try:
with open(path) as f:
return json.load(f)
except (OSError, ValueError, TypeError):
return {}
def normalize_mac(mac):
return ":".join("%02x" % int(part, 16) for part in mac.split(":"))
def arp_table():
"""{ip: mac} from this Mac's ARP cache (macOS prints MACs without leading zeros)."""
out = subprocess.run(["/usr/sbin/arp", "-an"], capture_output=True, text=True, timeout=5).stdout
return {ip: normalize_mac(mac) for ip, mac in re.findall(r"\(([\d.]+)\) at ((?:[0-9a-fA-F]{1,2}:){5}[0-9a-fA-F]{1,2}) ", out)}
def dante_label(cfg):
for t in all_tiles(cfg):
for c in t.get("checks", []):
if c["type"] == "dante":
return c["conn"]
return None
def poll_dante_map(snap, cfg, comp, path, arp=arp_table, survey=None):
"""Every Dante device that is present now: where it is plugged in and whether that port
powers it. Kept for devices that later disappear, so a hold can still find them."""
switches, label = cfg.get("dante_switches"), dante_label(cfg)
if not switches or not label:
return
# Polled every 30 s so the first map appears soon after start; a full survey
# of the switches only every dante_map_seconds once one has succeeded.
if time.time() - snap.dante_map_time < cfg.get("dante_map_seconds", 600):
return
try:
present = json.loads(snap.var(label, "devices") or "[]")
except ValueError:
return
ips = arp()
wanted = {}
for name in present:
ip = comp.variable(label, name + "_ip")
if ip and ips.get(ip):
wanted[name] = (ip, ips[ip])
if not wanted:
return
if survey is None:
import netgear
survey = netgear.survey
found = survey(switches, [mac for _, mac in wanted.values()])
new, now = dict(snap.dante_map), int(time.time())
for name, (ip, mac) in wanted.items():
entry = dict(new.get(name, {}), ip=ip, mac=mac, seen=now)
entry.update(found.get(mac, {}))
new[name] = entry
snap.dante_map, snap.dante_map_time = new, time.time()
log.info("dante map: %d devices, %d on PoE ports", len(new), sum(1 for e in new.values() if e.get("poe")))
if path:
tmp = path + ".tmp"
with open(tmp, "w") as f:
json.dump(new, f, indent=1, sort_keys=True)
os.replace(tmp, path)
def execute_poe(cfg, name, entry):
"""Power-cycle a Dante device's PoE port, found by its MAC now (or its last known port)."""
try:
import netgear
host, port, how = netgear.poe_cycle_device(cfg.get("dante_switches", []), entry["mac"], entry)
log.info("FIX power-cycled %s on switch %s port %s (%s)", name, host, port, how)
return True
except Exception as e:
log.warning("FIX power-cycle %s failed: %s", name, e)
return False
def execute_fix(cfg, fix_id, chk):
"""Run one entry from config "fixes" (in its own thread). Only listed commands ever run.
Returns True if the command ran and succeeded."""
fix = cfg["fixes"][fix_id]
kind, label = fix.get("kind", "ssh"), fix.get("label", fix_id)
if fix.get("disabled"): # switched off in the settings: never runs
log.info("FIX %s: switched off in the settings, not run", label)
return "TURNED\nOFF"
try:
if kind in ("ssh", "local"):
argv = ssh_cmd(cfg, fix["host"]) + [fix["run"]] if kind == "ssh" else ["/bin/zsh", "-c", fix["run"]]
r = subprocess.run(argv, capture_output=True, text=True, timeout=fix.get("timeout", 60))
detail = (r.stdout + r.stderr).strip().replace("\n", " ")[-200:]
host = cfg.get("hosts", {}).get(fix.get("host"), {}) if kind == "ssh" else {}
if r.returncode == 255 and host.get("mac"):
# ssh could not reach the PC: it is probably off or asleep. Wake it; hold again
# once it is up (the app starts only after someone is logged in).
wake_on_lan(host["mac"], host.get("wake_broadcast"))
log.info("FIX %s -> %s does not answer: Wake-on-LAN sent to %s", label, fix["host"], host["mac"])
return WAKING
if r.returncode == 255 and kind == "ssh":
# The machine itself is off the network (e.g. the Bible TV PC's Wi-Fi dropped,
# 2026-10-11): say so instead of a bare FIX FAILED.
log.info("FIX %s -> %s not reachable: %s", label, fix["host"], detail)
return "%s NOT\nREACHABLE" % fix.get("short", fix["host"].upper())
log.info("FIX %s -> %s%s", label, "ok" if r.returncode == 0 else "exit %d" % r.returncode,
(": " + detail) if detail else "")
conn = chk.get("conn") or fix.get("conn")
if r.returncode == 0 and fix.get("reconnect") and conn:
# The app is running now: make Companion reconnect to it instead of waiting for its retry.
comp = Companion(cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000))
ids = {c["label"]: c["id"] for c in comp.connections()}
ok = conn in ids and comp.restart_connection(ids[conn])
log.info("FIX %s -> Companion connection %s restarted %s", label, conn, "ok" if ok else "FAILED")
return r.returncode == 0
if kind == "spotify":
return play_spotify(cfg, fix, chk, label)
if kind == "atem":
ok = bool(ATEM_WATCHER) and ATEM_WATCHER.fix(fix["part"])
log.info("FIX %s -> %s", label, "switcher matches the restore file again" if ok else "FAILED")
return ok
if kind == "reconnect":
return reconnect_and_verify(cfg, fix, label)
if kind == "midi_connect":
# Like Connect in Audio MIDI Setup: add the peer back to the session it dropped out of.
args = (chk["session"], chk["peer"], chk["address"])
if chk.get("host"):
r = subprocess.run(ssh_cmd(cfg, chk["host"]) + ["/usr/bin/python3", "-"],
input=remote_midi_script("ns['connect_peer'](%r, %r, %r)" % args),
capture_output=True, text=True, timeout=30)
result = parse_remote_midi(r.stdout) if r.returncode == 0 else None
if result is None:
log.info("FIX %s -> %s on %s failed: %s", label, chk["peer"], chk["host"],
(r.stdout + r.stderr).strip().replace("\n", " ")[-200:])
return False
else:
import coremidi
result = coremidi.connect_peer(*args)
log.info("FIX %s -> %s at %s, session %s%s: %s", label, chk["peer"], chk["address"], chk["session"],
(" on " + chk["host"]) if chk.get("host") else "", result)
return result in ("asked", "already", "kicked")
if kind == "netgear_poe":
# Netgear AV switch on the Dante network: power-cycle one PoE port (netgear.py).
import netgear
netgear.poe_cycle(fix["switch"], fix["port"], fix.get("mac"))
log.info("FIX %s -> switch %s port %s power-cycled", label, fix["switch"], fix["port"])
return True
if kind == "ready":
# The READY page: re-sync the lineup, put the cross back up, copy tracks out, or look again.
import readiness
return readiness.run_action(cfg, fix["action"], chk)
if kind == "hyperdeck":
# Turn Companion's connection to the deck off and on: after a deck restarts, that is what
# brings it back green (a plain reconnect does not, 2026-10-11). The deck is never rebooted.
return reconnect_and_verify(cfg, fix, label, conn=chk["conn"], host=chk.get("ip"), toggle=True)
if kind == "companion_press":
# Press a button on one of Companion's own pages (the Audio page's FOH Backup), as a finger would.
status = make_companion(cfg).press(fix["page"], fix["row"], fix["col"])
where = "page %s row %s col %s" % (fix["page"], fix["row"], fix["col"])
if status == 200:
log.info("FIX %s -> pressed Companion %s", label, where)
return True
if status == 204:
log.info("FIX %s -> no button at %s (HTTP 204)", label, where)
return "NO BUTTON\nTHERE"
log.info("FIX %s -> HTTP %d", label, status)
return False
log.warning("FIX %s: unknown kind %r", label, kind)
except Exception as e:
log.warning("FIX %s failed: %s", label, e)
return False
def play_spotify(cfg, fix, chk, label, run=None):
"""Play the service playlist in Spotify on this Mac, opening Spotify first if it is closed."""
run = run or subprocess.run
sp_ = cfg.get("spotify") or {}
playlist = spotify_playlist(cfg, chk) or fix.get("playlist")
if not playlist:
log.warning("FIX %s: no playlist set", label)
return False
if run(["/usr/bin/pgrep", "-xq", "Spotify"], timeout=5).returncode != 0:
run(["/usr/bin/open", "-g", "-a", "Spotify"], timeout=15)
time.sleep(sp_.get("open_wait", 8))
run(["/usr/bin/osascript", "-e", 'tell application "Spotify" to play track "%s"' % playlist],
capture_output=True, text=True, timeout=15)
time.sleep(sp_.get("play_wait", 3))
r = run(["/usr/bin/osascript", "-e", 'tell application "Spotify" to player state as text'],
capture_output=True, text=True, timeout=10)
ok = r.returncode == 0 and r.stdout.strip() == "playing"
log.info("FIX %s -> %s", label, "playing" if ok else "not playing (%s)" % (r.stdout.strip() or r.stderr.strip()[-80:]))
return ok
def reconnect_and_verify(cfg, fix, label, sleep=None, conn=None, host=None, toggle=False):
"""Restart one Companion connection (toggle=True: switch it off and on again), then look again
after "verify_seconds". Returns True if it came back; otherwise what to tell the operator:
"hung_note" when the device still answers ping, else that it is not on the network."""
sleep = sleep or time.sleep
label_conn = conn or fix["conn"]
fix = dict(fix, conn=label_conn)
comp = Companion(cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000))
conn = {c["label"]: c for c in comp.connections()}.get(fix["conn"])
if conn is None:
log.info("FIX %s -> no Companion connection %s", label, fix["conn"])
return False
if toggle:
if conn.get("enabled") and not comp.disable_connection(conn["id"]):
log.info("FIX %s -> Companion refused to switch %s off", label, fix["conn"])
return False
try:
sleep(fix.get("off_seconds", 2))
finally: # whatever happens, it is switched back on
on = comp.enable_connection(conn["id"])
if not on:
log.info("FIX %s -> Companion did not switch %s back on", label, fix["conn"])
return "STILL OFF IN\nCOMPANION"
log.info("FIX %s -> %s switched off and on", label, fix["conn"])
elif not conn.get("enabled"):
# Switched off in Companion: turn it on (never call that a hung device).
ok = comp.enable_connection(conn["id"])
log.info("FIX %s -> %s was disabled in Companion: turned on %s", label, fix["conn"], "ok" if ok else "FAILED")
if not ok:
return "DISABLED IN\nCOMPANION"
elif not comp.restart_connection(conn["id"]):
log.info("FIX %s -> Companion refused to restart %s", label, fix["conn"])
return False
sleep(fix.get("verify_seconds", 20))
after = {c["label"]: c for c in comp.connections()}.get(fix["conn"]) or {}
status = (after.get("status") or {}).get("category")
if status == "good":
log.info("FIX %s -> %s reconnected", label, fix["conn"])
return True
if status not in ("warning", "error"): # still starting up, or unknown: no verdict yet
log.info("FIX %s -> %s is %s after %ss", label, fix["conn"], status, fix.get("verify_seconds", 20))
return "STILL\nCONNECTING"
host = host or companion_conn_hosts(companion_db_path(cfg)).get(fix["conn"])
if host and ping_once(host):
log.info("FIX %s -> %s still %s but %s answers ping: control port hung", label, fix["conn"], status, host)
return fix.get("hung_note", "STILL DOWN")
log.info("FIX %s -> %s still %s and %s does not answer ping", label, fix["conn"], status, host or "its address")
return "NO NETWORK\nCHECK POWER"
# --- ATEM switcher: audio and camera settings against the restore file -------
MIX_OPTIONS = {"Off": 1, "On": 2, "AudioFollowVideo": 4}
# ATEM Software Control's camera setting names -> camera control (category, parameter) and value order
CCU_PARAMS = {
("Lens", "ApertureFstop"): (0, 2, ["value"]),
("Video", "ManualWhiteBalance"): (1, 2, ["temperature", "tint"]),
("Video", "Exposure"): (1, 5, ["value"]),
("Video", "DetailLevel"): (1, 8, ["value"]),
("Video", "SensorGainInDecibels"): (1, 13, ["value"]),
("Video", "SensorNDFilter"): (1, 16, ["value"]),
("ColorCorrection", "LiftAdjust"): (8, 0, ["red", "green", "blue", "luma"]),
("ColorCorrection", "GammaAdjust"): (8, 1, ["red", "green", "blue", "luma"]),
("ColorCorrection", "GainAdjust"): (8, 2, ["red", "green", "blue", "luma"]),
("ColorCorrection", "OffsetAdjust"): (8, 3, ["red", "green", "blue", "luma"]),
("ColorCorrection", "ContrastAdjust"): (8, 4, ["pivot", "adjust"]),
("ColorCorrection", "LumaMix"): (8, 5, ["value"]),
("ColorCorrection", "ColorAdjust"): (8, 6, ["hue", "saturation"]),
}
def atem_reference(xml_path, compare=None):
"""The restore file (.xml from ATEM Software Control) as what atem_watch.js compares against:
every Fairlight source's mix option and fader, and every camera's settings above. `compare`
("compare" in config "atem") narrows it: {"audio": ["mixOption", "faderGain"], "camera": [...]}."""
import xml.etree.ElementTree as ET
compare = compare or {}
audio_fields = compare.get("audio", ["mixOption", "faderGain"])
camera_params = compare.get("camera") # parameter names, e.g. ["ApertureFstop", ...]
root = ET.parse(xml_path).getroot()
audio = {}
mixer = root.find("FairlightAudioMixer")
for inp in (mixer.iter("AudioInput") if mixer is not None else []):
for src in inp.iter("AudioSource"):
mix = MIX_OPTIONS.get(src.attrib.get("mixOption"))
if mix is not None:
want = {"mixOption": mix}
try:
gain = float(src.attrib.get("faderGain", "nan"))
except ValueError:
gain = float("nan")
if math.isfinite(gain) and "faderGain" in audio_fields: # the file saves dB; the switcher 1/100 dB
want["faderGain"] = int(round(gain * 100))
if "mixOption" in audio_fields or "faderGain" in want:
audio["%s/%s" % (inp.attrib["id"], src.attrib["id"])] = want
camera = {}
cc = root.find("CameraControl")
for p in (cc if cc is not None else []):
spec = CCU_PARAMS.get((p.attrib.get("category"), p.attrib.get("parameter")))
if camera_params is not None and p.attrib.get("parameter") not in camera_params:
continue
if spec and all(f in p.attrib for f in spec[2]):
camera["%s/%d/%d" % (p.attrib["device"], spec[0], spec[1])] = [float(p.attrib[f]) for f in spec[2]]
return {"audio": audio, "camera": camera}
def newest(pattern):
"""The path matching a pattern like ".../bmd-atem-*/..." with the highest version number."""
import glob
def version(p):
return [int(n) for n in re.findall(r"\d+", p)]
found = sorted(glob.glob(os.path.expanduser(pattern)), key=version)
return found[-1] if found else None
class AtemWatcher:
"""Runs atem_watch.js (Companion's node + its atem-connection library) and keeps its latest report."""
def __init__(self, cfg, snap, data_dir, stop):
a = cfg["atem"]
self.atem, self.snap, self.stop = a, snap, stop
self.proc, self.lock, self.fixed = None, threading.Lock(), {}
here = os.path.dirname(os.path.abspath(__file__))
self.script = os.path.join(here, "atem_watch.js")
self.ref_path = os.path.join(data_dir or here, "atem-reference.json")
self.node, self.library = newest(a["node"]), newest(a["library"])
if not self.node or not self.library:
raise OSError("Companion's node or its ATEM library not found")
xml = os.path.expanduser(a["reference"])
if not os.path.isabs(xml):
xml = os.path.join(here, xml)
with open(self.ref_path, "w") as f:
json.dump(atem_reference(xml, a.get("compare")), f)
def run(self):
while not self.stop.is_set():
try:
self.proc = subprocess.Popen([self.node, self.script, self.atem["host"], self.library, self.ref_path],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, bufsize=1)
errors = deque(maxlen=5) # read all the time, so a chatty stderr never stalls it
threading.Thread(target=lambda p=self.proc, e=errors: [e.append(l) for l in p.stderr],
daemon=True).start()
for line in self.proc.stdout:
if line.startswith("ATEM "):
data = json.loads(line[5:])
data["time"] = time.time()
self.snap.atem = data
elif line.startswith("FIXED "):
r = json.loads(line[6:])
self.fixed[r.get("part")] = r
if self.stop.is_set():
break
except (OSError, ValueError) as e:
log.warning("ATEM watcher: %s", e)
finally:
if self.proc and self.proc.poll() is None:
self.proc.terminate()
if self.proc and not self.stop.is_set():
err = [l.strip() for l in list(locals().get("errors") or []) if l.strip() and "ThreadedClass" not in l]
log.warning("ATEM watcher stopped (exit %s)%s", self.proc.poll(), (": " + err[-1][:200]) if err else "")
self.snap.atem = None
self.stop.wait(5)
def fix(self, part, timeout=8):
"""Send back what differs for "audio" or "camera"; True once the switcher matches again."""
with self.lock:
if not self.proc or self.proc.poll() is not None:
return False
self.fixed.pop(part, None)
self.proc.stdin.write("fix %s\n" % part)
self.proc.stdin.flush()
end = time.time() + timeout
while time.time() < end:
time.sleep(0.5)
report = self.snap.atem or {}
if part in self.fixed and report.get("connected") and not report.get(part) \
and report.get("time", 0) > time.time() - 3:
return True
r = self.fixed.get(part) or {}
log.info("ATEM fix %s: %s", part, r.get("error") or "%s still different" % len((self.snap.atem or {}).get(part, [])))
return False
ATEM_WATCHER = None
def check_atem(chk, snap, ctx):
"""chk["part"] "audio" or "camera": the switcher must match the restore file."""
a = snap.atem
if not a or not a.get("connected") or ctx["now"] - a.get("time", 0) > 30:
return [(chk["part"], Result(OK))] # watcher not reporting: no verdict
diffs = a.get(chk["part"]) or []
if diffs:
return [(chk["part"], Result(ERR, chk["msg"], info="; ".join(diffs[:3]), named=True))]
return [(chk["part"], Result(OK))]
CHECKS["atem"] = check_atem
def start_artnet_listener(snap, stop, port=6454):
"""Note when each source last sent Art-Net. Shares the port (SO_REUSEPORT) so it never
keeps another program from listening too. Passive: nothing is ever sent."""
import select as _select
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if hasattr(socket, "SO_REUSEPORT"):
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
try:
s.bind(("", port))
except OSError as e:
log.warning("Art-Net listener could not start on port %d: %s", port, e)
return None
def loop():
snap.artnet_since = time.time()
while not stop.is_set():
if _select.select([s], [], [], 0.5)[0]:
data, (src, _) = s.recvfrom(2048)
if data[:8] == b"Art-Net\x00":
snap.artnet_last = dict(snap.artnet_last, **{src: time.time()})
s.close()
t = threading.Thread(target=loop, name="artnet", daemon=True)
t.start()
return t
class Periodic(threading.Thread):
def __init__(self, name, interval, fn, stop):
super().__init__(name=name, daemon=True)
self.interval, self.fn, self.stop = interval, fn, stop
self.failing = False
def run(self):
while not self.stop.is_set():
start = time.time()
try:
self.fn()
if self.failing:
log.info("%s recovered", self.name)
self.failing = False
except Exception as e:
if not self.failing:
log.warning("%s failing: %s", self.name, e)
self.failing = True
self.stop.wait(max(0.05, self.interval - (time.time() - start)))
# --- Main --------------------------------------------------------------------
def check_windows(windows, where):
for w in windows or []:
if not isinstance(w, dict) or not isinstance(w.get("days"), list) or not w["days"]:
raise ValueError("%s: a window needs days, start and end" % where)
for d in w["days"]:
if d not in DAYS:
raise ValueError("%s: unknown day %r" % (where, d))
for k in ("start", "end"):
if not re.fullmatch(r"([01]\d|2[0-3]):[0-5]\d", str(w.get(k, ""))):
raise ValueError("%s: %s must be HH:MM" % (where, k))
def check_host(cfg, host, where):
"""A machine the watchdog logs in to: listed in "hosts", with an ssh address, and not this Mac."""
h = cfg.get("hosts", {}).get(host)
if h is None:
raise ValueError("%s: unknown host %s" % (where, host))
if h.get("self"):
raise ValueError("%s: %s is this Mac - use a local fix or check" % (where, host))
if not h.get("ssh"):
raise ValueError("%s: host %s has no ssh address" % (where, host))
LAUNCHD_EXIT_CAP = 60 # launchd on macOS 26 applies at most 60 s of ExitTimeOut, whatever the plist says
def exit_timeout(cfg):
"""How long launchd should let the watchdog finish a running fix before killing it: the longest
fix plus its verify time plus a margin, never under a minute (install.sh writes it into the plist;
launchd caps what it honours at LAUNCHD_EXIT_CAP, so the shutdown budget is capped too)."""
longest = 0
for fix in cfg.get("fixes", {}).values():
longest = max(longest, fix.get("timeout", 60) + (fix.get("verify_seconds", 20)
if fix.get("kind") in ("reconnect", "hyperdeck") else 0))
return max(60, longest + 10)
def load_config(path):
with open(path) as f:
cfg = json.load(f)
for fid, fix in cfg.get("fixes", {}).items():
kind = fix.get("kind", "ssh")
if kind not in FIX_KINDS:
raise ValueError("fix %s: unknown kind %s" % (fid, kind))
if kind == "ssh":
check_host(cfg, fix.get("host"), "fix " + fid)
if kind in ("ssh", "local") and not fix.get("run"):
raise ValueError("fix %s: nothing to run" % fid)
if kind == "ready" and fix.get("action") not in READY_ACTIONS:
raise ValueError("fix %s: unknown READY action %s" % (fid, fix.get("action")))
if kind == "companion_press":
where = [fix.get("page"), fix.get("row"), fix.get("col")]
if (any(isinstance(v, bool) or not isinstance(v, int) for v in where)
or where[0] < 1 or where[1] < 0 or where[2] < 0):
raise ValueError("fix %s: a Companion button needs page, row and col (whole numbers)" % fid)
if fix.get("failover") and fix.get("never_auto"):
raise ValueError("fix %s: a failover fix runs by itself - it cannot be marked never to" % fid)
if fix.get("undo") and fix["undo"] not in cfg.get("fixes", {}):
raise ValueError("fix %s: unknown undo fix %s" % (fid, fix["undo"]))
check_windows(cfg.get("service_windows"), "service_windows")
check_windows((cfg.get("auto_fix") or {}).get("windows"), "auto_fix")
check_windows((cfg.get("failover") or {}).get("windows"), "failover")
check_windows((cfg.get("site") or {}).get("live_services"), "site.live_services")
ids = set()
for t in all_tiles(cfg):
if t["id"] in ids:
raise ValueError("duplicate tile id %s" % t["id"])
ids.add(t["id"])
for c in t.get("checks", []):
if c["type"] not in CHECKS:
raise ValueError("tile %s: unknown check type %s" % (t["id"], c["type"]))
for t in all_tiles(cfg):
for m in t.get("members", []):
if m not in ids:
raise ValueError("tile %s: unknown member %s" % (t["id"], m))
for c in t.get("checks", []):
fixes = c.get("fix") if isinstance(c.get("fix"), dict) else {"": c.get("fix")}
for f in fixes.values():
if f and f not in (RESTART_CONNECTION, DANTE_AUTO) and f not in cfg.get("fixes", {}):
raise ValueError("tile %s: unknown fix %s" % (t["id"], f))
if (f and f not in (RESTART_CONNECTION, DANTE_AUTO) and cfg["fixes"][f].get("kind", "ssh") == "ssh"
and cfg["fixes"][f]["host"] not in cfg.get("hosts", {})):
raise ValueError("fix %s: unknown host" % f)
if c["type"] == "remote" or (c["type"] == "midi_peer" and c.get("host")):
check_host(cfg, c["host"], "tile " + t["id"])
if t.get("hold_always") and t["hold_always"] not in cfg.get("fixes", {}):
raise ValueError("tile %s: unknown hold_always fix %s" % (t["id"], t["hold_always"]))
for f in (cfg.get("auto_fix") or {}).get("fixes", []):
if f not in cfg.get("fixes", {}):
raise ValueError("auto_fix: unknown fix %s" % f)
if cfg["fixes"][f].get("never_auto"):
raise ValueError("auto_fix: %s is marked never to run by itself" % f)
if re.search(r"pkill|taskkill|killall", cfg["fixes"][f].get("run", "")) and not cfg["fixes"][f].get("restart_ok"):
raise ValueError("auto_fix: %s closes a running program - that stays a hold" % f)
for k in cfg.get("restart_page", {}).get("keys", []):
if k["id"] in ids:
raise ValueError("restart key %s has a tile's id" % k["id"])
if k["id"] in {r["id"] for r in ready_keys(cfg)}:
raise ValueError("restart key %s has a READY key's id" % k["id"])
if k["fix"] not in cfg.get("fixes", {}):
raise ValueError("restart key %s: unknown fix %s" % (k["id"], k["fix"]))
for w in k.get("watch", []):
if not w.startswith("conn:") and w not in cfg.get("fixes", {}):
raise ValueError("restart key %s: unknown fix %s in watch" % (k["id"], w))
rp = cfg.get("ready_page")
if rp:
check_windows((cfg.get("ready") or {}).get("due"), "ready.due")
if rp.get("page") in (cfg.get("page"), (cfg.get("restart_page") or {}).get("page")):
raise ValueError("ready_page: page %s is already the Status or Restart page" % rp.get("page"))
for rk in ready_keys(cfg):
if "pos" not in rk:
raise ValueError("READY key %s has no place on the READY page" % rk["id"])
for c in rk.get("checks", []):
if c["type"] in ("ready", "ready_manual") and not c.get("item"):
raise ValueError("READY key %s: a %s check needs an item" % (rk["id"], c["type"]))
if c["type"] == "ready" and c["item"] not in READY_ITEMS:
raise ValueError("READY key %s: unknown item %s" % (rk["id"], c["item"]))
if sum(1 for t in cfg["tiles"] if t.get("kind") == "ready") > 1:
raise ValueError("at most one READY key on the Status page")
return cfg
def ping_hosts(cfg, snap=None, fast=False):
"""The hosts to ping every PING_INTERVAL seconds (plus the gateway), or with fast=True the ones
whose check asks for less (Studio B every second, so the failover sees it die within 2-3 s)."""
quick, slow = [], []
for t in all_tiles(cfg):
for c in t.get("checks", []):
if c["type"] == "ping":
host = ping_target(c, snap) if snap else c.get("host")
(quick if c.get("interval_seconds", PING_INTERVAL) < PING_INTERVAL else slow).append(host)
if fast:
return list(dict.fromkeys(h for h in quick if h))
gw = cfg.get("network", {}).get("gateway")
return list(dict.fromkeys(h for h in slow + ([gw] if gw else []) if h and h not in quick))
def http_checks(cfg):
"""[(url, interval)]: the Resi cloud every 30 s, a LAN app as often as its check asks."""
found = {}
for t in all_tiles(cfg):
for c in t.get("checks", []):
if c["type"] == "http":
found[c["url"]] = min(found.get(c["url"], 999), c.get("interval_seconds", 30))
return list(found.items())
def make_companion(cfg):
return Companion(cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000))
def start_pollers(cfg, snap, stop, data_dir=None):
fast, slow = needed_variables(cfg)
c1, c2, c3, c4 = (make_companion(cfg) for _ in range(4))
iface = cfg.get("network", {}).get("interface")
threads = [
Periodic("connections", 1.0, lambda: poll_connections(c1, snap), stop),
Periodic("variables", 1.0, lambda: poll_variables(c2, snap, fast), stop),
Periodic("slow-variables", 3.0, lambda: poll_variables(c3, snap, slow), stop),
Periodic("conn-hosts", 60.0, lambda: poll_conn_hosts(snap, cfg), stop),
Periodic("spotify", 3.0, lambda: poll_spotify(snap, cfg), stop),
Periodic("pings", float(PING_INTERVAL), lambda: poll_pings(snap, ping_hosts(cfg, snap), iface), stop),
Periodic("fast-pings", 1.0, lambda: poll_pings(snap, ping_hosts(cfg, snap, fast=True), None), stop),
Periodic("midi", 0.5, lambda: poll_midi(snap, cfg), stop),
Periodic("system", 10.0, lambda: poll_system(snap, cfg), stop),
Periodic("bonjour", 15.0, lambda: poll_mdns(snap), stop),
Periodic("http", 5.0, lambda: poll_http(snap, http_checks(cfg)), stop),
Periodic("remote", 1.0, lambda: poll_remote(snap, cfg, last=_remote_last), stop), # runs the groups that are due
Periodic("dante-map", 30.0, lambda: poll_dante_map(snap, cfg, c4, dante_map_path(data_dir)), stop),
]
for t in threads:
t.start()
threads += start_remote_midi_streams(snap, cfg, stop)
if cfg.get("atem") and any(c["type"] == "atem" for t in all_tiles(cfg) for c in t.get("checks", [])):
global ATEM_WATCHER
try:
ATEM_WATCHER = AtemWatcher(cfg, snap, data_dir, stop)
t = threading.Thread(target=ATEM_WATCHER.run, name="atem", daemon=True)
t.start()
threads.append(t)
except (OSError, ValueError) as e:
log.warning("ATEM checks off: %s", e)
if any(c["type"] == "artnet" for t in all_tiles(cfg) for c in t.get("checks", [])):
start_artnet_listener(snap, stop)
return threads
def snapshot_summary(snap, engine, now):
"""Everything the watchdog knows right now, in plain JSON (the settings page reads it instead of
asking the gear again): what Companion reports, pings, MIDI sessions here and on the other Macs,
helpers and programs, web checks, Spotify, the switcher, the Dante map, and every key's faults."""
out = {
"time": now,
"conns": {label: {"enabled": c.get("enabled"), "category": c.get("category"), "message": c.get("message")}
for label, c in snap.conns.items()},
"conns_fresh": snap.conns_fresh(now),
"conn_hosts": dict(snap.conn_hosts),
"ping": dict(snap.ping),
"link": snap.link,
"midi": snap.midi if snap.midi_time else None,
"remote_midi": {h: v for h, v in snap.remote_midi.items()},
"remote_midi_age": {h: round(now - t, 1) for h, t in snap.remote_midi_time.items()},
"mdns": sorted(snap.mdns) if snap.mdns_time else None,
"launchd": dict(snap.launchd),
"procs": dict(snap.procs),
"http": dict(snap.http),
"disk": dict(snap.disk),
"prefs": {"%s/%s" % k: v for k, v in snap.prefs.items()},
"remote": {k.replace("\0", "|"): v for k, v in snap.remote.items()},
"spotify": snap.spotify,
"spotify_known_songs": len(snap.spotify_tracks),
"atem": ({"connected": snap.atem.get("connected"), "audio": snap.atem.get("audio"), "camera": snap.atem.get("camera"),
"age": round(now - snap.atem.get("time", now), 1)} if snap.atem else None),
"dante_map": snap.dante_map,
"dante_map_age": round(now - snap.dante_map_time, 1) if snap.dante_map_time else None,
"artnet": {src: round(now - t, 1) for src, t in snap.artnet_last.items()},
"midi_rx": {name: {"connected": l.connected, "age": None if l.last_time is None else round(now - l.last_time, 1)}
for name, l in snap.listeners.items()},
"service_on": engine.service.on(now),
"alarms_override": engine.service.override,
"auto_fix": {"on": engine.auto_on, "paused": engine.auto_paused},
"failover": dict(engine.failed),
"burst": engine.burst and {"start": engine.burst["start"], "conns": sorted(engine.burst["conns"]),
"cleared": engine.burst["cleared"]},
"live": list(engine.live_now(now)),
"ready": snap.ready,
"faults": {},
}
for key, tr in engine.trackers.items():
if tr.confirmed and key in engine.meta:
tile_id, chk, sub = engine.meta[key]
out["faults"].setdefault(tile_id, []).append(
{"check": check_identity(chk), "type": chk["type"], "sub": sub, "level": tr.result.level,
"msg": tr.result.msg, "since": tr.since, "acked": tr.acked})
return out
def post_webhook(url, text):
if not url:
return
try:
req = urllib.request.Request(url, data=json.dumps({"text": text}).encode(),
headers={"Content-Type": "application/json"})
urllib.request.urlopen(req, timeout=8)
except (OSError, ValueError) as e:
log.warning("webhook failed: %s", e)
HOLD_MAX_AGE = 15 # seconds: a hold older than this (queued while the watchdog was down) is dropped
def handle_commands(text, engine, comp, snap, now):
for part in text.split(";"):
words = part.strip().split()
if not words:
continue
verb, arg = words[0].lower(), (words[1] if len(words) > 1 else "")
if verb == "ack":
log.info("ACK %s (%d faults)", arg, engine.ack(arg))
elif verb == "ackall":
log.info("ACK all (%d faults)", engine.ack("all"))
elif verb == "service":
target = {"on": True, "off": False}.get(arg, not engine.service.on(now))
engine.service.set(target, now)
log.info("SERVICE %s (button)", "ON" if target else "OFF")
elif verb in ("restart", "force"):
# Holds carry Companion's clock ("force rs_x 1791496225"). One pressed while the watchdog
# was down must not fire minutes later in a burst with the others.
stamp = words[2] if len(words) > 2 else None
if stamp is not None and (not stamp.isdigit() or now - int(stamp) > HOLD_MAX_AGE):
log.info("%s %s: pressed %s, too long ago - ignored", verb.upper(), arg,
"at %s" % stamp if stamp.isdigit() else "at an unknown time")
elif stamp is None and verb == "force":
log.info("FORCE %s: no time on it - ignored", arg)
elif verb == "restart": # a 2 second hold on a status button
engine.run_fixes(arg, snap, comp, now)
else: # a Restart page key, held 2 seconds
engine.run_force(arg, snap, comp, now)
elif verb == "test":
# The settings page: "test ".
if len(words) < 5 or not words[4].isdigit() or now - int(words[4]) > HOLD_MAX_AGE:
log.info("TEST %s: no time on it, or too long ago - ignored", arg)
else:
engine.run_test(arg, words[2], words[3], snap, comp, now)
elif verb == "done": # a tap on a READY page key a person checks
engine.toggle_done(arg, now)
elif verb == "resurvey": # the settings page: look at the Dante switches now
snap.dante_map_time = 0.0
log.info("RESURVEY: Dante map survey requested")
else:
log.warning("unknown command %r", part)
def run(cfg, data_dir, dry_run=False, stop=None, config_sha=None):
cfg = effective_config(cfg)
stop = stop or threading.Event()
names = cfg["variables"]
comp = make_companion(cfg)
snap = Snapshot()
engine = Engine(cfg, os.path.join(data_dir, "state.json"), data_dir)
snap.dante_map = load_dante_map(dante_map_path(data_dir))
site = cfg.get("site", {})
prefix = site.get("alert_prefix") or site.get("ingest_name") or "Status panel"
alert_after = cfg.get("timing", {}).get("alert_after_seconds", 30)
def on_term(signum, frame):
log.info("stop requested: %d fix(es) running will finish first", len(engine.running))
engine.stopping = True
stop.set()
try:
signal.signal(signal.SIGTERM, on_term)
except ValueError: # not the main thread (tests)
pass
threads = start_pollers(cfg, snap, stop, data_dir)
if cfg.get("ready_page") and not (cfg.get("ready") or {}).get("disabled"):
import readiness
poller = readiness.Poller(cfg, snap, stop, data_dir, is_live=lambda: engine.live_now(time.time())[0])
poller.start()
threads.append(poller)
if not dry_run:
try:
# A hold pressed while the watchdog was down must not fire on a config it was not meant for.
queued = comp.get_custom(names["command"])
if queued and queued.strip():
comp.set_custom(names["command"], "")
log.info("dropped commands queued before the start: %s", queued.strip())
except (OSError, http.client.HTTPException) as e:
log.warning("could not read the command variable at start: %s", e)
stop.wait(2.5) # let the first polls land before judging anything
last_json, last_write, last_eval, last_snapshot = None, 0.0, 0.0, 0.0
down_since, alerted, warned_missing = None, False, set()
log.info("status panel running (code %s, %s)", VERSION, "dry run" if dry_run else "live")
try:
write_json(os.path.join(data_dir, "started.json"),
{"pid": os.getpid(), "started_at": time.time(), "config_sha": config_sha, "version": VERSION})
except OSError as e:
log.warning("started.json: %s", e)
while not stop.is_set():
now = time.time()
try:
if not dry_run:
cmd = comp.get_custom(names["command"])
if cmd and cmd.strip():
comp.set_custom(names["command"], "")
handle_commands(cmd, engine, comp, snap, now)
last_eval = 0.0 # redraw now, so a press shows at once
if now - last_eval >= 0.5: # redraw twice a second
last_eval = now
tiles = engine.step(snap, now)
if not dry_run:
try:
engine.auto_fix(snap, comp, now)
except Exception: # a bug here must never take the panel down
log.exception("auto-fix failed; the panel carries on without it this round")
try:
engine.failover(snap, comp, now)
except Exception:
log.exception("failover failed; the panel carries on without it this round")
for kind, text in engine.events:
log.info("%s %s", kind, text)
engine.events = []
if now - last_snapshot >= 5:
last_snapshot = now
try:
write_json(os.path.join(data_dir, "snapshot.json"), snapshot_summary(snap, engine, now))
except (OSError, TypeError, ValueError) as e:
log.warning("snapshot.json: %s", e)
js = json.dumps(tiles, separators=(",", ":"), sort_keys=True)
if dry_run:
if js != last_json:
log.info("tiles: %s", " | ".join(tiles[t]["x"].replace("\n", " ") for t in sorted(tiles)))
last_json = js
else:
if js != last_json or now - last_write > 10:
ok, status = comp.set_custom(names["state"], js)
if ok:
last_json, last_write = js, now
elif names["state"] not in warned_missing:
log.warning("cannot set custom:%s (HTTP %s) - create it in Companion", names["state"], status)
warned_missing.add(names["state"])
ok, status = comp.set_custom(names["heartbeat"], str(int(now)))
if not ok and names["heartbeat"] not in warned_missing:
log.warning("cannot set custom:%s (HTTP %s) - create it in Companion", names["heartbeat"], status)
warned_missing.add(names["heartbeat"])
if down_since is not None:
log.info("Companion reachable again")
if alerted:
post_webhook(cfg.get("alert_webhook"), "%s: Companion is answering again." % prefix)
down_since, alerted = None, False
except (OSError, http.client.HTTPException) as e:
if down_since is None:
down_since = now
log.warning("Companion unreachable: %s", e)
elif not alerted and now - down_since > alert_after:
post_webhook(cfg.get("alert_webhook"), "%s: Companion has not answered for %d seconds."
% (prefix, alert_after))
alerted = True
stop.wait(0.25)
# A fix that is running finishes before the process goes (launchd waits up to the exit timeout,
# capped at LAUNCHD_EXIT_CAP, for us).
budget = time.time() + max(5, min(exit_timeout(cfg), LAUNCHD_EXIT_CAP) - 5)
for t in engine.fix_threads:
t.join(timeout=max(0.1, budget - time.time()))
if any(t.is_alive() for t in engine.fix_threads):
log.warning("stopping with a fix still running")
else:
log.info("stopped")
for t in threads:
t.join(timeout=2)
def run_once(cfg):
"""Poll everything synchronously, judge with no debounce, print the tiles."""
cfg = effective_config(cfg)
snap = Snapshot()
comp = make_companion(cfg)
fast, slow = needed_variables(cfg)
poll_connections(comp, snap)
poll_variables(comp, snap, fast + slow)
everything = ping_hosts(cfg) + ping_hosts(cfg, fast=True)
poll_pings(snap, everything, cfg.get("network", {}).get("interface"))
poll_pings(snap, everything, None) # pings need two misses to count
poll_system(snap, cfg)
poll_mdns(snap)
poll_http(snap, http_checks(cfg))
poll_remote(snap, cfg)
try:
poll_midi(snap, cfg)
except Exception as e:
print("midi: %s" % e)
if cfg.get("ready_page"):
import readiness
snap.ready = readiness.look(cfg)
engine = Engine(cfg, None)
tiles = engine.step(snap, time.time(), debounce=0)
rows = {}
for t in cfg["tiles"]:
if "pos" in t:
rows.setdefault(t["pos"][0], {})[t["pos"][1]] = t["id"]
for r in sorted(rows):
for c in sorted(rows[r]):
tid = rows[r][c]
tile = tiles[tid]
lv = next((k for k, ((b, f), _) in COLORS.items() if b == tile["b"] and f == tile["f"]), "?")
print("r%dc%d %-8s %-6s %s" % (r, c, tid, lv, tile["x"].replace("\n", " | ")))
print("hidden:", ", ".join("%s=%s" % (t["id"], tiles[t["id"]]["x"].replace("\n", " | "))
for t in cfg["tiles"] if "pos" not in t))
return tiles
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--config", default=os.path.join(HERE, "config.json"))
ap.add_argument("--data-dir", default=HERE)
ap.add_argument("--once", action="store_true", help="poll once, print the tiles, write nothing")
ap.add_argument("--dry-run", action="store_true", help="run but never write to Companion")
ap.add_argument("--log", default=os.path.expanduser("~/Library/Logs/status-panel.log"))
ap.add_argument("--check", action="store_true", help="load and validate the config, then exit")
ap.add_argument("--exit-timeout", action="store_true", help="print the launchd exit timeout this config needs")
args = ap.parse_args(argv)
cfg = load_config(args.config)
if args.check:
print("config OK: %d keys, %d fixes" % (len(cfg["tiles"]), len(cfg.get("fixes", {}))))
return
if args.exit_timeout:
print(exit_timeout(cfg))
return
if args.once:
run_once(cfg)
return
handler = logging.handlers.RotatingFileHandler(args.log, maxBytes=1_000_000, backupCount=3)
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
log.addHandler(handler)
if sys.stderr.isatty():
log.addHandler(logging.StreamHandler())
log.setLevel(logging.INFO)
if not (3, 9) <= sys.version_info[:2] <= (3, 12):
log.warning("python %s: the Dante switch reader (telnetlib) needs 3.9-3.12", sys.version.split()[0])
with open(args.config, "rb") as f:
config_sha = hashlib.sha256(f.read()).hexdigest()
run(cfg, args.data_dir, dry_run=args.dry_run, config_sha=config_sha)
if __name__ == "__main__":
main()