admin
f6031cfa16
A section's score is now its peak dB above the noise floor capped by the sharpest rise within ONSET_SECONDS (0.5 s). Real events (voices, impacts, barks) rise fast and keep their full prominence; a gradual swell that outruns the 30 s floor blocks (gusts, distant approaching cars) still flags but scores near zero, so score-ranked review (chips, U/I highlights, "Highlights only" mode) surfaces events first. A section starting in a file's first 0.5 s is scored against the floor instead, so events cut off by a file split are not punished as swells. Old cached analyses carry now-wrong scores, so the cache gains a leading "detector" version key (DETECTOR_VERSION = 2) checked by both _cached_analysis_params() and the /api/analyze cache hit path; v1 caches never match and are recomputed on the next analyse. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
163 lines
6.1 KiB
Python
163 lines
6.1 KiB
Python
"""Tests for the adaptive loud-section detector in web.py."""
|
||
|
||
import math
|
||
|
||
from web import _cut_filename, _loud_sections, _noise_floor_db, _recording_start
|
||
|
||
WINDOW_DUR = 0.1 # 100 ms windows, as produced by WINDOW_SAMPLES at 48 kHz
|
||
|
||
|
||
def _run(rms, margin_db=12.0, min_gap=2.0, min_duration=0.5):
|
||
duration = len(rms) * WINDOW_DUR
|
||
return _loud_sections(rms, WINDOW_DUR, duration, margin_db, min_gap, min_duration)
|
||
|
||
|
||
def test_burst_above_quiet_floor_is_detected():
|
||
rms = [0.002] * 1200 # 2 min of quiet ambience (−54 dBFS)
|
||
rms[600:610] = [0.05] * 10 # 1 s burst at −26 dBFS (+28 dB)
|
||
sections = _run(rms)
|
||
assert len(sections) == 1
|
||
s = sections[0]
|
||
assert s['start'] == 60.0
|
||
assert 60.5 <= s['end'] <= 62.0
|
||
assert 26.0 <= s['score'] <= 30.0
|
||
|
||
|
||
def test_slow_ambience_swell_is_not_detected():
|
||
# Level rises 20 dB over 10 minutes and back down — e.g. rain setting in.
|
||
# The old fixed threshold would have flagged the entire loud half.
|
||
up = [0.002 * 10 ** (i / 6000 * 1.0) for i in range(6000)] # −54 → −34 dB
|
||
down = list(reversed(up))
|
||
assert _run(up + down) == []
|
||
|
||
|
||
def test_burst_still_detected_on_loud_ambience():
|
||
# Same +28 dB prominence as the quiet test, but on a 20 dB louder floor.
|
||
rms = [0.02] * 1200
|
||
rms[600:610] = [0.5] * 10
|
||
sections = _run(rms)
|
||
assert len(sections) == 1
|
||
assert sections[0]['start'] == 60.0
|
||
|
||
|
||
def test_min_rms_floor_suppresses_blips_in_digital_silence():
|
||
rms = [0.000001] * 1200
|
||
rms[600:605] = [0.004] * 5 # −48 dBFS: audible blip, but below MIN_RMS+12dB
|
||
assert _run(rms) == []
|
||
rms[600:605] = [0.05] * 5 # −26 dBFS clearly clears the clamped floor
|
||
assert len(_run(rms)) == 1
|
||
|
||
|
||
def test_min_gap_merges_nearby_bursts():
|
||
rms = [0.002] * 1200
|
||
rms[600:605] = [0.05] * 5
|
||
rms[615:620] = [0.05] * 5 # 1 s gap < 2 s min_gap → one section
|
||
rms[900:905] = [0.05] * 5 # 28 s away → separate section
|
||
sections = _run(rms)
|
||
assert len(sections) == 2
|
||
assert sections[0]['start'] == 60.0
|
||
assert sections[0]['end'] >= 61.5
|
||
assert sections[1]['start'] == 90.0
|
||
|
||
|
||
def test_min_duration_drops_subsecond_blips():
|
||
# Isolated single-window pops (clicks, single raindrops) spaced wider than
|
||
# min_gap must not each become their own section — this is what used to
|
||
# produce thousands of zero-length sections per day.
|
||
rms = [0.002] * 1200
|
||
for i in range(600, 660, 30): # 0.1 s blips, 3 s apart (> min_gap)
|
||
rms[i] = 0.05
|
||
assert _run(rms) == []
|
||
# With the filter disabled they are all reported
|
||
assert len(_run(rms, min_duration=0.0)) == 2
|
||
|
||
|
||
def test_min_duration_keeps_sections_at_or_above_it():
|
||
rms = [0.002] * 1200
|
||
rms[600:605] = [0.05] * 5 # exactly 0.5 s
|
||
sections = _run(rms, min_duration=0.5)
|
||
assert len(sections) == 1
|
||
assert sections[0]['start'] == 60.0
|
||
|
||
|
||
def test_min_duration_applies_after_gap_merging():
|
||
# Two sub-min_duration blips within min_gap merge into one section whose
|
||
# loud span exceeds min_duration — the merged section must survive.
|
||
rms = [0.002] * 1200
|
||
rms[600] = 0.05
|
||
rms[610] = 0.05 # 1 s apart < 2 s min_gap → merged, 1.1 s span
|
||
sections = _run(rms, min_duration=1.0)
|
||
assert len(sections) == 1
|
||
assert sections[0]['start'] == 60.0
|
||
assert sections[0]['end'] >= 61.0
|
||
|
||
|
||
def test_slow_swell_scores_near_zero_sharp_burst_scores_high():
|
||
# A ~10 s swell rises faster than the 30 s floor blocks can track, so it
|
||
# still flags — but its score must collapse (onset cap) so it ranks at the
|
||
# bottom, while a sharp burst with the same peak keeps its full score.
|
||
rms = [0.002] * 1200
|
||
for k in range(80): # rise −54 → −28 dB over 8 s
|
||
rms[400 + k] = 0.002 * 10 ** (26 * (k / 80) / 20)
|
||
rms[480:500] = [0.04] * 20 # hold 2 s at −28 dB
|
||
for k in range(80): # fall back over 8 s
|
||
rms[500 + k] = 0.04 * 10 ** (-26 * (k / 80) / 20)
|
||
rms[900:910] = [0.04] * 10 # sharp 1 s burst, same peak
|
||
sections = _run(rms)
|
||
assert len(sections) == 2
|
||
swell, burst = sections
|
||
assert swell['start'] < 50.0 < 90.0 <= burst['start']
|
||
assert burst['score'] >= 24.0 # ≈ full 26 dB prominence
|
||
assert swell['score'] <= 5.0 # capped by its slow onset
|
||
assert swell['score'] < burst['score']
|
||
|
||
|
||
def test_burst_at_file_start_keeps_full_score():
|
||
# No pre-event history to measure a rise against: the onset cap falls back
|
||
# to prominence above the floor, so events cut off by a file split are not
|
||
# punished as swells.
|
||
rms = [0.05] * 10 + [0.002] * 1190
|
||
sections = _run(rms)
|
||
assert len(sections) == 1
|
||
assert sections[0]['start'] == 0.0
|
||
assert sections[0]['score'] >= 24.0
|
||
|
||
|
||
def test_noise_floor_tracks_blocks_and_ignores_short_events():
|
||
quiet_db = 20 * math.log10(0.002)
|
||
db = [quiet_db] * 1200
|
||
db[600:650] = [-20.0] * 50 # 5 s event must not raise its own floor
|
||
floor = _noise_floor_db(db, WINDOW_DUR)
|
||
assert len(floor) == len(db)
|
||
assert all(abs(f - quiet_db) < 1.0 for f in floor)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Filename parsing / cut naming
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_recording_start_parses_standard_name():
|
||
from datetime import datetime
|
||
assert _recording_start("20260523_220000") == datetime(2026, 5, 23, 22, 0, 0)
|
||
|
||
|
||
def test_recording_start_rejects_nonstandard_name():
|
||
assert _recording_start("radio1_20260523") is None
|
||
assert _recording_start("notes") is None
|
||
|
||
|
||
def test_cut_filename_uses_wall_clock_span():
|
||
# Recording started 22:00:00; cut covers 22:31:30 → 22:32:30.
|
||
name = _cut_filename("20260523_220000", ".flac", 1890, 1950)
|
||
assert name == "20260523_22-31-30_22-32-30.flac"
|
||
|
||
|
||
def test_cut_filename_rolls_over_the_hour():
|
||
name = _cut_filename("20260523_220000", ".wav", 3590, 3661)
|
||
assert name == "20260523_22-59-50_23-01-01.wav"
|
||
|
||
|
||
def test_cut_filename_falls_back_for_nonstandard_name():
|
||
name = _cut_filename("mixtape", ".mp3", 740, 750.4)
|
||
assert name == "mixtape_cut_740s-750s.mp3"
|