Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.templating import Jinja2Templates | |
| from fastapi.staticfiles import StaticFiles | |
| import pandas as pd | |
| from prophet import Prophet | |
| import matplotlib.pyplot as plt | |
| import os | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| from pydantic import BaseModel | |
| import json | |
| app = FastAPI() | |
| # Mount static files | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| # Templates | |
| templates = Jinja2Templates(directory="templates") | |
| UPLOAD_FOLDER = 'static' | |
| if not os.path.exists(UPLOAD_FOLDER): | |
| os.makedirs(UPLOAD_FOLDER) | |
| # Load data | |
| yield_file = 'ICRISAT-District_Level_Data_30_Years.csv' | |
| soil_file = 'SoilHealthScores_by_District_2.csv' | |
| yield_df = pd.read_csv(yield_file) | |
| soil_df = pd.read_csv(soil_file) | |
| # Helper functions | |
| def get_soil_category(score): | |
| if score == 0: | |
| return "No Soil Health Data" | |
| elif score >= 4.5: | |
| return "Very Excellent Soil Health" | |
| elif score >= 4: | |
| return "Excellent Soil Health" | |
| elif score >= 3: | |
| return "Good Soil Health" | |
| elif score >= 2: | |
| return "Poor Soil Health" | |
| else: | |
| return "Very Poor Soil Health" | |
| def calculate_climate_score(yield_cat, soil_cat): | |
| score_map = { | |
| "Highly Recommended Crop": 90, | |
| "Good Crop": 70, | |
| "Poor Crop": 50, | |
| "Very Poor Crop": 30, | |
| "Very Excellent Soil Health": 95, | |
| "Excellent Soil Health": 85, | |
| "Good Soil Health": 65, | |
| "Poor Soil Health": 45, | |
| "Very Poor Soil Health": 25, | |
| "No Soil Health Data": 0 | |
| } | |
| return int((score_map[yield_cat] * 0.6) + (score_map[soil_cat] * 0.4)) | |
| soil_df['Soil_Category'] = soil_df['SoilHealthScore'].apply(get_soil_category) | |
| yield_columns = [col for col in yield_df.columns if 'YIELD (Kg per ha)' in col] | |
| base_crop_names = {col.split(' YIELD')[0]: col for col in yield_columns} | |
| def calculate_loan(c, predicted_yield, yield_cat, soil_cat, climate_score): | |
| crop_base_prices_per_hectare = { | |
| "RICE": 75000, | |
| "WHEAT": 65000, | |
| "KHARIF SORGHUM": 60000, | |
| "RABI SORGHUM": 62000, | |
| "SORGHUM": 61000, | |
| "PEARL MILLET": 50000, | |
| "MAIZE": 55000, | |
| "FINGER MILLET": 77000, | |
| "BARLEY": 48000, | |
| "CHICKPEA": 90000, | |
| "PIGEONPEA": 95000, | |
| "MINOR PULSES": 85000, | |
| "GROUNDNUT": 110000, | |
| "SESAMUM": 130000, | |
| "RAPESEED AND MUSTARD": 100000, | |
| "SAFFLOWER": 95000, | |
| "CASTOR": 88000, | |
| "LINSEED": 90000, | |
| "SUNFLOWER": 102000, | |
| "SOYABEAN": 98000, | |
| "OILSEEDS": 94000, | |
| "SUGARCANE": 150000, | |
| "COTTON": 120000 | |
| } | |
| base_loan = crop_base_prices_per_hectare[c] | |
| yield_multiplier = { | |
| "Highly Recommended Crop": 1.5, | |
| "Good Crop": 1.2, | |
| "Poor Crop": 0.8, | |
| "Very Poor Crop": 0.5 | |
| } | |
| soil_multiplier = { | |
| "Very Excellent Soil Health": 1.5, | |
| "Excellent Soil Health": 1.3, | |
| "Good Soil Health": 1.1, | |
| "Poor Soil Health": 0.9, | |
| "Very Poor Soil Health": 0.7, | |
| "No Soil Health Data": 0.5 | |
| } | |
| climate_multiplier = climate_score / 100 | |
| loan_amount = base_loan * yield_multiplier[yield_cat] * soil_multiplier[soil_cat] * climate_multiplier | |
| return round(loan_amount, 2) | |
| async def home(request: Request): | |
| return templates.TemplateResponse("index.html", {"request": request, "crops": base_crop_names.keys()}) | |
| async def get_crops(): | |
| return {"crops": list(base_crop_names.keys())} | |
| class PredictInput(BaseModel): | |
| crop: str | |
| district: str | |
| land_area: str | |
| async def predict(request: Request, input_data: PredictInput): | |
| crop_input = input_data.crop | |
| district_input = input_data.district | |
| land_area = input_data.land_area | |
| if not crop_input or not district_input or crop_input not in base_crop_names: | |
| raise HTTPException(status_code=400, detail="Invalid input. Please enter a valid crop and district.") | |
| yield_col = base_crop_names[crop_input] | |
| district_yield = yield_df[yield_df['Dist Name'] == district_input] | |
| district_soil = soil_df[soil_df['Dist Name'] == district_input] | |
| if district_yield.empty or district_soil.empty: | |
| raise HTTPException(status_code=400, detail="District data not found.") | |
| ts_data = district_yield[['Year', yield_col]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| if predicted_yield > 1000: | |
| yield_cat = "Highly Recommended Crop" | |
| color = "green" | |
| elif predicted_yield > 500: | |
| yield_cat = "Good Crop" | |
| color = "yellow" | |
| elif predicted_yield > 200: | |
| yield_cat = "Poor Crop" | |
| color = "orange" | |
| else: | |
| yield_cat = "Very Poor Crop" | |
| color = "red" | |
| soil_score = district_soil['SoilHealthScore'].values[0] | |
| soil_cat = district_soil['Soil_Category'].values[0] | |
| climate_score = calculate_climate_score(yield_cat, soil_cat) | |
| plt.figure(figsize=(10, 5)) | |
| model.plot(forecast) | |
| image_path = os.path.join(UPLOAD_FOLDER, "forecast.png") | |
| plt.savefig(image_path) | |
| plt.close() | |
| loan_amount = calculate_loan(crop_input, predicted_yield, yield_cat, soil_cat, climate_score) | |
| best_crop = None | |
| max_yield = 0 | |
| for crop, column in base_crop_names.items(): | |
| ts_data = district_yield[['Year', column]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| if len(ts_data) >= 5: | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| if predicted_yield > max_yield: | |
| max_yield = predicted_yield | |
| best_crop = crop | |
| plt.figure(figsize=(10, 5)) | |
| ts_data = district_yield[['Year', yield_col]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| plt.plot(ts_data['ds'], ts_data['y'], label=f'{crop_input} Yield') | |
| best_crop_data = district_yield[['Year', base_crop_names[best_crop]]].dropna() | |
| best_crop_data.columns = ['ds', 'y'] | |
| best_crop_data['ds'] = pd.to_datetime(best_crop_data['ds'], format='%Y') | |
| plt.plot(best_crop_data['ds'], best_crop_data['y'], label=f'{best_crop} Yield', linestyle='--') | |
| plt.xlabel('Year') | |
| plt.ylabel('Yield (Kg/ha)') | |
| plt.title(f"Yield Comparison for {crop_input} and Best Crop ({best_crop}) in {district_input}") | |
| plt.legend() | |
| image_path2 = os.path.join(UPLOAD_FOLDER, "forecast2.png") | |
| plt.grid(True) | |
| plt.savefig(image_path2) | |
| plt.close() | |
| result = { | |
| "crop": crop_input, | |
| "district": district_input, | |
| "predicted_yield": f"{round(predicted_yield, 2)}Kg/ha", | |
| "loan_amount": float(loan_amount) * float(land_area), | |
| "best_crop": best_crop, | |
| "yield_cat": yield_cat, | |
| "color": color, | |
| "soil_health": soil_cat, | |
| "climate_score": climate_score | |
| } | |
| return templates.TemplateResponse("index.html", { | |
| "request": request, | |
| "result": result, | |
| "image_path": image_path, | |
| "image_path2": image_path2, | |
| "crops": base_crop_names.keys() | |
| }) | |
| async def api_predict(crop: str, district: str, land: str): | |
| if not crop or not district: | |
| raise HTTPException(status_code=400, detail="Missing crop or district in request.") | |
| return await predict2(crop, district, land) | |
| async def predict2(crop_input: str, district_input: str, area: str): | |
| if not crop_input or not district_input or crop_input not in base_crop_names: | |
| raise HTTPException(status_code=400, detail="Invalid input. Please enter a valid crop and district.") | |
| yield_col = base_crop_names[crop_input] | |
| district_yield = yield_df[yield_df['Dist Name'] == district_input] | |
| district_soil = soil_df[soil_df['Dist Name'] == district_input] | |
| if district_yield.empty or district_soil.empty: | |
| raise HTTPException(status_code=400, detail="District data not found.") | |
| ts_data = district_yield[['Year', yield_col]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| if predicted_yield > 1000: | |
| yield_cat = "Highly Recommended Crop" | |
| color = "green" | |
| elif predicted_yield > 500: | |
| yield_cat = "Good Crop" | |
| color = "yellow" | |
| elif predicted_yield > 200: | |
| yield_cat = "Poor Crop" | |
| color = "orange" | |
| else: | |
| yield_cat = "Very Poor Crop" | |
| color = "red" | |
| soil_score = district_soil['SoilHealthScore'].values[0] | |
| soil_cat = district_soil['Soil_Category'].values[0] | |
| climate_score = calculate_climate_score(yield_cat, soil_cat) | |
| loan_amount = calculate_loan(crop_input, predicted_yield, yield_cat, soil_cat, climate_score) | |
| best_crop = None | |
| max_yield = 0 | |
| for crop, column in base_crop_names.items(): | |
| ts_data = district_yield[['Year', column]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| if len(ts_data) >= 5: | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| if predicted_yield > max_yield: | |
| max_yield = predicted_yield | |
| best_crop = crop | |
| top_crops = [] | |
| crop_yields = [] | |
| for crop, column in base_crop_names.items(): | |
| ts_data = district_yield[['Year', column]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| if len(ts_data) >= 5: | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| crop_yields.append((crop, predicted_yield)) | |
| top_crops = sorted(crop_yields, key=lambda x: x[1], reverse=True)[:3] | |
| top_crops_array = [crop for crop, yield_value in top_crops] | |
| ts_data_json = [{"Year": str(year.year), "Yield": yield_value} for year, yield_value in zip(ts_data['ds'], ts_data['y'])] | |
| best_crop_data = district_yield[['Year', base_crop_names[best_crop]]].dropna() | |
| best_crop_data.columns = ['ds', 'y'] | |
| best_crop_data['ds'] = pd.to_datetime(best_crop_data['ds'], format='%Y') | |
| best_crop_data_json = [{"Year": str(year.year), "Yield": yield_value} for year, yield_value in zip(best_crop_data['ds'], best_crop_data['y'])] | |
| result = { | |
| "crop": crop_input, | |
| "district": district_input, | |
| "predicted_yield": f"{round(predicted_yield, 2)} Kg/ha", | |
| "yield_category": yield_cat, | |
| "best_crop": top_crops_array, | |
| "soil_health": soil_cat, | |
| "score": climate_score, | |
| "loan_amount": f"{float(loan_amount)*float(area)}", | |
| } | |
| return result | |
| async def api_map(crop: str, district: str, land: str): | |
| if not crop or not district: | |
| raise HTTPException(status_code=400, detail="Missing crop or district in request.") | |
| return await map_data(crop, district, land) | |
| async def map_data(crop_input: str, district_input: str, area: str): | |
| if not crop_input or not district_input or crop_input not in base_crop_names: | |
| raise HTTPException(status_code=400, detail="Invalid input. Please enter a valid crop and district.") | |
| yield_col = base_crop_names[crop_input] | |
| district_yield = yield_df[yield_df['Dist Name'] == district_input] | |
| district_soil = soil_df[soil_df['Dist Name'] == district_input] | |
| if district_yield.empty or district_soil.empty: | |
| raise HTTPException(status_code=400, detail="District data not found.") | |
| ts_data = district_yield[['Year', yield_col]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| best_crop = None | |
| max_yield = 0 | |
| for crop, column in base_crop_names.items(): | |
| ts_data = district_yield[['Year', column]].dropna() | |
| ts_data.columns = ['ds', 'y'] | |
| ts_data['ds'] = pd.to_datetime(ts_data['ds'], format='%Y') | |
| if len(ts_data) >= 5: | |
| model = Prophet(yearly_seasonality=True, growth='flat') | |
| model.fit(ts_data) | |
| future = model.make_future_dataframe(periods=1, freq='YS') | |
| forecast = model.predict(future) | |
| predicted_yield = max(forecast.iloc[-1]['yhat'], 0) | |
| if predicted_yield > max_yield: | |
| max_yield = predicted_yield | |
| best_crop = crop | |
| ts_data_json = [{"Year": str(year.year), "Yield": yield_value} for year, yield_value in zip(ts_data['ds'], ts_data['y'])] | |
| best_crop_data = district_yield[['Year', base_crop_names[best_crop]]].dropna() | |
| best_crop_data.columns = ['ds', 'y'] | |
| best_crop_data['ds'] = pd.to_datetime(best_crop_data['ds'], format='%Y') | |
| best_crop_data_json = [{"Year": str(year.year), "Yield": yield_value} for year, yield_value in zip(best_crop_data['ds'], best_crop_data['y'])] | |
| result = { | |
| "ts_data": ts_data_json, | |
| "best_crop_data": best_crop_data_json | |
| } | |
| return result | |
| # To run the application: | |
| # if __name__ == "__main__": | |
| # import uvicorn | |
| # uvicorn.run(app, host="0.0.0.0", port=8000) |