diff --git a/backend/services/longform_render.py b/backend/services/longform_render.py index 380c0fbb..0835b656 100644 --- a/backend/services/longform_render.py +++ b/backend/services/longform_render.py @@ -28,6 +28,7 @@ from __future__ import annotations import hashlib import json +import math import os import re from dataclasses import dataclass @@ -168,6 +169,105 @@ def build_loudnorm_filter(preset: Optional[str]) -> Optional[str]: return f"loudnorm=I={p.i}:TP={p.tp}:LRA={p.lra}" +@dataclass(frozen=True) +class MeasuredLoudness: + """The five loudnorm measure-pass values (FFmpeg JSON keys), all finite + floats. Fed back into the second (apply) pass as ``measured_*`` + ``offset``.""" + input_i: float + input_tp: float + input_lra: float + input_thresh: float + target_offset: float + + +def build_loudnorm_measure_filter(preset: Optional[str]) -> Optional[str]: + """First-pass loudnorm filter (``print_format=json``) for ``preset``, or + ``None`` for off/unknown — mirrors :func:`build_loudnorm_filter`'s lookup + (no whitespace stripping) so the same values count as 'no filter'.""" + if not preset: + return None + p = LOUDNESS_PRESETS.get(preset.lower()) + if p is None: + return None + return f"loudnorm=I={p.i}:TP={p.tp}:LRA={p.lra}:print_format=json" + + +def parse_loudnorm_measure(stderr_text: Optional[str]) -> Optional[MeasuredLoudness]: + """Extract the loudnorm measure JSON from ffmpeg stderr → MeasuredLoudness, + or ``None`` on ANY failure (caller falls back to single-pass). FFmpeg prints + the JSON object amid other non-JSON lines (and possibly a config dump block), + so we take the LAST balanced ``{...}`` via a linear brace-depth scan — no + regex (CodeQL-safe), O(n), no backtracking — then json.loads + coerce/validate + the five required keys to finite floats.""" + if not stderr_text: + return None + # Find the last balanced top-level {...} block via a single linear scan. + start = -1 + depth = 0 + block = None + for i, ch in enumerate(stderr_text): + if ch == "{": + if depth == 0: + start = i + depth += 1 + elif ch == "}": + if depth > 0: + depth -= 1 + if depth == 0 and start != -1: + block = stderr_text[start:i + 1] # keep scanning → last wins + if block is None: + return None + try: + obj = json.loads(block) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(obj, dict): + return None + keys = ("input_i", "input_tp", "input_lra", "input_thresh", "target_offset") + vals = {} + for k in keys: + if k not in obj: + return None + try: + v = float(obj[k]) + except (TypeError, ValueError): + return None + if not math.isfinite(v): # rejects "-inf"/"inf"/"nan" (silent clip) + return None + vals[k] = v + return MeasuredLoudness(**vals) + + +def build_loudnorm_apply_filter( + preset: Optional[str], measured: Optional["MeasuredLoudness"], +) -> Optional[str]: + """Second-pass (apply) loudnorm filter feeding the measured values back in. + ``None`` for off/unknown preset OR when ``measured`` is None (so a caller + that forgot to branch never emits ``measured_I=None``).""" + if not preset or measured is None: + return None + p = LOUDNESS_PRESETS.get(preset.lower()) + if p is None: + return None + return ( + f"loudnorm=I={p.i}:TP={p.tp}:LRA={p.lra}" + f":measured_I={measured.input_i}:measured_TP={measured.input_tp}" + f":measured_LRA={measured.input_lra}:measured_thresh={measured.input_thresh}" + f":offset={measured.target_offset}:linear=true:print_format=summary" + ) + + +def build_loudnorm_measure_cmd(ffmpeg: str, concat_list_path: str, filt: str) -> list[str]: + """Pure argv for the measure pass: decode the concat list, run the + print_format=json loudnorm filter, discard audio to the portable null muxer. + Input segment is byte-identical to build_render_cmd so measured == muxed.""" + return [ + ffmpeg, "-y", "-hide_banner", "-loglevel", "info", + "-f", "concat", "-safe", "0", "-i", str(concat_list_path), + "-af", filt, "-f", "null", "-", + ] + + # ── FFMETADATA ────────────────────────────────────────────────────────────── def build_ffmetadata( @@ -239,6 +339,7 @@ def build_render_cmd( bitrate: str = "128k", cover_path: Optional[str] = None, loudness: Optional[str] = None, + measured: Optional[MeasuredLoudness] = None, ) -> list[str]: """Pure argv for muxing chapter WAVs + FFMETADATA into a tagged, chapter-marked audio file. @@ -271,7 +372,10 @@ def build_render_cmd( if embed_cover: cmd += ["-map", "2:v", "-disposition:v", "attached_pic"] - filt = build_loudnorm_filter(loudness) + # Two-pass apply when measured values are present; else single-pass. Both + # return None for a non-preset loudness, so the `if filt:` guard below + # gives an off-render no -af (byte-identical to today). + filt = build_loudnorm_apply_filter(loudness, measured) if measured is not None else build_loudnorm_filter(loudness) if filt: cmd += ["-af", filt] diff --git a/tests/test_longform_render.py b/tests/test_longform_render.py index be2f47f9..bca5190b 100644 --- a/tests/test_longform_render.py +++ b/tests/test_longform_render.py @@ -10,15 +10,38 @@ import pytest from services.longform_render import ( LOUDNESS_PRESETS, + MeasuredLoudness, build_concat_list, build_ffmetadata, + build_loudnorm_apply_filter, build_loudnorm_filter, + build_loudnorm_measure_cmd, + build_loudnorm_measure_filter, build_render_cmd, chapter_cache_key, + parse_loudnorm_measure, prune_cache_dir, validate_cover_image, ) +# A verified ffmpeg loudnorm measure-JSON fixture (n8.1.1 shape). +_MEASURE_JSON = """[Parsed_loudnorm_0 @ 0x55] +{ + "input_i" : "-21.75", + "input_tp" : "-18.06", + "input_lra" : "0.00", + "input_thresh" : "-31.75", + "output_i" : "-19.02", + "output_tp" : "-3.01", + "normalization_type" : "dynamic", + "target_offset" : "0.05" +} +[out#0/null @ 0x66] video:0kB audio:1kB +size=N/A time=00:00:10 +""" +_MEASURED = MeasuredLoudness(input_i=-21.75, input_tp=-18.06, input_lra=0.0, + input_thresh=-31.75, target_offset=0.05) + # ── loudness ──────────────────────────────────────────────────────────────── @@ -258,3 +281,95 @@ def test_prune_cache_evicts_oldest_first(tmp_path): def test_prune_cache_missing_dir_is_safe(tmp_path): assert prune_cache_dir(str(tmp_path / "nope")) == (0, 0) + + +# ── #28 two-pass loudnorm: measure filter ─────────────────────────────────── + +def test_measure_filter_goldens(): + assert build_loudnorm_measure_filter("acx") == "loudnorm=I=-19.0:TP=-3.0:LRA=11.0:print_format=json" + assert build_loudnorm_measure_filter("podcast") == "loudnorm=I=-16.0:TP=-1.5:LRA=11.0:print_format=json" + assert build_loudnorm_measure_filter("ACX") == build_loudnorm_measure_filter("acx") + + +@pytest.mark.parametrize("val", [None, "", "off", "none", "bogus", " acx "]) +def test_measure_filter_off_unknown_whitespace_is_none(val): + # mirrors single-pass: no strip, so " acx " is unknown → None + assert build_loudnorm_measure_filter(val) is None + + +# ── parse_loudnorm_measure ────────────────────────────────────────────────── + +def test_parse_measure_success_picks_last_block_and_ignores_extra_keys(): + m = parse_loudnorm_measure(_MEASURE_JSON) + assert m == _MEASURED + + +def test_parse_measure_last_block_wins_over_config_dump(): + text = '{"input_i":"1"}\nconfig\n' + _MEASURE_JSON + assert parse_loudnorm_measure(text) == _MEASURED + + +@pytest.mark.parametrize("bad", [ + None, "", " ", "no braces here", '{ "input_i": "-1"', # no/unbalanced + '{ "input_i": "-1", }', '{ input_i: -1 }', # malformed json + '{ "input_i":"-1","input_tp":"-3","input_lra":"0","input_thresh":"-30" }', # missing target_offset + '[1,2,3]', '"scalar"', # not an object +]) +def test_parse_measure_failure_matrix_returns_none(bad): + assert parse_loudnorm_measure(bad) is None + + +@pytest.mark.parametrize("badval", ['"n/a"', '""', '"-inf"', '"inf"', '"nan"']) +def test_parse_measure_rejects_nonnumeric_and_nonfinite(badval): + text = ('{ "input_i":%s,"input_tp":"-3","input_lra":"0",' + '"input_thresh":"-30","target_offset":"0.0" }') % badval + assert parse_loudnorm_measure(text) is None + + +# ── apply filter ──────────────────────────────────────────────────────────── + +def test_apply_filter_golden(): + f = build_loudnorm_apply_filter("acx", _MEASURED) + assert f == ( + "loudnorm=I=-19.0:TP=-3.0:LRA=11.0:measured_I=-21.75:measured_TP=-18.06" + ":measured_LRA=0.0:measured_thresh=-31.75:offset=0.05:linear=true:print_format=summary" + ) + + +@pytest.mark.parametrize("preset,measured", [ + ("off", _MEASURED), (None, _MEASURED), ("bogus", _MEASURED), ("acx", None), +]) +def test_apply_filter_none_cases(preset, measured): + assert build_loudnorm_apply_filter(preset, measured) is None + + +# ── measure cmd argv ──────────────────────────────────────────────────────── + +def test_measure_cmd_exact_argv(): + assert build_loudnorm_measure_cmd("ffmpeg", "c.txt", "FILT") == [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "info", + "-f", "concat", "-safe", "0", "-i", "c.txt", + "-af", "FILT", "-f", "null", "-", + ] + + +# ── build_render_cmd measured branch ──────────────────────────────────────── + +def test_render_cmd_two_pass_apply_when_measured(): + cmd = build_render_cmd("ffmpeg", "c.txt", "m.ff", "o.m4b", loudness="acx", measured=_MEASURED) + af = cmd[cmd.index("-af") + 1] + assert "measured_I=-21.75" in af and "linear=true" in af + + +def test_render_cmd_single_pass_when_measured_none(): + cmd = build_render_cmd("ffmpeg", "c.txt", "m.ff", "o.m4b", loudness="acx", measured=None) + af = cmd[cmd.index("-af") + 1] + assert af == "loudnorm=I=-19.0:TP=-3.0:LRA=11.0" # single-pass, no measured_* + + +def test_render_cmd_off_emits_no_af_even_with_stray_measured(): + cmd = build_render_cmd("ffmpeg", "c.txt", "m.ff", "o.m4b", loudness="off", measured=_MEASURED) + assert "-af" not in cmd + + cmd2 = build_render_cmd("ffmpeg", "c.txt", "m.ff", "o.m4b") # default: no loudness/measured + assert "-af" not in cmd2