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

LIGHT TIMING: the real offset up to a full 16th, no flag on a cue already on a solid hit, hold moves the cues

Browse Source 
- Detection in 2.5 ms steps; where a hit starts is halfway up its rise; each song's cues on solid hits
  set its "on time"; a cue already on a solid hit is never named; "about a 16th" when it is
- Hold: moves the off cues onto their hit in the open set through a status panel command added to
  AbletonOSC (tools/abletonosc), keeping each cue's end; the key asks for a save until the set is saved

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-22 10:31:33 -04:00
co-authored by Claude Opus 5.5
parent d4e4092a9d
commit 625d424bda
8 changed files with 268 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
readiness.py
+145 -37
View File
@@ -849,28 +849,36 @@ def mix_plan(rc, live):
HIT_SCRIPT = r"""
import aifc, audioop, json, wave
P = json.loads(%r)
FRAME = 0.005
FRAME = 0.0025
def attack(env, i):
# how far the drums climb within 10 ms of frame i, from the quietest point in the 20 ms before it
after = max(env[i:i + 4] or [0])
return max(0, after - min(env[max(0, i - 8):i] or [after]))
def look(path, sec, span):
aiff = path.lower().endswith((".aif", ".aiff"))
f = aifc.open(path, "rb") if aiff else wave.open(path, "rb")
try:
sr, w, ch = f.getframerate(), f.getsampwidth(), f.getnchannels()
a = max(0, int((sec - span - 0.03) * sr))
a = max(0, int((sec - span - 0.05) * sr))
f.setpos(a)
hop, n = int(FRAME * sr), int((2 * span + 0.03) / FRAME)
data = f.readframes(hop * (n + 1))
data = f.readframes(int((2 * span + 0.1) * sr))
finally:
f.close()
if aiff:
data = audioop.byteswap(data, w) # AIFF is big-endian
size = hop * w * ch
rms = [audioop.rms(data[i * size:(i + 1) * size], w) for i in range(n)]
rise = [max(0, rms[i] - sum(rms[max(0, i - 4):i]) / max(1, min(4, i))) for i in range(n)]
times = [a / sr + i * FRAME for i in range(n)]
near = [i for i in range(n) if abs(times[i] - sec) <= span]
at = max([rise[i] for i in near if abs(times[i] - sec) <= 0.012] or [0])
best = max(near, key=lambda i: rise[i]) if near else None
return [round(times[best] - sec, 4), rise[best], at] if best is not None else None
hop = int(FRAME * sr) * w * ch
env = [audioop.rms(data[i:i + hop], w) for i in range(0, len(data) - hop + 1, hop)]
t0 = a / sr - sec
near = [i for i in range(len(env)) if abs(t0 + i * FRAME) <= span]
if not near:
return None
att = [attack(env, i) for i in range(len(env))]
best = max(near, key=lambda i: att[i])
at = max([att[i] for i in near if abs(t0 + i * FRAME) <= 0.010] or [0])
# where that hit starts: the first frame that climbs halfway from the quiet before it to its peak
lo, hi = min(env[max(0, best - 8):best] or [0]), max(env[best:best + 4] or [0])
start = next((j for j in range(max(0, best - 8), best + 4) if env[j] >= lo + 0.5 * (hi - lo)), best)
return [round(t0 + start * FRAME, 4), att[best], at]
out = []
for path, sec, span in P["cues"]:
try:
@@ -880,10 +888,12 @@ for path, sec, span in P["cues"]:
print(json.dumps(out))
"""
HITS = {} # (set, its mtime) -> the hit list, so an unchanged set is not listened to again
MOVED = {} # (set, its mtime) -> beats a hold moved in Live, until the set is saved
def cue_times(rc, summary):
"""[(beat, note, song group, file, second in the file, search span)] for every flash and FX cue over a drum stem."""
"""[(beat, note, song group, file, second in the file, search span, a 16th in seconds)] for every flash and FX
cue over a drum stem. The search reaches out to almost a full 16th either way."""
groups = song_groups(summary)
owner = {t["id"]: g for g, members in groups.items() for t in members}
tempo = summary["tempo"]
@@ -897,7 +907,7 @@ def cue_times(rc, summary):
seen.add(beat)
sec = seconds_at(tempo, beat) - seconds_at(tempo, src[1]) + src[5]
sixteenth = seconds_at(tempo, beat + 0.25) - seconds_at(tempo, beat)
out.append((beat, note, owner.get(src[0], ""), src[3], sec, min(0.08, 0.45 * sixteenth)))
out.append((beat, note, owner.get(src[0], ""), src[3], sec, min(0.95 * sixteenth, 0.25), sixteenth))
return out
@@ -906,9 +916,35 @@ def bar_of(beat):
return "%d.%d.%d" % (beat // 4 + 1, beat % 4 + 1, (beat % 1) * 4 + 1)
def off_cues(rc, cues, hits):
"""[(cue, ms off against the song's own cues on solid hits)] for cues on a quiet spot with a strong hit near.
A cue that already sits on a solid hit is never named, even with a stronger hit close by (a double hit)."""
tol = rc.get("light_tolerance_ms", 25) / 1000.0
strengths = {}
for c, h in zip(cues, hits):
if isinstance(h, list):
strengths.setdefault(c[3], []).append(h[1])
typical = {f: sorted(v)[len(v) // 2] for f, v in strengths.items()}
solid = lambda c, h: h[2] >= 0.5 * typical.get(c[3], 0) or (h[1] > 0 and h[2] >= 0.6 * h[1])
# each song's stem starts its hits a few ms after the grid: what its cues on solid hits show is "on time"
base = {}
for c, h in zip(cues, hits):
if isinstance(h, list) and solid(c, h):
base.setdefault(c[2], []).append(h[0])
base = {song: sorted(v)[len(v) // 2] for song, v in base.items()}
out = []
for c, h in zip(cues, hits):
if not isinstance(h, list):
continue
dt = h[0] - base.get(c[2], 0.0)
if abs(dt) > tol and h[1] >= 0.4 * typical.get(c[3], 0) and h[1] > 0 and not solid(c, h):
out.append((c, dt))
return out
def eval_light_timing(rc, look, hits):
"""Cues whose hit is clearly a few ms off: the strongest hit near the cue is at least `light_tolerance_ms`
away, is a real hit for that song, and is clearly stronger than what is under the cue itself."""
"""Cues whose hit is clearly off (see off_cues), named with song and bar; cues a hold moved in Live
count as done until the set is saved and listened to again."""
line = Line()
if not look.get("set"):
line.fault(ERR, "NO SET", "no set to check")
@@ -918,29 +954,23 @@ def eval_light_timing(rc, look, hits):
line.info = "NO CUES ON DRUMS"
return line
names = look.get("light_names") or {}
tol = rc.get("light_tolerance_ms", 25) / 1000.0
by_file = {}
for c, h in zip(cues, hits):
if isinstance(h, list):
by_file.setdefault(c[3], []).append(h[1])
typical = {f: sorted(v)[len(v) // 2] for f, v in by_file.items()}
off, broken = [], set()
for c, h in zip(cues, hits):
if not isinstance(h, list):
broken.add(os.path.basename(c[3]))
continue
dt, best, at = h
if abs(dt) > tol and best >= 0.3 * typical.get(c[3], 0) and best > 0 and (at == 0 or best / at >= 1.5):
off.append((c, dt))
said = ["%s bar %s: %s is %d ms %s the hit" % (song or "?", bar_of(beat), names.get(str(note), "note %d" % note),
abs(dt) * 1000, "before" if dt > 0 else "after")
for (beat, note, song, _, _, _), dt in off]
moved = set(MOVED.get((look.get("set"), look.get("mtime"))) or [])
off = [(c, dt) for c, dt in off_cues(rc, cues, hits) if c[0] not in moved]
said = []
for (beat, note, song, _, _, _, sixteenth), dt in off:
how = "%d ms %s the hit" % (abs(dt) * 1000, "before" if dt > 0 else "after")
if abs(dt) >= 0.75 * sixteenth:
how += " (about a 16th)"
said.append("%s bar %s: %s is %s" % (song or "?", bar_of(beat), names.get(str(note), "note %d" % note), how))
if off:
line.fault(WARN, "%d CUES OFF" % len(off) if len(off) > 1 else "CUE OFF %s" % bar_of(off[0][0][0]), said[0])
line.notes += said[1:]
for f in sorted(broken):
line.notes.append("could not read %s" % f)
line.info = "%d CUES ON TIME" % (len(cues) - len(off) - sum(1 for h in hits if not isinstance(h, list)))
if moved:
line.fault(WARN, "SAVE SET (%d MOVED)" % len(moved), "a hold moved %d cue%s onto the hit in Live: save the set "
"(Cmd+S), or Cmd+Z in Live to take it back" % (len(moved), "" if len(moved) == 1 else "s"))
broken = sorted({os.path.basename(c[3]) for c, h in zip(cues, hits) if not isinstance(h, list)})
line.notes += ["could not read %s" % f for f in broken]
line.info = "%d CUES ON TIME" % (len(cues) - len(off) - len(moved) - sum(1 for h in hits if not isinstance(h, list)))
return line
@@ -956,6 +986,76 @@ def light_hits(cfg, rc, look):
return HITS[key]
def moves(rc, summary, off):
"""[(beat, note, new beat)] that put each off cue (every flash/FX note on that 16th) onto its hit, on the
16th grid when the hit is within 10 ms of it."""
out = []
wanted = set(rc.get("light_hit_notes", range(30, 40)))
for c, dt in off:
beat, sixteenth = c[0], c[6]
new = beat + dt * 0.25 / sixteenth
grid = round(new * 4) / 4
if abs(grid - new) * 4 * sixteenth <= 0.010:
new = grid
for b, note in summary.get("cues") or []:
if b == beat and note in wanted:
out.append((beat, note, round(new, 4)))
return out
MOVE_SCRIPT = r"""
import json, socket, struct, time
P = json.loads(%r)
def pad(b):
return b + b"\0" * (4 - len(b) %% 4)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.bind(("127.0.0.1", 11001))
except OSError:
print(json.dumps({"error": "OSC PORT BUSY"})); raise SystemExit
s.settimeout(0.5)
done = []
for beat, note, new in P["moves"]:
s.sendto(pad(b"/live/statuspanel/move_note") + pad(b",sfif") + pad(P["track"].encode()) + struct.pack(">fif", beat, note, new),
("127.0.0.1", 11000))
end = time.time() + 3
while time.time() < end:
try:
b, _ = s.recvfrom(4096)
except socket.timeout:
continue
if b.startswith(b"/live/statuspanel/move_note"):
done.append([beat, note, struct.unpack(">i", b[-4:])[0]])
break
else:
print(json.dumps({"error": "NO MOVE COMMAND"})); raise SystemExit
print(json.dumps({"done": done}))
"""
def fix_light_timing(cfg, rc):
"""A hold: move every off cue onto its hit in the open set (AbletonOSC's status panel command)."""
light = LAST.get("light") or {}
if not light.get("off"):
return True
todo = moves(rc, light["summary"], light["off"])
try:
r = run_remote(cfg, rc["tracks_host"], MOVE_SCRIPT % json.dumps(
{"track": rc.get("lighting_track", "LIGHTING CUES"), "moves": todo}), timeout=60)
finally:
hand_back(cfg, rc["tracks_host"])
if r.get("error"):
return r["error"]
moved = sorted({b for b, _, n in r["done"] if n})
if moved:
MOVED.setdefault(light["key"], [])
MOVED[light["key"]] = sorted(set(MOVED[light["key"]]) | set(moved))
log.info("FIX lighting cues moved onto the hit in Live: %s%s", ", ".join(bar_of(b) for b in moved) or "none",
" (not found in the open set: %s)" % ", ".join(bar_of(b) for b, _, n in r["done"] if not n)
if any(not n for _, _, n in r["done"]) else "")
return True if moved else "NOT IN OPEN SET"
# --- Song keys --------------------------------------------------------------------------
NOTES = {"C": 0, "D": 2, "E": 4, "F": 5, "G": 7, "A": 9, "B": 11}
@@ -1646,7 +1746,13 @@ def look(cfg, now=None, data_dir=None, sources=None):
lines["tune"] = eval_tune(rc, plan, al)
if rc.get("light_timing_check", True):
try:
lines["light_timing"] = eval_light_timing(rc, al, src["hits"](al) if al.get("set") else [])
hits = src["hits"](al) if al.get("set") else []
lines["light_timing"] = eval_light_timing(rc, al, hits)
if al.get("set"):
cues = cue_times(rc, al["summary"])
LAST["light"] = {"key": (al["set"], al.get("mtime")), "summary": al["summary"],
"off": [(c, dt) for c, dt in off_cues(rc, cues, hits)
if c[0] not in set(MOVED.get((al["set"], al.get("mtime"))) or [])]}
except Exception as e: # noqa: BLE001 - a stem it cannot read must not hide the rest
failed(["light_timing"], e)
if rc.get("ltc_note", "LTC"):
@@ -1803,6 +1909,8 @@ def run_action(cfg, action, chk=None, sleep=time.sleep):
return "NO SET TO COPY"
log.info("FIX tracks copy started to %s", ", ".join(r["started"]) or "nowhere (all there)")
return True
if action == "light_timing":
return fix_light_timing(cfg, rc)
if action == "ableton_mix":
_, _, changes = mix_plan(rc, live_look(cfg, rc))
if not changes: