diff options
Diffstat (limited to 'lib/project/src/voiceforge/audio.py')
| -rw-r--r-- | lib/project/src/voiceforge/audio.py | 128 |
1 files changed, 0 insertions, 128 deletions
diff --git a/lib/project/src/voiceforge/audio.py b/lib/project/src/voiceforge/audio.py deleted file mode 100644 index 7fee0c2..0000000 --- a/lib/project/src/voiceforge/audio.py +++ /dev/null @@ -1,128 +0,0 @@ -"""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")} |