Grok Bot
52 lines
2.2 KiB
Python
52 lines
2.2 KiB
Python
"""Measure audio with ffmpeg: RMS and peak per 50 ms window. The checker and the assembler both use this."""
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
|
|
|
|
def ffmpeg(args):
|
|
r = subprocess.run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y"] + args, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
raise RuntimeError("ffmpeg failed: " + r.stderr.strip()[-500:])
|
|
|
|
|
|
def probe(path):
|
|
out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries",
|
|
"stream=codec_name,sample_rate,channels,bit_rate:format=duration", "-of", "default=nw=1", path],
|
|
capture_output=True, text=True, check=True).stdout
|
|
return dict(line.split("=", 1) for line in out.strip().splitlines())
|
|
|
|
|
|
def windows(path):
|
|
"""[(rms_db, peak_db)] for consecutive 50 ms windows of the decoded file."""
|
|
sr = int(probe(path)["sample_rate"])
|
|
n = sr // 20
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
rms_file, peak_file = os.path.join(tmp, "rms.txt"), os.path.join(tmp, "peak.txt")
|
|
af = ("asetnsamples=n=%d:p=0,astats=metadata=1:reset=1,"
|
|
"ametadata=print:key=lavfi.astats.Overall.RMS_level:file=%s,"
|
|
"ametadata=print:key=lavfi.astats.Overall.Peak_level:file=%s" % (n, rms_file, peak_file))
|
|
r = subprocess.run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-i", path, "-af", af, "-f", "null", "-"],
|
|
capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
raise RuntimeError("ffmpeg failed: " + r.stderr.strip()[-500:])
|
|
read = lambda p, key: [float(v) for v in re.findall(key + r"=(-?[\d.]+|-inf|inf)", open(p).read())]
|
|
rms, peak = read(rms_file, "RMS_level"), read(peak_file, "Peak_level")
|
|
return list(zip(rms, peak))
|
|
|
|
|
|
def db_mean_power(values):
|
|
"""Overall level in dB of equal-length windows given as dB RMS."""
|
|
if not values:
|
|
return float("-inf")
|
|
p = sum(10 ** (v / 10) for v in values) / len(values)
|
|
return 10 * math.log10(p) if p > 0 else float("-inf")
|
|
|
|
|
|
def overall(path):
|
|
"""(rms_db, peak_db) of the whole file."""
|
|
w = windows(path)
|
|
return db_mean_power([r for r, _ in w]), max(p for _, p in w)
|