prop_guard: keep a song's timeline running and the stage on Worship (Southcoast, live 2026-10-25)
SC loses lyrics and lights when timecode drops, and the first slide's macro (a communication action to Companion, which sets the look and stage layout) sometimes misses the stage layout. A small LaunchAgent on the ProPresenter Mac (org.waterschurch.prop-guard) watches ProPresenter's own API: a song's timeline that stops mid-song is started again at once (left alone if stopped again right after, or on the last slide), and in a song's first seconds the Stage Display is put on Worship if it is not on a worship layout. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+161
@@ -0,0 +1,161 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ProPresenter guard: keeps a song's lyrics (and the cues on its timeline) running and the band's stage display
|
||||
on the worship layout, fixing either at once. Runs on the ProPresenter Mac (LaunchAgent org.waterschurch.prop-guard),
|
||||
talking only to ProPresenter's own network API on this Mac. Stdlib only.
|
||||
|
||||
Built 2026-10-25 for Southcoast, where timecode sometimes drops out (lyrics and lights stop) and the first slide's
|
||||
macro does not always bring the stage display to the worship layout. A song is a presentation with a timeline
|
||||
(every song in the plan has one; nothing else does).
|
||||
|
||||
TIMECODE the song's timeline follows the tracks' LTC. When it was running and stops in the middle of the song
|
||||
(not on the last slide), the timeline is started again at once from where it stopped, so the lyrics and
|
||||
cues carry on; when the LTC comes back, ProPresenter follows it again. Logged with the time it stopped.
|
||||
STAGE a few seconds after a song becomes active (its first slide fired), the Stage Display must be on a
|
||||
layout whose name has "Worship" in it; if the macro did not get it there, it is set to "Worship".
|
||||
Only in the first seconds of a song: a layout someone picks later is left alone.
|
||||
|
||||
Status (for the status panel): ~/Library/Application Support/status-panel/prop-guard.json; log:
|
||||
~/Library/Logs/prop-guard.log.
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
CFG = {
|
||||
"api": "http://127.0.0.1:50098",
|
||||
"stage_screen": "Stage Display",
|
||||
"song_layout": "Worship", # set when the stage is on none of the "accept" layouts
|
||||
"accept": ["worship"], # a layout whose name contains one of these is fine for a song
|
||||
"stage_after": 2.5, # seconds after the song starts before the stage is judged
|
||||
"stage_window": 12.0, # ... and until when it is put right
|
||||
"resume": True, # start a stopped timeline again (False: only log)
|
||||
"stopped_for": 1.0, # seconds stopped before it counts (ProPresenter blinks while it seeks)
|
||||
"poll": 0.4,
|
||||
}
|
||||
STATUS = os.path.expanduser("~/Library/Application Support/status-panel/prop-guard.json")
|
||||
log = logging.getLogger("prop-guard")
|
||||
|
||||
|
||||
def api(path, timeout=2.0):
|
||||
try:
|
||||
with urllib.request.urlopen(CFG["api"] + path, timeout=timeout) as r:
|
||||
body = r.read()
|
||||
return json.loads(body) if body.strip() else {}
|
||||
except (urllib.error.URLError, OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
class Guard:
|
||||
def __init__(self, now=time.time):
|
||||
self.now = now
|
||||
self.song = None # uuid of the active song
|
||||
self.started = 0.0
|
||||
self.ran = False # its timeline has run (so a stop means the LTC went away)
|
||||
self.last_t = None
|
||||
self.stopped_at = None
|
||||
self.slides = 0
|
||||
self.stage_fixed = False
|
||||
self.resumed_at = None # when the guard last started the timeline again
|
||||
self.hands_off = False # stopped again right after a restart: someone means it, leave this song alone
|
||||
self.state = {"started": now(), "timecode_drops": [], "stage_fixes": [], "now": {}}
|
||||
|
||||
def stage(self):
|
||||
for m in api("/v1/stage/layout_map") or []:
|
||||
if (m.get("screen") or {}).get("name") == CFG["stage_screen"]:
|
||||
return m["screen"], m.get("layout") or {}
|
||||
return None, {}
|
||||
|
||||
def set_layout(self, screen):
|
||||
layouts = api("/v1/stage/layouts") or []
|
||||
want = next((l["id"] for l in layouts if l["id"]["name"] == CFG["song_layout"]), None)
|
||||
if not want:
|
||||
log.warning("STAGE: no layout called %s", CFG["song_layout"])
|
||||
return False
|
||||
return api("/v1/stage/screen/%s/layout/%s" % (screen["uuid"], want["uuid"])) is not None
|
||||
|
||||
def tick(self):
|
||||
now = self.now()
|
||||
pres = (api("/v1/presentation/active") or {}).get("presentation")
|
||||
if not pres or not pres.get("has_timeline"):
|
||||
if self.song:
|
||||
log.info("SONG over: %s", self.name)
|
||||
self.song = None
|
||||
self.state["now"] = {"song": None, "time": now}
|
||||
return
|
||||
uuid, name = pres["id"]["uuid"], pres["id"]["name"]
|
||||
if uuid != self.song:
|
||||
self.song, self.name, self.started, self.ran, self.last_t, self.stopped_at = uuid, name, now, False, None, None
|
||||
self.stage_fixed, self.resumed_at, self.hands_off = False, None, False
|
||||
self.slides = sum(len(g.get("slides") or []) for g in pres.get("groups") or [])
|
||||
log.info("SONG %s (%d slides)", name, self.slides)
|
||||
tl = api("/v1/presentation/active/timeline") or {}
|
||||
running, t = bool(tl.get("is_running")), tl.get("current_time")
|
||||
idx = (api("/v1/presentation/slide_index") or {}).get("presentation_index") or {}
|
||||
slide = idx.get("index")
|
||||
at_end = slide is not None and self.slides and slide >= self.slides - 1
|
||||
if running:
|
||||
if not self.ran:
|
||||
log.info("TIMECODE running in %s at %.1f s", name, t or 0)
|
||||
self.ran, self.stopped_at = True, None
|
||||
elif self.ran and not at_end and not self.hands_off:
|
||||
self.stopped_at = self.stopped_at or now
|
||||
if self.resumed_at and now - self.resumed_at < 5:
|
||||
self.hands_off = True # it was stopped again at once: an operator, not the LTC
|
||||
log.info("TIMECODE: %s stopped again right after it was started: left alone for this song", name)
|
||||
elif now - self.stopped_at >= CFG["stopped_for"]:
|
||||
drop = {"song": name, "at": t, "slide": slide, "time": now}
|
||||
if CFG["resume"]:
|
||||
ok = api("/v1/presentation/active/timeline/play") is not None
|
||||
after = api("/v1/presentation/active/timeline") or {}
|
||||
drop["resumed"] = bool(ok and after.get("is_running"))
|
||||
self.resumed_at = now
|
||||
log.warning("TIMECODE DROPPED in %s at %.2f s (slide %s)%s", name, t or 0, slide,
|
||||
"" if not CFG["resume"] else (": timeline started again" if drop.get("resumed") else ": COULD NOT START IT"))
|
||||
self.state["timecode_drops"] = (self.state["timecode_drops"] + [drop])[-20:]
|
||||
self.stopped_at = None
|
||||
self.last_t = t
|
||||
screen, layout = self.stage()
|
||||
lname = layout.get("name") or ""
|
||||
ok_layout = any(a in lname.lower() for a in CFG["accept"])
|
||||
if screen and not ok_layout and CFG["stage_after"] <= now - self.started <= CFG["stage_window"] and not self.stage_fixed:
|
||||
fixed = self.set_layout(screen)
|
||||
self.stage_fixed = True
|
||||
fix = {"song": name, "was": lname, "time": now, "fixed": fixed}
|
||||
self.state["stage_fixes"] = (self.state["stage_fixes"] + [fix])[-20:]
|
||||
log.warning("STAGE: %s started with the stage on %r: %s", name, lname, "set to " + CFG["song_layout"] if fixed else "COULD NOT SET IT")
|
||||
self.state["now"] = {"song": name, "running": running, "t": t, "slide": slide, "stage": lname, "time": now}
|
||||
|
||||
def save(self):
|
||||
try:
|
||||
os.makedirs(os.path.dirname(STATUS), exist_ok=True)
|
||||
with open(STATUS + ".tmp", "w") as f:
|
||||
json.dump(self.state, f)
|
||||
os.replace(STATUS + ".tmp", STATUS)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) > 1 and os.path.exists(sys.argv[1]):
|
||||
CFG.update(json.load(open(sys.argv[1])))
|
||||
logging.basicConfig(filename=os.path.expanduser("~/Library/Logs/prop-guard.log"), level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s")
|
||||
log.info("prop guard started: %s", json.dumps(CFG))
|
||||
g, saved = Guard(), 0.0
|
||||
while True:
|
||||
try:
|
||||
g.tick()
|
||||
except Exception: # noqa: BLE001 - keep guarding
|
||||
log.exception("tick failed")
|
||||
if time.time() - saved > 2:
|
||||
g.save()
|
||||
saved = time.time()
|
||||
time.sleep(CFG["poll"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user