Skip to content

Core Utilities

Small, focused modules that every higher-level LunaVox component depends on. Each one is a single-point-of-truth enforced by AGENT.md rules: other code imports from here rather than probing the OS, re-reading paths, or spinning up Rich consoles directly.

Platform detection

platform

Host-platform helpers.

All sys.platform / platform.system() branches in the Python side live here. The rule mirrors the C++ side: other modules must not probe the OS directly — they import from here so the selection logic stays in one file and future platforms only require one edit.

The build.factory / build.get_*_class entry points are the single allowed exception because they bind a host tag to an entire Builder class rather than to a scalar.

shared_lib_name

shared_lib_name(stem: str) -> str

Return the platform-appropriate shared-library filename for stem.

Windows: {stem}.dll; macOS: lib{stem}.dylib; Linux / other POSIX: lib{stem}.so.

Source code in src/lunavox/core/platform.py
def shared_lib_name(stem: str) -> str:
    """Return the platform-appropriate shared-library filename for ``stem``.

    Windows: ``{stem}.dll``; macOS: ``lib{stem}.dylib``; Linux / other POSIX:
    ``lib{stem}.so``.
    """
    if sys.platform == "win32":
        return f"{stem}.dll"
    if sys.platform == "darwin":
        return f"lib{stem}.dylib"
    return f"lib{stem}.so"

executable_suffix

executable_suffix() -> str

Empty on POSIX, .exe on Windows. Useful when resolving binary paths like build/lunavox-cli.

Source code in src/lunavox/core/platform.py
def executable_suffix() -> str:
    """Empty on POSIX, ``.exe`` on Windows. Useful when resolving binary
    paths like ``build/lunavox-cli``."""
    return ".exe" if sys.platform == "win32" else ""

is_windows

is_windows() -> bool
Source code in src/lunavox/core/platform.py
def is_windows() -> bool:
    return sys.platform == "win32"

is_macos

is_macos() -> bool
Source code in src/lunavox/core/platform.py
def is_macos() -> bool:
    return sys.platform == "darwin"

is_linux

is_linux() -> bool
Source code in src/lunavox/core/platform.py
def is_linux() -> bool:
    return sys.platform.startswith("linux")

Project root resolution

resolve_project_root

resolve_project_root(explicit_root: Path | None = None) -> Path

Find and return the project root directory, creating 'models' and 'lib' if missing.

Source code in src/lunavox/core/project.py
def resolve_project_root(explicit_root: Path | None = None) -> Path:
    """Find and return the project root directory, creating 'models' and 'lib' if missing."""
    candidate = None

    if explicit_root is not None:
        candidate = explicit_root.resolve()
        if not _looks_like_project_root(candidate):
            raise RuntimeError(f"Invalid --project-root: {candidate}")
    else:
        env_root = os.environ.get("LUNAVOX_PROJECT_ROOT", "").strip()
        if env_root:
            env_candidate = Path(env_root).resolve()
            if _looks_like_project_root(env_candidate):
                candidate = env_candidate

        if candidate is None:
            cwd = Path.cwd().resolve()
            probe_paths = [cwd, *cwd.parents]
            for p in probe_paths:
                if _looks_like_project_root(p):
                    candidate = p
                    break

    if candidate is None:
        raise RuntimeError(
            "Could not locate LunaVox project root. Run inside the repository or pass --project-root."
        )

    # Ensure required runtime directories exist
    for sub in ["models", "lib"]:
        target = candidate / sub
        if not target.exists():
            target.mkdir(parents=True, exist_ok=True)

    return candidate

Dependency management

DependencyPolicy dataclass

DependencyPolicy(yes: bool = False, no_install: bool = False)

ensure_dependency_group

ensure_dependency_group(
    group: str, policy: DependencyPolicy, project_root: Path | None = None
) -> None
Source code in src/lunavox/core/deps.py
def ensure_dependency_group(
    group: str, policy: DependencyPolicy, project_root: Path | None = None
) -> None:
    if group not in DEPENDENCY_GROUPS:
        raise RuntimeError(f"Unknown dependency group: {group}")

    missing = missing_modules(DEPENDENCY_GROUPS[group])
    if not missing:
        return

    missing_text = ", ".join(f"{name} ({pip_name})" for name, pip_name in missing)
    install_cmd = f'{sys.executable} -m pip install "lunavox[{group}]"'
    fallback_cmd = ""
    if project_root is not None and (project_root / "pyproject.toml").exists():
        fallback_cmd = f'{sys.executable} -m pip install ".[{group}]"'
    install_cmd_display = install_cmd.replace("[", "\\[").replace("]", "\\]")

    if policy.no_install:
        raise RuntimeError(
            "Missing required dependencies: "
            + missing_text
            + "\nAuto install disabled by --no-install.\nInstall manually:\n  "
            + install_cmd
            + (f"\nOr from local project:\n  {fallback_cmd}" if fallback_cmd else "")
        )

    non_interactive = not (sys.stdin.isatty() and sys.stdout.isatty())
    if non_interactive and not policy.yes:
        raise RuntimeError(
            "Missing required dependencies: "
            + missing_text
            + "\nNon-interactive mode detected. Re-run with --yes or install manually:\n  "
            + install_cmd
            + (f"\nOr from local project:\n  {fallback_cmd}" if fallback_cmd else "")
        )

    should_install = policy.yes
    if not should_install:
        should_install = Confirm.ask(
            f"Missing dependencies for '{group}': {missing_text}\nInstall now using `{install_cmd_display}`?",
            default=True,
        )

    if not should_install:
        raise RuntimeError(
            "Dependency installation declined.\nInstall manually:\n  "
            + install_cmd
            + (f"\nOr from local project:\n  {fallback_cmd}" if fallback_cmd else "")
        )

    env = os.environ.copy()
    try:
        # Prioritize local project installation if we are in the project root
        if fallback_cmd:
            subprocess.run(
                [sys.executable, "-m", "pip", "install", f".[{group}]"],
                check=True,
                env=env,
                cwd=str(project_root),
            )
        else:
            subprocess.run(
                [sys.executable, "-m", "pip", "install", f"lunavox[{group}]"],
                check=True,
                env=env,
            )
    except subprocess.CalledProcessError:
        # Fallback to name-based install if local fail (or vice versa if we swapped them)
        if not fallback_cmd:
            raise
        subprocess.run(
            [sys.executable, "-m", "pip", "install", f"lunavox[{group}]"],
            check=True,
            env=env,
        )

    still_missing = missing_modules(DEPENDENCY_GROUPS[group])
    if still_missing:
        still_text = ", ".join(f"{name} ({pip_name})" for name, pip_name in still_missing)
        raise RuntimeError(
            "Dependency installation finished but modules are still missing: " + still_text
        )
    console.print(f"[success]Dependencies for '{group}' are ready.[/success]")

has_module

has_module(name: str) -> bool
Source code in src/lunavox/core/deps.py
def has_module(name: str) -> bool:
    return importlib.util.find_spec(name) is not None

missing_modules

missing_modules(modules: Iterable[tuple[str, str]]) -> list[tuple[str, str]]
Source code in src/lunavox/core/deps.py
def missing_modules(modules: Iterable[tuple[str, str]]) -> list[tuple[str, str]]:
    return [(name, pip_name) for name, pip_name in modules if not has_module(name)]

Session logging

logging

Unified process-wide logger for the Python side.

Every writer — CLI session header, build driver stage output, conversion pipeline subprocess capture, GUI-bridged C++ log callback — appends to the same logs/latest.log through this module. A single threading.Lock serializes writes so concurrent tasks (e.g. Rich status spinner + build subprocess + model download) cannot corrupt each other.

Design notes:

  • The log file is a plain append stream. No level filtering and no handler chain; the consumer decides what to write, we just persist.
  • session_start(path) truncates and writes a session header. The CLI calls this once per invocation; nothing else should.
  • Writers call append(text). The module is a no-op until session_start has been called, so importing the logger from a library module is safe even in contexts where no log file exists (tests, sdist builds, IDE completion).
  • Keep this file dependency-free: no Rich, no lunavox imports. The UI console lives in core.ui and is separate on purpose — users can silence the console without losing the log, and vice versa.

session_start

session_start(path: Path, header: str = '') -> None

Truncate path and register it as the process-wide log file.

Subsequent :func:append calls will go here. Safe to call multiple times per process — the last call wins (CLI does this once at startup; tests may rebind to a tmp path).

Source code in src/lunavox/core/logging.py
def session_start(path: Path, header: str = "") -> None:
    """Truncate ``path`` and register it as the process-wide log file.

    Subsequent :func:`append` calls will go here. Safe to call multiple
    times per process — the last call wins (CLI does this once at
    startup; tests may rebind to a tmp path).
    """
    global _log_path
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with _lock:
        _log_path = path
        with open(path, "w", encoding="utf-8") as f:
            if header:
                f.write(header)
                if not header.endswith("\n"):
                    f.write("\n")

append

append(text: str) -> None

Append text to the active session log. No-op if unbound.

Source code in src/lunavox/core/logging.py
def append(text: str) -> None:
    """Append ``text`` to the active session log. No-op if unbound."""
    if _log_path is None:
        return
    with _lock, open(_log_path, "a", encoding="utf-8") as f:
        f.write(text)
        if not text.endswith("\n"):
            f.write("\n")