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

READY page: LED CROSS checks the Resolume columns Sunday's slides call up

Browse Source 
- For every item in the plan's ProPresenter playlist: the macros its slides fire (read on the
  ProP Mac from Configuration/Macros and the .pro files), the RossTalk "CC page/row/col" each
  macro sends to Companion, the Companion buttons that one presses in turn, down to Resolume's
  /composition/columns/N/connect. Each such column must hold a clip on some layer (one whose
  ignore-column-trigger is not On), and Resolume must be on the Service deck.
- COL N EMPTY / COL N MISSING name the column and the items that use it. The hold only puts
  Resolume back on its deck; it never plays a clip. "Something is playing" is no longer checked.
- ready.prop_host replaces ready.led_clip.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-15 14:43:49 -04:00
co-authored by Claude Opus 5.5
parent 5adab6c518
commit c9ffbc6828
6 changed files with 264 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
readiness.py
+195 -64
View File
@@ -15,14 +15,15 @@ snap.ready, which the READY page keys draw (check type "ready" in status_panel.p
the tracks play at that song's start (worked out from the set's LTC clip and tempo map)
backgrounds every song's slides have a background (ProPresenter labels a slide with
its media's file name)
led_cross Resolume is on the Service deck and something is playing on the cross
led_cross every Resolume column Sunday's slides call up holds content, on the Service deck
(slide -> macro -> RossTalk "CC page/row/col" to Companion -> Resolume column)
lineup the Band Lineup board (ProdDash) shows this plan and exactly the people on it
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
What can be put right is put right by a hold (fix kind "ready", run_action below): re-sync
the lineup from PCO, put the cross background back up, copy the tracks to a location that
the lineup from PCO, put Resolume back on its deck, copy the tracks to a location that
has none, or simply look again now.
Nothing here writes to Companion, ProPresenter or Ableton's set; the Ableton set is read
@@ -52,8 +53,6 @@ MEDIA = re.compile(r"\.(mp4|mov|m4v|mxf|mpg|mpeg|avi|webm|jpg|jpeg|png|gif|heic|
WAKE = threading.Event() # set by a hold: look again now instead of waiting for the interval
LAST = {} # the latest look (run_action needs the plan's date)
MEMORY = {} # what is worth remembering across looks: the cross's clip ("led_clip")
MEMORY_FILE = "ready-memory.json"
class ReadyError(Exception):
@@ -291,7 +290,7 @@ def prop_look(rc, plan):
try:
p = (http_json(base + "/v1/presentation/" + uuid, timeout=6) or {}).get("presentation") or {}
slides = [s for g in p.get("groups") or [] for s in g.get("slides") or []]
cache[uuid] = {"name": (p.get("id") or {}).get("name"), "type": p.get("type"),
cache[uuid] = {"name": (p.get("id") or {}).get("name"), "type": p.get("type"), "path": p.get("presentation_path"),
"timeline": bool(p.get("has_timeline")), "cues": p.get("total_cues") or 0,
"slides": [{"enabled": s.get("enabled", True), "label": s.get("label") or "",
"text": s.get("text") or "", "notes": s.get("notes") or ""} for s in slides]}
@@ -735,31 +734,181 @@ def eval_lineup(rc, plan, state):
return line
# --- Resolume (the LED cross) ------------------------------------------------------------------
# --- Resolume: the columns Sunday's slides call up -------------------------------------------
# A ProPresenter slide fires a macro (the .pro holds the macro's UUID); the macro sends RossTalk
# "CC page/row/column" to Companion; that Companion button (and the buttons it presses in turn)
# sends Resolume "/composition/columns/N/connect". Every column found that way must hold content.
def connected_clips(comp):
"""[(layer number, clip number, name)] of every clip playing in the composition (1-based, as the API wants)."""
out = []
for li, layer in enumerate(comp.get("layers") or []):
for ci, clip in enumerate(layer.get("clips") or []):
state = ((clip.get("connected") or {}).get("value") or "")
if state.startswith("Connected"):
out.append((li + 1, ci + 1, ((clip.get("name") or {}).get("value") or "")))
MACRO_SCRIPT = r"""
import json, os, subprocess
P = json.loads(%r)
def show_dir():
try:
d = subprocess.run(["defaults", "read", "com.renewedvision.propresenter", "applicationShowDirectory"],
capture_output=True, text=True, timeout=5).stdout.strip()
if d:
return os.path.expanduser(d)
except Exception:
pass
return os.path.expanduser("~/Documents/ProPresenter")
def varint(b, i):
s = r = 0
while True:
x = b[i]; i += 1; r |= (x & 0x7f) << s
if not x & 0x80:
return r, i
s += 7
def fields(b):
out, i = [], 0
while i < len(b):
k, i = varint(b, i); num, wt = k >> 3, k & 7
if wt == 0: v, i = varint(b, i)
elif wt == 1: v, i = b[i:i + 8], i + 8
elif wt == 5: v, i = b[i:i + 4], i + 4
elif wt == 2:
n, i = varint(b, i); v = b[i:i + n]; i += n
else: raise ValueError(wt)
out.append((num, wt, v))
return out
def get(b, *path):
for p in path:
if not isinstance(b, bytes):
return None
f = [v for n, w, v in fields(b) if n == p]
if not f:
return None
b = f[0]
return b if isinstance(b, bytes) else None
macros, raw = {}, {}
for num, wt, v in fields(open(os.path.join(show_dir(), "Configuration", "Macros"), "rb").read()):
if num != 2 or wt != 2:
continue
try:
uuid = get(v, 1, 1).decode(); name = (get(v, 2) or b"").decode()
except Exception:
continue
cc = []
for n, w, act in fields(v):
comm = get(act, 32) if n == 4 and w == 2 else None
if comm is None:
continue
dev, fmt = (get(comm, 1, 2) or b"").decode(), (get(comm, 2) or b"").decode()
for n2, w2, cmd in fields(comm):
if n2 == 4 and w2 == 2:
cc.append([dev, fmt, (get(cmd, 2) or b"").decode()])
macros[uuid] = {"name": name, "cc": cc}
raw[uuid] = v
for u, v in raw.items(): # a macro that runs other macros
macros[u]["calls"] = [o for o in raw if o != u and o.encode() in v]
used = {}
for p in P["paths"]:
try:
b = open(p, "rb").read()
except OSError:
used[p] = None
continue
used[p] = [u for u in macros if u.encode() in b]
print(json.dumps({"macros": macros, "used": used}))
"""
COLUMN = re.compile(r"/composition/columns/(\d+)/connect")
LOCATION = re.compile(r"^\s*(\d+)/(\d+)/(\d+)\s*$")
PRESSES = ("button_pressrelease", "button_press", "button_release", "button_pressrelease_condition")
def eval_resolume(rc, comp, memory):
def companion_page(cfg, n):
"""One Companion page as Companion exports it (read-only)."""
comp = cfg.get("companion") or {}
url = "http://%s:%d/int/export/page/%d?includeSecrets=false" % (comp.get("host", "127.0.0.1"), comp.get("port", 8000), n)
with urllib.request.urlopen(url, timeout=6) as r:
raw = r.read()
try:
raw = __import__("gzip").decompress(raw)
except OSError:
pass
return json.loads(raw).get("page") or {}
def button_columns(get_page, location, seen=None):
"""The Resolume columns a Companion button connects, following the buttons it presses."""
seen = set() if seen is None else seen
m = LOCATION.match(location or "")
if not m or location in seen or len(seen) > 40:
return set()
seen.add(location)
page, row, col = (int(x) for x in m.groups())
ctl = (((get_page(page) or {}).get("controls") or {}).get(str(row)) or {}).get(str(col)) or {}
cols = set()
def walk(actions):
for a in actions or []:
opts = {k: (v.get("value") if isinstance(v, dict) else v) for k, v in (a.get("options") or {}).items()}
for v in opts.values():
for hit in COLUMN.findall(str(v)):
cols.add(int(hit))
if "olumn" in str(a.get("definitionId")) and str(opts.get("column", "")).isdigit():
cols.add(int(opts["column"]))
if a.get("connectionId") == "internal" and a.get("definitionId") in PRESSES:
cols.update(button_columns(get_page, str(opts.get("location", "")), seen))
for kids in (a.get("children") or {}).values():
walk(kids)
for step in (ctl.get("steps") or {}).values():
for actions in (step.get("action_sets") or {}).values():
walk(actions)
return cols
def macro_columns(macros, uuid, get_page, seen=None):
"""The columns one ProPresenter macro calls up (through the Companion buttons it presses)."""
seen = set() if seen is None else seen
if uuid in seen or uuid not in macros:
return set()
seen.add(uuid)
cols = set()
for dev, fmt, value in macros[uuid].get("cc") or []:
if fmt.strip().upper().startswith("CC"):
cols |= button_columns(get_page, value.replace(":", "/"))
for other in macros[uuid].get("calls") or []:
cols |= macro_columns(macros, other, get_page, seen)
return cols
def column_has_content(comp, n):
"""Column n (1-based) holds a clip on some layer that a column trigger plays."""
for layer in comp.get("layers") or []:
clips = layer.get("clips") or []
if n - 1 < len(clips):
c = clips[n - 1]
ignore = (c.get("ignorecolumntrigger") or {}).get("value") # "Layer Determined" / "On" / "Off"
if ((c.get("connected") or {}).get("value") or "Empty") != "Empty" and ignore not in ("On", True):
return True
return False
def eval_columns(rc, prop, found, get_page, comp):
"""Every Resolume column the plan's ProPresenter items call up holds content, on the right deck."""
line = Line()
deck = next((d for d in comp.get("decks") or [] if (d.get("selected") or {}).get("value")), None)
want = rc.get("resolume_deck", "")
if want and (deck is None or ((deck.get("name") or {}).get("value") or "") != want):
line.fault(ERR, "WRONG DECK", "Resolume is on deck %s, not %s" % (((deck or {}).get("name") or {}).get("value"), want))
playing = connected_clips(comp)
if not playing:
line.fault(ERR, "NOTHING ON", "no clip is playing in %s" % ((comp.get("name") or {}).get("value") or "the composition"))
else:
memory["led_clip"] = playing[0][2]
line.info = short(playing[0][2], 30)
line.fault(ERR, "WRONG DECK", "Resolume is on deck %s, not %s: the columns would be another deck's" % (
((deck or {}).get("name") or {}).get("value"), want))
macros, used = found.get("macros") or {}, found.get("used") or {}
wanted = {} # column -> [items that call it up]
for it in prop.get("items") or []:
path = (it.get("pres") or {}).get("path")
for uuid in used.get(path) or []:
for col in macro_columns(macros, uuid, get_page):
wanted.setdefault(col, []).append(it["name"])
columns = comp.get("columns") or []
for col in sorted(wanted):
name = ((columns[col - 1].get("name") or {}).get("value") if col <= len(columns) else "") or ""
who = ", ".join(dict.fromkeys(wanted[col]))
if col > len(columns):
line.fault(ERR, "COL %d MISSING" % col, "%s call up column %d; Resolume has %d columns" % (who, col, len(columns)))
elif not column_has_content(comp, col):
line.fault(ERR, "COL %d EMPTY" % col, "column %d (%s) is empty in Resolume; %s call it up" % (col, name, who))
line.info = "%d COLUMNS OK" % len(wanted) if wanted else "NO COLUMNS USED"
return line
@@ -812,25 +961,6 @@ class Line:
"info": self.info}
def load_memory(data_dir):
try:
return json.load(open(os.path.join(data_dir, MEMORY_FILE))) if data_dir else {}
except (OSError, ValueError):
return {}
def save_memory(data_dir, memory):
if not data_dir:
return
try:
tmp = os.path.join(data_dir, MEMORY_FILE + ".tmp")
with open(tmp, "w") as f:
json.dump(memory, f)
os.replace(tmp, os.path.join(data_dir, MEMORY_FILE))
except OSError as e:
log.warning("%s: %s", MEMORY_FILE, e)
def look(cfg, now=None, data_dir=None, sources=None):
"""Everything the READY page shows, as {"time", "plan", "items": {item: {"faults", "details", "info"}}}.
`sources` replaces the readers (tests)."""
@@ -840,12 +970,12 @@ def look(cfg, now=None, data_dir=None, sources=None):
ableton=lambda plan: ableton_look(cfg, rc, plan),
lineup=lambda: http_json(rc.get("lineup_url", "").rstrip("/") + "/state", timeout=6),
resolume=lambda: http_json(rc.get("resolume_url", "").rstrip("/") + "/api/v1/composition", timeout=6),
macros=lambda paths: run_remote(cfg, rc["prop_host"], MACRO_SCRIPT % json.dumps({"paths": paths}), timeout=60),
companion=lambda n: companion_page(cfg, n),
prop_bundle=lambda: run_remote(cfg, rc["prop_bundle_host"], BUNDLE_SCRIPT % json.dumps(
{"folder": rc.get("prop_bundle_folder", ""), "pattern": rc.get("prop_bundle_pattern", "*.probundle")}), timeout=30))
src.update(sources or {})
lines = {}
if not MEMORY:
MEMORY.update(load_memory(data_dir))
def failed(names, err):
msg = str(err) if isinstance(err, ReadyError) else "CHECK FAILED"
@@ -862,13 +992,31 @@ def look(cfg, now=None, data_dir=None, sources=None):
("lights", rc.get("tracks_host")), ("prop", rc.get("prop_url")),
("timecode", rc.get("prop_url")), ("backgrounds", rc.get("prop_url")),
("lineup", rc.get("lineup_url")), ("mics", True),
("track_bundle", rc.get("locations")), ("prop_bundle", rc.get("prop_bundle_host"))) if on], e)
("track_bundle", rc.get("locations")), ("prop_bundle", rc.get("prop_bundle_host")),
("led_cross", rc.get("resolume_url") and rc.get("prop_host"))) if on], e)
if plan is not None:
prop = None
if rc.get("prop_url"):
try:
lines.update(eval_prop(rc, plan, src["prop"](plan)))
prop = src["prop"](plan)
lines.update(eval_prop(rc, plan, prop))
except Exception as e: # noqa: BLE001
failed(["prop", "timecode", "backgrounds"], e)
failed(["prop", "timecode", "backgrounds"] + (["led_cross"] if rc.get("resolume_url") and rc.get("prop_host") else []), e)
if rc.get("resolume_url") and rc.get("prop_host") and prop is not None:
try:
pages = {}
def get_page(n):
if n not in pages:
try:
pages[n] = src["companion"](n)
except (OSError, ValueError):
pages[n] = {}
return pages[n]
paths = sorted({(it.get("pres") or {}).get("path") for it in prop["items"]} - {None})
lines["led_cross"] = eval_columns(rc, prop, src["macros"](paths), get_page, src["resolume"]())
except Exception as e: # noqa: BLE001
failed(["led_cross"], e if isinstance(e, ReadyError) else ReadyError("RESOLUME UNREACHABLE"))
if rc.get("tracks_host"):
try:
al = src["ableton"](plan)
@@ -895,14 +1043,6 @@ def look(cfg, now=None, data_dir=None, sources=None):
lines["prop_bundle"] = eval_prop_bundle(rc, src["prop_bundle"](), week_start(now, rc.get("reset_day", "mon")))
except Exception as e: # noqa: BLE001
failed(["prop_bundle"], e)
if rc.get("resolume_url"):
try:
before = dict(MEMORY)
lines["led_cross"] = eval_resolume(rc, src["resolume"](), MEMORY)
if MEMORY != before:
save_memory(data_dir, MEMORY)
except Exception as e: # noqa: BLE001
failed(["led_cross"], e if isinstance(e, ReadyError) else ReadyError("RESOLUME UNREACHABLE"))
out = {"time": now, "items": {k: v.as_dict() for k, v in lines.items()}}
if plan is not None:
out["plan"] = {"id": plan["id"], "title": plan["title"], "dates": plan["dates"], "time": plan["time"]}
@@ -998,16 +1138,7 @@ def run_action(cfg, action, chk=None, sleep=time.sleep):
log.info("FIX Resolume deck %s selected", want)
sleep(1)
comp = http_json(base, timeout=6) or {}
if connected_clips(comp):
return True
name = rc.get("led_clip") or MEMORY.get("led_clip") # set in config, else the last one seen playing
for li, layer in enumerate(comp.get("layers") or []):
for ci, clip in enumerate(layer.get("clips") or []):
if name and ((clip.get("name") or {}).get("value") or "") == name:
http_json("%s/layers/%d/clips/%d/connect" % (base, li + 1, ci + 1), method="POST", timeout=6)
log.info("FIX LED cross: %s playing again (layer %d clip %d)", name, li + 1, ci + 1)
return True
return "NO CLIP KNOWN"
return True # an empty column is filled by hand in Resolume
if action == "ltc_to_pco":
return write_ltc(rc, LAST.get("ltc") or [], LAST.get("service_type"), LAST.get("plan_id"))
if action == "copy_tracks":