File size: 5,993 Bytes
5feba25 20948c2 5feba25 20948c2 5feba25 20948c2 5feba25 20948c2 5feba25 | 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | import joblib
import logging
from pathlib import Path
from typing import List
from app.config import MODEL_PATH, CLASS_NAMES, FEATURE_RANGES
from app.schemas import PredictionResponse
logger = logging.getLogger(__name__)
class Predictor:
"""Load and use ML model for predictions"""
def __init__(self, model_path: str = None):
"""
Initialize predictor with model path.
Args:
model_path: Path to model file (default: config.MODEL_PATH)
"""
if model_path is None:
model_path = MODEL_PATH
self.model_path = Path(model_path)
self.model = None
self._load_model()
def _load_model(self):
"""Load model from disk using joblib"""
if not self.model_path.exists():
raise FileNotFoundError(f"Model not found at {self.model_path}")
try:
self.model = joblib.load(self.model_path)
logger.info(f"✅ Model loaded from {self.model_path}")
logger.info(f" Model type: {type(self.model).__name__}")
logger.info(f" Expected features: {self.model.n_features_in_}")
except Exception as e:
logger.error(f"❌ Failed to load model: {e}")
raise
def predict(self, feature_vector: List[float]) -> PredictionResponse:
"""
Make prediction on feature vector.
Args:
feature_vector: List of 16 floats in correct order
Returns:
PredictionResponse with prediction, probability, risk level
Raises:
RuntimeError: If model not loaded
ValueError: If feature vector wrong size
"""
if self.model is None:
raise RuntimeError("Model not loaded")
if len(feature_vector) != 16:
raise ValueError(f"Expected 16 features, got {len(feature_vector)}")
try:
# Make prediction on single sample
# Note: sklearn expects 2D array [n_samples, n_features]
prediction_class = int(self.model.predict([feature_vector])[0])
# Get probability/confidence array for all classes
proba = self.model.predict_proba([feature_vector])[0]
probability = float(max(proba))
# SMART LOGIC: If top prediction is "Other/Unknown" (class 7), use second-best
OTHER_UNKNOWN_CLASS = 7
if prediction_class == OTHER_UNKNOWN_CLASS:
# Find second-highest confidence
top_two_indices = (-proba).argsort()[:2] # Get top 2 class indices
prediction_class = int(top_two_indices[1]) # Use second-best class
probability = float(proba[prediction_class]) # Get its confidence
logger.info(f"⚠️ Top prediction was Other/Unknown, using second-best: {CLASS_NAMES[prediction_class]}")
# Map class number to class name
class_name = CLASS_NAMES[prediction_class] if prediction_class < len(CLASS_NAMES) else "Unknown"
# Determine risk level based on probability
if probability >= 0.8:
risk_level = "High"
elif probability >= 0.6:
risk_level = "Medium"
else:
risk_level = "Low"
# Create friendly explanation based on the predicted condition
# Focus on actionable advice without technical model details
if class_name == "Healthy":
explanation = "Keep up a healthy lifestyle!"
elif class_name == "Arthritis":
explanation = "Consider low-impact exercises and consult with your doctor about pain management options."
elif class_name == "Asthma":
explanation = "Work with your healthcare provider on an asthma action plan and manage triggers."
elif class_name == "Cancer":
explanation = "Please consult with an oncologist immediately for proper evaluation and care."
elif class_name == "Diabetes":
explanation = "Monitor your blood sugar levels and work with your healthcare provider on a diabetes management plan."
elif class_name == "Hypertension":
explanation = "Monitor your blood pressure regularly and follow your doctor's guidance on medication and lifestyle changes."
elif class_name == "Obesity":
explanation = "Consider speaking with a nutritionist or healthcare provider about a healthy weight management plan."
elif class_name == "Other/Unknown":
explanation = "Please consult with a healthcare provider for proper evaluation and personalized advice."
else:
explanation = "Please consult with a healthcare professional for proper diagnosis and treatment."
result = PredictionResponse(
prediction=prediction_class,
probability=probability,
risk_level=risk_level,
explanation=explanation
)
logger.info(f"✅ Prediction: {class_name} (class {prediction_class}, confidence: {probability*100:.1f}%)")
return result
except Exception as e:
logger.error(f"❌ Prediction failed: {e}")
raise
# Global predictor instance (loaded once)
_predictor_instance = None
def get_predictor() -> Predictor:
"""
Get or create global predictor instance (lazy loading).
Returns:
Predictor instance
Raises:
FileNotFoundError: If model file not found
"""
global _predictor_instance
if _predictor_instance is None:
_predictor_instance = Predictor()
return _predictor_instance
def reload_predictor() -> Predictor:
"""
Force reload of predictor (useful for testing).
Returns:
New Predictor instance
"""
global _predictor_instance
_predictor_instance = Predictor()
return _predictor_instance
|