Files
Nextcue/source/npm_vendor.py
T
christianandClaude Opus 5.5 1a658875bc Resi cue countdowns on ProPresenter and in Companion
The Resi -> ProPresenter bridge, its installer and site kit, and the
Companion module resi-studio-cues, as running at North Attleboro.

Bridge (resi_to_propresenter.py)
- Follows a Resi decoder's playhead and keeps ProPresenter's NextCue timer
  counting down to the next cue; optional cue name as a stage message, sent
  once per cue (ProPresenter flashes it on every send) and off by default.
- Playhead estimate no longer learns a cadence from the partial first gap
  after a start, so a 30s Resi doesn't read as paused half the time.
- Backs off on 429, holds 2 min before retrying a failed startup login, and
  logs Resi's actual reason (wrong password, USER_LOCKED, rate limit).
- --setup asks for the login, decoder, ProPresenter and timer.

install_bridge.sh installs it as a login service (outside ~/Documents,
which launchd may not read); make_kit.sh builds the kit for another site.

Companion module resi-studio-cues 0.1.4 (source/, package committed)
- Same playhead fix; login and request hold-off (15s doubling to 5 min);
  a 401/403 on a fresh session is a permission, not a reason to sign in
  again; init never fails on a Resi problem, so Companion can't restart it
  into a login loop.

Tests: test_playhead.py, test_stage_message.py, test_rate_limit.py and
source/backoff_test.js run offline; test_end_to_end.py drives a real
ProPresenter and refuses to run without an explicit flag.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 18:11:01 -04:00

106 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""Assemble a fully-nested node_modules tree from the npm registry, without npm.
Fully nested (no hoisting) so Node's upward resolution can never pick a wrong
version: every package's deps live inside its own node_modules/.
"""
import json
import os
import shutil
import subprocess
import sys
import tarfile
import urllib.parse
import urllib.request
CACHE = os.path.abspath("_npmcache")
os.makedirs(CACHE, exist_ok=True)
_meta = {}
def meta(name):
if name not in _meta:
path = os.path.join(CACHE, name.replace("/", "_") + ".json")
if not os.path.exists(path):
url = "https://registry.npmjs.org/" + urllib.parse.quote(name, safe="")
urllib.request.urlretrieve(url, path)
with open(path) as fh:
_meta[name] = json.load(fh)
return _meta[name]
def parse(v):
core = v.split("-")[0].split("+")[0]
return tuple(int(x) for x in core.split("."))
def satisfies(version, rng):
if "-" in version: # never auto-pick a prerelease
return False
rng = rng.strip()
v = parse(version)
if rng in ("", "*", "latest", "x"):
return True
if rng.startswith("^"):
b = parse(rng[1:])
if v < b:
return False
if b[0] > 0:
return v[0] == b[0]
if b[1] > 0:
return v[0] == 0 and v[1] == b[1]
return v[0] == 0 and v[1] == 0 and v[2] == b[2]
if rng.startswith("~"):
b = parse(rng[1:])
return v >= b and v[0] == b[0] and v[1] == b[1]
if rng.startswith(">="):
return v >= parse(rng[2:])
if rng[0].isdigit():
return version == rng
raise ValueError("unsupported range %r" % rng)
def pick(name, rng):
m = meta(name)
ok = [v for v in m["versions"] if satisfies(v, rng)]
if not ok:
raise SystemExit("no version of %s satisfies %s" % (name, rng))
best = max(ok, key=parse)
return best, m["versions"][best]
def install(name, rng, target_nm, chain, log):
version, info = pick(name, rng)
if (name, version) in chain:
return # already provided by an ancestor
dest = os.path.join(target_nm, *name.split("/"))
if os.path.exists(dest):
return
os.makedirs(os.path.dirname(dest), exist_ok=True)
tgz = os.path.join(CACHE, "%s-%s.tgz" % (name.replace("/", "_"), version))
if not os.path.exists(tgz):
urllib.request.urlretrieve(info["dist"]["tarball"], tgz)
tmp = dest + ".tmp"
shutil.rmtree(tmp, ignore_errors=True)
with tarfile.open(tgz) as tf:
tf.extractall(tmp)
inner = os.path.join(tmp, "package")
shutil.move(inner if os.path.isdir(inner) else tmp, dest)
shutil.rmtree(tmp, ignore_errors=True)
log.append("%s%s@%s" % (" " * len(chain), name, version))
deps = info.get("dependencies") or {}
if deps:
child_nm = os.path.join(dest, "node_modules")
for dname, drng in sorted(deps.items()):
install(dname, drng, child_nm, chain | {(name, version)}, log)
if __name__ == "__main__":
root_nm, name, rng = sys.argv[1], sys.argv[2], sys.argv[3]
os.makedirs(root_nm, exist_ok=True)
log = []
install(name, rng, root_nm, frozenset(), log)
print("\n".join(log))
print("\n%d packages" % len(log))