modeling_fastplms.py

"""Generated bridge to the unchanged FastPLMs package sources."""

import base64
import hashlib
import importlib
import importlib.util
import sys
import tempfile
from importlib.metadata import PackageNotFoundError, distribution
from io import BytesIO
from pathlib import Path
from zipfile import ZIP_DEFLATED, ZipFile

from .fastplms_bundle import RUNTIME_DATA, RUNTIME_HASH

if RUNTIME_HASH != "17b8f83a33e63d941e3edfb8db2d8381286046b572f94194e576ca394d997597":
    raise RuntimeError("FastPLMs runtime identity differs from the bridge.")

_RUNTIME_TEMPORARIES = []

def _archive_runtime_hashes(payload):
    result = {}
    with ZipFile(BytesIO(payload)) as archive:
        for member in archive.infolist():
            name = member.filename
            parts = Path(name).parts
            if (
                member.is_dir()
                or "\\" in name
                or not parts
                or parts[0] != "fastplms"
                or len(parts) < 2
                or any(part in {"", ".", ".."} for part in parts)
                or Path(name).suffix in {".pyc", ".pyo"}
                or member.flag_bits & 0x1
                or member.compress_type != ZIP_DEFLATED
                or member.external_attr >> 16 != 0o100644
            ):
                raise RuntimeError("Embedded FastPLMs archive has an unsafe path.")
            relative = Path(*parts[1:]).as_posix()
            if relative in result:
                raise RuntimeError("Embedded FastPLMs archive repeats a path.")
            result[relative] = hashlib.sha256(archive.read(member)).hexdigest()
    return result

def _ensure_runtime():
    payload = base64.b85decode("".join(RUNTIME_DATA))
    if hashlib.sha256(payload).hexdigest() != RUNTIME_HASH:
        raise RuntimeError("Embedded FastPLMs runtime hash mismatch.")
    expected = _archive_runtime_hashes(payload)
    temporary = tempfile.TemporaryDirectory(prefix="fastplms-artifact-runtime-")
    try:
        runtime_root = Path(temporary.name)
        with ZipFile(BytesIO(payload)) as archive:
            for member in archive.infolist():
                target = runtime_root.joinpath(*Path(member.filename).parts)
                target.parent.mkdir(parents=True, exist_ok=True)
                with target.open("xb") as handle:
                    handle.write(archive.read(member))
        package_root = runtime_root / "fastplms"
        if _runtime_file_hashes(package_root) != expected:
            raise RuntimeError(
                "Private FastPLMs runtime differs from the embedded archive."
            )
    except BaseException:
        temporary.cleanup()
        raise
    _RUNTIME_TEMPORARIES.append(temporary)
    return package_root

def _runtime_file_hashes(package_root):
    result = {}
    for path in sorted(package_root.rglob("*")):
        relative = path.relative_to(package_root)
        if path.is_symlink():
            raise RuntimeError("Private FastPLMs runtime contains a symlink.")
        if path.is_dir():
            continue
        if path.suffix in {".pyc", ".pyo"}:
            raise RuntimeError("Private FastPLMs runtime contains bytecode.")
        if not path.is_file():
            raise RuntimeError("Private FastPLMs runtime contains a non-file entry.")
        result[relative.as_posix()] = hashlib.sha256(path.read_bytes()).hexdigest()
    return result

def _installed_runtime_digest(installed_root, relative):
    candidate = installed_root / relative
    if candidate.is_file():
        return hashlib.sha256(candidate.read_bytes()).hexdigest()
    if relative != "kernels.lock":
        return None
    try:
        installed_distribution = distribution("fastplms")
    except PackageNotFoundError:
        return None
    for entry in installed_distribution.files or ():
        normalized = str(entry).replace("\\", "/")
        if normalized.endswith(".dist-info/kernels.lock"):
            lock_path = Path(installed_distribution.locate_file(entry))
            if lock_path.is_file():
                return hashlib.sha256(lock_path.read_bytes()).hexdigest()
    return None

def _extend_loaded_package_paths(package_root):
    for name, module in list(sys.modules.items()):
        if name != "fastplms" and not name.startswith("fastplms."):
            continue
        paths = getattr(module, "__path__", None)
        if paths is None:
            continue
        relative = name.split(".")[1:]
        candidate = package_root.joinpath(*relative)
        candidate_text = str(candidate)
        if candidate.is_dir() and candidate_text not in paths:
            paths.append(candidate_text)

def _merge_runtime(installed, package_root):
    incoming = _runtime_file_hashes(package_root)
    known = dict(getattr(installed, "__fastplms_artifact_runtime_files__", {}))
    installed_root_text = getattr(
        installed, "__fastplms_artifact_installed_root__", None
    )
    if not known:
        installed_file = getattr(installed, "__file__", None)
        if installed_file is None:
            raise RuntimeError(
                "The loaded fastplms package has no source path and cannot be verified "
                "against the embedded artifact runtime."
            )
        installed_root = Path(installed_file).resolve().parent
        for relative, digest in incoming.items():
            if _installed_runtime_digest(installed_root, relative) != digest:
                raise RuntimeError(
                    "The installed FastPLMs runtime differs from this artifact at "
                    f"{relative!r}. Install the artifact's matching FastPLMs release "
                    "or use a separate Python process."
                )
        installed_root_text = str(installed_root)
        installed.__fastplms_artifact_installed_root__ = installed_root_text
    conflicts = sorted(
        relative
        for relative, digest in incoming.items()
        if relative in known and known[relative] != digest
    )
    if conflicts:
        raise RuntimeError(
            "FastPLMs artifacts contain incompatible runtime sources at "
            + ", ".join(repr(path) for path in conflicts[:5])
            + ". Load incompatible releases in separate Python processes."
        )
    if installed_root_text is not None:
        installed_root = Path(installed_root_text)
        for relative, digest in incoming.items():
            if relative in known:
                continue
            if _installed_runtime_digest(installed_root, relative) != digest:
                raise RuntimeError(
                    "The installed FastPLMs runtime differs from this artifact at "
                    f"{relative!r}. Install the artifact's matching FastPLMs release "
                    "or use a separate Python process."
                )
    known.update(incoming)
    installed.__fastplms_artifact_runtime_files__ = known
    roots = list(getattr(installed, "__fastplms_artifact_runtime_roots__", ()))
    if str(package_root) not in roots:
        roots.append(str(package_root))
    installed.__fastplms_artifact_runtime_roots__ = tuple(roots)
    temporaries = list(
        getattr(installed, "__fastplms_artifact_runtime_temporaries__", ())
    )
    for temporary in _RUNTIME_TEMPORARIES:
        if temporary not in temporaries:
            temporaries.append(temporary)
    installed.__fastplms_artifact_runtime_temporaries__ = tuple(temporaries)
    hashes = set(getattr(installed, "__fastplms_artifact_runtime_hashes__", ()))
    hashes.add(RUNTIME_HASH)
    installed.__fastplms_artifact_runtime_hashes__ = frozenset(hashes)
    _extend_loaded_package_paths(package_root)
    return installed

def _import_without_bytecode(module_name):
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        return importlib.import_module(module_name)
    finally:
        sys.dont_write_bytecode = previous

def _install_runtime():
    installed = sys.modules.get("fastplms")
    hashes = getattr(installed, "__fastplms_artifact_runtime_hashes__", ())
    if RUNTIME_HASH in hashes:
        return installed
    package_root = _ensure_runtime()
    if installed is not None:
        return _merge_runtime(installed, package_root)
    spec = importlib.util.spec_from_file_location(
        "fastplms",
        package_root / "__init__.py",
        submodule_search_locations=[str(package_root)],
    )
    if spec is None or spec.loader is None:
        raise ImportError("Unable to load the embedded FastPLMs runtime.")
    package = importlib.util.module_from_spec(spec)
    package.__fastplms_artifact_runtime_hash__ = RUNTIME_HASH
    package.__fastplms_artifact_runtime_hashes__ = frozenset({RUNTIME_HASH})
    package.__fastplms_artifact_runtime_files__ = _runtime_file_hashes(package_root)
    package.__fastplms_artifact_runtime_roots__ = (str(package_root),)
    package.__fastplms_artifact_runtime_temporaries__ = tuple(
        _RUNTIME_TEMPORARIES
    )
    sys.modules["fastplms"] = package
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        try:
            spec.loader.exec_module(package)
        except BaseException:
            sys.modules.pop("fastplms", None)
            raise
    finally:
        sys.dont_write_bytecode = previous
    return package

_install_runtime()
_module_225 = _import_without_bytecode("fastplms.models.ankh.modeling_ankh")
FastAnkhConfig = _module_225.FastAnkhConfig
FastAnkhConfig.__module__ = __name__
FastAnkhForConditionalGeneration = _module_225.FastAnkhForConditionalGeneration
FastAnkhForConditionalGeneration.__module__ = __name__
FastAnkhForMaskedLMExtension = _module_225.FastAnkhForMaskedLMExtension
FastAnkhForMaskedLMExtension.__module__ = __name__
FastAnkhForSequenceClassification = _module_225.FastAnkhForSequenceClassification
FastAnkhForSequenceClassification.__module__ = __name__
FastAnkhForTokenClassification = _module_225.FastAnkhForTokenClassification
FastAnkhForTokenClassification.__module__ = __name__
FastAnkhModel = _module_225.FastAnkhModel
FastAnkhModel.__module__ = __name__