Spaces:
Sleeping
Sleeping
File size: 4,637 Bytes
725d1d2 6752520 725d1d2 371053c 6752520 371053c 725d1d2 371053c 6752520 371053c 725d1d2 6752520 371053c 6752520 371053c 725d1d2 371053c 725d1d2 6752520 725d1d2 6752520 725d1d2 6752520 725d1d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | import asyncio
import json
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Dict, Optional, Tuple
from huggingface_hub import hf_hub_download
import joblib
import pandas as pd
class ModelService:
def __init__(self, model_repo: str, model_file: str, feature_file: str, metadata_file: str):
self.model_repo = model_repo
self.model_file = model_file
self.feature_file = feature_file
self.metadata_file = metadata_file
self.pipeline = None
self.expected_features = None
self.load_time = None
self.model_metadata: Optional[Dict[str, Any]] = None
self.executor = ThreadPoolExecutor(max_workers=4)
self._load_model()
self._load_model_metadata()
def _download_from_hf(self, filename: str) -> str:
path = hf_hub_download(repo_id=self.model_repo, filename=filename, cache_dir="models")
return path
def _load_model(self):
try:
start_time = time.time()
model_path = self._download_from_hf(self.model_file)
feature_path = self._download_from_hf(self.feature_file)
self.pipeline = joblib.load(model_path)
self.expected_features = joblib.load(feature_path)
self.load_time = round(time.time() - start_time, 3)
print("Model Loaded Successfully!")
except Exception as e:
self.pipeline = None
self.load_time = None
print(f"❌ Model failed to load: {e}")
def _load_model_metadata(self):
try:
metadata_path = self._download_from_hf(self.metadata_file)
with open(metadata_path, "r") as f:
self.model_metadata = json.load(f)
print("Metadata loaded!")
except Exception as e:
print("❌ Failed to load metadata:", e)
self.model_metadata = None
def get_model_info(self) -> Dict[str, Any]:
# Safety check
if self.model_metadata is None:
self._load_model_metadata()
return {
"path": self.model_path,
"load_time_sec": self.load_time,
"type": self.model_metadata["model_info"]["model_type"],
"pipeline": self.model_metadata["model_info"]["model_name"],
}
def is_model_loaded(self) -> bool:
return self.pipeline is not None
async def preprocess_data(self, data_dict: Dict[str, Any]) -> pd.DataFrame:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
self.executor, self._preprocess_sync, data_dict
)
def _preprocess_sync(self, data_dict: Dict[str, Any]) -> pd.DataFrame:
df = pd.DataFrame([data_dict])
# Reindex ke expected_features
if self.expected_features:
missing_model_features = [
c for c in self.expected_features if c not in df.columns
]
if missing_model_features:
raise ValueError(f"Missing model features: {missing_model_features}")
df = df.reindex(columns=self.expected_features, fill_value=0)
return df
async def predict(
self, df: pd.DataFrame
) -> Tuple[int, Optional[float], Optional[float]]:
if not self.is_model_loaded():
raise RuntimeError("Model not loaded")
loop = asyncio.get_event_loop()
return await loop.run_in_executor(self.executor, self._predict_sync, df)
def _predict_sync(
self, df: pd.DataFrame
) -> Tuple[int, Optional[float], Optional[float], Optional[str]]:
# Prediksi label
pred = self.pipeline.predict(df)[0]
# Prediksi probabilitas
if hasattr(self.pipeline, "predict_proba"):
proba = self.pipeline.predict_proba(df)[0]
proba_no, proba_yes = float(proba[0]), float(proba[1])
else:
proba_yes = None
proba_no = None
# Tentukan priority/confidence berdasarkan probabilitas yes
if proba_yes is not None:
if proba_yes >= 0.7:
priority = "HIGH"
elif proba_yes >= 0.5:
priority = "MEDIUM"
elif proba_yes >= 0.3:
priority = "LOW"
else:
priority = "VERY LOW"
else:
priority = None
return int(pred), proba_yes, proba_no, priority
def get_complete_model_info(self) -> (Dict[str, Any] | None):
if self.model_metadata is None:
self._load_model_metadata()
return self.model_metadata |