feat(tools): take-lens — read a recorded take through the gig log, and find the trims

PLN asked for "a master split per track loaded time, cleanup parts obviously
'practice or plugging in' that are scarce VS intro/outro (can we detect this? did we
record play metadata as the ardour records?)".

Three datasets existed and had never been joined:
  - Ardour names sources `Take<N>_Tidal <orbit>-<seq>%<L|R>.wav`, so the take number
    plus the frame count plus the mtime give an exact wall-clock window with no
    session-XML parsing at all.
  - gig-log has coalesced MIDI CC at 1 Hz: 29862 events over this take, on 46
    distinct controllers. Hands-idle is the clearest signature of "plugging in".
  - per-orbit RMS/peak says what the SOUND did.

The third one is not optional, and that is the design insight. Idle hands ALONE is
ambiguous: you can play a part for three minutes without touching a knob. So a
candidate is classified by what the audio was doing underneath it — `dead` (nothing
above -60 dBFS: safe to cut), `loop` (a couple of orbits droning, hands off:
probably plugging in), `playing` (most orbits going: a real part, KEEP). On take 94
that distinction earned its keep immediately: of three hands-idle runs, two were
real parts at 8 and 4 orbits average, and only the 75s tail was actually dead.

FINDINGS ON TAKE 94 (21.1 min, 20:08:33 -> 20:29:41, 12 orbits)
  - 8 of 12 orbits peak OVER 0 dBFS, worst d4 at +9.63 dB. The files are 32-bit
    FLOAT, so this is headroom and not damage — a gain cut recovers it exactly, and
    it only becomes real clipping on integer export or at the master bus. Labelling
    it "CLIPPING" would have been a false alarm sending him to re-record material
    that is completely intact, so the tool now reads the subtype and says which it
    is. A uniform -16 dB on the record bus lands the hottest orbit on the -6 dBFS
    source target.
  - d12 silent for the whole take (-78 dBFS).
  - no track timeline: nothing published the loaded track while it recorded. The
    tool says so plainly instead of guessing, because a confident wrong boundary is
    worse than none — the existing Bandcamp splits carry crossfade bleed for exactly
    that reason. Takes from d5baca32 onward split exactly.

A MEASUREMENT TRAP, RECORDED IN THE CODE
The first pass estimated duration as size/(sr*3), assuming 24-bit. These files are
32-bit float. That put take 94 at 28.2 min starting 20:01:30 instead of 21.1 min
starting 20:08:33 — and manufactured a "6m15s dead intro" that never existed, purely
from a 7-minute window shift. The fallback now assumes 4 bytes and, more
importantly, exact lengths come from the frame count and estimated ones are labelled
"do not cut from it".

Also: logs_covering() checks EVERY log's [first,last] range rather than assuming the
newest one holds the window, because a take can straddle a recorder restart — which
is exactly what happened today.

It proposes, it does not cut: output is a report plus an EDL JSON for
armada/tide-table/edl_render.py. Trims are a taste call and the ear is the gate.

tools/take-lens.py

+#!/usr/bin/env python3
+"""take-lens — read a recorded Ardour take through the gig log.
+
+WHY THIS EXISTS (2026-07-29)
+
+PLN recorded 28 minutes of OPAL practice and asked for "a master, split per track
+loaded time, cleanup parts obviously 'practice or plugging in' that are scarce VS
+intro/outro (can we detect this?)".
+
+The answer needed three datasets that had never been joined:
+
+  1. WHAT WAS RECORDED — Ardour writes one file per orbit per take, named
+     `Take<N>_Tidal <orbit>-<seq>%<L|R>.wav`. The take number and the file length
+     give an exact wall-clock window, because the mtime is the moment recording
+     stopped. No session XML parsing needed.
+  2. WHAT THE HANDS DID — `tools/gig-log.py` logs coalesced MIDI CC at 1 Hz. Hands
+     idle for six minutes is the single clearest signature of "plugging in".
+  3. WHAT THE SOUND DID — per-orbit RMS and peak. Needed because idle hands alone
+     are ambiguous: you can play a track for three minutes without touching a knob.
+     Only `idle hands AND flat audio` means dead air.
+
+Plus, from 2026-07-29 onward, `k:"track"` records (see gig-log.py) give the loaded
+.tidal at every moment — which turns "split per track" from inference into a read.
+Takes recorded BEFORE that lands have no track timeline; this tool says so plainly
+rather than guessing, because a confident wrong boundary is worse than no boundary
+(the existing Bandcamp splits carry crossfade bleed for exactly that reason).
+
+WHAT IT DOES NOT DO
+It proposes; it does not cut. Output is a report plus an EDL JSON for
+`armada/tide-table/edl_render.py`. Trims are a taste call and the ear is the gate.
+
+USAGE
+  tools/take-lens.py takes                    # what takes exist, with wall-clock windows
+  tools/take-lens.py lens 94                  # correlate take 94: hands, sound, tracks
+  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)
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+import time
+from pathlib import Path
+
+SESSION = Path(os.environ.get(
+    "PARVAGUES_ARDOUR_SESSION",
+    os.path.expanduser("~/Work/Sound/Ardour/Tidal Live")))
+LOG_DIR = Path(os.environ.get(
+    "GIG_LOG_DIR", os.path.expanduser("~/.local/share/parvagues/gig-log")))
+
+# Ardour source naming for this session. The %L/%R suffix is the channel: each orbit
+# records as TWO mono files, so only %L is read for activity (a stereo orbit whose
+# channels differ in ENERGY would be a routing bug, not a musical event).
+SRC_RE = re.compile(r"Take(\d+)_Tidal (\d+)-(\d+)%([LR])\.wav$")
+
+BAR = " ▁▂▃▄▅▆▇█"
+
+# A hands-idle run at least this long is a TRIM CANDIDATE — but only when the audio
+# agrees. 45s was chosen because it is longer than any musical phrase in the set at
+# 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
+# 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.
+SILENT_DBFS = -60.0
+
+
+def sparkline(vals: list[float], lo: float | None = None,
+              hi: float | None = None) -> str:
+    """Eight-level sparkline. Empty string for no data.
+
+    Scaled to the data unless bounds are given. A near-zero baseline is NOT turned
+    into a percentage — the right lens per control, learned the hard way.
+    """
+    if not vals:
+        return ""
+    lo = min(vals) if lo is None else lo
+    hi = max(vals) if hi is None else hi
+    if hi <= lo:
+        return BAR[0] * len(vals)
+    out = []
+    for v in vals:
+        f = (v - lo) / (hi - lo)
+        out.append(BAR[max(0, min(8, int(round(8 * f))))])
+    return "".join(out)
+
+
+def hhmmss(t: float) -> str:
+    return time.strftime("%H:%M:%S", time.localtime(t))
+
+
+def hhmm(t: float) -> str:
+    return time.strftime("%H:%M", time.localtime(t))
+
+
+# ---------------------------------------------------------------- takes discovery
+
+def find_takes(session: Path = SESSION) -> dict[int, dict]:
+    """Map take number -> {orbits: {n: path}, dur_s, start, end, sr}.
+
+    Duration comes from the frame count via soundfile when available, falling back
+    to size/(sr*bytes) — the fallback is only used for the listing, never for a cut.
+    `end` is the file mtime: Ardour writes until the transport stops, so the last
+    write IS the end of the take. `start` is end - duration.
+    """
+    try:
+        import soundfile as sf
+    except ImportError:
+        sf = None
+    takes: dict[int, dict] = {}
+    inter = session / "interchange"
+    if not inter.is_dir():
+        return takes
+    for f in inter.rglob("*.wav"):
+        m = SRC_RE.search(f.name)
+        if not m or m.group(4) != "L":
+            continue
+        take, orbit = int(m.group(1)), int(m.group(2))
+        st = f.stat()
+        t = takes.setdefault(take, {"orbits": {}, "dur_s": 0.0, "end": 0.0,
+                                    "sr": 48000, "exact": False})
+        t["orbits"][orbit] = f
+        t["end"] = max(t["end"], st.st_mtime)
+        if sf is not None:
+            try:
+                info = sf.info(str(f))
+                t["sr"] = info.samplerate
+                t["subtype"] = info.subtype           # PCM_24, FLOAT, ...
+                t["dur_s"] = max(t["dur_s"], info.frames / info.samplerate)
+                t["exact"] = True
+                continue
+            except Exception:
+                pass
+        # Fallback only, and only for the listing. NOTE this guessed 3 bytes/frame
+        # and was WRONG by 33% on this session, which are 32-bit FLOAT: it put the
+        # take's start 7 minutes early and invented a 6-minute dead intro that did
+        # not exist. Never cut from an estimated length.
+        t["dur_s"] = max(t["dur_s"], st.st_size / (t["sr"] * 4))
+    for t in takes.values():
+        t["start"] = t["end"] - t["dur_s"]
+    return takes
+
+
+def cmd_takes(session: Path) -> int:
+    takes = find_takes(session)
+    if not takes:
+        print(f"take-lens: no Take*_Tidal sources under {session}/interchange",
+              file=sys.stderr)
+        return 1
+    print(f"session: {session}")
+    print("take  orbits  minutes  window                    len")
+    for n in sorted(takes):
+        t = takes[n]
+        miss = [o for o in range(1, 13) if o not in t["orbits"]]
+        flag = "" if not miss else f"  MISSING orbits {miss}"
+        mark = "" if t["exact"] else "  (length estimated)"
+        print(f"{n:>4}  {len(t['orbits']):>6}  {t['dur_s']/60:>7.1f}  "
+              f"{hhmmss(t['start'])} -> {hhmmss(t['end'])}{mark}{flag}")
+    return 0
+
+
+# ------------------------------------------------------------------- the gig log
+
+def logs_covering(t0: float, t1: float, d: Path = LOG_DIR) -> list[Path]:
+    """Every gig log whose [first, last] sample range overlaps the window.
+
+    A take can straddle a log rotation (a restart of the recorder mid-session), so
+    this must never assume the newest log holds everything. Reads only the first and
+    last lines to decide — the files run to tens of MB.
+    """
+    out = []
+    for p in sorted(d.glob("gig-*.jsonl")):
+        try:
+            with p.open("rb") as f:
+                first = f.readline().decode("utf-8", "replace")
+                try:
+                    f.seek(-4096, os.SEEK_END)
+                except OSError:
+                    f.seek(0)
+                tail = f.read().decode("utf-8", "replace").splitlines()
+            a = json.loads(first).get("t")
+            b = None
+            for line in reversed(tail):
+                try:
+                    b = json.loads(line).get("t")
+                except json.JSONDecodeError:
+                    continue
+                if b:
+                    break
+            if a is None or b is None:
+                continue
+            if a <= t1 and b >= t0:
+                out.append(p)
+        except (OSError, json.JSONDecodeError, KeyError):
+            continue
+    return out
+
+
+def read_window(paths: list[Path], t0: float, t1: float) -> dict:
+    """Pull every record inside the window, grouped by kind."""
+    got = {"cc": [], "note": [], "s": [], "track": [], "mark": []}
+    for p in paths:
+        with p.open() as f:
+            for line in f:
+                if '"t":' not in line:
+                    continue
+                try:
+                    r = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                t = r.get("t")
+                k = r.get("k")
+                if t is None or k not in got or not (t0 <= t <= t1):
+                    continue
+                got[k].append(r)
+    for v in got.values():
+        v.sort(key=lambda r: r["t"])
+    return got
+
+
+# ------------------------------------------------------------------ audio energy
+
+def orbit_energy(path: Path, bucket_s: float, nb: int, sr: int) -> dict:
+    """Per-bucket RMS and peak dBFS for one orbit, plus overall peak.
+
+    Read sequentially in bucket-sized blocks: one pass, no seeking, constant memory.
+    A whole 28-minute orbit is ~240 MB and takes about a second off warm cache.
+    """
+    import numpy as np
+    import soundfile as sf
+    n = int(bucket_s * sr)
+    rms = [0.0] * nb
+    pk = [0.0] * nb
+    i = 0
+    with sf.SoundFile(str(path)) as f:
+        while i < nb:
+            blk = f.read(n, dtype="float32", always_2d=False)
+            if blk is None or len(blk) == 0:
+                break
+            if blk.ndim > 1:
+                blk = blk.mean(axis=1)
+            rms[i] = float(np.sqrt(np.mean(np.square(blk, dtype="float64"))))
+            pk[i] = float(np.max(np.abs(blk)))
+            i += 1
+
+    def db(x: float) -> float:
+        return 20.0 * __import__("math").log10(x) if x > 1e-12 else -120.0
+
+    return {"rms_db": [db(v) for v in rms], "peak_db": [db(v) for v in pk],
+            "peak_all_db": db(max(pk) if pk else 0.0)}
+
+
+# ------------------------------------------------------------------------- lens
+
+def cmd_lens(take_n: int, bucket_s: float, session: Path,
+             audio: bool, edl_out: Path | None) -> int:
+    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"]
+    nb = max(1, int(dur // bucket_s) + 1)
+
+    print(f"take {take_n}  {hhmmss(t0)} -> {hhmmss(t1)}  "
+          f"{dur/60:.1f} min  {len(t['orbits'])} orbits  {sr} Hz")
+    if not t["exact"]:
+        print("  NOTE length is estimated from file size — do not cut from it")
+
+    paths = logs_covering(t0, t1)
+    if not paths:
+        print("  no gig log covers this window — hands + xruns unavailable")
+        rec = {"cc": [], "note": [], "s": [], "track": [], "mark": []}
+    else:
+        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)
+
+    # --- 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,
+                     hi=float(len(t["orbits"]) or 1)) if energy else ""
+    lvl = sparkline(loud, 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]}")
+        if orbs:
+            print(f"    orbits {orbs[i:i+W]}")
+            print(f"    level  {lvl[i:i+W]}")
+
+    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}%)")
+
+    if energy:
+        peaks = {o: e["peak_all_db"] for o, e in energy.items()}
+        hot = sorted(peaks.items(), key=lambda kv: -kv[1])
+        sub = t.get("subtype", "?")
+        # A float file is NOT clipped by exceeding 0 dBFS — 32-bit float carries
+        # values above 1.0 losslessly, so the waveform is intact and a gain cut
+        # fully recovers it. Calling that "clipping" would be a false alarm that
+        # sends PLN re-recording material that is actually fine. It only becomes
+        # real clipping on export to integer, or at the master bus. In a PCM file
+        # the same number means samples were truncated and the audio IS damaged.
+        isfloat = "FLOAT" in str(sub).upper()
+        print(f"\nper-orbit peak dBFS (source, pre-master; {sub}):")
+        if isfloat:
+            print("  32-bit float: over 0 dBFS is HEADROOM, not damage — a gain cut")
+            print("  recovers it exactly. It clips only on int export or the master.")
+        for orb, pdb in hot:
+            if pdb >= 0.0:
+                flag = "  <-- OVER 0 (recoverable)" if isfloat else "  <-- CLIPPED"
+            elif pdb > -3.0:
+                flag = "  <-- hot"
+            else:
+                flag = ""
+            print(f"  d{orb:<3} {pdb:>7.2f}{flag}")
+        over = [o for o, p in peaks.items() if p >= 0.0]
+        if over:
+            worst = max(peaks.values())
+            print(f"  {len(over)}/{len(peaks)} orbits over 0 dBFS, worst +{worst:.1f} dB")
+            print(f"  -> a uniform -{worst + 6:.0f} dB on the record bus would put the")
+            print(f"     hottest orbit at the -6 dBFS source target.")
+        silent = [o for o, p in peaks.items() if p <= SILENT_DBFS]
+        if silent:
+            print(f"  silent all take: {['d%d' % o for o in silent]}")
+
+    # --- track timeline ------------------------------------------------------
+    print("")
+    if rec["track"]:
+        print("track timeline (from the editor — ground truth):")
+        for r in rec["track"]:
+            print(f"  {hhmmss(r['t'])}  {r.get('path')}")
+    else:
+        print("track timeline: NONE for this take.")
+        print("  Nothing published the loaded track while it recorded (gig-log gained")
+        print("  k:\"track\" on 2026-07-29). Boundaries inside this take can only be")
+        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})
+
+    print(f"\nhands-idle runs >= {IDLE_RUN_S:.0f}s: {len(cands)}")
+    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']})")
+
+    if edl_out:
+        edl = {
+            "take": take_n, "session": str(session),
+            "start_epoch": t0, "end_epoch": t1, "dur_s": dur, "sr": sr,
+            "bucket_s": bucket_s,
+            "orbits": sorted(t["orbits"]),
+            "track_timeline": [{"t": r["t"], "offset_s": r["t"] - t0,
+                                "path": r.get("path")} for r in rec["track"]],
+            "trim_candidates": cands,
+            "peak_dbfs": {f"d{o}": e["peak_all_db"] for o, e in energy.items()},
+            "hands": {"cc": sum(cc_n), "notes": sum(notes),
+                      "idle_buckets": idle, "buckets": nb},
+            "generated_by": "tools/take-lens.py",
+        }
+        edl_out.write_text(json.dumps(edl, indent=2) + "\n")
+        print(f"\nEDL -> {edl_out}")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    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")
+    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,
+                      help="seconds per bucket (default 15)")
+    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")
+    a = ap.parse_args(argv)
+    if a.cmd == "takes":
+        return cmd_takes(a.session)
+    return cmd_lens(a.take, a.bucket, a.session, not a.no_audio, a.edl)
+
+
+if __name__ == "__main__":
+    sys.exit(main())