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