Spaces:
Sleeping
Sleeping
| 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) | |
| } | |
| 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)}") | |
| async def root(): | |
| return {"message": "Fertilizer Recommendation API is running. Use /predict_fertilizer endpoint to send input data."} | |
| 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 | |
| } | |