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
|
"""Command-line entry point; configuration errors fail before model downloads."""
import argparse
from dataclasses import asdict, fields
import json
import math
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import tomllib
from rich.console import Console
from . import __version__
from .audio import ffmpeg_run, loudness, measure
from .config import PROFILES, Settings, resolve, toml
from .pipeline import check_target, process, targets_for
from .progress import Display
from .setup import data_dir, ensure_ffmpeg
def parser():
app = argparse.ArgumentParser(prog="producer", description="Local speech cleanup and podcast mastering. Originals are never overwritten.")
app.add_argument("--version", action="version", version=__version__)
sub = app.add_subparsers(dest="command", required=True)
for name in ("process", "preview", "config", "analyze"):
cmd = sub.add_parser(name)
cmd.add_argument("--config", type=Path, help="TOML configuration file")
cmd.add_argument("--profile", choices=PROFILES, default=None)
cmd.add_argument("--set", action="append", default=[], metavar="KEY=VALUE", help="Override any setting using TOML value syntax")
for field in fields(Settings):
if field.name == "profile":
continue
default = getattr(Settings(), field.name)
option = "--" + field.name.replace("_", "-")
if isinstance(default, bool):
cmd.add_argument(option, action=argparse.BooleanOptionalAction, default=None)
else:
cmd.add_argument(option, type=type(default), default=None)
cmd.add_argument("--no-denoise", dest="denoiser", action="store_const", const="none")
if name == "config":
cmd.add_argument("--json", action="store_true")
else:
cmd.add_argument("inputs", type=Path, nargs="+" if name == "process" else 1)
if name in ("process", "preview"):
cmd.add_argument("--output-dir", type=Path, default=None,
help="Directory for generated outputs (default: beside each input)")
cmd.add_argument("--overwrite", action="store_true", help="Replace generated outputs, never input files")
if name == "process":
cmd.add_argument("-o", "--output", type=Path, help="Explicit WAV destination (one input only)")
if name == "preview":
cmd.add_argument("--start", type=float, default=0)
cmd.add_argument("--duration", type=float, default=30)
cmd.add_argument("--profiles", nargs="+", choices=PROFILES, default=["natural", "narrator", "radio"])
sub.add_parser("doctor", help="Check tools and driver without downloading AI")
setup = sub.add_parser("setup", help="Download and self-test the isolated AI backend")
setup.add_argument("--device", choices=("auto", "cpu", "cuda"), default="auto")
return app
def settings_for(args, profile=None):
overrides = {field.name: getattr(args, field.name) for field in fields(Settings)
if field.name != "profile" and getattr(args, field.name, None) is not None}
for item in args.set:
key, sep, value = item.partition("=")
if not sep:
raise ValueError("--set expects KEY=VALUE")
# Strings can be supplied without TOML quotes for CLI convenience.
try:
parsed = tomllib.loads("value = " + value)["value"]
except tomllib.TOMLDecodeError:
parsed = value
overrides[key.strip()] = parsed
return resolve(args.config, profile or args.profile, overrides)
def require_input(source: Path) -> None:
if not source.is_file() or source.suffix.lower() != ".wav":
raise ValueError(f"Input must be an existing WAV file: {source}")
def main():
app = parser()
args = app.parse_args()
console = Console(stderr=True)
try:
if args.command == "doctor":
from .ai import _cuda_available
ffmpeg = ensure_ffmpeg()
version = subprocess.run([ffmpeg, "-version"], capture_output=True, text=True, check=True).stdout.splitlines()[0]
console.print(version, markup=False)
console.print(f"FFmpeg: {ffmpeg}\nData: {data_dir()}\nCUDA driver available: {_cuda_available()}\nAI is installed and self-tested by './producer.sh setup'.", markup=False)
return
if args.command == "setup":
from .ai import ensure_ai
with Display() as display:
ensure_ffmpeg()
python = ensure_ai(args.device, display.update, verify=True)
console.print(f"AI ready: {python}", markup=False)
return
settings = settings_for(args)
if args.command == "config":
print(json.dumps(asdict(settings), indent=2) if args.json else toml(settings), end="\n" if args.json else "")
return
with Display() as display:
if args.command == "analyze":
stats = measure(args.inputs[0], display.update)
stats["loudness"] = loudness(ensure_ffmpeg(), args.inputs[0], settings, stats["duration_seconds"], display.update)
print(json.dumps(stats, indent=2, allow_nan=False))
return
if args.command == "process":
if args.output and len(args.inputs) != 1:
raise ValueError("--output requires exactly one input")
if args.output and args.output_dir:
raise ValueError("--output and --output-dir cannot be combined")
# Without an explicit destination, outputs are written beside
# each input; --output-dir keeps every result in one directory.
outputs = [args.output or (args.output_dir or p.parent) / f"{p.stem}.{settings.profile}.wav"
for p in args.inputs]
# Preflight the whole plan so a predictable conflict cannot
# fail a batch after earlier inputs have already been published.
sources = {p.resolve() for p in args.inputs}
claimed: set[Path] = set()
for source, output in zip(args.inputs, outputs):
require_input(source)
targets = targets_for(output.absolute(), settings)
resolved = {target.resolve() for target in targets}
if len(resolved) < len(targets) or resolved & claimed:
raise ValueError("Inputs have duplicate output names; process them separately with --output")
claimed |= resolved
if resolved & sources or any(
target.exists() and os.path.samefile(source, target) for target in targets):
raise ValueError("An output would overwrite an input recording")
for target in targets:
check_target(source, target, args.overwrite)
for source, output in zip(args.inputs, outputs):
report = process(source, output, settings, display.update, args.overwrite)
console.print(f"Written: {output} | {report['master_loudness']['input_i']} LUFS | {report['master_loudness']['input_tp']} dBTP", markup=False)
for warning in report["warnings"]:
console.print(f"Warning: {warning}", markup=False)
return
if not math.isfinite(args.start) or args.start < 0 or not 0 < args.duration <= 300:
raise ValueError("Preview start must be >= 0 and duration must be in (0, 300] seconds")
if len(set(args.profiles)) != len(args.profiles):
raise ValueError("Preview profiles must be unique")
if any(item.partition("=")[0].strip() == "profile" for item in args.set):
raise ValueError("Preview profiles come from --profiles; remove the profile override from --set")
source = args.inputs[0]
require_input(source)
original = source.resolve()
output_dir = args.output_dir or source.parent
# Preflight every preview destination before extracting the excerpt.
claimed: set[Path] = set()
for profile in args.profiles:
output = output_dir / f"{source.stem}.{profile}.preview.wav"
if output.resolve() == original:
raise ValueError("Preview output would overwrite the original")
targets = [output, output.with_suffix(".report.json")]
resolved = {target.resolve() for target in targets}
if len(resolved) < len(targets) or resolved & claimed:
raise ValueError("Preview outputs have duplicate names")
claimed |= resolved
for target in targets:
check_target(original, target, args.overwrite)
output_dir.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix=".preview-", dir=output_dir) as tmp:
excerpt = Path(tmp) / "excerpt.wav"
ffmpeg_run(ensure_ffmpeg(), ["-ss", str(args.start), "-i", str(original), "-t", str(args.duration),
"-map", "0:a:0", "-c:a", "pcm_f32le", str(excerpt)], "Extracting preview", args.duration, display.update)
for profile in args.profiles:
configured = settings_for(args, profile)
# Equal loudness prevents a louder preset winning an unfair A/B.
configured.normalize = True
configured.limiter = True
output = output_dir / f"{source.stem}.{profile}.preview.wav"
report = process(excerpt, output, configured, display.update, args.overwrite,
provenance={"original": str(original), "preview_start_seconds": args.start,
"requested_duration_seconds": args.duration,
"normalization_forced": True})
console.print(f"Preview: {output} | {report['master_loudness']['input_i']} LUFS | {report['master_loudness']['input_tp']} dBTP", markup=False)
for warning in report["warnings"]:
console.print(f"Warning: {warning}", markup=False)
except KeyboardInterrupt:
console.print("Cancelled; temporary files removed. Completed outputs may remain.")
raise SystemExit(130)
except (ValueError, OSError, RuntimeError, subprocess.SubprocessError) as error:
console.print(f"Error: {error}", markup=False)
raise SystemExit(1)
if __name__ == "__main__":
main()
|
