File size: 2,415 Bytes
49fd357
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Model inference for pricing predictions."""

import logging
from pathlib import Path
from typing import Union

import joblib
import pandas as pd

from src.ml.preprocessing import normalize_categories

logger = logging.getLogger(__name__)

DEFAULT_MODEL_PATH = (
    Path(__file__).parent.parent.parent / "models" / "best_model.joblib"
)

_predictor_instance: "PricingPredictor | None" = None


class ModelNotFoundError(Exception):
    """Raised when the model file cannot be found."""


class PricingPredictor:
    """Predictor class for car rental pricing."""

    def __init__(self, model_path: Union[str, Path] = DEFAULT_MODEL_PATH) -> None:
        """Initialize predictor with trained model.

        Args:
            model_path: Path to the trained model file.

        Raises:
            ModelNotFoundError: If the model file does not exist.
        """
        self.model_path = Path(model_path)
        self.model = None
        self._load_model()

    def _load_model(self) -> None:
        """Load model from disk.

        Raises:
            ModelNotFoundError: If the model file does not exist.
        """
        if not self.model_path.exists():
            logger.error("Model file not found: %s", self.model_path)
            raise ModelNotFoundError(f"Model not found at {self.model_path}")

        logger.info("Loading model from %s", self.model_path)
        self.model = joblib.load(self.model_path)
        logger.info("Model loaded successfully")

    def predict_from_features(self, cars: list[dict]) -> list[int]:
        """Make predictions from car feature dictionaries.

        Args:
            cars: List of dictionaries with car features.

        Returns:
            List of predicted prices (rounded to int).
        """
        logger.debug("Predicting for %d cars", len(cars))
        df = pd.DataFrame(cars)
        df = normalize_categories(df)
        predictions = self.model.predict(df)
        return [int(round(p)) for p in predictions]


def get_predictor(
    model_path: Union[str, Path] = DEFAULT_MODEL_PATH,
) -> PricingPredictor:
    """Get singleton predictor instance.

    Args:
        model_path: Path to the trained model file.

    Returns:
        Singleton PricingPredictor instance.
    """
    global _predictor_instance
    if _predictor_instance is None:
        _predictor_instance = PricingPredictor(model_path)
    return _predictor_instance