#!/usr/bin/env python3 """ resi_probe.py — does the Resi cloud hand us live cue + playhead data for OUR account? One read-only pass over the endpoints a Companion "cue countdown" module would need: login POST /api/v3/login?newToken=true who am i GET /api_v2.svc/users/me venues GET /api/v3/customers/{customerId}/venues decoders GET /api_v2.svc/users/{venueUuid}/players <- playhead lives here event GET /api/v3/customers/{customerId}/events/{eventId} cues GET /api_v2.svc/streamprofiles/{eventProfileId}/events/{eventId}/cues Nothing is created, changed or deleted. Your password is read with getpass (never echoed), sent only to https://central.resi.io, and never written to disk. This talks to Resi's INTERNAL API — the one studio.resi.io itself uses. It is undocumented and unsupported; don't open a Resi ticket about it. Usage: python3 resi_probe.py --user you@yourchurch.org python3 resi_probe.py --user you@yourchurch.org --watch 10 # 10s playhead watch python3 resi_probe.py --user you@yourchurch.org --save ./resi-capture """ import argparse import getpass import http.cookiejar import json import os import ssl import sys import time import urllib.error import urllib.parse import urllib.request BASE = "https://central.resi.io" _jar = http.cookiejar.CookieJar() _opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(_jar)) _token = None _save_dir = None def _cookie(name): for c in _jar: if c.name == name: return c.value return None def _headers(): h = {"Accept": "application/json", "User-Agent": "resi-probe/1.0"} if _token: h["Authorization"] = "X-Bearer " + _token return h def _request(method, path, body=None): url = path if path.startswith("http") else BASE + path data = None h = _headers() if body is not None: data = json.dumps(body).encode() h["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=h, method=method) try: with _opener.open(req, timeout=20) as resp: raw = resp.read() except urllib.error.HTTPError as e: raise RuntimeError("HTTP %s on %s %s — %s" % (e.code, method, path, e.read()[:300].decode("utf-8", "replace"))) except ssl.SSLError as e: raise RuntimeError("TLS failure (%s). If this Mac's Python has no CA bundle, run the " "'Install Certificates.command' that ships with your Python install." % e) except urllib.error.URLError as e: raise RuntimeError("Could not reach %s — %s" % (url, e.reason)) if not raw: return None try: return json.loads(raw) except ValueError: return raw.decode("utf-8", "replace") def get(path): out = _request("GET", path) if _save_dir is not None: name = urllib.parse.quote(path.split("?")[0].strip("/"), safe="") + ".json" with open(os.path.join(_save_dir, name), "w") as fh: json.dump(out, fh, indent=2, sort_keys=True) return out def login(username, password): global _token _request("POST", "/api/v3/login?newToken=true", {"userName": username, "password": password}) _token = _cookie("livingas1") if not _token: raise RuntimeError("Login returned 2xx but set no 'livingas1' cookie — the auth shape has changed. " "Cookies seen: %s" % ", ".join(c.name for c in _jar)) def ms(position): """'01:02:03.500' -> milliseconds. Returns None on anything unparseable.""" if not position: return None try: hh, mm_, rest = position.split(":") return int((int(hh) * 3600 + int(mm_) * 60 + float(rest)) * 1000) except (ValueError, AttributeError): return None def hms(milliseconds): s, sign = int(abs(milliseconds) // 1000), "-" if milliseconds < 0 else "" return "%s%02d:%02d:%02d" % (sign, s // 3600, (s % 3600) // 60, s % 60) def main(): global _save_dir ap = argparse.ArgumentParser(description="Read-only probe of the Resi internal API.") ap.add_argument("--user", help="Resi Studio login (email)") ap.add_argument("--watch", type=int, default=0, metavar="SEC", help="after the report, poll the playhead for SEC seconds to prove it advances") ap.add_argument("--save", metavar="DIR", help="write each raw JSON response to DIR (contains venue/decoder names and uuids)") args = ap.parse_args() if args.save: os.makedirs(args.save, exist_ok=True) _save_dir = args.save username = args.user or input("Resi login (email): ").strip() password = getpass.getpass("Password (not echoed, not stored): ") print("\n== auth ==") login(username, password) print(" login OK, bearer token acquired") me = get("/api_v2.svc/users/me") customer_id = me.get("customerId") print(" user : %s" % me.get("userName")) print(" org : %s (customerId %s)" % (me.get("customerName"), customer_id)) if not customer_id: sys.exit(" no customerId on this account — stop here, nothing else will resolve") print("\n== venues ==") venues = get("/api/v3/customers/%s/venues" % customer_id) or [] if not venues: print(" none returned. Either this login has no venue access, or the plan has no Multisite.") for v in venues: print(" %-40s %s" % (v.get("userName") or v.get("name"), v.get("uuid"))) print("\n== decoders / players ==") live = [] for v in venues: players = [] try: players = get("/api_v2.svc/users/%s/players" % v.get("uuid")) or [] except RuntimeError as e: print(" [%s] players call failed: %s" % (v.get("userName"), e)) continue if not players: print(" [%s] no players" % v.get("userName")) for p in players: print(" [%s] %-28s pos=%-14s eventId=%s" % (v.get("userName"), p.get("userName"), p.get("position"), p.get("eventId"))) extra = sorted(set(p) - {"playerHardwareId", "userName", "position", "eventId"}) if extra: print(" other fields on this player: %s" % ", ".join(extra)) if p.get("eventId"): live.append((v, p)) if not live: print("\nNo decoder is on an event right now. Re-run this DURING a service —" "\nposition and eventId are what the whole countdown hangs on.") return for v, p in live: print("\n== event + cues for %s / %s ==" % (v.get("userName"), p.get("userName"))) event = get("/api/v3/customers/%s/events/%s" % (customer_id, p.get("eventId"))) profile_id = event.get("eventProfileId") print(" event : %s" % event.get("name")) print(" startTime : %s" % event.get("startTime")) print(" profileId : %s" % profile_id) if not profile_id: print(" no eventProfileId — cannot fetch cues for this event") continue cues = get("/api_v2.svc/streamprofiles/%s/events/%s/cues?canISetCues=1" % (profile_id, p.get("eventId"))) or [] cues = sorted(cues, key=lambda c: ms(c.get("position")) or 0) print(" %d cue(s):" % len(cues)) here = ms(p.get("position")) for c in cues: at = ms(c.get("position")) marker = " " if here is not None and at is not None: marker = "->" if at > here else " " print(" %s %-14s %s" % (marker, c.get("position"), c.get("name"))) if here is None: print("\n playhead unparseable (%r) — countdown not possible from this field" % p.get("position")) continue nxt = next((c for c in cues if (ms(c.get("position")) or 0) > here), None) print("\n THE BUTTON WOULD READ:") print(" Playback position : %s" % hms(here)) if nxt: print(" Next cue : %s" % nxt.get("name")) print(" Next cue in : %s" % hms((ms(nxt.get("position")) or 0) - here)) else: print(" Next cue : (none left on this event)") if args.watch: v, p = live[0] print("\n== watching %s for %ds (does the playhead advance?) ==" % (p.get("userName"), args.watch)) first = last = None deadline = time.time() + args.watch while time.time() < deadline: players = get("/api_v2.svc/users/%s/players" % v.get("uuid")) or [] now = next((x.get("position") for x in players if x.get("playerHardwareId") == p.get("playerHardwareId")), None) print(" %s %s" % (time.strftime("%H:%M:%S"), now)) if first is None: first = ms(now) last = ms(now) time.sleep(2) if first is not None and last is not None: moved = last - first print(" moved %s over ~%ds -> %s" % (hms(moved), args.watch, "PLAYING (a play/pause variable is derivable)" if moved > 0 else "frozen / paused")) if __name__ == "__main__": try: main() except RuntimeError as exc: sys.exit("\nFAILED: %s" % exc) except KeyboardInterrupt: sys.exit("\ninterrupted")