#!/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, "ltc_file": "~/Library/Application Support/status-panel/ltc.json", # LTC Listen.app: the timecode really arriving "ltc_lost_after": 2.0, # seconds without timecode in a song that had it: TIMECODE LOST "status_port": 8095, # GET /timecode: 200 fine, 503 lost (the status panel's http check) } 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": [], "ltc_drops": [], "now": {}} self.ltc_seen = False # timecode heard during this song self.ltc_lost_at = None # when it stopped (None: running, or not heard yet this song) self.lost = False # TIMECODE LOST, for the status panel 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 ltc(self): try: with open(os.path.expanduser(CFG["ltc_file"])) as f: d = json.load(f) except (OSError, ValueError): return None return d if self.now() - (d.get("time") or 0) < 5 else None # LTC Listen not writing: no verdict def watch_ltc(self, name, now): """LTC Listen hears the tracks' timecode on the same SoundGrid input ProPresenter uses. In a song that had it, timecode gone for ltc_lost_after seconds is TIMECODE LOST (logged with where it stopped) until it comes back.""" d = self.ltc() if d is None or name is None: self.ltc_seen, self.ltc_lost_at, self.lost = False, None, False return if d.get("running"): if self.lost: log.warning("TIMECODE BACK in %s after %.1f s at %s", name, now - self.ltc_lost_at, d.get("tc")) self.state["ltc_drops"][-1]["back"] = now self.ltc_seen, self.ltc_lost_at, self.lost = True, None, False self.last_tc = d.get("tc") return if not self.ltc_seen: return self.ltc_lost_at = self.ltc_lost_at or now if not self.lost and now - self.ltc_lost_at >= CFG["ltc_lost_after"]: self.lost = True log.warning("TIMECODE DROPPED in %s at %s: no timecode on SoundGrid for %.1f s", name, getattr(self, "last_tc", "?"), now - self.ltc_lost_at) self.state["ltc_drops"] = (self.state["ltc_drops"] + [{"song": name, "tc": getattr(self, "last_tc", None), "time": self.ltc_lost_at}])[-20:] def tick(self): now = self.now() pres = (api("/v1/presentation/active") or {}).get("presentation") song_name = pres["id"]["name"] if pres and pres.get("has_timeline") else None if song_name != getattr(self, "ltc_song", None): self.ltc_seen, self.ltc_lost_at, self.lost, self.ltc_song = False, None, False, song_name self.watch_ltc(song_name, now) 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 serve_status(guard): """GET /timecode: 200 "RUNNING" / "OK", or 503 "TIMECODE LOST" (SC's status panel checks it).""" import http.server import threading class H(http.server.BaseHTTPRequestHandler): def do_GET(self): lost = guard.lost body = b"TIMECODE LOST" if lost else b"OK" self.send_response(503 if lost else 200) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(body) def log_message(self, *a): pass srv = http.server.ThreadingHTTPServer(("127.0.0.1", CFG["status_port"]), H) threading.Thread(target=srv.serve_forever, daemon=True).start() 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 try: serve_status(g) except OSError as e: log.warning("status port %s: %s", CFG["status_port"], e) 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()