diora-web/podcasts/silence.py

"""Skip silence in the browser: what to cut out of an episode's pauses, found once on the server.

Android shortens pauses while it plays (diora-android PauseCutter.kt). A browser
can't listen to the audio of most podcast servers (no CORS: Web Audio hears only
zeros), so the server reads the file once and keeps the result on the episode
(PodcastEpisode.silences); the browser then runs fast through those parts while
it plays (podcasts-web.js).

The rules are PauseCutter's, so a pause sounds the same everywhere: the audio
is judged in 10 ms windows against the recording itself — silence is well below
the recent speech level *and* close to the recording's noise floor, never above
-30 dB — and a pause of 90 ms or more keeps a fifth of its length (at least
60 ms, at most 250 ms), half before and half after the part that goes.
"""
import logging
import math
import os
import subprocess
import tempfile
import threading
import warnings

import requests

from django.db import close_old_connections

with warnings.catch_warnings():
    warnings.simplefilter('ignore', DeprecationWarning)
    try:
        import audioop  # the RMS of a window in C; gone from Python 3.13 (then: audioop-lts)
    except ImportError:  # pragma: no cover
        audioop = None

log = logging.getLogger(__name__)

VERSION = 2            # what's stored on the episode; an older result is found anew
RATE = 8000            # Hz, mono: enough to judge loudness
WINDOW = 0.01          # s
TIMEOUT = 900          # seconds for decoding one episode
MAX_BYTES = 800 * 1024 * 1024   # a larger file isn't an episode to analyse
MAX_RUNNING = 2        # episodes analysed at the same time

# PauseCutter.Params
MIN_PAUSE_MS = 90
KEEP_MIN_MS = 60
KEEP_RATIO = 0.2
KEEP_MAX_MS = 250
FADE_MS = 12
BELOW_SPEECH_DB = 20.0
ABOVE_NOISE_DB = 12.0
CEILING_DB = -30.0
FLOOR_DB = -70.0
DECAY_DB_PER_WINDOW = 6.0 * WINDOW   # 6 dB per second
NOISE_SHARE = 0.1
NOISE_WINDOWS = 1000                 # 10 s
NOISE_MIN_DB = -90.0
NOISE_MAX_DB = -35.0

_running = set()
_lock = threading.Lock()
_slots = threading.BoundedSemaphore(MAX_RUNNING)


class _NoiseHistory:
    """The last windows' levels in 1 dB steps, for a running percentile."""
    BINS = 121

    def __init__(self):
        self.ring = [0] * NOISE_WINDOWS
        self.bins = [0] * self.BINS
        self.count = 0
        self.at = 0

    def add(self, db):
        b = min(max(int(db + 120.0), 0), self.BINS - 1)
        if self.count == NOISE_WINDOWS:
            self.bins[self.ring[self.at]] -= 1
        else:
            self.count += 1
        self.ring[self.at] = b
        self.bins[b] += 1
        self.at = (self.at + 1) % NOISE_WINDOWS

    def level(self):
        if self.count < NOISE_WINDOWS // 10:
            return -60.0  # not enough seen yet
        target = max(int(self.count * NOISE_SHARE), 1)
        seen = 0
        for b, n in enumerate(self.bins):
            seen += n
            if seen >= target:
                return b - 120.0
        return 0.0


def _cut_of(start, end):
    """The part of the pause [start, end) (window indexes) that goes, in seconds, or None."""
    p = round((end - start) * WINDOW * 1000)   # ms
    if p < MIN_PAUSE_MS:
        return None
    keep = min(max(round(p * KEEP_RATIO), KEEP_MIN_MS), KEEP_MAX_MS, p)
    if p - keep < FADE_MS:
        return None
    head = keep // 2
    tail = keep - head
    return [round(start * WINDOW + head / 1000, 3), round(end * WINDOW - tail / 1000, 3)]


def cuts(levels):
    """[[from, to], …] in seconds: what to leave out, for 10 ms window levels in dB."""
    level = -20.0                  # recent speech level
    noise = -60.0                  # noise floor
    history = _NoiseHistory()
    out = []
    start = None
    for i, db in enumerate(levels):
        threshold = max(min(level - BELOW_SPEECH_DB, noise + ABOVE_NOISE_DB, CEILING_DB), FLOOR_DB)
        silent = db < threshold
        level = db if db > level else max(FLOOR_DB, level - DECAY_DB_PER_WINDOW)
        history.add(db)
        noise = min(max(history.level(), NOISE_MIN_DB), NOISE_MAX_DB)
        if silent:
            if start is None:
                start = i
        elif start is not None:
            cut = _cut_of(start, i)
            if cut:
                out.append(cut)
            start = None
    return out  # a pause running to the end stays as it is


def _level_db(window):
    if audioop is not None:
        rms = audioop.rms(window, 2)
    else:  # pragma: no cover
        samples = memoryview(window).cast('h')
        rms = math.sqrt(sum(s * s for s in samples) / len(samples))
    return -120.0 if rms < 1 else 20 * math.log10(rms / 32768)


def levels_of(path):
    """10 ms window levels (dB full scale) of the audio file at `path`."""
    import imageio_ffmpeg
    proc = subprocess.Popen(
        [imageio_ffmpeg.get_ffmpeg_exe(), '-nostdin', '-hide_banner', '-loglevel', 'error',
         '-vn', '-sn', '-dn', '-i', path, '-ac', '1', '-ar', str(RATE), '-f', 's16le', '-'],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    size = int(RATE * WINDOW) * 2
    levels = []
    rest = b''
    timer = threading.Timer(TIMEOUT, proc.kill)
    timer.start()
    try:
        while True:
            chunk = proc.stdout.read(1 << 20)
            if not chunk:
                break
            data = rest + chunk
            whole = len(data) // size * size
            for i in range(0, whole, size):
                levels.append(_level_db(data[i:i + size]))
            rest = data[whole:]
        err = proc.stderr.read().decode('utf-8', 'replace')
        if proc.wait() != 0:
            raise RuntimeError((err.strip().splitlines()[-1:] or [f'ffmpeg failed ({proc.returncode})'])[0][:200])
    finally:
        timer.cancel()
    return levels


def analyze(url):
    """What to cut out of the audio at `url`. Downloaded here first: the static ffmpeg
    build can't look up host names (it crashes doing so), it reads a local file."""
    fd, path = tempfile.mkstemp(suffix='.audio')
    try:
        with os.fdopen(fd, 'wb') as out, requests.get(
                url, stream=True, timeout=(10, 60), headers={'User-Agent': 'diora podcast silence'}) as r:
            if r.status_code >= 400:
                raise RuntimeError(f'HTTP {r.status_code}')
            size = 0
            for chunk in r.iter_content(1 << 20):
                size += len(chunk)
                if size > MAX_BYTES:
                    raise RuntimeError('file too large')
                out.write(chunk)
        return cuts(levels_of(path))
    finally:
        try:
            os.remove(path)
        except OSError:
            pass


def current(silences):
    """Whether a stored result is from these rules (else it's found anew)."""
    return bool(silences) and silences.get('v') == VERSION


def ensure(episode_id):
    """Find the episode's cuts in the background, unless that runs already."""
    with _lock:
        if episode_id in _running:
            return
        _running.add(episode_id)
    threading.Thread(target=_work, args=(episode_id,), daemon=True).start()


def _work(episode_id):
    from .models import PodcastEpisode
    try:
        with _slots:
            close_old_connections()
            ep = PodcastEpisode.objects.filter(pk=episode_id).only('id', 'audio_url').first()
            if ep is None:
                return
            try:
                result = {'v': VERSION, 'spans': analyze(ep.audio_url)}
            except Exception as e:  # unreachable file, timeout, not audio: remembered, not retried
                log.warning('silences of episode %s failed: %s', episode_id, e)
                result = {'v': VERSION, 'spans': [], 'error': str(e)[:200] or type(e).__name__}
            PodcastEpisode.objects.filter(pk=episode_id).update(silences=result)
    finally:
        with _lock:
            _running.discard(episode_id)
        close_old_connections()