File size: 8,637 Bytes
59b15f5 |
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 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 |
"""
Multi-Variant FastAPI REST API for Milk Spoilage Classification
This API supports multiple model variants with different feature subsets.
Perfect for Custom GPT integration - allows selecting the optimal model
based on available data and prediction needs.
"""
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import joblib
import numpy as np
from typing import Dict, Optional, List
import os
import json
from pathlib import Path
# Load all model variants
VARIANTS_DIR = Path("model/variants")
if not VARIANTS_DIR.exists():
# Try alternate path for local development
VARIANTS_DIR = Path(__file__).parent.parent.parent / "model" / "variants"
# Load variants config
config_path = VARIANTS_DIR / "variants_config.json"
if not config_path.exists():
raise FileNotFoundError(f"variants_config.json not found at {config_path}")
with open(config_path) as f:
VARIANTS_CONFIG = json.load(f)
# Load all model files
MODELS = {}
for variant_id in VARIANTS_CONFIG['variants'].keys():
model_path = VARIANTS_DIR / f"{variant_id}.joblib"
if model_path.exists():
MODELS[variant_id] = joblib.load(model_path)
else:
print(f"Warning: Model file not found for variant {variant_id}")
print(f"✓ Loaded {len(MODELS)} model variants: {list(MODELS.keys())}")
# Create FastAPI app
app = FastAPI(
title="Milk Spoilage Classification API (Multi-Variant)",
description="""
AI-powered milk spoilage classification with multiple model variants.
**10 Model Variants Available:**
- **baseline**: All features (best accuracy: 95.8%)
- **scenario_1_days14_21**: Days 14 & 21 only (94.2%)
- **scenario_3_day21**: Day 21 only (93.7%)
- **scenario_4_day14**: Day 14 only (87.4%)
- **scenario_2_days7_14**: Days 7 & 14 (87.3%)
- **scenario_6_spc_all**: SPC only - all days (78.3%)
- **scenario_8_spc_7_14**: SPC days 7 & 14 (73.3%)
- **scenario_9_tgn_7_14**: TGN days 7 & 14 (73.1%)
- **scenario_7_tgn_all**: TGN only - all days (69.9%)
- **scenario_5_day7**: Day 7 only (62.8%)
Select the variant based on your available data. If you have all measurements,
use 'baseline' for best accuracy. If you only have partial data, choose the
appropriate scenario variant.
""",
version="2.0.0"
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
max_age=3600,
)
# Request/Response models
class PredictionInput(BaseModel):
spc_d7: Optional[float] = Field(None, description="Standard Plate Count at Day 7 (log CFU/mL)", ge=0.0, le=10.0)
spc_d14: Optional[float] = Field(None, description="Standard Plate Count at Day 14 (log CFU/mL)", ge=0.0, le=10.0)
spc_d21: Optional[float] = Field(None, description="Standard Plate Count at Day 21 (log CFU/mL)", ge=0.0, le=10.0)
tgn_d7: Optional[float] = Field(None, description="Total Gram-Negative at Day 7 (log CFU/mL)", ge=0.0, le=10.0)
tgn_d14: Optional[float] = Field(None, description="Total Gram-Negative at Day 14 (log CFU/mL)", ge=0.0, le=10.0)
tgn_d21: Optional[float] = Field(None, description="Total Gram-Negative at Day 21 (log CFU/mL)", ge=0.0, le=10.0)
model_variant: str = Field(
"baseline",
description="Model variant to use (baseline, scenario_1_days14_21, scenario_3_day21, etc.)"
)
class Config:
json_schema_extra = {
"example": {
"spc_d7": 2.1,
"spc_d14": 4.7,
"spc_d21": 6.4,
"tgn_d7": 1.0,
"tgn_d14": 3.7,
"tgn_d21": 5.3,
"model_variant": "baseline"
}
}
class VariantInfo(BaseModel):
variant_id: str
name: str
description: str
features: List[str]
test_accuracy: float
class PredictionOutput(BaseModel):
prediction: str = Field(..., description="Predicted spoilage class")
probabilities: Dict[str, float] = Field(..., description="Probability for each class")
confidence: float = Field(..., description="Confidence score (max probability)")
variant_used: VariantInfo = Field(..., description="Information about the model variant used")
def extract_features(input_data: PredictionInput, required_features: List[str]) -> np.ndarray:
"""Extract required features from input data."""
feature_map = {
'SPC_D7': input_data.spc_d7,
'SPC_D14': input_data.spc_d14,
'SPC_D21': input_data.spc_d21,
'TGN_D7': input_data.tgn_d7,
'TGN_D14': input_data.tgn_d14,
'TGN_D21': input_data.tgn_d21,
}
# Check for missing required features
missing = [f for f in required_features if feature_map[f] is None]
if missing:
raise HTTPException(
status_code=400,
detail=f"Missing required features for variant: {', '.join(missing)}"
)
# Extract and convert from log to raw CFU/mL
features = [10 ** feature_map[f] for f in required_features]
return np.array([features])
@app.get("/")
async def root():
"""Root endpoint with API information."""
return {
"message": "Milk Spoilage Classification API - Multi-Variant",
"version": "2.0.0",
"variants_available": len(MODELS),
"endpoints": {
"predict": "/predict",
"variants": "/variants",
"health": "/health",
"docs": "/docs"
}
}
@app.get("/variants", tags=["Variants"])
async def list_variants():
"""List all available model variants with their metadata."""
variants_list = []
for variant_id, metadata in VARIANTS_CONFIG['variants'].items():
variants_list.append({
"variant_id": variant_id,
"name": metadata['name'],
"description": metadata['description'],
"features": metadata['features'],
"test_accuracy": metadata['test_accuracy'],
"n_features": len(metadata['features'])
})
# Sort by test accuracy descending
variants_list.sort(key=lambda x: x['test_accuracy'], reverse=True)
return {
"total_variants": len(variants_list),
"variants": variants_list
}
@app.post("/predict", response_model=PredictionOutput, tags=["Prediction"])
async def predict(input_data: PredictionInput):
"""
Predict milk spoilage type using the specified model variant.
**How to choose a variant:**
- If you have all 6 measurements → use 'baseline' (best accuracy)
- If you only have Day 21 data → use 'scenario_3_day21'
- If you only have Day 14 data → use 'scenario_4_day14'
- If you only have SPC measurements → use 'scenario_6_spc_all'
- etc.
The API will validate that you've provided all required features for the selected variant.
"""
# Validate variant exists
if input_data.model_variant not in MODELS:
raise HTTPException(
status_code=400,
detail=f"Unknown variant '{input_data.model_variant}'. Use /variants to see available options."
)
# Get model and metadata
model = MODELS[input_data.model_variant]
variant_meta = VARIANTS_CONFIG['variants'][input_data.model_variant]
required_features = variant_meta['features']
# Extract features
try:
features = extract_features(input_data, required_features)
except HTTPException as e:
raise e
# Make prediction
prediction = model.predict(features)[0]
probabilities = model.predict_proba(features)[0]
# Format response
prob_dict = {
str(cls): float(prob)
for cls, prob in zip(model.classes_, probabilities)
}
variant_info = VariantInfo(
variant_id=input_data.model_variant,
name=variant_meta['name'],
description=variant_meta['description'],
features=required_features,
test_accuracy=variant_meta['test_accuracy']
)
return PredictionOutput(
prediction=str(prediction),
probabilities=prob_dict,
confidence=float(max(probabilities)),
variant_used=variant_info
)
@app.get("/health", tags=["Health"])
async def health_check():
"""Health check endpoint."""
return {
"status": "healthy",
"models_loaded": len(MODELS),
"variants": list(MODELS.keys()),
"classes": MODELS['baseline'].classes_.tolist() if 'baseline' in MODELS else []
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|