Upload 4 files

final_model.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a6e8f57ad9c823b75d7c622047b4c0486ade556788767ba079b353a6df018e44
+size 32528345

label_encoder.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:7f768bc83fabe555d39a6a0087b4a3aefbba39029dd57f0d07d7dc6b072a66da
+size 985

main.py

@@ -0,0 +1,129 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+import pandas as pd
+import joblib
+import os
+import logging
+import numpy as np
+from typing import List, Dict
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="Crop Recommendation API")
+
+# Define model file paths
+MODEL_PATH = r"final_model.joblib"
+ENCODER_PATH = r"label_encoder.joblib"
+
+# Load model and encoder at startup
+try:
+    final_RF = joblib.load(MODEL_PATH)
+    label_encoder = joblib.load(ENCODER_PATH)
+    VALID_CROPS = list(label_encoder.classes_)
+    logger.info("Model and encoder loaded successfully")
+except Exception as e:
+    logger.error(f"Failed to load model or encoder: {str(e)}")
+    raise Exception(f"Failed to load model or encoder: {str(e)}")
+
+
+# Pydantic model for input validation
+class CropInput(BaseModel):
+    N: float = Field(..., ge=0, le=200, description="Nitrogen content in soil (kg/ha)")
+    P: float = Field(
+        ..., ge=0, le=200, description="Phosphorus content in soil (kg/ha)"
+    )
+    K: float = Field(..., ge=0, le=200, description="Potassium content in soil (kg/ha)")
+    temperature: float = Field(..., ge=0, le=50, description="Temperature in Celsius")
+    ph: float = Field(..., ge=0, le=14, description="Soil pH value")
+    rainfall: float = Field(..., ge=0, le=2000, description="Rainfall in millimeters")
+
+
+def get_top_n_classes(
+    model, X_df: pd.DataFrame, label_encoder, n: int = 5
+) -> List[Dict]:
+    """Get the top N predicted classes with their probabilities."""
+    try:
+        probs = model.predict_proba(X_df)[0]
+        top_indices = np.argsort(probs)[::-1][:n]
+        top_probs = probs[top_indices]
+        top_labels = label_encoder.inverse_transform(top_indices)
+        return [
+            {"crop": label, "probability": float(prob)}
+            for label, prob in zip(top_labels, top_probs)
+        ]
+    except Exception as e:
+        logger.error(f"Error in get_top_n_classes: {str(e)}")
+        raise ValueError(f"Failed to compute top classes: {str(e)}")
+
+
+# Synchronous prediction function
+def predict_crop(input_data: Dict) -> Dict:
+    try:
+        # Convert input to DataFrame
+        input_df = pd.DataFrame([input_data])
+
+        # Validate required columns
+        required_cols = ["N", "P", "K", "temperature", "ph", "rainfall"]
+        missing_cols = set(required_cols) - set(input_df.columns)
+        if missing_cols:
+            raise ValueError(f"Missing required columns: {missing_cols}")
+
+        # Get top 5 predictions
+        top_predictions = get_top_n_classes(final_RF, input_df, label_encoder, n=5)
+
+        return {"predictions": top_predictions, "status": "success"}
+    except Exception as e:
+        logger.error(f"Prediction error: {str(e)}")
+        return {"predictions": [], "status": "failure", "error": str(e)}
+
+
+@app.post("/predict_crop")
+async def predict_crop_endpoint(input_data: CropInput):
+    try:
+        # Check if files exist
+        for path in [MODEL_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_crop(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": "Crop Recommendation API is running. Use /predict_crop endpoint to send input data."
+    }
+
+
+@app.get("/valid_inputs")
+async def get_valid_inputs():
+    return {
+        "N": {"min": 0, "max": 200, "unit": "kg/ha"},
+        "P": {"min": 0, "max": 200, "unit": "kg/ha"},
+        "K": {"min": 0, "max": 200, "unit": "kg/ha"},
+        "temperature": {"min": 0, "max": 50, "unit": "Celsius"},
+        "ph": {"min": 0, "max": 14, "unit": "pH"},
+        "rainfall": {"min": 0, "max": 2000, "unit": "mm"},
+        "possible_crops": VALID_CROPS,
+    }
+
+
+# conda activate crop_new
+# uvicorn crop_recommender:app --host 0.0.0.0 --port 8004

requirements.txt

@@ -0,0 +1,9 @@
+fastapi==0.112.2
+uvicorn==0.32.1
+pandas==2.2.3
+scikit-learn==1.6.1
+numpy==2.0.2
+joblib==1.4.2
+pydantic==2.10.3
+markupsafe==2.1.5
+anyio==4.7.0