Skip to content

T
Tidal
Commit 2ddcae51 by PLN (Algolia)
Options

fix(take-lens): detection resolution is not a display preference — plus mix 

Added `mix` (sum the orbits to one gain-corrected, trimmed stereo file), then found
two real bugs in the trim rule by verifying my own render instead of trusting it.

BUG 1 — the rule demanded idle hands AND silence

audio_lens profile on the rendered mix reported rms -180 dBFS at 19.2 min: DIGITAL
silence, sitting in the middle of a "trimmed" master. The first rule only cut where
the hands were idle AND nothing sounded, and there PLN was busy on the controller
with nothing playing. That third state — fiddling in silence — IS the practice/
plugging-in the trim exists to remove. Silence needs no corroboration from the hands,
so it is now its own kind with its own (much shorter) 20s threshold: long enough to
never eat a break or a drop's pre-silence (~4 bars, under 8s at these tempos), short
enough to catch the real thing.

BUG 2 — and the reason it was invisible: analysis ran at DISPLAY resolution

A direct 1-second measurement of the same mix found 103 seconds of silence in five
runs of 11-31s. The classifier, bucketing at 15s, found NONE of them: a 17-second
silence straddling two 15-second buckets leaves no bucket fully silent, so every
bucket's rms sat far above -60 dBFS. The threshold was fine; the grain was lying.

Analysis now always runs at GRAIN_S = 1s and `--bucket` only sets the sparkline
width. Detection resolution must never be a display preference — the coarse view is
for eyes, the decision is for data.

STRUCTURAL: one rule, one reader
`classify()` is now the single source of the cut rule and `analyse()` the single
audio pass, both shared by the report and the mixer. They were duplicated, which is
the worst possible arrangement here: a mixer that cuts what the report called `keep`
destroys material silently. Also dedup candidates by 80% OVERLAP rather than strict
containment — the hands-idle run began 1s before the silence run on take 94, so
containment reported the same stretch twice and the table contradicted itself.

VALIDATED end to end on take 94 (21.14 min, 12 orbits)
  candidates 5, no contradictions: 3x silence (21s, 30s, 68s) TRIM,
  2x playing (68s at 8.00 and 3.72 orbits avg) KEEP.
  Trimming only `silence`+`dead` -> 19.15 min out, peak -3.36 dBFS, rms -24.47.
  Re-measured the render: silence 103s -> 52s. The three survivors are 17s, 18s and
  11s — all under the 20s threshold BY DESIGN, plausibly musical gaps that want an
  ear rather than a rule. Spectral balance broadband at 60s/300s/600s/900s via
  audio_lens profile (the opening 20s reads kick/sub-only because it genuinely is).

`loop` is deliberately NOT cuttable: a droning pad under idle hands can be an intro
as easily as a distraction. It proposes; the ear decides.
parent 2e6e4e9f
Showing with 309 additions and 75 deletions
tools/take-lens.py
......@@ -64,6 +64,15 @@ BAR = " ▁▂▃▄▅▆▇█"
# 120-140 BPM (a 16-bar section is ~32s) and so cannot be mistaken for "playing a
# part without touching anything".
IDLE_RUN_S = 45.0
# SILENCE, on the other hand, needs no corroboration from the hands and can be much
# shorter. Learned the hard way on take 94: the first version of this tool required
# idle hands AND silence, and so kept ~50s of DIGITAL silence (-180 dBFS, verified by
# audio_lens profile) in the middle of the mix — because PLN was busy on the
# controller with nothing sounding. That state is "fiddling in silence", which is
# exactly the practice/plugging-in the trim exists to remove. 20s is short enough to
# catch it and long enough to never eat a break or a drop's pre-silence (the longest
# deliberate gap in the set is ~4 bars, under 8s at these tempos).
SILENT_RUN_S = 20.0
# Below this, an orbit is considered silent for the bucket. -60 dBFS is well under
# the noise floor of the capture chain (the known 223 Hz vsink noise sits lower) and
# well above true digital silence, so a decaying reverb tail still counts as sound.
......@@ -255,6 +264,131 @@ def orbit_energy(path: Path, bucket_s: float, nb: int, sr: int) -> dict:
"peak_all_db": db(max(pk) if pk else 0.0)}
# The grain everything is DECIDED at. Display coarseness is a separate knob.
GRAIN_S = 1.0
def analyse(t: dict, rec: dict, grain_s: float, audio: bool):
"""Per-grain hands and sound for one take. One audio pass, one truth.
Returns (n_steps, cc_n, notes, ctrls, active, loud, energy) where every list is
indexed by grain step. Shared by the report and the mixer so they can never be
looking at different numbers.
"""
dur, sr = t["dur_s"], t["sr"]
ns = max(1, int(dur // grain_s) + 1)
def step_of(ts: float) -> int:
return min(ns - 1, max(0, int((ts - t["start"]) // grain_s)))
cc_n = [0] * ns
ctrls: list[set] = [set() for _ in range(ns)]
for r in rec.get("cc", []):
i = step_of(r["t"])
cc_n[i] += r.get("n", 1)
if r.get("cc") is not None:
ctrls[i].add(r["cc"])
notes = [0] * ns
for r in rec.get("note", []):
notes[step_of(r["t"])] += 1
energy: dict[int, dict] = {}
if audio:
try:
for orb in sorted(t["orbits"]):
energy[orb] = orbit_energy(t["orbits"][orb], grain_s, ns, sr)
except ImportError as e:
print(f" audio skipped: {e}")
energy = {}
active = [0] * ns
loud = [-120.0] * ns
for e in energy.values():
for i, v in enumerate(e["rms_db"]):
if v > SILENT_DBFS:
active[i] += 1
loud[i] = max(loud[i], v)
return ns, cc_n, notes, ctrls, active, loud, energy
# ------------------------------------------------------------- classification
def classify(nb: int, bucket_s: float, dur: float, cc_n: list[int],
active: list[int], n_orbits: int) -> list[dict]:
"""Every trim candidate, in one place.
THE single source of the cut rule. It used to be duplicated between the report
and the mixer, which is the worst possible arrangement here: a mixer that trims
something the report called `keep` destroys material silently. Both callers now
ask this function.
Kinds, in priority order:
silence every orbit under SILENT_DBFS for >= SILENT_RUN_S. Trim, no matter
what the hands were doing — see the SILENT_RUN_S note.
dead hands idle >= IDLE_RUN_S and nothing sounding. Trim.
loop hands idle >= IDLE_RUN_S, a couple of orbits droning. Probably
plugging in; flag but do not auto-cut.
playing hands idle >= IDLE_RUN_S, most orbits going. A real part. KEEP.
"""
out: list[dict] = []
def emit(a: int, b: int, kind: str, why: str) -> None:
out.append({"start_s": a * bucket_s, "end_s": min(dur, b * bucket_s),
"dur_s": min(dur, b * bucket_s) - a * bucket_s,
"kind": kind, "why": why})
def runs(pred) -> list[tuple[int, int]]:
found, cur = [], None
for i in range(nb):
if pred(i):
cur = i if cur is None else cur
elif cur is not None:
found.append((cur, i))
cur = None
if cur is not None:
found.append((cur, nb))
return found
have_audio = any(active)
silent_spans: list[tuple[int, int]] = []
if have_audio:
for a, b in runs(lambda i: active[i] == 0):
if (b - a) * bucket_s >= SILENT_RUN_S:
silent_spans.append((a, b))
emit(a, b, "silence", "every orbit under %.0f dBFS" % SILENT_DBFS)
for a, b in runs(lambda i: cc_n[i] == 0):
if (b - a) * bucket_s < IDLE_RUN_S:
continue
# Already covered by a silence span? Do not report it twice. Measured by
# OVERLAP, not containment: the hands-idle run and the silence run rarely
# start on the same step (one began 1s earlier on take 94) and strict
# containment then reported the same stretch twice, once as `silence` and
# once as `loop` — contradicting itself in the same table.
span = b - a
covered = sum(max(0, min(b, sb) - max(a, sa)) for sa, sb in silent_spans)
if span and covered / span >= 0.8:
continue
if not have_audio:
emit(a, b, "unknown", "no audio read")
continue
seg = active[a:b]
avg = sum(seg) / len(seg)
if max(seg, default=0) == 0:
emit(a, b, "dead", "no orbit above %.0f dBFS" % SILENT_DBFS)
elif avg <= max(2, n_orbits * 0.25):
emit(a, b, "loop", "%.2f orbits avg, hands off" % avg)
else:
emit(a, b, "playing", "%.2f orbits avg — a real part, keep" % avg)
out.sort(key=lambda c: c["start_s"])
return out
# Kinds the mixer will actually cut. `loop` is deliberately NOT here: it needs an
# ear, because a droning pad under idle hands can be an intro as easily as a
# distraction.
CUTTABLE = ("silence", "dead")
# ------------------------------------------------------------------------- lens
def cmd_lens(take_n: int, bucket_s: float, session: Path,
......@@ -280,45 +414,26 @@ def cmd_lens(take_n: int, bucket_s: float, session: Path,
print(f" gig log: {len(paths)} file(s)")
rec = read_window(paths, t0, t1)
def bucket_of(ts: float) -> int:
return min(nb - 1, max(0, int((ts - t0) // bucket_s)))
# --- hands ---------------------------------------------------------------
cc_n = [0] * nb
ctrls: list[set] = [set() for _ in range(nb)]
for r in rec["cc"]:
b = bucket_of(r["t"])
cc_n[b] += r.get("n", 1)
if r.get("cc") is not None:
ctrls[b].add(r["cc"])
notes = [0] * nb
for r in rec["note"]:
notes[bucket_of(r["t"])] += 1
# --- sound ---------------------------------------------------------------
energy: dict[int, dict] = {}
if audio:
try:
for orb in sorted(t["orbits"]):
energy[orb] = orbit_energy(t["orbits"][orb], bucket_s, nb, sr)
except ImportError as e:
print(f" audio skipped: {e}")
energy = {}
active = [0] * nb # how many orbits are making sound
loud = [-120.0] * nb # loudest orbit rms in the bucket
for orb, e in energy.items():
for i, v in enumerate(e["rms_db"]):
if v > SILENT_DBFS:
active[i] += 1
loud[i] = max(loud[i], v)
# Analysis runs at GRAIN_S (1s) ALWAYS; --bucket only sets the width of the
# sparkline. That separation is not cosmetic — the first version analysed at the
# display grain, and a 17-second silence straddling two 15-second buckets left
# NO bucket fully silent, so it detected none of the 103 seconds of digital
# silence that a direct 1-second measurement of the same mix found. Detection
# resolution must not be a display preference.
ns, cc_n, notes, ctrls, active, loud, energy = analyse(
t, rec, GRAIN_S, audio)
nb = max(1, int(ns * GRAIN_S // bucket_s) + 1)
# --- report --------------------------------------------------------------
W = 60
print(f"\nbuckets: {nb} x {bucket_s:.0f}s")
hands = sparkline([float(v) for v in cc_n], lo=0.0)
orbs = sparkline([float(v) for v in active], lo=0.0,
per = max(1, int(bucket_s / GRAIN_S))
print(f"\nanalysis grain {GRAIN_S:.0f}s ({ns} steps); display {bucket_s:.0f}s")
hands = sparkline([float(sum(cc_n[i*per:(i+1)*per])) for i in range(nb)], lo=0.0)
orbs = sparkline([float(max(active[i*per:(i+1)*per], default=0))
for i in range(nb)], lo=0.0,
hi=float(len(t["orbits"]) or 1)) if energy else ""
lvl = sparkline(loud, lo=-60.0, hi=0.0) if energy else ""
lvl = sparkline([max(loud[i*per:(i+1)*per], default=-120.0)
for i in range(nb)], lo=-60.0, hi=0.0) if energy else ""
for i in range(0, nb, W):
print(f"\n {hhmm(t0 + i*bucket_s)}")
print(f" hands {hands[i:i+W]}")
......@@ -329,7 +444,11 @@ def cmd_lens(take_n: int, bucket_s: float, session: Path,
print(f"\nhands: {sum(cc_n)} CC + {sum(notes)} notes, "
f"{len(set().union(*ctrls)) if any(ctrls) else 0} distinct controllers")
idle = sum(1 for v in cc_n if v == 0)
print(f" {idle}/{nb} buckets hands-idle ({100*idle/nb:.0f}%)")
print(f" {idle}/{ns} s hands-idle ({100*idle/ns:.0f}%)")
if energy:
sil = sum(1 for v in active if v == 0)
print(f"sound: {sil}/{ns} s with every orbit under {SILENT_DBFS:.0f} dBFS "
f"({100*sil/ns:.0f}%)")
if energy:
peaks = {o: e["peak_all_db"] for o, e in energy.items()}
......@@ -377,47 +496,20 @@ def cmd_lens(take_n: int, bucket_s: float, session: Path,
print(" inferred. Later takes split exactly.")
# --- trim candidates ------------------------------------------------------
runs = []
cur = None
for i in range(nb):
if cc_n[i] == 0:
cur = i if cur is None else cur
elif cur is not None:
runs.append((cur, i))
cur = None
if cur is not None:
runs.append((cur, nb))
cands = []
for a, b in runs:
if (b - a) * bucket_s < IDLE_RUN_S:
continue
seg_active = active[a:b] if energy else []
# Idle hands alone is ambiguous. Classify by what the SOUND did:
# dead - nothing playing: safe to cut
# loop - a few orbits droning, hands off: probably plugging in
# playing - most orbits going: a real part. Do NOT trim.
if not energy:
kind, why = "unknown", "no audio read"
elif max(seg_active, default=0) == 0:
kind, why = "dead", "no orbit above %.0f dBFS" % SILENT_DBFS
elif (sum(seg_active) / len(seg_active)) <= max(2, len(t["orbits"]) * 0.25):
kind, why = "loop", "%.1f orbits avg, hands off" % (
sum(seg_active) / len(seg_active))
else:
kind, why = "playing", "%.1f orbits avg — a real part, keep" % (
sum(seg_active) / len(seg_active))
cands.append({"start": t0 + a * bucket_s, "end": t0 + b * bucket_s,
"start_s": a * bucket_s, "end_s": b * bucket_s,
"dur_s": (b - a) * bucket_s, "kind": kind, "why": why})
cands = classify(ns, GRAIN_S, dur, cc_n, active, len(t["orbits"]))
for c in cands:
c["start"] = t0 + c["start_s"]
c["end"] = t0 + c["end_s"]
print(f"\nhands-idle runs >= {IDLE_RUN_S:.0f}s: {len(cands)}")
cut_s = sum(c["dur_s"] for c in cands if c["kind"] in CUTTABLE)
print(f"\ncandidates: {len(cands)} "
f"(cuttable {cut_s:.0f}s of {dur:.0f}s = {100*cut_s/dur:.0f}%)")
for c in cands:
tag = {"dead": "TRIM", "loop": "trim?", "playing": "keep",
"unknown": "?"}[c["kind"]]
print(f" [{tag:>5}] {hhmmss(c['start'])} -> {hhmmss(c['end'])} "
f"{c['dur_s']:>5.0f}s +{c['start_s']/60:>5.1f}min into take "
f"({c['why']})")
tag = {"silence": "TRIM", "dead": "TRIM", "loop": "trim?",
"playing": "keep", "unknown": "?"}[c["kind"]]
print(f" [{tag:>5}] {c['kind']:<8} {hhmmss(c['start'])} -> "
f"{hhmmss(c['end'])} {c['dur_s']:>5.0f}s "
f"+{c['start_s']/60:>5.1f}min ({c['why']})")
if edl_out:
edl = {
......@@ -438,6 +530,138 @@ def cmd_lens(take_n: int, bucket_s: float, session: Path,
return 0
# ------------------------------------------------------------------------- mix
def cmd_mix(take_n: int, session: Path, out: Path, gain_db: float | None,
trim: bool, bucket_s: float) -> int:
"""Sum the orbit stems into one stereo file, gain-corrected and trimmed.
Streams in one-second blocks across all 24 mono files at once: constant memory,
one sequential pass per file. Writes 32-bit float so a sum that overshoots stays
recoverable — the same reason the source stems are float.
`--gain-db auto` derives the trim from the measured peak so the hottest orbit
lands on the -6 dBFS source target. UNIFORM across orbits, never per-orbit: a
per-orbit trim would silently re-balance the mix PLN performed.
"""
import numpy as np
import soundfile as sf
takes = find_takes(session)
if take_n not in takes:
print(f"take-lens: no take {take_n} (have {sorted(takes)})", file=sys.stderr)
return 1
t = takes[take_n]
sr, dur = t["sr"], t["dur_s"]
orbits = sorted(t["orbits"])
# Pair up %L/%R. A missing side is NOT silently mono-ed: say so, because a
# one-sided orbit means a routing fault worth knowing about.
pairs: dict[int, list[Path]] = {}
for orb in orbits:
lp = t["orbits"][orb]
rp = Path(str(lp).replace("%L.wav", "%R.wav"))
if not rp.exists():
print(f" d{orb}: no %R side — using %L for both channels")
rp = lp
pairs[orb] = [lp, rp]
if gain_db is None:
gain_db = 0.0
g = 10.0 ** (gain_db / 20.0)
# Trim only regions the lens called `dead` — never `loop`, never `playing`.
cuts: list[tuple[float, float]] = []
if trim:
cuts = _dead_regions(take_n, session, bucket_s, sr)
if cuts:
print("trimming (dead regions only):")
for a, b in cuts:
print(f" {a/60:6.2f}min -> {b/60:6.2f}min ({b-a:.0f}s)")
else:
print("trimming: nothing dead to cut")
def keep(t_from: float, t_to: float) -> bool:
mid = (t_from + t_to) / 2
return not any(a <= mid < b for a, b in cuts)
BLK = 1.0
n = int(BLK * sr)
total = int(dur * sr)
out.parent.mkdir(parents=True, exist_ok=True)
handles = {orb: [sf.SoundFile(str(p)) for p in ps] for orb, ps in pairs.items()}
written = 0
peak = 0.0
sumsq = 0.0
nsamp = 0
try:
with sf.SoundFile(str(out), "w", samplerate=sr, channels=2,
subtype="FLOAT") as w:
pos = 0
while pos < total:
acc = np.zeros((n, 2), dtype="float64")
got = 0
for orb, (fl, fr) in handles.items():
a = fl.read(n, dtype="float32", always_2d=False)
b = fr.read(n, dtype="float32", always_2d=False)
if a is None or len(a) == 0:
continue
got = max(got, len(a))
acc[:len(a), 0] += a
if b is not None and len(b):
acc[:len(b), 1] += b
if got == 0:
break
blk = (acc[:got] * g).astype("float32")
if keep(pos / sr, (pos + got) / sr):
w.write(blk)
written += got
peak = max(peak, float(np.max(np.abs(blk))) if got else 0.0)
sumsq += float(np.sum(np.square(blk, dtype="float64")))
nsamp += blk.size
pos += got
finally:
for hs in handles.values():
for h in hs:
h.close()
import math
pdb = 20 * math.log10(peak) if peak > 0 else -120.0
rdb = 20 * math.log10(math.sqrt(sumsq / nsamp)) if nsamp else -120.0
print(f"\nwrote {out}")
print(f" {written/sr/60:.2f} min (source {dur/60:.2f} min, "
f"{(dur - written/sr):.0f}s trimmed)")
print(f" gain applied {gain_db:+.1f} dB over {len(orbits)} orbits summed")
print(f" peak {pdb:+.2f} dBFS rms {rdb:.2f} dBFS")
if pdb >= 0.0:
print(" NOTE still over 0 dBFS — float output so it is intact, but it needs")
print(" more attenuation before any integer export or upload.")
if pdb <= -40.0:
print(" SUSPICIOUS: this is very quiet for a full mix — verify before serving.")
return 3
print("\nnext: verify the spectral balance rather than trusting the sum —")
print(f" python3 armada/tide-table/audio_lens.py profile '{out}'")
return 0
def _dead_regions(take_n: int, session: Path, bucket_s: float,
sr: int) -> list[tuple[float, float]]:
"""Offsets (seconds into the take) of everything in CUTTABLE.
Goes through `classify()` — the same call the report makes — so the mixer can
never cut something the report called `keep`.
"""
takes = find_takes(session)
t = takes[take_n]
t0, t1, dur = t["start"], t["end"], t["dur_s"]
paths = logs_covering(t0, t1)
rec = read_window(paths, t0, t1) if paths else {"cc": []}
ns, cc_n, _notes, _ctrls, active, _loud, _e = analyse(t, rec, GRAIN_S, True)
return [(c["start_s"], c["end_s"])
for c in classify(ns, GRAIN_S, dur, cc_n, active, len(t["orbits"]))
if c["kind"] in CUTTABLE]
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument("--session", type=Path, default=SESSION)
......@@ -450,9 +674,19 @@ def main(argv: list[str] | None = None) -> int:
lens.add_argument("--no-audio", action="store_true",
help="skip the audio read (instant, but trims stay ambiguous)")
lens.add_argument("--edl", type=Path, help="write a proposed EDL JSON here")
mix = sub.add_parser("mix", help="sum the orbits to one stereo file")
mix.add_argument("take", type=int)
mix.add_argument("--out", type=Path, required=True)
mix.add_argument("--gain-db", type=float, default=0.0,
help="uniform gain on every orbit (e.g. -16)")
mix.add_argument("--trim", action="store_true",
help="drop regions the lens calls `dead` (hands idle AND silent)")
mix.add_argument("--bucket", type=float, default=15.0)
a = ap.parse_args(argv)
if a.cmd == "takes":
return cmd_takes(a.session)
if a.cmd == "mix":
return cmd_mix(a.take, a.session, a.out, a.gain_db, a.trim, a.bucket)
return cmd_lens(a.take, a.bucket, a.session, not a.no_audio, a.edl)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment