christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

READY: RESI VOX - the Resi song goes out with its lead singer's mic

Browse Source 
PCO's song description names the lead ('Genesis lead', RESI marks the song, else the last before the
message), the lineup screen gives their VOX, and the Resi AJA's Lead Vocal channel (21) is read from
the card with Dante's own read-only query. Wrong: red, and the hold / auto-fix presses that VOX's AJA
button on page 13 and reads the card back. Two leads, a lead not on the lineup, a PCO Audio note naming
another VOX, or VOX 5+ are flagged only. Checked on 2026-10-25's real data: Genesis, VOX 4, AJA on VOX 4;
the Audio note says V2, so it is flagged and nothing switches. Applied on NA after the services.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-25 07:30:36 -04:00
co-authored by Claude Opus 5.5
parent 0ecec5f112
commit 32e071e169
8 changed files with 330 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
readiness.py
+197 -1
View File
@@ -30,6 +30,10 @@ snap.ready, which the READY page keys draw (check type "ready" in status_panel.p
mics (a tap-to-confirm key) which mics this weekend needs, from PCO
track_bundle every location has this weekend's Ableton project folder on the shared drive
prop_bundle a ProPresenter bundle was saved to the locations' folder this week
resi_vox the Resi song (marked RESI in PCO, else the last song before the message) goes out to the locations
with its lead singer's mic: the lead from PCO ("Genesis lead"), their VOX number from the lineup
screen, and what the Resi AJA's "Lead Vocal" channel really receives, read from the card over Dante.
The hold presses that VOX's button on Companion's page 13.
start every service PCO lists this week (Sunday and special services such as First / Together
Tuesday) has its Ableton set; the hold builds the missing ones (ableton_start.py)
@@ -1750,7 +1754,7 @@ def look(cfg, now=None, data_dir=None, sources=None):
macros=lambda paths: run_remote(cfg, rc["prop_host"], MACRO_SCRIPT % json.dumps({"paths": paths}), timeout=60),
companion=lambda n: companion_page(cfg, n), live=lambda: live_look(cfg, rc),
hits=lambda al: light_hits(cfg, rc, al), services=lambda: upcoming_services(rc, now),
start=lambda svcs: start_look(cfg, rc, svcs),
start=lambda svcs: start_look(cfg, rc, svcs), resi_now=lambda: resi_now(cfg, rc.get("resi_vox") or {}),
prop_bundle=lambda plan: run_remote(cfg, rc["prop_bundle_host"], BUNDLE_SCRIPT % json.dumps(
{"folder": prop_bundle_folder(rc, plan), "pattern": rc.get("prop_bundle_pattern", "*.probundle")}), timeout=30))
src.update(sources or {})
@@ -1856,6 +1860,16 @@ def look(cfg, now=None, data_dir=None, sources=None):
lines["start"] = eval_start(rc, services, src["start"](services), now)
except Exception as e: # noqa: BLE001
failed(["start"], e)
if rc.get("resi_vox"):
try:
try:
lineup = src["lineup"]() if rc.get("lineup_url") else None
except (OSError, ValueError):
lineup = None # PCO's team positions are used instead
now_vox, got = src["resi_now"]()
lines["resi_vox"] = eval_resi_vox(rc, plan, lineup, now_vox, got)[0]
except Exception as e: # noqa: BLE001
failed(["resi_vox"], e)
mics = Line()
mics.info = mics_needed(rc, plan)
lines["mics"] = mics
@@ -2083,6 +2097,186 @@ def eval_start(rc, services, look, now):
return line
# --- RESI VOX: the lead singer's mic to the locations ----------------------------------------------------
# The Resi song goes from NA to every other location with one vocal: the Resi AJA's "Lead Vocal" channel, which
# Companion's page 13 VOX1-4 AJA buttons subscribe to the wireless receiver's channel 1-4 (BOH-AD4Q-A). The lineup
# screen (ProdDash) says which VOX each singer is on; PCO's song description says who leads ("Genesis lead").
DANTE_ARC = 4440
def resi_song(plan):
"""The song that goes out over Resi: the one PCO marks RESI (description), else the last before the message."""
songs, last = [], None
for it in plan["items"]:
if it["type"] == "header" and "message" in it["title"].lower():
break
if it["type"] == "song" and it["position"] in ("during", ""):
songs.append(it)
last = it
marked = [it for it in plan["items"] if it["type"] == "song" and re.search(r"\bRESI\b", it.get("description") or "", re.I)]
return marked[-1] if marked else last
def song_leads(item):
""""Genesis lead" -> ["Genesis"]; "Mike and DJ lead" -> ["Mike", "DJ"]; nobody named -> []."""
for line in (item.get("description") or "").splitlines():
m = re.match(r"^\W*(.+?)\s+leads?\b", line.strip(), re.I)
if m:
return [n.strip() for n in re.split(r"\s*(?:,|&|\+|/|\band\b)\s*", m.group(1), flags=re.I) if n.strip()]
return []
def vox_of(label):
m = re.search(r"\bVOX\s*(\d+)", label or "", re.I)
return int(m.group(1)) if m else None
def lead_vox(name, lineup, team):
"""(VOX number, full name, where it came from) for a first name ("Genesis"): the lineup screen first (it is
what the band sees), else the plan's team positions; None when nobody or more than one person fits."""
def fits(full):
return bool(full) and norm(full).split()[:len(norm(name).split())] == norm(name).split()
slots = [s for s in ((lineup or {}).get("state") or {}).get("slots") or [] if s.get("person") and fits(s["person"].get("name"))]
voxed = [(vox_of(s.get("label")), s["person"]["name"]) for s in slots if vox_of(s.get("label"))]
if len({v for v, _ in voxed}) == 1:
return voxed[0][0], voxed[0][1], "lineup"
if voxed:
return None, ", ".join(n for _, n in voxed), "lineup"
people = [(vox_of(m.get("position")), m["name"]) for m in team or [] if fits(m.get("name")) and vox_of(m.get("position"))
and m.get("status") != "D"]
if len({v for v, _ in people}) == 1:
return people[0][0], people[0][1], "PCO"
return None, ", ".join(n for _, n in people), "PCO"
def dante_rx(ip, channel, timeout=2.0):
"""What a Dante device's receive channel is subscribed to, asked the way Dante Controller asks (read only):
{"name", "tx_channel", "tx_device", "status"}, or None if it does not answer."""
page = (channel - 1) // 16
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(timeout)
try:
seq = int(time.time() * 1000) & 0xffff
s.sendto(bytes.fromhex("27ff0010%04x3000" % seq + "0000000100%x10000" % page), (ip, DANTE_ARC))
d, _ = s.recvfrom(4096)
except OSError:
return None
finally:
s.close()
def cstr(o):
if not o or o >= len(d):
return ""
e = d.find(b"\0", o)
return d[o:e if e >= 0 else len(d)].decode("utf-8", "replace")
for i in range(16):
rec = d[12 + i * 20:32 + i * 20]
if len(rec) < 20:
break
num, _, _, tx_chan, tx_dev, rx_name, status, _, _, _ = struct.unpack(">10H", rec)
if num == channel:
return {"name": cstr(rx_name), "tx_channel": cstr(tx_chan), "tx_device": cstr(tx_dev), "status": status}
return None
def companion_var(cfg, conn, name):
comp = cfg.get("companion") or {}
url = "http://%s:%s/api/variable/%s/%s/value" % (comp.get("host", "127.0.0.1"), comp.get("port", 8000),
urllib.parse.quote(conn), urllib.parse.quote(name))
with urllib.request.urlopen(url, timeout=5) as r:
return r.read().decode("utf-8", "replace").strip()
def resi_now(cfg, rv):
"""The VOX number the Resi AJA's lead channel receives now (None: something else), and what it is."""
ip = companion_var(cfg, rv.get("dante_conn", "audinate-dantecontroller"), rv["aja"] + "_ip")
got = dante_rx(ip, rv.get("channel", 21)) if ip and ip != "Not found" else None
if got is None:
raise ReadyError("AJA NO ANSWER")
if got["tx_device"] != rv.get("mics", "BOH-AD4Q-A"):
return None, got
n = vox_of(got["tx_channel"])
if n is None:
try:
names = json.loads(companion_var(cfg, rv.get("dante_conn", "audinate-dantecontroller"), rv.get("mics", "BOH-AD4Q-A") + "_tx_names"))
n = names.index(got["tx_channel"]) + 1
except (ValueError, OSError):
n = None
return n, got
def eval_resi_vox(rc, plan, lineup, now_vox, now_got):
"""(Line, the VOX the lead is on or None): what the locations hear on the Resi song."""
line, rv = Line(), rc.get("resi_vox") or {}
buttons = {int(k) for k in (rv.get("buttons") or {})}
song = resi_song(plan)
if song is None:
line.fault(WARN, "NO RESI SONG", "PCO has no song before the message")
return line, None
leads = song_leads(song)
where = "%s (%s)" % (song["title"], "marked RESI" if re.search(r"\bRESI\b", song.get("description") or "", re.I)
else "the last song before the message")
if not leads:
line.fault(WARN, "NO LEAD " + short(song["title"], 12), "%s: PCO's description names no lead (\"Genesis lead\")" % where)
return line, None
found = [(n,) + lead_vox(n, lineup, plan.get("team")) for n in leads]
voxes = {v for _, v, _, _ in found if v}
for n, v, who, src in found:
if v is None:
line.fault(WARN, "%s NOT ON LINEUP" % short(n, 10), "%s leads %s, but %s" % (n, where, (
"more than one person fits: " + who) if who else "nobody of that name has a VOX on the lineup or in PCO"))
want = None
if len(voxes) == 1 and all(v for _, v, _, _ in found):
want = voxes.pop()
elif len(voxes) > 1:
line.fault(WARN, "%d LEADS: %s" % (len(found), " ".join(short(n, 6) for n, _, _, _ in found)),
"%s has %s leading (VOX %s): pick the one Resi gets by hand" % (
where, " and ".join(n for n, _, _, _ in found), ", ".join(str(v) for _, v, _, _ in found)))
audio = (song.get("notes") or {}).get(rv.get("note", "Audio"), "")
said = re.search(r"\bV(?:OX)?\s*(\d)\b", audio or "", re.I)
if want and said and int(said.group(1)) != want:
line.fault(WARN, "PCO SAYS V%s, LEAD V%d" % (said.group(1), want), "%s: the Audio note says %s, but %s is on VOX %d "
"on the lineup: check which is right (nothing is switched while they differ)" % (where, audio, found[0][2], want))
want = None
if want and buttons and want not in buttons:
line.fault(ERR, "LEAD ON VOX %d: NO BUTTON" % want, "%s is on VOX %d; page 13 only has VOX %s AJA" % (
found[0][2], want, ", ".join(str(b) for b in sorted(buttons))))
want = None
have = "VOX %d" % now_vox if now_vox else "%s @ %s" % (now_got.get("tx_channel") or "nothing", now_got.get("tx_device") or "-")
if want and now_vox != want:
line.fault(ERR, "RESI %s NOT VOX %d" % (("V%d" % now_vox) if now_vox else "MIC WRONG", want),
"%s: %s leads (VOX %d on the %s); the Resi AJA's lead channel gets %s" % (
where, found[0][2], want, found[0][3], have))
who = next(((w or n).split()[0] for n, v, w, _ in found if v and v == now_vox), "")
line.info = ("VOX %d %s" % (now_vox, short(who, 10))).strip() if now_vox else have[:16]
line.notes.append("%s: lead %s; AJA %s receives %s" % (where, ", ".join("%s = VOX %s (%s)" % (w or n, v, s) for n, v, w, s in found),
now_got.get("name"), have))
return line, want
def fix_resi_vox(cfg, rc, sleep=time.sleep):
"""Press the lead's VOX button on Companion (page 13), then read the AJA back."""
import status_panel
plan = pco_plan(rc, time.time())
rv = rc.get("resi_vox") or {}
lineup = http_json(rc.get("lineup_url", "").rstrip("/") + "/state", timeout=6) if rc.get("lineup_url") else None
now_vox, got = resi_now(cfg, rv)
line, want = eval_resi_vox(rc, plan, lineup, now_vox, got)
if not want:
return (line.faults[0][1] if line.faults else "NOTHING TO DO")[:20]
if now_vox == want:
return True
page, row, col = rv["buttons"][str(want)]
status = status_panel.make_companion(cfg).press(page, row, col)
log.info("FIX RESI VOX: VOX %s -> VOX %d (Companion %d/%d/%d: %s)", now_vox, want, page, row, col, status)
for _ in range(10):
sleep(1)
if resi_now(cfg, rv)[0] == want:
return True
return "STILL VOX %s" % (resi_now(cfg, rv)[0] or "?")
# --- Fixes (a hold on a READY key: fix kind "ready") ------------------------------------------------
def write_ltc(rc, songs, service_type, plan_id):
@@ -2170,6 +2364,8 @@ def run_action(cfg, action, chk=None, sleep=time.sleep):
return "NO AGENT" if r.get("no_agent") else "STILL NOT MOUNTED"
if action == "light_timing":
return fix_light_timing(cfg, rc)
if action == "resi_vox":
return fix_resi_vox(cfg, rc, sleep)
if action == "start_week":
if IS_LIVE[0]():
return "SERVICE ON"