christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LTC timecode goes to the other campuses' plans too: Apollo Beach, Southcoast, Woonsocket

Browse Source 
The TIMECODE key now reads each campus's plan for the same weekend, pairs its songs with North
Attleboro's by title in North Attleboro's order (a song out of order, or one NA does not play, is
left alone and said so), checks their LTC notes, and the hold writes the timecode into every
campus's plan. ready.ltc_campuses lists the campuses (name, two letters, PCO service type).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-19 10:26:13 -04:00
co-authored by Claude Fable 5.1
parent 87bfe7c162
commit 3ccaea0076
6 changed files with 140 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
readiness.py
+71 -9
View File
@@ -201,9 +201,12 @@ def pick_plan(plans, now):
return (upcoming[0][1], upcoming[0][0]) if upcoming else (None, None)
def pco_plan(rc, now):
"""This weekend's plan: {"id", "title", "dates", "time", "items": [...], "team": [...]}."""
def pco_plan(rc, now, service_type=None, team=True):
"""This weekend's plan: {"id", "title", "dates", "time", "items": [...], "team": [...]}.
`service_type` reads another campus's plan for the same weekend (its team is not needed)."""
auth, st = pco_login(rc)
if service_type:
st = str(service_type)
try:
future, _ = pco_get(auth, "/service_types/%s/plans?filter=future&order=sort_date&per_page=4" % st, pages=1)
past, _ = pco_get(auth, "/service_types/%s/plans?filter=past&order=-sort_date&per_page=2" % st, pages=1)
@@ -227,7 +230,7 @@ def pco_plan(rc, now):
out_items.append({"id": it.get("id"), "note_ids": note_ids, "title": a.get("title") or "", "type": a.get("item_type") or "",
"position": a.get("service_position") or "", "description": a.get("description") or "",
"notes": ns})
members, inc = pco_get(auth, "/service_types/%s/plans/%s/team_members?include=team&per_page=100" % (st, pid))
members, inc = pco_get(auth, "/service_types/%s/plans/%s/team_members?include=team&per_page=100" % (st, pid)) if team else ([], [])
teams = {i["id"]: (i.get("attributes") or {}).get("name") for i in inc if i.get("type") == "Team"}
team = []
for m in members:
@@ -637,6 +640,46 @@ def song_ltc(rc, plan, look):
return out
def match_in_order(na_songs, campus_songs):
"""Pair a campus's songs with the home campus's by title, keeping the home order: a campus song
whose place breaks that order (or that the home campus does not play) is skipped, that song only.
Returns ([(campus item, home song)], [skipped titles])."""
pairs, skipped, last = [], [], -1
for it in campus_songs:
idx = next((i for i, s in enumerate(na_songs) if i > last and same(it["title"], s["title"])), None)
if idx is None:
skipped.append(it["title"])
continue
pairs.append((it, na_songs[idx]))
last = idx
return pairs, skipped
def campus_ltc(rc, na_songs, campus, plan):
"""One other campus's plan against the home timecode: its songs in home order, each with the
timecode it should carry and what its LTC note says now."""
note = rc.get("ltc_note", "LTC")
pairs, skipped = match_in_order(na_songs, plan_songs(rc, plan))
songs = [{"item": it, "title": it["title"], "tracks": s["tracks"], "pco": (it["notes"].get(note) or "").strip()} for it, s in pairs]
return {"name": campus.get("name") or campus.get("short") or str(campus.get("service_type")), "short": campus.get("short") or (campus.get("name") or "?")[:2].upper(),
"service_type": str(campus.get("service_type")), "plan_id": plan["id"], "dates": plan.get("dates", ""), "songs": songs, "skipped": skipped}
def eval_campus_ltc(rc, result, line):
"""The other campus's notes must say the home timecode for every matched song."""
tag = result["short"]
for s in result["songs"]:
if not s["pco"]:
line.fault(ERR, "%s NO LTC %s" % (tag, short(s["title"], 12)), "%s: %s has no %s note; the tracks start it at %s" % (
result["name"], s["title"], rc.get("ltc_note", "LTC"), s["tracks"]))
elif s["pco"] != s["tracks"]:
line.fault(ERR, "%s LTC WRONG %s" % (tag, short(s["title"], 10)), "%s: %s says %s, the tracks start it at %s" % (
result["name"], s["title"], s["pco"], s["tracks"]))
if result["skipped"]:
line.notes.append("%s: not in this campus's order, left alone: %s" % (result["name"], ", ".join(result["skipped"])))
return line
def eval_ltc(rc, plan, look, line):
"""PCO's LTC note for every song must be the timecode the tracks play at that song's start."""
for s in song_ltc(rc, plan, look):
@@ -950,14 +993,14 @@ class Line:
"""One checklist line: its faults [(level, msg, detail)] and, when it is fine, what to show."""
def __init__(self):
self.faults, self.info = [], ""
self.faults, self.info, self.notes = [], "", []
def fault(self, level, msg, detail=""):
if all(m != msg for _, m, _ in self.faults):
self.faults.append((level, msg, detail))
def as_dict(self):
return {"faults": [[lv, m] for lv, m, _ in self.faults], "details": [d for _, _, d in self.faults if d],
return {"faults": [[lv, m] for lv, m, _ in self.faults], "details": [d for _, _, d in self.faults if d] + list(self.notes),
"info": self.info}
@@ -966,7 +1009,7 @@ def look(cfg, now=None, data_dir=None, sources=None):
`sources` replaces the readers (tests)."""
rc = cfg.get("ready") or {}
now = now or time.time()
src = dict(pco=lambda: pco_plan(rc, now), prop=lambda plan: prop_look(rc, plan),
src = dict(pco=lambda: pco_plan(rc, now), pco_campus=lambda st: pco_plan(rc, now, st, team=False), prop=lambda plan: prop_look(rc, plan),
ableton=lambda plan: ableton_look(cfg, rc, plan),
lineup=lambda: http_json(rc.get("lineup_url", "").rstrip("/") + "/state", timeout=6),
resolume=lambda: http_json(rc.get("resolume_url", "").rstrip("/") + "/api/v1/composition", timeout=6),
@@ -1023,9 +1066,23 @@ def look(cfg, now=None, data_dir=None, sources=None):
lines.update(eval_ableton(rc, plan, al))
if rc.get("ltc_note", "LTC"):
eval_ltc(rc, plan, al, lines.setdefault("timecode", Line()))
LAST["ltc"] = [dict(s, item={"id": s["item"]["id"], "note_ids": s["item"]["note_ids"]})
for s in song_ltc(rc, plan, al)]
na_songs = song_ltc(rc, plan, al)
LAST["ltc"] = [dict(s, item={"id": s["item"]["id"], "note_ids": s["item"]["note_ids"]}) for s in na_songs]
LAST["service_type"], LAST["plan_id"] = plan.get("service_type"), plan["id"]
others = []
for campus in rc.get("ltc_campuses") or []:
if not campus.get("service_type"):
continue
try:
cplan = src["pco_campus"](campus["service_type"])
res = campus_ltc(rc, na_songs, campus, cplan)
eval_campus_ltc(rc, res, lines["timecode"])
others.append(dict(res, songs=[dict(s, item={"id": s["item"]["id"], "note_ids": s["item"]["note_ids"]}) for s in res["songs"]]))
except Exception as e: # noqa: BLE001 - one campus's plan down must not hide the home one
tag = campus.get("short") or (campus.get("name") or "?")[:2].upper()
msg = str(e) if isinstance(e, ReadyError) else "PCO UNREACHABLE"
lines["timecode"].fault(WARN, "%s %s" % (tag, msg), "%s: %s" % (campus.get("name") or tag, e))
LAST["ltc_other"] = others
if rc.get("locations"):
lines["track_bundle"] = eval_track_bundle(rc, al)
except Exception as e: # noqa: BLE001
@@ -1140,7 +1197,12 @@ def run_action(cfg, action, chk=None, sleep=time.sleep):
comp = http_json(base, timeout=6) or {}
return True # an empty column is filled by hand in Resolume
if action == "ltc_to_pco":
return write_ltc(rc, LAST.get("ltc") or [], LAST.get("service_type"), LAST.get("plan_id"))
result = write_ltc(rc, LAST.get("ltc") or [], LAST.get("service_type"), LAST.get("plan_id"))
for other in LAST.get("ltc_other") or []:
r = write_ltc(rc, other["songs"], other["service_type"], other["plan_id"])
if r is not True and result is True:
result = "%s: %s" % (other["short"], r)
return result
if action == "copy_tracks":
plan = LAST.get("plan")
if not plan: