ISR/tests/test_web.py

"""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.command = 'GET'          # _send skips the body on HEAD
        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()