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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
"""Offline mono voice mastering with float intermediates and atomic outputs."""
from dataclasses import asdict
import fcntl
import json
import math
import os
from pathlib import Path
import tempfile
from . import __version__
from .audio import ffmpeg_run, loudness, measure
from .config import Settings
from .setup import ensure_ffmpeg
def filters(settings: Settings) -> list[str]:
s = settings
result = []
if s.expansion:
result.append(f"agate=threshold={10 ** (s.expansion_threshold_db / 20)}:ratio={s.expansion_ratio}:range={10 ** (-s.expansion_range_db / 20)}:attack=10:release=250:detection=rms")
if s.eq:
result += [f"equalizer=f=140:t=q:w=0.7:g={s.warmth_db}",
f"equalizer=f=300:t=q:w=0.8:g={s.mud_db}",
f"equalizer=f=3500:t=q:w=0.7:g={s.presence_db}"]
if s.compression:
result.append(f"acompressor=threshold={10 ** (s.compressor_threshold_db / 20)}:ratio={s.compressor_ratio}:attack={s.compressor_attack_ms}:release={s.compressor_release_ms}:knee=4:makeup=1:detection=rms")
if s.deess:
result.append(f"deesser=i={s.deess_intensity}:m={s.deess_amount}:f=0.5")
if s.lowpass:
result.append(f"lowpass=f={s.lowpass_hz}:p=2")
return result or ["anull"]
def targets_for(destination: Path, settings: Settings) -> list[Path]:
"""Every path a successful run publishes: master WAV, report, optional MP3."""
targets = [destination, destination.with_suffix(".report.json")]
if settings.mp3:
targets.append(destination.with_suffix(".mp3"))
return targets
def check_target(source: Path, target: Path, overwrite: bool) -> None:
"""Preflight one published path; the original recording is never touched."""
if target.resolve() == source or (target.exists() and os.path.samefile(source, target)):
raise ValueError("Refusing to overwrite the original recording")
# lexists also rejects dangling symlinks, which publication would only hit late.
if os.path.lexists(target) and not overwrite:
raise FileExistsError(f"Output exists: {target}; use --overwrite to replace outputs")
if target.exists() and not target.is_file():
raise ValueError(f"Output is not a regular file: {target}")
def publish(source: Path, target: Path, overwrite: bool):
if overwrite:
os.replace(source, target)
else:
# Link is atomic and fails if another process created the destination.
os.link(source, target)
source.unlink()
def process(source: Path, destination: Path, settings: Settings, progress,
overwrite=False, provenance: dict | None = None) -> dict:
source, destination = source.resolve(), destination.absolute()
if not source.is_file() or source.suffix.lower() != ".wav":
raise ValueError(f"Input must be an existing WAV file: {source}")
if destination.suffix.lower() != ".wav":
raise ValueError("Master output must have a .wav extension")
report_path = destination.with_suffix(".report.json")
for target in targets_for(destination, settings):
check_target(source, target, overwrite)
ffmpeg = ensure_ffmpeg()
original = measure(source, progress, "Analyzing original")
if original["sample_peak_dbfs"] < -100:
raise ValueError("Recording is silent or too quiet to master safely")
channel = settings.channel
if original["channels"] == 2 and channel == "auto":
raise ValueError("Stereo input: select --channel left, right, or mix explicitly. Auto downmix could cancel your voice.")
if original["channels"] == 1 and channel == "right":
raise ValueError("Cannot select the right channel of a mono recording")
duration = original["duration_seconds"]
warnings = []
if original["near_full_scale_samples"]:
warnings.append("Input has near-full-scale samples: inspect for clipping; lost peaks cannot be reliably restored.")
if max(abs(x) for x in original["dc_offset"]) > 0.01:
warnings.append("Significant input DC offset detected; enable highpass filtering to remove it.")
destination.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix=".voiceforge-", dir=destination.parent) as temporary:
work = Path(temporary)
prepared, cleaned, shaped = [work / name for name in ("prepared.wav", "cleaned.wav", "shaped.wav")]
input_filters = []
if original["channels"] == 2:
input_filters.append({"left": "pan=mono|c0=c0", "right": "pan=mono|c0=c1",
"mix": "pan=mono|c0=0.5*c0+0.5*c1"}[channel])
if settings.highpass:
input_filters.append(f"highpass=f={settings.highpass_hz}:p=2")
if settings.hum_hz:
input_filters += [f"equalizer=f={settings.hum_hz * n}:t=q:w=25:g=-18" for n in (1, 2, 3)]
ffmpeg_run(ffmpeg, ["-i", str(source), "-map", "0:a:0", "-af",
",".join(input_filters or ["anull"]), "-ar", "48000", "-ac", "1",
"-c:a", "pcm_f32le", "-rf64", "auto", str(prepared)], "Preparing mono 48 kHz audio", duration, progress)
if settings.denoiser == "deepfilter" and settings.denoise_strength > 0:
from .ai import denoise
denoise(prepared, cleaned, settings.device, settings.denoise_strength, progress)
elif settings.denoiser == "fft" and settings.denoise_strength > 0:
ffmpeg_run(ffmpeg, ["-i", str(prepared), "-af",
f"afftdn=nr={max(0.01, settings.fft_reduction_db * settings.denoise_strength)}:tn=1",
"-c:a", "pcm_f32le", "-rf64", "auto", str(cleaned)], "FFT noise reduction", duration, progress)
else:
cleaned = prepared
clean_stats = measure(cleaned, progress, "Measuring cleaned voice")
if clean_stats["sample_peak_dbfs"] < -100:
raise ValueError("The selected audio is silent or too quiet to master safely; "
"check --channel, the mix, and cleanup settings")
gain = 0.0
if settings.leveling:
speech_db = clean_stats["speech_level_estimate_dbfs"]
if speech_db <= -110:
warnings.append("Speech level estimate is unreliable; leveling skipped. "
"Inspect levels or rely on normalization.")
else:
desired = settings.level_target_db - speech_db
gain = max(-settings.max_gain_db, min(settings.max_gain_db, desired))
if abs(desired) > settings.max_gain_db:
warnings.append("Input level correction reached its configured gain limit.")
chain = ([f"volume={gain}dB"] if gain else []) + filters(settings)
ffmpeg_run(ffmpeg, ["-i", str(cleaned), "-af", ",".join(chain),
"-c:a", "pcm_f32le", "-rf64", "auto", str(shaped)], "EQ, dynamics and de-essing", duration, progress)
before = loudness(ffmpeg, shaped, settings, duration, progress, "Loudness analysis (pass 1)")
if settings.normalize and any(before.get(k) is None for k in ("input_i", "input_tp", "input_lra", "input_thresh", "target_offset")):
raise ValueError("Cannot normalize: insufficient measurable audio after processing")
final_filters = []
if settings.normalize:
final_filters.append(f"loudnorm=I={settings.target_lufs}:TP={settings.true_peak_db}:LRA={settings.loudness_range}:"
f"measured_I={before['input_i']}:measured_TP={before['input_tp']}:measured_LRA={before['input_lra']}:"
f"measured_thresh={before['input_thresh']}:offset={before['target_offset']}:linear=true:print_format=json")
elif settings.limiter:
final_filters += ["aresample=192000", f"alimiter=limit={10 ** (settings.true_peak_db / 20)}:level=false:latency=true"]
final_filters.append(f"aresample={settings.sample_rate}:output_sample_bits=24:dither_method=triangular")
master = work / "master.wav"
render_log = ffmpeg_run(ffmpeg, ["-i", str(shaped), "-af", ",".join(final_filters),
"-ar", str(settings.sample_rate), "-c:a", "pcm_s24le", "-rf64", "auto", str(master)],
"Mastering (pass 2)", duration, progress)
# Bounded disk use: drop consumed intermediates instead of waiting for
# the temporary directory teardown at the end of the run.
shaped.unlink(missing_ok=True)
for intermediate in {prepared, cleaned}:
intermediate.unlink(missing_ok=True)
normalization_mode = "disabled"
if settings.normalize:
start, end = render_log.rfind("{"), render_log.rfind("}")
if start < 0 or end < start:
raise RuntimeError("FFmpeg did not report the applied normalization mode")
normalization_mode = json.loads(render_log[start:end + 1])["normalization_type"]
final = loudness(ffmpeg, master, settings, duration, progress, "Verifying exported master")
final_stats = measure(master, progress, "Checking exported samples")
if abs(final_stats["duration_seconds"] - duration) > 0.02:
raise RuntimeError("Output duration differs from input by more than 20 ms")
if (final_stats["channels"] != 1 or final_stats["sample_rate"] != settings.sample_rate
or final_stats["subtype"] != "PCM_24"):
raise RuntimeError("Exported master is not the expected mono PCM_24 WAV")
if settings.normalize and (final["input_i"] is None or abs(final["input_i"] - settings.target_lufs) > 0.5):
warnings.append("Master misses the loudness target by more than 0.5 LU; inspect the report before publishing.")
if settings.limiter and final["input_tp"] is not None and final["input_tp"] > settings.true_peak_db + 0.1:
warnings.append("Master exceeds the true-peak target by more than 0.1 dB; inspect before publishing.")
if final_stats["near_full_scale_samples"]:
warnings.append("Export has near-full-scale samples. Enable limiting or lower gains.")
mp3_stats = None
if settings.mp3:
mp3 = work / "delivery.mp3"
ffmpeg_run(ffmpeg, ["-i", str(master), "-c:a", "libmp3lame", "-b:a",
f"{settings.mp3_bitrate}k", str(mp3)], "Encoding MP3", duration, progress)
mp3_stats = loudness(ffmpeg, mp3, settings, duration, progress, "Verifying decoded MP3")
if mp3_stats["input_tp"] is not None and mp3_stats["input_tp"] > settings.true_peak_db + 0.1:
warnings.append("MP3 encoding increased true peak beyond the configured ceiling; use more headroom.")
report = {"voiceforge_version": __version__, "source": str(source),
"output": str(destination), "settings": asdict(settings), "input": original,
"cleaned": clean_stats, "level_correction_db": gain,
"pre_master_loudness": before, "master_loudness": final,
"master": final_stats, "mp3_loudness": mp3_stats,
"normalization_mode": normalization_mode,
"warnings": warnings}
if provenance:
report["provenance"] = provenance
staged_report = work / "report.json"
staged_report.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n")
# Publish only after all processing and verification succeeded. The
# directory lock serializes publication so concurrent jobs targeting
# the same output directory cannot interleave WAV/report/MP3 sets.
publication_fd = os.open(destination.parent, os.O_RDONLY)
try:
fcntl.flock(publication_fd, fcntl.LOCK_EX)
if settings.mp3:
publish(mp3, destination.with_suffix(".mp3"), overwrite)
publish(staged_report, report_path, overwrite)
publish(master, destination, overwrite)
finally:
os.close(publication_fd)
return report
|
