Spaces:
Sleeping
Sleeping
File size: 5,550 Bytes
bbd5f9c | 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 | #!/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)
|