From b6221e0d639de46ba76e2433085de00aa2a73b60 Mon Sep 17 00:00:00 2001 From: christian Date: Sun, 25 Oct 2026 08:06:29 -0400 Subject: [PATCH] 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 --- prop_guard.py | 161 +++++++++++++++++++++++++++++++++++++++ tests/test_prop_guard.py | 92 ++++++++++++++++++++++ 2 files changed, 253 insertions(+) create mode 100644 prop_guard.py create mode 100644 tests/test_prop_guard.py diff --git a/prop_guard.py b/prop_guard.py new file mode 100644 index 0000000..a81faf7 --- /dev/null +++ b/prop_guard.py @@ -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() diff --git a/tests/test_prop_guard.py b/tests/test_prop_guard.py new file mode 100644 index 0000000..670d446 --- /dev/null +++ b/tests/test_prop_guard.py @@ -0,0 +1,92 @@ +"""prop_guard.py (2026-10-25, Southcoast): a song's timeline started again when the timecode drops, and the stage +display put on the worship layout when the first slide's macro did not, against a fake ProPresenter API.""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import prop_guard as pg # noqa: E402 + + +class FakeProP: + def __init__(self): + self.t, self.running, self.slide, self.layout, self.calls, self.song = 0.0, False, 0, "MC", [], "Jehovah" + + def __call__(self, path, timeout=2.0): + self.calls.append(path) + if path == "/v1/presentation/active": + return {"presentation": {"id": {"uuid": "U-" + self.song, "name": self.song}, "has_timeline": True, + "groups": [{"slides": [{}] * 10}]}} if self.song else {"presentation": None} + if path == "/v1/presentation/active/timeline": + return {"is_running": self.running, "current_time": self.t} + if path == "/v1/presentation/active/timeline/play": + self.running = True + return {} + if path == "/v1/presentation/slide_index": + return {"presentation_index": {"index": self.slide}} + if path == "/v1/stage/layout_map": + return [{"screen": {"uuid": "S0", "name": "Stage Display"}, "layout": {"name": self.layout}}] + if path == "/v1/stage/layouts": + return [{"id": {"uuid": "L1", "name": "Worship"}}, {"id": {"uuid": "L2", "name": "MC"}}] + if path.startswith("/v1/stage/screen/S0/layout/"): + self.layout = "Worship" if path.endswith("L1") else self.layout + return {} + return None + + +class TestGuard(unittest.TestCase): + def setUp(self): + self.fake, self.clock = FakeProP(), [1000.0] + self.orig = pg.api + pg.api = self.fake + self.g = pg.Guard(now=lambda: self.clock[0]) + + def tearDown(self): + pg.api = self.orig + + def at(self, t): + self.clock[0] = 1000.0 + t + self.g.tick() + + def test_stage_put_on_worship_when_the_macro_missed(self): + self.at(0) + self.assertEqual(self.fake.layout, "MC") # judged only after a few seconds + self.at(3) + self.assertEqual(self.fake.layout, "Worship") + self.fake.layout = "Message" # someone picks another layout later: left alone + self.at(20) + self.assertEqual(self.fake.layout, "Message") + + def test_a_timecode_drop_starts_the_timeline_again(self): + self.fake.layout = "Worship" + self.fake.running = True + self.at(0) + self.at(10) + self.fake.running, self.fake.slide = False, 4 # LTC gone mid-song + self.at(10.5) + self.assertNotIn("/v1/presentation/active/timeline/play", self.fake.calls) # a blink is not a drop + self.at(11.6) + self.assertTrue(self.fake.running) + self.assertEqual(len(self.g.state["timecode_drops"]), 1) + self.assertTrue(self.g.state["timecode_drops"][0]["resumed"]) + + def test_stopped_on_purpose_or_at_the_end_is_left_alone(self): + self.fake.layout, self.fake.running = "Worship", True + self.at(0) + self.fake.running, self.fake.slide = False, 4 + self.at(1) + self.at(2.1) # restarted + self.fake.running = False # and stopped again at once: an operator + self.at(3) + self.at(10) + self.assertEqual(self.fake.calls.count("/v1/presentation/active/timeline/play"), 1) + self.fake.song, self.fake.running, self.fake.slide = "Owe You Praise", True, 0 + self.at(20) + self.fake.running, self.fake.slide = False, 9 # its last slide: the song is over + self.at(21) + self.at(25) + self.assertEqual(self.fake.calls.count("/v1/presentation/active/timeline/play"), 1) + + +if __name__ == "__main__": + unittest.main()