Add Beryl Agent and harness source
This commit is contained in:
@@ -0,0 +1,234 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Apply the PiecesAI delivery-audio standard without re-encoding video."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
VOLUME_RE = re.compile(r"(?P<name>mean_volume|max_volume): (?P<value>-?inf|-?\d+(?:\.\d+)?) dB")
|
||||
|
||||
|
||||
def _binary(name: str) -> str:
|
||||
path = shutil.which(name)
|
||||
if path is None:
|
||||
raise RuntimeError(f"required binary not found: {name}")
|
||||
return path
|
||||
|
||||
|
||||
def _run(command: list[str], *, capture: bool = False) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
command,
|
||||
check=True,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE if capture else None,
|
||||
stderr=subprocess.PIPE if capture else None,
|
||||
)
|
||||
|
||||
|
||||
def _probe(ffprobe: str, media_path: Path) -> dict:
|
||||
result = _run(
|
||||
[
|
||||
ffprobe,
|
||||
"-v",
|
||||
"error",
|
||||
"-show_entries",
|
||||
"stream=index,codec_type,codec_name,width,height,r_frame_rate,sample_rate,channels",
|
||||
"-show_entries",
|
||||
"format=duration,size",
|
||||
"-of",
|
||||
"json",
|
||||
str(media_path),
|
||||
],
|
||||
capture=True,
|
||||
)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
|
||||
def _volume_stats(ffmpeg: str, media_path: Path) -> dict[str, float]:
|
||||
result = _run(
|
||||
[
|
||||
ffmpeg,
|
||||
"-hide_banner",
|
||||
"-i",
|
||||
str(media_path),
|
||||
"-af",
|
||||
"volumedetect",
|
||||
"-f",
|
||||
"null",
|
||||
"-",
|
||||
],
|
||||
capture=True,
|
||||
)
|
||||
stats: dict[str, float] = {}
|
||||
for match in VOLUME_RE.finditer(result.stderr):
|
||||
stats[match.group("name")] = float(match.group("value"))
|
||||
if set(stats) != {"mean_volume", "max_volume"}:
|
||||
raise RuntimeError(f"could not read volume statistics from {media_path}")
|
||||
return stats
|
||||
|
||||
|
||||
def _audio_filter(*, gain: float, peak_limit: float, denoise: str) -> str:
|
||||
filters: list[str] = []
|
||||
if denoise == "afftdn":
|
||||
filters.append("afftdn=nr=10:nf=-45:tn=1")
|
||||
elif denoise == "anlmdn":
|
||||
filters.append("anlmdn=s=1e-5:p=0.002:r=0.006:m=15")
|
||||
filters.extend(
|
||||
(
|
||||
f"volume={gain:.6f}",
|
||||
f"alimiter=limit={peak_limit:.6f}:attack=5:release=50:level=false",
|
||||
)
|
||||
)
|
||||
return ",".join(filters)
|
||||
|
||||
|
||||
def finalize_audio(
|
||||
input_path: Path,
|
||||
output_path: Path,
|
||||
*,
|
||||
gain: float = 2.0,
|
||||
peak_limit: float = 0.89,
|
||||
denoise: str = "none",
|
||||
audio_bitrate: str = "256k",
|
||||
sample_rate: int = 48_000,
|
||||
force: bool = False,
|
||||
) -> dict:
|
||||
input_path = input_path.resolve()
|
||||
output_path = output_path.resolve()
|
||||
if input_path == output_path:
|
||||
raise ValueError("input and output paths must differ")
|
||||
if not input_path.is_file():
|
||||
raise FileNotFoundError(input_path)
|
||||
if output_path.exists() and not force:
|
||||
raise FileExistsError(f"output exists; pass --force to replace it: {output_path}")
|
||||
if gain <= 0:
|
||||
raise ValueError("gain must be greater than zero")
|
||||
if not 0 < peak_limit <= 1:
|
||||
raise ValueError("peak limit must be within (0, 1]")
|
||||
|
||||
ffmpeg = _binary("ffmpeg")
|
||||
ffprobe = _binary("ffprobe")
|
||||
source_probe = _probe(ffprobe, input_path)
|
||||
stream_types = {stream.get("codec_type") for stream in source_probe.get("streams", [])}
|
||||
if not {"video", "audio"}.issubset(stream_types):
|
||||
raise ValueError("input must contain both video and audio streams")
|
||||
|
||||
source_volume = _volume_stats(ffmpeg, input_path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = output_path.with_name(
|
||||
f".{output_path.stem}.{uuid.uuid4().hex}.tmp{output_path.suffix or '.mp4'}"
|
||||
)
|
||||
try:
|
||||
_run(
|
||||
[
|
||||
ffmpeg,
|
||||
"-hide_banner",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-y",
|
||||
"-i",
|
||||
str(input_path),
|
||||
"-map",
|
||||
"0:v:0",
|
||||
"-map",
|
||||
"0:a:0",
|
||||
"-map_metadata",
|
||||
"0",
|
||||
"-c:v",
|
||||
"copy",
|
||||
"-af",
|
||||
_audio_filter(gain=gain, peak_limit=peak_limit, denoise=denoise),
|
||||
"-c:a",
|
||||
"aac",
|
||||
"-b:a",
|
||||
audio_bitrate,
|
||||
"-ar",
|
||||
str(sample_rate),
|
||||
"-movflags",
|
||||
"+faststart",
|
||||
str(temporary),
|
||||
]
|
||||
)
|
||||
output_probe = _probe(ffprobe, temporary)
|
||||
output_volume = _volume_stats(ffmpeg, temporary)
|
||||
# AAC can overshoot the linear limiter slightly. The standard keeps at
|
||||
# least 0.5 dB of encoded-sample headroom.
|
||||
if output_volume["max_volume"] > -0.5:
|
||||
raise RuntimeError(
|
||||
f"unsafe output peak: {output_volume['max_volume']:.1f} dB; "
|
||||
"lower --peak-limit"
|
||||
)
|
||||
os.replace(temporary, output_path)
|
||||
finally:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
return {
|
||||
"input": str(input_path),
|
||||
"output": str(output_path),
|
||||
"video_mode": "stream_copy",
|
||||
"gain": gain,
|
||||
"gain_db": 20 * math.log10(gain),
|
||||
"peak_limit": peak_limit,
|
||||
"denoise": denoise,
|
||||
"audio_codec": "aac",
|
||||
"audio_bitrate": audio_bitrate,
|
||||
"sample_rate": sample_rate,
|
||||
"source_volume": source_volume,
|
||||
"output_volume": output_volume,
|
||||
"source_probe": source_probe,
|
||||
"output_probe": output_probe,
|
||||
}
|
||||
|
||||
|
||||
def _parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Boost delivery audio while stream-copying the video track."
|
||||
)
|
||||
parser.add_argument("input", type=Path)
|
||||
parser.add_argument("output", type=Path)
|
||||
parser.add_argument("--gain", type=float, default=2.0)
|
||||
parser.add_argument("--peak-limit", type=float, default=0.89)
|
||||
parser.add_argument("--denoise", choices=("none", "afftdn", "anlmdn"), default="none")
|
||||
parser.add_argument("--audio-bitrate", default="256k")
|
||||
parser.add_argument("--sample-rate", type=int, default=48_000)
|
||||
parser.add_argument("--receipt", type=Path)
|
||||
parser.add_argument("--force", action="store_true")
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = _parser().parse_args()
|
||||
try:
|
||||
receipt = finalize_audio(
|
||||
args.input,
|
||||
args.output,
|
||||
gain=args.gain,
|
||||
peak_limit=args.peak_limit,
|
||||
denoise=args.denoise,
|
||||
audio_bitrate=args.audio_bitrate,
|
||||
sample_rate=args.sample_rate,
|
||||
force=args.force,
|
||||
)
|
||||
except (FileNotFoundError, FileExistsError, RuntimeError, ValueError) as exc:
|
||||
print(f"error: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
encoded = json.dumps(receipt, ensure_ascii=False, indent=2)
|
||||
if args.receipt:
|
||||
args.receipt.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.receipt.write_text(encoded + "\n", encoding="utf-8")
|
||||
print(encoded)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user