adminandClaude Opus 5
a836948882
list_files sorted on mtime while `date` — and so the DATE column, the day grouping and both date filters — came from _recording_start(path.stem). The two disagree for any file whose mtime drifted from its name: a copy, a restore, an ffmpeg round-trip. Such a file landed in the right day group but at the wrong position inside it, and could drag its whole day group to the wrong place on the page. 'date' is fixed-width zero-padded, so sorting the string is already chronological — no extra field needed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
286 lines
10 KiB
Python
286 lines
10 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_listing_is_ordered_by_the_filename_clock_not_mtime(tmp_path):
|
||
# mtime is the last write (≈ the end of the recording) and drifts from the
|
||
# name on any copy or round-trip. Here the older recording is touched last.
|
||
import os
|
||
early, late = tmp_path / '20260523_080000.wav', tmp_path / '20260523_090000.wav'
|
||
_write_wav(early)
|
||
_write_wav(late)
|
||
os.utime(late, (1000, 1000)) # newest recording, oldest mtime
|
||
|
||
names = [f['name'] for f in web.list_files(str(tmp_path))]
|
||
assert names == ['20260523_090000.wav', '20260523_080000.wav']
|
||
|
||
|
||
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()
|