ISR/isrweb/loudness.py

"""Adaptive loud-section detection.

Pure computation: RMS windows -> a rolling noise floor -> scored sections.
Nothing here knows about HTTP, caching or the archive on disk, and nothing
here imports the rest of the app — so the detector can be worked on, and
tested, on its own. That matters because this is the part most likely to
change: if flagging (rather than ranking) ever needs improving, the next
step is a spectral filter or an optional Silero VAD over candidate sections,
and it all lands in this file.

**Detection is adaptive — do not regress it to an absolute threshold.** A
fixed RMS threshold flagged every ambience change (passing cars, rain) and
produced ~600 useless sections a day; that is why it was replaced.
"""

import math
import struct
import wave

try:
    import numpy as np
    NUMPY_AVAILABLE = True
except ImportError:
    NUMPY_AVAILABLE = False

try:
    import soundfile as sf
    SOUNDFILE_AVAILABLE = True
except ImportError:
    SOUNDFILE_AVAILABLE = False

# ---------------------------------------------------------------------------
# Tuning
# ---------------------------------------------------------------------------

WINDOW_SAMPLES   = 4800    # 100 ms at 48 kHz
MARGIN_DB        = 12.0    # sections must rise this many dB above the noise floor
MIN_GAP_SECONDS  = 2.0     # merge loud sections separated by less than this
MIN_DURATION_SECONDS = 0.5 # discard loud sections shorter than this

NOISE_BLOCK_SECONDS = 30.0 # noise floor is estimated per block of this length
NOISE_PERCENTILE    = 20   # percentile of windowed dB levels taken as the floor
MIN_RMS             = 0.002  # ≈ −54 dBFS; the floor never drops below this, so
                             # digital silence does not make every tiny sound loud
ONSET_SECONDS       = 0.5  # a section's score is capped by its sharpest dB rise
                           # within this span, so slow swells rank low

# Bumped whenever section detection/scoring changes in a way that makes old
# cached results wrong (not just differently parameterised). Caches written
# with another version never match and are recomputed on the next analyse.
DETECTOR_VERSION = 3   # v3: a section open at EOF ends at its last loud
                       # window, not at the file duration

def _compute_rms_windows_wav(wf, channels: int, sampwidth: int, framerate: int,
                              window_samples: int):
    """Yield rms_0_to_1 for every window in the open wave file."""
    while True:
        raw = wf.readframes(window_samples)
        if not raw:
            break
        n_samp = len(raw) // (sampwidth * channels)
        if n_samp == 0:
            break

        if NUMPY_AVAILABLE:
            arr = np.frombuffer(raw[:n_samp * sampwidth * channels], dtype='<i2')
            if channels > 1:
                arr = arr.reshape(-1, channels).mean(axis=1)
            rms = float(np.sqrt(np.mean(arr.astype(np.float64) ** 2))) / 32768.0
        else:
            fmt = f'<{n_samp * channels}h'
            samples = struct.unpack(fmt, raw[:n_samp * sampwidth * channels])
            mono = samples[::channels] if channels > 1 else samples
            rms = math.sqrt(sum(s * s for s in mono) / len(mono)) / 32768.0

        yield round(rms, 5)


def _noise_floor_db(db_values: list, window_dur: float) -> list:
    """Per-window background noise floor in dBFS.

    The floor is the NOISE_PERCENTILE-th percentile of the windowed dB levels
    in each NOISE_BLOCK_SECONDS block, then min-smoothed over ±2 neighbouring
    blocks so an event spanning a whole block cannot raise its own floor.
    Tracks slow ambience changes (day/night, rain, traffic hum) so detection
    is relative to "how loud it normally is right now"."""
    n = len(db_values)
    block = max(1, int(round(NOISE_BLOCK_SECONDS / window_dur)))
    floors = []
    for i in range(0, n, block):
        chunk = sorted(db_values[i:i + block])
        floors.append(chunk[int(len(chunk) * NOISE_PERCENTILE / 100)])
    smoothed = [min(floors[max(0, b - 2):b + 3]) for b in range(len(floors))]
    return [smoothed[min(i // block, len(smoothed) - 1)] for i in range(n)]


def _loud_sections(rms_values: list, window_dur: float, duration: float,
                   margin_db: float, min_gap: float = MIN_GAP_SECONDS,
                   min_duration: float = MIN_DURATION_SECONDS) -> list:
    """Sections whose level rises at least margin_db above the local noise
    floor. Each section carries a 'score': its peak dB above the floor,
    capped by the sharpest rise observed within ONSET_SECONDS. Real events
    (voices, impacts, barks) have steep onsets, so their cap equals their
    peak; a swell that drifts up slower than the noise-floor blocks can track
    (wind, a distant approaching car) still flags but scores near zero, so
    score-ranked review (chips, U/I highlights) surfaces events first.

    Sections shorter than min_duration (after min_gap merging) are discarded:
    without this, every isolated 100 ms window that pops above the floor — a
    click, a single raindrop — becomes its own section and a day can drown in
    thousands of sub-second clips."""
    db = [20 * math.log10(max(r, 1e-6)) for r in rms_values]
    floor = _noise_floor_db(db, window_dur)
    min_db = 20 * math.log10(MIN_RMS)
    onset_win = max(1, int(round(ONSET_SECONDS / window_dur)))

    sections = []
    start_t = None
    last_loud_t = None
    peak = 0.0
    onset = 0.0

    def _score():
        return round(max(0.0, min(peak, onset)), 1)

    for i, d in enumerate(db):
        t = i * window_dur
        floor_eff = max(floor[i], min_db)
        if d >= floor_eff + margin_db:
            if start_t is None:
                start_t = t
                peak = 0.0
                onset = 0.0
            last_loud_t = t
            peak = max(peak, d - floor_eff)
            # Rise within the onset span; a section starting before the file
            # has history is measured against the floor instead (an event cut
            # off by a file split must not be punished as a swell).
            rise = d - db[i - onset_win] if i >= onset_win else d - floor_eff
            onset = max(onset, rise)
        else:
            if start_t is not None and (t - last_loud_t) > min_gap:
                end_t = last_loud_t + window_dur
                if end_t - start_t >= min_duration - 1e-9:
                    sections.append({'start': round(start_t, 1),
                                     'end':   round(end_t, 1),
                                     'score': _score()})
                start_t = None
                last_loud_t = None

    # A section still open at EOF ends at its last loud window, exactly like
    # every section closed inside the loop — not at the file duration, which
    # used to stretch it by up to min_gap of trailing silence and put that
    # silence into the clip served for it.
    if start_t is not None and (last_loud_t + window_dur - start_t) >= min_duration - 1e-9:
        sections.append({'start': round(start_t, 1),
                         'end':   round(min(last_loud_t + window_dur, duration), 1),
                         'score': _score()})

    return sections


def _package_result(rms_values: list, framerate: int, n_frames: int,
                    window_samples: int, margin_db: float,
                    min_gap: float = MIN_GAP_SECONDS,
                    min_duration: float = MIN_DURATION_SECONDS) -> dict:
    window_dur = window_samples / framerate
    duration   = n_frames / framerate

    # Note: no RMS data is returned — the UI is screen-reader oriented and
    # draws no waveform, so sections + duration is all it needs. The window
    # length went the same way: it only ever existed to place RMS bars.
    return {
        'sections':    _loud_sections(rms_values, window_dur, duration, margin_db, min_gap, min_duration),
        'duration':    round(duration, 2),
    }


def analyze_wav(path: Path, window_samples: int = WINDOW_SAMPLES,
                margin_db: float = MARGIN_DB,
                min_gap: float = MIN_GAP_SECONDS,
                min_duration: float = MIN_DURATION_SECONDS) -> dict:
    try:
        with wave.open(str(path), 'rb') as wf:
            channels  = wf.getnchannels()
            sampwidth = wf.getsampwidth()
            framerate = wf.getframerate()
            n_frames  = wf.getnframes()
            # _compute_rms_windows_wav decodes as int16 unconditionally, so an
            # 8- or 24-bit file would yield garbage RMS and therefore garbage
            # sections — silently. The recorder only writes 16-bit, but --dir
            # can point at any directory.
            if sampwidth != 2:
                return {'error': f'only 16-bit WAV files can be analysed '
                                 f'(this one is {sampwidth * 8}-bit)'}
            rms_values = list(_compute_rms_windows_wav(
                wf, channels, sampwidth, framerate, window_samples))
    except Exception as e:
        return {'error': str(e)}

    return _package_result(rms_values, framerate, n_frames, window_samples, margin_db, min_gap, min_duration)


def analyze_flac(path: Path, window_samples: int = WINDOW_SAMPLES,
                 margin_db: float = MARGIN_DB,
                 min_gap: float = MIN_GAP_SECONDS,
                 min_duration: float = MIN_DURATION_SECONDS) -> dict:
    """Analyse a FLAC file for loudness. Requires numpy and soundfile."""
    if not NUMPY_AVAILABLE or not SOUNDFILE_AVAILABLE:
        return {'error': 'FLAC analysis requires: pip install numpy soundfile'}

    try:
        with sf.SoundFile(path) as f:
            framerate = f.samplerate
            channels  = f.channels
            n_frames  = len(f)

            rms_values = []
            while True:
                frames = f.read(window_samples, dtype='float32', always_2d=True)
                if len(frames) == 0:
                    break
                mono = frames.mean(axis=1) if channels > 1 else frames[:, 0]
                rms  = float(np.sqrt(np.mean(mono.astype(np.float64) ** 2)))
                rms_values.append(round(rms, 5))
    except Exception as e:
        return {'error': str(e)}

    return _package_result(rms_values, framerate, n_frames, window_samples, margin_db, min_gap, min_duration)