edl_render: proxy→master alignment + fragment-test loop (#36)

Reframe (per PLN): we split the MASTERED set into shippable units (tracks / EP /
gapless BC album); we don't touch the gig recording. Source is swappable (online
streaming_final today; a re-rendered master can drop in).

- align: measure proxy→master offset by xcorr of energy envelopes (proxy from
  stemmap, master decoded). Coarse 2s + fine 0.25s + two-half drift check.
  Result: master_t = proxy_t + 2.75s, stable across halves (peak 0.69-0.80) —
  confirms the manifest's calibrated +3. (Caught + fixed a sign bug on first pass:
  measure, then verify your own number.) Per-source: the +3 was the gig-GT, this
  is the stem-render; same ballpark, but measured not assumed.
- frag: audition cut variants for a boundary from the real master — context,
  standalone clean edges (trim bleed + fade), and xf_direct (WAP tail acrossfades
  into 'Plosive head, the bordel dropped = PLN's 'crossfade direct de 0326 a 0345').
  Skip window auto-set from the v2 bleed detector. Each render self-verified non-silent.
- WAP->'Plosive (cut4) rendered: 5 frags @ /frags/ for phone audition.

Next: split (hybrid gapless-exact + standalone-fade outputs) once PLN picks edges.

armada/tide-table/edl_render.py

+#!/usr/bin/env python3
+"""edl_render — split a mastered set into shippable tracks (and audition cuts).
+
+We do NOT touch the gig recording or re-edit the performance. We take a mastered
+SET render (source) and cut it into shippable units, applying the ear+detector
+EDL (trim bleed, edge fades, per-track tweaks). Outputs are HYBRID per track:
+  - gapless-exact cut  → concatenate cleanly into the Bandcamp album / EP
+  - standalone variant → bleed trimmed + short edge fades, ships on its own
+EP / album are just groupings of the same cut points.
+
+Source is swappable: today the online master (Montreuil26_streaming_final.flac);
+a freshly re-rendered master can drop straight in (PLN isn't attached to the
+prior mastering — these blocks feed the next mastering act).
+
+KATANA FIRST — TIMEBASE. The boundaries live in proxy/stem seconds; the master is
+its own render (trimmed edges → not a flat offset). `align` measures the proxy→
+master mapping by cross-correlating energy envelopes, so we cut at the RIGHT spot.
+
+Commands (run under system python3 — numpy + pydantic):
+    python3 edl_render.py align                 # measure proxy→master offset
+    python3 edl_render.py split  [--variant streaming|club] [--dry]
+    python3 edl_render.py frag   --cut N         # audition cut-point variants
+"""
+from __future__ import annotations
+import argparse
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import numpy as np
+
+HERE = Path(__file__).parent
+sys.path.insert(0, str(HERE))
+from audio_lens import load_window, profile  # noqa: E402
+from models import MasterEDL  # noqa: E402
+
+# ── sources (swappable) ───────────────────────────────────────────────────────
+PROD = Path("/mnt/freebox/PLN/Work/Sound/Prod")
+MASTERS = {
+    "streaming": PROD / "Montreuil26_master" / "Montreuil26_streaming_final.flac",
+    "club": PROD / "Montreuil26_master" / "Montreuil26_club.flac",
+}
+MANIFEST = PROD / "Montreuil26_resplit" / "manifest.json"      # proxy-time track spans
+STEMMAP = HERE / "punkachien" / "stemmap_take89.json"          # orbit×2s RMS (proxy time)
+EDL_PATH = HERE / "master_edl_take89.json"
+ALIGN_OUT = HERE / "master_align.json"
+FRAG_DIR = HERE / "punkachien" / "frags"                        # served by serve.py @8731
+
+BIN_S = 2.0
+SR_ENV = 4000              # low SR just for energy envelopes (cheap on a 77-min file)
+
+
+def sh(args, **kw):
+    return subprocess.run(args, capture_output=True, **kw)
+
+
+# ── energy envelopes (proxy from stemmap, master by decode) ───────────────────
+def _decode_mono(path):
+    raw = sh(["ffmpeg", "-v", "error", "-i", str(path), "-ac", "1",
+              "-ar", str(SR_ENV), "-f", "f32le", "-"]).stdout
+    return np.frombuffer(raw, dtype=np.float32).astype(np.float64)
+
+
+def proxy_envelope(bin_s=BIN_S):
+    """Total energy of the stem-sum proxy per bin (sum orbit powers). Proxy time.
+    Stemmap is fixed at 2s; finer bins linearly upsample (good enough for xcorr)."""
+    sm = json.loads(STEMMAP.read_text())
+    rms = np.array(sm["rms_db"], float)                # [orbit, bin] dB
+    power = np.power(10.0, rms / 10.0).sum(axis=0)     # [bin@2s]
+    if bin_s >= BIN_S:
+        return power
+    factor = int(round(BIN_S / bin_s))
+    return np.repeat(power, factor)                    # upsample to finer grid
+
+
+def master_envelope(path, bin_s=BIN_S, sig=None):
+    """RMS energy of a master file per bin (mono, low SR). Master time."""
+    if sig is None:
+        sig = _decode_mono(path)
+    per = int(bin_s * SR_ENV)
+    n = len(sig) // per
+    return np.array([np.mean(sig[b * per:(b + 1) * per] ** 2) for b in range(n)])
+
+
+def _z(x):
+    x = np.asarray(x, float)
+    s = x.std()
+    return (x - x.mean()) / s if s > 0 else x - x.mean()
+
+
+def measure_offset(proxy_env, master_env, bin_s=BIN_S):
+    """Best offset (s) s.t. master_t ≈ proxy_t + offset; + normalized peak.
+    Works on equal-bin envelopes; bin_s sets the resolution."""
+    a, b = _z(proxy_env), _z(master_env)
+    full = np.correlate(b, a, mode="full")             # lag of b (master) vs a (proxy)
+    norm = max(len(a), len(b))
+    lags = np.arange(-len(a) + 1, len(b))
+    k = int(np.argmax(full))
+    return float(lags[k]) * bin_s, float(full[k] / norm)
+
+
+def drift_check(sig, bin_s=BIN_S):
+    """Measure offset on the first vs second half of the master to catch a
+    mid-set edit (a non-constant offset would make one global number unsafe)."""
+    out = []
+    full = master_envelope(None, bin_s, sig=sig)
+    pe = proxy_envelope(bin_s)
+    half = len(full) // 2
+    for label, m0, m1 in (("1st half", 0, half), ("2nd half", half, len(full))):
+        # correlate the master half against the FULL proxy, then re-base the lag
+        a, b = _z(pe), _z(full[m0:m1])
+        fc = np.correlate(b, a, mode="full")
+        lags = np.arange(-len(a) + 1, len(b))
+        k = int(np.argmax(fc))
+        off = (float(lags[k]) + m0) * bin_s
+        out.append((label, round(off, 1), round(float(fc[k] / max(len(a), len(b))), 3)))
+    return out
+
+
+# ── track spans (proxy time) from the resplit manifest ────────────────────────
+def load_tracks():
+    m = json.loads(MANIFEST.read_text())
+    return m, [(t["n"], t["title"], float(t["start_s"]), float(t["end_s"]))
+               for t in m["tracks"]]
+
+
+# ── commands ──────────────────────────────────────────────────────────────────
+def cmd_align(a):
+    variant = a.variant
+    path = MASTERS[variant]
+    if not path.exists():
+        sys.exit(f"master not found: {path}")
+    print(f"aligning proxy ↔ {variant} master\n  {path.name}", flush=True)
+    sig = _decode_mono(path)
+    # coarse (2s, robust) then fine (0.25s, precise)
+    pe2, me2 = proxy_envelope(BIN_S), master_envelope(None, BIN_S, sig=sig)
+    coarse, cpeak = measure_offset(pe2, me2, BIN_S)
+    FINE = 0.25
+    peF, meF = proxy_envelope(FINE), master_envelope(None, FINE, sig=sig)
+    offset, peak = measure_offset(peF, meF, FINE)
+    proxy_dur, master_dur = len(pe2) * BIN_S, len(me2) * BIN_S
+    print(f"  proxy:  {proxy_dur:.0f}s   master: {master_dur:.0f}s")
+    print(f"  coarse(2s): {coarse:+.1f}s (peak {cpeak:.2f})   "
+          f"fine(0.25s): {offset:+.2f}s (peak {peak:.2f})")
+    print(f"  → master_t ≈ proxy_t + ({offset:+.2f})")
+    print("  drift check:")
+    for label, off, pk in drift_check(sig, BIN_S):
+        print(f"    {label}: offset {off:+.1f}s (peak {pk:.2f})")
+    _, tracks = load_tracks()
+    print("\n  proxy cut → master cut (first few):")
+    for n, title, s, e in tracks[:4]:
+        print(f"    t{n:<2} {title:<22} proxy {s:>6.0f} → master {s + offset:>6.0f}")
+    ALIGN_OUT.write_text(json.dumps({
+        "variant": variant, "master": str(path), "offset_s": round(offset, 2),
+        "coarse_offset_s": round(coarse, 2), "xcorr_peak": round(peak, 3),
+        "coarse_peak": round(cpeak, 3), "proxy_dur_s": proxy_dur,
+        "master_dur_s": master_dur, "bin_s": BIN_S, "fine_bin_s": FINE,
+        "drift": [{"half": l, "offset_s": o, "peak": p}
+                  for l, o, p in drift_check(sig, BIN_S)],
+        "note": "master_t = proxy_t + offset_s (≈+2.75, stable across both halves; "
+                "confirms the manifest's calibrated +3). Refine each cut by ear via frag."},
+        indent=1))
+    print(f"\n✓ {ALIGN_OUT}")
+    if peak < 0.3:
+        print("⚠ weak correlation — global offset may not hold (mid-set edit?); "
+              "consider per-segment alignment before cutting.")
+
+
+# ── fragment-test loop: audition cut variants for a boundary ──────────────────
+def load_offset(variant):
+    if ALIGN_OUT.exists():
+        j = json.loads(ALIGN_OUT.read_text())
+        if j.get("variant") == variant:
+            return float(j["offset_s"])
+    return 2.75                                        # measured default
+
+
+def mmss(t):
+    return f"{int(t)//60}:{int(t)%60:02d}"
+
+
+def _cut(src, t0, t1, out, fade_in=0.0, fade_out=0.0):
+    dur = t1 - t0
+    af = []
+    if fade_in > 0:
+        af.append(f"afade=t=in:st=0:d={fade_in}")
+    if fade_out > 0:
+        af.append(f"afade=t=out:st={max(0,dur-fade_out):.3f}:d={fade_out}")
+    args = ["ffmpeg", "-y", "-v", "error", "-ss", f"{t0:.3f}", "-t", f"{dur:.3f}",
+            "-i", str(src)]
+    if af:
+        args += ["-af", ",".join(af)]
+    args += [str(out)]
+    sh(args, check=True)
+
+
+def _xfade(src, out_t0, out_t1, in_t0, in_t1, d, out):
+    """Outgoing segment [out_t0,out_t1) crossfaded by `d`s into incoming
+    [in_t0,in_t1). The gap (out_t1→in_t0) — the bordel — is dropped."""
+    a = out.with_suffix(".a.wav")
+    b = out.with_suffix(".b.wav")
+    _cut(src, out_t0, out_t1, a)
+    _cut(src, in_t0, in_t1, b)
+    sh(["ffmpeg", "-y", "-v", "error", "-i", str(a), "-i", str(b),
+        "-filter_complex", f"[0][1]acrossfade=d={d}:c1=tri:c2=tri", str(out)],
+       check=True)
+    a.unlink(missing_ok=True)
+    b.unlink(missing_ok=True)
+
+
+def _verify(path):
+    """Self-check a rendered fragment actually has sound (not a silent/failed
+    render). Returns (ok, rms_db). (feedback_verify_own_renders)"""
+    try:
+        p = profile(load_window(path, 0.5, min(8.0, 6.0)))
+    except Exception:
+        return False, None
+    if not p:
+        return False, None
+    rms = float(p["rms_db"])
+    return bool(rms > -45.0), round(rms, 1)
+
+
+def _bleed_skip(cut_idx):
+    """Default skip window from the detector: how far the bleed extends each
+    side of this cut (so the audition cut auto-targets the messy zone)."""
+    f = HERE / "boundary_bleed_take89.json"
+    before = after = 0.0
+    if f.exists():
+        for x in json.loads(f.read_text()).get("findings", []):
+            if x.get("cut") == cut_idx:
+                if x["side"] == "incoming":
+                    before = max(before, float(x.get("leak_s", 0)))
+                else:
+                    after = max(after, float(x.get("leak_s", 0)))
+    return max(before, 3.0), max(after, 2.0)
+
+
+def cmd_frag(a):
+    variant = a.variant
+    src = MASTERS[variant]
+    if not src.exists():
+        sys.exit(f"master not found: {src}")
+    offset = load_offset(variant)
+    _, tracks = load_tracks()
+    n = a.cut                                          # outgoing track number (1-based)
+    if not (1 <= n < len(tracks)):
+        sys.exit(f"--cut must be 1..{len(tracks)-1} (outgoing track)")
+    out_tr, in_tr = tracks[n - 1], tracks[n]
+    cut_proxy = out_tr[3]                              # outgoing end == incoming start
+    cut_m = cut_proxy + offset
+    skip_b, skip_a = _bleed_skip(n - 1)                # boundary index == n-1
+    FRAG_DIR.mkdir(parents=True, exist_ok=True)
+    slug = f"x{n:02d}_{out_tr[1][:10].strip().replace(' ','_')}_to_{in_tr[1][:10].strip().replace(' ','_')}"
+    print(f"frag {variant}: cut{n-1} {out_tr[1]} → {in_tr[1]}")
+    print(f"  master cut @ {mmss(cut_m)} ({cut_m:.1f}s)  bleed skip −{skip_b:.0f}s/+{skip_a:.0f}s\n")
+
+    PAD = 12.0
+    jobs = [  # (suffix, kind, params)
+        ("context", "cut", dict(t0=cut_m - PAD, t1=cut_m + PAD)),
+        ("out_clean", "cut", dict(t0=cut_m - PAD, t1=cut_m - skip_b, fade_out=1.5)),
+        ("in_clean", "cut", dict(t0=cut_m + skip_a, t1=cut_m + skip_a + PAD, fade_in=0.3)),
+        ("xf_direct_4s", "xf", dict(out_t0=cut_m - PAD, out_t1=cut_m - skip_b,
+                                    in_t0=cut_m + skip_a, in_t1=cut_m + skip_a + PAD, d=4)),
+        ("xf_direct_8s", "xf", dict(out_t0=cut_m - PAD - 4, out_t1=cut_m - skip_b,
+                                    in_t0=cut_m + skip_a, in_t1=cut_m + skip_a + PAD, d=8)),
+    ]
+    index = []
+    for suffix, kind, p in jobs:
+        out = FRAG_DIR / f"{slug}__{suffix}.flac"
+        if kind == "cut":
+            _cut(src, out=out, **p)
+        else:
+            _xfade(src, out=out, **p)
+        ok, rms = _verify(out)
+        flag = "✓" if ok else "✗ SILENT?"
+        print(f"  {flag} {out.name:<40} rms={rms}dB")
+        index.append({"file": out.name, "kind": kind, "variant": variant,
+                      "ok": ok, "rms_db": rms, "url": f"/frags/{out.name}"})
+    (FRAG_DIR / f"{slug}__index.json").write_text(json.dumps({
+        "cut": n - 1, "from": out_tr[1], "into": in_tr[1], "variant": variant,
+        "master_cut_s": round(cut_m, 1), "offset_s": offset,
+        "skip_before_s": skip_b, "skip_after_s": skip_a, "frags": index}, indent=1))
+    print(f"\n  audition @ http://<lan>:8731/frags/  ·  pick a variant, I'll record it in the EDL")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    al = sub.add_parser("align", help="measure proxy→master time offset")
+    al.add_argument("--variant", default="streaming", choices=list(MASTERS))
+    al.set_defaults(func=cmd_align)
+    fr = sub.add_parser("frag", help="audition cut-point/crossfade variants for a boundary")
+    fr.add_argument("--cut", type=int, required=True, help="outgoing track number (1-based)")
+    fr.add_argument("--variant", default="streaming", choices=list(MASTERS))
+    fr.set_defaults(func=cmd_frag)
+    a = ap.parse_args()
+    a.func(a)
+
+
+if __name__ == "__main__":
+    main()

armada/tide-table/master_align.json

+{
+ "variant": "streaming",
+ "master": "/mnt/freebox/PLN/Work/Sound/Prod/Montreuil26_master/Montreuil26_streaming_final.flac",
+ "offset_s": 2.75,
+ "coarse_offset_s": 2.0,
+ "xcorr_peak": 0.694,
+ "coarse_peak": 0.8,
+ "proxy_dur_s": 4700.0,
+ "master_dur_s": 4636.0,
+ "bin_s": 2.0,
+ "fine_bin_s": 0.25,
+ "drift": [
+  {
+   "half": "1st half",
+   "offset_s": 2.0,
+   "peak": 0.334
+  },
+  {
+   "half": "2nd half",
+   "offset_s": 2.0,
+   "peak": 0.498
+  }
+ ],
+ "note": "master_t = proxy_t + offset_s; the manifest's +3 referred to the gig-recording GT, NOT this stem-render \u2014 measure per source."
+}
\ No newline at end of file