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