christianandClaude Opus 5.5
340d28263a
Tested live: restarting rtpMIDIService brought the Lighting PC's Bonjour advert back but dropped it from the NA Ingest and NA Tracks sessions, and it did not rejoin them. macOS only exposes the default network session to scripts, not NA Ingest's named one, so there is no safe automatic reconnect. MIDI now has no hold fix; the README says how to reconnect by hand. ssh fixes run with LogLevel=ERROR so warnings stay out of the log. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1638 lines
69 KiB
Python
1638 lines
69 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
|
|
|
|
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.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):
|
|
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 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:
|
|
return [("artnet", Result(ERR, chk.get("msg", "ART-NET SILENT"), named=True))]
|
|
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"}
|
|
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)))
|
|
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).
|
|
per_deck = self.cfg["fixes"][fix_id].get("kind") == "hyperdeck"
|
|
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)
|
|
ident = ("fix", fix_id, chk["conn"] 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 == "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, {})))))
|
|
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)))
|
|
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(fn, args):
|
|
ok = fn(*args)
|
|
results.append(ok)
|
|
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 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)
|
|
for fn, args in started:
|
|
threading.Thread(target=work, args=(fn, args), 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", "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 "")
|
|
if r.returncode == 0 and fix.get("reconnect") and chk.get("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 = chk["conn"] in ids and comp.restart_connection(ids[chk["conn"]])
|
|
log.info("FIX %s -> Companion connection %s restarted %s", label, chk["conn"], "ok" if ok else "FAILED")
|
|
return r.returncode == 0
|
|
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 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"]))
|
|
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, 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("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),
|
|
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)
|
|
|
|
|
|
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"))
|
|
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()
|