RESI VOX reads what each page 13 VOX AJA button really sends (now DQ-Rack 33-36) from Companion's store; a source no button sends is MIC WRONG

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-27 12:53:17 -04:00
co-authored by Claude Opus 5.5
parent a4d5fbe1aa
commit 12a9e506e1
3 changed files with 78 additions and 2 deletions
+44 -1
View File
@@ -2409,12 +2409,55 @@ def companion_var(cfg, conn, name):
return r.read().decode("utf-8", "replace").strip()
def vox_sources(cfg, rv):
"""{VOX n: (Dante device, channel number)}: what each VOX AJA button on Companion really sends, read from
Companion's own store (2026-10-27 the buttons were moved from the AD4Q-A's 1-4 to the DQ-Rack's 33-36 and the
panel kept expecting the AD4Q-A). Empty when it cannot be read."""
import sqlite3
db = os.path.expanduser((cfg.get("companion") or {}).get("db") or "~/Library/Application Support/companion/v5.0/db.sqlite")
out = {}
try:
con = sqlite3.connect("file:%s?mode=ro" % urllib.parse.quote(db), uri=True, timeout=2)
try:
for n, (page, row, col) in (rv.get("buttons") or {}).items():
pg = con.execute("select value from pages where id=?", (int(page),)).fetchone()
cid = ((json.loads(pg[0]).get("controls") or {}).get(str(row)) or {}).get(str(col)) if pg else None
ctl = con.execute("select value from controls where id=?", (cid,)).fetchone() if cid else None
for step in (json.loads(ctl[0]).get("steps") or {}).values() if ctl else []:
for acts in (step.get("action_sets") or {}).values():
for a in acts or []:
o = a.get("options") or {}
dev = (o.get("sourceDevice") or {}).get("value")
if (o.get("destinationDevice") or {}).get("value") == rv["aja"] and dev:
ch = (o.get("sourceChannel_" + dev) or {}).get("value")
if ch:
out[int(n)] = (dev, int(ch))
finally:
con.close()
except (sqlite3.Error, ValueError, KeyError, TypeError, AttributeError):
return {}
return out
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")
conn = rv.get("dante_conn", "audinate-dantecontroller")
ip = companion_var(cfg, conn, 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")
every = vox_sources(cfg, rv)
srcs = {k: v for k, v in every.items() if v[0] == got["tx_device"]}
if every and not srcs:
return None, got # not what any VOX button sends: a hold presses the lead's
if srcs:
# The channel's number on its device: from the device's channel names, else the name's own number ("33 DQ-VOX 1").
try:
ch = json.loads(companion_var(cfg, conn, got["tx_device"] + "_tx_names")).index(got["tx_channel"]) + 1
except (ValueError, OSError):
m = re.match(r"^\s*(\d+)\b", got["tx_channel"] or "")
ch = int(m.group(1)) if m else None
return next((k for k, (_, c) in srcs.items() if c == ch), None), got
if got["tx_device"] != rv.get("mics", "BOH-AD4Q-A"):
return None, got
n = vox_of(got["tx_channel"])