guts/pipeline/gen_audio.py

#!/usr/bin/env python3
"""GUTS audio pack — 100% procedural synthesis in numpy, on-device, $0 (Lane D).

Ported from PROCITY's gen_audio.py (round-11 audio round) — same DSP toolkit, same reasons:
  * $0, no model download, runs in the mflux venv (numpy only).
  * Deterministic + seeded — same name, same sound, forever.
  * Perfect SEAMLESS LOOPS (the tail is crossfaded over the head) — neural gens can't loop.
  * Parody-safe by construction: every sample is an original oscillator/noise render.
  * Precise budget control (<=10 MB shipped, TECH.md).

Pipeline:  synth (numpy float32) -> WAV (stdlib wave, int16) -> ffmpeg -> OGG/Opus + M4A/AAC.
Loudness:  beds -> ffmpeg loudnorm (EBU R128, -16 LUFS); sfx -> peak-normalized in-synth
           (loudnorm is unreliable under ~3 s).
Evidence:  --evidence writes a spectrogram sheet + a NUMERIC loop-seam check to
           docs/shots/laneD/ (PROCESS.md: audio evidence = spectrogram + loop-seam check).

    <mflux-py> pipeline/gen_audio.py --dry-run     # list the pack + budget, no render
    <mflux-py> pipeline/gen_audio.py               # render -> web/assets/audio/
    <mflux-py> pipeline/gen_audio.py --only sfx    # one category
    <mflux-py> pipeline/gen_audio.py --wav-only    # keep .genaudio/*.wav, skip ffmpeg (debug)

<mflux-py> = ~/Documents/MODELBEAST/venvs/mflux/bin/python
House law: the game runs silent-and-happy with zero assets — none of this blocks anyone.
"""
import json, math, os, shutil, subprocess, sys, wave, zlib
import numpy as np

SR = 44100
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RAW = os.path.join(ROOT, "pipeline", ".genaudio")            # WAV scratch (git-ignored)
OUT = os.path.join(ROOT, "web", "assets", "audio")           # shipped OGG/M4A
SHOTS = os.path.join(ROOT, "docs", "shots", "laneD")
BATCH = os.path.join(ROOT, "pipeline", "batch_audio.json")
for d in (RAW, OUT, SHOTS):
    os.makedirs(d, exist_ok=True)

# m3ultra's ffmpeg is homebrew and NOT on the non-interactive ssh PATH — find it properly.
FF = (os.environ.get("GUTS_FFMPEG") or shutil.which("ffmpeg")
      or ("/opt/homebrew/bin/ffmpeg" if os.path.exists("/opt/homebrew/bin/ffmpeg") else "ffmpeg"))
XF = 0.75                                                    # loop crossfade seconds


# ─────────────────────────────────────────────────────────────────────────────
# DSP toolkit (pure numpy) — ported from PROCITY pipeline/gen_audio.py
# ─────────────────────────────────────────────────────────────────────────────
def seed_of(name):
    return zlib.crc32(name.encode()) & 0xFFFFFFFF


def rng(name):
    return np.random.default_rng(seed_of(name))


def tarr(dur):
    return np.arange(int(dur * SR)) / SR


def _phase(f, dur):
    n = int(dur * SR)
    f = np.full(n, f, dtype=np.float64) if np.isscalar(f) else np.resize(f, n)
    return np.cumsum(2 * np.pi * f / SR)


def osc(f, dur, kind="sine"):
    ph = _phase(f, dur)
    if kind == "sine":
        return np.sin(ph)
    if kind == "tri":
        return 2 / np.pi * np.arcsin(np.sin(ph))
    if kind == "saw":
        return 2 * ((ph / (2 * np.pi)) % 1.0) - 1
    if kind == "square":
        return np.sign(np.sin(ph))
    raise ValueError(kind)


def noise(dur, r=None, kind="white"):
    n = int(dur * SR)
    r = r if r is not None else np.random.default_rng(0)
    x = r.standard_normal(n)
    if kind == "pink":                      # ~-3 dB/oct, vectorised one-pole cascade
        out = np.zeros(n)
        for a, g in ((0.99765, 0.0990460), (0.96300, 0.2965164), (0.57000, 1.0526913)):
            out += lp_pole(x * g, a)
        return (out + x * 0.1848) / 4
    return x


def lp_pole(x, a):
    """y[n] = a*y[n-1] + x[n] — via lfilter-free cumulative trick (scipy is not installed)."""
    y = np.empty_like(x)
    acc = 0.0
    for i in range(len(x)):                 # honest loop; pack is small, runs in ~1 s
        acc = a * acc + x[i]
        y[i] = acc
    return y


def lp1(x, cutoff):
    a = math.exp(-2 * math.pi * cutoff / SR)
    return lp_pole(x * (1 - a), a)


def hp1(x, cutoff):
    return x - lp1(x, cutoff)


def bandpass(x, lo, hi):
    return hp1(lp1(x, hi), lo)


def adsr(dur, a=0.01, d=0.1, s=0.7, r=0.1):
    n = int(dur * SR)
    ai, di = int(a * SR), int(d * SR)
    ri = min(int(r * SR), max(n - ai - di, 0))
    si = max(n - ai - di - ri, 0)
    return np.concatenate([
        np.linspace(0, 1, ai), np.linspace(1, s, di),
        np.full(si, s), np.linspace(s, 0, ri)])[:n]


def fade(x, fin=0.01, fout=0.01):
    n = len(x); ai = int(fin * SR); ao = int(fout * SR)
    if ai: x[:ai] *= np.linspace(0, 1, ai)
    if ao: x[-ao:] *= np.linspace(1, 0, ao)
    return x


def peaknorm(x, target_db=-3.0):
    p = np.max(np.abs(x)) or 1.0
    return x * (10 ** (target_db / 20) / p)


def softclip(x, drive=1.0):
    return np.tanh(x * drive)


def seamless(x, xf=XF):
    """Loop of length len(x)-xf*SR that plays end->start seamlessly, by crossfading the tail
    (xf s) over the head. Render xf LONGER than the loop you want."""
    k = int(xf * SR)
    if k <= 0 or len(x) <= k:
        return x
    body, tail = x[:-k].copy(), x[-k:]
    w = np.linspace(0, 1, k)
    if body.ndim == 2:
        w = w[:, None]
    body[:k] = body[:k] * w + tail * (1 - w)
    return body


def stereo(l, r=None):
    r = l if r is None else r
    n = max(len(l), len(r))
    out = np.zeros((n, 2))
    out[:len(l), 0] = l; out[:len(r), 1] = r
    return out


def pan(x, p):
    p = (p + 1) / 2
    return stereo(x * math.cos(p * math.pi / 2), x * math.sin(p * math.pi / 2))


def mix(*layers):
    n = max(len(a) for a in layers)
    ch = 2 if any(a.ndim == 2 for a in layers) else 1
    out = np.zeros((n, 2) if ch == 2 else n)
    for a in layers:
        if ch == 2 and a.ndim == 1:
            a = stereo(a)
        out[:len(a)] += a
    return out


def schroeder_reverb(x, mix_amt=0.2, decay=0.5):
    """4 combs + 2 allpasses. The gut is a wet cave; everything gets some of this."""
    mono = x.mean(axis=1) if x.ndim == 2 else x
    out = np.zeros(len(mono))
    for ms, g in ((29.7, 0.805), (37.1, 0.827), (41.1, 0.783), (43.7, 0.764)):
        d = int(ms * SR / 1000)
        buf = np.zeros(len(mono) + d)
        buf[:len(mono)] = mono
        for i in range(d, len(buf)):
            buf[i] += buf[i - d] * g * decay
        out += buf[:len(mono)]
    out /= 4
    for ms, g in ((5.0, 0.7), (1.7, 0.7)):
        d = int(ms * SR / 1000)
        y = np.zeros(len(out) + d)
        y[:len(out)] = out
        for i in range(d, len(y)):
            y[i] += y[i - d] * -g
        out = y[:len(out)]
    wet = out / (np.max(np.abs(out)) or 1)
    if x.ndim == 2:
        return x * (1 - mix_amt) + stereo(wet) * mix_amt
    return x * (1 - mix_amt) + wet * mix_amt


# ─────────────────────────────────────────────────────────────────────────────
# the GUTS pack (round 1 = proof: 1 bed + 4 sfx; full pack is round 2)
# ─────────────────────────────────────────────────────────────────────────────
def bed_esophagus(name):
    """Wet cavernous drone + muffled resting heartbeat + peristalsis whooshes.
    Loop = 24 s exactly, so the 1.2 s heartbeat (50 bpm) and 4 s whoosh both divide it and
    survive the crossfade without a stutter."""
    r = rng(name)
    LOOP = 24.0
    dur = LOOP + XF

    # low wet drone — two detuned saws, heavily lowpassed, slow swell
    d1 = osc(55.0, dur, "saw") + osc(55.0 * 1.006, dur, "saw") * 0.8
    drone = lp1(d1 * 0.25, 190) * (0.75 + 0.25 * np.sin(2 * np.pi * tarr(dur) / 12.0))

    # cavern air — pink noise in the 200–900 band, breathing at 0.25 Hz
    air = bandpass(noise(dur, r, "pink"), 200, 900) * 0.30
    air *= 0.6 + 0.4 * np.sin(2 * np.pi * tarr(dur) / 8.0)

    # muffled heartbeat (lub-dub), 50 bpm
    hb = np.zeros(int(dur * SR))
    period = 1.2
    for i in range(int(dur / period) + 1):
        for off, amp in ((0.0, 1.0), (0.16, 0.62)):          # lub, dub
            st = int((i * period + off) * SR)
            if st >= len(hb):
                continue
            th = 0.22
            body = osc(np.linspace(58, 34, int(th * SR)), th, "sine") * adsr(th, 0.004, 0.05, 0.25, 0.14)
            seg = body * amp
            n = min(len(seg), len(hb) - st)
            hb[st:st + n] += seg[:n]
    hb = lp1(hb, 120) * 0.55

    # peristalsis whoosh every 4 s — filtered noise sweeping past
    wh = np.zeros(int(dur * SR))
    for i in range(int(dur / 4.0) + 1):
        st = int(i * 4.0 * SR)
        wd = 1.8
        seg = noise(wd, np.random.default_rng(seed_of(name) + i))
        sweep = np.linspace(300, 1600, int(wd * SR))
        seg = bandpass(seg, 200, 2000) * np.sin(np.pi * np.linspace(0, 1, len(seg))) ** 2
        seg *= 0.18 * (0.7 + 0.3 * math.sin(i))
        n = min(len(seg), len(wh) - st)
        if n > 0:
            wh[st:st + n] += seg[:n]

    body = mix(pan(drone, -0.15), pan(air, 0.2), pan(hb, 0.0), pan(wh, 0.35))
    body = schroeder_reverb(body, mix_amt=0.28, decay=0.55)
    body = softclip(body * 0.8, 1.1)
    return seamless(body, XF)


def sfx_pellet(name):
    """Lysozyme cannon — a short wet bright zap, not a laser pew."""
    dur = 0.20
    r = rng(name)
    body = osc(np.linspace(1500, 380, int(dur * SR)), dur, "saw") * adsr(dur, 0.001, 0.04, 0.25, 0.12)
    wet = bandpass(noise(dur, r) * adsr(dur, 0.001, 0.03, 0.1, 0.15), 800, 5200) * 0.5
    x = lp1(body * 0.6 + wet, 6000)
    return peaknorm(fade(x, 0.001, 0.02), -4.0)


def sfx_hit_squelch(name):
    """Wet impact into tissue: a low slap + a burst of high spatter."""
    dur = 0.34
    r = rng(name)
    thud = osc(np.linspace(180, 55, int(dur * SR)), dur, "sine") * adsr(dur, 0.002, 0.09, 0.18, 0.2)
    spat = bandpass(noise(dur, r), 700, 4500) * adsr(dur, 0.001, 0.05, 0.06, 0.25)
    x = softclip(thud * 0.9 + spat * 0.55, 1.4)
    x = schroeder_reverb(x, mix_amt=0.16, decay=0.4)
    return peaknorm(fade(x, 0.001, 0.03), -3.5)


def sfx_boost(name):
    """ENDO-1 boost — rising filtered whoosh with a pressure thump underneath."""
    dur = 0.85
    r = rng(name)
    air = noise(dur, r)
    env = np.linspace(0, 1, int(dur * SR)) ** 1.6
    air = bandpass(air, 220, 5200) * env * np.sin(np.pi * np.linspace(0, 1, int(dur * SR))) ** 0.7
    sub = osc(np.linspace(38, 120, int(dur * SR)), dur, "sine") * adsr(dur, 0.05, 0.2, 0.6, 0.3) * 0.6
    x = softclip(air * 0.7 + sub, 1.2)
    return peaknorm(fade(x, 0.01, 0.06), -3.5)


def sfx_pickup(name):
    """Pickup chime — soft, green, organic (ART_BIBLE pickups #7dffb0). Two-note bell."""
    dur = 0.5
    x = np.zeros(int(dur * SR))
    for f, amp, delay in ((784.0, 1.0, 0.0), (1174.7, 0.7, 0.07)):
        st = int(delay * SR)
        seg = (osc(f, dur - delay, "sine") + 0.35 * osc(f * 2.01, dur - delay, "sine")) \
            * adsr(dur - delay, 0.004, 0.18, 0.22, 0.28) * amp
        x[st:st + len(seg)] += seg
    x = schroeder_reverb(lp1(x, 7000), mix_amt=0.22, decay=0.45)
    return peaknorm(fade(x, 0.002, 0.05), -4.0)


PACK = {
    "bed-esophagus":    ("bed", bed_esophagus,   dict(loop=True,  gain=0.5, key="esophagus")),
    "sfx-pellet":       ("sfx", sfx_pellet,      dict(loop=False, gain=0.5, key="pellet")),
    "sfx-hit-squelch":  ("sfx", sfx_hit_squelch, dict(loop=False, gain=0.7, key="hit_squelch")),
    "sfx-boost":        ("sfx", sfx_boost,       dict(loop=False, gain=0.6, key="boost")),
    "sfx-pickup":       ("sfx", sfx_pickup,      dict(loop=False, gain=0.5, key="pickup")),
}


# ─────────────────────────────────────────────────────────────────────────────
# io + evidence
# ─────────────────────────────────────────────────────────────────────────────
def write_wav(path, x):
    if x.ndim == 1:
        x = stereo(x)
    peak = np.max(np.abs(x))
    if peak > 0.95:
        x = x * (0.95 / peak)
    pcm = (np.clip(x, -1, 1) * 32767).astype("<i2")
    with wave.open(path, "wb") as w:
        w.setnchannels(2); w.setsampwidth(2); w.setframerate(SR)
        w.writeframes(pcm.tobytes())


def encode(wav, base, category):
    ogg = os.path.join(OUT, base + ".ogg")
    m4a = os.path.join(OUT, base + ".m4a")
    af = ["-af", "loudnorm=I=-16:TP=-1.5:LRA=11"] if category == "bed" else []
    br = "80k" if category == "bed" else "64k"
    subprocess.run([FF, "-y", "-i", wav, *af, "-c:a", "libopus", "-b:a", br, "-ar", "48000", ogg],
                   check=True, capture_output=True)
    subprocess.run([FF, "-y", "-i", wav, *af, "-c:a", "aac", "-b:a", "128k", "-ar", "44100", m4a],
                   check=True, capture_output=True)
    return os.path.getsize(ogg), os.path.getsize(m4a)


def loop_seam(x):
    """THE loop-seam number. Ratio of the wrap-around step (last sample -> first sample) to
    the RMS step inside the signal. ~1 = the seam is indistinguishable from ordinary motion;
    >>1 = an audible click. Reported for every looping bed; vibes are not evidence."""
    m = x.mean(axis=1) if x.ndim == 2 else x
    inner = np.sqrt(np.mean(np.diff(m) ** 2)) or 1e-9
    wrap = abs(m[0] - m[-1])
    return float(wrap / inner)


def spectrogram(x, w=420, h=180):
    """|STFT| -> log-magnitude image column-per-frame. numpy only (no matplotlib on the box)."""
    m = x.mean(axis=1) if x.ndim == 2 else x
    n = 1024
    hop = max(1, (len(m) - n) // w)
    win = np.hanning(n)
    cols = []
    for i in range(w):
        st = i * hop
        seg = m[st:st + n]
        if len(seg) < n:
            seg = np.pad(seg, (0, n - len(seg)))
        S = np.abs(np.fft.rfft(seg * win))
        cols.append(20 * np.log10(S + 1e-6))
    img = np.array(cols).T[:h * 2:2][::-1]              # low freqs at the bottom
    img = np.clip((img + 60) / 70, 0, 1)
    return img


def evidence(rendered):
    """Spectrogram sheet + loop-seam table -> docs/shots/laneD/. Import PIL late: the render
    path itself must not depend on it."""
    from PIL import Image, ImageDraw
    rows = len(rendered)
    W, H, PADL = 420, 180, 150
    sheet = Image.new("RGB", (W + PADL + 20, rows * (H + 14) + 14), (10, 12, 14))
    d = ImageDraw.Draw(sheet)
    for i, (name, x, meta) in enumerate(rendered):
        y = 8 + i * (H + 14)
        img = spectrogram(x, W, H)
        g = (img * 255).astype(np.uint8)
        rgb = np.stack([g, (g * 0.85).astype(np.uint8), (g * 0.55).astype(np.uint8)], axis=-1)
        sheet.paste(Image.fromarray(rgb, "RGB"), (PADL, y))
        d.text((8, y + 6), name, fill=(90, 255, 210))
        d.text((8, y + 22), f"{meta['seconds']:.2f}s", fill=(150, 170, 180))
        if meta.get("loop"):
            d.text((8, y + 38), f"seam {meta['loop_seam']:.2f}", fill=(150, 170, 180))
        d.text((8, y + 54 if meta.get("loop") else y + 38), f"{meta['ogg_kb']}KB ogg", fill=(150, 170, 180))
    p = os.path.join(SHOTS, "audio_spectrograms.png")
    sheet.save(p, "PNG")
    return p


if __name__ == "__main__":
    only = sys.argv[sys.argv.index("--only") + 1] if "--only" in sys.argv else ""
    wav_only = "--wav-only" in sys.argv
    todo = {k: v for k, v in PACK.items() if not only or only in k or only == v[0]}

    if "--dry-run" in sys.argv:
        print(f"{len(todo)} audio assets (procedural numpy synth, $0):")
        for k, (cat, _, meta) in sorted(todo.items()):
            print(f"  [{cat}] {k:18s} loop={str(meta['loop']):5s} key={meta['key']}")
        sys.exit(0)

    print(f"rendering {len(todo)} assets @ {SR}Hz (numpy synth -> {os.path.basename(FF)} opus+aac)")
    log = json.load(open(BATCH)) if os.path.exists(BATCH) else {}
    rendered, tot_ogg, tot_m4a = [], 0, 0
    for k, (cat, gen, meta) in todo.items():
        x = gen(k)
        wav = os.path.join(RAW, k + ".wav")
        write_wav(wav, x)
        secs = len(x) / SR
        seam = loop_seam(x) if meta["loop"] else None
        if wav_only:
            print(f"  [{cat}] {k:18s} {secs:5.2f}s  (wav only)"
                  + (f"  seam={seam:.2f}" if seam is not None else ""))
            continue
        og, m4 = encode(wav, k, cat)
        tot_ogg += og; tot_m4a += m4
        info = dict(category=cat, key=meta["key"], loop=meta["loop"], gain=meta["gain"],
                    seconds=round(secs, 2), ogg_kb=og // 1024, m4a_kb=m4 // 1024)
        if seam is not None:
            info["loop_seam"] = round(seam, 3)
        log[k] = info
        rendered.append((k, x, info))
        print(f"  [{cat}] {k:18s} {secs:5.2f}s  ogg={og//1024:>4}KB  m4a={m4//1024:>4}KB"
              + (f"  seam={seam:.2f}" if seam is not None else ""))
    if wav_only:
        sys.exit(0)
    json.dump(log, open(BATCH, "w"), indent=2, sort_keys=True)
    total = (tot_ogg + tot_m4a) / 1e6
    print(f"\ntotal shipped: ogg {tot_ogg/1e6:.2f}MB + m4a {tot_m4a/1e6:.2f}MB = {total:.2f}MB "
          f"(budget 10MB)")
    if "--evidence" in sys.argv and rendered:
        print(f"evidence -> {evidence(rendered)}")