Skip to content

T
Tidal
Commit cc9cbf48 by PLN (Algolia)
Options

feat(take-lens): pack a take so it travels with its performance — and fix the trim PLN caught 

Two things in one file, both from the same evening's practice recording.

=== PART 1: `pack` — the performance must travel with the audio (#106, #107)

"ensure recordings have the midi too, keep this from blowing up tracking either
ardour rec or compressing to avoid storing hours of no-move?"

Today the audio is permanent and the performance is not. Ardour's sources live
forever under the session; the MIDI and the track timeline sit in a gig-log
directory that prune() deletes on a timer. So the 29862 CC events that explain
take 94 were on a countdown while the audio they explain was not. `pack` writes a
sidecar next to the audio: meta.json (window, sr, subtype, per-orbit peak dBFS,
silent orbits, xrun delta, gear + boot state), midi.jsonl (every cc/note/track/mark
record in the window, copied verbatim out of the log so pruning cannot reach it),
and edl.json.

The sidecar states its own honest limit in a field: gig-log COALESCES CC per
(port, channel, controller) per second, keeping count + first/last/min/max. That is
a faithful summary of a knob sweep and is what makes the log cheap, but a .mid
rendered from it would be a reconstruction, not a recording. The JSONL is the truth.

`--compress` then wavpacks every source and verifies each one with `wvunpack -vm`
before reporting a single byte saved — an archiver that reports a ratio without
checking is reporting a hope. The .wav is NEVER deleted; removal stays a separate
human step after the take has been heard. FLAC was rejected outright: it cannot
store 32-bit float, and routing through s32 would hard-clip the 8 orbits sitting
over 0 dBFS while looking like a win on the silent one.

Two measurement bugs found by running it on real data rather than trusting it:
 - xrun is a RUNNING TOTAL in gig-log's `s` records, not a per-tick delta (its own
   CUMULATIVE tuple exists for exactly this mistake). Summing it reported 58,668,469
   xruns for a 21-minute take. Rebased to last-minus-first: 230, a real number.
 - Records are not events. A `cc` record is a one-second bucket carrying `n`, so
   printing the record count understated the performance 25x (1167 vs 29862).

=== PART 2: the trim threshold was protecting every bad transition

PLN heard the first machine-trimmed mix and found the hole: "silence around 18 is
not normal its a bad transition ahah in a real mixing work wed cut it proper and
fade or at least trim most". The gap he heard WAS detected and reported, and
survived anyway. SILENT_RUN_S was 20 s; the three gaps that made it into the mix
measured 17 s, 18 s and 11 s. Every one sat just under the bar. A threshold chosen
so the trimmer could never eat a musical break had silently become a threshold that
preserves every failed transition instead.

Three changes, all of them his rule rather than a tuning:

1. CLAMP, don't binary-cut. The bar drops to 10 s, but a detected silence is no
   longer removed whole — the last 2 s before the music resumes is kept as a
   breath. That is "or at least trim most" implemented literally, and it makes the
   lower threshold safe: the worst case if the rule fires on something musical is
   now a shortened rest, not a missing one. classify() computes both the detected
   region and the removed region so the report and the mixer cannot disagree.

2. FADE every seam. Every cut this tool had ever made was a butt-join, which in a
   live room is a click. Each removal now fades out into the cut and back in out of
   it over 250 ms, applied sample-accurately — the old mixer decided keep/drop per
   one-second block by its midpoint, quantising every boundary to +-0.5 s.
   Verified on the re-rendered take: max sample-to-sample step at the five seams is
   0.0001-0.0029 against a p99.99 of 0.0562 for the take as a whole, i.e. the joins
   are 20-100x below ordinary musical transients.

3. BRIDGE blips. A dead patch is rarely clean end to end, and one orbit's tail
   crossing -60 dBFS for a second splits it into pieces that each fall under the
   bar. Only single-orbit blips are bridged: a real hit lights several orbits, and
   that is what keeps this from swallowing a one-shot.

Result on take 94: 143 s trimmed instead of 119 s, all five gaps closed including
the one at +18.0 min he named, output 18.75 min, peak -3.36 dBFS.

=== `selftest`

Added because two of these rules could not be validated on real material — the
bridge fires zero times on take 94 and a correct fade is inaudible by construction.
A rule you have never seen fire is a hope, not a rule. It found a bug on the first
run, in the test's own arithmetic: asserting "no span survives a multi-orbit hit"
was wrong because the left half is exactly SILENT_RUN_S long and qualifies by
itself. The property that actually matters is that no trimmed span ever CONTAINS a
moment when the music was playing, and that is what it now asserts.

Correction to a verification, worth recording: the first check of the re-rendered
mix reported an 11 s silence still present. It did not exist. I measured the output
against -60 dBFS while the mix carries a uniform -16 dB, so I was testing at an
effective -44 dBFS at source. At the matched threshold exactly one rest over 4 s
remains, 9.25 s, correctly left alone as musical. The right lens per control, again.
parent b5061e23
Showing with 409 additions and 29 deletions
tools/take-lens.py
......@@ -35,6 +35,9 @@ USAGE
tools/take-lens.py lens 94 --bucket 10 # finer time resolution
tools/take-lens.py lens 94 --edl OUT.json # also write a proposed EDL
tools/take-lens.py lens 94 --no-audio # MIDI only (instant, no 3 GB read)
tools/take-lens.py pack 94 # sidecar meta+midi+edl beside the audio
tools/take-lens.py pack 94 --compress # ...and wavpack it, verified lossless
tools/take-lens.py mix 94 --out M.wav --trim --gain-db -16
"""
from __future__ import annotations
......@@ -69,10 +72,29 @@ IDLE_RUN_S = 45.0
# 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
# exactly the practice/plugging-in the trim exists to remove.
#
# 10s, not 20s. The 20s bar was chosen so the trimmer could never eat a musical break
# (4 bars is under 8s at 120-140 BPM) — but PLN heard the result and it had become the
# opposite guarantee: "silence around 18 is not normal its a bad transition ahah in a
# real mixing work wed cut it proper and fade or at least trim most". All three gaps
# that survived the first trim measured 17s, 18s and 11s — every one of them sat just
# under the bar. A threshold that protects musical rests was preserving every bad
# transition instead.
SILENT_RUN_S = 10.0
# ...and the answer to a bad transition is not "delete every trace of it" either. A
# reported silence is CLAMPED, not removed: the last SILENT_KEEP_S before the music
# resumes is kept as a breath, and everything before that goes. That is "or at least
# trim most" implemented literally, and it means the rule stays safe even if it fires
# on something musical — the worst case is a shortened rest, not a missing one.
SILENT_KEEP_S = 2.0
# Every cut this tool made before today was a BUTT-JOIN, which in a live room is a
# click. Each removal now fades out into the cut and back in out of it. 250ms is long
# enough to be inaudible as a transient and far too short to smear a downbeat.
SEAM_FADE_S = 0.25
# A dead patch split by a one-second tail crossing the threshold is still one dead
# patch. Blips no longer than this, with at most one orbit sounding, are bridged.
SILENT_BRIDGE_S = 2.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.
......@@ -328,12 +350,21 @@ def classify(nb: int, bucket_s: float, dur: float, cc_n: list[int],
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.
Each candidate carries BOTH the detected region (`start_s`/`end_s`) and the
region a mixer would actually remove (`cut_start_s`/`cut_end_s`), because for
`silence` those differ: the tail is kept as a breath. Computing the cut here
rather than in the mixer is the whole point of this function — a mixer that
derives its own bounds is a mixer that can silently disagree with the report.
"""
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,
s, e = a * bucket_s, min(dur, b * bucket_s)
# A silence is clamped to a breath; anything else is removed whole.
cs, ce = (s, max(s, e - SILENT_KEEP_S)) if kind == "silence" else (s, e)
out.append({"start_s": s, "end_s": e, "dur_s": e - s,
"cut_start_s": cs, "cut_end_s": ce, "cut_dur_s": ce - cs,
"kind": kind, "why": why})
def runs(pred) -> list[tuple[int, int]]:
......@@ -351,7 +382,23 @@ def classify(nb: int, bucket_s: float, dur: float, cc_n: list[int],
have_audio = any(active)
silent_spans: list[tuple[int, int]] = []
if have_audio:
for a, b in runs(lambda i: active[i] == 0):
# CLOSE the silence first. A dead patch is rarely digitally clean end to
# end: one orbit's tail crossing -60 dBFS for a second splits it into
# pieces that each fall under SILENT_RUN_S, and the whole patch survives.
# Measured on take 94: a 21s dead stretch came through as 4.25 + 11.25 +
# 4.25 with two blips between, so the trim removed none of it.
#
# Only ONE orbit may be sounding in a bridged blip. That is the difference
# between a decaying tail and a musical event — a real hit lights several
# orbits — and it is what keeps this from swallowing a one-shot.
sil = [active[i] == 0 for i in range(nb)]
bridge = max(1, int(SILENT_BRIDGE_S / bucket_s))
for a, b in runs(lambda i: not sil[i]):
if (a > 0 and b < nb and (b - a) <= bridge
and max(active[a:b], default=99) <= 1):
for i in range(a, b):
sil[i] = True
for a, b in runs(lambda i: sil[i]):
if (b - a) * bucket_s >= SILENT_RUN_S:
silent_spans.append((a, b))
emit(a, b, "silence", "every orbit under %.0f dBFS" % SILENT_DBFS)
......@@ -373,12 +420,17 @@ def classify(nb: int, bucket_s: float, dur: float, cc_n: list[int],
continue
seg = active[a:b]
avg = sum(seg) / len(seg)
# A hands-idle run can CONTAIN a shorter silence without being mostly
# silent (25% overlap fails the 0.8 dedup above). Say so, or the table
# reads as self-contradictory: a span labelled "a real part, keep" sitting
# across a stretch the next line marks TRIM.
hole = (f"; holds {covered:.0f}s of trimmed silence" if covered else "")
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)
emit(a, b, "loop", "%.2f orbits avg, hands off%s" % (avg, hole))
else:
emit(a, b, "playing", "%.2f orbits avg — a real part, keep" % avg)
emit(a, b, "playing", "%.2f orbits avg — a real part, keep%s" % (avg, hole))
out.sort(key=lambda c: c["start_s"])
return out
......@@ -501,15 +553,20 @@ def cmd_lens(take_n: int, bucket_s: float, session: Path,
c["start"] = t0 + c["start_s"]
c["end"] = t0 + c["end_s"]
cut_s = sum(c["dur_s"] for c in cands if c["kind"] in CUTTABLE)
cut_s = sum(c["cut_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 = {"silence": "TRIM", "dead": "TRIM", "loop": "trim?",
"playing": "keep", "unknown": "?"}[c["kind"]]
# A silence shows what is REMOVED vs what is DETECTED, because they differ:
# the last SILENT_KEEP_S is left as a breath before the music resumes.
cut = ("" if c["kind"] not in CUTTABLE or
abs(c["cut_dur_s"] - c["dur_s"]) < 0.05
else f" cut {c['cut_dur_s']:.0f}s, {SILENT_KEEP_S:.0f}s kept")
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']})")
f"+{c['start_s']/60:>5.1f}min ({c['why']}){cut}")
if edl_out:
edl = {
......@@ -570,20 +627,18 @@ def cmd_mix(take_n: int, session: Path, out: Path, gain_db: float | None,
gain_db = 0.0
g = 10.0 ** (gain_db / 20.0)
# Trim only regions the lens called `dead` — never `loop`, never `playing`.
# Trim only what classify() marked CUTTABLE — never `loop`, never `playing`.
cuts: list[tuple[float, float]] = []
if trim:
cuts = _dead_regions(take_n, session, bucket_s, sr)
cuts = cut_regions(take_n, session, sr)
if cuts:
print("trimming (dead regions only):")
print("trimming (silence clamped to a breath, dead removed whole):")
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)
print("trimming: nothing to cut")
cuts_smp = [(int(a * sr), int(b * sr)) for a, b in cuts]
fade = int(SEAM_FADE_S * sr)
BLK = 1.0
n = int(BLK * sr)
......@@ -613,10 +668,16 @@ def cmd_mix(take_n: int, session: Path, out: Path, gain_db: float | None,
if got == 0:
break
blk = (acc[:got] * g).astype("float32")
if keep(pos / sr, (pos + got) / sr):
# Sample-accurate, not block-accurate. The old version decided
# keep/drop per one-second block by its midpoint, which quantised
# every boundary to +-0.5s and butt-joined the result.
mask, env = _seam_env(pos, got, cuts_smp, fade, np)
blk *= env[:, None]
blk = blk[mask]
if len(blk):
w.write(blk)
written += got
peak = max(peak, float(np.max(np.abs(blk))) if got else 0.0)
written += len(blk)
peak = max(peak, float(np.max(np.abs(blk))))
sumsq += float(np.sum(np.square(blk, dtype="float64")))
nsamp += blk.size
pos += got
......@@ -644,12 +705,231 @@ def cmd_mix(take_n: int, session: Path, out: Path, gain_db: float | None,
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.
# ------------------------------------------------------------------------ pack
# Where a packed take lives. Beside the audio, deliberately: the entire point is that
# the performance data stops living in a directory that a prune timer owns.
def pack_dir(session: Path, take_n: int) -> Path:
return session / "takes" / f"take{take_n}"
Goes through `classify()` — the same call the report makes — so the mixer can
never cut something the report called `keep`.
def cmd_pack(take_n: int, session: Path, out_dir: Path | None,
compress: bool) -> int:
"""Make a take self-describing, then make it affordable.
Two problems with one answer. Today the audio is permanent and the performance
is not: Ardour's sources live forever under the session while the MIDI and the
track timeline sit in `~/.local/share/parvagues/gig-log/`, which `prune()`
deletes on a timer. So the 29862 CC events that explain take 94 are on a
countdown and the audio they explain is not. Copying the window's records out,
next to the audio, ends that.
And 12 orbits x stereo x 32-bit float x 48 kHz is 277 MB/min — 16.6 GB/hour of
practice. `wavpack -h` gets ~4x on real material (measured per-orbit: 42% on the
busiest, 7% on a sparse one) and is verified lossless here on every file before
anything is reported as archived.
THE .WAV IS NEVER DELETED. Removal is a separate, explicit, human step after the
take has been heard. A silent data-losing archiver is not worth 4x.
"""
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]
t0, t1, dur, sr = t["start"], t["end"], t["dur_s"], t["sr"]
if not t["exact"]:
print("take-lens: length is ESTIMATED for this take — refusing to pack, "
"because every window bound below would be a guess", file=sys.stderr)
return 1
out = out_dir or pack_dir(session, take_n)
out.mkdir(parents=True, exist_ok=True)
print(f"take {take_n} {hhmmss(t0)} -> {hhmmss(t1)} {dur/60:.1f} min")
print(f"pack -> {out}")
# --- the performance ------------------------------------------------------
paths = logs_covering(t0, t1)
rec = read_window(paths, t0, t1) if paths else {
"cc": [], "note": [], "s": [], "track": [], "mark": []}
midi_p = out / f"take{take_n}.midi.jsonl"
n_midi = 0
with midi_p.open("w") as f:
# Sorted across kinds so the file reads as one timeline, and offset_s added
# so a reader never has to know the take's epoch to place an event.
for r in sorted((r for k in ("cc", "note", "track", "mark")
for r in rec.get(k, [])), key=lambda r: r["t"]):
r = dict(r, offset_s=round(r["t"] - t0, 3))
f.write(json.dumps(r, separators=(",", ":")) + "\n")
n_midi += 1
# Records != events: a `cc` record is a one-second bucket carrying `n`. Printing
# only the record count would understate the performance by ~25x on this take.
n_ev = sum(r.get("n", 1) for k in ("cc", "note") for r in rec.get(k, []))
print(f" midi {n_midi:>7} records ({n_ev} events) -> {midi_p.name}")
# --- the measurements -----------------------------------------------------
ns, cc_n, notes, ctrls, active, loud, energy = analyse(t, rec, GRAIN_S, True)
cands = classify(ns, GRAIN_S, dur, cc_n, active, len(t["orbits"]))
# xrun is a RUNNING TOTAL since the log started, not a per-tick delta — see
# gig-log.py's CUMULATIVE tuple, which exists for exactly this mistake. Summing
# it reported 58,668,469 xruns for a 21-minute take. Window value = last - first,
# the same rebase gig-log's own slice_session() does.
ss = [r for r in rec.get("s", []) if r.get("xrun") is not None]
xruns = max(0, int(ss[-1]["xrun"]) - int(ss[0]["xrun"])) if len(ss) > 1 else 0
hdr = None
for p in paths:
try:
with p.open() as f:
h = json.loads(f.readline())
if h.get("k") == "hdr":
hdr = h
break
except (OSError, json.JSONDecodeError):
continue
meta = {
"take": take_n,
"session": str(session),
"start_epoch": t0, "end_epoch": t1,
"start_local": hhmmss(t0), "end_local": hhmmss(t1),
"iso": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(t0)),
"dur_s": dur, "sr": sr, "subtype": t.get("subtype"),
"orbits": sorted(t["orbits"]),
"peak_dbfs": {f"d{o}": round(e["peak_all_db"], 2)
for o, e in energy.items()},
"silent_orbits": [f"d{o}" for o, e in energy.items()
if e["peak_all_db"] <= SILENT_DBFS],
"hands": {"cc": sum(cc_n), "notes": sum(notes),
"controllers": sorted(set().union(*ctrls)) if any(ctrls) else [],
"idle_s": sum(1 for v in cc_n if v == 0), "steps": ns},
"xruns": xruns,
"gig_logs": [p.name for p in paths],
"track_timeline": [{"offset_s": round(r["t"] - t0, 3),
"path": r.get("path")} for r in rec["track"]],
"gear": (hdr or {}).get("gear"),
"boot": {k: (hdr or {}).get(k)
for k in ("mode", "gov", "fmax", "gpu", "iso")} if hdr else None,
# The honest limit, stated IN the artifact so a future reader cannot miss it.
"midi_fidelity": (
"COALESCED, not sample-accurate. gig-log groups CC per "
"(port, channel, controller) per second, keeping count + first/last/"
"min/max. That is a faithful summary of a knob sweep and is what makes "
"the log cheap, but a .mid rendered from it would be a reconstruction, "
"not a recording. The JSONL is the truth."),
"generated_by": "tools/take-lens.py pack",
}
if not rec["track"]:
meta["track_timeline_note"] = (
"NONE. Nothing published the loaded track while this take recorded "
"(gig-log gained k:\"track\" on 2026-07-29). Boundaries inside this "
"take can only be inferred — do not synthesise them.")
(out / f"take{take_n}.meta.json").write_text(json.dumps(meta, indent=2) + "\n")
print(f" meta {len(t['orbits'])} orbits, {sum(cc_n)} CC, {xruns} xruns"
f"{'' if rec['track'] else ', NO track timeline'}")
edl = {"take": take_n, "session": str(session),
"start_epoch": t0, "end_epoch": t1, "dur_s": dur, "sr": sr,
"orbits": sorted(t["orbits"]),
"track_timeline": meta["track_timeline"],
"trim_candidates": cands,
"seam_fade_s": SEAM_FADE_S, "silence_keep_s": SILENT_KEEP_S,
"generated_by": "tools/take-lens.py pack"}
(out / f"take{take_n}.edl.json").write_text(json.dumps(edl, indent=2) + "\n")
cut_s = sum(c["cut_dur_s"] for c in cands if c["kind"] in CUTTABLE)
print(f" edl {len(cands)} candidates, {cut_s:.0f}s cuttable "
f"({100*cut_s/dur:.0f}%)")
if not compress:
print("\n(no --compress: sidecar only)")
return 0
return _archive(t, out, take_n)
def _archive(t: dict, out: Path, take_n: int) -> int:
"""wavpack every source of the take, verify each, report. Never deletes."""
import shutil
import subprocess
if not shutil.which("wavpack") or not shutil.which("wvunpack"):
print("\ncompress: wavpack/wvunpack not on PATH — skipped", file=sys.stderr)
return 2
srcs: list[Path] = []
for lp in t["orbits"].values():
srcs.append(lp)
rp = Path(str(lp).replace("%L.wav", "%R.wav"))
if rp.exists():
srcs.append(rp)
wvdir = out / "wv"
wvdir.mkdir(exist_ok=True)
print(f"\ncompress: {len(srcs)} sources -> {wvdir}")
raw = comp = 0
bad = []
for i, s in enumerate(sorted(srcs), 1):
dst = wvdir / (s.stem + ".wv")
if not dst.exists():
r = subprocess.run(["wavpack", "-h", "-q", "-y", str(s), "-o", str(dst)],
capture_output=True, text=True)
if r.returncode != 0:
bad.append((s.name, "wavpack: " + r.stderr.strip()[:120]))
continue
# -vm verifies the decoded md5 against the one stored at encode time. An
# archiver that reports a ratio without checking is reporting a hope.
v = subprocess.run(["wvunpack", "-vm", "-q", str(dst)],
capture_output=True, text=True)
if v.returncode != 0:
bad.append((s.name, "VERIFY FAILED: " + v.stderr.strip()[:120]))
continue
raw += s.stat().st_size
comp += dst.stat().st_size
print(f" [{i:>2}/{len(srcs)}] {s.stat().st_size/1e6:>7.1f} MB -> "
f"{dst.stat().st_size/1e6:>6.1f} MB {dst.name}")
if raw:
print(f"\n {raw/1e9:.2f} GB -> {comp/1e9:.2f} GB "
f"({100*comp/raw:.0f}%, saved {(raw-comp)/1e9:.2f} GB) — "
f"all verified lossless")
if bad:
print(f"\n {len(bad)} FAILED:", file=sys.stderr)
for name, why in bad:
print(f" {name}: {why}", file=sys.stderr)
return 1
print(" the .wav sources are UNTOUCHED. Removing them is a separate, human")
print(" step — after the take has been heard, never before.")
return 0
def _seam_env(p0: int, n: int, cuts_smp: list[tuple[int, int]], fade: int, np):
"""Keep-mask + fade envelope for the sample range [p0, p0+n).
Every removal fades OUT over the last `fade` samples before it and back IN over
the first `fade` samples after it. Overlapping fades take the minimum, so two
cuts a few hundred ms apart cannot sum back above unity.
This is a cut-with-fades, not a crossfade: nothing overlaps, so the timeline
stays a plain concatenation and the reported cut durations remain exact.
"""
idx = np.arange(p0, p0 + n)
keep = np.ones(n, dtype=bool)
env = np.ones(n, dtype="float32")
for cs, ce in cuts_smp:
keep &= ~((idx >= cs) & (idx < ce))
if fade <= 0:
continue
m = (idx >= cs - fade) & (idx < cs)
if m.any():
env[m] = np.minimum(env[m], (cs - idx[m]) / fade)
m = (idx >= ce) & (idx < ce + fade)
if m.any():
env[m] = np.minimum(env[m], (idx[m] - ce + 1) / fade)
return keep, env
def cut_regions(take_n: int, session: Path, sr: int) -> list[tuple[float, float]]:
"""Offsets (seconds into the take) of everything a mixer should remove.
Goes through `classify()` — the same call the report makes — and uses the
`cut_*` bounds it computed, so the mixer can never cut something the report
called `keep`, nor remove more of a silence than the report said it would.
"""
takes = find_takes(session)
t = takes[take_n]
......@@ -657,9 +937,96 @@ def _dead_regions(take_n: int, session: Path, bucket_s: float,
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"])
return [(c["cut_start_s"], c["cut_end_s"])
for c in classify(ns, GRAIN_S, dur, cc_n, active, len(t["orbits"]))
if c["kind"] in CUTTABLE]
if c["kind"] in CUTTABLE and c["cut_dur_s"] > 0]
# -------------------------------------------------------------------- selftest
def cmd_selftest() -> int:
"""Exercise the cut rules on synthetic input, with no rig and no audio.
Written because two of the rules below could not be validated on take 94: the
silence BRIDGE never fires there (measured: zero bridgeable blips), and the
seam fade is inaudible by construction. A rule you have never seen fire is not
a rule, it is a hope — same lesson as check-boot's Pass 5.
"""
import math
fails = []
def check(name: str, ok: bool, detail: str = "") -> None:
print(f" {'PASS' if ok else 'FAIL'} {name}{' ' + detail if detail else ''}")
if not ok:
fails.append(name)
print("classify:")
n = 60
# A 20s silence in the middle of a playing take.
act = [4] * n
for i in range(20, 40):
act[i] = 0
c = classify(n, 1.0, float(n), [5] * n, act, 12)
sil = [x for x in c if x["kind"] == "silence"]
check("a 20s silence is detected", len(sil) == 1)
if sil:
check("...and CLAMPED, not removed",
abs(sil[0]["cut_dur_s"] - (20 - SILENT_KEEP_S)) < 1e-6,
f"cut {sil[0]['cut_dur_s']:.0f}s of {sil[0]['dur_s']:.0f}s")
check("...leaving the breath at the END (music resumes into it)",
abs(sil[0]["cut_end_s"] - (sil[0]["end_s"] - SILENT_KEEP_S)) < 1e-6)
# Shorter than the bar: a musical rest must survive untouched.
act = [4] * n
for i in range(20, 28):
act[i] = 0
check("an 8s rest is NOT cut (4 bars at 120-140 BPM)",
not [x for x in classify(n, 1.0, float(n), [5] * n, act, 12)
if x["kind"] == "silence"])
# The bridge: a 20s dead patch split by a 1s single-orbit tail.
act = [4] * n
for i in range(20, 40):
act[i] = 0
act[30] = 1
c = classify(n, 1.0, float(n), [5] * n, act, 12)
sil = [x for x in c if x["kind"] == "silence"]
check("a 1-orbit tail does not split a dead patch",
len(sil) == 1 and sil[0]["dur_s"] >= 19,
f"{len(sil)} span(s)")
# ...but a real event (several orbits) must break it. The property is NOT "no
# span survives" — the left half is exactly SILENT_RUN_S long and legitimately
# qualifies on its own. The property is that no span CONTAINS the hit, i.e. the
# trim can never remove a moment where the music was actually playing.
act[30] = 5
sil = [x for x in classify(n, 1.0, float(n), [5] * n, act, 12)
if x["kind"] == "silence"]
swallowed = [x for x in sil if x["start_s"] <= 30 < x["end_s"]]
check("a MULTI-orbit hit is never inside a trimmed span", not swallowed,
f"{len(sil)} span(s), none covering the hit")
print("seam fade:")
try:
import numpy as np
except ImportError:
print(" SKIP numpy unavailable")
else:
cuts = [(1000, 2000)]
keep, env = _seam_env(0, 3000, cuts, 100, np)
check("the cut region is dropped", int((~keep).sum()) == 1000)
check("fade-out reaches ~0 at the cut edge", env[999] < 0.02,
f"env[999]={env[999]:.3f}")
check("fade-in starts at ~0 after the cut", env[2000] < 0.02,
f"env[2000]={env[2000]:.3f}")
check("unity away from any seam",
float(env[0]) == 1.0 and float(env[2999]) == 1.0)
# Two cuts closer together than the fade must not sum above unity.
_, env2 = _seam_env(0, 3000, [(1000, 1100), (1150, 1250)], 200, np)
check("overlapping fades never exceed unity", float(env2.max()) <= 1.0)
print(f"\n{'ALL PASS' if not fails else str(len(fails)) + ' FAILED: ' + ', '.join(fails)}")
return 0 if not fails else 1
def main(argv: list[str] | None = None) -> int:
......@@ -667,6 +1034,7 @@ def main(argv: list[str] | None = None) -> int:
ap.add_argument("--session", type=Path, default=SESSION)
sub = ap.add_subparsers(dest="cmd", required=True)
sub.add_parser("takes", help="list recorded takes with wall-clock windows")
sub.add_parser("selftest", help="exercise the cut rules on synthetic input")
lens = sub.add_parser("lens", help="correlate one take: hands, sound, tracks")
lens.add_argument("take", type=int)
lens.add_argument("--bucket", type=float, default=15.0,
......@@ -674,6 +1042,14 @@ 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")
pk = sub.add_parser("pack", help="sidecar the take's MIDI/meta/EDL beside the "
"audio, and optionally archive it")
pk.add_argument("take", type=int)
pk.add_argument("--out", type=Path,
help="pack dir (default <session>/takes/take<N>/)")
pk.add_argument("--compress", action="store_true",
help="also wavpack every source, verified lossless "
"(the .wav is never deleted)")
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)
......@@ -685,6 +1061,10 @@ def main(argv: list[str] | None = None) -> int:
a = ap.parse_args(argv)
if a.cmd == "takes":
return cmd_takes(a.session)
if a.cmd == "selftest":
return cmd_selftest()
if a.cmd == "pack":
return cmd_pack(a.take, a.session, a.out, a.compress)
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