path: root/app/backends/audiocpp/build.py

"""Checkout lifecycle: clone location, ggml patches, binary build, uninstall."""

import contextlib
import io
import re
import shlex
import shutil
from datetime import datetime
from pathlib import Path
from typing import List, Optional

from backends import common, servers
from backends.common import APP_DIR
from .catalog import _BACKEND_TOKEN_RE
from .constants import AUDIOCPP_DIR_NAME, PATCH_DIR

def uninstall(*, emit=None, cancel=None) -> int:
    """Remove the audio.cpp backend entirely: stop its server, delete the checkout.

    The checkout (``app/audio.cpp``) holds the built binary, the downloaded
    models, and the server.json, so removing the directory uninstalls the
    backend. A running server this tool started is stopped first
    (best-effort).

    EMIT is accepted for registry symmetry with the other backends but is
    unused here — this uninstall has no subprocess phase, and its prints are
    captured by the task view when run in the TUI. CANCEL is a
    ``threading.Event`` honored between phases only (after the server has
    been stopped, before the checkout is deleted), so a started phase always
    completes and the uninstall never tears halfway. Returns the exit code
    (130 when cancelled before a remaining phase).
    """
    # Only stop when a pid file exists: without one this tool never
    # started the server, so the "not started by this tool" notice would
    # be uninstall-time noise.
    if servers.pid_for("audiocpp") is not None:
        servers.stop("audiocpp")
    if common.cancel_requested(cancel):
        return 130
    checkout = find_local_checkout()
    if checkout is None:
        print("[INFO] No audio.cpp checkout to remove.")
        return 0
    print(f"[INFO] Removing audio.cpp checkout {checkout}...")
    shutil.rmtree(checkout, ignore_errors=True)
    print("[OK] audio.cpp removed.")
    return 0


def find_local_checkout() -> Optional[Path]:
    """Return the managed audio.cpp checkout at ``app/audio.cpp``.

    Returns the path only when it contains a ``model_specs`` directory;
    the checkout is installed there by the setup wizard and nowhere else.
    """
    try:
        resolved = (APP_DIR / AUDIOCPP_DIR_NAME).resolve()
    except OSError:
        return None
    if (resolved / "model_specs").is_dir():
        return resolved
    return None


def find_audiocpp_server_bin(audiocpp_dir: Path) -> Optional[Path]:
    """Return the built audiocpp_server binary, or None when not built.

    Scans ``audiocpp_dir/build/*`` for a build directory containing
    ``bin/audiocpp_server`` (``.exe`` allowed on Windows). When several
    builds exist the first (alphabetical) is returned.
    """
    build_root = audiocpp_dir / "build"
    if not build_root.is_dir():
        return None
    try:
        build_dirs = sorted(build_root.iterdir(),
                            key=lambda p: p.name.lower())
    except OSError:
        return None
    for build_dir in build_dirs:
        if not build_dir.is_dir():
            continue
        for name in ("audiocpp_server", "audiocpp_server.exe"):
            server = build_dir / "bin" / name
            if server.exists():
                return server
    return None


def built_server_binary(audiocpp_dir: Path, backend: str) -> Optional[Path]:
    """Return the built audiocpp_server for BACKEND, or None.

    Like ``find_audiocpp_server_bin`` but limited to build directories whose
    name carries the BACKEND token (``-cuda-``, ``-vulkan-``, ``-hip-``,
    ``-cpu-``; ``-metal-`` counts as ``cpu``). A checkout with builds for
    several backends is asked which one to use without re-offering a build
    for a backend that is already built.
    """
    build_root = audiocpp_dir / "build"
    if not build_root.is_dir():
        return None
    try:
        build_dirs = sorted(build_root.iterdir(),
                            key=lambda p: p.name.lower())
    except OSError:
        return None
    for build_dir in build_dirs:
        if not build_dir.is_dir():
            continue
        match = _BACKEND_TOKEN_RE.search(build_dir.name.lower())
        if not match:
            continue
        token = "cpu" if match.group(1) == "metal" else match.group(1)
        if token != backend:
            continue
        for name in ("audiocpp_server", "audiocpp_server.exe"):
            server = build_dir / "bin" / name
            if server.exists():
                return server
    return None


def find_build_script(audiocpp_dir: Path) -> Optional[Path]:
    """Return the audio.cpp build helper script to run, or None.

    Prefers ``scripts/build_linux.sh``; otherwise the first
    ``scripts/build_*.sh`` it finds. (Windows ``.bat`` scripts are not run
    automatically — build manually there.)
    """
    scripts = audiocpp_dir / "scripts"
    if not scripts.is_dir():
        return None
    preferred = scripts / "build_linux.sh"
    if preferred.exists():
        return preferred
    try:
        candidates = sorted(scripts.glob("build_*.sh"),
                            key=lambda p: p.name.lower())
    except OSError:
        return None
    return candidates[0] if candidates else None


GGML_PATCHES = [
    {
        "file": "ggml-top-k-cuda-iterator.patch",
        "target": "external/ggml/src/ggml-cuda/top-k.cu",
        "marker": r"#\s*include\s*<cuda/iterator>",
        "label": "top-k.cu: add #include <cuda/iterator> (CCCL 3.x build fix)",
    },
]


def apply_ggml_patches(audiocpp_dir: Path, *, emit=None, cancel=None) -> int:
    """Apply the shipped ggml build patches to an audio.cpp checkout.

    Idempotent: a patch whose marker already matches its target is skipped
    (it is either already applied, or the fork re-vendored a fixed ggml). A
    patch that no longer applies because the vendored file changed shape is a
    loud, non-interactive failure — the build is aborted so the user
    re-evaluates the patch instead of hitting a known nvcc break minutes
    later. Returns 0 when every patch is applied or already present, 1 on
    drift, 130 when cancelled.
    """
    for patch in GGML_PATCHES:
        if cancel is not None and cancel.is_set():
            return 130
        target = audiocpp_dir / patch["target"]
        if not target.is_file():
            print(f"[INFO] {patch['file']}: target {patch['target']} not "
                  f"present in this checkout; skipping")
            continue
        try:
            text = target.read_text(encoding="utf-8", errors="ignore")
        except OSError as exc:
            print(f"[WARNING] {patch['file']}: could not read {target}: "
                  f"{exc}; skipping")
            continue
        if re.search(patch["marker"], text):
            print(f"[OK] {patch['file']}: fix already present, skipping")
            continue
        patch_path = PATCH_DIR / patch["file"]
        if not patch_path.is_file():
            print(f"[ERROR] {patch['file']}: patch file not found at "
                  f"{patch_path}; cannot apply")
            return 1
        check_argv = ["git", "-C", str(audiocpp_dir), "apply", "--check",
                      "--whitespace=nowarn", str(patch_path)]
        check_rc = common.run_console_subprocess(
            check_argv, emit=emit, cancel=cancel)
        if check_rc == 130 or (cancel is not None and cancel.is_set()):
            return 130
        if check_rc != 0:
            print(f"[ERROR] {patch['file']}: no longer applies to "
                  f"{patch['target']} (git apply --check exit {check_rc}). "
                  f"The audio.cpp fork's vendored ggml changed shape and "
                  f"still lacks the fix. Re-evaluate {patch_path}: "
                  f"regenerate the patch, or drop this entry if the fork "
                  f"now ships the fix.")
            return 1
        apply_argv = ["git", "-C", str(audiocpp_dir), "apply",
                      "--whitespace=nowarn", str(patch_path)]
        rc = common.run_console_subprocess(
            apply_argv, emit=emit, cancel=cancel)
        if rc == 130 or (cancel is not None and cancel.is_set()):
            return 130
        if rc != 0:
            print(f"[ERROR] {patch['file']}: git apply failed (exit {rc})")
            return rc
        print(f"[OK] {patch['file']}: applied ({patch['label']})")
    return 0


def build_audiocpp(audiocpp_dir: Path, backend: str, *,
                   emit=None, cancel=None) -> int:
    """Build audiocpp_server for BACKEND, streaming output.

    With EMIT None the build script runs on the console (inherits the
    terminal); with EMIT given (the in-TUI task view) its output streams line
    by line to EMIT so the view can show progress, and CANCEL aborts it.

    On the EMIT (TUI) path the build output is also tee'd to
    ``app/logs/audiocpp_build_<timestamp>.log`` so it survives the curses
    session; when the build fails (and was not cancelled) a post-TUI notice
    with the copy-pastable command and the log path is queued for the console
    (see ``backends.common.record_post_tui_notice``).

    Returns the build script's exit code (non-zero when the script is
    missing).
    """
    script = find_build_script(audiocpp_dir)
    if script is None:
        message = (f"[ERROR] No build script found in {audiocpp_dir}/scripts; "
                   "build audiocpp_server manually (see the audio.cpp README)")
        print(message)
        if emit is not None:
            common.record_post_tui_notice(message)
        return 1
    argv = ["sh", str(script), "--backend", backend, "--target",
            "audiocpp_server", "--deployment-build"]
    command = f"cd {audiocpp_dir} && {shlex.join(argv)}"
    if emit is None:
        print(f"[INFO] Building audiocpp_server for {backend} ({command})...")
        patch_rc = apply_ggml_patches(audiocpp_dir, cancel=cancel)
        if patch_rc == 130 or (cancel is not None and cancel.is_set()):
            return 130
        if patch_rc != 0:
            print("[ERROR] ggml build patches could not be applied; "
                  "aborting audiocpp_server build. See the messages above "
                  "and re-evaluate app/backends/patches/.")
            return patch_rc
        return common.run_console_subprocess(argv, cwd=audiocpp_dir)
    return _build_audiocpp_tui(emit, cancel, argv, command, audiocpp_dir)


def _build_audiocpp_tui(emit, cancel, argv: List[str], command: str,
                        audiocpp_dir: Path) -> int:
    """Run the build on the TUI path: tee output to a log file.

    The ggml patch step runs first, inside the same log: every emitted
    line (patch status, build output) is also written (and flushed) to
    ``app/logs/audiocpp_build_<timestamp>.log``. On failure a summary (the
    copy-pastable COMMAND and the log path) is emitted into the TUI,
    written to the log, and queued as a post-TUI console notice. A
    cancelled build (CANCEL set) is not reported as a failure, but its
    partial output stays in the log file.
    """
    log_path = common.LOG_DIR / (
        f"audiocpp_build_{datetime.now():%Y%m%d_%H%M%S}.log")
    log_path.parent.mkdir(parents=True, exist_ok=True)
    log_handle = log_path.open("w", encoding="utf-8")

    def tee(line: str) -> None:
        log_handle.write(line + "\n")
        log_handle.flush()
        emit(line)

    class _TeeWriter(io.TextIOBase):
        """Route print() output from the patch step into the log too."""

        def write(self, s: str) -> int:
            for line in s.splitlines():
                if line:
                    tee(line)
            return len(s)

    try:
        with contextlib.redirect_stdout(_TeeWriter()):
            patch_rc = apply_ggml_patches(audiocpp_dir, emit=tee,
                                          cancel=cancel)
        if patch_rc == 130 or (cancel is not None and cancel.is_set()):
            return 130
        if patch_rc != 0:
            notice = ("[ERROR] ggml build patches could not be applied; "
                      "aborting audiocpp_server build. See the messages "
                      "above and re-evaluate app/backends/patches/.")
            tee(notice)
            common.record_post_tui_notice(notice)
            return patch_rc
        tee(f"[INFO] Building audiocpp_server ({command})...")
        rc = common.run_console_subprocess(
            argv, cwd=audiocpp_dir, emit=tee, cancel=cancel)
        if rc != 0 and (cancel is None or not cancel.is_set()):
            notice = (f"[ERROR] audio.cpp build failed (exit code {rc}).\n"
                      f"  Build log: {log_path}\n"
                      f"  Troubleshoot by re-running this command:\n"
                      f"    {command}")
            for line in notice.splitlines():
                tee(line)
            common.record_post_tui_notice(notice)
    finally:
        log_handle.close()
    return rc


def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None:
    """Print remediation when audiocpp_server is missing (troubleshooting).

    The hub starts and stops the server itself, so a working install gets
    no manual launch instructions. When no binary was built, though, the
    user needs to know how to build and run it by hand. The commands are
    prefixed with ``cd <checkout> &&`` because the server discovers
    model_specs/<family>.json relative to its working directory.
    """
    if find_audiocpp_server_bin(audiocpp_dir) is not None:
        return
    print("\n[INFO] audiocpp_server binary not found. Build it first, e.g.:")
    script = find_build_script(audiocpp_dir)
    if script is not None:
        print(f"  sh {script} --backend <cuda|vulkan|hip|cpu> "
              "--target audiocpp_server --deployment-build")
    print(f"  then run: cd {audiocpp_dir} && ./build/<platform>-<backend>"
          f"-release/bin/audiocpp_server --config {output_path}")