"""pv-at — acceptance tests that drive the real rig and assert on real audio.

    python3 tools/at --list
    python3 tools/at                     # run every case
    python3 tools/at --case sub_is_low   # one case
    python3 tools/at --json              # machine-readable, for gig-up.sh

Exit codes: 0 all passed, 1 a case failed, 2 the rig is not running (which is
NOT a failure — it is an unmet precondition, and conflating the two is how a
green suite comes to mean nothing).

Why the cases look like this
----------------------------
Each case asserts on the lens that can actually SEE the property:

    "is it playing"      -> peak, and the ALL-vs-SOME rule below
    "is the bass low"    -> spectral centroid; never the sample's name
    "does the pad hold"  -> DUTY over the time axis, never an aggregate
    "does the riser rise"-> the slope of the timeline
    "does the filter cut"-> centroid shift, because rms is blind to a filter

The ALL-vs-SOME rule, learned the hard way: if EVERY declared orbit is silent,
the track did not compile — that is a Haskell problem, not an audio one. If
SOME are silent, it compiled and you have a real audio fault. Reporting them the
same way sent a whole evening chasing SuperDirt for a missing pair of parens.
"""

from __future__ import annotations

import argparse
import json
import math
import statistics
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from at.harness import (  # noqa: E402
    TOOLS, Shape, _load, hush, measure, play, rig_state,
)

lcxl = _load("lcxl_init", TOOLS / "lcxl-init.py")

TRACK = Path(__file__).resolve().parent / "fixtures" / "claude.tidal"
ORBITS = [1, 2, 3, 4, 5, 6, 7, 8]

# CCs this suite is allowed to move. 77-84 are MIDI-learned to Ardour track
# gains (CC77 is global gain: down = total silence), so they are not merely
# avoided here — they are excluded by construction.
SAFE_CC = {41, 44, 49, 50, 51, 73, 74, 75}
FORBIDDEN_CC = set(range(77, 85))


@dataclass
class Result:
    name: str
    passed: bool
    detail: str
    measured: str = ""


Case = Callable[[dict[int, Shape]], Result]
CASES: dict[str, tuple[str, Case]] = {}


def case(name: str, doc: str):
    def deco(fn: Case):
        CASES[name] = (doc, fn)
        return fn
    return deco


def ok(name: str, detail: str, measured: str = "") -> Result:
    return Result(name, True, detail, measured)


def no(name: str, detail: str, measured: str = "") -> Result:
    return Result(name, False, detail, measured)


# --------------------------------------------------------------------------
# Cases that read the baseline capture
# --------------------------------------------------------------------------

@case("track_compiles", "every declared orbit makes sound (ALL silent = a compile error)")
def _track_compiles(base: dict[int, Shape]) -> Result:
    silent = [dn for dn, sh in base.items() if sh.silent]
    live = [dn for dn, sh in base.items() if not sh.silent]
    if not silent:
        return ok("track_compiles", f"{len(live)}/{len(base)} orbits sounding")
    if not live:
        return no("track_compiles",
                  "ALL orbits silent — the block did not COMPILE. This is a "
                  "Haskell error, not an audio fault. Check the Pulsar console.",
                  f"silent: {silent}")
    return no("track_compiles",
              f"{len(silent)} orbit(s) silent while others play — a real audio "
              f"fault (fader, cut group, sample, or bus)",
              f"silent: {silent}")


@case("sub_is_low", "d3 is genuinely sub-register (>60% of energy below 300 Hz)")
def _sub_is_low(base: dict[int, Shape]) -> Result:
    sh = base.get(3)
    if sh is None or sh.silent:
        return no("sub_is_low", "d3 made no sound")
    measured = (f"{sh.low_ratio*100:.1f}% of energy below 300 Hz "
                f"(centroid {sh.centroid_hz:.0f} Hz, which the noise floor "
                f"inflates — see band_ratio)")
    if sh.low_ratio > 0.6:
        return ok("sub_is_low", "d3 really is sub-register", measured)
    return no("sub_is_low",
              "d3 does not carry its energy low — validate by analysis, never "
              "infer register from a sample name (meth_bass is a wobble, not a "
              "sub)", measured)


@case("stab_is_bright", "d4 is genuinely bright (>35% of energy above 1 kHz)")
def _stab_is_bright(base: dict[int, Shape]) -> Result:
    sh = base.get(4)
    if sh is None or sh.silent:
        return no("stab_is_bright", "d4 made no sound")
    sub = base.get(3)
    if sub is None or sub.silent:
        return no("stab_is_bright", "d3 made no sound, nothing to compare against")
    measured = (f"d4 {sh.high_ratio*100:.1f}% vs d3 {sub.high_ratio*100:.1f}% "
                f"of energy above 1 kHz")
    # COMPARATIVE, not an absolute threshold. The first version demanded
    # ">35% above 1 kHz" and failed d4 at 32.5% — a number I had picked out of
    # the air, and the honest response to a threshold that close is to fix the
    # question, not to nudge the threshold until it passes. "The stab is
    # brighter than the sub" is a real musical invariant; "32.5% is too low" is
    # not a claim I can defend.
    if sh.high_ratio > sub.high_ratio * 4 and sh.high_ratio > 0.15:
        return ok("stab_is_bright", "d4 sits well above d3 in the spectrum", measured)
    return no("stab_is_bright", "d4 is not clearly brighter than the sub", measured)


@case("pad_sustains", "d5 holds instead of attacking and dying (duty > 0.6)")
def _pad_sustains(base: dict[int, Shape]) -> Result:
    """The 2026-07-28 regression.

    desire.tidal's d5 measured peak -11.5 / rms -53.3 — a 41.8 dB crest with a
    duty near zero. On an aggregate that is indistinguishable from a quiet
    sustained part; on the time axis it is obviously an attack followed by
    reverb tail and silence. PLN heard it as "plays a few secs then only clics".
    """
    sh = base.get(5)
    if sh is None or sh.silent:
        return no("pad_sustains", "d5 made no sound")
    if sh.duty > 0.6:
        return ok("pad_sustains", "d5 sustains",
                  f"duty {sh.duty*100:.0f}%, crest {sh.crest_db:.1f} dB")
    return no("pad_sustains",
              "d5 attacks then dies — a click plus a tail, not a pad. Suspect a "
              "shared cut group, a truncating envelope, or a closed filter.",
              f"duty {sh.duty*100:.0f}%, crest {sh.crest_db:.1f} dB")


@case("hats_are_dense", "d2 delivers a real onset rate (> 2/s)")
def _hats_dense(base: dict[int, Shape]) -> Result:
    sh = base.get(2)
    if sh is None or sh.silent:
        return no("hats_are_dense", "d2 made no sound")
    if sh.onsets_per_s > 2.0:
        return ok("hats_are_dense", "d2 is percussive",
                  f"{sh.onsets_per_s:.1f} onsets/s")
    return no("hats_are_dense", "d2 is too sparse to be the hat orbit",
              f"{sh.onsets_per_s:.1f} onsets/s")


@case("riser_rises", "d7's level climbs across the window (the time-axis test)")
def _riser_rises(base: dict[int, Shape]) -> Result:
    sh = base.get(7)
    if sh is None or sh.silent:
        return no("riser_rises", "d7 made no sound")
    live = [b for b in sh.timeline if b != -math.inf]
    if len(live) < 6:
        return no("riser_rises", "not enough bins to judge a slope",
                  f"{len(live)} live bins")
    # PHASE-INDEPENDENT. Comparing the first half of the window to the second
    # half looked reasonable and was a coin flip: d7 is driven by `slow 4 saw`,
    # whose period (~7.8 s at 124) is comparable to the capture window, so a
    # fixed window lands on a random phase of the ramp. Consecutive runs of the
    # identical pattern scored +7.1 dB and -20.9 dB.
    #
    # What actually defines a riser is that SOMEWHERE it climbs a long way: the
    # largest rise from any bin to any LATER bin (the max "draw-up"). That is
    # true at every phase, and a steady or falling orbit cannot fake it.
    best, running_min = 0.0, live[0]
    for v in live[1:]:
        running_min = min(running_min, v)
        best = max(best, v - running_min)
    span = max(live) - min(live)
    measured = f"max draw-up {best:.1f} dB over a {span:.1f} dB span"
    if best > 10.0:
        return ok("riser_rises", "d7 climbs", measured)
    return no("riser_rises",
              "d7 never climbs far — an aggregate would never have caught this",
              measured)


@case("cut_groups_distinct", "no two orbits share a cut group (static check)")
def _cut_groups(base: dict[int, Shape]) -> Result:
    sys.path.insert(0, str(TOOLS))
    from pvlint.core import load as pv_load
    from pvlint.rules import check as pv_check
    findings = [f for f in pv_check(pv_load(str(TRACK))) if f.rule == "PV004"]
    if not findings:
        return ok("cut_groups_distinct", "every orbit owns its cut group")
    return no("cut_groups_distinct", findings[0].message)


# --------------------------------------------------------------------------
# The control case — needs its own captures, so it runs separately
# --------------------------------------------------------------------------

def control_case(seconds: float) -> Result:
    """Sweep gF1 (CC49) and assert d4's brightness actually moves.

    Measured on the CENTROID, not rms: a filter changes the spectrum and leaves
    the level alone, so an rms-based assertion here would report "no change" on
    a filter that is working perfectly.

    gF1 is BIPOLAR — 64 is bypass, low is a closing low-pass, high is a climbing
    high-pass. So the sweep runs centre -> low, and the expected direction is
    DOWN. Comparing two arbitrary points on a bipolar control is how you get a
    confident wrong answer.
    """
    name = "filter_bites"
    cc = 49
    assert cc in SAFE_CC and cc not in FORBIDDEN_CC

    # Resolve the port BY NAME on every call. A port resolved once at boot and
    # cached is this rig's single most common silent failure: a device event
    # renumbers it, the handle goes stale, and nothing reports an error.
    try:
        port = lcxl.sc_midi_input_port()
    except Exception:
        port = None
    if not port:
        return Result(name, False, "could not resolve an LCXL sequencer port "
                                   "— cannot send virtual CC", "skipped")

    # Baseline TWICE. Patterns vary per cycle on their own, so a single
    # baseline cannot tell a real effect from ordinary drift.
    lcxl.send_cc(port, cc, 64)
    time.sleep(2.0)
    b1 = measure([4], seconds)[4]
    b2 = measure([4], seconds)[4]
    drift = abs(b2.centroid_hz - b1.centroid_hz)
    base_c = statistics.mean([b1.centroid_hz, b2.centroid_hz])

    lcxl.send_cc(port, cc, 10)          # well below centre = closing low-pass
    time.sleep(2.0)
    lowered = measure([4], seconds)[4]

    lcxl.send_cc(port, cc, 64)          # restore bypass, always
    swing = abs(lowered.centroid_hz - base_c)
    measured = (f"centroid {base_c:.0f} -> {lowered.centroid_hz:.0f} Hz "
                f"(swing {swing:.0f}, drift {drift:.0f})")

    if base_c <= 0:
        return Result(name, False, "no baseline signal on d4", measured)
    if swing < 2 * max(drift, 1.0):
        return Result(name, False,
                      "INCONCLUSIVE — the swing is not clearly bigger than the "
                      "pattern's own drift. Never read a percentage off a "
                      "near-zero baseline.", measured)
    if lowered.centroid_hz < base_c:
        return Result(name, True, "gF1 closes the low-pass as expected", measured)
    return Result(name, False, "gF1 moved brightness the WRONG way", measured)


# --------------------------------------------------------------------------

def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(prog="pv-at", description=__doc__.split("\n")[0])
    ap.add_argument("--list", action="store_true")
    ap.add_argument("--case", action="append", help="run only these cases")
    ap.add_argument("--seconds", type=float, default=10.0, help="capture window")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--no-controls", action="store_true",
                    help="skip the case that sends virtual CC")
    ap.add_argument("--keep-playing", action="store_true",
                    help="do not hush at the end")
    args = ap.parse_args(argv)

    if args.list:
        for n, (doc, _) in CASES.items():
            print(f"  {n:22s} {doc}")
        print(f"  {'filter_bites':22s} gF1 (CC49) measurably changes d4's brightness")
        return 0

    rig = rig_state()
    if not rig.ready:
        msg = f"pv-at: rig not ready — {rig.why_not()}"
        print(json.dumps({"status": "rig_not_ready", "why": rig.why_not()})
              if args.json else msg, file=sys.stderr)
        return 2

    if not args.json:
        print(f"pv-at: playing {TRACK.name} and measuring {len(ORBITS)} orbits "
              f"over {args.seconds:g}s\n")

    play(TRACK, settle=8.0)
    base = measure(ORBITS, args.seconds)

    selected = args.case or list(CASES)
    results: list[Result] = []
    for n in selected:
        if n in CASES:
            results.append(CASES[n][1](base))
    if not args.no_controls and (not args.case or "filter_bites" in selected):
        results.append(control_case(min(args.seconds, 8.0)))

    if not args.keep_playing:
        hush()

    if args.json:
        print(json.dumps({
            "status": "ran",
            "shapes": {f"d{dn}": vars(sh) for dn, sh in base.items()},
            "results": [vars(r) for r in results],
        }, default=str))
    else:
        for dn in ORBITS:
            print(f"  {base[dn]}")
        print()
        for r in results:
            mark = "PASS" if r.passed else "FAIL"
            print(f"  [{mark}] {r.name:22s} {r.detail}")
            if r.measured:
                print(f"         {r.measured}")
        n_ok = sum(1 for r in results if r.passed)
        print(f"\npv-at: {n_ok}/{len(results)} passed")

    return 0 if all(r.passed for r in results) else 1


if __name__ == "__main__":
    sys.exit(main())
