routes/ML_DecisionTree_IrisClassifier.py

from fastapi import APIRouter
from pydantic import BaseModel
import joblib
from sklearn.datasets import load_iris
from typing import Any, cast

from .config_huggingface import build_model_url, download_artifact_if_needed

router = APIRouter(tags=["Machine Learning"])

class IrisFeatures(BaseModel):
    sepal_length: float = 1.0  
    sepal_width: float = 1.0
    petal_length: float = 1.0
    petal_width: float = 1.0


MODEL_STATE: dict[str, Any] = {
    "model": None,
    "error": None,
}

MODEL_URL = build_model_url("ML_DecisionTree_IrisClassifier.joblib")

iris = cast(Any, load_iris())


def _ensure_model_loaded() -> None:
    if MODEL_STATE["model"] is not None:
        return

    try:
        model_path = download_artifact_if_needed(MODEL_URL)
        MODEL_STATE["model"] = joblib.load(model_path)
        MODEL_STATE["error"] = None
    except Exception as e:
        MODEL_STATE["error"] = str(e)
        raise

@router.put("/models/irisClassifier")
def iris_classifier(data: IrisFeatures):
    import numpy as np

    try:
        _ensure_model_loaded()
    except Exception:
        detail = "Model not loaded."
        if MODEL_STATE["error"]:
            detail = f"Model not loaded: {MODEL_STATE['error']}"
        return {"error": detail, "status": 500}

    model = cast(Any, MODEL_STATE["model"])
    test_data = np.array([[
        data.sepal_length,
        data.sepal_width,
        data.petal_length,
        data.petal_width
    ]])
    raw_prediction = model.predict(test_data)[0]

    # Some serialized models output numeric indices, others output class labels.
    if isinstance(raw_prediction, (int, np.integer)):
        class_idx = int(raw_prediction)
        if class_idx < 0 or class_idx >= len(iris.target_names):
            return {"error": f"Invalid prediction index: {class_idx}", "status": 500}
        return {"prediction": str(iris.target_names[class_idx])}

    return {"prediction": str(raw_prediction)}