GameCult/AquaSynth
Watch
9
0
Fork
You've already forked AquaSynth
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

aqua-ear: audio perception tool (spectrogram, onsets, detectors, compare) #24

Merged
aqua merged 1 commit from aqua/aqua-ear into main 2026-11-10 17:36:55 +00:00
Conversation 0 Commits 1 Files changed 5 +701
5 changed files with 701 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
tools/aqua-ear/.gitignore vendored Normal file
View file

@ -0,0 +1 @@
__pycache__/

83
tools/aqua-ear/README.md Normal file
View file

@ -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.

451
tools/aqua-ear/aqua_ear.py Normal file
View file

@ -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/<name>)")
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())

164
tools/aqua-ear/test_aqua_ear.py Normal file
View file

@ -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()

2
tools/aqua-feel/README.md
View file

@ -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
```