1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
"""Streaming measurement and FFmpeg execution. No shell interpolation."""
from collections import deque
import json
import math
import os
from pathlib import Path
import selectors
import signal
import subprocess
import time
import numpy as np
import soundfile as sf
from .setup import terminate_group
def ffmpeg_run(ffmpeg: str, args: list[str], stage: str, duration: float,
progress) -> str:
command = [ffmpeg, "-hide_banner", "-nostdin", "-y", "-nostats",
"-progress", "pipe:1"] + args
progress(stage, 0, duration or None)
diagnostics = deque(maxlen=128)
pending = b""
completed = 0.0
last_advance = time.monotonic()
with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
start_new_session=True) as process:
try:
with selectors.DefaultSelector() as selector:
for stream in (process.stdout, process.stderr):
os.set_blocking(stream.fileno(), False)
selector.register(stream, selectors.EVENT_READ)
while selector.get_map() or process.poll() is None:
for key, _ in selector.select(timeout=0.2):
data = os.read(key.fd, 8192)
if not data:
selector.unregister(key.fileobj)
continue
if key.fileobj is process.stderr:
diagnostics.append(data)
else:
pending += data
while b"\n" in pending:
line, pending = pending.split(b"\n", 1)
if line.startswith(b"out_time_us="):
try:
value = max(0.0, int(line.split(b"=", 1)[1]) / 1e6)
except ValueError:
continue
if value > completed:
last_advance = time.monotonic()
completed = value
label = stage
if time.monotonic() - last_advance > 60:
label += " (no new audio progress for 60s; still waiting)"
progress(label, min(completed, duration), duration or None)
code = process.wait()
output = b"".join(diagnostics).decode(errors="replace")
if code:
raise RuntimeError(f"{stage} failed (exit {code}):\n{output[-8000:]}")
progress(stage, duration, duration or None)
return output
except BaseException:
terminate_group(process)
raise
def measure(path: Path, progress, stage="Analyzing audio") -> dict:
"""100ms RMS histogram is bounded in memory, including multi-hour input."""
with sf.SoundFile(path) as source:
if source.channels not in (1, 2):
raise ValueError("Only mono or stereo voice recordings are supported")
if not len(source):
raise ValueError("The recording is empty")
peaks = np.zeros(source.channels)
sums = np.zeros(source.channels)
squares = np.zeros(source.channels)
clipped = 0
histogram = np.zeros(121, dtype=np.int64)
frames = 0
progress(stage, 0, len(source))
for audio in source.blocks(blocksize=max(1, source.samplerate // 10),
dtype="float64", always_2d=True):
if not np.isfinite(audio).all():
raise ValueError("Recording contains non-finite samples")
peaks = np.maximum(peaks, np.max(np.abs(audio), axis=0))
sums += audio.sum(axis=0)
squares += (audio * audio).sum(axis=0)
clipped += int(np.count_nonzero(np.abs(audio) >= 0.9999))
rms = float(np.sqrt(np.mean(audio * audio)))
db = 20 * math.log10(max(rms, 1e-6))
histogram[int(np.clip(round(db) + 120, 0, 120))] += 1
frames += len(audio)
progress(stage, frames, len(source))
cumulative = histogram.cumsum()
def percentile(p):
return int(np.searchsorted(cumulative, max(1, math.ceil(cumulative[-1] * p)))) - 120
quiet, speech = percentile(0.1), percentile(0.8)
if speech - quiet < 15:
# Pause-dominated recordings: the 80th percentile of all blocks
# sits in room tone, so estimate the level from the loud tail
# instead of mistaking sparse active speech for silence.
speech = max(speech, percentile(0.98))
noise_available = speech - quiet >= 15 and quiet > -110
return {"frames": frames, "duration_seconds": frames / source.samplerate,
"sample_rate": source.samplerate, "channels": source.channels,
"subtype": source.subtype,
"sample_peak_dbfs": float(20 * np.log10(max(float(peaks.max()), 1e-12))),
"channel_rms_dbfs": (20 * np.log10(np.maximum(np.sqrt(squares / frames), 1e-12))).tolist(),
"dc_offset": (sums / frames).tolist(), "near_full_scale_samples": clipped,
"speech_level_estimate_dbfs": speech,
"quiet_blocks_dbfs": quiet,
"noise_floor_estimate_dbfs": quiet if noise_available else None,
"noise_floor_confidence": "low (quiet-block heuristic, not speech recognition)" if noise_available else "insufficient room tone"}
def loudness(ffmpeg, source, settings, duration, progress, stage="Measuring loudness"):
output = ffmpeg_run(ffmpeg, ["-i", str(source), "-map", "0:a:0", "-af",
f"loudnorm=I={settings.target_lufs}:TP={settings.true_peak_db}:LRA={settings.loudness_range}:print_format=json",
"-f", "null", "-"], stage, duration, progress)
start, end = output.rfind("{"), output.rfind("}")
if start < 0 or end < start:
raise RuntimeError("FFmpeg did not return loudness measurements")
raw = json.loads(output[start:end + 1])
return {key: (float(value) if math.isfinite(float(value)) else None)
for key, value in raw.items() if key != "normalization_type"} | {
"normalization_type": raw.get("normalization_type")}
|
