From 8760b395afffe3cd327c30d0d8b3270cb6e21531 Mon Sep 17 00:00:00 2001 From: Aqua Date: Tue, 10 Nov 2026 17:36:03 +0000 Subject: [PATCH] aqua-ear: audio perception tool (spectrogram, onsets, envelope, detectors, compare) Python (numpy, scipy, matplotlib, soundfile). Onsets at 1 ms on a full-band and a >2 kHz channel; double hit, click/step/DC, clipping/true peak, overlapping tails, abrupt end detectors with stated assumptions; compare mode. True peak and BS.1770 loudness ported from aqua-feel/metrics.js (cross-checked against it in a test). 25 synthetic-fixture tests, each detector with a near-miss. Co-Authored-By: Claude Sonnet 5.5 --- tools/aqua-ear/.gitignore | 1 + tools/aqua-ear/README.md | 83 ++++++ tools/aqua-ear/aqua_ear.py | 451 ++++++++++++++++++++++++++++++++ tools/aqua-ear/test_aqua_ear.py | 164 ++++++++++++ tools/aqua-feel/README.md | 2 + 5 files changed, 701 insertions(+) create mode 100644 tools/aqua-ear/.gitignore create mode 100644 tools/aqua-ear/README.md create mode 100644 tools/aqua-ear/aqua_ear.py create mode 100644 tools/aqua-ear/test_aqua_ear.py diff --git a/tools/aqua-ear/.gitignore b/tools/aqua-ear/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/tools/aqua-ear/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/tools/aqua-ear/README.md b/tools/aqua-ear/README.md new file mode 100644 index 0000000..ccff787 --- /dev/null +++ b/tools/aqua-ear/README.md @@ -0,0 +1,83 @@ +# aqua-ear + +Audio perception for AquaSynth: give it a wav, ogg or flac and it writes pictures a vision model can read, an onset list at 1 ms, a level +envelope, and five detectors that each print one line with the numbers behind it. A compare mode does the same for two files and says what +changed. It is the counterpart of `../aqua-feel` (which renders and measures AquaSynth's own patches): aqua-ear looks at any finished audio +file, including a decoded Ogg Vorbis clip. + +**Nothing here hears anything.** Every number is a signal fact about a decoded buffer. A detector firing is a thing to look at or listen to, +not a defect; a detector staying quiet is not a pass. + +## Run + +``` +# once: numpy, scipy, matplotlib and soundfile (libsndfile does the Vorbis). Nothing else; no network at run time. +python3 -m venv ear-venv && ear-venv/bin/pip install numpy scipy matplotlib soundfile +# (this sandbox: /var/lib/mind-aqua-self/ogg-venv/bin/python already has all four) + +python aqua_ear.py analyze clip.ogg --out out/clip # one file +python aqua_ear.py compare before.ogg after.ogg --out out/cmp # two files, B relative to A +python aqua_ear.py --help # add --strict to either to exit 3 when any detector fires +python -m unittest -v test_aqua_ear # 25 tests on synthetic fixtures generated in the test; ~6 s +``` + +Exit code: 0 whatever the findings, 2 on a tool error (unreadable file), 3 only with `--strict` and a firing detector. + +## Output + +`analyze` writes, to `--out`: + +| file | what | +|---|---| +| `spectrogram.png` | whole file, log frequency 40 Hz to Nyquist, dBFS colour scale (80 dB range, a full-scale sine reads 0 dBFS), onsets as labelled dashed lines. Fonts are 15 pt and up on a 1600 px canvas | +| `zoom.png` | one row per onset window (15 ms before to 85 ms after; overlapping windows merge; at most 6): waveform on the left, 5.3 ms-frame spectrogram on the right, onset times printed in boxes | +| `envelope.png` | level in dBFS (1 ms envelope) against ms, onsets marked | +| `onsets.json`, `onsets.txt` | onset list, 1 ms resolution, with `band` (`full` or `hf`), peak level and peak time | +| `report.json`, `report.txt` | file facts (peak, true peak, BS.1770 loudness, rms), onsets, the five verdict lines, and every threshold used | + +`compare` writes `compare.png` (A and B spectrograms stacked on one time axis, envelopes overlaid), `compare.txt`/`compare.json` (onset diff +with matches inside 3 ms, per-detector verdict diff with CHANGED marked, B minus A deltas for peak, true peak, loudness, rms, duration) and the +full single-file reports in `a/` and `b/`. + +## Onsets + +Two channels, both on a 1 ms grid. `full`: Hilbert magnitude of the signal, 1 ms smoothing, onset where the level rises 4 dB above its trough +of the last 20 ms (trough clamped to max(-60 dBFS, file peak -45 dB)); the reported ms is where the rise crosses trough +2 dB, so it is the start of +the rise, good to a millisecond or two. `hf`: the same detector on the signal above 2 kHz (4th-order Butterworth, zero-phase), ignored below +-75 dBFS. The `hf` channel exists because a tick riding on a tone (the Dings Void tick, 20 dB under the tone) barely moves the full-band level and is +plain in the high band. An onset that lands while the previous one is still rising is merged into it, and a precursor that never dips is one onset. + +## Detectors + +Each verdict line carries its own stated assumption. All boundaries are the `TH` table at the top of `aqua_ear.py` and are printed in `report.json`. + +| detector | fires when | notes | +|---|---|---| +| `double_hit` | two onsets in one band closer than 80 ms, within one note, with peaks within 20 dB | **one note** = the full-band level between the two never falls more than 15 dB below the first onset's peak (for `hf` onsets: below the level there). Two staccato notes that die between hits are not a flam; two hits 150 ms apart are not; a precursor 30 dB under the main hit is reported as "precursor", not fired | +| `click_step_dc` | `\|mean\|` >= 0.005 (-46 dBFS), or a discontinuity: second difference >= 0.02 and >= 25x its local (+-2.5 ms) mean | labelled `step` if the level 2 ms after differs from 2 ms before by half the jump or more, else `click`. A bright continuous tone does not fire; a sample-accurate jump on a quiet low tone (below 0.02) does not either | +| `clipping_true_peak` | any sample over 1.0, or a run of 3+ equal samples at >= 0.99, or true peak > 0 dBTP | true peak is 4x oversampled. Plateaus below 0.99 (a limiter at -2 dBFS) are not reported as clipping. A note at -1 dBFS does not fire. After a lossy encode a clip shows as overs and true peak, not equal samples | +| `overlapping_tails` | onsets >= 80 ms apart where the lowest level in the 5 ms before the new onset is above -40 dB re the previous note's peak | a long ring-out always trips this; it states a fact about overlap, not about whether the overlap is wanted | +| `abrupt_end` | last 3 ms above -40 dB re file peak and above -70 dBFS, and not 6 dB below the same window 9 ms earlier | a fade, a decayed tail, or trailing digital silence are clear | + +## Reused from aqua-feel, and how + +`true_peak` (the Hann-windowed 48-tap sinc, 4x) and `loudness_lufs` (BS.1770-4 K-weighting, 400 ms blocks, -70 absolute and -10 LU relative gates) +are ports of `../aqua-feel/metrics.js` `truePeak` and `integratedLoudness`, about 25 lines, not calls into it. Reason: aqua-ear is Python for +numpy/scipy/matplotlib and libsndfile (the only Vorbis reader on the box); calling JS would add node as a run-time dependency and a process per file. +A test runs both on the same buffer when `node` is present and requires agreement (true peak to 1e-5, loudness to 0.01 LU); it skips otherwise. +Coefficients are 48 kHz; other sample rates are resampled to 48 kHz for loudness only. Multichannel files are mixed to mono for everything. +aqua-feel's own onset and clip logic is not reused: it is tuned to its rendered patches (0.17 ms peak follower, first-onset only). + +## What it cannot judge + +Timbre. Annoyance. Whether a sound is pleasant, harsh, cheap, muddy, clear in a room or readable as its object. Musical feel, timing feel, +how an onset interacts with the next one in a listener's ear. Anything a listener hears as character. Whether a click is audible (it measures +size, not audibility: a 0.02 second-difference click on a quiet tone can be inaudible, and a barely-over-threshold tail can be a swamp). It also +cannot see what is under -60 dBFS in the `full` channel, cannot separate two onsets inside one rising attack, and reads pitch only as +a smear on the spectrogram. Verdicts are signal facts: the thresholds are tripwires chosen after looking at synthetic fixtures and the Dings Void +clips, not taste. + +## Planned versus live + +Live: everything above. Not built: per-onset spectral-flux detector for noise-on-noise, an audibility model, batch mode over a directory, `--strict` +per-detector selection. None of these is needed by a current consumer. diff --git a/tools/aqua-ear/aqua_ear.py b/tools/aqua-ear/aqua_ear.py new file mode 100644 index 0000000..de221c3 --- /dev/null +++ b/tools/aqua-ear/aqua_ear.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +"""aqua-ear: audio perception tool. Measures signal facts in a wav/ogg/flac and draws pictures a vision model can read. +It does not hear anything: no detector is a verdict of good or bad. See README.md.""" +import argparse, json, os, sys, tempfile + +os.environ.setdefault("MPLCONFIGDIR", os.path.join(tempfile.gettempdir(), "aqua-ear-mpl")) +import numpy as np +import soundfile as sf +import scipy.signal as ss +from scipy.ndimage import uniform_filter1d +import matplotlib +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +# ---- thresholds: every detector's boundary lives here, stated in the report evidence --------------------------------- +TH = dict( + floor_dbfs=-60.0, # onsets below this absolute level are ignored + rel_floor_db=-45.0, # ... and below this many dB under the file peak + onset_rise_db=4.0, # envelope must rise this far above its recent trough to count as an onset + trough_window_ms=20, # how far back the trough is looked for + flam_gap_ms=80.0, # two onsets closer than this may be a flam/double hit + flam_balance_db=20.0, # ... and the two onsets' peaks are within this many dB (a much quieter first onset is a precursor, reported not fired) + one_note_db=15.0, # ... if the envelope between them never fell more than this below the first onset's peak + dc_abs=0.005, # |mean| at or above this is a DC offset (-46 dBFS) + step_abs=0.02, # a discontinuity must jump at least this much (second difference, -34 dBFS) ... + step_ratio=25.0, # ... and be this many times the local mean |second difference| (+-2.5 ms) + clip_run=3, # consecutive plateau samples at the ceiling that make a clipped run + clip_level=0.99, # plateau must be at or above this |x| + tp_limit_dbtp=0.0, # true peak above this is an inter-sample over + overlap_gap_ms=80.0, # onsets this far apart or more are separate notes ... + overlap_tail_db=-40.0, # ... and overlap if the previous note's tail is above this many dB re its peak at the new onset + end_decay_db=6.0, # ... unless it has fallen at least this far in the preceding 9 ms (a fade) + end_tail_db=-40.0, # abrupt end: the last 3 ms is louder than this many dB re file peak ... + end_abs_dbfs=-70.0, # ... and louder than this absolute level + hf_cut_hz=2000.0, # transient channel: onsets also detected on the signal above this (4th-order Butterworth, zero-phase) + hf_floor_dbfs=-75.0, # ... ignoring anything whose high band peaks below this + match_ms=3.0, # compare mode: onsets within this are the same onset +) +FONT = 15 +plt.rcParams.update({"font.size": FONT, "axes.titlesize": FONT + 3, "axes.labelsize": FONT + 1, + "xtick.labelsize": FONT, "ytick.labelsize": FONT, "legend.fontsize": FONT - 1, + "axes.grid": True, "grid.alpha": 0.25}) + +def db(x): return 20 * np.log10(np.maximum(x, 1e-9)) + +# ---- loading ---------------------------------------------------------------------------------------------------------- +def load(path): + x, sr = sf.read(path, dtype="float64", always_2d=True) + ch = x.shape[1] + return x.mean(axis=1), sr, ch + +# ---- ported from tools/aqua-feel/metrics.js (JS lines ported, not called: keeps this tool one language, no node at run time) +def true_peak(x, sr): + """4x oversampled true peak, same Hann-windowed 48-tap sinc as metrics.js truePeak (linear). Returns (tp, sample_peak).""" + T = 12 + k = np.arange(2 * T) - T + 1 + peak = float(np.max(np.abs(x))) if len(x) else 0.0 + tp = peak + for ph in (0.25, 0.5, 0.75): + t = k - ph + h = np.sinc(t) * (0.5 + 0.5 * np.cos(np.pi * t / T)) + y = _corr(x, h, T) # metrics.js: y[n] = sum_i x[n+i-T+1] h[i] + tp = max(tp, float(np.max(np.abs(y))) if len(y) else 0.0) + return tp, peak + +def _corr(x, h, T): + full = np.correlate(x, h, mode="full") # full[j] = sum_i x[j-(len(h)-1)+i] h[i], so y[n] = full[n+T] + return full[T:T + len(x)] + +K1 = ([1.53512485958697, -2.69169618940638, 1.19839281085285], [1, -1.69065929318241, 0.73248077421585]) +K2 = ([1, -2, 1], [1, -1.99004745483398, 0.99007225036621]) + +def loudness_lufs(x, sr): + """BS.1770-4 integrated, mono, as metrics.js integratedLoudness. Coefficients are 48 kHz, so other rates are resampled.""" + if sr != 48000: + from math import gcd + g = gcd(48000, sr); x = ss.resample_poly(x, 48000 // g, sr // g) + k = ss.lfilter(*K2, ss.lfilter(*K1, x)) + B, H = 19200, 4800 + n = max(1, int(np.ceil(max(0, len(k) - B) / H)) + 1) + ms = [] + for i in range(n): + seg = k[i * H:i * H + B] + ms.append(float(np.sum(seg * seg)) / B) + ms = np.array(ms) + l = lambda e: -0.691 + 10 * np.log10(np.maximum(e, 1e-30)) + p1 = ms[l(ms) > -70] + if not len(p1): return None + rel = l(p1.mean()) - 10 + p2 = p1[l(p1) > rel] + return float(l(p2.mean())) if len(p2) else None + +# ---- envelope and onsets ------------------------------------------------------------------------------------------------ +def envelope_1ms(x, sr): + """Hilbert magnitude, 1 ms moving average, then max per 1 ms block: env_db[k] is the level at k ms. Ripple-free for tones.""" + if len(x) == 0: return np.array([-180.0]) + pad = np.zeros(int(sr * 0.05)) # zero-pad: the FFT Hilbert is circular and rings at both ends otherwise + a = np.abs(ss.hilbert(np.concatenate([pad, x, pad])))[len(pad):len(pad) + len(x)] + a = uniform_filter1d(a, max(1, int(round(sr * 0.001))), mode="nearest") + blk = sr / 1000.0 + nb = int(np.ceil(len(a) / blk)) + idx = np.minimum((np.arange(len(a)) / blk).astype(int), nb - 1) + env = np.zeros(nb); np.maximum.at(env, idx, a) + return db(env) + +def detect_onsets(env_db, peak_db, abs_floor=None): + """Rise detector on the 1 ms envelope. Onset = first ms (within 10 ms before the trigger) at trough+2 dB or the floor.""" + floor = max(TH["floor_dbfs"] if abs_floor is None else abs_floor, peak_db + TH["rel_floor_db"]) + W, rise = TH["trough_window_ms"], TH["onset_rise_db"] + out, in_event, ev_peak, since, low = [], False, -1e9, 0, 0 + for i in range(len(env_db)): + e = env_db[i] + if in_event: + low = low + 1 if e - env_db[i - 1] < rise / W else 0 # rise/W = 0.2 dB/ms: slower than that cannot reach a trigger inside the trough window + if low >= 2: in_event, since = False, i - 1 # the event is over once the envelope has stopped rising for 2 ms + continue + lo = max(since, i - W) + trough = max(env_db[lo:i + 1].min(), floor) + if e >= floor and e - trough >= rise: + lvl = trough + 2.0 + j = i + while j > max(lo, i - 10) and env_db[j - 1] >= lvl: j -= 1 + out.append(j); in_event, ev_peak, low = True, e, 0 + return out + +def onset_table(env_db, onsets, band="full"): + rows = [] + for n, t in enumerate(onsets): + nxt = onsets[n + 1] if n + 1 < len(onsets) else len(env_db) + seg = env_db[t:min(nxt, t + 80) + 1] + rows.append(dict(band=band, ms=int(t), peak_dbfs=round(float(seg.max()), 1), peak_ms=int(t + seg.argmax()), + level_at_onset_dbfs=round(float(env_db[t]), 1))) + return rows + +# ---- detectors: each returns dict(name, fired, verdict, evidence) ----------------------------------------------------------- +def det_flam(env_db, rows, band="full"): + """Pairs of onsets in one band closer than flam_gap_ms. 'One note' is judged on the full-band envelope between them.""" + tag = "" if band == "full" else f"[{band}] " + pairs = [] + for a, b in zip(rows, rows[1:]): + gap = b["ms"] - a["ms"] + if gap >= TH["flam_gap_ms"]: continue + between = env_db[a["ms"]:b["ms"] + 1] + ref = a["peak_dbfs"] if band == "full" else float(between.max()) + floor_db = float(between.min()) - ref + diff = round(abs(a["peak_dbfs"] - b["peak_dbfs"]), 1) + pairs.append(dict(band=band, first_ms=a["ms"], second_ms=b["ms"], gap_ms=gap, min_between_db_re_first_peak=round(floor_db, 1), + one_note=bool(floor_db >= -TH["one_note_db"]), peak_difference_db=diff, comparable=bool(diff <= TH["flam_balance_db"]))) + hits = [p for p in pairs if p["one_note"] and p["comparable"]] + if hits: + verdict = "FIRES double hit / flam: " + "; ".join(f"{tag}onsets at {p['first_ms']} and {p['second_ms']} ms, {p['gap_ms']} ms apart, level between never below {p['min_between_db_re_first_peak']} dB, peaks {p['peak_difference_db']} dB apart" for p in hits) + elif pairs: + verdict = f"no flam: {tag}close onsets exist but " + "; ".join( + (f"onsets at {p['first_ms']} and {p['second_ms']} ms differ by {p['peak_difference_db']} dB in peak (precursor or grace swell, not a second hit)" if p["one_note"] + else f"the envelope fell to {p['min_between_db_re_first_peak']} dB between {p['first_ms']} and {p['second_ms']} ms (separate notes)") for p in pairs) + else: + verdict = f"no flam: {tag}no two onsets closer than {TH['flam_gap_ms']:g} ms ({len(rows)} onset(s))" + return dict(name="double_hit", fired=bool(hits), verdict=verdict, evidence=pairs) + +def det_click_dc(x, sr): + m = float(np.mean(x)) if len(x) else 0.0 + dc_fire = abs(m) >= TH["dc_abs"] + e = np.abs(x[2:] - 2 * x[1:-1] + x[:-2]) if len(x) > 3 else np.zeros(0) + ev = [] + if len(e): + w = max(3, int(sr * 0.005) | 1) + loc = uniform_filter1d(e, w, mode="nearest") + cand = np.where((e >= TH["step_abs"]) & (e >= TH["step_ratio"] * loc))[0] + groups = [] + for c in cand: + if groups and c - groups[-1][-1] <= 2: groups[-1].append(c) + else: groups.append([c]) + h = max(2, int(sr * 0.002)) + for g in groups[:50]: + c = g[int(np.argmax(e[g]))] + 1 # sample index of the middle of the second difference + before = x[max(0, c - h):c - 1]; after = x[c + 1:c + 1 + h] + lvl = float(after.mean() - before.mean()) if len(before) and len(after) else 0.0 + jump = float(abs(x[c] - x[c - 1])) + kind = "step" if abs(lvl) >= 0.5 * jump else "click" + ev.append(dict(ms=round(c * 1000 / sr, 2), kind=kind, jump=round(jump, 4), level_shift=round(lvl, 4))) + fired = dc_fire or bool(ev) + parts = [] + parts.append(f"DC offset {m:+.5f} ({db(abs(m)):.1f} dBFS) {'AT/ABOVE' if dc_fire else 'below'} {TH['dc_abs']} limit") + parts.append(f"{len(ev)} discontinuity(ies)" + (": " + ", ".join(f"{d['kind']} at {d['ms']} ms jump {d['jump']}" for d in ev[:6]) if ev else "")) + return dict(name="click_step_dc", fired=fired, verdict=("FIRES " if fired else "none: ") + "; ".join(parts) + + f" [assumption: discontinuity = second difference >= {TH['step_abs']} and >= {TH['step_ratio']:g}x its local mean]", + evidence=dict(dc=m, discontinuities=ev)) + +def det_clip(x, sr): + tp, sp = true_peak(x, sr) + over = int(np.sum(np.abs(x) > 1.0)) + runs, run = 0, 0 + ax = np.abs(x) + for i in range(len(x)): + if ax[i] >= TH["clip_level"] and (run == 0 or abs(x[i] - x[i - 1]) < 1e-6 and ax[i - 1] >= TH["clip_level"]): + run += 1 + if run == TH["clip_run"]: runs += 1 + else: + run = 1 if ax[i] >= TH["clip_level"] else 0 + tpdb = float(db(tp)) + fired = over > 0 or runs > 0 or tpdb > TH["tp_limit_dbtp"] + v = (f"sample peak {db(sp):.2f} dBFS, true peak {tpdb:.2f} dBTP, {over} sample(s) over 1.0, {runs} clipped plateau run(s) " + f"of >={TH['clip_run']} equal samples >= {TH['clip_level']}") + return dict(name="clipping_true_peak", fired=fired, verdict=("FIRES " if fired else "none: ") + v + + f" [fires on any over, any plateau run, or true peak > {TH['tp_limit_dbtp']:g} dBTP]", + evidence=dict(sample_peak_dbfs=float(db(sp)), true_peak_dbtp=tpdb, over_samples=over, plateau_runs=runs)) + +def det_overlap(env_db, rows): + hits, near = [], [] + for a, b in zip(rows, rows[1:]): + gap = b["ms"] - a["ms"] + if gap < TH["overlap_gap_ms"]: continue + tail = float(env_db[max(0, b["ms"] - 5):b["ms"] + 1].min()) - a["peak_dbfs"] # lowest level in the 5 ms before the new onset + (hits if tail > TH["overlap_tail_db"] else near).append(dict(prev_ms=a["ms"], next_ms=b["ms"], gap_ms=gap, tail_db_re_prev_peak=round(tail, 1))) + if hits: + v = "FIRES overlapping tails: " + "; ".join(f"onset at {h['next_ms']} ms arrives while the note from {h['prev_ms']} ms is still at {h['tail_db_re_prev_peak']} dB re its peak" for h in hits) + else: + v = f"none: {len(near)} later onset(s) arrived on a tail below {TH['overlap_tail_db']:g} dB" if near else "none: fewer than two separate notes" + return dict(name="overlapping_tails", fired=bool(hits), verdict=v + f" [assumption: separate notes >= {TH['overlap_gap_ms']:g} ms apart; overlap = tail above {TH['overlap_tail_db']:g} dB re previous peak (lowest level in the 5 ms before the new onset)]", evidence=hits + near) + +def det_end(x, sr, peak): + n = max(1, int(sr * 0.003)) + tail = float(np.max(np.abs(x[-n:]))) if len(x) else 0.0 + ref = float(np.max(np.abs(x[-4 * n:-3 * n]))) if len(x) >= 4 * n else tail # the 3 ms window ending 9 ms before the end + drop = float(db(ref) - db(tail)) + last = float(x[-1]) if len(x) else 0.0 + rel = float(db(tail) - db(peak)) + fired = peak > 0 and rel > TH["end_tail_db"] and db(tail) > TH["end_abs_dbfs"] and drop < TH["end_decay_db"] + v = f"last 3 ms peak {db(tail):.1f} dBFS ({rel:.1f} dB re file peak), {drop:.1f} dB below the same window 9 ms earlier, final sample {last:+.5f}" + return dict(name="abrupt_end", fired=bool(fired), verdict=("FIRES abrupt end cut: " if fired else "none: ") + v + + f" [assumption: abrupt if the last 3 ms is above {TH['end_tail_db']:g} dB re file peak and above {TH['end_abs_dbfs']:g} dBFS and has not fallen {TH['end_decay_db']:g} dB since 9 ms earlier]", + evidence=dict(tail_dbfs=float(db(tail)), tail_db_re_peak=rel, drop_db=drop, final_sample=last)) + +# ---- analysis --------------------------------------------------------------------------------------------------------------- +def analyze(path): + x, sr, ch = load(path) + tp, sp = true_peak(x, sr) + env = envelope_1ms(x, sr) + peak_db = float(env.max()) + onsets = detect_onsets(env, peak_db) if sp > 1e-6 else [] + rows = onset_table(env, onsets) + hf_rows = [] + if sr / 2 > TH["hf_cut_hz"] * 1.2 and sp > 1e-6: + hf = ss.sosfiltfilt(ss.butter(4, TH["hf_cut_hz"], "highpass", fs=sr, output="sos"), x) + env_hf = envelope_1ms(hf, sr) + if env_hf.max() > TH["hf_floor_dbfs"]: + hf_rows = onset_table(env_hf, detect_onsets(env_hf, float(env_hf.max()), TH["hf_floor_dbfs"]), "hf") + f_full, f_hf = det_flam(env, rows), det_flam(env, hf_rows, "hf") + flam = dict(name="double_hit", fired=f_full["fired"] or f_hf["fired"], evidence=f_full["evidence"] + f_hf["evidence"], + verdict=f_full["verdict"] + " | " + f_hf["verdict"] + f" [assumption: one note = full-band level between the two onsets never below -{TH['one_note_db']:g} dB re the first (hf: re the level there); peaks within {TH['flam_balance_db']:g} dB; hf = above {TH['hf_cut_hz']:g} Hz]") + dets = [flam, det_click_dc(x, sr), det_clip(x, sr), det_overlap(env, rows), det_end(x, sr, sp)] + lufs = loudness_lufs(x, sr) + info = dict(file=os.path.basename(path), sample_rate=sr, channels=ch, duration_ms=round(len(x) * 1000 / sr, 1), + sample_peak_dbfs=round(float(db(sp)), 2), true_peak_dbtp=round(float(db(tp)), 2), + loudness_lufs=None if lufs is None else round(lufs, 2), rms_dbfs=round(float(db(np.sqrt(np.mean(x * x)))), 2) if len(x) else None) + return dict(info=info, onsets=rows, onsets_hf=hf_rows, detectors=dets, thresholds=TH), x, sr, env + +# ---- pictures --------------------------------------------------------------------------------------------------------------- +def stft_db(x, sr, nperseg, hop): + w = ss.windows.hann(nperseg) + f, t, Z = ss.stft(x, sr, window=w, nperseg=nperseg, noverlap=nperseg - hop, boundary="zeros", padded=True) + mag = np.abs(Z) * 2 # scipy's default 'spectrum' scaling is 1/sum(window), so a full-scale sine reads 0 dBFS + return f, t * 1000.0, db(mag) + +def draw_spec(ax, x, sr, nperseg, hop, fmin=40, label=True): + f, tms, S = stft_db(x, sr, nperseg, hop) + keep = f >= fmin + top = S.max() + m = ax.pcolormesh(tms, f[keep], S[keep], shading="auto", cmap="magma", vmin=top - 80, vmax=top) + ax.set_xlim(0, len(x) * 1000 / sr) + ax.set_yscale("log"); ax.set_ylim(fmin, sr / 2) + ticks = [t for t in (50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000) if fmin <= t <= sr / 2] + ax.set_yticks(ticks); ax.set_yticklabels([f"{t/1000:g}k" if t >= 1000 else str(t) for t in ticks]) + ax.grid(False) + if label: ax.set_ylabel("frequency (Hz, log)") + return m + +COL = {"spec": {"full": "#00e5ff", "hf": "#ffd400"}, "plain": {"full": "#c0392b", "hf": "#d35400"}} +def items(rep): return sorted([(o["ms"], o["band"]) for o in rep["onsets"] + rep.get("onsets_hf", [])]) + +def mark_onsets(ax, its, bg="spec", lo=None, hi=None): + """its: [(ms, band)]. Dashed line plus a big boxed label per onset; hf (transient channel) onsets are yellow/orange and say 'hf'.""" + for k, (t, band) in enumerate(its): + if lo is not None and not (lo <= t <= hi): continue + c = COL[bg][band] + ax.axvline(t, color=c, lw=1.8, ls="--" if band == "full" else ":") + ax.annotate((f"{t:.0f} ms" if band == "full" else f"hf {t:.0f} ms"), (t, 1.0), xycoords=("data", "axes fraction"), xytext=(3, -6 - 19 * (k % 3)), + textcoords="offset points", color="white" if (bg == "plain" or band == "full") else "black", fontsize=FONT, fontweight="bold", va="top", + bbox=dict(facecolor=c, alpha=0.9, pad=2, lw=0)) + +def title_for(rep): + i = rep["info"] + return f"{i['file']} {i['duration_ms']:g} ms {i['sample_rate']} Hz peak {i['sample_peak_dbfs']} dBFS TP {i['true_peak_dbtp']} dBTP" + +def plot_spectrogram(rep, x, sr, path): + ons = items(rep) + dur = len(x) * 1000 / sr + fig, ax = plt.subplots(figsize=(16, 7.5), dpi=100) + m = draw_spec(ax, x, sr, 2048 if dur > 400 else 1024, 128) + mark_onsets(ax, ons) + ax.set_xlabel("time (ms)"); ax.set_title("Spectrogram (dBFS, onsets dashed)\n" + title_for(rep), loc="left") + cb = fig.colorbar(m, ax=ax, pad=0.01); cb.set_label("dBFS per bin (80 dB range)", fontsize=FONT) + fig.tight_layout(); fig.savefig(path); plt.close(fig) + +def zoom_windows(ons, dur, pre=15, post=85, cap=6): + wins = [] + for t in ons: + a, b = max(0.0, t - pre), min(dur, t + post) + if wins and a <= wins[-1][1]: wins[-1][1] = b; wins[-1][2].append(t) + else: wins.append([a, b, [t]]) + return wins[:cap], max(0, len(wins) - cap) + +def plot_zoom(rep, x, sr, path): + ons = items(rep) + wins, dropped = zoom_windows([t for t, _ in ons], len(x) * 1000 / sr) + if not wins: return False + fig, axes = plt.subplots(len(wins), 2, figsize=(16, 4.2 * len(wins)), dpi=100, squeeze=False, gridspec_kw=dict(width_ratios=[1, 1.3])) + for r, (a, b, ts) in enumerate(wins): + seg = x[int(a * sr / 1000):int(b * sr / 1000)] + tt = a + np.arange(len(seg)) * 1000 / sr + pad_ms = 8.0 # analyse a wider slice so the spectrogram has no zero-padded edge frames inside the window + a2 = max(0.0, a - pad_ms); wide = x[int(a2 * sr / 1000):int((b + pad_ms) * sr / 1000)] + axw, axs = axes[r] + axw.plot(tt, seg, color="#1f4e9c", lw=1.2); axw.set_xlim(a, b); axw.set_ylim(-1.05 * max(np.abs(seg).max(), 1e-3), 1.05 * max(np.abs(seg).max(), 1e-3)) + mark_onsets(axw, ons, "plain", a, b); axw.set_ylabel("amplitude"); axw.set_title(f"waveform {a:.0f}-{b:.0f} ms", loc="left") + ff, tm, SS = stft_db(wide, sr, 256, 16); k = ff >= 100 + axs.pcolormesh(tm + a2, ff[k], SS[k], shading="auto", cmap="magma", vmin=SS.max() - 70, vmax=SS.max()) + axs.set_yscale("log"); axs.set_ylim(100, sr / 2); axs.set_xlim(a, b); axs.grid(False) + tk = [t for t in (200, 500, 1000, 2000, 5000, 10000, 20000) if t <= sr / 2] + axs.set_yticks(tk); axs.set_yticklabels([f"{t/1000:g}k" if t >= 1000 else str(t) for t in tk]) + mark_onsets(axs, ons, "spec", a, b); axs.set_title("spectrogram (5.3 ms frames)", loc="left") + if r == len(wins) - 1: axw.set_xlabel("time (ms)"); axs.set_xlabel("time (ms)") + note = f" ({dropped} further window(s) not drawn)" if dropped else "" + fig.suptitle("Zoom on each onset" + note, fontsize=FONT + 4, x=0.01, ha="left") + fig.tight_layout(rect=(0, 0, 1, 0.97)); fig.savefig(path); plt.close(fig) + return True + +def plot_envelope(rep, env, path, other=None, labels=("", "")): + fig, ax = plt.subplots(figsize=(16, 6), dpi=100) + ax.plot(np.arange(len(env)), env, color="#1f4e9c", lw=2, label=labels[0] or "level") + if other is not None: ax.plot(np.arange(len(other[0])), other[0], color="#e67e22", lw=2, label=labels[1]) + lo = max(-100, min(env.min(), -80)); ax.set_ylim(lo, 3) + mark_onsets(ax, items(rep), "plain") + if other is not None: + for t in other[1]: ax.axvline(t, color="#e67e22", ls=":", lw=2) + ax.axhline(0, color="k", lw=1) + ax.set_xlabel("time (ms)"); ax.set_ylabel("level (dBFS, 1 ms envelope)") + ax.set_title("Level envelope (onsets dashed)\n" + title_for(rep), loc="left") + if other is not None or labels[0]: ax.legend(loc="lower right") + fig.tight_layout(); fig.savefig(path); plt.close(fig) + +# ---- text --------------------------------------------------------------------------------------------------------------- +def report_text(rep): + i = rep["info"] + L = [title_for(rep), f"loudness {i['loudness_lufs']} LUFS (mono, BS.1770 gated) rms {i['rms_dbfs']} dBFS channels {i['channels']} (mixed to mono)", "", + f"onsets ({len(rep['onsets'])}), 1 ms resolution:"] + L += [f" {o['ms']:6d} ms peak {o['peak_dbfs']:6.1f} dBFS at {o['peak_ms']} ms" for o in rep["onsets"]] or [" none"] + L += ["", f"transient (hf, above {TH['hf_cut_hz']:g} Hz) onsets ({len(rep['onsets_hf'])}):"] + L += [f" {o['ms']:6d} ms peak {o['peak_dbfs']:6.1f} dBFS (hf level) at {o['peak_ms']} ms" for o in rep["onsets_hf"]] or [" none"] + L += ["", "detectors (signal facts, not verdicts of good or bad):"] + L += [f" [{d['name']}] {d['verdict']}" for d in rep["detectors"]] + return "\n".join(L) + "\n" + +def write_single(path, outdir, quiet=False): + os.makedirs(outdir, exist_ok=True) + rep, x, sr, env = analyze(path) + plot_spectrogram(rep, x, sr, os.path.join(outdir, "spectrogram.png")) + plot_zoom(rep, x, sr, os.path.join(outdir, "zoom.png")) + plot_envelope(rep, env, os.path.join(outdir, "envelope.png")) + json.dump(rep, open(os.path.join(outdir, "report.json"), "w"), indent=1, default=float) + json.dump(rep["onsets"] + rep["onsets_hf"], open(os.path.join(outdir, "onsets.json"), "w"), indent=1) + open(os.path.join(outdir, "onsets.txt"), "w").write("".join(f"{o['ms']} ms\t{o['band']}\tpeak {o['peak_dbfs']} dBFS\n" for o in sorted(rep["onsets"] + rep["onsets_hf"], key=lambda o: o["ms"]))) + txt = report_text(rep) + open(os.path.join(outdir, "report.txt"), "w").write(txt) + return rep, x, sr, env, txt + +def match_onsets(a, b): + tol, used, pairs, only_a = TH["match_ms"], set(), [], [] + for t in a: + c = [(abs(t - u), j, u) for j, u in enumerate(b) if j not in used and abs(t - u) <= tol] + if c: _, j, u = min(c); used.add(j); pairs.append((t, u)) + else: only_a.append(t) + return pairs, only_a, [u for j, u in enumerate(b) if j not in used] + +def compare(pa, pb, outdir): + os.makedirs(outdir, exist_ok=True) + ra, xa, sra, ea, _ = write_single(pa, os.path.join(outdir, "a")) + rb, xb, srb, eb, _ = write_single(pb, os.path.join(outdir, "b")) + oa, ob = [o["ms"] for o in ra["onsets"]], [o["ms"] for o in rb["onsets"]] + pairs, only_a, only_b = match_onsets(oa, ob) + ia, ib = ra["info"], rb["info"] + ha, hb = [o["ms"] for o in ra["onsets_hf"]], [o["ms"] for o in rb["onsets_hf"]] + hpairs, honly_a, honly_b = match_onsets(ha, hb) + deltas = {k: (None if ia[k] is None or ib[k] is None else round(ib[k] - ia[k], 2)) for k in ("sample_peak_dbfs", "true_peak_dbtp", "loudness_lufs", "rms_dbfs", "duration_ms")} + ddiff = [dict(name=da["name"], a=da["fired"], b=db_["fired"], changed=da["fired"] != db_["fired"], a_verdict=da["verdict"], b_verdict=db_["verdict"]) + for da, db_ in zip(ra["detectors"], rb["detectors"])] + cmp = dict(a=ia["file"], b=ib["file"], onsets_matched=[dict(a_ms=p, b_ms=q, shift_ms=q - p) for p, q in pairs], onsets_only_in_a=only_a, onsets_only_in_b=only_b, + hf_onsets_matched=[dict(a_ms=p, b_ms=q, shift_ms=q - p) for p, q in hpairs], hf_onsets_only_in_a=honly_a, hf_onsets_only_in_b=honly_b, + deltas_b_minus_a=deltas, detector_diff=ddiff) + json.dump(cmp, open(os.path.join(outdir, "compare.json"), "w"), indent=1) + L = [f"A = {ia['file']} B = {ib['file']}", "", + f"onsets A {oa}", f"onsets B {ob}", + f"matched (within {TH['match_ms']:g} ms): " + (", ".join(f"{p}->{q}" for p, q in pairs) or "none"), + f"only in A: {only_a} only in B: {only_b}", "", + f"transient (hf) onsets A {ha}", f"transient (hf) onsets B {hb}", + f"hf matched: " + (", ".join(f"{p}->{q}" for p, q in hpairs) or "none") + f" only in A: {honly_a} only in B: {honly_b}", "", + "deltas B-A: " + " ".join(f"{k} {v:+}" if v is not None else f"{k} n/a" for k, v in deltas.items()), "", "detector diff:"] + for d in ddiff: + L.append(f" [{d['name']}] A {'FIRES' if d['a'] else 'clear'} | B {'FIRES' if d['b'] else 'clear'}" + (" <-- CHANGED" if d["changed"] else "")) + L.append(f" A: {d['a_verdict']}"); L.append(f" B: {d['b_verdict']}") + open(os.path.join(outdir, "compare.txt"), "w").write("\n".join(L) + "\n") + # picture: stacked spectrograms on one time axis, envelopes overlaid + dmax = max(len(xa) / sra, len(xb) / srb) * 1000 + fig, axes = plt.subplots(3, 1, figsize=(16, 15), dpi=100, gridspec_kw=dict(height_ratios=[3, 3, 2.4])) + for ax, x, sr, r, nm in ((axes[0], xa, sra, ra, "A"), (axes[1], xb, srb, rb, "B")): + m = draw_spec(ax, x, sr, 1024, 64) + mark_onsets(ax, items(r)); ax.set_xlim(0, dmax) + ax.set_title(f"{nm}: {title_for(r)}", loc="left") + axes[1].set_xlabel("time (ms)") + e = axes[2] + e.plot(np.arange(len(ea)), ea, color="#1f4e9c", lw=2, label="A " + ia["file"]); e.plot(np.arange(len(eb)), eb, color="#e67e22", lw=2, label="B " + ib["file"]) + for t in oa: e.axvline(t, color="#1f4e9c", ls="--", lw=1.6) + for t in ob: e.axvline(t, color="#e67e22", ls=":", lw=2.2) + e.set_xlim(0, dmax); e.set_ylim(-100, 3); e.set_xlabel("time (ms)"); e.set_ylabel("level (dBFS)"); e.legend(loc="lower right") + e.set_title("Level envelopes (A dashed onsets, B dotted onsets)", loc="left") + fig.tight_layout(); fig.savefig(os.path.join(outdir, "compare.png")); plt.close(fig) + return ra, rb, "\n".join(L) + "\n" + +# ---- CLI ------------------------------------------------------------------------------------------------------------------------ +def main(argv=None): + p = argparse.ArgumentParser(prog="aqua_ear.py", description="Audio perception tool: spectrogram + onset zooms, onset list, level envelope and signal-fact detectors for one wav/ogg/flac, or a comparison of two. Measures; does not hear.") + sub = p.add_subparsers(dest="cmd", required=True) + a = sub.add_parser("analyze", help="analyze one file"); a.add_argument("file"); a.add_argument("--out", help="output dir (default ./aqua-ear-out/)") + c = sub.add_parser("compare", help="compare two files (B relative to A)"); c.add_argument("a"); c.add_argument("b"); c.add_argument("--out", help="output dir (default ./aqua-ear-out/compare)") + for s in (a, c): s.add_argument("--strict", action="store_true", help="exit 3 if any detector fires (default: findings never change the exit code)") + args = p.parse_args(argv) + try: + if args.cmd == "analyze": + out = args.out or os.path.join("aqua-ear-out", os.path.splitext(os.path.basename(args.file))[0]) + rep, _, _, _, txt = write_single(args.file, out); print(txt + f"wrote {out}/ (spectrogram.png zoom.png envelope.png onsets.json onsets.txt report.json report.txt)") + fired = any(d["fired"] for d in rep["detectors"]) + else: + out = args.out or os.path.join("aqua-ear-out", "compare") + ra, rb, txt = compare(args.a, args.b, out); print(txt + f"wrote {out}/ (compare.png compare.txt compare.json, a/ and b/ single reports)") + fired = any(d["fired"] for r in (ra, rb) for d in r["detectors"]) + except Exception as e: + print(f"aqua-ear: error: {e}", file=sys.stderr); return 2 + return 3 if args.strict and fired else 0 + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/aqua-ear/test_aqua_ear.py b/tools/aqua-ear/test_aqua_ear.py new file mode 100644 index 0000000..3bdba95 --- /dev/null +++ b/tools/aqua-ear/test_aqua_ear.py @@ -0,0 +1,164 @@ +"""Tests for aqua-ear on synthetic fixtures generated here (no committed binaries). Each detector has a case where it must fire and a +near-miss where it must not. Run: python -m unittest -v test_aqua_ear (from this directory, with the README's venv).""" +import contextlib, io, json, os, shutil, struct, subprocess, tempfile, unittest +import numpy as np +import soundfile as sf +import aqua_ear as ae + +SR = 48000 +rng = np.random.default_rng(7) + +def t_(sec): return np.arange(int(SR * sec)) / SR + +def note(f=440.0, amp=0.5, dur=0.6, tau=0.15, attack=0.01, fade=0.03): + """A decaying sine with a linear attack and (optionally) a closing fade to exact zero: the clean reference note.""" + t = t_(dur); x = amp * np.sin(2 * np.pi * f * t) * np.exp(-t / tau) + x *= np.minimum(1, t / attack) + if fade: n = int(fade * SR); x[-n:] *= np.linspace(1, 0, n) + return x + +def burst(x, at_ms, amp=0.6, ms=3): + n = int(ms * SR / 1000); i = int(at_ms * SR / 1000) + x[i:i + n] += amp * np.sign(rng.standard_normal(n)) * np.exp(-np.arange(n) / (n / 4)) + return x + +def rep_of(x, d, name): + p = os.path.join(d, name + ".wav"); sf.write(p, x.astype(np.float32), SR, subtype="FLOAT") + return ae.analyze(p)[0] + +def fired(rep, name): return {d["name"]: d["fired"] for d in rep["detectors"]}[name] + +def quiet(argv): + with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): return ae.main(argv) + +class T(unittest.TestCase): + @classmethod + def setUpClass(c): c.d = tempfile.mkdtemp(prefix="aqua-ear-test-") + @classmethod + def tearDownClass(c): shutil.rmtree(c.d, ignore_errors=True) + def rep(self, x, name="f"): return rep_of(x, self.d, name) + + # --- onsets + def test_onset_times_within_2ms(self): + x = np.zeros(int(SR * 0.5)); x[int(0.1 * SR):] += note(dur=0.4, attack=0.002, fade=0)[:len(x) - int(0.1 * SR)] + r = self.rep(x, "on"); self.assertEqual(len(r["onsets"]), 1); self.assertAlmostEqual(r["onsets"][0]["ms"], 100, delta=2) + def test_clean_note_all_clear(self): + r = self.rep(note(), "clean") + self.assertEqual(len(r["onsets"]), 1) + self.assertEqual([d["name"] for d in r["detectors"] if d["fired"]], []) + + # --- double hit / flam + def double(self, gap_ms): return burst(burst(note(amp=0.3, tau=0.4, dur=0.8), 100), 100 + gap_ms) + def test_flam_fires_25ms(self): + r = self.rep(self.double(25), "flam25"); self.assertTrue(fired(r, "double_hit")) + h = [p for p in next(d for d in r["detectors"] if d["name"] == "double_hit")["evidence"] if p["one_note"]][0] + self.assertAlmostEqual(h["gap_ms"], 25, delta=2) + def test_flam_near_miss_150ms_apart(self): + self.assertFalse(fired(self.rep(self.double(150), "flam150"), "double_hit")) + def test_flam_near_miss_separate_staccato_notes(self): + x = np.zeros(int(SR * 0.4)) + for at in (50, 90): # 40 ms apart, each decays 60 dB in under 20 ms: two notes, not one + n = note(dur=0.05, tau=0.003, attack=0.001, fade=0); i = int(at * SR / 1000); x[i:i + len(n)] += n + self.assertFalse(fired(self.rep(x, "stac"), "double_hit")) + def test_flam_near_miss_quiet_precursor(self): + x = np.zeros(int(SR * 0.6)) + for at, amp, tau, dur in ((88, 0.01, 0.01, 0.02), (100, 0.3, 0.3, 0.5)): # a -40 dBFS blip 12 ms before the real note: 30 dB apart, a precursor not a second hit + n = note(amp=amp, tau=tau, dur=dur, attack=0.001 if dur < 0.1 else 0.004, fade=0); i = int(at * SR / 1000); x[i:i + len(n)] += n[:len(x) - i] + r = self.rep(x, "pre"); d = next(d for d in r["detectors"] if d["name"] == "double_hit") + self.assertFalse(d["fired"]); self.assertTrue(any(p["one_note"] and not p["comparable"] for p in d["evidence"])) # it saw the pair and declined it + def test_hf_transient_channel_finds_a_tick_the_full_band_envelope_misses(self): + x = note(amp=0.3, tau=0.5, dur=0.8, f=300) + for at in (200, 225): x = burst(x, at, amp=0.02, ms=2) # clicks 20 dB under the tone: the broadband envelope barely moves + r = self.rep(x, "tick"); self.assertEqual(len(r["onsets"]), 1) + self.assertGreaterEqual(len(r["onsets_hf"]), 2); self.assertTrue(fired(r, "double_hit")) + def test_hf_near_miss_ticks_150ms_apart(self): + x = note(amp=0.3, tau=0.5, dur=0.8, f=300) + for at in (200, 350): x = burst(x, at, amp=0.02, ms=2) + r = self.rep(x, "tick150"); self.assertGreaterEqual(len(r["onsets_hf"]), 2); self.assertFalse(fired(r, "double_hit")) + + # --- click / step / DC + def test_dc_fires_and_near_miss(self): + self.assertTrue(fired(self.rep(note() + 0.05, "dc"), "click_step_dc")) + self.assertFalse(fired(self.rep(note() + 0.001, "dc_small"), "click_step_dc")) + def test_step_fires(self): + x = note(dur=0.6, tau=1.0, fade=0); x[int(0.3 * SR):] += 0.3 + d = next(d for d in self.rep(x, "step")["detectors"] if d["name"] == "click_step_dc") + ev = d["evidence"]["discontinuities"]; self.assertTrue(d["fired"]); self.assertEqual(ev[0]["kind"], "step"); self.assertAlmostEqual(ev[0]["ms"], 300, delta=1) + def test_click_fires(self): + x = note(); x[int(0.3 * SR)] += 0.6 + ev = next(d for d in self.rep(x, "click")["detectors"] if d["name"] == "click_step_dc")["evidence"]["discontinuities"] + self.assertEqual(ev[0]["kind"], "click") + def test_discontinuity_near_miss_bright_continuous_tone(self): + x = 0.5 * np.sin(2 * np.pi * 8000 * t_(0.3)) * np.minimum(1, t_(0.3) / 0.01); x[-1440:] *= np.linspace(1, 0, 1440) + self.assertFalse(fired(self.rep(x, "bright"), "click_step_dc")) + + # --- clipping / true peak + def test_clipped_tone_fires(self): + x = np.clip(1.5 * np.sin(2 * np.pi * 440 * t_(0.3)), -1, 1) + d = next(d for d in self.rep(x, "clip")["detectors"] if d["name"] == "clipping_true_peak") + self.assertTrue(d["fired"]); self.assertGreater(d["evidence"]["plateau_runs"], 0) + def test_minus_1_dbfs_not_clipped(self): + r = self.rep(0.891 * np.sin(2 * np.pi * 440 * t_(0.3)), "m1"); d = next(d for d in r["detectors"] if d["name"] == "clipping_true_peak") + self.assertFalse(d["fired"]); self.assertAlmostEqual(d["evidence"]["sample_peak_dbfs"], -1.0, delta=0.05) + def test_plateau_below_the_ceiling_is_not_clipping(self): + x = np.clip(1.2 * np.sin(2 * np.pi * 440 * t_(0.3)), -0.8, 0.8) # flat tops, but at -1.9 dBFS: a limiter shape, not a converter over + d = next(d for d in self.rep(x, "lim")["detectors"] if d["name"] == "clipping_true_peak"); self.assertFalse(d["fired"]) + def test_intersample_over_fires_without_sample_over(self): + x = 1.3 * np.sin(2 * np.pi * (SR / 4) * t_(0.1) + np.pi / 4) # samples sit at +-0.919, the waveform between them reaches 1.3 + d = next(d for d in self.rep(x, "isp")["detectors"] if d["name"] == "clipping_true_peak") + self.assertTrue(d["fired"]); self.assertEqual(d["evidence"]["over_samples"], 0); self.assertAlmostEqual(d["evidence"]["true_peak_dbtp"], 20 * np.log10(1.3), delta=0.3) + + # --- overlapping tails + def pair(self, gap_ms, tau): + x = np.zeros(int(SR * 1.2)) + for at in (100, 100 + gap_ms): + n = note(dur=0.6, tau=tau, fade=0, attack=0.003); i = int(at * SR / 1000); x[i:i + len(n)] += n[:len(x) - i] + return x + def test_overlap_fires(self): + self.assertTrue(fired(self.rep(self.pair(150, 0.2), "ovl"), "overlapping_tails")) + def test_overlap_near_miss_short_tails(self): + self.assertFalse(fired(self.rep(self.pair(300, 0.015), "novl"), "overlapping_tails")) + + # --- abrupt end + def test_abrupt_end_fires(self): + x = note(tau=1.0, fade=0, dur=0.3) # still at about -9 dBFS when the file ends + self.assertTrue(fired(self.rep(x, "cut"), "abrupt_end")) + def test_abrupt_end_near_miss_faded_and_silent_tail(self): + self.assertFalse(fired(self.rep(note(tau=1.0, fade=0.03, dur=0.3), "fade"), "abrupt_end")) + x = np.concatenate([note(tau=1.0, fade=0, dur=0.3), np.zeros(4800)]); self.assertFalse(fired(self.rep(x, "sil"), "abrupt_end")) + self.assertFalse(fired(self.rep(note(tau=0.05, fade=0, dur=0.6), "decayed"), "abrupt_end")) + + # --- ports from aqua-feel and I/O + def test_loudness_and_true_peak_known_answers(self): + r = self.rep(0.5 * np.sin(2 * np.pi * 440 * t_(2)), "lk")["info"] + self.assertAlmostEqual(r["loudness_lufs"], -9.7, delta=0.2); self.assertAlmostEqual(r["sample_peak_dbfs"], -6.02, delta=0.05) + def test_ports_match_aqua_feel_metrics_js(self): + node = shutil.which("node"); js = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "aqua-feel", "metrics.js") + if not node or not os.path.exists(js): self.skipTest("node or aqua-feel/metrics.js absent") + x = (0.8 * np.sin(2 * np.pi * 997 * t_(1.0)) + 0.3 * rng.standard_normal(SR)).astype(np.float32) + raw = os.path.join(self.d, "x.f32"); x.tofile(raw) + code = f"import {{truePeak,integratedLoudness}} from 'file://{os.path.abspath(js)}';import fs from 'fs';const b=fs.readFileSync('{raw}');const x=new Float32Array(b.buffer,b.byteOffset,b.length/4);console.log(JSON.stringify([truePeak(x),integratedLoudness(x)]))" + tp, lu = json.loads(subprocess.run([node, "--input-type=module", "-e", code], capture_output=True, text=True, check=True).stdout) + mine_tp, _ = ae.true_peak(x.astype(np.float64), SR) + self.assertAlmostEqual(mine_tp, tp, delta=1e-5); self.assertAlmostEqual(ae.loudness_lufs(x.astype(np.float64), SR), lu, delta=0.01) + def test_formats_wav_ogg_flac(self): + x = note() + for ext in ("wav", "ogg", "flac"): + p = os.path.join(self.d, "fmt." + ext); sf.write(p, x, SR) + r = ae.analyze(p)[0]; self.assertEqual(len(r["onsets"]), 1, ext) + + # --- CLI and compare + def test_cli_exit_codes_and_outputs(self): + p = os.path.join(self.d, "dh.wav"); sf.write(p, self.double(25), SR); out = os.path.join(self.d, "o") + self.assertEqual(quiet(["analyze", p, "--out", out]), 0) # findings never fail the run + for f in ("spectrogram.png", "zoom.png", "envelope.png", "onsets.json", "onsets.txt", "report.json", "report.txt"): self.assertTrue(os.path.getsize(os.path.join(out, f)) > 0, f) + self.assertEqual(quiet(["analyze", p, "--out", out, "--strict"]), 3) + self.assertEqual(quiet(["analyze", os.path.join(self.d, "missing.wav"), "--out", out]), 2) + def test_compare_reports_the_difference(self): + a, b = os.path.join(self.d, "ca.wav"), os.path.join(self.d, "cb.wav"); sf.write(a, self.double(25), SR); sf.write(b, note(amp=0.3, tau=0.4, dur=0.8), SR) + out = os.path.join(self.d, "cmp"); self.assertEqual(quiet(["compare", a, b, "--out", out]), 0) + c = json.load(open(os.path.join(out, "compare.json"))) + self.assertTrue(next(d for d in c["detector_diff"] if d["name"] == "double_hit")["changed"]); self.assertTrue(len(c["onsets_only_in_a"]) >= 2) + self.assertTrue(os.path.getsize(os.path.join(out, "compare.png")) > 0) + +if __name__ == "__main__": unittest.main() diff --git a/tools/aqua-feel/README.md b/tools/aqua-feel/README.md index ede673a..c4c0e65 100644 --- a/tools/aqua-feel/README.md +++ b/tools/aqua-feel/README.md @@ -7,6 +7,8 @@ JSON and a static HTML gallery, and gates the numbers against thresholds. listener to check, not verdicts: a render can pass every threshold and sound wrong, and fail one and sound fine. The WAVs and the gallery exist so a person can press play. +To look at a finished audio file (wav/ogg/flac: spectrogram, onsets at 1 ms, level envelope, click/clip/double-hit/overlap/abrupt-end detectors, two-file compare) rather than a rendered patch, use `../aqua-ear`. + ## Run ``` -- 2.47.3