christianandClaude Opus 5.5
b8748be655
STUDIO-A-Ingest now maps to a local dvs_ingest fix (same helper as the other Macs, which NA Ingest also needs). A hold on a failing button that has no automatic fix now shows NO AUTO FIX instead of NOTHING TO FIX. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1370 lines
56 KiB
Python
1370 lines
56 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?value=%s" % (
|
|
urllib.parse.quote(name, safe=""), urllib.parse.quote(value, safe=""))
|
|
status, text = self._request("POST", path)
|
|
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
|
|
|
|
|
|
# --- 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.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)
|
|
|
|
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):
|
|
return [("conn", conn_result(snap, chk["conn"], ctx["now"], chk.get("service_only", False)))]
|
|
|
|
|
|
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 check_ping(chk, snap, ctx):
|
|
fails = snap.ping.get(chk["host"])
|
|
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):
|
|
if not snap.midi_time:
|
|
return [("peer", Result(OK))]
|
|
sess = snap.midi.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 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,
|
|
}
|
|
|
|
# Checks whose connection a long press may restart (Dante is shared, so not that one).
|
|
RESTARTABLE = ("conn", "hyperdeck", "shure", "psm", "tascam")
|
|
|
|
|
|
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")]
|
|
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.notes = {} # tile id -> hold feedback {"text", "level", "until"}
|
|
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)))
|
|
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
|
|
fix_id = chk.get("fix")
|
|
if isinstance(fix_id, dict):
|
|
fix_id = fix_id.get(sub)
|
|
if fix_id:
|
|
if chk["type"] == "hyperdeck" and hyperdeck_recording(snap, chk["conn"]):
|
|
continue
|
|
# One run per fix, except a HyperDeck reboot, which is per deck.
|
|
per_deck = self.cfg["fixes"][fix_id].get("kind") == "hyperdeck_reboot"
|
|
ident = ("fix", fix_id, chk.get("ip", "") if per_deck else "")
|
|
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)
|
|
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 == "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")
|
|
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((what, chk))
|
|
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 = []
|
|
|
|
def work(fix_id, chk):
|
|
ok = self.executor(self.cfg, fix_id, chk)
|
|
results.append(ok is not False)
|
|
if len(results) == len(started): # the last one to finish reports for all
|
|
done = time.time()
|
|
if all(results):
|
|
self.note(tile_id, "CHECKING\n" + summary, "fixing", done, 20)
|
|
else:
|
|
self.note(tile_id, "FIX FAILED\n" + summary, "failed", done, 20)
|
|
for fix_id, chk in started:
|
|
threading.Thread(target=work, args=(fix_id, chk), daemon=True).start()
|
|
return plan
|
|
|
|
|
|
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 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 poll_midi(snap, cfg):
|
|
import coremidi
|
|
snap.midi, snap.midi_time = coremidi.sessions(), time.time()
|
|
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 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", "StrictHostKeyChecking=accept-new", "-o", "ControlMaster=auto",
|
|
"-o", "ControlPath=/tmp/status-panel-ssh-%C", "-o", "ControlPersist=120", cfg["hosts"][host]["ssh"]]
|
|
|
|
|
|
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 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:]
|
|
log.info("FIX %s -> %s%s", label, "ok" if r.returncode == 0 else "exit %d" % r.returncode,
|
|
(": " + detail) if detail else "")
|
|
return r.returncode == 0
|
|
if kind == "hyperdeck_reboot":
|
|
with socket.create_connection((chk["ip"], 9993), timeout=3) as s:
|
|
s.settimeout(3)
|
|
try:
|
|
s.recv(1024) # the deck's greeting
|
|
except socket.timeout:
|
|
pass
|
|
s.sendall(b"reboot\n")
|
|
log.info("FIX %s %s -> reboot sent", label, chk["conn"])
|
|
return True
|
|
log.warning("FIX %s: unknown kind %r", label, kind)
|
|
except Exception as e:
|
|
log.warning("FIX %s failed: %s", label, e)
|
|
return False
|
|
|
|
|
|
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 cfg.get("fixes", {}):
|
|
raise ValueError("tile %s: unknown fix %s" % (t["id"], f))
|
|
if f 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"]))
|
|
return cfg
|
|
|
|
|
|
def ping_hosts(cfg):
|
|
hosts = [c["host"] for t in cfg["tiles"] for c in t.get("checks", []) if c["type"] == "ping"]
|
|
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):
|
|
fast, slow = needed_variables(cfg)
|
|
c1, c2, c3 = make_companion(cfg), make_companion(cfg), make_companion(cfg)
|
|
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("pings", 3.0, lambda: poll_pings(snap, ping_hosts(cfg), 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),
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
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)
|
|
|
|
|
|
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 == "restart": # sent by a 2 second hold
|
|
engine.run_fixes(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"))
|
|
threads = start_pollers(cfg, snap, stop)
|
|
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()
|