Armia-Gamal/AgriTech-API
1
1import os2import joblib3import requests4import numpy as np5from fastapi import FastAPI, HTTPException6from pydantic import BaseModel, Field7 8app = FastAPI(title="NABTA AgriTech API (Irrigation & Crop Recommendation)")9 10BASE_DIR = os.path.dirname(os.path.abspath(__file__))11API_KEY = os.getenv("API_KEY", "4bfbf537c3bf4dd1b3e01516263005")12 13# ==============================================================================14# CONFIGURATIONS15# ==============================================================================16CROP_KC_LOOKUP = {17 "Rice": 1.20, "Banana": 1.20, "Tomato": 1.15, "Potato": 1.15, "Cabbage": 1.05,18 "Maize": 1.20, "Wheat": 1.15, "Cotton": 1.15, "Soybean": 1.15, "Onion": 1.05,19 "Sunflower": 1.10, "Coffee": 1.05, "Mango": 1.10, "Sugarcane": 1.25, "Sorghum": 1.00,20 "Barley": 1.10, "Chickpea": 1.00, "Groundnut": 1.0521}22 23CROP_MAD_FACTORS = {24 "Rice": 0.40, "Banana": 1.00, "Tomato": 0.95, "Potato": 0.90, "Cabbage": 0.95,25 "Maize": 0.85, "Wheat": 0.80, "Cotton": 0.75, "Soybean": 0.75, "Onion": 1.05,26 "Sunflower": 0.70, "Coffee": 0.85, "Mango": 0.65, "Sugarcane": 0.80, "Sorghum": 0.60,27 "Barley": 0.75, "Chickpea": 0.55, "Groundnut": 0.7028}29 30# ==============================================================================31# LOAD MODELS32# ==============================================================================33try:34 clf_model = joblib.load(os.path.join(BASE_DIR, "decision_model.pkl"))35 reg_model = joblib.load(os.path.join(BASE_DIR, "water_req_model.pkl"))36 crop_model = joblib.load(os.path.join(BASE_DIR, "crop_rec_model.pkl"))37 encoders = joblib.load(os.path.join(BASE_DIR, "encoders.pkl"))38 feat_info = joblib.load(os.path.join(BASE_DIR, "feature_lists.pkl"))39 40 CROP_LIST = sorted(list(encoders['crop_type'].classes_))41 SOIL_LIST = sorted(list(encoders['Soil_Type'].classes_))42 SOIL_FC = feat_info['soil_fc']43 SOIL_WP = feat_info['soil_wp']44 45 SOIL_MOISTURE_RANGES = {46 soil: {47 "min": round(max(0.0, SOIL_WP[soil] - 0.02), 2),48 "max": round(SOIL_FC[soil] + 0.04, 2),49 }50 for soil in SOIL_LIST51 }52 print("✅ All AgriTech V3.1 Models Loaded Successfully!")53except Exception as e:54 print(f"❌ Error loading ML Models: {e}")55 CROP_LIST, SOIL_LIST, SOIL_MOISTURE_RANGES = [], [], {}56 57# ==============================================================================58# SCHEMAS (Pydantic)59# ==============================================================================60class CoordsRequest(BaseModel):61 lat: float62 lon: float63 64class IrrigationRequest(BaseModel):65 city: str66 crop: str67 soil: str68 moisture: float69 70class RecommendationRequest(BaseModel):71 city: str72 soil: str73 n_val: float = Field(80.0, alias="N")74 p_val: float = Field(45.0, alias="P")75 k_val: float = Field(80.0, alias="K")76 77# ==============================================================================78# HELPERS79# ==============================================================================80def fetch_weather(lat=None, lon=None, city=None):81 query = f"{lat},{lon}" if lat and lon else city82 if not query: return None83 url = f"https://api.weatherapi.com/v1/current.json?key={API_KEY}&q={query}&aqi=no"84 try:85 res = requests.get(url, timeout=8)86 if res.status_code == 200:87 d = res.json()88 return {89 "temp": d["current"]["temp_c"],90 "hum": d["current"]["humidity"],91 "wind": d["current"]["wind_kph"],92 "rain": d["current"].get("precip_mm", 0),93 "city": d["location"]["name"] + "، " + d["location"]["country"],94 }95 return None96 except:97 return None98 99def infer_climate_zone(temp, rain):100 if temp > 30 and rain < 15: return "Arid"101 elif temp > 24 and rain < 35: return "Semi-arid"102 elif temp > 25 and rain >= 70: return "Tropical"103 elif 19 <= temp <= 24 and 25 <= rain <= 50: return "Mediterranean"104 else: return "Temperate"105 106def calc_depletion(sm, soil, crop_name):107 fc = SOIL_FC.get(soil, 0.28)108 wp = SOIL_WP.get(soil, 0.12)109 dep = float(np.clip((fc - sm) / (fc - wp + 1e-5), 0, 1.5))110 smpfc = float(np.clip(sm / (fc + 1e-5), 0, 1.2))111 crop_factor = CROP_MAD_FACTORS.get(crop_name, 0.75)112 mad_fraction = np.clip(0.45 * crop_factor, 0.15, 0.65)113 mad = wp + mad_fraction * (fc - wp)114 return dep, smpfc, fc, wp, mad115 116# ==============================================================================117# ROUTES118# ==============================================================================119@app.get("/")120async def root():121 return {"message": "Welcome to NABTA AgriTech API 🌿", "docs_url": "/docs"}122 123@app.get("/meta-data")124async def get_meta_data():125 return {126 "crops": CROP_LIST,127 "soils": SOIL_LIST,128 "soil_moisture_ranges": SOIL_MOISTURE_RANGES129 }130 131@app.post("/weather")132async def get_weather(payload: CoordsRequest):133 weather = fetch_weather(lat=payload.lat, lon=payload.lon)134 if weather: return weather135 raise HTTPException(status_code=400, detail="تعذر جلب الطقس الإحداثي الحالي")136 137@app.post("/predict-irrigation")138async def predict_irrigation(payload: IrrigationRequest):139 if payload.crop not in CROP_LIST or payload.soil not in SOIL_LIST:140 raise HTTPException(status_code=400, detail="نوع المحصول أو التربة غير مدعوم")141 142 moisture_range = SOIL_MOISTURE_RANGES[payload.soil]143 if not (moisture_range['min'] <= payload.moisture <= moisture_range['max']):144 raise HTTPException(status_code=400, detail=f"الرطوبة لتربة {payload.soil} يجب أن تكون بين {moisture_range['min']*100}% و {moisture_range['max']*100}%")145 146 w = fetch_weather(city=payload.city)147 if not w: raise HTTPException(status_code=400, detail="فشل سحب الطقس الفعلي")148 149 zone_name = infer_climate_zone(w['temp'], w['rain'])150 crop_enc = encoders['crop_type'].transform([payload.crop])[0]151 soil_enc = encoders['Soil_Type'].transform([payload.soil])[0]152 zone_enc = encoders['Climate_Zone'].transform([zone_name])[0]153 154 dep, smpfc, fc, wp, mad = calc_depletion(payload.moisture, payload.soil, payload.crop)155 156 X1 = np.array([[w['temp'], w['hum'], w['rain'], dep, smpfc, crop_enc, soil_enc, fc, wp]])157 prediction = int(clf_model.predict(X1)[0])158 confidence = round(float(clf_model.predict_proba(X1)[0][prediction]) * 100, 1)159 160 water_amount = 0.0161 if prediction == 1:162 ra_lookup = {"Tropical": 30, "Arid": 33, "Temperate": 22, "Semi-arid": 29, "Mediterranean": 26}163 eto = round(0.0023 * (w['temp'] + 17.8) * (9**0.5) * 0.408 * ra_lookup.get(zone_name, 28), 2)164 kc_val = CROP_KC_LOOKUP.get(payload.crop, 1.10)165 X2 = np.array([[w['temp'], w['hum'], w['rain'], crop_enc, zone_enc, eto, kc_val, payload.moisture]])166 water_amount = round(float(reg_model.predict(X2)[0]), 1)167 168 return {169 "weather": w,170 "climate_zone": zone_name,171 "needs_irrigation": prediction == 1,172 "decision": "🚿 ري الآن" if prediction == 1 else "✅ لا يحتاج ري",173 "water_amount_liters_per_m2": water_amount,174 "confidence_percentage": confidence,175 "soil_moisture_percentage": round(payload.moisture * 100, 1),176 "mad_percentage": round(mad * 100, 1)177 }178 179@app.post("/recommend-crop")180async def recommend_crop(payload: RecommendationRequest):181 if payload.soil not in SOIL_LIST: raise HTTPException(status_code=400, detail="نوع التربة غير مدعوم")182 183 w = fetch_weather(city=payload.city)184 if not w: raise HTTPException(status_code=400, detail="تعذر جلب الطقس")185 186 soil_enc = encoders['Soil_Type'].transform([payload.soil])[0]187 N_P_ratio = payload.n_val / (payload.p_val + 1)188 N_K_ratio = payload.n_val / (payload.k_val + 1)189 NPK_total = payload.n_val + payload.p_val + payload.k_val190 191 X3 = np.array([[payload.n_val, payload.p_val, payload.k_val, w['temp'], w['hum'], w['rain'], soil_enc, N_P_ratio, N_K_ratio, NPK_total]])192 193 prob = crop_model.predict_proba(X3)[0]194 top3_idx = np.argsort(prob)[::-1][:3]195 return {196 "city": w['city'],197 "soil_type": payload.soil,198 "top_3_crops": [{"crop": CROP_LIST[i], "confidence": round(float(prob[i]) * 100, 1)} for i in top3_idx]199 }