christianandClaude Opus 5.5
a0c72183bd
- Page 23 "Restart": hold a key 2 s to restart DVS (4 Macs), ProPresenter,
Ableton, Church Translate, ProdCom, ProdDash, captions, the bridge or the
Tascam connection even when the status page says it is fine ("force",
shared 60 s guard). RESTART PAGE key on page 19, STATUS PAGE back.
- Holds carry Companion's clock; ones queued while the watchdog was down
are dropped instead of firing together later.
- TASCAM hold always reconnects and checks again after 20 s: HUNG - POWER
CYCLE IT (pings, control port hung), NO NETWORK, or turns a disabled
connection on. Tascam pinged at its Companion address.
- INGEST: ProdCom (:24480), ProdDash's page (:24500), the caption screen;
DVS on Ingest restarts ProdCom afterwards. RESI: the encoder (10.1.10.202).
- Windows starts check the app really started (FIX FAILED otherwise); OBS's
start task runs elevated (OBS is "Run as administrator"); setup script
does that for any such app and gives sshd restart-on-failure.
- push_page.py checks what is on the target page, not its name (pages 22
and 97 are named "PAGE" but hold buttons); --which restart.
- Companion custom variables are written in the request body (no URL limit).
Reviewed by a 3-lens review with per-finding verification; 7 confirmed
issues fixed. Tested live: OBS hold opens OBS; Restart page TASCAM key.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1900 lines
83 KiB
Python
1900 lines
83 KiB
Python
#!/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 http.client
|
|
import json
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import re
|
|
import shutil
|
|
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")
|
|
|
|
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", "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)),
|
|
"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")
|
|
|
|
def __init__(self, level, msg="", info="", service_only=False, named=False):
|
|
self.level = level
|
|
self.msg = msg
|
|
self.info = info
|
|
self.service_only = service_only
|
|
self.named = named
|
|
|
|
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 enable_connection(self, conn_id):
|
|
status, _ = self._request("POST", "/api/connections/%s/enable" % urllib.parse.quote(conn_id, safe=""))
|
|
return status == 200
|
|
|
|
|
|
# --- 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.peer_addr = {} # (host or "", session, peer) -> address it was last seen at
|
|
self.conn_hosts = {} # Companion connection label -> its "host" setting
|
|
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)
|
|
|
|
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_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 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))]
|
|
|
|
|
|
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,
|
|
"mdns": check_mdns, "midi_rx": check_midi_rx, "remote": check_remote, "artnet": check_artnet,
|
|
}
|
|
|
|
# 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 cfg["tiles"]:
|
|
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 ServiceMode:
|
|
"""On inside a scheduled window; a button press overrides until the next window edge."""
|
|
|
|
def __init__(self, windows, path):
|
|
self.windows, self.path = windows, path
|
|
self.override, self.last_sched = None, None
|
|
if not path:
|
|
return
|
|
try:
|
|
with open(path) as f:
|
|
saved = json.load(f)
|
|
self.override, self.last_sched = saved.get("override"), saved.get("last_sched")
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def scheduled(self, now):
|
|
dt = datetime.fromtimestamp(now)
|
|
hm = dt.strftime("%H:%M")
|
|
day = DAYS[dt.weekday()]
|
|
return any(day in w["days"] and w["start"] <= hm < w["end"] for w in self.windows)
|
|
|
|
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):
|
|
if not self.path:
|
|
return
|
|
try:
|
|
tmp = self.path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump({"override": self.override, "last_sched": self.last_sched}, f)
|
|
os.replace(tmp, self.path)
|
|
except OSError as e:
|
|
log.warning("could not save service mode: %s", e)
|
|
|
|
|
|
# --- Engine ------------------------------------------------------------------
|
|
|
|
class Engine:
|
|
def __init__(self, cfg, state_path=None):
|
|
self.cfg = cfg
|
|
self.tiles = {t["id"]: t for t in cfg["tiles"]}
|
|
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.service = ServiceMode(cfg.get("service_windows", []), state_path)
|
|
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.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 = {} # fix identity -> last run time
|
|
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.deck_ips = {} # HyperDeck connection -> last address Companion reported
|
|
self.now = 0.0
|
|
|
|
# raw results for every check key
|
|
def evaluate(self, snap, now, service_on):
|
|
ctx = {"now": now, "service": service_on, "any_recording": False, "dante_all": {}}
|
|
for t in self.cfg["tiles"]:
|
|
for chk in t.get("checks", []):
|
|
if chk["type"] == "dante":
|
|
ctx["dante_all"].setdefault(chk["conn"], set()).update(chk["devices"])
|
|
for t in self.cfg["tiles"]:
|
|
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
|
|
raw = {}
|
|
for t in self.cfg["tiles"]:
|
|
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():
|
|
bad = bool(c.get("enabled")) and c.get("category") == "error"
|
|
if bad:
|
|
bad_now.add(label)
|
|
if bad and not self.prev_conn_bad.get(label):
|
|
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", 180):
|
|
stamp = datetime.fromtimestamp(self.burst["start"]).strftime("%H:%M")
|
|
out.append(("burst", Result(WARN, "BLIP " + stamp)))
|
|
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())
|
|
event = tr.update(res, now, 0 if chk["type"] == "network" else chk.get("debounce_seconds", deb),
|
|
self.flap_window)
|
|
if event == "fault":
|
|
self.events.append(("FAULT", "%s %s" % (self.tiles[tile_id]["label"], res.msg)))
|
|
elif event == "clear":
|
|
self.events.append(("CLEAR", "%s %s" % (self.tiles[tile_id]["label"], tr.last_fault_msg)))
|
|
# 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):
|
|
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)))
|
|
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))
|
|
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 "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
|
|
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):
|
|
"""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 = 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
|
|
# 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(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 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=60):
|
|
plan = self.fix_plan(tile_id, snap)
|
|
always = self.tiles.get(tile_id, {}).get("hold_always")
|
|
if not plan and always:
|
|
# e.g. TASCAM: its hold reconnects even when it looks fine (its control port hangs).
|
|
return self.run_force(tile_id, snap, comp, now, cooldown, fix_id=always)
|
|
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
|
|
started, names, waited = [], [], 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 == "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()))
|
|
started.append((self.executor, (self.cfg, what, chk), fix.get("short", what.upper())))
|
|
if not names:
|
|
self.note(tile_id, "JUST TRIED\nWAIT 1 MIN" 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)
|
|
return plan
|
|
self.note(tile_id, "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)
|
|
elif said: # the fix says what to do next
|
|
self.note(tile_id, said[0], "failed", done, 60)
|
|
elif all(r is not False for r in results):
|
|
self.note(tile_id, "CHECKING\n" + summary, "fixing", done, 20)
|
|
else:
|
|
self.note(tile_id, "FIX FAILED\n" + summary, "failed", done, 20)
|
|
said = []
|
|
for fn, args, short in started:
|
|
threading.Thread(target=work, args=(fn, args, short), daemon=True).start()
|
|
return plan
|
|
|
|
def run_force(self, key_id, snap, comp, now, cooldown=60, fix_id=None):
|
|
"""Run one fix whatever the status says: a Restart page key, or a button's "hold_always".
|
|
Shares the 60 s guard with holds, so a hold and a restart key never double up."""
|
|
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"]
|
|
ident = ("fix", fix_id, "")
|
|
if now - self.fix_last.get(ident, 0) < cooldown:
|
|
log.info("FORCE %s: %s ran less than %ds ago, skipped", 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\nWAIT 1 MIN", "idle", now, 4)
|
|
return None
|
|
self.fix_last[ident] = now
|
|
fix = self.cfg["fixes"][fix_id]
|
|
log.info("FORCE %s: %s", key_id, fix.get("label", fix_id))
|
|
self.note(key_id, "RESTARTING", "fixing", now, 180)
|
|
|
|
def work():
|
|
r = self.executor(self.cfg, fix_id, {})
|
|
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)
|
|
threading.Thread(target=work, daemon=True).start()
|
|
return fix_id
|
|
|
|
|
|
def display_level(res, acked, service_on):
|
|
if res.level == OFF:
|
|
return "off"
|
|
if res.level == OK:
|
|
return "ok"
|
|
if res.level == WARN:
|
|
return "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", "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 render(engine, service_on):
|
|
out = {}
|
|
for kid, key in engine.restart_keys.items():
|
|
note = engine.notes.get(kid)
|
|
if note and engine.now < note["until"]:
|
|
out[kid] = colored(key["label"] + "\n" + note["text"], note["level"])
|
|
else:
|
|
out[kid] = colored(key["label"] + "\nHOLD 2s", "idle")
|
|
for tile in engine.cfg["tiles"]:
|
|
tid, kind = tile["id"], tile.get("kind")
|
|
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] != "warn")
|
|
text = ("%d DOWN" % n_bad if n_bad else "%d WARN" % len(faults)) + "\n" + first
|
|
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, 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, 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
|
|
|
|
|
|
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()
|
|
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 cfg["tiles"] 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)})
|
|
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):
|
|
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
|
|
newest = max(vers, key=lambda d: tuple(int(x) for x in d[1:].split(".")))
|
|
return os.path.join(root, newest, "db.sqlite")
|
|
|
|
|
|
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 cfg["tiles"] 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)
|
|
|
|
|
|
def poll_midi(snap, cfg):
|
|
import coremidi
|
|
snap.midi, snap.midi_time = coremidi.sessions(), time.time()
|
|
remember_peers(snap, "", snap.midi)
|
|
for t in cfg["tiles"]:
|
|
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 cfg["tiles"] 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 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):
|
|
"""One shell line that runs every check and prints SP<n>=<exit code>."""
|
|
return "; ".join("(%s) >/dev/null 2>&1; echo SP%d=$?" % (cmd, i) for i, cmd in enumerate(cmds))
|
|
|
|
|
|
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 poll_remote(snap, cfg):
|
|
by_host = {}
|
|
for t in cfg["tiles"]:
|
|
for c in t.get("checks", []):
|
|
if c["type"] == "remote":
|
|
by_host.setdefault(c["host"], [])
|
|
if c["cmd"] not in by_host[c["host"]]:
|
|
by_host[c["host"]].append(c["cmd"])
|
|
result = dict(snap.remote)
|
|
for host, cmds in by_host.items():
|
|
try:
|
|
out = subprocess.run(ssh_cmd(cfg, host) + [remote_script(cmds)], capture_output=True,
|
|
text=True, timeout=20).stdout
|
|
except subprocess.TimeoutExpired:
|
|
out = ""
|
|
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 hyperdeck_probe(ip, timeout=3, port=9993):
|
|
"""Ask a HyperDeck directly what it is doing, on its control port 9993.
|
|
|
|
Returns its transport status ("record", "preview", "stopped", ...), "busy" if it
|
|
refused another client, "answered" if it greeted but did not report a status,
|
|
"hung" if it accepted the connection and said nothing at all, or "unreachable".
|
|
Only "hung" justifies a reboot.
|
|
"""
|
|
try:
|
|
s = socket.create_connection((ip, port), timeout=timeout)
|
|
except OSError:
|
|
return "unreachable"
|
|
buf = b""
|
|
with s:
|
|
s.settimeout(timeout)
|
|
try:
|
|
s.sendall(b"transport info\r\n")
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
chunk = s.recv(4096)
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
m = re.search(rb"208 transport info:.*?status: (\w+)", buf, re.S)
|
|
if m:
|
|
return m.group(1).decode().lower()
|
|
if b"120 connection rejected" in buf:
|
|
return "busy"
|
|
except socket.timeout:
|
|
pass
|
|
except OSError:
|
|
return "unreachable" if not buf else "answered"
|
|
return "answered" if buf else "hung"
|
|
|
|
|
|
# --- 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 cfg["tiles"]:
|
|
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)
|
|
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
|
|
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 == "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")
|
|
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 == "hyperdeck":
|
|
port = chk.get("port", 9993)
|
|
verdict = hyperdeck_probe(chk["ip"], port=port)
|
|
if verdict == "hung":
|
|
with socket.create_connection((chk["ip"], port), timeout=3) as s:
|
|
s.sendall(b"reboot\r\n")
|
|
log.info("FIX %s %s -> deck accepted a connection but did not answer: reboot sent",
|
|
label, chk["conn"])
|
|
return True
|
|
if verdict == "unreachable":
|
|
log.info("FIX %s %s -> deck does not answer on the network: nothing sent", label, chk["conn"])
|
|
return False
|
|
# The deck answers (and may be recording): leave it alone, reconnect Companion instead.
|
|
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 = chk["conn"] in ids and comp.restart_connection(ids[chk["conn"]])
|
|
log.info("FIX %s %s -> deck is fine (%s): Companion connection restarted %s",
|
|
label, chk["conn"], verdict, "ok" if ok else "FAILED")
|
|
return bool(ok)
|
|
log.warning("FIX %s: unknown kind %r", label, kind)
|
|
except Exception as e:
|
|
log.warning("FIX %s failed: %s", label, e)
|
|
return False
|
|
|
|
|
|
def reconnect_and_verify(cfg, fix, label, sleep=time.sleep):
|
|
"""Restart one Companion connection, 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
|
|
(its control side has hung - only a power-cycle helps), else that it is not on the network."""
|
|
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 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 = 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"
|
|
|
|
|
|
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 load_config(path):
|
|
with open(path) as f:
|
|
cfg = json.load(f)
|
|
ids = set()
|
|
for t in cfg["tiles"]:
|
|
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 cfg["tiles"]:
|
|
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" and c["host"] not in cfg.get("hosts", {}):
|
|
raise ValueError("tile %s: unknown host %s" % (t["id"], c["host"]))
|
|
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 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["fix"] not in cfg.get("fixes", {}):
|
|
raise ValueError("restart key %s: unknown fix %s" % (k["id"], k["fix"]))
|
|
return cfg
|
|
|
|
|
|
def ping_hosts(cfg, snap=None):
|
|
hosts = [ping_target(c, snap) if snap else c.get("host")
|
|
for t in cfg["tiles"] for c in t.get("checks", []) if c["type"] == "ping"]
|
|
hosts = [h for h in hosts if h]
|
|
gw = cfg.get("network", {}).get("gateway")
|
|
return list(dict.fromkeys(hosts + ([gw] if gw else [])))
|
|
|
|
|
|
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 cfg["tiles"]:
|
|
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("pings", 3.0, lambda: poll_pings(snap, ping_hosts(cfg, snap), iface), stop),
|
|
Periodic("midi", 2.0, 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", 10.0, lambda: poll_remote(snap, cfg), stop),
|
|
Periodic("remote-midi", 5.0, lambda: poll_remote_midi(snap, cfg), stop),
|
|
Periodic("dante-map", 30.0, lambda: poll_dante_map(snap, cfg, c4, dante_map_path(data_dir)), stop),
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
if any(c["type"] == "artnet" for t in cfg["tiles"] for c in t.get("checks", [])):
|
|
start_artnet_listener(snap, stop)
|
|
return threads
|
|
|
|
|
|
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)
|
|
else:
|
|
log.warning("unknown command %r", part)
|
|
|
|
|
|
def run(cfg, data_dir, dry_run=False, stop=None):
|
|
stop = stop or threading.Event()
|
|
names = cfg["variables"]
|
|
comp = make_companion(cfg)
|
|
snap = Snapshot()
|
|
engine = Engine(cfg, os.path.join(data_dir, "service-mode.json"))
|
|
snap.dante_map = load_dante_map(dante_map_path(data_dir))
|
|
threads = start_pollers(cfg, snap, stop, data_dir)
|
|
stop.wait(2.5) # let the first polls land before judging anything
|
|
last_json, last_write, last_eval = None, 0.0, 0.0
|
|
down_since, alerted, warned_missing = None, False, set()
|
|
log.info("status panel running (%s)", "dry run" if dry_run else "live")
|
|
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 >= 1.0:
|
|
last_eval = now
|
|
tiles = engine.step(snap, now)
|
|
for kind, text in engine.events:
|
|
log.info("%s %s", kind, text)
|
|
engine.events = []
|
|
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"), "NA Ingest: Companion is answering again.")
|
|
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 > 30:
|
|
post_webhook(cfg.get("alert_webhook"), "NA Ingest: Companion has not answered for 30 seconds.")
|
|
alerted = True
|
|
stop.wait(0.25)
|
|
for t in threads:
|
|
t.join(timeout=10)
|
|
|
|
|
|
def run_once(cfg):
|
|
"""Poll everything synchronously, judge with no debounce, print the tiles."""
|
|
snap = Snapshot()
|
|
comp = make_companion(cfg)
|
|
fast, slow = needed_variables(cfg)
|
|
poll_connections(comp, snap)
|
|
poll_variables(comp, snap, fast + slow)
|
|
poll_pings(snap, ping_hosts(cfg), cfg.get("network", {}).get("interface"))
|
|
poll_pings(snap, ping_hosts(cfg), 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)
|
|
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"))
|
|
args = ap.parse_args(argv)
|
|
cfg = load_config(args.config)
|
|
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)
|
|
run(cfg, args.data_dir, dry_run=args.dry_run)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|