ProdDash/modules/resi-studio-cues/resi_to_propresenter.py

#!/usr/bin/env python3
"""
resi_to_propresenter.py — drive a ProPresenter countdown timer from Resi cue points.

Watches a Resi decoder's playhead, works out how long until its next cue, and keeps a
ProPresenter timer armed with that number. ProPresenter renders the countdown itself, so
the stage screen is smooth and styled; this only pokes it when something actually changes.

No Companion, no Stream Deck. Python 3 only — nothing to install.

    python3 resi_to_propresenter.py --list     # show venues/decoders, then fill in the config
    python3 resi_to_propresenter.py            # run it
    python3 resi_to_propresenter.py --dry-run  # watch what it WOULD send

The Resi password is read from the macOS Keychain, never stored here. Add it once:

    security add-generic-password -a "$USER" -s resi-propresenter -w

Resi's API is undocumented and unsupported; don't open a Resi ticket about it.
"""

import argparse
import http.cookiejar
import json
import math
import os
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))


def config_path():
    """Settings live OUTSIDE the checkout so a git pull never touches them — the same
    rule ProdDash follows. A config sitting next to the script still wins, so a copy
    dropped on a USB stick keeps working."""
    env = os.environ.get("RESI_PP_CONFIG")
    if env:
        return os.path.expanduser(env)
    beside = os.path.join(HERE, "resi_pp_config.json")
    if os.path.exists(beside):
        return beside
    return os.path.expanduser("~/Library/Application Support/ResiCues/config.json")


CONFIG_PATH = config_path()
RESI = "https://central.resi.io"
KEYCHAIN_SERVICE = "resi-propresenter"

DEFAULTS = {
    "resi_user": "",
    "player": "",             # playerHardwareId; blank = use the only one found
    "propresenter": "http://10.1.14.225:1025",
    "timer_name": "NextCue",
    "stage_message": True,    # also push the cue NAME as a stage message
    "poll_seconds": 1.0,      # how often to ask Resi where the playhead is
    "cue_refresh_seconds": 5, # how often to re-read the cue list
    "resync_seconds": 30,     # re-push to ProPresenter even when nothing changed
    "position_trim_ms": 0,    # only if the whole chain reads consistently behind
}


# --------------------------------------------------------------------------- utilities
def now_ms():
    """Wall clock in milliseconds. A seam so the estimator can be tested on a fake clock."""
    return time.time() * 1000


def log(msg):
    print("%s  %s" % (time.strftime("%H:%M:%S"), msg), flush=True)


def hms(ms):
    s = max(0, int(ms // 1000))
    return "%02d:%02d:%02d" % (s // 3600, (s % 3600) // 60, s % 60)


def position_to_ms(pos):
    """'01:02:03.500' -> milliseconds, or None."""
    try:
        hh, mm, rest = pos.split(":")
        return int((int(hh) * 3600 + int(mm) * 60 + float(rest)) * 1000)
    except (ValueError, AttributeError):
        return None


def keychain_password(account):
    try:
        out = subprocess.run(
            ["security", "find-generic-password", "-a", account, "-s", KEYCHAIN_SERVICE, "-w"],
            capture_output=True, text=True, timeout=10)
    except (OSError, subprocess.SubprocessError) as e:
        raise SystemExit("Could not run the Keychain helper: %s" % e)
    if out.returncode != 0:
        return None
    return out.stdout.rstrip("\n")


def wait_for_password(account):
    """Running as a login agent, a missing password is something to wait out rather
    than die on — the agent would otherwise relaunch forever before it is set up."""
    told = False
    while True:
        pw = keychain_password(account)
        if pw:
            return pw
        if not told:
            log("No Resi password in the Keychain for '%s'. Add it once with:" % account)
            log('    security add-generic-password -a "%s" -s %s -w'
                % (account, KEYCHAIN_SERVICE))
            log("Waiting for it...")
            told = True
        time.sleep(60)


# ------------------------------------------------------------------------------- Resi
class Resi:
    def __init__(self, username, password):
        self.username, self.password = username, password
        self.jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar))
        self.token = None
        self.logged_in_at = 0
        self.customer_id = None
        self.no_cache_bust = False

    def _cookie(self, name):
        for c in self.jar:
            if c.name == name:
                return c.value
        return None

    def _call(self, method, path, body=None):
        url = RESI + path
        headers = {"Accept": "application/json", "User-Agent": "resi-to-propresenter/1.0"}
        if self.token:
            headers["Authorization"] = "X-Bearer " + self.token
        data = None
        if body is not None:
            data = json.dumps(body).encode()
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        with self.opener.open(req, timeout=20) as resp:
            raw = resp.read()
        return json.loads(raw) if raw else None

    def login(self):
        self._call("POST", "/api/v3/login?newToken=true",
                   {"userName": self.username, "password": self.password})
        self.token = self._cookie("livingas1")
        if not self.token:
            raise RuntimeError("login succeeded but returned no session cookie")
        self.logged_in_at = time.time()
        me = self._call("GET", "/api_v2.svc/users/me")
        self.customer_id = me.get("customerId")
        if not self.customer_id:
            raise RuntimeError("this Resi account has no customerId")
        log("Resi: signed in as %s (%s)" % (me.get("userName"), me.get("customerName")))

    def ensure_session(self):
        if not self.token or time.time() - self.logged_in_at > 50 * 60:
            self.login()

    def get(self, path):
        self.ensure_session()
        try:
            return self._call("GET", path)
        except urllib.error.HTTPError as e:
            if e.code in (401, 403):
                log("Resi: session expired, signing in again")
                self.login()
                return self._call("GET", path)
            raise

    def venues(self):
        return self.get("/api/v3/customers/%s/venues" % self.customer_id) or []

    def players(self, venue_uuid):
        path = "/api_v2.svc/users/%s/players" % venue_uuid
        if not self.no_cache_bust:
            try:
                return self.get(path + "?_=%d" % int(time.time() * 1000)) or []
            except urllib.error.HTTPError as e:
                if 400 <= e.code < 500:
                    self.no_cache_bust = True
                    log("Resi: dropping the cache-busting parameter (%s)" % e.code)
                else:
                    raise
        return self.get(path) or []

    def event(self, event_id):
        return self.get("/api/v3/customers/%s/events/%s" % (self.customer_id, event_id))

    def cues(self, profile_id, event_id):
        raw = self.get("/api_v2.svc/streamprofiles/%s/events/%s/cues?canISetCues=1"
                       % (profile_id, event_id)) or []
        out = []
        for c in raw:
            ms = position_to_ms(c.get("position"))
            if ms is not None:
                out.append({"name": c.get("name") or "", "ms": ms})
        out.sort(key=lambda c: c["ms"])
        return out


# -------------------------------------------------------------------------- estimator
class Playhead:
    """Resi republishes `position` only every few seconds, irregularly, and each reading
    is already slightly stale. Every reading is kept as media-time-minus-wall-time; the
    LARGEST recent value is the least-delayed reading and is the one trusted. The
    estimate is then just "now plus that offset", correct to within Resi's own pipeline
    lag and smooth between updates."""

    WINDOW_MS = 120_000

    def __init__(self, trim_ms=0):
        self.trim_ms = trim_ms
        self.offsets = []       # (arrival_ms, offset_ms)
        self.spreads = []       # how far honest readings sit apart: this link's noise
        self.gaps = []          # how often the playhead actually moves
        self.base = None
        self.last_media = None
        self.last_new_at = None
        self.last_est = None
        self.seeked = False

    def note(self, media_ms):
        now = now_ms()
        if media_ms != self.last_media:
            off = media_ms - now
            noise = max(self.spreads) if self.spreads else 0
            threshold = max(5000, noise * 1.5 + 4000)
            if self.base is not None:
                delta = off - self.base
                if abs(delta) > threshold:
                    # seek, fired cue, or resume after a pause: the window describes
                    # the old position and has to go
                    self.offsets = []
                    self.seeked = True
                    if abs(delta) > 10000:
                        self.last_est = None
                elif delta < 0:
                    self.spreads = (self.spreads + [-delta])[-20:]
            if self.last_new_at is not None:
                gap = now - self.last_new_at
                # The gap either side of a pause is not Resi's cadence -- recording it
                # would inflate the estimate and make the NEXT pause slow to notice.
                if 0 < gap < 60000 and (not self.gaps or gap <= 3 * max(self.gaps)):
                    self.gaps = (self.gaps + [gap])[-8:]
            self.offsets.append((now, off))
            self.last_media = media_ms
            self.last_new_at = now
        kept = [o for o in self.offsets if now - o[0] <= self.WINDOW_MS]
        # Always keep the newest reading even once it is older than the window. If the
        # window empties during a long pause there is no offset left to compare against,
        # so the resume looks like an ordinary reading instead of a discontinuity.
        if not kept and self.offsets:
            kept = [self.offsets[-1]]
        self.offsets = kept
        self.base = max(o[1] for o in self.offsets) if self.offsets else None

    def read(self):
        """-> (estimated_media_ms, playing, cadence_ms) or (None, False, 0)."""
        if self.last_new_at is None:
            return None, False, 0
        now = now_ms()
        cadence = max(self.gaps) if self.gaps else 0
        # Only call it paused once the playhead has been still for clearly longer than
        # Resi's own reporting cadence -- a fixed timeout mistakes a slow update for a stop.
        pause_after = max(15000, min(45000, cadence * 2.5 + 6000))
        playing = self.base is not None and (now - self.last_new_at) < pause_after
        if not playing:
            return (self.last_est if self.last_est is not None else self.last_media), False, cadence
        est = now + self.base + self.trim_ms
        if self.last_est is not None and est < self.last_est:
            est = self.last_est          # the window is a max; never run backwards
        self.last_est = est
        return est, True, cadence

    def take_seek(self):
        was, self.seeked = self.seeked, False
        return was


# ------------------------------------------------------------------- ProPresenter side
class ProPresenter:
    def __init__(self, base, timer_name, dry_run=False):
        self.base = base.rstrip("/")
        self.timer = timer_name
        self.dry_run = dry_run
        self.quoted = urllib.parse.quote(timer_name, safe="")

    def _call(self, method, path, body=None):
        if self.dry_run:
            log("   would %s %s %s" % (method, path, json.dumps(body) if body is not None else ""))
            return True
        url = self.base + path
        data = None
        headers = {}
        if body is not None:
            data = json.dumps(body).encode()
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with urllib.request.urlopen(req, timeout=8):
                return True
        except urllib.error.HTTPError as e:
            log("ProPresenter %s %s -> HTTP %s" % (method, path, e.code))
        except urllib.error.URLError as e:
            log("ProPresenter unreachable at %s (%s)" % (self.base, e.reason))
        return False

    def arm(self, seconds):
        return self._call("PUT", "/v1/timer/%s/start" % self.quoted, {
            "id": {"name": self.timer},
            "allows_overrun": False,
            "countdown": {"duration": int(seconds)},
        })

    def stop(self):
        return self._call("GET", "/v1/timer/%s/stop" % self.quoted)

    def message(self, text):
        return self._call("PUT", "/v1/stage/message", text)

    def clear_message(self):
        return self._call("DELETE", "/v1/stage/message")

    def check(self):
        try:
            with urllib.request.urlopen(self.base + "/v1/timers", timeout=8) as r:
                timers = json.loads(r.read())
        except Exception as e:
            raise SystemExit("Cannot reach ProPresenter at %s (%s).\n"
                             "Check the address and that its API is enabled in Settings -> Network."
                             % (self.base, e))
        names = [t.get("id", {}).get("name") for t in timers]
        if self.timer not in names:
            raise SystemExit("ProPresenter has no timer called '%s'. It has: %s\n"
                             "Create a countdown timer with that exact name, or set "
                             "\"timer_name\" in the config." % (self.timer, ", ".join(map(str, names))))
        log("ProPresenter: found timer '%s'" % self.timer)


# ------------------------------------------------------------------------------- glue
def load_config():
    cfg = dict(DEFAULTS)
    if os.path.exists(CONFIG_PATH):
        with open(CONFIG_PATH) as fh:
            cfg.update(json.load(fh))
    else:
        os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
        with open(CONFIG_PATH, "w") as fh:
            json.dump(cfg, fh, indent=2)
        print("Wrote a starter config to %s — fill in resi_user, then re-run." % CONFIG_PATH)
        raise SystemExit(0)
    if not cfg["resi_user"]:
        raise SystemExit("Set \"resi_user\" in %s first." % CONFIG_PATH)
    return cfg


def find_player(resi, wanted):
    """-> (venue_uuid, player) for the configured decoder."""
    found = []
    for v in resi.venues():
        for p in resi.players(v.get("uuid")):
            found.append((v, p))
    if not found:
        raise SystemExit("Resi returned no decoders for this account.")
    if wanted:
        for v, p in found:
            if p.get("playerHardwareId") == wanted:
                return v.get("uuid"), p
        raise SystemExit("No decoder with playerHardwareId '%s'. Run --list to see them." % wanted)
    if len(found) == 1:
        v, p = found[0]
        return v.get("uuid"), p
    raise SystemExit("This account has %d decoders — run --list and set \"player\" in the config."
                     % len(found))


def do_list(resi):
    for v in resi.venues():
        print("\n%s" % (v.get("userName") or v.get("name")))
        players = resi.players(v.get("uuid"))
        if not players:
            print("    (no decoders)")
        for p in players:
            print("    %-30s playerHardwareId: %s" % (p.get("userName"), p.get("playerHardwareId")))
            print("    %-30s position: %s   eventId: %s"
                  % ("", p.get("position"), p.get("eventId")))
    print("\nPut the playerHardwareId you want into \"player\" in %s" % CONFIG_PATH)


def run(cfg, dry_run):
    resi = Resi(cfg["resi_user"], wait_for_password(cfg["resi_user"]))
    resi.login()
    pp = ProPresenter(cfg["propresenter"], cfg["timer_name"], dry_run)
    if not dry_run:
        pp.check()

    venue_uuid, player = find_player(resi, cfg["player"])
    hardware_id = player.get("playerHardwareId")
    log("Watching decoder '%s' (%s)" % (player.get("userName"), hardware_id))

    head = Playhead(cfg["position_trim_ms"])
    cues, event_id, profile_id = [], None, None
    last_cue_load = 0.0
    last_push = 0.0
    pushed_cue = None      # (name, ms) currently armed in ProPresenter
    pushed_playing = None

    while True:
        loop_started = time.time()
        try:
            players = resi.players(venue_uuid)
            me = next((p for p in players if p.get("playerHardwareId") == hardware_id), None)
            if me is None:
                log("decoder not in the players list right now")
            else:
                media = position_to_ms(me.get("position"))
                if media is not None:
                    head.note(media)

                # follow the event the decoder is actually on
                if me.get("eventId") and me.get("eventId") != event_id:
                    event_id = me.get("eventId")
                    ev = resi.event(event_id) or {}
                    profile_id = ev.get("eventProfileId")
                    cues = resi.cues(profile_id, event_id) if profile_id else []
                    last_cue_load = time.time()
                    log("event %s — %d cue(s): %s"
                        % (ev.get("name") or event_id, len(cues),
                           ", ".join(c["name"] for c in cues) or "none"))
                elif not me.get("eventId"):
                    event_id, cues = None, []

                seeked = head.take_seek()
                if profile_id and event_id and (
                        seeked or time.time() - last_cue_load >= cfg["cue_refresh_seconds"]):
                    fresh = resi.cues(profile_id, event_id)
                    last_cue_load = time.time()
                    if fresh != cues:
                        log("cue list changed — %d cue(s)" % len(fresh))
                        cues = fresh

                est, playing, cadence = head.read()
                nxt = next((c for c in cues if c["ms"] > est), None) if est is not None else None

                identity = (nxt["name"], nxt["ms"]) if nxt else None
                due = (identity != pushed_cue
                       or playing != pushed_playing
                       or seeked
                       or time.time() - last_push >= cfg["resync_seconds"])

                if due:
                    if playing and nxt:
                        remaining = max(0, (nxt["ms"] - est) / 1000.0)
                        pp.arm(math.ceil(remaining))
                        if cfg["stage_message"]:
                            pp.message(nxt["name"])
                        log("%s  ->  %s in %s   (playhead %s, Resi updates every %ss)"
                            % (cfg["timer_name"], nxt["name"], hms(nxt["ms"] - est),
                               hms(est), round(cadence / 1000) if cadence else "?"))
                    elif not playing:
                        pp.stop()
                        log("playback stopped — timer held")
                    else:
                        pp.stop()
                        if cfg["stage_message"]:
                            pp.clear_message()
                        log("no cues left on this event")
                    pushed_cue, pushed_playing, last_push = identity, playing, time.time()

        except urllib.error.URLError as e:
            log("network: %s" % e.reason)
        except Exception as e:                       # keep running through a bad service day
            log("error: %s" % e)

        time.sleep(max(0.2, cfg["poll_seconds"] - (time.time() - loop_started)))


def main():
    ap = argparse.ArgumentParser(description="Drive a ProPresenter timer from Resi cue points.")
    ap.add_argument("--list", action="store_true", help="list venues and decoders, then exit")
    ap.add_argument("--dry-run", action="store_true", help="print what would be sent, touch nothing")
    args = ap.parse_args()

    cfg = load_config()
    if args.list:
        pw = keychain_password(cfg["resi_user"])
        if not pw:
            raise SystemExit(
                "No Resi password in the Keychain for '%s'. Add it once with:\n"
                '    security add-generic-password -a "%s" -s %s -w'
                % (cfg["resi_user"], cfg["resi_user"], KEYCHAIN_SERVICE))
        resi = Resi(cfg["resi_user"], pw)
        resi.login()
        do_list(resi)
        return
    try:
        run(cfg, args.dry_run)
    except KeyboardInterrupt:
        print("\nstopped")


if __name__ == "__main__":
    main()