slop rewriteHEADmain

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()