adminandClaude Opus 5
b661e166ed
_compute_rms_windows_wav honours sampwidth for frame arithmetic but decodes as int16 unconditionally, so an 8- or 24-bit WAV produced meaningless RMS values and therefore meaningless sections, silently. The recorder only writes 16-bit, but --dir points at arbitrary directories. analyze_wav now refuses anything else, turning a silent wrong answer into a visible one. _loud_sections closed mid-file sections at last_loud_t + window_dur but the trailing one at the file duration. A file whose last loud event is followed by less than min_gap of quiet reported a section stretched by up to min_gap seconds, and the clip served for it carried that silence. Now capped at the last loud window, consistent with every other section and with the score, which only ever covered the loud windows. DETECTOR_VERSION 2 -> 3 so existing caches are recomputed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
273 lines
9.7 KiB
Python
273 lines
9.7 KiB
Python
"""Tests for the adaptive loud-section detector in web.py."""
|
||
|
||
import io
|
||
import math
|
||
import wave
|
||
|
||
import web
|
||
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_section_open_at_eof_ends_at_last_loud_window():
|
||
# Burst ends 1 s before EOF — less than min_gap, so the section is still
|
||
# open when the windows run out. It must end where the sound ended, not at
|
||
# the file duration; otherwise the clip carries a second of trailing
|
||
# silence and disagrees with every section closed mid-file.
|
||
rms = [0.002] * 1200
|
||
rms[1180:1190] = [0.05] * 10 # loud 118.0 → 119.0, file is 120.0 s
|
||
sections = _run(rms)
|
||
assert len(sections) == 1
|
||
assert sections[0]['start'] == 118.0
|
||
assert sections[0]['end'] == 119.0
|
||
|
||
|
||
def test_section_running_to_the_very_end_is_capped_at_duration():
|
||
rms = [0.002] * 1200
|
||
rms[1190:1200] = [0.05] * 10 # loud right up to EOF
|
||
sections = _run(rms)
|
||
assert len(sections) == 1
|
||
assert sections[0]['end'] == 120.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"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# /api/clip error handling
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class _OfflineHandler(web._Handler):
|
||
"""A _Handler with no socket, for driving handler methods directly."""
|
||
|
||
def __init__(self, recordings_dir):
|
||
self.recordings_dir = str(recordings_dir)
|
||
self.analyses_dir = str(recordings_dir)
|
||
self.wfile = io.BytesIO()
|
||
self.close_connection = False
|
||
self.codes = []
|
||
|
||
def send_response(self, code):
|
||
self.codes.append(code)
|
||
|
||
def send_header(self, key, value):
|
||
pass
|
||
|
||
def end_headers(self):
|
||
pass
|
||
|
||
|
||
def _write_wav(path, seconds=2, framerate=8000):
|
||
with wave.open(str(path), 'wb') as wf:
|
||
wf.setnchannels(1)
|
||
wf.setsampwidth(2)
|
||
wf.setframerate(framerate)
|
||
wf.writeframes(b'\x00\x00' * (seconds * framerate))
|
||
|
||
|
||
def test_non_16_bit_wav_is_rejected_rather_than_misread(tmp_path):
|
||
# RMS windowing decodes as int16 regardless of the declared sample width,
|
||
# so an 8-bit file used to produce meaningless sections with no warning.
|
||
path = tmp_path / 'eight_bit.wav'
|
||
with wave.open(str(path), 'wb') as wf:
|
||
wf.setnchannels(1)
|
||
wf.setsampwidth(1)
|
||
wf.setframerate(8000)
|
||
wf.writeframes(b'\x80' * 8000)
|
||
result = web.analyze_wav(path)
|
||
assert 'error' in result
|
||
assert '16-bit' in result['error']
|
||
|
||
|
||
def test_16_bit_wav_still_analyses(tmp_path):
|
||
path = tmp_path / 'ok.wav'
|
||
_write_wav(path)
|
||
assert 'error' not in web.analyze_wav(path)
|
||
|
||
|
||
def test_clip_failure_after_headers_closes_connection_instead_of_writing_json(tmp_path):
|
||
# Once the audio headers are out, an error body would be a second response
|
||
# on a keep-alive connection that already promised audio/wav plus an exact
|
||
# Content-Length — desyncing every later request on that connection.
|
||
_write_wav(tmp_path / 'rec.wav')
|
||
h = _OfflineHandler(tmp_path)
|
||
|
||
def boom(path, start, end):
|
||
h._send_clip_headers(b'HDR', 10, 1, 2)
|
||
raise RuntimeError('decode exploded')
|
||
|
||
h._clip_wav = boom
|
||
h._api_clip({'file': ['rec.wav'], 'start': ['0'], 'end': ['1']})
|
||
|
||
assert h.close_connection is True
|
||
assert h.codes == [200] # no second status line
|
||
assert b'error' not in h.wfile.getvalue() # no JSON body
|
||
|
||
|
||
def test_clip_failure_before_headers_still_reports_json(tmp_path):
|
||
_write_wav(tmp_path / 'rec.wav')
|
||
h = _OfflineHandler(tmp_path)
|
||
|
||
def boom(path, start, end):
|
||
raise RuntimeError('could not open')
|
||
|
||
h._clip_wav = boom
|
||
h._api_clip({'file': ['rec.wav'], 'start': ['0'], 'end': ['1']})
|
||
|
||
assert h.close_connection is False
|
||
assert h.codes == [500]
|
||
assert b'could not open' in h.wfile.getvalue()
|