Upload 5 files

ExoMACModel/ExoMAC-KKT/exoplanet_class_labels.json

@@ -0,0 +1 @@
+["CANDIDATE", "CONFIRMED", "FALSE POSITIVE"]

ExoMACModel/ExoMAC-KKT/exoplanet_feature_columns.json

@@ -0,0 +1 @@
+["koi_depth", "koi_duration", "koi_impact", "koi_period", "koi_prad", "koi_slogg", "koi_sma", "koi_smet", "koi_snr", "koi_srad", "koi_steff", "duty_cycle", "log_koi_period", "log_koi_depth", "log_koi_snr", "teq_proxy"]

ExoMACModel/ExoMAC-KKT/exoplanet_metadata.json

@@ -0,0 +1,10 @@
+{
+  "best_model_name": "RandomForest",
+  "n_features": 16,
+  "labels": [
+    "CANDIDATE",
+    "CONFIRMED",
+    "FALSE POSITIVE"
+  ],
+  "created": "2025-10-05T20:49:33.128163Z"
+}

ExoMACModel/__init__.py

@@ -0,0 +1,3 @@
+from ExoMACModel.main import ExoMACModel
+
+__all__ = ["ExoMACModel"]

ExoMACModel/main.py

@@ -0,0 +1,331 @@
+from __future__ import annotations
+
+import json
+import os
+import threading
+from typing import Dict, Tuple, Optional, List
+from pathlib import Path
+
+import joblib
+import numpy as np
+import pandas as pd
+from huggingface_hub import hf_hub_download, snapshot_download
+
+
+class _Singleton(type):
+    """Thread-safe Singleton metaclass (una instancia por proceso)."""
+    _instances: Dict[type, object] = {}
+    _lock = threading.Lock()
+
+    def __call__(cls, *args, **kwargs):
+        # Double-checked locking
+        if cls not in cls._instances:
+            with cls._lock:
+                if cls not in cls._instances:
+                    cls._instances[cls] = super().__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class ExoMACModel(metaclass=_Singleton):
+    """
+    Misión-agnóstico: cargador de modelo (Pipeline sklearn) entrenado con Kepler/K2/TESS.
+    - Descarga artefactos desde Hugging Face SOLO si no existen localmente.
+    - Guarda/lee desde una carpeta local del proyecto (por defecto: ./models/ExoMAC-KKT).
+    - Exposición de helpers de predicción y de features ingenierizadas.
+    """
+
+    DEFAULT_REPO = "ZapatoProgramming/ExoMAC-KKT"
+    _FILENAMES = {
+        "model":  "exoplanet_best_model.joblib",
+        "feats":  "exoplanet_feature_columns.json",
+        "labels": "exoplanet_class_labels.json",
+        "meta":   "exoplanet_metadata.json",
+    }
+
+    def __init__(
+        self,
+        repo_id: Optional[str] = None,
+        token: Optional[str] = None,
+        prefer_snapshot: bool = True,
+        allow_patterns: Optional[List[str]] = None,
+        local_dir: Optional[str | os.PathLike] = None,
+        always_download: bool = False,
+        verbose: bool = True,
+    ):
+        """
+        Args:
+            repo_id: Hugging Face repo id. Por defecto 'ZapatoProgramming/ExoMAC-KKT'.
+            token:   Token HF si el repo es privado.
+            prefer_snapshot: Si True, usa snapshot_download (descarga por patrón).
+            allow_patterns:  Patrones a descargar cuando prefer_snapshot=True.
+            local_dir:       Carpeta donde se guardan/leen artefactos en tu proyecto.
+            always_download: Si True, fuerza descarga (útil para actualizar).
+            verbose:         Imprime mensajes útiles.
+        """
+        self.repo_id = repo_id or self.DEFAULT_REPO
+        self.token = token
+        self.prefer_snapshot = prefer_snapshot
+        self.allow_patterns = allow_patterns or ["artifacts/*", "*.joblib", "*.json"]
+        self.local_dir = Path(local_dir or (Path("models") / self.repo_id.split("/")[-1]))
+        self.local_dir.mkdir(parents=True, exist_ok=True)
+        self.always_download = always_download
+        self.verbose = verbose
+
+        self._model = None
+        self._feature_columns: List[str] = []
+        self._class_labels: List[str] = []
+        self._metadata: Dict = {}
+
+        self._load_artifacts()
+
+    # ------------------------- PUBLIC API -------------------------
+
+    @property
+    def model(self):
+        return self._model
+
+    @property
+    def feature_columns(self) -> List[str]:
+        return list(self._feature_columns)
+
+    @property
+    def class_labels(self) -> List[str]:
+        return list(self._class_labels)
+
+    @property
+    def metadata(self) -> Dict:
+        return dict(self._metadata)
+
+    def predict(
+        self,
+        params: Dict[str, float],
+        return_proba: bool = True,
+        compute_engineered_if_missing: bool = True,
+    ) -> Tuple[str, Optional[Dict[str, float]]]:
+        """
+        Predice una etiqueta y (opcionalmente) probabilidades para un dict de features.
+        - Rellena features ingenierizadas si el modelo las espera y no están.
+        """
+        if compute_engineered_if_missing:
+            params = self._ensure_engineered_features(dict(params))
+
+        X = pd.DataFrame([params], dtype=float).reindex(columns=self._feature_columns)
+        y_idx = int(self._model.predict(X)[0])
+        label = self._class_labels[y_idx]
+
+        if not return_proba:
+            return label, None
+
+        proba = None
+        try:
+            p = self._model.predict_proba(X)[0]
+            proba = {lbl: float(prob) for lbl, prob in zip(self._class_labels, p)}
+        except Exception:
+            pass
+        return label, proba
+
+    def predict_with_debug(self, params: Dict[str, float]) -> Tuple[str, Optional[Dict[str, float]]]:
+        """
+        Igual que predict(), pero imprime features reconocidas/desconocidas y faltantes.
+        """
+        params2 = self._ensure_engineered_features(dict(params))
+        X = pd.DataFrame([params2], dtype=float).reindex(columns=self._feature_columns)
+
+        recognized = [c for c in self._feature_columns if c in params2]
+        unknown = [k for k in params2.keys() if k not in self._feature_columns]
+        missing = X.columns[X.iloc[0].isna()].tolist()
+
+        print(f"Recognized: {len(recognized)}/{len(self._feature_columns)}")
+        if recognized:
+            print("  •", ", ".join(recognized[:16]) + (" ..." if len(recognized) > 16 else ""))
+        if unknown:
+            print(f"Unknown keys: {len(unknown)}")
+            if unknown:
+                print("  •", ", ".join(unknown[:16]) + (" ..." if len(unknown) > 16 else ""))
+        if missing:
+            print(f"Missing (imputed): {len(missing)}")
+            if missing:
+                print("  •", ", ".join(missing[:16]) + (" ..." if len(missing) > 16 else ""))
+
+        return self.predict(params2, return_proba=True, compute_engineered_if_missing=False)
+
+    # ------------------------- INTERNALS -------------------------
+
+    def _load_artifacts(self) -> None:
+        """
+        1) Si ya existen archivos locales y always_download=False -> NO descarga.
+        2) Si faltan archivos o always_download=True -> descarga (snapshot o per-file).
+        3) Carga el modelo + metadata desde disco.
+        """
+        paths: Optional[Dict[str, str]] = None
+
+        # (0) Intentar leer desde local sin tocar red
+        if not self.always_download:
+            local_paths = self._try_local_paths()
+            if local_paths is not None:
+                paths = local_paths
+                if self.verbose:
+                    print(f"[ExoMAC] Using cached artifacts in {self.local_dir}")
+            else:
+                if self.verbose:
+                    print(f"[ExoMAC] Local artifacts not found. Will download to {self.local_dir}.")
+
+        # (1) Descargar si hace falta
+        if paths is None:
+            if self.prefer_snapshot:
+                # Descarga patrones a la carpeta local (la API ya no usa symlinks)
+                snapshot_download(
+                    repo_id=self.repo_id,
+                    token=self.token,
+                    allow_patterns=self.allow_patterns,
+                    local_dir=str(self.local_dir),
+                )
+                paths = self._resolve_from_dir(self.local_dir)
+            else:
+                paths = {}
+                for key, fname in self._FILENAMES.items():
+                    paths[key] = self._get_artifact_to_local_dir(fname)
+
+        # (2) Cargar desde disco
+        self._model = joblib.load(paths["model"])
+        self._feature_columns = json.load(open(paths["feats"], "r", encoding="utf-8"))
+        self._class_labels    = json.load(open(paths["labels"], "r", encoding="utf-8"))
+        self._metadata        = json.load(open(paths["meta"], "r", encoding="utf-8"))
+
+        if self.verbose:
+            print(f"[ExoMAC] Loaded model from {paths['model']}")
+
+    # --- Local path helpers ---
+
+    def _have_all_files(self, base: Path) -> bool:
+        """¿Están TODOS los artefactos (en artifacts/ o raíz) en 'base'?"""
+        base = Path(base)
+        for _, name in self._FILENAMES.items():
+            p1 = base / "artifacts" / name
+            p2 = base / name
+            if not (p1.exists() or p2.exists()):
+                return False
+        return True
+
+    def _try_local_paths(self) -> Optional[Dict[str, str]]:
+        """Devuelve rutas locales si todo existe; si falta algo, None."""
+        if self._have_all_files(self.local_dir):
+            return self._resolve_from_dir(self.local_dir)
+        return None
+
+    def _resolve_from_dir(self, base_dir: Path | str) -> Dict[str, str]:
+        """
+        Selecciona artifacts/<name> si existe; si no, <base>/<name>.
+        """
+        base_dir = Path(base_dir)
+        out: Dict[str, str] = {}
+        for key, name in self._FILENAMES.items():
+            p1 = base_dir / "artifacts" / name
+            p2 = base_dir / name
+            if p1.exists():
+                out[key] = str(p1)
+            elif p2.exists():
+                out[key] = str(p2)
+            else:
+                raise FileNotFoundError(f"Could not find {name} under {base_dir}")
+        return out
+
+    def _get_artifact_to_local_dir(self, fname: str) -> str:
+        """
+        Descarga a self.local_dir con hf_hub_download (si tu versión soporta local_dir).
+        Si no, descarga a la caché global y copia a self.local_dir.
+        """
+        self.local_dir.mkdir(parents=True, exist_ok=True)
+
+        for candidate in (f"artifacts/{fname}", fname):
+            try:
+                # huggingface_hub >= 0.23 soporta local_dir
+                path = hf_hub_download(
+                    repo_id=self.repo_id,
+                    filename=candidate,
+                    token=self.token,
+                    local_dir=str(self.local_dir),
+                )
+                return path
+            except TypeError:
+                # Fallback: versión antigua sin local_dir
+                cache_path = hf_hub_download(
+                    repo_id=self.repo_id,
+                    filename=candidate,
+                    token=self.token,
+                )
+                dst = self.local_dir / Path(candidate).name
+                os.makedirs(self.local_dir, exist_ok=True)
+                if not os.path.exists(dst):
+                    from shutil import copy2
+                    copy2(cache_path, dst)
+                return str(dst)
+            except Exception:
+                # prueba siguiente candidato (raíz en lugar de artifacts/)
+                continue
+
+        raise FileNotFoundError(f"Could not download {fname} from {self.repo_id}")
+
+    # --- Engineered features helpers ---
+
+    def _ensure_engineered_features(self, d: Dict[str, float]) -> Dict[str, float]:
+        """
+        Rellena features ingenierizadas si el modelo las espera y no están:
+        - duty_cycle, log_koi_period, log_koi_depth, teq_proxy
+        - koi_snr/log_koi_snr  o  snr_proxy/log_snr_proxy (proxy)
+        """
+        need = set(self._feature_columns)
+
+        # Duty cycle
+        if "duty_cycle" in need and "duty_cycle" not in d:
+            if all(k in d for k in ("koi_duration", "koi_period")) and d.get("koi_period"):
+                d["duty_cycle"] = d["koi_duration"] / (d["koi_period"] * 24.0)
+
+        # Logs
+        if "log_koi_period" in need and "log_koi_period" not in d and d.get("koi_period", 0) > 0:
+            d["log_koi_period"] = np.log10(d["koi_period"])
+        if "log_koi_depth" in need and "log_koi_depth" not in d and d.get("koi_depth", 0) > 0:
+            d["log_koi_depth"] = np.log10(d["koi_depth"])
+
+        # teq_proxy (simple)
+        if "teq_proxy" in need and "teq_proxy" not in d and "koi_steff" in d:
+            d["teq_proxy"] = d["koi_steff"]
+
+        # SNR real o proxy
+        if "koi_snr" in need and "koi_snr" not in d:
+            d["koi_snr"] = np.nan
+        if "log_koi_snr" in need and "log_koi_snr" not in d and d.get("koi_snr", 0) > 0:
+            d["log_koi_snr"] = np.log10(d["koi_snr"])
+
+        if "snr_proxy" in need and "snr_proxy" not in d:
+            if all(k in d for k in ("koi_depth", "koi_duration", "koi_period")) and d.get("koi_period", 0) > 0:
+                d["snr_proxy"] = d["koi_depth"] * np.sqrt(max(d["koi_duration"] / (d["koi_period"] * 24.0), 1e-12))
+        if "log_snr_proxy" in need and "log_snr_proxy" not in d and d.get("snr_proxy", 0) > 0:
+            d["log_snr_proxy"] = np.log10(d["snr_proxy"])
+
+        return d
+
+
+# ------------------------- DEMO -------------------------
+if __name__ == "__main__":
+    # Primera ejecución: descargará a ./models/ExoMAC-KKT si no existe.
+    model = ExoMACModel(
+        local_dir="./ExoMACModel/ExoMAC-KKT",
+        prefer_snapshot=True,
+        always_download=False,   # <- ejecuciones siguientes NO vuelven a descargar
+        verbose=True,
+    )
+
+    # Subsecuentes: misma instancia (singleton) y SIN descarga.
+    same_model = ExoMACModel(local_dir="./ExoMACModel/ExoMAC-KKT")
+    assert model is same_model
+
+    # Ejemplo mínimo de predicción
+    params = {
+        "koi_period": 12.0, "koi_duration": 3.5, "koi_depth": 600.0, "koi_impact": 0.20,
+        "koi_prad": 2.1, "koi_slogg": 4.4, "koi_sma": 0.10, "koi_smet": 0.0,
+        "koi_srad": 1.0, "koi_steff": 5700.0, "koi_snr": 12.0,
+    }
+    label, proba = model.predict_with_debug(params)
+    print("Predicted:", label)
+    print("Local dir:", model.local_dir.resolve())