import os import numpy as np import pandas as pd import torch import timm import torch.nn as nn from flask import Flask, request, jsonify, render_template_string from torchvision import transforms from PIL import Image from math import radians, cos, sin, sqrt, atan2 import joblib import requests from dotenv import load_dotenv from werkzeug.utils import secure_filename # ========================================================= # ENV & APP CONFIG # ========================================================= load_dotenv() OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY") app = Flask(__name__) UPLOAD_FOLDER = "/tmp/uploads" os.makedirs(UPLOAD_FOLDER, exist_ok=True) # ========================================================= # GLOBAL PLACEHOLDERS (LAZY LOADING) # ========================================================= soil_model = None crop_model = None crop_model_label = None soil_df = None agri_df = None class_labels = [ "Alluvial Soil", "Black Soil", "Clay Soil", "Non Soil", "Red Soil" ] num_classes = len(class_labels) # ========================================================= # MODEL & DATA LOADERS (LAZY) # ========================================================= def load_soil_model(): global soil_model if soil_model is None: model = timm.create_model( "vit_base_patch16_224", pretrained=False, num_classes=num_classes ) model.head = nn.Linear(model.head.in_features, num_classes) state = torch.load( "models/best_vit_model.pth", map_location=torch.device("cpu") ) model.load_state_dict(state) model.eval() soil_model = model return soil_model def load_crop_model(): global crop_model, crop_model_label if crop_model is None: crop_model = joblib.load("models/model_random_forest.joblib") crop_model_label = joblib.load("models/label_encoder.joblib") return crop_model, crop_model_label def load_dataframes(): global soil_df, agri_df if soil_df is None: soil_df = pd.read_csv("data/soil_data.csv") if agri_df is None: agri_df = pd.read_csv("data/tips_menanam_dan_manfaat_tanaman.csv") # ========================================================= # IMAGE PREPROCESS # ========================================================= def preprocess_image(img_path): transform = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize(mean=[0.5]*3, std=[0.5]*3) ]) img = Image.open(img_path).convert("RGB") return transform(img).unsqueeze(0) # ========================================================= # SOIL PREDICTION # ========================================================= def predict_soil_type(img_path): model = load_soil_model() img_tensor = preprocess_image(img_path) with torch.no_grad(): outputs = model(img_tensor) probabilities = torch.softmax(outputs, dim=1).numpy()[0] idx = int(np.argmax(probabilities)) return class_labels[idx], float(probabilities[idx]) # ========================================================= # HAVERSINE & SOIL DATA # ========================================================= def find_nearest_soil_data_weighted(soil_type, lat, lon): load_dataframes() filtered = soil_df[soil_df["Soil_Type"] == soil_type].copy() if filtered.empty: return None user_lat, user_lon = radians(lat), radians(lon) def haversine(row): lat2, lon2 = radians(row["Location_Latitude"]), radians(row["Location_Longitude"]) dlat, dlon = lat2 - user_lat, lon2 - user_lon a = sin(dlat/2)**2 + cos(user_lat)*cos(lat2)*sin(dlon/2)**2 c = 2 * atan2(sqrt(a), sqrt(1-a)) return 6371 * c filtered["Distance_km"] = filtered.apply(haversine, axis=1) row = filtered.nsmallest(1, "Distance_km").iloc[0] return { "latitude": float(row["Location_Latitude"]), "longitude": float(row["Location_Longitude"]), "pH": float(row["pH"]), "N": float(row["Nitrogen_N_ppm"]), "P": float(row["Phosphorus_P_ppm"]), "K": float(row["Potassium_K_ppm"]), "distance_km": float(row["Distance_km"]) } # ========================================================= # WEATHER & GEO # ========================================================= def get_weather_data(lat, lon): url = ( f"https://api.openweathermap.org/data/2.5/weather" f"?lat={lat}&lon={lon}&appid={OPENWEATHER_API_KEY}&units=metric" ) res = requests.get(url, timeout=10) if res.status_code != 200: return None data = res.json() return { "temperature": float(data["main"]["temp"]), "humidity": float(data["main"]["humidity"]) } def get_location_name(lat, lon): try: url = f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json" headers = {"User-Agent": "soil-api/1.0"} res = requests.get(url, headers=headers, timeout=10) if res.status_code == 200: return res.json().get("display_name", "Tidak ditemukan") except Exception: pass return "Tidak ditemukan" # ========================================================= # FARMING TIPS # ========================================================= def get_farming_tips(df, crop_name): match = df[df["Nama Tanaman"].str.lower() == str(crop_name).lower()] if not match.empty: row = match.iloc[0] return { "Nama Tanaman": row["Nama Tanaman"], "Tips Menanam": row.get("Tips Menanam", "Tidak tersedia"), "Manfaat": row.get("Manfaat", "Tidak tersedia") } return {"Tanaman": crop_name, "Pesan": "Data tidak tersedia"} # ========================================================= # ROUTES # ========================================================= @app.route("/") def index(): return render_template_string("""

🌱 Soil & Crop Recommendation API

POST /analyze with image, lat, lon

""") @app.route("/analyze", methods=["POST"]) def analyze(): if "image" not in request.files: return jsonify({"error": "Gambar tidak ditemukan"}), 400 try: lat = float(request.form.get("lat")) lon = float(request.form.get("lon")) except: return jsonify({"error": "Koordinat tidak valid"}), 400 image_file = request.files["image"] filename = secure_filename(image_file.filename) image_path = os.path.join(UPLOAD_FOLDER, filename) image_file.save(image_path) soil_type, soil_acc = predict_soil_type(image_path) nearest = find_nearest_soil_data_weighted(soil_type, lat, lon) if not nearest: return jsonify({"error": "Data tanah tidak ditemukan"}), 404 weather = get_weather_data(lat, lon) if not weather: return jsonify({"error": "Gagal mengambil cuaca"}), 500 crop_model, crop_label = load_crop_model() input_df = pd.DataFrame([{ "temperature": weather["temperature"], "humidity": weather["humidity"], "ph": nearest["pH"], "N": nearest["N"], "P": nearest["P"], "K": nearest["K"] }]) proba = crop_model.predict_proba(input_df)[0] top_idx = np.argsort(proba)[::-1][:5] load_dataframes() crops = [ crop_label.inverse_transform([crop_model.classes_[i]])[0] for i in top_idx ] percentages = [round(float(proba[i]) * 100, 2) for i in top_idx] tips = [get_farming_tips(agri_df, c) for c in crops] return jsonify({ "soil_type": soil_type, "soil_accuracy": round(soil_acc * 100, 2), "location": get_location_name(nearest["latitude"], nearest["longitude"]), "weather": weather, "recommended_crops": [ {"crop": c, "percentage": p} for c, p in zip(crops, percentages) ], "farming_tips": tips }) # ========================================================= # RUN SERVER (HF SPACES) # ========================================================= if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)