src/api.py

#!/usr/bin/env python3
"""
FastAPI Crop Yield Prediction API
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import pandas as pd
import numpy as np
import joblib
import warnings
import os
from datetime import datetime

# Import the existing predictor classes
from crop_yield_predictor import CropYieldPredictor, validate_json_input

warnings.filterwarnings('ignore')

# Initialize FastAPI app
app = FastAPI(
    title="Crop Yield Prediction API",
    description="API for predicting crop yields using Random Forest model",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# Pydantic models for request and response
class CropPredictionRequest(BaseModel):
    year: int = Field(..., description="Crop year (e.g., 2024)", example=2024)
    state: str = Field(..., description="State name", example="Punjab")
    crop: str = Field(..., description="Crop name", example="Rice")
    season: str = Field(..., description="Season", example="Kharif")
    area: float = Field(..., description="Area in hectares", example=10.0)
    production: float = Field(..., description="Production in tons", example=25.0)
    rainfall: Optional[float] = Field(1000.0, description="Annual rainfall in mm", example=1200)
    fertilizer: Optional[float] = Field(50.0, description="Fertilizer usage in kg", example=75)
    pesticide: Optional[float] = Field(5.0, description="Pesticide usage in kg", example=8)

class CropPredictionResponse(BaseModel):
    model: str = Field(..., description="Model used for prediction", example="Random Forest")
    predicted_yield: str = Field(..., description="Predicted yield with units", example="2017.7 kg/hectare")
    total_expected_production: str = Field(..., description="Total expected production with units", example="20.18 tons")
    assessment: str = Field(..., description="Yield assessment", example="Good yield expected")

class ErrorResponse(BaseModel):
    error: str = Field(..., description="Error message")

# Global predictor instance
predictor = None

@app.on_event("startup")
async def startup_event():
    """Initialize the predictor on startup"""
    global predictor
    try:
        predictor = CropYieldPredictor(quiet=True)
        print("✅ Crop Yield Predictor initialized successfully")
    except Exception as e:
        print(f"❌ Failed to initialize predictor: {e}")
        raise e

@app.get("/")
async def root():
    """Root endpoint"""
    return {
        "message": "Crop Yield Prediction API. Use /docs for interactive API documentation.",
        "version": "1.0.0",
        "endpoints": {
            "predict": "/predict",
            "health": "/health",
            "docs": "/docs",
            "available_options": "/available-options"
        }
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "model_loaded": predictor is not None
    }

@app.post("/predict", response_model=CropPredictionResponse, responses={400: {"model": ErrorResponse}})
async def predict_yield(request: CropPredictionRequest):
    """
    Predict crop yield based on input parameters
    
    Returns prediction with model type, predicted yield, total production, and assessment.
    """
    try:
        if predictor is None:
            raise HTTPException(status_code=500, detail="Predictor not initialized")
        
        # Convert request to dictionary format
        input_data = request.dict()
        
        # Validate and normalize input
        validated_data = validate_json_input(input_data)
        
        # Make prediction
        prediction, _ = predictor.predict_yield(validated_data)
        
        if isinstance(prediction, str) and 'Error' in prediction:
            raise HTTPException(status_code=400, detail=prediction)
        
        # Calculate total production
        total_production = (prediction * validated_data['Area']) / 1000.0  # Convert to tons
        
        # Determine assessment
        if prediction > 3000:
            assessment = "Excellent yield expected"
        elif prediction > 2000:
            assessment = "Good yield expected"
        elif prediction > 1000:
            assessment = "Moderate yield expected"
        else:
            assessment = "Low yield expected"
        
        # Format response
        response = CropPredictionResponse(
            model="Random Forest",
            predicted_yield=f"{round(prediction, 2)} kg/hectare",
            total_expected_production=f"{round(total_production, 2)} tons",
            assessment=assessment
        )
        
        return response
        
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")

@app.get("/available-options")
async def get_available_options():
    """Get available crops, states, and seasons"""
    try:
        if predictor is None:
            raise HTTPException(status_code=500, detail="Predictor not initialized")
        
        return {
            "states": predictor.get_state_options(),
            "crops": predictor.get_crop_options(),
            "seasons": predictor.get_season_options()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)