app.py

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
    }