TRACKS OUT builds the locations' sets (Location Setup PDF): SC and WS in their own PCO keys (pitched stems transposed, locator, Waves Tune, pads, CtW/Salvation), consolidated in Live; AB as NA; key check per location; never edits a set open in Live
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+188
-5
@@ -49,10 +49,12 @@ import xml.etree.ElementTree as ET
|
||||
|
||||
STATUS = os.path.expanduser("~/Library/Caches/status-panel-start.json")
|
||||
LOCK = os.path.expanduser("~/Library/Caches/status-panel-start.lock")
|
||||
LOC_STATUS = os.path.expanduser("~/Library/Caches/status-panel-locations.json") # TRACKS OUT's job
|
||||
NOTES = {"C": 0, "D": 2, "E": 4, "F": 5, "G": 7, "A": 9, "B": 11}
|
||||
FINISHED = re.compile(r"^(.*?)\s*\[\s*([A-G](?:#|b)?m?)\s*\]\s*(?:-\s*CUSTOM)?\s*$", re.I)
|
||||
NUMBERED = re.compile(r"^\s*(\d+)\s*-")
|
||||
SALVATION = re.compile(r"salvation", re.I)
|
||||
LOCATOR = re.compile(r"^\s*(\d+)\s*-\s*(.*?)\s*-\s*([A-G](?:#|b)?m?)\s*$") # "3- THE LORD IS HERE - B"
|
||||
# Stems that have no key: never transposed (the workflow: "drums, percussion, aux drums, and other unpitched
|
||||
# sounds"). Sub bass, timpani and FX are pitched. A bare LOOP stem is a drum loop (SYNTH / VOX LOOP are not).
|
||||
UNPITCHED = (r"\b(DRUMS?|PERC(USSION)?|SHAKERS?|TAMB(OURINE)?S?|CYMBALS?|TOMS|KICK|SNARE|HI ?HATS?|HATS|CLAPS?|SNAPS"
|
||||
@@ -146,16 +148,16 @@ def template_for(routing, name):
|
||||
class Status:
|
||||
"""The job's progress, written where the READY look reads it."""
|
||||
|
||||
def __init__(self, data):
|
||||
self.data = data
|
||||
def __init__(self, data, path=None):
|
||||
self.data, self.path = data, path or STATUS
|
||||
self.write()
|
||||
|
||||
def write(self, **kw):
|
||||
self.data.update(kw, time=time.time())
|
||||
tmp = STATUS + ".tmp"
|
||||
tmp = self.path + ".tmp"
|
||||
with open(tmp, "w") as f:
|
||||
json.dump(self.data, f)
|
||||
os.replace(tmp, STATUS)
|
||||
os.replace(tmp, self.path)
|
||||
|
||||
|
||||
# --- reading and writing sets ----------------------------------------------------------------------
|
||||
@@ -1528,6 +1530,186 @@ def consolidate(path, cfg, status, log, live=None):
|
||||
live.close()
|
||||
|
||||
|
||||
# --- the locations' sets (TRACKS OUT) ----------------------------------------------------------------
|
||||
# Waters_Worship_Location_Setup.pdf: each location's set is made from the finished NA set. AB keeps NA's keys; WS
|
||||
# and SC are put in the keys their own Planning Center plan has: the pitched stems transposed the workflow's way
|
||||
# (then consolidated in Live), the locator, the Waves Tune key cue and the pads changed with them, the Call to
|
||||
# Worship and Salvation following the song they go with.
|
||||
|
||||
def location_keys(ls, songs):
|
||||
"""[time, locator, number, title, key in the set, key the location wants (None: left as it is)] for every
|
||||
numbered locator, in order. `songs`: the location plan's songs ({"title", "key"})."""
|
||||
rows = []
|
||||
for l in sorted(ls.findall("Locators/Locators/Locator"), key=lambda l: fval(l, "Time")):
|
||||
m = LOCATOR.match(val(l, "Name") or "")
|
||||
if not m:
|
||||
continue
|
||||
title = m.group(2).strip()
|
||||
want = next((x["key"] for x in songs if key_of(x.get("key")) and (norm(x["title"]) == norm(title) or close_titles(x["title"], title))), None)
|
||||
rows.append([fval(l, "Time"), l, int(m.group(1)), title, m.group(3), want])
|
||||
for i, r in enumerate(rows):
|
||||
if r[5]:
|
||||
continue
|
||||
if norm(r[3]).startswith("CALL TO WORSHIP"): # the first song's key
|
||||
nxt = next((x for x in rows[i + 1:] if x[5]), None)
|
||||
if nxt and same_key(r[4], nxt[4]):
|
||||
r[5] = nxt[5]
|
||||
elif SALVATION.search(r[3]): # the key of the song before it, as in NA
|
||||
prev = next((x for x in reversed(rows[:i]) if x[5] and same_key(x[4], r[4])), None)
|
||||
if prev:
|
||||
r[5] = prev[5]
|
||||
return rows
|
||||
|
||||
|
||||
def localize(path, loc, songs, cfg, log):
|
||||
"""One location's set (a copy of NA's) put in the location's keys, in place (the old .als kept in Backup/).
|
||||
Returns how many locators changed key."""
|
||||
root = load(path)
|
||||
ls = root.find("LiveSet")
|
||||
proj = os.path.dirname(path)
|
||||
rows = location_keys(ls, songs)
|
||||
for x in songs:
|
||||
if not any(norm(x["title"]) == norm(r[3]) or close_titles(x["title"], r[3]) for r in rows):
|
||||
log.append("%s: %s IS NOT IN THE SET" % (loc, x["title"].upper()))
|
||||
for r in rows:
|
||||
if not r[5] and not norm(r[3]).startswith("CALL TO WORSHIP") and not SALVATION.search(r[3]):
|
||||
log.append("%s: NO %s IN ITS PLAN, LEFT IN %s" % (loc, r[3].upper(), r[4]))
|
||||
todo = [(i, r) for i, r in enumerate(rows) if r[5] and semitones(r[4], r[5])]
|
||||
if not todo:
|
||||
return 0
|
||||
by_name = {norm(tname(t)): t for t in tracks(ls)}
|
||||
pads_t, tune_t = by_name.get(norm(cfg["pad_track"])), by_name.get(norm(cfg["tune_track"]))
|
||||
unpitched = cfg.get("unpitched") or UNPITCHED
|
||||
files = Files(proj)
|
||||
groups = [(song_span(family(ls, g))[0], family(ls, g)) for g in song_groups(ls, cfg["keep_groups"])]
|
||||
for i, r in todo:
|
||||
t0, t1 = r[0] - 0.01, (rows[i + 1][0] - 0.01 if i + 1 < len(rows) else 1e12)
|
||||
semis = semitones(r[4], r[5])
|
||||
setv(r[1], "Name", "%d- %s - %s" % (r[2], r[3], r[5]))
|
||||
tune = tune_for(cfg["_tunes"], r[5])
|
||||
for c in clips(tune_t) if tune_t is not None else []:
|
||||
if t0 <= span(c)[0] < t1:
|
||||
if tune:
|
||||
setv(c, "Name", tune[0])
|
||||
setv(c, "ProgramChange", tune[1])
|
||||
else:
|
||||
log.append("%s: NO WAVES TUNE CUE FOR %s" % (loc, r[5]))
|
||||
pad = cfg["_pads"].get((key_of(r[5]) or (None,))[0])
|
||||
ev = events(pads_t) if pads_t is not None else None
|
||||
for c in list(ev) if ev is not None else []:
|
||||
a, b = span(c)
|
||||
if t0 <= a < t1:
|
||||
if pad:
|
||||
new = pad_clip(c, pad, a, b, tempo_at(ls, a))
|
||||
files.adopt(new, None, "")
|
||||
ev.insert(list(ev).index(c), new)
|
||||
ev.remove(c)
|
||||
else:
|
||||
log.append("%s: NO PAD IN %s" % (loc, r[5]))
|
||||
n = 0
|
||||
for start, fam in groups:
|
||||
if start is not None and t0 <= start < t1:
|
||||
for t in fam:
|
||||
if t.tag == "AudioTrack" and pitched(tname(t), unpitched):
|
||||
for c in clips(t):
|
||||
transpose(c, int(fval(c, "PitchCoarse")) + semis, tempo_at(ls, span(c)[0]))
|
||||
n += 1
|
||||
log.append("%s: %s %s>%s%s" % (loc, r[3].upper(), r[4], r[5], " (%d CLIPS)" % n if n else ""))
|
||||
problems = check(root)
|
||||
if problems:
|
||||
raise RuntimeError("%s SET CHECK: %s" % (loc, problems[0]))
|
||||
os.makedirs(os.path.join(proj, "Backup"), exist_ok=True)
|
||||
shutil.copyfile(path, os.path.join(proj, "Backup", "%s [before %s keys %s].als" % (
|
||||
os.path.splitext(os.path.basename(path))[0], loc, time.strftime("%Y-%m-%d %H%M%S"))))
|
||||
files.copy(proj)
|
||||
save(root, path)
|
||||
return len(todo)
|
||||
|
||||
|
||||
def copy_project(src, dst):
|
||||
"""NA's project to a location's folder, the way TRACKS OUT always has: under a temporary name, without
|
||||
Backup/, the set renamed for the location, then put in place."""
|
||||
tmp = dst + ".copying"
|
||||
os.makedirs(tmp, exist_ok=True)
|
||||
subprocess.run(["/usr/bin/rsync", "-a", "--exclude", "Backup/", src + "/", tmp + "/"], check=True, capture_output=True)
|
||||
old = os.path.join(tmp, os.path.basename(src) + ".als")
|
||||
if os.path.exists(old):
|
||||
os.rename(old, os.path.join(tmp, os.path.basename(dst) + ".als"))
|
||||
if os.path.exists(dst):
|
||||
raise RuntimeError("%s APPEARED WHILE COPYING" % os.path.basename(dst))
|
||||
os.rename(tmp, dst)
|
||||
|
||||
|
||||
def run_locations(cfg):
|
||||
"""TRACKS OUT: every location gets this weekend's set; WS and SC in their own keys, consolidated in Live."""
|
||||
status = Status({"state": "running", "started": time.time(), "step": "STARTING", "results": [], "log": []}, LOC_STATUS)
|
||||
log = status.data["log"]
|
||||
try:
|
||||
cfg["_pads"] = pads_index(cfg.get("pads_folder"))
|
||||
cfg["_tunes"] = tune_index(cfg.get("tune_library"))
|
||||
base = os.path.expanduser(cfg["projects"])
|
||||
|
||||
def folder(loc):
|
||||
return os.path.join(base, loc, cfg["folder"].format(date=cfg["date"], loc=loc))
|
||||
src = folder(cfg["site"])
|
||||
src_als = os.path.join(src, os.path.basename(src) + ".als")
|
||||
if not os.path.exists(src_als):
|
||||
raise RuntimeError("NO %s SET" % cfg["site"])
|
||||
if pending(load(src_als), cfg.get("unpitched") or UNPITCHED):
|
||||
raise RuntimeError("%s IS NOT CONSOLIDATED YET" % cfg["site"])
|
||||
work = []
|
||||
for loc in cfg["locations"]:
|
||||
dst = folder(loc)
|
||||
if not os.path.exists(dst) and not os.path.isdir(dst + ".copying"):
|
||||
status.write(step="COPYING TO %s" % loc)
|
||||
copy_project(src, dst)
|
||||
log.append("COPIED TO %s" % loc)
|
||||
songs = (cfg.get("transpose") or {}).get(loc)
|
||||
if songs is None:
|
||||
continue
|
||||
path = os.path.join(dst, os.path.basename(dst) + ".als")
|
||||
if not os.path.exists(path):
|
||||
log.append("%s: NO %s" % (loc, os.path.basename(path)))
|
||||
continue
|
||||
if live_set() == path:
|
||||
# Live holds its own copy: a change on disk would be lost on its next save (or overwrite work in it).
|
||||
log.append("%s: ITS SET IS OPEN IN LIVE: OPEN ANOTHER SET, THEN HOLD TRACKS OUT" % loc)
|
||||
continue
|
||||
status.write(step="%s KEYS" % loc)
|
||||
changed = localize(path, loc, songs, cfg, log)
|
||||
status.data["results"].append({"loc": loc, "keys_changed": changed})
|
||||
work.append(path)
|
||||
live, before = None, live_set()
|
||||
try:
|
||||
for path in work if cfg.get("consolidate", True) else []:
|
||||
if not pending(load(path), cfg.get("unpitched") or UNPITCHED):
|
||||
continue
|
||||
try:
|
||||
live = live or Live()
|
||||
res = consolidate(path, cfg, status, log, live)
|
||||
except RuntimeError as e:
|
||||
log.append("%s: CONSOLIDATE IN LIVE: %s" % (os.path.basename(path)[:-4].upper(), e))
|
||||
continue
|
||||
status.data["results"].append(res)
|
||||
status.write()
|
||||
# Live goes back to the set it had open (NA's).
|
||||
if live and before and os.path.exists(before) and live_set() != before:
|
||||
status.write(step="REOPENING %s" % os.path.basename(before)[:-4].upper()[:14])
|
||||
subprocess.run(["/usr/bin/open", "-a", live_app(), before], capture_output=True, timeout=30)
|
||||
end = time.time() + 180
|
||||
while time.time() < end and live_set() != before:
|
||||
time.sleep(2)
|
||||
if live_set() != before:
|
||||
log.append("OPEN %s IN LIVE AGAIN" % os.path.basename(before)[:-4].upper())
|
||||
finally:
|
||||
if live:
|
||||
live.close()
|
||||
status.write(state="done", step="DONE")
|
||||
except Exception as e: # noqa: BLE001 - the key shows what stopped it
|
||||
import traceback
|
||||
status.write(state="failed", step=str(e)[:60].upper(), trace=traceback.format_exc()[-2000:])
|
||||
|
||||
|
||||
# --- the job ----------------------------------------------------------------------------------
|
||||
|
||||
def projects_dir(cfg):
|
||||
@@ -1654,7 +1836,8 @@ def main(P):
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) > 1: # the detached job
|
||||
try:
|
||||
run(json.load(open(sys.argv[1])))
|
||||
job = json.load(open(sys.argv[1]))
|
||||
(run_locations if job.get("job") == "locations" else run)(job)
|
||||
finally:
|
||||
try:
|
||||
os.remove(LOCK)
|
||||
|
||||
Reference in New Issue
Block a user