MeshChatX/meshchatx/android_codec2.py

# SPDX-License-Identifier: 0BSD

"""Load Codec2 native libraries before pycodec2/LXST on Chaquopy Android."""

from __future__ import annotations

import ctypes
import importlib
import logging
import os
import sys
from pathlib import Path

logger = logging.getLogger(__name__)

_codec2_preload_error: str | None = None
_codec2_preload_ok = False
_codec2_preload_attempted = False


def _is_chaquopy_android() -> bool:
    try:
        import java  # noqa: F401
    except ImportError:
        return False
    return True


def _cdll_load(path_or_name: str):
    """Load a shared library with RTLD_GLOBAL when the platform supports it.

    pycodec2.so declares NEEDED libcodec2.so. Loading with RTLD_GLOBAL lets the
    later dlopen of the extension resolve that dependency.
    """
    mode = getattr(ctypes, "RTLD_GLOBAL", None)
    if mode is None:
        return ctypes.CDLL(path_or_name)
    return ctypes.CDLL(path_or_name, mode=mode)


def _libcodec2_candidates() -> list[Path]:
    """Return candidate paths for libcodec2.so without importing pycodec2.

    import pycodec2 loads the extension which already needs libcodec2.so.
    Searching sys.path on disk avoids that chicken-and-egg failure.
    """
    candidates: list[Path] = []
    seen: set[str] = set()

    def add(path: Path) -> None:
        key = str(path)
        if key in seen:
            return
        seen.add(key)
        candidates.append(path)

    explicit = Path(os.environ.get("MESHCHAT_LIBCODEC2_PATH", "") or "")
    if explicit.is_file():
        add(explicit)

    for entry in sys.path:
        if not entry:
            continue
        root = Path(entry)
        add(root / "pycodec2" / "libcodec2.so")
        add(root / "chaquopy" / "lib" / "libcodec2.so")
        # Some Chaquopy layouts nest native libs under site-packages directly.
        add(root / "libcodec2.so")

    # Extracted APK native lib dirs (ABI-specific jniLibs sync target).
    for env_key in ("MESHCHAT_NATIVE_LIB_DIR", "ANDROID_NATIVE_LIBRARY_DIR"):
        native_dir = Path(os.environ.get(env_key, "") or "")
        if native_dir.is_dir():
            add(native_dir / "libcodec2.so")

    return candidates


def _java_system_load_library(name: str = "codec2") -> bool:
    """Ask the Android runtime to load a jniLibs shared object by soname."""
    try:
        from java.lang import System as JavaSystem

        JavaSystem.loadLibrary(name)
        return True
    except Exception as exc:
        logger.debug("Java System.loadLibrary(%s) failed: %s", name, exc)
        return False


def _java_system_load_absolute(path: Path) -> bool:
    """Load an absolute .so path via Java System.load."""
    try:
        from java.lang import System as JavaSystem

        JavaSystem.load(str(path))
        return True
    except Exception as exc:
        logger.debug("Java System.load(%s) failed: %s", path, exc)
        return False


def ensure_codec2_native_library(*, force: bool = False) -> bool:
    """Preload libcodec2.so so import pycodec2 works on Android.

    Chaquopy installs chaquopy-libcodec2 separately from pycodec2. The
    extension module only declares a NEEDED entry for libcodec2.so. Without
    preloading or bundling the shared library next to pycodec2.so, imports
    fail at runtime with dlopen errors.

    Always prefer absolute-path loads with RTLD_GLOBAL even after a successful
    Java System.loadLibrary. On some Android or Chaquopy linker setups the
    Java load alone is not enough for the Python extension dlopen.
    """
    global _codec2_preload_attempted, _codec2_preload_error, _codec2_preload_ok

    if _codec2_preload_ok and not force:
        return True
    if _codec2_preload_attempted and not force and _codec2_preload_error is not None:
        # Previous attempt failed. Allow one more try when force=True only.
        return False

    _codec2_preload_attempted = True
    _codec2_preload_error = None

    if not _is_chaquopy_android():
        _codec2_preload_ok = True
        return True

    loaded_any = False
    if _java_system_load_library("codec2"):
        logger.info("Loaded Codec2 via Java System.loadLibrary(codec2)")
        loaded_any = True

    try:
        _cdll_load("libcodec2.so")
        loaded_any = True
    except OSError:
        pass

    last_error: str | None = None
    for lib_path in _libcodec2_candidates():
        if not lib_path.is_file():
            continue
        if _java_system_load_absolute(lib_path):
            loaded_any = True
            logger.info("Loaded Codec2 via Java System.load(%s)", lib_path)
        try:
            _cdll_load(str(lib_path))
            logger.info("Loaded Codec2 native library from %s", lib_path)
            loaded_any = True
            break
        except OSError as exc:
            last_error = f"{lib_path}: {exc}"

    if loaded_any:
        _codec2_preload_ok = True
        _codec2_preload_error = None
        return True

    _codec2_preload_ok = False
    _codec2_preload_error = last_error or "libcodec2.so not found on Android"
    logger.warning("Codec2 native preload failed: %s", _codec2_preload_error)
    return False


def _install_pycodec2_ctypes_fallback() -> tuple[bool, str | None]:
    """Install a ctypes pycodec2 stand-in when the Cython extension cannot load.

    Some Android vendor wheels ship an empty pycodec2.so (no PyInit). libcodec2.so
    is still present. Expose the same Codec2 API LXST expects via ctypes.
    """
    try:
        from meshchatx import pycodec2_ctypes

        ok, err = pycodec2_ctypes.probe()
        if not ok:
            return False, err
        # Drop a broken partial import so LXST picks up the stand-in.
        sys.modules.pop("pycodec2", None)
        sys.modules["pycodec2"] = pycodec2_ctypes
        logger.warning(
            "Using ctypes Codec2 fallback (native pycodec2 extension unavailable)"
        )
        return True, None
    except Exception as exc:
        return False, str(exc)


def probe_pycodec2() -> tuple[bool, str | None]:
    """Import pycodec2 after preload and report whether Codec2 works."""
    if _is_chaquopy_android() and not ensure_codec2_native_library():
        # Retry once in case native libs appeared after an early failed attempt.
        if not ensure_codec2_native_library(force=True):
            # Still try ctypes against whatever path we can find.
            return _install_pycodec2_ctypes_fallback()
    try:
        import pycodec2

        c2 = pycodec2.Codec2(1600)
        _ = c2.samples_per_frame()
        return True, None
    except Exception as exc:
        native_error = str(exc)
        if _is_chaquopy_android():
            ok, fallback_error = _install_pycodec2_ctypes_fallback()
            if ok:
                return True, None
            return False, fallback_error or native_error
        return False, native_error


def ensure_lxst_codec2_binding() -> bool:
    """Ensure LXST.Codecs.Codec2 is bound after a successful pycodec2 probe.

    Soft-import patches set Codec2 to None when the first LXST.Codecs import
    fails. Reload Codecs after preload so telephony can use Codec2 modes.
    """
    ok, _err = probe_pycodec2()
    if not ok:
        return False
    try:
        codecs_mod = importlib.import_module("LXST.Codecs")

        if getattr(codecs_mod, "Codec2", None) is not None:
            return True
        codecs_mod = importlib.reload(codecs_mod)
        if getattr(codecs_mod, "Codec2", None) is None:
            return False
        # Refresh Telephony bindings that may have imported Codec2 as None.
        try:
            telephony_mod = importlib.import_module("LXST.Primitives.Telephony")
            importlib.reload(telephony_mod)
        except Exception as tel_exc:
            logger.debug("LXST Telephony reload skipped: %s", tel_exc)
        return True
    except Exception as exc:
        logger.warning("LXST Codec2 binding failed: %s", exc)
        return False


def codec2_preload_error() -> str | None:
    """Return the last preload failure message, if any."""
    return _codec2_preload_error


def reset_codec2_preload_state_for_tests() -> None:
    """Clear preload memoization (tests only)."""
    global _codec2_preload_attempted, _codec2_preload_error, _codec2_preload_ok
    _codec2_preload_attempted = False
    _codec2_preload_ok = False
    _codec2_preload_error = None