status-indicators/prop_guard.py

#!/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()