hooks.py

"""Does this 30 seconds contain the hook? — the intra-track half of the pick.

The market ranking asks "what is charting that sounds like this". That is a
question about the outside world, and on its own it cannot see the moment on a
record that makes people stop scrolling. This module asks the other question,
about the record alone: **which part of THIS track is the hook**.

Three signals, all computed from work the app already pays for:

  repeats  — the section family. Every 30-second window is already embedded
             as three 10-second CLAP tiles on a 5-second grid, and those tiles
             are cached. Comparing the cached tiles against each other costs a
             single 39x39 dot product (4 milliseconds on a 3-minute track) and
             says how many other places in the record sound like this one.
             A chorus recurs; an intro does not. Counted, not summed, so the
             FIRST chorus is not penalised for arriving early — the same
             choice audio-brief's `cutpoints.py` makes.

  voice    — how much energy sits in the centre of the stereo image between
             300 Hz and 4 kHz, which is where a lead vocal usually is. This is
             a WEAK signal and is weighted as one; see "What was measured".

  lift     — how loud and how busy this passage is against the track's own
             average. A hook is normally at or above the record's own level,
             an intro or a breakdown below it.

Every signal is scored against the track's own average or its own strongest
window, never against other records. "Strong" here means strong for this
track — the reader is choosing between parts of one record, not comparing
records, and a cross-track scale would be a calibration nobody has done.

What was measured (2026-08-21, `Joshua Baraka - YoYo MAIN.mp3`, a full master,
against a Demucs vocal stem taken as ground truth):

  * Demucs itself — 95 seconds and 1.7 GB of peak memory for a 3-minute track
    on a fast laptop. Not affordable on a free CPU Space that is already
    holding a 1.9 GB CLAP checkpoint.
  * Open-Unmix (umxl), the light separator — 11 seconds and 1.2 GB on the same
    laptop, so roughly a minute on two shared vCPUs. Also rejected.
  * librosa's `nn_filter` foreground (its own vocal-separation recipe) — 33
    seconds, 770 MB, and no better than the cheap options below.
  * Harmonic energy in the vocal band (HPSS) — separated vocal from
    instrumental with an AUC of 0.79 against the stem.
  * Centre-channel energy in the vocal band — AUC 0.86, and it costs 0.15
    seconds. Best of the cheap options, so it is the one here.
  * BUT: plain loudness scored 0.87 on the same test. Once you hold loudness
    constant, no cheap method beat "how loud is it" at finding the voice.

So `voice` is honestly a "voice or lead instrument out front" measure, not
vocal detection, it carries the smallest weight of the three, and the UI says
so. The signal that does the real work here is `repeats`.

Weights are hand-set on a first pass. They order windows; they are not a
measurement of anything, and nothing has been fitted to labelled data.
"""

from __future__ import annotations

import numpy as np

# Hand-set first pass. `repeats` leads because it is the one signal that was
# measured to work; `voice` trails because it was measured to be weak.
W_REPEAT, W_VOICE, W_LIFT = 0.45, 0.25, 0.30

# Tiles closer together than this overlap or sit in the same phrase, so a
# match between them says nothing about the section recurring. Mirrors the
# `width=` guard on librosa's recurrence matrix.
TILE_MIN_GAP_S = 15.0
# A tile pair counts as "the same part of the record" above this percentile of
# the track's own off-diagonal similarities. Relative on purpose: CLAP
# similarities inside one track sit in a narrow, track-dependent band.
TILE_PCT = 80.0

# Wording cut-offs, as a share of the track's OWN strongest window.
STRONG_OF_BEST, MEDIUM_OF_BEST = 0.85, 0.65


def relative(x: np.ndarray, cap: float = 2.0) -> np.ndarray:
    """Scale a curve by the track's own mean, so 0.5 means average and 1.0
    means twice the average.

    Min-max normalising would be wrong for the same reason it is wrong for
    window affinity: on a track that is uniformly loud it stretches noise
    across the whole 0–1 range and invents a winner.
    """
    x = np.asarray(x, dtype=np.float64)
    m = float(np.mean(x))
    if m <= 1e-12:
        return np.zeros_like(x)
    return np.clip(x / m, 0.0, cap) / cap


def tile_repetition(tiles: np.ndarray, tile_starts: np.ndarray,
                    min_gap_s: float = TILE_MIN_GAP_S,
                    pct: float = TILE_PCT) -> np.ndarray:
    """Per-tile count of how many distant tiles sound like this one, scaled so
    the most-repeated tile is 1.0.

    `tiles` must be L2-normed, so the Gram matrix is cosine similarity.
    Returns zeros when nothing recurs, which leaves the other two signals to
    decide rather than inventing a ranking.
    """
    tiles = np.asarray(tiles, dtype=np.float64)
    starts = np.asarray(tile_starts, dtype=np.float64)
    n = len(starts)
    if n < 3 or tiles.ndim != 2 or tiles.shape[0] != n:
        return np.zeros(max(n, 0))
    sim = tiles @ tiles.T
    far = np.abs(starts[:, None] - starts[None, :]) >= min_gap_s
    if not far.any():
        return np.zeros(n)
    thresh = float(np.percentile(sim[far], pct))
    count = ((sim > thresh) & far).sum(axis=1).astype(np.float64)
    top = float(count.max())
    return count / top if top > 0 else count


def window_hooks(starts: list[float], curves: dict, tiles: np.ndarray | None,
                 tile_starts: np.ndarray | None,
                 window_s: float = 30.0) -> dict:
    """-> {hook, repeats, voice, lift, label, method} — one entry per window.

    Never raises. A missing signal contributes 0.5 (neutral) rather than 0, so
    a track whose beat or stereo read failed is not pushed to the bottom of
    its own ranking.
    """
    n = len(starts)
    zero = {"hook": [0.5] * n, "repeats": [0.5] * n, "voice": [0.5] * n,
            "lift": [0.5] * n, "label": ["unread"] * n, "method": "none"}
    if n == 0:
        return {**zero, "hook": [], "repeats": [], "voice": [], "lift": [],
                "label": []}
    try:
        return _window_hooks(starts, curves, tiles, tile_starts, window_s)
    except Exception as exc:  # noqa: BLE001 — degradation is the feature
        return {**zero, "method": f"failed: {type(exc).__name__}: {exc}"}


def _window_hooks(starts, curves, tiles, tile_starts, window_s):
    t = np.asarray(curves.get("times") or [], dtype=np.float64)
    have_curves = t.size > 0
    if have_curves:
        voice_c = relative(curves["voice"])
        lift_c = 0.5 * relative(curves["rms"]) + 0.5 * relative(curves["onset"])

    rep_t = None
    if tiles is not None and tile_starts is not None and len(tile_starts) >= 3:
        rep_t = tile_repetition(tiles, tile_starts)
        tile_starts = np.asarray(tile_starts, dtype=np.float64)

    repeats, voice, lift = [], [], []
    for s in starts:
        if have_curves:
            sel = (t >= s) & (t < s + window_s)
            voice.append(float(voice_c[sel].mean()) if sel.any() else 0.5)
            lift.append(float(lift_c[sel].mean()) if sel.any() else 0.5)
        else:
            voice.append(0.5)
            lift.append(0.5)
        if rep_t is not None:
            tsel = (tile_starts >= s) & (tile_starts < s + window_s)
            repeats.append(float(rep_t[tsel].mean()) if tsel.any() else 0.5)
        else:
            repeats.append(0.5)

    hook = [W_REPEAT * r + W_VOICE * v + W_LIFT * l
            for r, v, l in zip(repeats, voice, lift)]
    best = max(hook) if hook else 0.0
    method = "+".join(p for p in (
        "clap-tiles" if rep_t is not None else "",
        "centre-channel+level" if have_curves else "") if p) or "none"
    return {
        "hook": [round(h, 4) for h in hook],
        "repeats": [round(r, 3) for r in repeats],
        "voice": [round(v, 3) for v in voice],
        "lift": [round(l, 3) for l in lift],
        "label": [label(h, best) for h in hook],
        "method": method,
    }


def label(hook: float, best: float) -> str:
    """Plain words for one window's hook score, read against the strongest
    window in the same track. Never a cross-track claim."""
    if best <= 1e-9:
        return "unread"
    share = hook / best
    if share >= STRONG_OF_BEST:
        return "strong"
    if share >= MEDIUM_OF_BEST:
        return "medium"
    return "quiet"