Groundwork for accurate ACX mastering: the pure, ffmpeg-free pieces of the
two-pass loudnorm upgrade, layered over the existing single-pass builders
(which stay). The async measure orchestrator + SSE wiring into the render path
is slice 2.
- `MeasuredLoudness` (frozen dataclass: the 5 measure-pass floats).
- `build_loudnorm_measure_filter(preset)` — first pass (+print_format=json);
mirrors build_loudnorm_filter's lookup (no strip) so the same inputs map to
"no filter".
- `parse_loudnorm_measure(stderr)` — extracts the LAST balanced {...} via a
linear brace-depth scan (NO regex → CodeQL-safe), json.loads + coerces the 5
keys to finite floats; returns None on the full failure matrix (absent/empty/
unbalanced/malformed/missing-key/non-numeric/non-finite "-inf"/array/scalar).
Rejecting "-inf" is the silent-clip path → single-pass fallback.
- `build_loudnorm_apply_filter(preset, measured)` — second pass feeding
measured_*/offset back in with linear=true; None for off/unknown OR measured
is None.
- `build_loudnorm_measure_cmd(ffmpeg, concat, filt)` — exact 16-element argv,
input segment byte-identical to build_render_cmd (measured == muxed),
portable `-f null -` sink (no /dev/null or NUL).
- `build_render_cmd` gains `measured: Optional[MeasuredLoudness] = None`: apply
two-pass when present, else single-pass; off-render still emits no -af. The
`measured=None` default keeps every existing caller + argv byte-identical.
Loudness stays opt-in (default None) → default cross-platform behavior unchanged.
Tests: 28 cases — measure-filter goldens + off/unknown/whitespace; parser
success (last-block-wins, ignores extra keys) + full failure matrix +
non-finite rejection; apply-filter golden + None cases; exact measure argv;
build_render_cmd two-pass/single-pass/off branches. Backend pytest green (71).
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Palash Debnath
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
e297cbfee3
commit
e1c8c3bc0d
@@ -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]
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user