christianandClaude Opus 5.5
0d655bff3c
Hold START on Monday: last weekend's songs go to Finished Songs (a changed one parks the version it came from as z...), then each service PCO lists this week (weekend + Special Services / Morning Worship...; First / Together Tuesday in Projects/FIRST TUESDAY) gets 'YYMMDD NA Project' built from the newest weekend set: locators, tempo map and TEMPO clips, Waves Tune cues, songs from Finished Songs (transposed Complex Pro when only another key exists), markers / guide / lighting cues on the dedicated tracks, pads, routing and send A at 0 dB, audio copied in with an Ableton Project Info folder. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1353 lines
57 KiB
Python
1353 lines
57 KiB
Python
#!/usr/bin/env python3
|
||
"""START: the week's Ableton sets, built the way the worship workflow (Waters_Worship_Ableton_Workflow.pdf)
|
||
does them by hand on Mondays. readiness.py sends this file to the Tracks Mac and runs it there, detached
|
||
(stdlib only, /usr/bin/python3 3.9), with the services Planning Center lists for the coming days: Sunday and
|
||
every special service (First / Together Tuesday, Morning Worship, ...).
|
||
|
||
1 the previous weekend set's songs go to Finished Songs as "TITLE [KEY]": the song group with its
|
||
arrangement markers, lighting cues and guide inside it, the audio copied in. A song that changed parks
|
||
the old folder as "zTITLE [KEY]" first; an unchanged one is left as it is.
|
||
2 each new set ("261101 NA Project", "261103 NA First Tuesday Project") starts from the newest weekend
|
||
set: its songs, pads, Waves Tune cues and marker, guide and lighting clips go; the dedicated groups,
|
||
the numbered locators (with their computer-key mappings), the loop locators and the loop cues stay.
|
||
3 the locators are named "N- TITLE - KEY" from PCO, the tempo map and the TEMPO label clips follow the
|
||
songs, and a Waves Tune key cue goes at every locator
|
||
4 every song comes in from Finished Songs at its locator. In another key it is transposed the way the
|
||
workflow says (warp on, Complex Pro, formants 0, transpose) and flagged, because consolidating it is
|
||
done in Live. Its markers, guide and lighting cues go onto the dedicated tracks.
|
||
5 pads: the Call to Worship pad in the first song's key, an ending pad per song past its loop, Salvation
|
||
6 every stem routed per the routing template, send A (in-ears) at 0 dB
|
||
|
||
A song too long for its slot pushes the loop locators, loop cues and later slots along (in whole bars);
|
||
more songs than slots add slots; fewer move Salvation up, as the sets have done by hand.
|
||
|
||
Every audio file the set uses is copied into its project folder (as Collect All and Save does), so the
|
||
project can be sent to the locations as it is. Nothing is overwritten: a set that is there is left alone,
|
||
and a new set is built in "<name>.building" and only renamed when it is complete.
|
||
|
||
Progress and the result go to ~/Library/Caches/status-panel-start.json (the START key on the READY page).
|
||
"""
|
||
import copy
|
||
import gzip
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
import shutil
|
||
import struct
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
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")
|
||
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)
|
||
AUDIO_EVENTS = "DeviceChain/MainSequencer/Sample/ArrangerAutomation/Events"
|
||
MIDI_EVENTS = "DeviceChain/MainSequencer/ClipTimeable/ArrangerAutomation/Events"
|
||
INITIAL = -63072000.0 # the time Live gives an envelope's starting value
|
||
|
||
|
||
# --- small helpers ----------------------------------------------------------------------------
|
||
|
||
def fmt(v):
|
||
if isinstance(v, bool):
|
||
return "true" if v else "false"
|
||
if isinstance(v, float):
|
||
if v == int(v) and abs(v) < 1e15:
|
||
return str(int(v))
|
||
return repr(round(v, 10))
|
||
return str(v)
|
||
|
||
|
||
def val(e, path, default=None):
|
||
x = e.find(path) if path else e
|
||
return x.get("Value") if x is not None else default
|
||
|
||
|
||
def fval(e, path, default=0.0):
|
||
v = val(e, path)
|
||
return float(v) if v not in (None, "") else default
|
||
|
||
|
||
def setv(e, path, v):
|
||
x = e.find(path) if path else e
|
||
if x is not None:
|
||
x.set("Value", fmt(v))
|
||
|
||
|
||
def norm(text):
|
||
t = (text or "").upper().replace("&", " AND ").replace("’", "'")
|
||
t = re.sub(r"\((LIVE|ACOUSTIC|FEAT[^)]*|RADIO[^)]*)\)", " ", t)
|
||
t = re.sub(r"[^A-Z0-9 ]+", " ", t.replace("'", ""))
|
||
return " ".join(t.split())
|
||
|
||
|
||
def key_of(text):
|
||
m = re.match(r"^\s*([A-Ga-g])(#|b|♯|♭)?\s*(m|min|minor)?\s*$", text or "")
|
||
if not m:
|
||
return None
|
||
pc = NOTES[m.group(1).upper()] + {"#": 1, "♯": 1, "b": -1, "♭": -1}.get(m.group(2) or "", 0)
|
||
return pc % 12, bool(m.group(3))
|
||
|
||
|
||
def same_key(a, b, mode=True):
|
||
ka, kb = key_of(a), key_of(b)
|
||
return bool(ka and kb) and ka[0] == kb[0] and (not mode or ka[1] == kb[1])
|
||
|
||
|
||
def semitones(have, want):
|
||
"""The shortest transposition from `have` to `want` (-6..+5)."""
|
||
a, b = key_of(have), key_of(want)
|
||
if not (a and b):
|
||
return None
|
||
return (b[0] - a[0] + 6) % 12 - 6
|
||
|
||
|
||
def bar_up(x, bar=4.0):
|
||
return math.ceil(x / bar - 1e-9) * bar
|
||
|
||
|
||
def stem_name(name):
|
||
s = re.sub(r"\[.*?\]", " ", (name or "").upper())
|
||
s = re.sub(r"^\s*\d+\s*-\s*", "", s)
|
||
return " ".join(s.split())
|
||
|
||
|
||
def template_for(routing, name):
|
||
n = stem_name(name)
|
||
if n in routing:
|
||
return routing[n]
|
||
base = re.sub(r"\s*\d+$", "", n)
|
||
fam = {tuple(v) for k, v in routing.items() if re.sub(r"\s*\d+$", "", k) == base}
|
||
return list(fam.pop()) if len(fam) == 1 else None
|
||
|
||
|
||
class Status:
|
||
"""The job's progress, written where the READY look reads it."""
|
||
|
||
def __init__(self, data):
|
||
self.data = data
|
||
self.write()
|
||
|
||
def write(self, **kw):
|
||
self.data.update(kw, time=time.time())
|
||
tmp = STATUS + ".tmp"
|
||
with open(tmp, "w") as f:
|
||
json.dump(self.data, f)
|
||
os.replace(tmp, STATUS)
|
||
|
||
|
||
# --- reading and writing sets ----------------------------------------------------------------------
|
||
|
||
def load(path):
|
||
with gzip.open(path) as f:
|
||
return ET.parse(f).getroot()
|
||
|
||
|
||
def save(root, path):
|
||
data = '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode")
|
||
tmp = path + ".tmp"
|
||
with gzip.open(tmp, "wb") as f:
|
||
f.write(data.encode("utf-8"))
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def tracks(ls):
|
||
return list(ls.find("Tracks"))
|
||
|
||
|
||
def tname(t):
|
||
return val(t, "Name/EffectiveName", "") or ""
|
||
|
||
|
||
def set_name(t, name):
|
||
setv(t, "Name/EffectiveName", name)
|
||
setv(t, "Name/UserName", name)
|
||
|
||
|
||
def group_of(t):
|
||
return val(t, "TrackGroupId", "-1")
|
||
|
||
|
||
def family(ls, top):
|
||
"""`top` and every track under it, in set order."""
|
||
ids, out = {top.get("Id")}, []
|
||
for t in tracks(ls):
|
||
if t is top or group_of(t) in ids:
|
||
ids.add(t.get("Id"))
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def events(t):
|
||
"""The arrangement's clip list of a track (None for a group or return)."""
|
||
return t.find(AUDIO_EVENTS) if t.tag == "AudioTrack" else t.find(MIDI_EVENTS) if t.tag == "MidiTrack" else None
|
||
|
||
|
||
def clips(t):
|
||
ev = events(t)
|
||
return list(ev) if ev is not None else []
|
||
|
||
|
||
def span(c):
|
||
return fval(c, "CurrentStart"), fval(c, "CurrentEnd")
|
||
|
||
|
||
def warped(c):
|
||
return c.tag != "AudioClip" or val(c, "IsWarped") == "true"
|
||
|
||
|
||
def seconds(c):
|
||
"""An unwarped audio clip's length in the sample (its loop values are seconds)."""
|
||
return fval(c, "Loop/LoopEnd") - fval(c, "Loop/LoopStart")
|
||
|
||
|
||
def clip_tempo(c):
|
||
s, e = span(c)
|
||
secs = seconds(c)
|
||
return (e - s) / secs * 60.0 if secs > 0 and not warped(c) else None
|
||
|
||
|
||
def place(c, start, bpm=None, end=None):
|
||
"""Move a clip to `start`; an unwarped audio clip is as long as its audio lasts at `bpm`."""
|
||
s, e = span(c)
|
||
if end is None:
|
||
end = start + seconds(c) * bpm / 60.0 if (not warped(c) and bpm) else start + (e - s)
|
||
setv(c, "CurrentStart", float(start))
|
||
setv(c, "CurrentEnd", float(end))
|
||
c.set("Time", fmt(float(start)))
|
||
|
||
|
||
def renumber(parent, tag=None):
|
||
for i, e in enumerate(x for x in parent if tag is None or x.tag == tag):
|
||
e.set("Id", str(i))
|
||
|
||
|
||
def is_pointee(e):
|
||
return e.tag in ("AutomationTarget", "Pointee") or e.tag.endswith("ModulationTarget")
|
||
|
||
|
||
def max_pointee(root):
|
||
return max([int(e.get("Id")) for e in root.iter() if is_pointee(e) and (e.get("Id") or "").isdigit()] or [0])
|
||
|
||
|
||
def remap_pointees(elems, next_id):
|
||
"""New ids for everything automation can point at in `elems` (tracks from another set), and the
|
||
envelopes that point at them; returns the next free id."""
|
||
m = {}
|
||
for root in elems:
|
||
for e in root.iter():
|
||
if is_pointee(e) and e.get("Id") is not None:
|
||
m[e.get("Id")] = str(next_id)
|
||
e.set("Id", str(next_id))
|
||
next_id += 1
|
||
for root in elems:
|
||
for e in root.iter("PointeeId"):
|
||
if e.get("Value") in m:
|
||
e.set("Value", m[e.get("Value")])
|
||
return next_id
|
||
|
||
|
||
def shift_automation(t, off):
|
||
for ev in t.findall("AutomationEnvelopes/Envelopes/AutomationEnvelope/Automation/Events"):
|
||
for e in ev:
|
||
tm = float(e.get("Time", "0"))
|
||
if tm > INITIAL + 1:
|
||
e.set("Time", fmt(tm + off))
|
||
|
||
|
||
def fit_slots(t, n_scenes):
|
||
"""As many session clip slots as the set has scenes."""
|
||
for path, tag in (("DeviceChain/MainSequencer/ClipSlotList", "ClipSlot"),
|
||
("DeviceChain/FreezeSequencer/ClipSlotList", "ClipSlot"), ("Slots", "GroupTrackSlot")):
|
||
lst = t.find(path)
|
||
if lst is None:
|
||
continue
|
||
slots = list(lst)
|
||
while len(slots) > n_scenes:
|
||
lst.remove(slots.pop())
|
||
while slots and len(slots) < n_scenes:
|
||
s = copy.deepcopy(slots[-1])
|
||
v = s.find("ClipSlot/Value")
|
||
if v is not None:
|
||
for x in list(v):
|
||
v.remove(x)
|
||
lst.append(s)
|
||
slots.append(s)
|
||
renumber(lst)
|
||
|
||
|
||
def fit_sends(t, holders, send_a):
|
||
"""The set's sends (one per return track), send A at `send_a`."""
|
||
sends = t.find("DeviceChain/Mixer/Sends")
|
||
if sends is None:
|
||
return
|
||
have = list(sends)
|
||
while len(have) > len(holders):
|
||
sends.remove(have.pop())
|
||
while len(have) < len(holders):
|
||
h = copy.deepcopy(holders[len(have)])
|
||
sends.append(h)
|
||
have.append(h)
|
||
renumber(sends)
|
||
if have and send_a is not None:
|
||
setv(have[0], "Send/Manual", float(send_a))
|
||
|
||
|
||
def route(t, want, targets):
|
||
"""Output to a template's (upper, lower) display pair, using the routing target the set has for it."""
|
||
r = t.find("DeviceChain/AudioOutputRouting")
|
||
target = targets.get(tuple(want)) if want else None
|
||
if r is None or target is None:
|
||
return False
|
||
setv(r, "Target", target)
|
||
setv(r, "UpperDisplayString", want[0])
|
||
setv(r, "LowerDisplayString", want[1])
|
||
return True
|
||
|
||
|
||
def audio_info(path):
|
||
"""(frames, sample rate) of a WAV or AIFF file, or None."""
|
||
try:
|
||
with open(path, "rb") as f:
|
||
head = f.read(12)
|
||
if head[:4] == b"RIFF" and head[8:12] == b"WAVE":
|
||
rate = align = None
|
||
while True:
|
||
ch = f.read(8)
|
||
if len(ch) < 8:
|
||
return None
|
||
cid, size = ch[:4], struct.unpack("<I", ch[4:])[0]
|
||
if cid == b"fmt ":
|
||
fmt_ = f.read(size)
|
||
rate, align = struct.unpack("<I", fmt_[4:8])[0], struct.unpack("<H", fmt_[12:14])[0]
|
||
if size % 2:
|
||
f.read(1)
|
||
elif cid == b"data":
|
||
return (size // align, rate) if align else None
|
||
else:
|
||
f.seek(size + (size % 2), 1)
|
||
if head[:4] == b"FORM" and head[8:12] in (b"AIFF", b"AIFC"):
|
||
while True:
|
||
ch = f.read(8)
|
||
if len(ch) < 8:
|
||
return None
|
||
cid, size = ch[:4], struct.unpack(">I", ch[4:])[0]
|
||
if cid == b"COMM":
|
||
d = f.read(size)
|
||
frames = struct.unpack(">I", d[2:6])[0]
|
||
exp = struct.unpack(">H", d[8:10])[0] & 0x7FFF
|
||
mant = struct.unpack(">Q", d[10:18])[0]
|
||
return frames, int(round(mant * 2.0 ** (exp - 16383 - 63)))
|
||
f.seek(size + (size % 2), 1)
|
||
except (OSError, struct.error, ZeroDivisionError):
|
||
return None
|
||
return None
|
||
|
||
|
||
# --- files: every audio file the set uses goes into its project ------------------------------------------
|
||
|
||
def ref_source(fr, proj):
|
||
"""The file a FileRef names, wherever it really is now (a moved or renamed project still counts)."""
|
||
rel, path = val(fr, "RelativePath") or "", val(fr, "Path") or ""
|
||
found = []
|
||
if rel and proj:
|
||
found.append(os.path.join(proj, rel))
|
||
if path:
|
||
found.append(path if os.path.isabs(path) else os.path.join(proj or "", path))
|
||
if proj and os.path.isabs(path) and "/Samples/" in path: # the project was renamed after saving
|
||
found.append(os.path.join(proj, "Samples", path.split("/Samples/", 1)[1]))
|
||
return next((p for p in found if os.path.isfile(p)), None)
|
||
|
||
|
||
class Files:
|
||
"""The copies a set needs: (source, project-relative destination), made at the end."""
|
||
|
||
def __init__(self, proj):
|
||
self.proj, self.todo, self.missing = proj, {}, []
|
||
|
||
def adopt(self, clip, src_proj, sub):
|
||
"""Point an audio clip at its own copy under Samples/Imported/<sub>/ in this project."""
|
||
fr = clip.find("SampleRef/FileRef")
|
||
if fr is None:
|
||
return
|
||
src = ref_source(fr, src_proj)
|
||
if src is None:
|
||
self.missing.append(val(fr, "Path") or val(fr, "RelativePath") or "?")
|
||
return
|
||
rel = os.path.join("Samples", "Imported", sub, os.path.basename(src)) if sub else os.path.join("Samples", "Imported", os.path.basename(src))
|
||
prev = self.todo.get(rel)
|
||
if prev and prev != src: # two different files with one name: keep both
|
||
stem, ext = os.path.splitext(rel)
|
||
i = 2
|
||
while self.todo.get("%s %d%s" % (stem, i, ext)) not in (None, src):
|
||
i += 1
|
||
rel = "%s %d%s" % (stem, i, ext)
|
||
self.todo[rel] = src
|
||
setv(fr, "RelativePathType", 3)
|
||
setv(fr, "RelativePath", rel)
|
||
setv(fr, "Path", os.path.join(self.proj, rel))
|
||
|
||
def keep_in_place(self, clip, src_proj):
|
||
"""A clip of the base set: the same project-relative place in the new project."""
|
||
fr = clip.find("SampleRef/FileRef")
|
||
if fr is None:
|
||
return
|
||
src = ref_source(fr, src_proj)
|
||
if src is None: # offline in the base set too: left as it was
|
||
return
|
||
if src_proj and os.path.abspath(src).startswith(os.path.abspath(src_proj) + os.sep):
|
||
rel = os.path.relpath(src, src_proj)
|
||
self.todo[rel] = src
|
||
if val(fr, "RelativePathType") == "3" or val(fr, "RelativePath"):
|
||
setv(fr, "RelativePathType", 3)
|
||
setv(fr, "RelativePath", rel)
|
||
if os.path.isabs(val(fr, "Path") or ""):
|
||
setv(fr, "Path", os.path.join(self.proj, rel))
|
||
else:
|
||
self.adopt(clip, src_proj, "")
|
||
|
||
def size(self):
|
||
return sum(os.path.getsize(s) for s in self.todo.values() if os.path.exists(s))
|
||
|
||
def copy(self, into, progress=None):
|
||
done = 0
|
||
for rel, src in sorted(self.todo.items()):
|
||
dst = os.path.join(into, rel)
|
||
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
||
if not (os.path.exists(dst) and os.path.getsize(dst) == os.path.getsize(src)):
|
||
# A plain copy, as Live's Collect All and Save makes: an APFS clone of a Google Drive file
|
||
# reads fine in a shell but Live could not open one (2026-10-22).
|
||
tmp = dst + ".part"
|
||
shutil.copyfile(src, tmp)
|
||
os.replace(tmp, dst)
|
||
if os.path.exists(src + ".asd") and not os.path.exists(dst + ".asd"):
|
||
shutil.copyfile(src + ".asd", dst + ".asd")
|
||
done += 1
|
||
if progress:
|
||
progress(done, len(self.todo))
|
||
|
||
|
||
# --- the library on the shared drive -----------------------------------------------------------------
|
||
|
||
def finished_index(folders):
|
||
"""[(folder path, title, key)] of every finished song (not z-parked, not unfinished)."""
|
||
out = []
|
||
for d in folders:
|
||
d = os.path.expanduser(d)
|
||
for n in sorted(os.listdir(d)) if os.path.isdir(d) else []:
|
||
m = FINISHED.match(n)
|
||
if not m or n.lower().startswith("z") or "UNFINISHED" in n.upper():
|
||
continue
|
||
out.append((os.path.join(d, n), m.group(1).strip(), m.group(2)))
|
||
return out
|
||
|
||
|
||
def finished_als(folder):
|
||
found = [os.path.join(folder, n) for n in os.listdir(folder) if n.endswith(".als") and not n.startswith(".")]
|
||
return max(found, key=os.path.getmtime) if found else None
|
||
|
||
|
||
def close_titles(a, b):
|
||
""""CAN'T HELP BUT PRAISE" for "Can't Help But Praise LM", but not "PRAISE" for "Owe You Praise"."""
|
||
wa, wb = norm(a).split(), norm(b).split()
|
||
short_, long_ = sorted((wa, wb), key=len)
|
||
if len(short_) < 2 or not long_:
|
||
return False
|
||
joined = " " + " ".join(long_) + " "
|
||
return (" " + " ".join(short_) + " ") in joined and len(short_) >= 0.6 * len(long_)
|
||
|
||
|
||
def pick_version(index, title, key):
|
||
"""The finished song for a PCO title: in its key if there is one, else the nearest key (to transpose)."""
|
||
exact = [x for x in index if norm(x[1]) == norm(title)]
|
||
near = [x for x in index if close_titles(x[1], title)]
|
||
for pool in (exact, near):
|
||
if not pool:
|
||
continue
|
||
same = [x for x in pool if same_key(x[2], key)]
|
||
if same:
|
||
return same[0], 0
|
||
scored = sorted(((abs(semitones(x[2], key)) if semitones(x[2], key) is not None else 99, x) for x in pool),
|
||
key=lambda sx: sx[0])
|
||
best = scored[0][1]
|
||
return best, semitones(best[2], key) or 0
|
||
return None, 0
|
||
|
||
|
||
def pads_index(folder):
|
||
folder = os.path.expanduser(folder or "")
|
||
out = {}
|
||
for n in sorted(os.listdir(folder)) if os.path.isdir(folder) else []:
|
||
m = re.match(r"^\s*([A-G](?:#|b)?)\s*-", n)
|
||
if m and n.lower().endswith((".wav", ".aif", ".aiff")):
|
||
out[key_of(m.group(1))[0]] = os.path.join(folder, n)
|
||
return out
|
||
|
||
|
||
def tune_index(folder):
|
||
"""{pitch class: (clip name, program)} from the Waves Tune clip library (major keys)."""
|
||
folder = os.path.expanduser(folder or "")
|
||
out = {}
|
||
for n in sorted(os.listdir(folder)) if os.path.isdir(folder) else []:
|
||
m = re.match(r"^\s*([A-G](?:#|b)?)(?:[:/][A-G](?:#|b)?)?\s*maj\b", n)
|
||
if not (m and n.endswith(".alc")):
|
||
continue
|
||
try:
|
||
pg = load(os.path.join(folder, n)).find(".//MidiClip/ProgramChange")
|
||
except (OSError, ET.ParseError, EOFError):
|
||
continue
|
||
if pg is not None:
|
||
out[key_of(m.group(1))[0]] = (n[:-4].replace(":", "/"), int(pg.get("Value")))
|
||
return out
|
||
|
||
|
||
def tune_for(tunes, key):
|
||
"""Waves Tune corrects to a major scale: a minor song uses its relative major."""
|
||
k = key_of(key)
|
||
if not k:
|
||
return None
|
||
return tunes.get((k[0] + 3) % 12 if k[1] else k[0])
|
||
|
||
|
||
# --- 1: the previous set's songs to Finished Songs ------------------------------------------------------
|
||
|
||
def song_groups(ls, keep):
|
||
"""Top-level groups that hold a song (audio clips), not the set's own groups."""
|
||
keep = {norm(k) for k in keep}
|
||
out = []
|
||
for t in tracks(ls):
|
||
if t.tag != "GroupTrack" or group_of(t) != "-1" or norm(tname(t)) in keep:
|
||
continue
|
||
fam = family(ls, t)
|
||
if any(clips(x) for x in fam if x.tag == "AudioTrack"):
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def song_span(fam):
|
||
cs = [span(c) for x in fam if x.tag == "AudioTrack" for c in clips(x)]
|
||
return (min(s for s, _ in cs), max(e for _, e in cs)) if cs else (None, None)
|
||
|
||
|
||
def locator_key(ls, start):
|
||
"""The key the set's locator gives the song starting at `start` ("3- JEHOVAH - B")."""
|
||
for l in ls.findall("Locators/Locators/Locator"):
|
||
if abs(fval(l, "Time") - start) <= 8 and NUMBERED.match(val(l, "Name") or ""):
|
||
m = re.search(r"-\s*([A-G](?:#|b)?m?)\s*$", val(l, "Name") or "")
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
|
||
def tempo_at(ls, beat):
|
||
tempo = ls.find("MasterTrack/DeviceChain/Mixer/Tempo")
|
||
tid = tempo.find("AutomationTarget").get("Id")
|
||
bpm = fval(tempo, "Manual", 120.0)
|
||
for env in ls.findall("MasterTrack/AutomationEnvelopes/Envelopes/AutomationEnvelope"):
|
||
if val(env, "EnvelopeTarget/PointeeId") == tid:
|
||
for e in env.findall("Automation/Events/FloatEvent"):
|
||
if float(e.get("Time")) <= beat + 1e-6:
|
||
bpm = float(e.get("Value"))
|
||
return bpm
|
||
|
||
|
||
def stems_signature(root, proj, skip_guide=True):
|
||
"""What a song is: its stems' file names and sizes (a re-consolidated song has new ones)."""
|
||
sig = []
|
||
for c in root.iter("AudioClip"):
|
||
fr = c.find("SampleRef/FileRef")
|
||
if fr is None:
|
||
continue
|
||
name = os.path.basename(val(fr, "Path") or val(fr, "RelativePath") or "")
|
||
if skip_guide and stem_name(os.path.splitext(name)[0]).startswith("GUIDE"):
|
||
continue
|
||
stem, ext = os.path.splitext(name)
|
||
sig.append((re.sub(r"-\d+$", "", stem) + ext, val(fr, "OriginalFileSize"))) # Collect's "Drums (Live)-1.wav"
|
||
return sorted(sig)
|
||
|
||
|
||
def helper_copy(src, gid, new_id, keep_from, keep_to, name):
|
||
"""A dedicated track (SECTIONS, LIGHTING CUES, +GUIDE) cut down to one song, inside its group."""
|
||
t = copy.deepcopy(src)
|
||
t.set("Id", str(new_id))
|
||
setv(t, "TrackGroupId", gid)
|
||
set_name(t, name)
|
||
ev = events(t)
|
||
for c in list(ev) if ev is not None else []:
|
||
s, e = span(c)
|
||
if not (s < keep_to and e > keep_from):
|
||
ev.remove(c)
|
||
renumber(ev) if ev is not None else None
|
||
return t
|
||
|
||
|
||
def export_songs(prev_path, cfg, status, log):
|
||
"""Step 1: every song of the previous weekend set to Finished Songs (new or changed ones only)."""
|
||
root = load(prev_path)
|
||
ls = root.find("LiveSet")
|
||
proj = os.path.dirname(prev_path)
|
||
dest_dir = os.path.expanduser(cfg.get("finished_out") or cfg["finished_folders"][0])
|
||
by_name = {norm(tname(t)): t for t in tracks(ls)}
|
||
dedicated = {k: by_name.get(norm(cfg[k])) for k in ("sections_track", "lighting_track", "guide_track")}
|
||
index = {n.upper(): n for n in os.listdir(dest_dir)} if os.path.isdir(dest_dir) else {}
|
||
done = []
|
||
for g in song_groups(ls, cfg["keep_groups"]):
|
||
fam = family(ls, g)
|
||
start, end = song_span(fam)
|
||
title = tname(g).strip()
|
||
key = locator_key(ls, start)
|
||
if not key or not title:
|
||
log.append("NOT SAVED %s: no key on its locator" % title)
|
||
continue
|
||
mine = stems_signature(_wrap(fam), proj)
|
||
versions = list(saved_versions(cfg, dest_dir, title))
|
||
# Saved already, under this name or a near one ("CAN'T HELP BUT PRAISE LM [Eb]" for a group called
|
||
# "CAN'T HELP BUT PRAISE"), parked or not: then it came from there and is left as it is.
|
||
if any(sig == mine for _, sig in versions):
|
||
continue
|
||
# Changed: it replaces the version it was made from (most of the same audio, same key), else a new name.
|
||
def overlap(sig):
|
||
a, b = set(sig), set(mine)
|
||
return len(a & b) / float(len(a | b) or 1)
|
||
live = [(n, sig) for n, sig in versions if not n.lower().startswith("z") and FINISHED.match(n) and same_key(FINISHED.match(n).group(2), key)
|
||
and os.path.dirname(n) == "" and n in index.values()]
|
||
best = max(live, key=lambda ns: overlap(ns[1]), default=None)
|
||
folder = best[0] if best and overlap(best[1]) >= 0.5 else "%s [%s]" % (title, key)
|
||
title = FINISHED.match(folder).group(1).strip()
|
||
existing = index.get(folder.upper())
|
||
status.write(step="SAVING %s" % title.upper()[:16])
|
||
out_root = copy.deepcopy(root)
|
||
out_ls = out_root.find("LiveSet")
|
||
keep_ids = {t.get("Id") for t in fam}
|
||
tr = out_ls.find("Tracks")
|
||
for t in list(tr):
|
||
if t.tag != "ReturnTrack" and t.get("Id") not in keep_ids:
|
||
tr.remove(t)
|
||
next_tid = max(int(t.get("Id")) for t in tracks(ls)) + 1
|
||
helpers = []
|
||
for k, label in (("sections_track", "SECTIONS"), ("lighting_track", "LIGHTING CUES"), ("guide_track", "Guide")):
|
||
if dedicated[k] is not None:
|
||
helpers.append(helper_copy(dedicated[k], g.get("Id"), next_tid, start - 0.01, end + 0.01, label))
|
||
next_tid += 1
|
||
first_child = [i for i, t in enumerate(list(tr)) if t.get("Id") in keep_ids][0] + 1
|
||
for i, h in enumerate(helpers):
|
||
tr.insert(first_child + i, h)
|
||
nxt = remap_pointees(helpers, max_pointee(out_root) + 1)
|
||
setv(out_ls, "NextPointeeId", nxt)
|
||
for l in out_ls.findall("Locators/Locators/Locator"):
|
||
out_ls.find("Locators/Locators").remove(l)
|
||
set_tempo(out_ls, [(0.0, tempo_at(ls, start))])
|
||
dest = os.path.join(dest_dir, folder)
|
||
building = dest + ".saving"
|
||
files = Files(building)
|
||
for t in tracks(out_ls):
|
||
for c in clips(t) if t.tag == "AudioTrack" else []:
|
||
files.adopt(c, proj, "")
|
||
shutil.rmtree(building, ignore_errors=True)
|
||
make_project(building)
|
||
if not cfg.get("no_copy"):
|
||
files.copy(building)
|
||
# Paths are written for the final folder name.
|
||
for c in out_root.iter("AudioClip"):
|
||
fr = c.find("SampleRef/FileRef")
|
||
if fr is not None and (val(fr, "Path") or "").startswith(building + os.sep):
|
||
setv(fr, "Path", dest + val(fr, "Path")[len(building):])
|
||
save(out_root, os.path.join(building, "%s.als" % title))
|
||
if existing:
|
||
parked = "z" + existing
|
||
while os.path.exists(os.path.join(dest_dir, parked)):
|
||
parked = "z" + parked
|
||
os.rename(os.path.join(dest_dir, existing), os.path.join(dest_dir, parked))
|
||
log.append("PARKED %s AS %s" % (existing, parked))
|
||
os.rename(building, dest)
|
||
index[folder.upper()] = folder
|
||
done.append(folder)
|
||
if files.missing:
|
||
log.append("%s: %d FILES MISSING" % (folder, len(files.missing)))
|
||
return done
|
||
|
||
|
||
def saved_versions(cfg, dest_dir, title):
|
||
"""The stems of every saved version of a song (any key, z-parked ones too)."""
|
||
dirs = [dest_dir] + [os.path.expanduser(d) for d in cfg["finished_folders"] if os.path.expanduser(d) != dest_dir]
|
||
for d in dirs:
|
||
for n in sorted(os.listdir(d)) if os.path.isdir(d) else []:
|
||
m = FINISHED.match(n.lstrip("zZvV") if n[:1] in "zZ" else n)
|
||
if not m or not (norm(m.group(1)) == norm(title) or close_titles(m.group(1), title)):
|
||
continue
|
||
als = finished_als(os.path.join(d, n)) if os.path.isdir(os.path.join(d, n)) else None
|
||
if not als:
|
||
continue
|
||
try:
|
||
yield (n if d == dest_dir else os.path.join(d, n)), stems_signature(load(als), os.path.join(d, n))
|
||
except (OSError, ET.ParseError, EOFError):
|
||
continue
|
||
|
||
|
||
def make_project(path):
|
||
"""A Live project folder. Live finds a project's root by its "Ableton Project Info" folder and looks for
|
||
the set's audio (relative to the project) from there: without it every clip is offline (2026-10-22)."""
|
||
os.makedirs(os.path.join(path, "Ableton Project Info"))
|
||
|
||
|
||
def _wrap(elems):
|
||
w = ET.Element("Tracks")
|
||
w.extend(elems)
|
||
return w
|
||
|
||
|
||
# --- 2..6: a new set ----------------------------------------------------------------------------
|
||
|
||
def set_tempo(ls, regions):
|
||
"""The master tempo: a step at the start of each (start, bpm) region."""
|
||
tempo = ls.find("MasterTrack/DeviceChain/Mixer/Tempo")
|
||
setv(tempo, "Manual", float(regions[0][1]))
|
||
tid = tempo.find("AutomationTarget").get("Id")
|
||
env = next((e for e in ls.findall("MasterTrack/AutomationEnvelopes/Envelopes/AutomationEnvelope")
|
||
if val(e, "EnvelopeTarget/PointeeId") == tid), None)
|
||
if env is None:
|
||
return
|
||
ev = env.find("Automation/Events")
|
||
for e in list(ev):
|
||
ev.remove(e)
|
||
pts = [(INITIAL, regions[0][1])]
|
||
prev = regions[0][1]
|
||
for start, bpm in regions[1:]:
|
||
if abs(bpm - prev) > 1e-6:
|
||
pts += [(start, prev), (start, bpm)]
|
||
prev = bpm
|
||
for i, (tm, v) in enumerate(pts):
|
||
ET.SubElement(ev, "FloatEvent", {"Id": str(i), "Time": fmt(float(tm)), "Value": fmt(float(v))})
|
||
|
||
|
||
class Slot:
|
||
def __init__(self, kind, start, loops=None, cues=None, kept=None, locator=None):
|
||
self.kind, self.start = kind, start
|
||
self.loops = loops or [] # [(rel beat, locator element)]
|
||
self.cues = cues or [] # [(rel beat, Loop Cue clip)]
|
||
self.kept = kept or [] # [(rel beat, track, clip)] marker/lighting/guide clips that belong to the slot
|
||
self.locator = locator
|
||
self.song = None
|
||
self.length = None # beats the song lasts
|
||
self.number = None
|
||
self.bpm = None
|
||
self.key = None
|
||
self.title = None
|
||
|
||
|
||
def base_slots(ls, loop_cue, dedicated):
|
||
"""The base set's slots: each numbered locator, its loop locators, its loop cues and (for Call to
|
||
Worship and Salvation) its own marker / lighting / guide clips."""
|
||
locs = sorted(ls.findall("Locators/Locators/Locator"), key=lambda l: fval(l, "Time"))
|
||
heads = [l for l in locs if NUMBERED.match(val(l, "Name") or "")]
|
||
if not heads:
|
||
raise RuntimeError("THE BASE SET HAS NO NUMBERED LOCATORS")
|
||
slots = []
|
||
for i, h in enumerate(heads):
|
||
start = fval(h, "Time")
|
||
end = fval(heads[i + 1], "Time") if i + 1 < len(heads) else 1e12
|
||
kind = "ctw" if i == 0 else "sal" if SALVATION.search(val(h, "Name") or "") else "song"
|
||
s = Slot(kind, start, locator=h)
|
||
s.loops = [(fval(l, "Time") - start, l) for l in locs
|
||
if l not in heads and (val(l, "Name") or "").strip().upper().startswith("LOOP") and start <= fval(l, "Time") < end]
|
||
s.cues = [(span(c)[0] - start, c) for c in (clips(loop_cue) if loop_cue is not None else [])
|
||
if start <= span(c)[0] < end and c.find(".//MidiNoteEvent") is not None]
|
||
if kind != "song":
|
||
for t in dedicated:
|
||
s.kept += [(span(c)[0] - start, t, c) for c in clips(t) if start <= span(c)[0] < end]
|
||
s.base_end = end
|
||
slots.append(s)
|
||
if slots[-1].kind != "sal":
|
||
slots[-1].kind = "sal"
|
||
return slots
|
||
|
||
|
||
def plan_layout(slots, songs, sal):
|
||
"""Slots for Call to Worship, the songs and Salvation, pushed along where a song is too long."""
|
||
ctw = slots[0]
|
||
base_songs = [s for s in slots if s.kind == "song"]
|
||
sal_slot = [s for s in slots if s.kind == "sal"][-1]
|
||
out = [ctw]
|
||
ctw.slot_len = (base_songs[0].start if base_songs else sal_slot.start) - ctw.start
|
||
delta = 0.0
|
||
for i, song in enumerate(songs):
|
||
if i < len(base_songs):
|
||
s = copy.copy(base_songs[i])
|
||
s.start = base_songs[i].start + delta
|
||
gap = (base_songs[i + 1].start if i + 1 < len(base_songs) else sal_slot.start) - base_songs[i].start
|
||
else: # more songs than the base set had slots: like the last one
|
||
tmpl = base_songs[-1] if base_songs else sal_slot
|
||
s = copy.copy(tmpl)
|
||
prev = out[-1]
|
||
s.start = bar_up(prev.start + prev.slot_len)
|
||
first = min([r for r, _ in tmpl.loops] or [0.0])
|
||
s.loops = [(r - first, l) for r, l in tmpl.loops]
|
||
s.cues = [(r - first, c) for r, c in tmpl.cues]
|
||
gap = None
|
||
s.loops, s.cues = list(s.loops), list(s.cues)
|
||
s.song, s.length = song, song.get("length")
|
||
length = s.length or 0.0
|
||
first_loop = min([r for r, _ in s.loops] + [r for r, _ in s.cues]) if (s.loops or s.cues) else None
|
||
if gap is None: # a new slot: its loop right after the song
|
||
push = bar_up(length + 4.0)
|
||
elif first_loop is not None and length + 4.0 > first_loop:
|
||
push = bar_up(length + 4.0 - first_loop)
|
||
else:
|
||
push = 0.0
|
||
if push:
|
||
s.loops = [(r + push, l) for r, l in s.loops]
|
||
s.cues = [(r + push, c) for r, c in s.cues]
|
||
tail = max([r for r, _ in s.cues] + [r for r, _ in s.loops] + [length])
|
||
s.slot_len = (gap + push) if gap is not None else bar_up(tail + 12.0)
|
||
if gap is not None:
|
||
delta += push
|
||
out.append(s)
|
||
# Salvation: in the first slot no song took (as "5- SALVATION" did when there were 3 songs), else after the last.
|
||
s = copy.copy(sal_slot)
|
||
s.loops, s.cues, s.kept = list(sal_slot.loops), list(sal_slot.cues), list(sal_slot.kept)
|
||
if len(songs) < len(base_songs):
|
||
s.start = base_songs[len(songs)].start + delta
|
||
elif len(songs) == len(base_songs):
|
||
s.start = sal_slot.start + delta
|
||
else:
|
||
prev = out[-1]
|
||
s.start = bar_up(prev.start + prev.slot_len)
|
||
s.song = sal
|
||
s.slot_len = max([r for r, _ in s.cues] + [r for r, _ in s.loops] + [64.0]) + 32.0
|
||
ctw.slot_len = out[1].start - ctw.start if len(out) > 1 else s.start - ctw.start
|
||
out.append(s)
|
||
for n, slot in enumerate(out, 1):
|
||
slot.number = n
|
||
return out
|
||
|
||
|
||
def clone_clip(tmpl, start, end, name):
|
||
c = copy.deepcopy(tmpl)
|
||
setv(c, "CurrentStart", float(start))
|
||
setv(c, "CurrentEnd", float(end))
|
||
c.set("Time", fmt(float(start)))
|
||
setv(c, "Name", name)
|
||
return c
|
||
|
||
|
||
def pad_clip(tmpl, path, start, end, bpm):
|
||
"""An ambient pad, unwarped (as the sets have them), from the start of the file."""
|
||
c = clone_clip(tmpl, start, end, os.path.splitext(os.path.basename(path))[0])
|
||
info = audio_info(path)
|
||
total = info[0] / float(info[1]) if info else 600.0
|
||
secs = min((end - start) * 60.0 / bpm, total)
|
||
setv(c, "CurrentEnd", float(start + secs * bpm / 60.0))
|
||
setv(c, "IsWarped", False)
|
||
for k, v in (("LoopStart", 0.0), ("LoopEnd", secs), ("StartRelative", 0.0), ("LoopOn", False),
|
||
("OutMarker", total), ("HiddenLoopStart", 0.0), ("HiddenLoopEnd", total)):
|
||
setv(c, "Loop/" + k, v)
|
||
sr = c.find("SampleRef")
|
||
fr = sr.find("FileRef")
|
||
setv(fr, "Path", path)
|
||
setv(fr, "RelativePath", "")
|
||
setv(fr, "RelativePathType", 0)
|
||
setv(fr, "OriginalFileSize", os.path.getsize(path))
|
||
setv(fr, "OriginalCrc", 0)
|
||
setv(sr, "LastModDate", int(os.path.getmtime(path)))
|
||
if info:
|
||
setv(sr, "DefaultDuration", info[0])
|
||
setv(sr, "DefaultSampleRate", info[1])
|
||
wm = c.find("WarpMarkers")
|
||
if wm is not None:
|
||
for x in list(wm):
|
||
wm.remove(x)
|
||
for i, (sec, beat) in enumerate(((0.0, 0.0), (total, total * bpm / 60.0))):
|
||
ET.SubElement(wm, "WarpMarker", {"Id": str(i), "SecTime": fmt(sec), "BeatTime": fmt(beat)})
|
||
return c
|
||
|
||
|
||
def transpose(c, semis, bpm):
|
||
"""The workflow's transpose: warp on, Complex Pro, formants 0, transpose (consolidating is left to Live)."""
|
||
if not warped(c):
|
||
s, e = span(c)
|
||
ls_, le = fval(c, "Loop/LoopStart"), fval(c, "Loop/LoopEnd")
|
||
total = fval(c, "Loop/HiddenLoopEnd", le)
|
||
to_beat = bpm / 60.0
|
||
setv(c, "IsWarped", True)
|
||
for k, v in (("LoopStart", ls_ * to_beat), ("LoopEnd", le * to_beat), ("StartRelative", 0.0),
|
||
("OutMarker", total * to_beat), ("HiddenLoopStart", 0.0), ("HiddenLoopEnd", total * to_beat)):
|
||
setv(c, "Loop/" + k, v)
|
||
wm = c.find("WarpMarkers")
|
||
if wm is not None:
|
||
for x in list(wm):
|
||
wm.remove(x)
|
||
for i, (sec, beat) in enumerate(((0.0, 0.0), (max(total, 1.0), max(total, 1.0) * to_beat))):
|
||
ET.SubElement(wm, "WarpMarker", {"Id": str(i), "SecTime": fmt(sec), "BeatTime": fmt(beat)})
|
||
setv(c, "WarpMode", 6)
|
||
setv(c, "ComplexProFormants", 0)
|
||
setv(c, "PitchCoarse", int(semis))
|
||
|
||
|
||
def dated(t):
|
||
"""The consolidation stamp in a track's name ("4-Guide [2026-09-28 141814]"), for picking the newest."""
|
||
m = re.search(r"\[(\d{4}-\d\d-\d\d \d{6})\]", tname(t))
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def song_parts(fam):
|
||
"""A finished song's group split into its stems and the three helper tracks the workflow puts in it
|
||
(one of each: a song saved twice can hold two guides; the newest is the one)."""
|
||
stems, sections, lighting, guide = [], [], [], []
|
||
for t in fam[1:]:
|
||
cs = clips(t)
|
||
if t.tag == "MidiTrack":
|
||
if any((val(c, "Name") or "").upper().startswith("LIGHTING CUES") for c in cs) or "LIGHT" in tname(t).upper():
|
||
lighting.append(t)
|
||
else:
|
||
sections.append(t)
|
||
elif t.tag == "AudioTrack" and stem_name(tname(t)).startswith("GUIDE"):
|
||
guide.append(t)
|
||
elif t.tag in ("AudioTrack", "GroupTrack"):
|
||
stems.append(t)
|
||
pick = lambda ts, key: clips(max(ts, key=key)) if ts else []
|
||
return (stems, pick(sections, lambda t: len(clips(t))), pick(lighting, lambda t: len(clips(t))),
|
||
pick(guide, lambda t: (dated(t), -fam.index(t))))
|
||
|
||
|
||
def load_song(entry, cfg):
|
||
"""A finished song: its tracks, helper clips, start beat and the tempo its audio was cut at."""
|
||
folder = entry[0]
|
||
als = finished_als(folder)
|
||
if not als:
|
||
return None
|
||
root = load(als)
|
||
ls = root.find("LiveSet")
|
||
tops = [t for t in tracks(ls) if t.tag == "GroupTrack" and group_of(t) == "-1"]
|
||
if not tops:
|
||
return None
|
||
fam = family(ls, tops[0])
|
||
stems, sections, lighting, guide = song_parts(fam)
|
||
audio = [c for t in stems for c in clips(t) if c.tag == "AudioClip"]
|
||
if not audio:
|
||
return None
|
||
start = min(span(c)[0] for c in audio)
|
||
tempos = sorted(t for t in (clip_tempo(c) for c in audio) if t)
|
||
bpm = tempos[len(tempos) // 2] if tempos else tempo_at(ls, start)
|
||
end = max(start + (seconds(c) * bpm / 60.0 if not warped(c) else span(c)[1] - span(c)[0]) for c in audio)
|
||
return {"root": root, "ls": ls, "proj": folder, "group": tops[0],
|
||
"stems": stems, "sections": sections, "lighting": lighting, "guide": guide,
|
||
"start": start, "bpm": round(bpm, 2), "length": end - start, "folder": os.path.basename(folder),
|
||
"title": entry[1], "key": entry[2]}
|
||
|
||
|
||
def build_set(svc, base_path, out_dir, name, cfg, status, log):
|
||
"""Steps 2-6 for one service; returns a summary."""
|
||
root = load(base_path)
|
||
ls = root.find("LiveSet")
|
||
base_proj = os.path.dirname(base_path)
|
||
final_proj = os.path.join(out_dir, name)
|
||
building = final_proj + ".building"
|
||
files = Files(final_proj)
|
||
by_name = {norm(tname(t)): t for t in tracks(ls)}
|
||
|
||
def track(k):
|
||
return by_name.get(norm(cfg[k]))
|
||
loop_cue, pads_t, tune_t, tempo_t = track("loop_cue_track"), track("pad_track"), track("tune_track"), track("tempo_track")
|
||
dedicated = [t for t in (track("sections_track"), track("lighting_track"), track("guide_track")) if t is not None]
|
||
cleared = [t for t in (pads_t, tune_t, track("loopguide_track")) if t is not None] + dedicated
|
||
tmpl = {"pad": (clips(pads_t) or [None])[0] if pads_t is not None else None,
|
||
"tune": (clips(tune_t) or [None])[0] if tune_t is not None else None,
|
||
"tempo": (clips(tempo_t) or [None])[0] if tempo_t is not None else None,
|
||
"cue": (clips(loop_cue) or [None])[0] if loop_cue is not None else None}
|
||
slots = base_slots(ls, loop_cue, dedicated)
|
||
|
||
# The songs, from Finished Songs.
|
||
index = finished_index(cfg["finished_folders"] + ([cfg["finished_out"]] if cfg.get("finished_out") else []))
|
||
songs = []
|
||
for s in svc["songs"]:
|
||
entry, semis = pick_version(index, s["title"], s["key"])
|
||
got = load_song(entry, cfg) if entry else None
|
||
info = {"title": s["title"], "key": s["key"], "pco_bpm": s.get("bpm")}
|
||
if got:
|
||
info.update(got, semis=semis, key=s["key"], from_key=got["key"])
|
||
if semis:
|
||
log.append("TRANSPOSE %s %s>%s: CONSOLIDATE IN LIVE" % (got["title"].upper(), got["key"], s["key"]))
|
||
if s.get("bpm") and abs(float(s["bpm"]) - got["bpm"]) > 0.6:
|
||
log.append("TEMPO %s: PCO %s, SONG %g" % (got["title"].upper(), s["bpm"], got["bpm"]))
|
||
else:
|
||
info.update(length=None, bpm=float(s.get("bpm") or 0) or None)
|
||
log.append("NEW SONG %s: BUILD IT BY HAND" % s["title"].upper())
|
||
songs.append(info)
|
||
if not songs:
|
||
raise RuntimeError("NO SONGS IN PCO")
|
||
first_bpm = next((x["bpm"] for x in songs if x.get("bpm")), 120.0)
|
||
sal = svc.get("salvation") or {}
|
||
sal_song = next((x for x in songs if sal.get("title") and norm(x["title"]) == norm(sal["title"])), None)
|
||
sal_bpm = (sal_song or {}).get("bpm") or (float(sal["bpm"]) if sal.get("bpm") else None) or songs[-1].get("bpm") or first_bpm
|
||
sal_key = sal.get("key") or songs[-1]["key"]
|
||
layout = plan_layout(slots, songs, {"title": "SALVATION", "key": sal_key, "bpm": sal_bpm})
|
||
|
||
# Where each display pair ("Ext. Out", "7/8 GTRS") really goes, from every track before any go.
|
||
targets = {}
|
||
for t in tracks(ls):
|
||
r = t.find("DeviceChain/AudioOutputRouting")
|
||
if r is not None and val(r, "UpperDisplayString"):
|
||
targets.setdefault((val(r, "UpperDisplayString"), val(r, "LowerDisplayString") or ""), val(r, "Target"))
|
||
# Out with last week: song groups and the per-song clips.
|
||
tr = ls.find("Tracks")
|
||
insert_at = None
|
||
for g in song_groups(ls, cfg["keep_groups"]):
|
||
for t in family(ls, g):
|
||
if insert_at is None:
|
||
insert_at = list(tr).index(t)
|
||
tr.remove(t)
|
||
for t in cleared + ([loop_cue] if loop_cue is not None else []) + ([tempo_t] if tempo_t is not None else []):
|
||
ev = events(t)
|
||
for c in list(ev) if ev is not None else []:
|
||
ev.remove(c)
|
||
loc_parent = ls.find("Locators/Locators")
|
||
numbered = {int(NUMBERED.match(val(l, "Name")).group(1)): l for l in loc_parent if NUMBERED.match(val(l, "Name") or "")}
|
||
for l in list(loc_parent):
|
||
loc_parent.remove(l)
|
||
|
||
n_scenes = len(list(ls.find("Scenes")))
|
||
holders = []
|
||
for t in tracks(ls):
|
||
sends = t.find("DeviceChain/Mixer/Sends")
|
||
if sends is not None and len(sends):
|
||
holders = list(sends)
|
||
break
|
||
routing = {stem_name(k): v for k, v in (cfg.get("routing") or {}).items()}
|
||
next_tid = max(int(t.get("Id")) for t in tracks(ls)) + 1
|
||
next_ptr = max(max_pointee(root), int(val(ls, "NextPointeeId", "0"))) + 1
|
||
if insert_at is None:
|
||
rets = [i for i, t in enumerate(list(tr)) if t.tag == "ReturnTrack"]
|
||
insert_at = rets[0] if rets else len(list(tr))
|
||
|
||
def add_clip(t, c):
|
||
ev = events(t)
|
||
ev.append(c)
|
||
renumber(ev)
|
||
|
||
regions, summary = [], {"name": name, "songs": [], "service": svc.get("label")}
|
||
for slot in layout:
|
||
s = slot.start
|
||
song = slot.song if slot.kind == "song" else None
|
||
if slot.kind == "ctw":
|
||
slot.key, slot.bpm, slot.title = songs[0]["key"], first_bpm, "CALL TO WORSHIP"
|
||
elif slot.kind == "sal":
|
||
slot.key, slot.bpm, slot.title = sal_key, sal_bpm, "SALVATION"
|
||
else:
|
||
slot.key, slot.bpm = song["key"], song.get("bpm") or first_bpm
|
||
slot.title = (song.get("title") or song["title"]).upper()
|
||
regions.append((s, slot.bpm))
|
||
# Locator: the base set's locator with this number (it carries the computer key), else a new one.
|
||
loc = numbered.get(slot.number)
|
||
if loc is None:
|
||
loc = copy.deepcopy(slot.locator)
|
||
pk = loc.find("Mapping/PersistentKeyString")
|
||
if pk is not None:
|
||
pk.set("Value", str(slot.number) if slot.number < 10 else "")
|
||
for k in ("Mapping/NoteOrController", "Mapping/LowerRangeNote", "Mapping/UpperRangeNote"):
|
||
setv(loc, k, -1)
|
||
setv(loc, "Mapping/IsNote", False)
|
||
setv(loc, "Time", float(s))
|
||
setv(loc, "Name", "%d- %s - %s" % (slot.number, slot.title, slot.key))
|
||
loc_parent.append(loc)
|
||
for r, l in slot.loops:
|
||
l = copy.deepcopy(l)
|
||
setv(l, "Time", float(s + r))
|
||
loc_parent.append(l)
|
||
if loop_cue is not None:
|
||
for r, c in slot.cues:
|
||
cc = copy.deepcopy(c)
|
||
place(cc, s + r)
|
||
add_clip(loop_cue, cc)
|
||
for r, t, c in slot.kept:
|
||
cc = copy.deepcopy(c)
|
||
place(cc, s + r)
|
||
if cc.tag == "AudioClip":
|
||
files.keep_in_place(cc, base_proj)
|
||
add_clip(t, cc)
|
||
# Waves Tune key cue at the locator.
|
||
tune = tune_for(cfg["_tunes"], slot.key)
|
||
if tune_t is not None and tmpl["tune"] is not None:
|
||
if tune:
|
||
c = clone_clip(tmpl["tune"], s, s + 1, tune[0])
|
||
setv(c, "ProgramChange", tune[1])
|
||
add_clip(tune_t, c)
|
||
else:
|
||
log.append("NO WAVES TUNE CUE FOR %s" % slot.key)
|
||
# The song.
|
||
if song and song.get("root") is not None:
|
||
off = s - song["start"]
|
||
fam_src = [song["group"]] + song["stems"]
|
||
new = [copy.deepcopy(t) for t in fam_src]
|
||
idmap = {}
|
||
for t in new:
|
||
idmap[t.get("Id")] = str(next_tid)
|
||
t.set("Id", str(next_tid))
|
||
next_tid += 1
|
||
for t in new:
|
||
if group_of(t) in idmap:
|
||
setv(t, "TrackGroupId", idmap[group_of(t)])
|
||
fit_slots(t, n_scenes)
|
||
fit_sends(t, holders, cfg.get("send_a", 1.0))
|
||
shift_automation(t, off)
|
||
for c in clips(t):
|
||
place(c, span(c)[0] + off, song["bpm"])
|
||
if c.tag == "AudioClip":
|
||
if song.get("semis"):
|
||
transpose(c, song["semis"], song["bpm"])
|
||
files.adopt(c, song["proj"], song["folder"])
|
||
if t.tag == "AudioTrack":
|
||
want = template_for(routing, tname(t)) if routing else None
|
||
if want and want[0] != "Master" and not route(t, want, targets):
|
||
log.append("NO ROUTE %s FOR %s" % ("/".join(want), stem_name(tname(t))))
|
||
set_name(new[0], song["title"].upper())
|
||
next_ptr = remap_pointees(new, next_ptr)
|
||
for i, t in enumerate(new):
|
||
tr.insert(insert_at + i, t)
|
||
insert_at += len(new)
|
||
for kind, k in (("sections", "sections_track"), ("lighting", "lighting_track"), ("guide", "guide_track")):
|
||
t = track(k)
|
||
if t is None:
|
||
continue
|
||
for c in song[kind]:
|
||
cc = copy.deepcopy(c)
|
||
place(cc, span(c)[0] + off, song["bpm"])
|
||
if cc.tag == "AudioClip":
|
||
files.adopt(cc, song["proj"], song["folder"])
|
||
a, b = span(cc)
|
||
if (cc.tag == "AudioClip") != (t.tag == "AudioTrack"):
|
||
continue
|
||
if any(span(x)[0] < b - 0.01 and span(x)[1] > a + 0.01 for x in clips(t)):
|
||
log.append("%s: A %s CLIP OVERLAPS, LEFT OUT" % (song["title"].upper(), cfg[k].upper()))
|
||
continue
|
||
add_clip(t, cc)
|
||
summary["songs"].append({"title": song["title"], "key": song["key"], "from": song["folder"],
|
||
"semis": song.get("semis", 0), "bpm": song["bpm"], "at": s})
|
||
elif song:
|
||
summary["songs"].append({"title": song["title"], "key": song["key"], "from": None, "at": s})
|
||
# Pads.
|
||
pad = cfg["_pads"].get((key_of(slot.key) or (None,))[0])
|
||
if pads_t is not None and tmpl["pad"] is not None and pad:
|
||
last_cue = max([r for r, _ in slot.cues] + [r for r, _ in slot.loops] + [0.0])
|
||
nxt = layout[layout.index(slot) + 1].start if slot is not layout[-1] else None
|
||
if slot.kind == "ctw":
|
||
a = s
|
||
b = min(nxt + 40.0, nxt + (layout[1].length or 40.0)) if nxt is not None else s + last_cue + 24
|
||
elif slot.kind == "sal":
|
||
a, b = s, s + last_cue + 24.0
|
||
elif not song.get("length"): # a song still to build: only the loop gets a pad
|
||
first = min([r for r, _ in slot.loops] + [r for r, _ in slot.cues] + [last_cue])
|
||
a, b = s + max(first - 32.0, 0.0), s + last_cue + 24.0
|
||
if nxt is not None:
|
||
b = min(b, nxt - 4.0)
|
||
else:
|
||
end = s + (song.get("length") or 0.0)
|
||
a = max(s, end - 64.0)
|
||
b = s + last_cue + 24.0
|
||
if nxt is not None:
|
||
b = min(b, nxt - 4.0)
|
||
b = max(b, end + 8.0)
|
||
c = pad_clip(tmpl["pad"], pad, a, b, slot.bpm)
|
||
files.adopt(c, None, "")
|
||
add_clip(pads_t, c)
|
||
elif pads_t is not None and not pad:
|
||
log.append("NO PAD IN %s" % slot.key)
|
||
|
||
# Tempo map and the TEMPO label clips (one per run of the same tempo).
|
||
set_tempo(ls, regions)
|
||
if tempo_t is not None and tmpl["tempo"] is not None:
|
||
last = layout[-1]
|
||
end_all = last.start + max([r for r, _ in last.cues] + [r for r, _ in last.loops] + [64.0]) + 32.0
|
||
runs = []
|
||
for start, bpm in regions:
|
||
if runs and abs(runs[-1][1] - bpm) < 1e-6:
|
||
continue
|
||
runs.append([start, bpm])
|
||
for i, (start, bpm) in enumerate(runs):
|
||
end = runs[i + 1][0] if i + 1 < len(runs) else end_all
|
||
c = clone_clip(tmpl["tempo"], start, end, "%.2f BPM" % bpm)
|
||
files.keep_in_place(c, base_proj)
|
||
add_clip(tempo_t, c)
|
||
# Clips the base set keeps (LTC, ...) come along into the new project.
|
||
for t in tracks(ls):
|
||
if t.tag == "AudioTrack" and t not in (pads_t, tempo_t) and t not in cleared:
|
||
for c in clips(t):
|
||
fr = c.find("SampleRef/FileRef")
|
||
if fr is not None and not (val(fr, "Path") or "").startswith(final_proj + os.sep):
|
||
files.keep_in_place(c, base_proj)
|
||
renumber(loc_parent)
|
||
setv(ls, "NextPointeeId", max(next_ptr, max_pointee(root) + 1))
|
||
problems = check(root)
|
||
if problems:
|
||
raise RuntimeError("SET CHECK: " + problems[0])
|
||
|
||
status.write(step="COPYING %s" % svc.get("label", ""), copy_bytes=files.size())
|
||
shutil.rmtree(building, ignore_errors=True)
|
||
make_project(building)
|
||
if not cfg.get("no_copy"):
|
||
files.copy(building, lambda d, n: status.write(step="COPYING %d/%d" % (d, n)))
|
||
save(root, os.path.join(building, name + ".als"))
|
||
if os.path.exists(final_proj):
|
||
raise RuntimeError("%s APPEARED WHILE BUILDING" % name)
|
||
os.rename(building, final_proj)
|
||
if files.missing:
|
||
log.append("%d AUDIO FILES MISSING" % len(files.missing))
|
||
summary["missing"] = files.missing[:20]
|
||
summary["path"] = os.path.join(final_proj, name + ".als")
|
||
return summary
|
||
|
||
|
||
def check(root):
|
||
"""What would make Live refuse the set (or misbehave): an empty list when it is sound."""
|
||
ls = root.find("LiveSet")
|
||
out = []
|
||
ids = [t.get("Id") for t in tracks(ls)]
|
||
if len(ids) != len(set(ids)):
|
||
out.append("TRACK IDS REPEAT")
|
||
for t in tracks(ls):
|
||
g = group_of(t)
|
||
if g != "-1" and g not in ids:
|
||
out.append("%s IN A GROUP THAT IS GONE" % tname(t))
|
||
ptr = [e.get("Id") for e in root.iter() if is_pointee(e)]
|
||
if len(ptr) != len(set(ptr)):
|
||
out.append("AUTOMATION IDS REPEAT")
|
||
if ptr and max(int(p) for p in ptr) >= int(val(ls, "NextPointeeId")):
|
||
out.append("NEXT POINTEE ID TOO LOW")
|
||
scenes = len(list(ls.find("Scenes")))
|
||
returns = len([t for t in tracks(ls) if t.tag == "ReturnTrack"])
|
||
for t in tracks(ls):
|
||
lst = t.find("DeviceChain/MainSequencer/ClipSlotList")
|
||
if lst is not None and len(lst) != scenes:
|
||
out.append("%s HAS %d CLIP SLOTS" % (tname(t), len(lst)))
|
||
sends = t.find("DeviceChain/Mixer/Sends")
|
||
if sends is not None and t.tag != "ReturnTrack" and len(sends) != returns:
|
||
out.append("%s HAS %d SENDS" % (tname(t), len(sends)))
|
||
prev = None
|
||
for c in sorted(clips(t), key=lambda c: span(c)[0]):
|
||
s, e = span(c)
|
||
if e <= s:
|
||
out.append("%s: AN EMPTY CLIP AT %g" % (tname(t), s))
|
||
if prev is not None and s < prev - 0.01:
|
||
out.append("%s: CLIPS OVERLAP AT %g" % (tname(t), s))
|
||
prev = e
|
||
return out
|
||
|
||
|
||
# --- the job ----------------------------------------------------------------------------------
|
||
|
||
def projects_dir(cfg):
|
||
return os.path.expanduser(cfg["projects"])
|
||
|
||
|
||
def weekend_pattern(cfg):
|
||
""""{date} {loc} Project" -> a pattern for the location's weekend sets (not "260909 NA FORGE Project")."""
|
||
f = re.escape(cfg.get("folder_format", "{date} {loc} Project"))
|
||
return "^" + f.replace(re.escape("{date}"), r"(\d{6})").replace(re.escape("{loc}"), re.escape(cfg["site_folder"])) + "$"
|
||
|
||
|
||
def newest_set(folder, before=None, pattern=r"^(\d{6}) "):
|
||
"""The newest dated set in a location folder ("261025 NA Project"), before `before` (YYMMDD)."""
|
||
best = None
|
||
for n in os.listdir(folder) if os.path.isdir(folder) else []:
|
||
m = re.match(pattern, n)
|
||
if not m or n.endswith((".building", ".copying")) or (before and m.group(1) >= before):
|
||
continue
|
||
d = os.path.join(folder, n)
|
||
als = [os.path.join(d, x) for x in os.listdir(d) if x.endswith(".als") and not x.startswith(".")] if os.path.isdir(d) else []
|
||
if als and (best is None or m.group(1) > best[0]):
|
||
best = (m.group(1), max(als, key=os.path.getmtime))
|
||
return best[1] if best else None
|
||
|
||
|
||
def target(cfg, svc):
|
||
folder = os.path.join(os.path.expanduser(cfg.get("out_projects") or cfg["projects"]), svc["folder"])
|
||
return folder, svc["name"]
|
||
|
||
|
||
def run(cfg):
|
||
status = Status({"state": "running", "started": time.time(), "step": "STARTING", "results": [], "log": []})
|
||
log = status.data["log"]
|
||
try:
|
||
cfg["_pads"] = pads_index(cfg.get("pads_folder"))
|
||
cfg["_tunes"] = tune_index(cfg.get("tune_library"))
|
||
site_dir = os.path.join(projects_dir(cfg), cfg["site_folder"])
|
||
todo = []
|
||
for svc in cfg["services"]:
|
||
folder, name = target(cfg, svc)
|
||
d = os.path.join(folder, name)
|
||
if os.path.isdir(d):
|
||
status.data["results"].append({"name": name, "skipped": "THERE ALREADY"})
|
||
continue
|
||
todo.append(svc)
|
||
if not todo:
|
||
status.write(state="done", step="NOTHING TO DO")
|
||
return
|
||
# 1. Last weekend's songs to Finished Songs (from the newest weekend set before the first new one).
|
||
first = min(s["date"] for s in todo)
|
||
prev = newest_set(site_dir, first, weekend_pattern(cfg))
|
||
if prev and cfg.get("save_songs", True):
|
||
status.write(step="SAVING SONGS")
|
||
saved = export_songs(prev, cfg, status, log)
|
||
status.write(saved=saved)
|
||
for svc in sorted(todo, key=lambda s: s["date"]):
|
||
folder, name = target(cfg, svc)
|
||
base = newest_set(site_dir, svc["date"], weekend_pattern(cfg))
|
||
if not base:
|
||
log.append("NO WEEKEND SET TO START %s FROM" % name)
|
||
continue
|
||
status.write(step="BUILDING %s" % svc["label"], base=os.path.basename(base))
|
||
os.makedirs(folder, exist_ok=True)
|
||
notes = []
|
||
res = build_set(svc, base, folder, name, cfg, status, notes)
|
||
res["notes"] = list(dict.fromkeys(notes))
|
||
log.extend(res["notes"])
|
||
status.data["results"].append(res)
|
||
status.write()
|
||
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:])
|
||
|
||
|
||
def main(P):
|
||
if P.get("mode") == "run":
|
||
# Detached, so a hold returns at once; one job at a time.
|
||
try:
|
||
fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
||
except FileExistsError:
|
||
if time.time() - os.path.getmtime(LOCK) < 3 * 3600:
|
||
print(json.dumps({"started": False, "busy": True}))
|
||
return
|
||
os.remove(LOCK)
|
||
fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
||
os.close(fd)
|
||
job = os.path.expanduser("~/Library/Caches/status-panel-start-job.json")
|
||
with open(job, "w") as f:
|
||
json.dump(P, f)
|
||
prog = os.path.expanduser("~/Library/Caches/status-panel-start.py")
|
||
src = globals().get("SOURCE")
|
||
if src is None:
|
||
with open(__file__) as f:
|
||
src = f.read()
|
||
with open(prog, "w") as f:
|
||
f.write(src)
|
||
subprocess.Popen(["/usr/bin/nice", "-n", "15", "/usr/bin/python3", prog, job],
|
||
stdout=open("/tmp/status-panel-start.log", "w"), stderr=subprocess.STDOUT, start_new_session=True)
|
||
print(json.dumps({"started": True}))
|
||
return
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1: # the detached job
|
||
try:
|
||
run(json.load(open(sys.argv[1])))
|
||
finally:
|
||
try:
|
||
os.remove(LOCK)
|
||
except OSError:
|
||
pass
|