bot_fastapi / app.py
httpsAkayush's picture
Add application file
7169e00
raw
history blame
30.9 kB
# app.py - Production-ready Hugging Face Spaces deployment
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
import torch
import numpy as np
import pandas as pd
import json
import gc
import os
import logging
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import PeftModel
from huggingface_hub import hf_hub_download
from typing import List, Dict, Any, Optional
import uvicorn
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Clear cache
torch.cuda.empty_cache()
gc.collect()
PARAMS = ["N","P","K","temperature","pH","rainfall","humidity"]
# Acceptable ranges
IGNORE_RANGES = {
"N": (-10, 10),
"P": (-10, 10),
"K": (-10, 10),
"temperature": (-0.2, 0.2),
"pH": (-0.2, 0.2),
"humidity": (-5, 5),
"rainfall": (-15, 15)
}
def evaluate_problems_and_diffs(required: np.ndarray, given: np.ndarray):
problems = []
diff_dict = {}
for i, param in enumerate(PARAMS):
diff = given[i] - required[i]
low, high = IGNORE_RANGES[param]
if not (low <= diff <= high):
status = "deficiency" if diff < 0 else "excess"
problems.append(f"{param}_{status}")
diff_dict[param] = diff
return problems, diff_dict
class AgriculturalAdvisor:
def __init__(self):
self.model = None
self.tokenizer = None
self.df1 = None
self.df2 = None
self.template = None
self.model_loaded = False
self.data_loaded = False
try:
self.load_data()
self.load_model()
logger.info("✅ Agricultural Advisor initialized successfully!")
except Exception as e:
logger.error(f"❌ Failed to initialize: {str(e)}")
def load_data(self):
"""Load datasets with fallback options"""
try:
# Try to load datasets
if os.path.exists('Crop_recommendation.csv'):
self.df1 = pd.read_csv('Crop_recommendation.csv')
logger.info("✅ Crop_recommendation.csv loaded")
else:
# Create fallback dataset
logger.warning("⚠️ Crop_recommendation.csv not found, creating fallback")
self.df1 = self.create_fallback_dataset()
if os.path.exists('sensor_Crop_Dataset.csv'):
self.df2 = pd.read_csv('sensor_Crop_Dataset.csv')
self.df2.rename(columns={"crop": "label"}, inplace=True)
self.df2 = self.df2.drop(["soil","variety"], axis=1, errors='ignore')
logger.info("✅ sensor_Crop_Dataset.csv loaded")
else:
logger.warning("⚠️ sensor_Crop_Dataset.csv not found")
self.df2 = pd.DataFrame()
# Load template
if os.path.exists("crop_template.json"):
with open("crop_template.json") as f:
self.template = json.load(f)
logger.info("✅ Template loaded")
else:
logger.warning("⚠️ Template not found, creating fallback")
self.template = self.create_fallback_template()
self.data_loaded = True
except Exception as e:
logger.error(f"❌ Error loading data: {str(e)}")
# Create minimal fallbacks
self.df1 = self.create_fallback_dataset()
self.df2 = pd.DataFrame()
self.template = self.create_fallback_template()
self.data_loaded = True
def create_fallback_dataset(self):
"""Create minimal dataset for demo"""
return pd.DataFrame({
'N': [80, 75, 85, 70, 90],
'P': [40, 35, 45, 30, 50],
'K': [67, 60, 70, 55, 75],
'temperature': [25, 27, 23, 30, 20],
'pH': [7.0, 6.8, 7.2, 6.5, 7.5],
'rainfall': [200, 180, 220, 150, 250],
'humidity': [60, 65, 55, 70, 50],
'label': ['rice', 'wheat', 'maize', 'cotton', 'sugarcane']
})
def create_fallback_template(self):
"""Create minimal template"""
return {
"rice": {
"N_deficiency": {
"Description": "Nitrogen deficiency causes yellowing of older leaves and stunted growth",
"Homemade/Natural Remedies": "Apply compost, farmyard manure, or green manures",
"Commercial Suggestions": "Apply urea fertilizer in split doses",
"Cultural Practices": "Use alternate wetting and drying irrigation",
"Crop-Specific Notes": "Critical during tillering stage"
},
"P_deficiency": {
"Description": "Phosphorus deficiency causes dark green to purplish leaves",
"Homemade/Natural Remedies": "Apply bone meal or rock phosphate",
"Commercial Suggestions": "Apply superphosphate as basal dose",
"Cultural Practices": "Maintain soil pH near neutral",
"Crop-Specific Notes": "Important for root and flower development"
}
},
"wheat": {
"N_deficiency": {
"Description": "Nitrogen deficiency in wheat causes chlorosis and poor tillering",
"Homemade/Natural Remedies": "Apply compost and green manures",
"Commercial Suggestions": "Apply urea in 2-3 splits",
"Cultural Practices": "Ensure proper drainage",
"Crop-Specific Notes": "Critical at tillering and grain filling"
}
}
}
def load_model(self):
"""Load model with error handling"""
try:
# Model configuration
base_model = "unsloth/gemma-3-1b-it"
adapter_path = "./unified_crop_model" # Local path
# Check if running on CPU or GPU
device = "cuda" if torch.cuda.is_available() else "cpu"
logger.info(f"🖥️ Using device: {device}")
# Configure quantization only for GPU
if device == "cuda":
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
bnb_4bit_compute_dtype="bfloat16"
)
self.model = AutoModelForCausalLM.from_pretrained(
base_model,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True
)
else:
# CPU inference
self.model = AutoModelForCausalLM.from_pretrained(
base_model,
torch_dtype=torch.float32,
trust_remote_code=True
)
# Try to load LoRA adapter
if os.path.exists(adapter_path):
try:
self.model = PeftModel.from_pretrained(
self.model,
adapter_path,
device_map="auto" if device == "cuda" else None
)
logger.info("✅ LoRA adapter loaded")
except Exception as e:
logger.warning(f"⚠️ Could not load LoRA adapter: {str(e)}")
logger.info("📝 Using base model without fine-tuning")
else:
logger.warning("⚠️ LoRA adapter not found, using base model")
# Load tokenizer
tokenizer_path = adapter_path if os.path.exists(adapter_path) else base_model
self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_path, trust_remote_code=True)
# Set pad token if not exists
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
self.model_loaded = True
logger.info("✅ Model loaded successfully!")
except Exception as e:
logger.error(f"❌ Failed to load model: {str(e)}")
self.model_loaded = False
def analyze_crop_conditions(self, crop, N, P, K, temp, humidity, pH, rainfall):
"""Analyze crop conditions with comprehensive error handling"""
if not self.data_loaded:
return "❌ Data not loaded properly. Please refresh the page."
if not self.model_loaded:
return "⚠️ Model not loaded. Providing basic analysis without AI recommendations."
try:
given = [N, P, K, temp, pH, rainfall, humidity]
# Find crop in datasets
if crop in self.df1['label'].values:
df = self.df1[self.df1['label']==crop]
elif not self.df2.empty and crop in self.df2['label'].values:
df = self.df2[self.df2['label']==crop]
else:
available_crops = list(self.df1['label'].unique())
return f"❌ Crop '{crop}' not found in database. Available crops: {', '.join(available_crops)}"
df_values = df.drop('label', axis=1)
df_array = np.array(df_values)
# MSE computation
mse_list = []
for row in df_array:
mse = np.mean((np.array(row) - np.array(given))**2)
mse_list.append(mse)
best_index = np.argmin(mse_list)
required = df_array[best_index].tolist()
problems, diff_dict = evaluate_problems_and_diffs(required, given)
if not problems:
return "✅ **Great!** No significant issues detected. Current conditions are within acceptable ranges for optimal growth."
# ==============================
# Detailed Default Template
# ==============================
default_template = {
"general": {
"nitrogen_deficiency": {
"Description": "Leaves appear pale or yellowish; growth may be slow.",
"Homemade/Natural Remedies": "Apply composted manure, cow dung, or green manure.",
"Commercial Suggestions": "Use balanced NPK fertilizer with higher nitrogen content.",
"Cultural Practices": "Rotate crops; avoid over-harvesting nitrogen-rich leaves.",
"Crop-Specific Notes": "Sensitive crops like leafy greens show symptoms faster."
},
"nitrogen_excess": {
"Description": "Excessive vegetative growth; flowering/fruiting may be delayed.",
"Homemade/Natural Remedies": "Limit nitrogen-rich organic inputs like fresh manure.",
"Commercial Suggestions": "Reduce nitrogen fertilizer; maintain balanced NPK ratios.",
"Cultural Practices": "Prune excess growth; monitor soil nutrient levels.",
"Crop-Specific Notes": "Fruit crops may produce fewer fruits if over-fertilized with nitrogen."
},
"phosphorus_deficiency": {
"Description": "Stunted growth; leaves may show dark green/purplish coloration.",
"Homemade/Natural Remedies": "Use bone meal, rock phosphate, or composted organic matter.",
"Commercial Suggestions": "Apply phosphorus-rich fertilizers like single superphosphate (SSP).",
"Cultural Practices": "Maintain soil pH around 6–7; avoid acidic soils.",
"Crop-Specific Notes": "Root crops may be most affected due to poor root development."
},
"phosphorus_excess": {
"Description": "Can interfere with micronutrient absorption (Zn, Fe).",
"Homemade/Natural Remedies": "Avoid adding extra phosphorus-containing amendments.",
"Commercial Suggestions": "Use balanced fertilizers; avoid repeated high P applications.",
"Cultural Practices": "Rotate crops to prevent phosphorus build-up.",
"Crop-Specific Notes": "Cereals are more sensitive to high phosphorus than legumes."
},
"potassium_deficiency": {
"Description": "Leaf edges turn brown, scorching; weak stems.",
"Homemade/Natural Remedies": "Add wood ash or composted banana peels.",
"Commercial Suggestions": "Apply potassium sulfate or muriate of potash.",
"Cultural Practices": "Ensure proper irrigation; avoid water stress.",
"Crop-Specific Notes": "Potato and tomato show clear leaf-edge symptoms."
},
"potassium_excess": {
"Description": "May reduce magnesium and calcium uptake.",
"Homemade/Natural Remedies": "Avoid excessive potassium-containing composts.",
"Commercial Suggestions": "Balance with magnesium/calcium fertilizers.",
"Cultural Practices": "Test soil regularly for K levels.",
"Crop-Specific Notes": "Leafy vegetables may show interveinal chlorosis if Mg is low."
},
"iron_deficiency": {
"Description": "Young leaves turn yellow with green veins (chlorosis).",
"Homemade/Natural Remedies": "Foliar spray with iron sulfate or iron chelates.",
"Commercial Suggestions": "Apply chelated iron to soil or foliage.",
"Cultural Practices": "Maintain soil pH below 7.5 for better uptake.",
"Crop-Specific Notes": "Fruit trees like apple and citrus are sensitive."
},
"iron_excess": {
"Description": "Can cause nutrient imbalance and toxicity.",
"Homemade/Natural Remedies": "Avoid iron-rich amendments in high-Fe soils.",
"Commercial Suggestions": "Test soil before adding iron fertilizers.",
"Cultural Practices": "Improve drainage in high-iron soils.",
"Crop-Specific Notes": "Rice paddies may tolerate slightly higher iron naturally."
},
"water_deficiency": {
"Description": "Wilting, leaf curl, and reduced yield.",
"Homemade/Natural Remedies": "Mulch soil to retain moisture; use organic matter.",
"Commercial Suggestions": "Implement drip or sprinkler irrigation.",
"Cultural Practices": "Schedule watering based on crop stage and weather.",
"Crop-Specific Notes": "Tomatoes and peppers are highly sensitive to water stress."
},
"water_excess": {
"Description": "Root rot, yellowing leaves, poor aeration.",
"Homemade/Natural Remedies": "Improve soil drainage using sand or organic matter.",
"Commercial Suggestions": "Raised beds; controlled irrigation.",
"Cultural Practices": "Avoid waterlogging; monitor soil moisture regularly.",
"Crop-Specific Notes": "Root crops like carrots and potatoes are prone to rot."
},
"pH_deficiency": {
"Description": "Soil too acidic (<5.5); stunted growth.",
"Homemade/Natural Remedies": "Apply wood ash or crushed eggshells.",
"Commercial Suggestions": "Use agricultural lime to raise pH.",
"Cultural Practices": "Test soil pH regularly; avoid acid-forming fertilizers.",
"Crop-Specific Notes": "Legumes prefer slightly acidic to neutral pH."
},
"pH_excess": {
"Description": "Soil too alkaline (>8); micronutrient deficiencies.",
"Homemade/Natural Remedies": "Incorporate organic matter like compost.",
"Commercial Suggestions": "Apply elemental sulfur to lower soil pH.",
"Cultural Practices": "Select tolerant crop varieties.",
"Crop-Specific Notes": "Tomatoes and spinach are sensitive to high pH."
},
"temperature_stress": {
"Description": "Too high or too low temperature affects growth and yield.",
"Homemade/Natural Remedies": "Shade nets or mulching to regulate temperature.",
"Commercial Suggestions": "Use protective covers or greenhouses.",
"Cultural Practices": "Plant at optimal seasonal windows.",
"Crop-Specific Notes": "Tomato, cucumber, and leafy greens are sensitive."
},
"pest_disease_issue": {
"Description": "Presence of pests or disease symptoms.",
"Homemade/Natural Remedies": "Neem oil, garlic extract, or organic sprays.",
"Commercial Suggestions": "Use approved pesticides or fungicides; follow IPM.",
"Cultural Practices": "Sanitation, crop rotation, resistant varieties.",
"Crop-Specific Notes": "Leafy vegetables and solanaceous crops need regular monitoring."
}
}
}
# selected issues dictionary
selected = {}
# Step 1: Check crop-specific template first
for prob in problems:
if prob in self.template.get(crop, {}):
selected[prob] = self.template[crop][prob]
# Step 2: If nothing found, use default template
if not selected:
for prob in problems:
if prob in default_template.get("general", {}):
selected[prob] = default_template["general"][prob]
# Step 3: If still nothing found, fallback message
if not selected:
issues_text = ', '.join(problems)
return f"⚠️ **Issues detected:** {issues_text}\n\n❗ No recommendations available even in the default template."
# Step 4: Build formatted output
context = f"Crop: {crop}\n"
for issue, details in selected.items():
context += f"\n## {issue.replace('_',' ').title()}\n"
for k, v in details.items():
context += f"💠 {k}: {v}\n"
# Generate AI recommendations if model available
ai_response = ""
if self.model_loaded:
try:
ai_response = self.generate_ai_recommendations(context)
except Exception as e:
logger.error(f"AI generation failed: {str(e)}")
ai_response = "AI recommendations temporarily unavailable."
# Format response
issues_summary = f"📊 **Issues Detected:** {', '.join(problems)}\n\n"
diff_summary = f"📈 **Parameter Differences:** {', '.join([f'{k}: {v:+.1f}' for k, v in diff_dict.items()])}\n\n"
# template_info = "📋 **Available Information:**\n"
# for issue, details in selected.items():
# template_info += f"\n**{issue.replace('_', ' ').title()}:**\n"
# template_info += f"• Description: {details.get('Description', 'N/A')}\n"
# template_info += f"• Natural Remedies: {details.get('Homemade/Natural Remedies', 'N/A')}\n"
# template_info += f"• Commercial Solutions: {details.get('Commercial Suggestions', 'N/A')}\n\n"
ai_section = f"🤖 **AI Recommendations:**\n{ai_response}\n" if ai_response else ""
return f"{issues_summary}{ai_section}"
except Exception as e:
logger.error(f"Analysis error: {str(e)}")
return f"❌ Error during analysis: {str(e)}"
def generate_ai_recommendations(self, context):
"""Generate AI recommendations with proper error handling"""
try:
messages = [
{
"role": "system",
"content": [{"type": "text", "text": "You are a helpful agronomy assistant. Based on soil conditions, suggest remedies for the detected crop issues."}]
},
{
"role": "user",
"content": [{"type": "text", "text": f"Here is reference info:\n{context}\n\nPlease give a concise recommendation."}]
}
]
inputs = self.tokenizer.apply_chat_template(
messages,
add_generation_prompt=True,
return_tensors="pt",
tokenize=True,
return_dict=True,
).to(self.model.device)
with torch.no_grad():
output = self.model.generate(
**inputs,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
pad_token_id=self.tokenizer.eos_token_id,
do_sample=True
)
# Decode response
response = self.tokenizer.decode(
output[0][inputs['input_ids'].shape[1]:],
skip_special_tokens=True
)
return response.strip()
except Exception as e:
logger.error(f"AI generation error: {str(e)}")
return f"AI recommendations temporarily unavailable due to: {str(e)}"
# Initialize advisor with error handling
logger.info("🚀 Initializing Agricultural Advisor...")
try:
advisor = AgriculturalAdvisor()
initialization_status = "✅ System Ready"
crops_available = list(advisor.df1['label'].unique())
except Exception as e:
logger.error(f"❌ Failed to initialize advisor: {str(e)}")
advisor = None
initialization_status = f"❌ Initialization Failed: {str(e)}"
crops_available = ["rice", "wheat", "maize"] # Fallback
# def get_crop_recommendations(crop, N, P, K, temperature, humidity, pH, rainfall):
# """Gradio interface function"""
# if advisor is None:
# return f"❌ System not initialized properly. Status: {initialization_status}"
# try:
# return advisor.analyze_crop_conditions(
# crop, N, P, K, temperature, humidity, pH, rainfall
# )
# except Exception as e:
# logger.error(f"Interface error: {str(e)}")
# return f"❌ Error processing request: {str(e)}"
## Pydantic models for API
class CropAnalysisRequest(BaseModel):
crop: str = Field(..., description="Name of the crop to analyze")
N: float = Field(..., ge=0, le=300, description="Nitrogen content (kg/ha)")
P: float = Field(..., ge=0, le=150, description="Phosphorus content (kg/ha)")
K: float = Field(..., ge=0, le=200, description="Potassium content (kg/ha)")
temperature: float = Field(..., ge=0, le=50, description="Temperature (°C)")
humidity: float = Field(..., ge=0, le=100, description="Humidity (%)")
pH: float = Field(..., ge=3, le=10, description="Soil pH level")
rainfall: float = Field(..., ge=0, le=2000, description="Rainfall (mm)")
class CropAnalysisResponse(BaseModel):
success: bool
message: str
recommendations: str
status: str
class SystemStatusResponse(BaseModel):
status: str
model_loaded: bool
data_loaded: bool
available_crops: List[str]
# Initialize advisor with error handling
logger.info("🚀 Initializing Agricultural Advisor...")
try:
advisor = AgriculturalAdvisor()
initialization_status = "✅ System Ready"
crops_available = list(advisor.df1['label'].unique())
except Exception as e:
logger.error(f"❌ Failed to initialize advisor: {str(e)}")
advisor = None
initialization_status = f"❌ Initialization Failed: {str(e)}"
crops_available = ["rice", "wheat", "maize"] # Fallback
# FastAPI app
app = FastAPI(
title="🌾 Agricultural Advisor API",
description="AI-powered agricultural advisor for crop recommendations based on soil and climate conditions",
version="1.0.0"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure as needed for production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/", response_class=HTMLResponse)
async def root():
"""Serve basic HTML interface"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>🌾 Agricultural Advisor API</title>
<style>
body { font-family: Arial, sans-serif; margin: 40px; background: #f5f5f5; }
.container { max-width: 800px; margin: 0 auto; background: white; padding: 30px; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }
h1 { color: #2e7d32; text-align: center; }
.endpoint { background: #f8f9fa; padding: 15px; margin: 10px 0; border-radius: 5px; border-left: 4px solid #4caf50; }
.method { color: #1976d2; font-weight: bold; }
.example { background: #e8f5e8; padding: 10px; margin: 10px 0; border-radius: 5px; font-family: monospace; }
pre { overflow-x: auto; }
</style>
</head>
<body>
<div class="container">
<h1>🌾 Agricultural Advisor API</h1>
<p>AI-powered agricultural advisor for crop recommendations based on soil and climate conditions.</p>
<h2>📋 Available Endpoints</h2>
<div class="endpoint">
<span class="method">GET</span> <strong>/status</strong>
<p>Get system status and available crops</p>
</div>
<div class="endpoint">
<span class="method">POST</span> <strong>/analyze</strong>
<p>Analyze crop conditions and get recommendations</p>
<div class="example">
<strong>Example Request:</strong>
<pre>{
"crop": "rice",
"N": 80,
"P": 40,
"K": 67,
"temperature": 25,
"humidity": 60,
"pH": 7.0,
"rainfall": 200
}</pre>
</div>
</div>
<div class="endpoint">
<span class="method">GET</span> <strong>/crops</strong>
<p>Get list of available crops</p>
</div>
<h2>📖 Documentation</h2>
<p>Visit <a href="/docs">/docs</a> for interactive API documentation</p>
<p>Visit <a href="/redoc">/redoc</a> for alternative documentation</p>
<h2>🔧 System Status</h2>
<p><strong>Status:</strong> """ + initialization_status + """</p>
<p><strong>Available Crops:</strong> """ + ", ".join(crops_available) + """</p>
</div>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.get("/status", response_model=SystemStatusResponse)
async def get_system_status():
"""Get system status"""
if advisor is None:
return SystemStatusResponse(
status=initialization_status,
model_loaded=False,
data_loaded=False,
available_crops=crops_available
)
return SystemStatusResponse(
status=initialization_status,
model_loaded=advisor.model_loaded,
data_loaded=advisor.data_loaded,
available_crops=crops_available
)
@app.get("/crops")
async def get_available_crops():
"""Get list of available crops"""
return {"crops": crops_available}
@app.post("/analyze", response_model=CropAnalysisResponse)
async def analyze_crop(request: CropAnalysisRequest):
"""Analyze crop conditions and provide recommendations"""
if advisor is None:
raise HTTPException(
status_code=503,
detail=f"System not initialized properly. Status: {initialization_status}"
)
try:
# Validate crop
if request.crop not in crops_available:
raise HTTPException(
status_code=400,
detail=f"Crop '{request.crop}' not available. Available crops: {', '.join(crops_available)}"
)
# Analyze crop conditions using the same method as Gradio version
recommendations = advisor.analyze_crop_conditions(
request.crop, request.N, request.P, request.K,
request.temperature, request.humidity, request.pH, request.rainfall
)
# Determine status based on recommendations
if "❌" in recommendations:
status = "error"
elif "⚠️" in recommendations:
status = "warning"
elif "✅" in recommendations:
status = "success"
else:
status = "info"
return CropAnalysisResponse(
success=True,
message="Analysis completed successfully",
recommendations=recommendations,
status=status
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Analysis error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Error processing request: {str(e)}"
)
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"system_status": initialization_status,
"timestamp": pd.Timestamp.now().isoformat()
}
if __name__ == "__main__":
uvicorn.run(
app,
host="0.0.0.0",
port=8000,
log_level="info"
)