Spaces:
Sleeping
Sleeping
File size: 4,084 Bytes
a754062 |
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 |
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import pandas as pd
import joblib
import os
import logging
from typing import Dict, List
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Fertilizer Recommendation API")
# Define model file paths
PIPELINE_PATH = r"fertilizer_pipeline.pkl"
ENCODER_PATH = r"fertilizer_label_encoder.pkl"
# Valid crops
VALID_CROPS = [
'Sugarcane', 'Jowar', 'Cotton', 'Rice', 'Wheat', 'Groundnut',
'Maize', 'Tur', 'Urad', 'Moong', 'Gram', 'Masoor', 'Soybean',
'Ginger', 'Turmeric', 'Grapes'
]
# Load pipeline and encoder at startup
try:
pipeline = joblib.load(PIPELINE_PATH)
label_encoder = joblib.load(ENCODER_PATH)
logger.info("Pipeline and encoder loaded successfully")
except Exception as e:
logger.error(f"Failed to load pipeline or encoder: {str(e)}")
raise Exception(f"Failed to load pipeline or encoder: {str(e)}")
# Pydantic model for input validation
class FertilizerInput(BaseModel):
Nitrogen: float = Field(..., ge=20, le=150, description="Nitrogen content in soil (kg/ha)")
Phosphorus: float = Field(..., ge=10, le=90, description="Phosphorus content in soil (kg/ha)")
Potassium: float = Field(..., ge=5, le=150, description="Potassium content in soil (kg/ha)")
pH: float = Field(..., ge=5.5, le=8.5, description="Soil pH value")
Rainfall: float = Field(..., ge=300, le=1700, description="Rainfall in millimeters")
Temperature: float = Field(..., ge=10, le=40, description="Temperature in Celsius")
Crop: str = Field(..., description="Crop type", enum=VALID_CROPS)
# Synchronous prediction function
def predict_fertilizer(input_data: Dict) -> Dict:
try:
# Convert input to DataFrame
input_df = pd.DataFrame([input_data])
# Validate required columns
required_cols = ['Nitrogen', 'Phosphorus', 'Potassium', 'pH', 'Rainfall', 'Temperature', 'Crop']
missing_cols = set(required_cols) - set(input_df.columns)
if missing_cols:
raise ValueError(f"Missing required columns: {missing_cols}")
# Predict
y_pred_encoded = pipeline.predict(input_df)
y_pred_label = label_encoder.inverse_transform(y_pred_encoded)[0]
return {
"fertilizer": y_pred_label,
"status": "success"
}
except Exception as e:
logger.error(f"Prediction error: {str(e)}")
return {
"fertilizer": "",
"status": "failure",
"error": str(e)
}
@app.post("/predict_fertilizer")
async def predict_fertilizer_endpoint(input_data: FertilizerInput):
try:
# Check if files exist
for path in [PIPELINE_PATH, ENCODER_PATH]:
if not os.path.exists(path):
raise HTTPException(status_code=500, detail=f"File not found: {path}")
# Convert Pydantic model to dict
input_dict = input_data.dict()
# Make prediction
result = predict_fertilizer(input_dict)
if result["status"] == "failure":
raise HTTPException(status_code=400, detail=result["error"])
return result
except Exception as e:
logger.error(f"Error processing prediction: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error processing prediction: {str(e)}")
@app.get("/")
async def root():
return {"message": "Fertilizer Recommendation API is running. Use /predict_fertilizer endpoint to send input data."}
@app.get("/valid_inputs")
async def get_valid_inputs():
return {
"Nitrogen": {"min": 20, "max": 150, "unit": "kg/ha"},
"Phosphorus": {"min": 10, "max": 90, "unit": "kg/ha"},
"Potassium": {"min": 5, "max": 150, "unit": "kg/ha"},
"pH": {"min": 5.5, "max": 8.5, "unit": "pH"},
"Rainfall": {"min": 300, "max": 1700, "unit": "mm"},
"Temperature": {"min": 10, "max": 40, "unit": "Celsius"},
"Crop": VALID_CROPS
}
|