faisaltitu/Drift-Detection

1"""2Predictor Module - Model Loading and Inference3"""4 5import json6import logging7from pathlib import Path8from typing import Dict, Optional, Tuple9 10import joblib11import numpy as np12import pandas as pd13from sklearn.ensemble import RandomForestRegressor14from sklearn.preprocessing import StandardScaler15 16logger = logging.getLogger(__name__)17 18# Paths19MODELS_DIR = Path(__file__).parent.parent / "models"20PRODUCTION_MODEL_DIR = MODELS_DIR / "production"21 22 23class Predictor:24    """Handles model loading and prediction."""25    26    def __init__(self):27        self.model: Optional[RandomForestRegressor] = None28        self.scaler: Optional[StandardScaler] = None29        self.metadata: Optional[Dict] = None30        self.model_version: Optional[int] = None31        self.feature_names = [32            "MedInc", "HouseAge", "AveRooms", "AveBedrms",33            "Population", "AveOccup", "Latitude", "Longitude"34        ]35        36    def load_model(self) -> bool:37        """38        Load the production model.39        40        Returns:41            True if successful42        """43        model_path = PRODUCTION_MODEL_DIR / "model.joblib"44        scaler_path = PRODUCTION_MODEL_DIR / "scaler.joblib"45        metadata_path = PRODUCTION_MODEL_DIR / "metadata.json"46        47        if not model_path.exists():48            logger.warning("Model not found: %s", model_path)49            return False50        51        try:52            self.model = joblib.load(model_path)53            self.scaler = joblib.load(scaler_path)54            55            with open(metadata_path, "r") as f:56                self.metadata = json.load(f)57            58            self.model_version = self.metadata.get("source_version", 59                                                    self.metadata.get("version", 0))60            61            logger.info("Model loaded: v%s", self.model_version)62            return True63            64        except Exception as e:65            logger.error("Error loading model: %s", e)66            return False67    68    def reload_model(self) -> bool:69        """Reload the model (useful after promotion)."""70        return self.load_model()71    72    def predict(self, features: Dict) -> Tuple[float, int]:73        """74        Make a prediction.75        76        Args:77            features: Dictionary of feature names to values78        79        Returns:80            Tuple of (prediction, model_version)81        """82        if self.model is None:83            raise RuntimeError("Model not loaded. Call load_model() first.")84        85        # Convert dict to DataFrame with feature names86        X = pd.DataFrame([[features[name] for name in self.feature_names]],87                         columns=self.feature_names)88        89        # Scale features90        X_scaled = self.scaler.transform(X)91        92        # Predict93        prediction = self.model.predict(X_scaled)[0]94        95        return float(prediction), self.model_version96    97    def is_loaded(self) -> bool:98        """Check if model is loaded."""99        return self.model is not None100 101 102# Global predictor instance103predictor = Predictor()104